Workers (#3)

* use worker pool to serve

* workers have made the benchmark slower somehow

* avoid returning an error when reading from a closed connection

Author: briansorahan

conn.go

@@ -24,7 +24,7 @@ type Conn interface {
 	net.Conn
 
 	Context() context.Context
-	Serve(Dispatcher) error
+	Serve(int, Dispatcher) error
 	Send(Packet) error
 	SendTo(net.Addr, Packet) error
 }

lolatency_test.go

@@ -36,7 +36,7 @@ func BenchmarkMessageSend(b *testing.B) {
 		ch  = make(chan struct{})
 		val = struct{}{}
 	)
-	go srv.Serve(osc.Dispatcher{
+	go srv.Serve(8, osc.Dispatcher{
 		"/ping": osc.Method(func(m osc.Message) error {
 			ch <- val
 			return nil
@@ -74,7 +74,7 @@ func BenchmarkMessageSendOneArgument(b *testing.B) {
 		ch  = make(chan struct{})
 		val = struct{}{}
 	)
-	go srv.Serve(osc.Dispatcher{
+	go srv.Serve(1, osc.Dispatcher{
 		"/ping": osc.Method(func(m osc.Message) error {
 			if _, err := m.Arguments[0].ReadInt32(); err != nil {
 				return err

udp.go

@@ -3,6 +3,7 @@ package osc
 import (
 	"context"
 	"net"
+	"strings"
 
 	"github.com/pkg/errors"
 )
@@ -93,25 +94,37 @@ func (conn *UDPConn) SendTo(addr net.Addr, p Packet) error {
 // Any errors returned from a dispatched method will be returned.
 // Note that this means that errors returned from a dispatcher method will kill your server.
 // If context.Canceled or context.DeadlineExceeded are encountered they will be returned directly.
-func (conn *UDPConn) Serve(dispatcher Dispatcher) error {
+func (conn *UDPConn) Serve(numWorkers int, dispatcher Dispatcher) error {
 	if dispatcher == nil {
 		return ErrNilDispatcher
 	}
-
 	for addr := range dispatcher {
 		if err := ValidateAddress(addr); err != nil {
 			return err
 		}
 	}
-
-	errChan := make(chan error)
-
+	var (
+		errChan = make(chan error)
+		ready   = make(chan Worker, numWorkers)
+	)
+	for i := 0; i < numWorkers; i++ {
+		go Worker{
+			DataChan:   make(chan Incoming),
+			Dispatcher: dispatcher,
+			ErrChan:    errChan,
+			Ready:      ready,
+		}.Run()
+	}
 	go func() {
 		for {
-			conn.serve(dispatcher, errChan)
+			if err := conn.serve(ready); err != nil {
+				if err == errClosed {
+					return
+				}
+				errChan <- err
+			}
 		}
 	}()
-
 	// If the connection is closed or the context is canceled then stop serving.
 	select {
 	case err := <-errChan:
@@ -124,38 +137,20 @@ func (conn *UDPConn) Serve(dispatcher Dispatcher) error {
 }
 
 // serve retrieves OSC packets.
-func (conn *UDPConn) serve(dispatcher Dispatcher, errChan chan error) {
+func (conn *UDPConn) serve(ready <-chan Worker) error {
 	data := make([]byte, bufSize)
-
 	_, sender, err := conn.ReadFromUDP(data)
 	if err != nil {
-		errChan <- err
-	}
-
-	switch data[0] {
-	case BundleTag[0]:
-		go func() {
-			bundle, err := ParseBundle(data, sender)
-			if err != nil {
-				errChan <- err
-			}
-			if err := dispatcher.Dispatch(bundle); err != nil {
-				errChan <- errors.Wrap(err, "dispatch bundle")
-			}
-		}()
-	case MessageChar:
-		go func() {
-			msg, err := ParseMessage(data, sender)
-			if err != nil {
-				errChan <- err
-			}
-			if err := dispatcher.Invoke(msg); err != nil {
-				errChan <- errors.Wrap(err, "dispatch message")
-			}
-		}()
-	default:
-		errChan <- ErrParse
+		// Tried non-blocking select on closeChan right before ReadFromUDP
+		// but that didn't stop us from reading a closed connection. [briansorahan]
+		if strings.Contains(err.Error(), "use of closed network connection") {
+			return errClosed
+		}
+		return err
 	}
+	worker := <-ready
+	worker.DataChan <- Incoming{Data: data, Sender: sender}
+	return nil
 }
 
 // SetContext sets the context associated with the conn.
@@ -168,3 +163,11 @@ func (conn *UDPConn) Close() error {
 	close(conn.closeChan)
 	return conn.udpConn.Close()
 }
+
+// Incoming represents incoming data.
+type Incoming struct {
+	Data   []byte
+	Sender net.Addr
+}
+
+var errClosed = errors.New("conn is closed")

udp_example_test.go

@@ -66,7 +66,7 @@ func ExampleUDPConn_pingpong() {
 }
 
 func serverDispatch(server *UDPConn, errChan chan error) {
-	errChan <- server.Serve(Dispatcher{
+	errChan <- server.Serve(1, Dispatcher{
 		"/ping": Method(func(msg Message) error {
 			fmt.Println("Server received ping.")
 			return server.SendTo(msg.Sender, Message{Address: "/pong"})
@@ -83,7 +83,7 @@ func serverDispatch(server *UDPConn, errChan chan error) {
 }
 
 func clientDispatch(client *UDPConn, errChan chan error, pongChan chan struct{}, closeChan chan struct{}) {
-	errChan <- client.Serve(Dispatcher{
+	errChan <- client.Serve(1, Dispatcher{
 		"/pong": Method(func(msg Message) error {
 			fmt.Println("Client received pong.")
 			close(pongChan)

udp_test.go

@@ -21,7 +21,7 @@ func TestInvalidAddress(t *testing.T) {
 	}
 	defer func() { _ = server.Close() }() // Best effort.
 
-	if err := server.Serve(Dispatcher{
+	if err := server.Serve(1, Dispatcher{
 		"/[": Method(func(msg Message) error {
 			return nil
 		}),
@@ -72,7 +72,7 @@ func TestDialUDPContext(t *testing.T) {
 	if c.Context() != ctxTimeout {
 		t.Fatalf("expected %+v to be %+v", ctxTimeout, c.Context())
 	}
-	if err := c.Serve(Dispatcher{}); err != context.DeadlineExceeded {
+	if err := c.Serve(1, Dispatcher{}); err != context.DeadlineExceeded {
 		t.Fatalf("expected context.DeadlineExceeded, got %+v", err)
 	}
 }
@@ -101,7 +101,7 @@ func testUDPServer(t *testing.T, dispatcher Dispatcher) (*UDPConn, *UDPConn, cha
 	errChan := make(chan error)
 
 	go func() {
-		errChan <- server.Serve(dispatcher)
+		errChan <- server.Serve(1, dispatcher)
 	}()
 
 	raddr, err := net.ResolveUDPAddr("udp", server.LocalAddr().String())
@@ -151,7 +151,7 @@ func TestUDPConnServe_ContextTimeout(t *testing.T) {
 	}
 	errChan := make(chan error)
 	go func() {
-		errChan <- server.Serve(Dispatcher{})
+		errChan <- server.Serve(1, Dispatcher{})
 	}()
 	select {
 	case <-time.After(200 * time.Millisecond):
@@ -180,7 +180,7 @@ func TestUDPConnServe_ReadError(t *testing.T) {
 		ctx:     context.Background(),
 	}
 	go func() {
-		errChan <- server.Serve(Dispatcher{
+		errChan <- server.Serve(1, Dispatcher{
 			"/close": Method(func(msg Message) error {
 				return server.Close()
 			}),
@@ -214,7 +214,7 @@ func TestUDPConnServe_NilDispatcher(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err := server.Serve(nil); err == nil {
+	if err := server.Serve(1, nil); err == nil {
 		t.Fatal("expected error, got nil")
 	}
 }
@@ -234,6 +234,8 @@ func TestUDPConnServe_BadInboundAddr(t *testing.T) {
 		if err := conn.Send(packet); err != nil {
 			t.Fatal(err)
 		}
+		t.Logf("sent message %s", string(packet.Bytes()))
+
 		if err := <-errChan; err == nil {
 			t.Fatalf("(packet %d) expected error, got nil", i)
 		}

worker.go

@@ -0,0 +1,43 @@
+package osc
+
+import (
+	"github.com/pkg/errors"
+)
+
+type Worker struct {
+	DataChan   chan Incoming
+	Dispatcher Dispatcher
+	ErrChan    chan error
+	Ready      chan<- Worker
+}
+
+func (w Worker) Run() {
+	w.Ready <- w
+
+	for incoming := range w.DataChan {
+		data := incoming.Data
+
+		switch data[0] {
+		case BundleTag[0]:
+			bundle, err := ParseBundle(data, incoming.Sender)
+			if err != nil {
+				w.ErrChan <- err
+			}
+			if err := w.Dispatcher.Dispatch(bundle); err != nil {
+				w.ErrChan <- errors.Wrap(err, "dispatch bundle")
+			}
+		case MessageChar:
+			msg, err := ParseMessage(data, incoming.Sender)
+			if err != nil {
+				w.ErrChan <- err
+			}
+			if err := w.Dispatcher.Invoke(msg); err != nil {
+				w.ErrChan <- errors.Wrap(err, "dispatch message")
+			}
+		default:
+			w.ErrChan <- ErrParse
+		}
+		// Announce the worker is ready again.
+		w.Ready <- w
+	}
+}