go-socket.io stress_test

stress_test.go

package socketio

import (
	"testing"
	"os"
	"flag"
	"strconv"
	"sync"
	"fmt"
	"http"
	"time"
)

const (
	SERVER_ADDR  = "localhost"
	PORT         = "9999"
	NUM_MESSAGES = 10000
)

func startServer(t *testing.T) {

	config := DefaultConfig
	config.QueueLength = 1000
	config.ReconnectTimeout = 1e6
	config.Origins = []string{"*"}

	server := NewSocketIO(&config)

	server.OnMessage(func(c *Conn, msg Message) {
		if err := c.Send(msg.Data()); err != nil {
			t.Logf("server echo send error: ", err)
		}
	})

	go func() {
		http.ListenAndServe(fmt.Sprintf("%s:%s", SERVER_ADDR, PORT), server.ServeMux())
	}()

}

func TestStressTest(t *testing.T) {

	clients := 1
	passes := 1
	msg_size := 150 // bytes
	numMessages := NUM_MESSAGES
	serverAddr := fmt.Sprintf("%s:%s", SERVER_ADDR, PORT)

	flag.Parse()
	if len(flag.Args()) > 0 {
		serverAddr = flag.Arg(0)
	}

	if v, err := strconv.Atoi(flag.Arg(1)); err == nil {
		passes = v
	}

	if v, err := strconv.Atoi(flag.Arg(2)); err == nil {
		clients = v
	}

	if v, err := strconv.Atoi(flag.Arg(3)); err == nil {
		if v > msg_size {
			msg_size = v
		}
	}

	if clients > 1 {
		t.Logf("\nTest starting with %d parallel clients...", clients)
	}

	startServer(t)

	time.Sleep(1e9)

	//msg := bytes.Repeat([]byte("X"), msg_size)

	for iters := 0; iters < passes; iters++ {
		t.Log("Starting pass", iters)
		clientDisconnect := new(sync.WaitGroup)
		clientDisconnect.Add(clients)

		for i := 0; i < clients; i++ {

			go func() {
				clientMessage := make(chan Message)

				client := NewWebsocketClient(SIOCodec{})
				client.OnMessage(func(msg Message) {
					clientMessage <- msg
				})
				client.OnDisconnect(func() {
					clientDisconnect.Done()
				})

				err := client.Dial("ws://"+serverAddr+"/socket.io/websocket", "http://"+serverAddr+"/")
				if err != nil {
					t.Fatal(err)
				}

				iook := make(chan bool)

				go func() {
					t.Logf("Sending %d messages of size %v bytes...", numMessages, msg_size)
					var err os.Error

					for i := 0; i < numMessages; i++ {
						if err = client.Send(make([]byte, msg_size)); err != nil {
							t.Fatal("Send ERROR:", err)
						}
					}

					iook <- true
				}()

				go func() {
					t.Log("Receiving messages...")
					for i := 0; i < numMessages; i++ {
						<-clientMessage
					}
					iook <- true
				}()

				for i := 0; i < 2; i++ {
					<-iook
				}

				go func() {
					if err = client.Close(); err != nil {
						t.Fatal("Close ERROR:", err)
					}
				}()
			}()
		}

		t.Log("Waiting for clients disconnect")
		clientDisconnect.Wait()

		fmt.Printf("Sent %v messages * %v concurrent clients = %v messages\n", numMessages, clients, numMessages*clients)

		time.Sleep(1e9)
	}

	time.Sleep(10e9)
}

Thanks!

I added some debugging into your test program and into go-socket.io client code and got the following
interesting sequence:

2011/08/27 11:47:52 client reader: bufio: buffer full via vIXNtcZcIUdDlO8z
2011/08/27 11:47:52 OnDisconnect from vIXNtcZcIUdDlO8z
2011/08/27 11:47:52 sio/conn: lost connection: vIXNtcZcIUdDlO8z[websocket]
2011/08/27 11:47:52 Send ERROR: write tcp 127.0.0.1:9999: broken pipe to vIXNtcZcIUdDlO8z

So what's happening here?

1. In reader() function at go-socket.io/client.go the wc.ws.Read() returns bufio: buffer full error.
2. socketio.WebsocketClient then invokes its' OnDisconnect callback and closes the underlaying socket.
3. The test program OnDisconnect callback is called.
4. The test program continues to write to the socket causing the broken pipe error, since the socket has been
   closed.

Why is this happening?
This is caused by a known bug in the websocket package: http://code.google.com/p/go/issues/detail?id=2134.
The server combines multiple response messages into single fragments of data that are sent simultaneously on the wire back towards the client. When the combined size of such a data fragment exceeds the size of the websocket packages bufio buffer size, which is by default 4096 bytes, the client can't read the packet and fails.
When I replaced the relevant code in the Go's websocket/client.go, the stress test went through fine without any
errors:

 func newClient(resourceName, host, origin, location, protocol string, rwc io.ReadWriteCloser, handshake handshaker) (ws *Conn, err os.Error) {
-       br := bufio.NewReader(rwc)
+       br, _ := bufio.NewReaderSize(rwc, 65536)

This problem is probably unrelated to the handshake packet being dropped due to channel saturation, but your test
gave us valuable information since we now known what is causing this kind of hiccups.

The new Go websocket package has been under work for some time already - I'm confident that it will be finalized soon,
and that it will make these problems go away.

JP

It seems that the new websocket package has landed in the Go tip yesterday.
I made the necessary changes to go-socket.io dev branch.

This problems seems to be solved now.

Oh cool! Another thing I discovered is that a previous recent release of new websocket code didnt work in Chrome beta. Apparently the draft was still changing. Someone had been working on the patches so I wonder if this latest release addresses that completely and it will now work in chrome beta. On Aug 27, 2011 2:55 AM, "madari" < reply@reply.github.com> wrote:

…

It seems that the new websocket package has landed in the Go tip yesterday. I made the necessary changes to go-socket.io dev branch. This problems seems to be solved now. ## Reply to this email directly or view it on GitHub: https://gist.github.com/1174699

Did you say the change was in the latest Go weekly? I'm using 6g version weekly.2011-08-17 9454, and I don't see that change in the websocket/client.go

No, it's in the tip. http://code.google.com/p/go/source/detail?r=88744261dd47fb1649898d281c17f5ff987e02df JP

…

On Aug 29, 2011, at 1:40 AM, justinfx wrote: Did you say the change was in the latest Go weekly? I'm using 6g version weekly.2011-08-17 9454, and I don't see that change in the websocket/client.go ## Reply to this email directly or view it on GitHub: https://gist.github.com/1174699