chore(dot/network): use sync.Pool for network message buffers

dot/network/config.go

@@ -97,6 +97,8 @@ type Config struct {
 
 	// telemetryInterval how often to send telemetry metrics
 	telemetryInterval time.Duration
+
+	noPreAllocate bool // internal option
 }
 
 // build checks the configuration, sets up the private key for the network service,

dot/network/notifications.go

@@ -28,7 +28,7 @@ import (
 
 var errCannotValidateHandshake = errors.New("failed to validate handshake")
 
-var maxHandshakeSize = unsafe.Sizeof(BlockAnnounceHandshake{}) //nolint
+const maxHandshakeSize = unsafe.Sizeof(BlockAnnounceHandshake{}) //nolint
 
 // Handshake is the interface all handshakes for notifications protocols must implement
 type Handshake interface {
@@ -226,7 +226,7 @@ func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtoc
 			return
 		}
 
-		hs, err := readHandshake(stream, decodeBlockAnnounceHandshake)
+		hs, err := s.readHandshake(stream, decodeBlockAnnounceHandshake)
 		if err != nil {
 			logger.Trace("failed to read handshake", "protocol", info.protocolID, "peer", peer, "error", err)
 			_ = stream.Close()
@@ -294,9 +294,11 @@ func (s *Service) broadcastExcluding(info *notificationsProtocol, excluding peer
 	}
 }
 
-func readHandshake(stream libp2pnetwork.Stream, decoder HandshakeDecoder) (Handshake, error) {
-	msgBytes := make([]byte, maxHandshakeSize)
-	tot, err := readStream(stream, msgBytes)
+func (s *Service) readHandshake(stream libp2pnetwork.Stream, decoder HandshakeDecoder) (Handshake, error) {
+	msgBytes := s.getMessageBuffer()
+	defer s.cleanupMessageBuffer(msgBytes)
+
+	tot, err := readStream(stream, msgBytes[:])
 	if err != nil {
 		return nil, err
 	}

dot/network/service.go

@@ -46,6 +46,8 @@ const (
 	lightID         = "/light/2"
 	blockAnnounceID = "/block-announces/1"
 	transactionsID  = "/transactions/1"
+
+	maxMessageSize = maxBlockResponseSize
 )
 
 var (
@@ -72,6 +74,9 @@ type Service struct {
 	mdns      *mdns
 	gossip    *gossip
 	syncQueue *syncQueue
+	bufPool   *sync.Pool
+	bufMap    *sync.Map //map[*[maxMessageSize]byte]struct{}
+	bufCount  int
 
 	notificationsProtocols map[byte]*notificationsProtocol // map of sub-protocol msg ID to protocol info
 	notificationsMu        sync.RWMutex
@@ -132,6 +137,30 @@ func NewService(cfg *Config) (*Service, error) {
 		return nil, err
 	}
 
+	// pre-allocate pool of buffers used to read from streams.
+	// initially allocate as many buffers as minimally necessary which is the number inbound streams we will have,
+	// which should equal min peers times the number of notifications protocols, which is currently 3.
+	//
+	// the purpose of the bufMap is to keep the reference to the buffers inside the *Service
+	// at all times to prevent the buffers from being garbage collected.
+	// with this addition, they are only garbage collected once the *Service is no longer
+	// used, which is when the node shuts down.
+	bufPool := new(sync.Pool)
+	bufPool.New = func() interface{} {
+		return [maxMessageSize]byte{}
+	}
+	bufMap := new(sync.Map)
+
+	var bufCount int
+	if !cfg.noPreAllocate {
+		for i := 0; i < cfg.MinPeers*3; i++ {
+			buf := [maxBlockResponseSize]byte{}
+			bufPool.Put(buf) //nolint
+			bufMap.Store(&buf, struct{}{})
+		}
+		bufCount = cfg.MinPeers * 3
+	}
+
 	network := &Service{
 		ctx:                    ctx,
 		cancel:                 cancel,
@@ -148,6 +177,9 @@ func NewService(cfg *Config) (*Service, error) {
 		lightRequest:           make(map[peer.ID]struct{}),
 		telemetryInterval:      cfg.telemetryInterval,
 		closeCh:                make(chan interface{}),
+		bufPool:                bufPool,
+		bufMap:                 bufMap,
+		bufCount:               bufCount,
 	}
 
 	network.syncQueue = newSyncQueue(network)
@@ -550,15 +582,39 @@ func isInbound(stream libp2pnetwork.Stream) bool {
 	return stream.Stat().Direction == libp2pnetwork.DirInbound
 }
 
+func (s *Service) getMessageBuffer() [maxMessageSize]byte {
+	buf := s.bufPool.Get()
+	if buf == nil {
+		return [maxMessageSize]byte{}
+	}
+
+	msgBytes, ok := buf.([maxMessageSize]byte)
+	if !ok {
+		return [maxMessageSize]byte{}
+	}
+
+	return msgBytes
+}
+
+func (s *Service) cleanupMessageBuffer(buf [maxMessageSize]byte) {
+	s.bufPool.Put(buf) //nolint
+
+	logger.Trace("message buffers allocated", "count", s.bufCount)
+	if s.bufCount >= s.cfg.MaxPeers*3 {
+		return
+	}
+
+	s.bufMap.Store(&buf, struct{}{})
+	s.bufCount++
+}
+
 func (s *Service) readStream(stream libp2pnetwork.Stream, decoder messageDecoder, handler messageHandler) {
-	var (
-		maxMessageSize uint64 = maxBlockResponseSize // TODO: determine actual max message size
-		msgBytes              = make([]byte, maxMessageSize)
-		peer                  = stream.Conn().RemotePeer()
-	)
+	peer := stream.Conn().RemotePeer()
+	msgBytes := s.getMessageBuffer()
+	defer s.cleanupMessageBuffer(msgBytes)
 
 	for {
-		tot, err := readStream(stream, msgBytes)
+		tot, err := readStream(stream, msgBytes[:])
 		if err == io.EOF {
 			continue
 		} else if err != nil {

dot/network/service_test.go

@@ -78,6 +78,8 @@ func createTestService(t *testing.T, cfg *Config) (srvc *Service) {
 		cfg.Syncer = newMockSyncer()
 	}
 
+	cfg.noPreAllocate = true
+
 	srvc, err := NewService(cfg)
 	require.NoError(t, err)
 

dot/network/sync.go

@@ -79,7 +79,7 @@ func (s *Service) handleSyncMessage(stream libp2pnetwork.Stream, msg Message) er
 	return nil
 }
 
-var (
+const (
 	blockRequestSize        uint32 = 128
 	blockRequestBufferSize  int    = 6
 	blockResponseBufferSize int    = 6