<commit_before>package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"github.com/vivowares/octopus/Godeps/_workspace/src/github.com/gorilla/websocket"
	"github.com/vivowares/octopus/Godeps/_workspace/src/github.com/parnurzeal/gorequest"
	"github.com/vivowares/octopus/Godeps/_workspace/src/github.com/satori/go.uuid"
	. "github.com/vivowares/octopus/utils"
	"log"
	"math/rand"
	"net/http"
	"net/url"
	"runtime"
	"strings"
	"sync"
	"time"
)

// go run tasks/benchmark.go  -host=172.16.58.3 -ports=8080:8081 -user=root -passwd=<PASSWORD> -fields=temperature:float -c=1 -p=1 -m=1

func main() {
	host := flag.String("host", "localhost", "the target server host")
	ports := flag.String("ports", "8080:8081", "the http port and device port")
	fields := flag.String("fields", "temperature:float", "fields that are used for bench test. Format: 'field1:type1,field2:type2'")
	user := flag.String("user", "root", "username for authenticating octopus")
	passwd := flag.String("passwd", "<PASSWORD>", "passwd for authenticating octopus")

	c := flag.Int("c", 1000, "number of concurrent clients")
	p := flag.Int("p", 100, "number of ping messages to send")
	m := flag.Int("m", 50, "number of payload messages to send")
	r := flag.Duration("r", 15*time.Second, "wait time for reading messages")
	w := flag.Duration("w", 2*time.Second, "wait time for writing messages")
	i := flag.Int("i", 5000, "wait milliseconds interval between each sends in client, randomized")
	I := flag.Int("I", 1000, "wait milliseconds interval between each connection, randomized")

	flag.Parse()

	runtime.GOMAXPROCS(runtime.NumCPU())

	_ports := strings.Split(*ports, ":")
	if len(_ports) != 2 {
		log.Fatalln("Invalid ports format, expecting <http port>:<device port>.")
	}
	httpPort := _ports[0]
	devicePort := _ports[1]

	log.Println("Login the octopus and get the auth token...")
	url := fmt.Sprintf("http://%s:%s/login", *host, httpPort)
	req := gorequest.New()
	response, bodyBytes, errs := req.Get(url).SetBasicAuth(*user, *passwd).EndBytes()
	if len(errs) > 0 {
		log.Fatalln(errs[0].Error())
	}
	if response.StatusCode != 200 {
		log.Fatalln("Unable to authenticate to Octopus. Please check the user/passwd pair.")
	}
	var loggedIn map[string]string
	err := json.Unmarshal(bodyBytes, &loggedIn)
	if err != nil {
		log.Fatalln("Unable to get auth response")
	}
	auth := loggedIn["auth_token"]
	if len(auth) > 0 {
		log.Println("Successfully logged in.")
	} else {
		log.Fatalln("Unable to get auth token, please check the server log.")
	}

	log.Println("Creating a channel for testing...")
	chanName := fmt.Sprintf("bench_channel_%d", time.Now().Unix())
	token := "123456789"
	url = fmt.Sprintf("http://%s:%s/channels", *host, httpPort)
	fieldDefs := strings.Split(*fields, ",")
	fieldMap := make(map[string]string)
	for _, def := range fieldDefs {
		pair := strings.Split(def, ":")
		fieldMap[pair[0]] = pair[1]
	}
	reqbody := map[string]interface{}{
		"name":          chanName,
		"description":   "bench test channel",
		"fields":        fieldMap,
		"access_tokens": []string{token},
	}
	asBytes, err := json.Marshal(reqbody)
	if err != nil {
		log.Fatalln(err.Error())
	}

	req = gorequest.New()
	response, bodyBytes, errs = req.Post(url).Set("AuthToken", auth).
		Send(string(asBytes)).EndBytes()
	if len(errs) > 0 {
		log.Fatalln(errs[0].Error())
	}
	if response.StatusCode != 201 {
		log.Fatalln("Unable to create test channel. Please check server log.")
	}

	var created map[string]string
	err = json.Unmarshal(bodyBytes, &created)
	if err != nil {
		log.Fatalln("Unable to get channel creation response")
	}
	chId := created["id"]
	if len(chId) > 0 {
		log.Println("Successfully created channel.")
	} else {
		log.Fatalln("Unable to get created channel Id. Please check server log.")
	}

	log.Println("Starting clients...")
	clients := make([]*WsClient, *c)
	var wg sync.WaitGroup
	wg.Add(*c)

	for _i := 0; _i < *c; _i++ {
		time.Sleep(time.Duration(rand.Intn(*I)) * time.Millisecond)
		go func(idx int) {
			defer wg.Done()
			c := &WsClient{
				Server:      fmt.Sprintf("%s:%s", *host, devicePort),
				ChannelId:   chId,
				DeviceId:    fmt.Sprintf("device-%d", idx),
				AccessToken: token,
				NPing:       *p,<commit_msg>fixed a bug in benchmark.go<commit_after>package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"github.com/vivowares/octopus/Godeps/_workspace/src/github.com/gorilla/websocket"
	"github.com/vivowares/octopus/Godeps/_workspace/src/github.com/parnurzeal/gorequest"
	"github.com/vivowares/octopus/Godeps/_workspace/src/github.com/satori/go.uuid"
	. "github.com/vivowares/octopus/utils"
	"log"
	"math/rand"
	"net/http"
	"net/url"
	"runtime"
	"strings"
	"sync"
	"time"
)

// ulimit -n 1048576; go run tasks/benchmark.go  -host=<host> -ports=8080:8081 -user=root -passwd=<PASSWORD> -fields=temperature:float -c=20000 -p=5 -m=5 -r=300s -w=10s -i=20000 -I=3 > bench.log 2>&1 &

func main() {
	host := flag.String("host", "localhost", "the target server host")
	ports := flag.String("ports", "8080:8081", "the http port and device port")
	fields := flag.String("fields", "temperature:float", "fields that are used for bench test. Format: 'field1:type1,field2:type2'")
	user := flag.String("user", "root", "username for authenticating octopus")
	passwd := flag.String("passwd", "<PASSWORD>", "passwd for authenticating octopus")

	c := flag.Int("c", 1000, "number of concurrent clients")
	p := flag.Int("p", 100, "number of ping messages to send")
	m := flag.Int("m", 50, "number of payload messages to send")
	r := flag.Duration("r", 15*time.Second, "wait time for reading messages")
	w := flag.Duration("w", 2*time.Second, "wait time for writing messages")
	i := flag.Int("i", 5000, "wait milliseconds interval between each sends in client, randomized")
	I := flag.Int("I", 1000, "wait milliseconds interval between each connection, randomized")

	flag.Parse()

	runtime.GOMAXPROCS(runtime.NumCPU())

	_ports := strings.Split(*ports, ":")
	if len(_ports) != 2 {
		log.Fatalln("Invalid ports format, expecting <http port>:<device port>.")
	}
	httpPort := _ports[0]
	devicePort := _ports[1]

	log.Println("Login the octopus and get the auth token...")
	url := fmt.Sprintf("http://%s:%s/login", *host, httpPort)
	req := gorequest.New()
	response, bodyBytes, errs := req.Get(url).SetBasicAuth(*user, *passwd).EndBytes()
	if len(errs) > 0 {
		log.Fatalln(errs[0].Error())
	}
	if response.StatusCode != 200 {
		log.Fatalln("Unable to authenticate to Octopus. Please check the user/passwd pair.")
	}
	var loggedIn map[string]string
	err := json.Unmarshal(bodyBytes, &loggedIn)
	if err != nil {
		log.Fatalln("Unable to get auth response")
	}
	auth := loggedIn["auth_token"]
	if len(auth) > 0 {
		log.Println("Successfully logged in.")
	} else {
		log.Fatalln("Unable to get auth token, please check the server log.")
	}

	log.Println("Creating a channel for testing...")
	chanName := fmt.Sprintf("bench_channel_%d", time.Now().UnixNano())
	token := "123456789"
	url = fmt.Sprintf("http://%s:%s/channels", *host, httpPort)
	fieldDefs := strings.Split(*fields, ",")
	fieldMap := make(map[string]string)
	for _, def := range fieldDefs {
		pair := strings.Split(def, ":")
		fieldMap[pair[0]] = pair[1]
	}
	reqbody := map[string]interface{}{
		"name":          chanName,
		"description":   "bench test channel",
		"fields":        fieldMap,
		"access_tokens": []string{token},
	}
	asBytes, err := json.Marshal(reqbody)
	if err != nil {
		log.Fatalln(err.Error())
	}

	req = gorequest.New()
	response, bodyBytes, errs = req.Post(url).Set("AuthToken", auth).
		Send(string(asBytes)).EndBytes()
	if len(errs) > 0 {
		log.Fatalln(errs[0].Error())
	}
	if response.StatusCode != 201 {
		log.Fatalln("Unable to create test channel. Please check server log.")
	}

	var created map[string]string
	err = json.Unmarshal(bodyBytes, &created)
	if err != nil {
		log.Fatalln("Unable to get channel creation response")
	}
	chId := created["id"]
	if len(chId) > 0 {
		log.Println("Successfully created channel.")
	} else {
		log.Fatalln("Unable to get created channel Id. Please check server log.")
	}

	log.Println("Starting clients...")
	clients := make([]*WsClient, *c)
	var wg sync.WaitGroup
	wg.Add(*c)

	for _i := 0; _i < *c; _i++ {
		time.Sleep(time.Duration(rand.Intn(*I)) * time.Millisecond)
		go func(idx int) {
			defer wg.Done()
			c := &WsClient{
				Server:      fmt.Sprintf("%s:%s", *host, devicePort),
				ChannelId:   chId,
				DeviceId:    fmt.Sprintf("device-%d-%d", idx, time.Now().UnixNano()),
				AccessToken: token,
				NPing:       *p,
{'fraction_non_alphanumeric': 0.17283372365339578, 'fraction_numerical': 0.020140515222482436, 'mean_word_length': 4.2852722772277225, 'pattern_counts': {'":': 12, '<': 14, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6111731', 'n_tokens_mistral': 3316, 'n_tokens_neox': 3045, 'n_words': 739}