Unix conn (#4)

* factor out serve helper and readSender interface

* factor out workerLoop

* added unix socket benchmark, test coverage for unix sockets

Author: briansorahan

lolatency_test.go

@@ -15,7 +15,7 @@ import (
 //
 // Update 3/11/2017 [briansorahan]
 // This benchmarks around 100us on my Dell Latitude E6510 with a single-core Core i7 @ 2.8GHz
-func BenchmarkMessageSend(b *testing.B) {
+func BenchmarkUDPSend(b *testing.B) {
 	laddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
 	if err != nil {
 		b.Fatal(err)
@@ -53,7 +53,7 @@ func BenchmarkMessageSend(b *testing.B) {
 }
 
 // Including a single argument does not seem to have much effect on latency [briansorahan].
-func BenchmarkMessageSendOneArgument(b *testing.B) {
+func BenchmarkUDPSendOneArgument(b *testing.B) {
 	laddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
 	if err != nil {
 		b.Fatal(err)
@@ -93,3 +93,42 @@ func BenchmarkMessageSendOneArgument(b *testing.B) {
 		<-ch
 	}
 }
+
+func BenchmarkUnixSend(b *testing.B) {
+	const network = "unixgram"
+
+	laddr, err := net.ResolveUnixAddr(network, osc.TempSocket())
+	if err != nil {
+		b.Fatal(err)
+	}
+	srv, err := osc.ListenUnix("unixgram", laddr)
+	if err != nil {
+		b.Fatal(err)
+	}
+	raddr, err := net.ResolveUnixAddr(network, srv.LocalAddr().String())
+	if err != nil {
+		b.Fatal(err)
+	}
+	conn, err := osc.DialUnix(network, nil, raddr)
+	if err != nil {
+		b.Fatal(err)
+	}
+	var (
+		ch  = make(chan struct{})
+		val = struct{}{}
+	)
+	go srv.Serve(8, osc.Dispatcher{
+		"/ping": osc.Method(func(m osc.Message) error {
+			ch <- val
+			return nil
+		}),
+	})
+	msg := osc.Message{Address: "/ping"}
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		conn.Send(msg)
+		<-ch
+	}
+}

osc.go

@@ -2,7 +2,12 @@ package osc
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
+	"net"
+	"strings"
+
+	"github.com/pkg/errors"
 )
 
 const (
@@ -91,3 +96,87 @@ func ReadBlob(length int32, data []byte) ([]byte, int64) {
 	}
 	return data[:idx], int64(idx)
 }
+
+// Incoming represents incoming data.
+type Incoming struct {
+	Data   []byte
+	Sender net.Addr
+}
+
+type netWriter interface {
+	SetWriteBuffer(bytes int) error
+	WriteTo([]byte, net.Addr) (int, error)
+}
+
+var errClosed = errors.New("conn is closed")
+
+func checkDispatcher(dispatcher Dispatcher) error {
+	if dispatcher == nil {
+		return ErrNilDispatcher
+	}
+	for addr := range dispatcher {
+		if err := ValidateAddress(addr); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// readSender knows how to read bytes and return the net.Addr
+// of the sender of the bytes.
+type readSender interface {
+	CloseChan() <-chan struct{}
+	Context() context.Context
+	read([]byte) (int, net.Addr, error)
+}
+
+func serve(r readSender, numWorkers int, dispatcher Dispatcher) error {
+	if err := checkDispatcher(dispatcher); err != nil {
+		return err
+	}
+	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 workerLoop(r, ready, errChan)
+
+	// If the connection is closed or the context is canceled then stop serving.
+	select {
+	case err := <-errChan:
+		return errors.Wrap(err, "error serving udp")
+	case <-r.CloseChan():
+	case <-r.Context().Done():
+		return r.Context().Err()
+	}
+	return nil
+}
+
+func workerLoop(r readSender, ready chan Worker, errChan chan error) {
+	for {
+		data := make([]byte, bufSize)
+		_, sender, err := r.read(data)
+		if err != nil {
+			// 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
+			}
+			errChan <- err
+			return
+		}
+
+		// Get the next worker.
+		worker := <-ready
+
+		// Assign them the data we just read.
+		worker.DataChan <- Incoming{Data: data, Sender: sender}
+	}
+}

udp.go

@@ -3,23 +3,22 @@ package osc
 import (
 	"context"
 	"net"
-	"strings"
 
 	"github.com/pkg/errors"
 )
 
 // udpConn includes exactly the methods we need from *net.UDPConn
 type udpConn interface {
 	net.Conn
+	netWriter
 
 	ReadFromUDP([]byte) (int, *net.UDPAddr, error)
-	SetWriteBuffer(bytes int) error
-	WriteTo([]byte, net.Addr) (int, error)
 }
 
 // UDPConn is an OSC connection over UDP.
 type UDPConn struct {
 	udpConn
+
 	closeChan chan struct{}
 	ctx       context.Context
 	errChan   chan error
@@ -65,17 +64,34 @@ func ListenUDPContext(ctx context.Context, network string, laddr *net.UDPAddr) (
 	return uc.initialize()
 }
 
+// Close closes the udp conn.
+func (conn *UDPConn) Close() error {
+	close(conn.closeChan)
+	return conn.udpConn.Close()
+}
+
+// CloseChan returns a channel that is closed when the connection gets closed.
+func (conn *UDPConn) CloseChan() <-chan struct{} {
+	return conn.closeChan
+}
+
+// Context returns the context associated with the conn.
+func (conn *UDPConn) Context() context.Context {
+	return conn.ctx
+}
+
 // initialize initializes a UDP connection.
 func (conn *UDPConn) initialize() (*UDPConn, error) {
 	if err := conn.udpConn.SetWriteBuffer(bufSize); err != nil {
 		return nil, errors.Wrap(err, "setting write buffer size")
 	}
 	return conn, nil
+
 }
 
-// Context returns the context associated with the conn.
-func (conn *UDPConn) Context() context.Context {
-	return conn.ctx
+// read reads bytes and returns the net.Addr of the sender.
+func (conn *UDPConn) read(data []byte) (int, net.Addr, error) {
+	return conn.ReadFromUDP(data)
 }
 
 // Send sends an OSC message over UDP.
@@ -95,79 +111,10 @@ func (conn *UDPConn) SendTo(addr net.Addr, p Packet) error {
 // 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(numWorkers int, dispatcher Dispatcher) error {
-	if dispatcher == nil {
-		return ErrNilDispatcher
-	}
-	for addr := range dispatcher {
-		if err := ValidateAddress(addr); err != nil {
-			return err
-		}
-	}
-	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 {
-			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:
-		return errors.Wrap(err, "error serving udp")
-	case <-conn.closeChan:
-	case <-conn.ctx.Done():
-		return conn.ctx.Err()
-	}
-	return nil
-}
-
-// serve retrieves OSC packets.
-func (conn *UDPConn) serve(ready <-chan Worker) error {
-	data := make([]byte, bufSize)
-	_, sender, err := conn.ReadFromUDP(data)
-	if err != nil {
-		// 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
+	return serve(conn, numWorkers, dispatcher)
 }
 
 // SetContext sets the context associated with the conn.
 func (conn *UDPConn) SetContext(ctx context.Context) {
 	conn.ctx = ctx
 }
-
-// Close closes the udp conn.
-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_test.go

@@ -37,7 +37,7 @@ func TestDialUDP(t *testing.T) {
 }
 
 func TestDialUDPSetWriteBufferError(t *testing.T) {
-	uc := &UDPConn{udpConn: errConn{}}
+	uc := &UDPConn{udpConn: errUDPConn{}}
 	_, err := uc.initialize()
 	if err == nil {
 		t.Fatal("expected error, got nil")
@@ -101,7 +101,10 @@ func testUDPServer(t *testing.T, dispatcher Dispatcher) (*UDPConn, *UDPConn, cha
 	errChan := make(chan error)
 
 	go func() {
-		errChan <- server.Serve(1, dispatcher)
+		if err := server.Serve(1, dispatcher); err != nil {
+			errChan <- err
+		}
+		close(errChan)
 	}()
 
 	raddr, err := net.ResolveUDPAddr("udp", server.LocalAddr().String())
@@ -125,16 +128,16 @@ func TestUDPConnSend_OK(t *testing.T) {
 	}
 }
 
-// errConn is an implementation of the udpConn interface that returns errors from all it's methods.
-type errConn struct {
+// errUDPConn is an implementation of the udpConn interface that returns errors from all it's methods.
+type errUDPConn struct {
 	udpConn
 }
 
-func (e errConn) ReadFromUDP(b []byte) (int, *net.UDPAddr, error) {
+func (e errUDPConn) ReadFromUDP(b []byte) (int, *net.UDPAddr, error) {
 	return 0, nil, errors.New("oops")
 }
 
-func (e errConn) SetWriteBuffer(bytes int) error {
+func (e errUDPConn) SetWriteBuffer(bytes int) error {
 	return errors.New("derp")
 }
 
@@ -176,7 +179,7 @@ func TestUDPConnServe_ReadError(t *testing.T) {
 		t.Fatal(err)
 	}
 	server := &UDPConn{
-		udpConn: errConn{udpConn: serverConn},
+		udpConn: errUDPConn{udpConn: serverConn},
 		ctx:     context.Background(),
 	}
 	go func() {