Major performance improvement in UDP processing by reusing buffer (#33)

* Connections as PacketConn slice, avoid runtime type assertions

* Add benchmarks for message processing for TCP and Datagram connections

The reason we can't simply use ListenUDP() and Dial() in the datagram
benchmark is that we wouldn't be able to ensure that all messages would
arrive on the server, even for a udp connection to localhost.
By using the fakePacketConn we ensure that all messages are delivered
making the datagram benchmark reliable.

* Major performance improvement in UDP processing by reusing buffers.

As we can see in the added benchmarks receiving messages from a UDP
connection used to be very slow, slower than using TCP which was a
little strange. By simply reusing the buffer used to receive datagrams
we can see a major improvement in processing time and memory
allocations.

benchmark                           old ns/op     new ns/op     delta
BenchmarkDatagramNoFormatting-4     16172         2668          -83.50%
BenchmarkTCPNoFormatting-4          7682          7644          -0.49%

benchmark                           old MB/s     new MB/s     speedup
BenchmarkDatagramNoFormatting-4     2.91         17.62        6.05x
BenchmarkTCPNoFormatting-4          6.12         6.15         1.00x

benchmark                           old allocs     new allocs     delta
BenchmarkDatagramNoFormatting-4     5              4
-20.00%
BenchmarkTCPNoFormatting-4          6              6              +0.00%

benchmark                           old bytes     new bytes     delta
BenchmarkDatagramNoFormatting-4     65905         368           -99.44%
BenchmarkTCPNoFormatting-4          464           464           +0.00%

Author: cezarsa

server.go

@@ -30,7 +30,7 @@ type TlsPeerNameFunc func(tlsConn *tls.Conn) (tlsPeer string, ok bool)
 
 type Server struct {
 	listeners               []net.Listener
-	connections             []net.Conn
+	connections             []net.PacketConn
 	wait                    sync.WaitGroup
 	doneTcp                 chan bool
 	datagramChannel         chan DatagramMessage
@@ -315,16 +315,12 @@ type DatagramMessage struct {
 	client  string
 }
 
-func (s *Server) goReceiveDatagrams(connection net.Conn) {
-	packetconn, ok := connection.(net.PacketConn)
-	if !ok {
-		panic("Connection is not a packet connection")
-	}
+func (s *Server) goReceiveDatagrams(packetconn net.PacketConn) {
 	s.wait.Add(1)
 	go func() {
 		defer s.wait.Done()
+		buf := make([]byte, 65536)
 		for {
-			buf := make([]byte, 65536)
 			n, addr, err := packetconn.ReadFrom(buf)
 			if err == nil {
 				// Ignore trailing control characters and NULs

server_bench_test.go

@@ -0,0 +1,105 @@
+package syslog
+
+import (
+	"bufio"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"gopkg.in/mcuadros/go-syslog.v2/format"
+)
+
+type noopFormatter struct{}
+
+func (noopFormatter) Parse() error {
+	return nil
+}
+
+func (noopFormatter) Dump() format.LogParts {
+	return format.LogParts{}
+}
+
+func (noopFormatter) Location(*time.Location) {}
+
+func (n noopFormatter) GetParser(l []byte) format.LogParser {
+	return n
+}
+
+func (n noopFormatter) GetSplitFunc() bufio.SplitFunc {
+	return nil
+}
+
+type handlerCounter struct {
+	expected int
+	current  int
+	done     chan struct{}
+}
+
+func (s *handlerCounter) Handle(logParts format.LogParts, msgLen int64, err error) {
+	s.current++
+	if s.current == s.expected {
+		close(s.done)
+	}
+}
+
+type fakePacketConn struct {
+	*io.PipeReader
+}
+
+func (c *fakePacketConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
+	n, err = c.PipeReader.Read(b)
+	return
+}
+func (c *fakePacketConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
+	return 0, nil
+}
+func (c *fakePacketConn) Close() error {
+	return nil
+}
+func (c *fakePacketConn) LocalAddr() net.Addr {
+	return nil
+}
+func (c *fakePacketConn) SetDeadline(t time.Time) error {
+	return nil
+}
+func (c *fakePacketConn) SetReadDeadline(t time.Time) error {
+	return nil
+}
+func (c *fakePacketConn) SetWriteDeadline(t time.Time) error {
+	return nil
+}
+
+func BenchmarkDatagramNoFormatting(b *testing.B) {
+	handler := &handlerCounter{expected: b.N, done: make(chan struct{})}
+	server := NewServer()
+	defer server.Kill()
+	server.SetFormat(noopFormatter{})
+	server.SetHandler(handler)
+	reader, writer := io.Pipe()
+	server.goReceiveDatagrams(&fakePacketConn{PipeReader: reader})
+	server.goParseDatagrams()
+	msg := []byte(exampleSyslog + "\n")
+	b.SetBytes(int64(len(msg)))
+	for i := 0; i < b.N; i++ {
+		writer.Write(msg)
+	}
+	<-handler.done
+}
+
+func BenchmarkTCPNoFormatting(b *testing.B) {
+	handler := &handlerCounter{expected: b.N, done: make(chan struct{})}
+	server := NewServer()
+	defer server.Kill()
+	server.SetFormat(noopFormatter{})
+	server.SetHandler(handler)
+	server.ListenTCP("127.0.0.1:0")
+	server.Boot()
+	conn, _ := net.DialTimeout("tcp", server.listeners[0].Addr().String(), time.Second)
+	msg := []byte(exampleSyslog + "\n")
+	b.SetBytes(int64(len(msg)))
+	for i := 0; i < b.N; i++ {
+		conn.Write(msg)
+	}
+	<-handler.done
+}