Updated benchmarks.

examples/bench/bench.go

@@ -1,103 +1,97 @@
-// Quick and dirty SockJS benchmark
-
 package main
 
 import (
 	"websocket"
+	"runtime"
+	"flag"
+	"rand"
 	"fmt"
 	"time"
-	"strconv"
-	"strings"
+	"syscall"
 	"json"
-	"rand"
-	"flag"
-	"runtime"
-	"math"
+	"strings"
+	"strconv"
 )
 
+// SockJS client
 type Stats struct
 {
+	sent int
+	recv int
+
 	start_time int64
 	end_time int64
 
-	client_id int
-
-	avg_ping int64
 	min_ping int64
 	max_ping int64
-
-	sent int
-	recv int
-
-	my_packets int
 }
 
-func NewStats() Stats {
-	n := Stats{}
-	n.sent = 0
-	n.recv = 0
+func newStats() *Stats {
+	n := new(Stats)
 	n.min_ping = ^int64(0)
 	return n
 }
 
-// Warmup time in seconds
-const WARMUP_TIME int64 = 1
-// Test time in seconds
-const TEST_TIME int64 = 10
+func clientReceiver(ch chan string, ws *websocket.Conn) {
+	var b = make([]byte, 512)
 
-func SockJSClient(client_id int, cs chan *Stats) {
-	// Error handler
-	defer func() {
-		if r := recover(); r != nil {
-			cs <- nil
+	for {
+		n, err := ws.Read(b);
+
+		if err != nil {
+			ch <- ""
+			break
 		}
-	}()
+
+		ch <- string(b[0:n])
+	}
+}
+
+func clientSender(client_id int, ch chan bool, result chan *Stats) {
 
 	ws, err := websocket.Dial("ws://localhost:8080/echo/0/0/websocket", "", "http://localhost/");
 	if err != nil {
 		panic("Dial: " + err.String())
 	}
 
-	var msg = make([]byte, 512)
-	n, err := ws.Read(msg)
-	if err != nil {
-		panic("Read: " + err.String())
-	}
+	defer func() {
+		if ws != nil {
+			ws.Close()
+		}
 
-	handshake := string(msg[0:n])
-	if handshake != "o" {
-		panic("Invalid SockJS handshake: " + handshake)
-	}
+		if r := recover(); r != nil {
+			result <- nil
+		}
+	}()
 
-	stats := NewStats()
-	start_time := time.Nanoseconds()
-	end_time := start_time + TEST_TIME * 1000000000
-	stats.start_time = start_time
+	reader := make(chan string)
 
-	for ; time.Nanoseconds() < end_time; {
-		val := fmt.Sprintf("[\"%d,%d\"]", client_id, time.Nanoseconds())
-		n, err = ws.Write([]byte(val))
-		if n != len(val) || err != nil {
-			panic("Send failed!")
-		}
+	go clientReceiver(reader, ws)
+
+	if handshake := <-reader; handshake != "o" {
+		panic("Invalid handshake!")
+	}
+
+	need_finish := false
+	num_pending := 0
 
-		stats.sent += 1
+	stats := newStats()
+	stats.start_time = time.Nanoseconds()
 
-		for {
-			n, err = ws.Read(msg)
-			if err != nil {
-				panic("Read failed: " + err.String())
+	for ; !need_finish || num_pending > 0; {
+		select {
+		case msg := <-reader:
+			if msg == "" {
+				break
 			}
 
-			data := string(msg[0:n])
-			if data[0] == 'a' {
-				dec := json.NewDecoder(strings.NewReader(data[1:]))
+			if msg[0] == 'a' {
+				dec := json.NewDecoder(strings.NewReader(msg[1:]))
 
 				var v []string
 
 				if err := dec.Decode(&v); err != nil {
 					panic("JSON decode failed: " + err.String())
-					return
 				}
 
 				stats.recv += 1
@@ -110,17 +104,16 @@ func SockJSClient(client_id int, cs chan *Stats) {
 					cid, _ := strconv.Atoi(parts[0])
 					if client_id == cid {
 						stamp, _ := strconv.Atoi64(parts[1])
-						delta := time.Nanoseconds() - stamp
 
-						if (stats.max_ping < delta) {
-							stats.max_ping = delta
+						dt := time.Nanoseconds() - stamp
+
+						if stats.min_ping > dt {
+							stats.min_ping = dt
 						}
-						if (stats.min_ping > delta) {
-							stats.min_ping = delta
+						if stats.max_ping < dt {
+							stats.max_ping = dt
 						}
-
-						stats.avg_ping = (stats.avg_ping + delta) / 2
-						stats.my_packets += 1
+						num_pending -= 1
 
 						got_response = true
 					}
@@ -130,63 +123,120 @@ func SockJSClient(client_id int, cs chan *Stats) {
 					break
 				}
 			} else
-			if data[0] == 'c' {
-				return
+			if msg[0] == 'c' {
+				panic("Got disconnect packet!")
+			}
+		case evt := <-ch:
+			if evt {
+				stamp := time.Nanoseconds()
+
+				msg := fmt.Sprintf("[\"%d,%d\"]", client_id, stamp)
+				ws.Write([]byte(msg))
+
+				stats.sent += 1
+
+				num_pending += 1
+			} else
+			{
+				need_finish = true
 			}
 		}
 	}
 
-	// Output statistics
 	stats.end_time = time.Nanoseconds()
-	cs <- &stats
+
+	result <- stats
 }
 
 var numCores = flag.Int("n", 1, "num of CPU cores to use")
-//var numClients = flag.Int("c", 1, "num of clients")
+var msgPerSec = flag.Int64("m", 2000, "number of messages per second")
+var msgTotal = flag.Int("t", 10000, "number of messages to send")
 
 // Entry point
 func main() {
-	flag.Parse()
-
 	rand.Seed(time.Nanoseconds())
 
+	flag.Parse()
 	runtime.GOMAXPROCS(*numCores)
 
 	// Number of clients to add for each ramp
-	ramps := [9]int{5,25,50,100,150,200,300,500,1000}
-	//ramps := [3]int{200,300,500}
+	ramps := []int{100,200,300,500,1000}
 
 	for i := 0; i < len(ramps); i++ {
 		num_clients := ramps[i]
 
-		cs := make(chan *Stats)
+		channels := make([]chan bool, num_clients)
+		stats := make(chan *Stats)
+
+		for j := 0; j < num_clients; j++ {
+			channels[j] = make(chan bool)
+			go clientSender(rand.Int(), channels[j], stats)
+		}
+
+		var msg_delay int64 = 1000000000 / *msgPerSec
+		var time_slot int64 = msg_delay
+
+		for j := 0; j < *msgTotal; j++ {
+			start := time.Nanoseconds()
+
+			channels[j % num_clients] <- true
+
+			delta_ts := time.Nanoseconds()
+			time_slot -= (delta_ts - start)
+			if time_slot > 0 {
+				syscall.Sleep(time_slot)
+			}
+
+			delta_end := time.Nanoseconds() - delta_ts
+			time_slot += (msg_delay - delta_end)
+		}
+
 		for j := 0; j < num_clients; j++ {
-			go SockJSClient(rand.Int(), cs)
+			channels[j] <- false
 		}
 
 		// Collect stats
 		var total_sent float64 = 0
 		var total_recv float64 = 0
+		var min_ping int64 = ^int64(0)
+		var max_ping int64 = 0
+		var avg_ping float64 = 0
+
 		errors := 0
 
 		for j := 0; j < num_clients; j++ {
-			stats := <-cs
+			data := <-stats
 
-			if stats != nil {
-				seconds := float64(stats.end_time - stats.start_time) / 1000000000
+			if data != nil {
+				seconds := float64(data.end_time - data.start_time) / 1000000000
 
-				total_sent += float64(stats.sent) / seconds
-				total_recv += float64(stats.recv) / seconds
+				total_sent += float64(data.sent) / seconds
+				total_recv += float64(data.recv) / seconds
 
-				if math.IsNaN(total_sent) {
-					fmt.Printf("--> %d, %d, %f\n", stats.sent, stats.recv, seconds)
+				if min_ping > data.min_ping {
+					min_ping = data.min_ping
 				}
+				if max_ping < data.max_ping {
+					max_ping = data.max_ping
+				}
+
+				avg_ping += float64((max_ping - min_ping) / 1000000) / seconds
 			} else
 			{
 				errors += 1
 			}
 		}
 
-		fmt.Printf("clients: %d, sent: %f, recv: %f, errors: %d\n", num_clients, total_sent, total_recv, errors)
+		flt_clients := float64(num_clients)
+		avg_ping /= flt_clients
+
+		fmt.Printf("clients: %d, sent: %f, recv: %f, min_ping: %d, max_ping: %d, avg_ping: %f, errors: %d\n",
+					num_clients,
+					total_sent,
+					total_recv,
+					min_ping / 1000000,
+					max_ping / 1000000,
+					avg_ping,
+					errors)
 	}
 }