UDP service listener performance enhancements

CHANGELOG.md

@@ -29,6 +29,7 @@
 - [#4600](https://github.com/influxdb/influxdb/pull/4600): ping endpoint can wait for leader
 - [#4648](https://github.com/influxdb/influxdb/pull/4648): UDP Client (v2 client)
 - [#4690](https://github.com/influxdb/influxdb/pull/4690): SHOW SHARDS now includes database and policy. Thanks @pires
+- [#4676](https://github.com/influxdb/influxdb/pull/4676): UDP service listener performance enhancements
 
 ### Bugfixes
 - [#4643](https://github.com/influxdb/influxdb/pull/4643): Fix panic during backup restoration. Thanks @oiooj

services/udp/config.go

@@ -17,13 +17,26 @@ const (
 	DefaultRetentionPolicy = ""
 
 	// DefaultBatchSize is the default UDP batch size.
-	DefaultBatchSize = 1000
+	DefaultBatchSize = 5000
 
 	// DefaultBatchPending is the default number of pending UDP batches.
-	DefaultBatchPending = 5
+	DefaultBatchPending = 10
 
 	// DefaultBatchTimeout is the default UDP batch timeout.
 	DefaultBatchTimeout = time.Second
+
+	// DefaultReadBuffer is the default buffer size for the UDP listener.
+	// Sets the size of the operating system's receive buffer associated with
+	// the UDP traffic. Keep in mind that the OS must be able
+	// to handle the number set here or the UDP listener will error and exit.
+	//
+	// DefaultReadBuffer = 0 means to use the OS default, which is usually too
+	// small for high UDP performance.
+	//
+	// Increasing OS buffer limits:
+	//     Linux:      sudo sysctl -w net.core.rmem_max=<read-buffer>
+	//     BSD/Darwin: sudo sysctl -w kern.ipc.maxsockbuf=<read-buffer>
+	DefaultReadBuffer = 0
 )
 
 type Config struct {
@@ -34,6 +47,7 @@ type Config struct {
 	RetentionPolicy string        `toml:"retention-policy"`
 	BatchSize       int           `toml:"batch-size"`
 	BatchPending    int           `toml:"batch-pending"`
+	ReadBuffer      int           `toml:"read-buffer"`
 	BatchTimeout    toml.Duration `toml:"batch-timeout"`
 }
 
@@ -64,5 +78,8 @@ func (c *Config) WithDefaults() *Config {
 	if d.BatchTimeout == 0 {
 		d.BatchTimeout = toml.Duration(DefaultBatchTimeout)
 	}
+	if d.ReadBuffer == 0 {
+		d.ReadBuffer = DefaultReadBuffer
+	}
 	return &d
 }

services/udp/service.go

@@ -17,9 +17,13 @@ import (
 	"github.com/influxdb/influxdb/tsdb"
 )
 
-// from https://en.wikipedia.org/wiki/User_Datagram_Protocol#Packet_structure
 const (
+	// Maximum UDP packet size
+	// see https://en.wikipedia.org/wiki/User_Datagram_Protocol#Packet_structure
 	UDPBufferSize = 65536
+
+	// Arbitrary, testing indicated that this doesn't typically get over 10
+	parserChanLen = 1000
 )
 
 // statistics gathered by the UDP package.
@@ -44,8 +48,9 @@ type Service struct {
 	wg   sync.WaitGroup
 	done chan struct{}
 
-	batcher *tsdb.PointBatcher
-	config  Config
+	parserChan chan []byte
+	batcher    *tsdb.PointBatcher
+	config     Config
 
 	PointsWriter interface {
 		WritePoints(p *cluster.WritePointsRequest) error
@@ -62,10 +67,11 @@ type Service struct {
 func NewService(c Config) *Service {
 	d := *c.WithDefaults()
 	return &Service{
-		config:  d,
-		done:    make(chan struct{}),
-		batcher: tsdb.NewPointBatcher(d.BatchSize, d.BatchPending, time.Duration(d.BatchTimeout)),
-		Logger:  log.New(os.Stderr, "[udp] ", log.LstdFlags),
+		config:     d,
+		done:       make(chan struct{}),
+		parserChan: make(chan []byte, parserChanLen),
+		batcher:    tsdb.NewPointBatcher(d.BatchSize, d.BatchPending, time.Duration(d.BatchTimeout)),
+		Logger:     log.New(os.Stderr, "[udp] ", log.LstdFlags),
 	}
 }
 
@@ -99,16 +105,26 @@ func (s *Service) Open() (err error) {
 		return err
 	}
 
+	if s.config.ReadBuffer != 0 {
+		err = s.conn.SetReadBuffer(s.config.ReadBuffer)
+		if err != nil {
+			s.Logger.Printf("Failed to set UDP read buffer to %d: %s",
+				s.config.ReadBuffer, err)
+			return err
+		}
+	}
+
 	s.Logger.Printf("Started listening on UDP: %s", s.config.BindAddress)
 
-	s.wg.Add(2)
+	s.wg.Add(3)
 	go s.serve()
-	go s.writePoints()
+	go s.parser()
+	go s.writer()
 
 	return nil
 }
 
-func (s *Service) writePoints() {
+func (s *Service) writer() {
 	defer s.wg.Done()
 
 	for {
@@ -138,35 +154,46 @@ func (s *Service) serve() {
 
 	s.batcher.Start()
 	for {
-		buf := make([]byte, UDPBufferSize)
 
 		select {
 		case <-s.done:
 			// We closed the connection, time to go.
 			return
 		default:
 			// Keep processing.
+			buf := make([]byte, UDPBufferSize)
+			n, _, err := s.conn.ReadFromUDP(buf)
+			if err != nil {
+				s.statMap.Add(statReadFail, 1)
+				s.Logger.Printf("Failed to read UDP message: %s", err)
+				continue
+			}
+			s.statMap.Add(statBytesReceived, int64(n))
+			s.parserChan <- buf[:n]
 		}
+	}
+}
 
-		n, _, err := s.conn.ReadFromUDP(buf)
-		if err != nil {
-			s.statMap.Add(statReadFail, 1)
-			s.Logger.Printf("Failed to read UDP message: %s", err)
-			continue
-		}
-		s.statMap.Add(statBytesReceived, int64(n))
+func (s *Service) parser() {
+	defer s.wg.Done()
 
-		points, err := models.ParsePoints(buf[:n])
-		if err != nil {
-			s.statMap.Add(statPointsParseFail, 1)
-			s.Logger.Printf("Failed to parse points: %s", err)
-			continue
-		}
+	for {
+		select {
+		case <-s.done:
+			return
+		case buf := <-s.parserChan:
+			points, err := models.ParsePoints(buf)
+			if err != nil {
+				s.statMap.Add(statPointsParseFail, 1)
+				s.Logger.Printf("Failed to parse points: %s", err)
+				return
+			}
 
-		for _, point := range points {
-			s.batcher.In() <- point
+			for _, point := range points {
+				s.batcher.In() <- point
+			}
+			s.statMap.Add(statPointsReceived, int64(len(points)))
 		}
-		s.statMap.Add(statPointsReceived, int64(len(points)))
 	}
 }