read buffer pool

go.mod

@@ -47,6 +47,7 @@ require (
 	github.com/jedisct1/go-minisign v0.0.0-20230811132847-661be99b8267
 	github.com/karalabe/hid v1.0.1-0.20240306101548-573246063e52
 	github.com/kylelemons/godebug v1.1.0
+	github.com/libp2p/go-buffer-pool v0.1.0
 	github.com/mattn/go-colorable v0.1.13
 	github.com/mattn/go-isatty v0.0.20
 	github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416
@@ -119,7 +120,6 @@ require (
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
-	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/mattn/go-runewidth v0.0.13 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/minio/sha256-simd v1.0.0 // indirect

p2p/discover/v5_udp.go

@@ -189,7 +189,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
 		clock:        cfg.Clock,
 		respTimeout:  cfg.V5RespTimeout,
 		// channels into dispatch
-		packetInCh:    make(chan ReadPacket, 1),
+		packetInCh:    make(chan ReadPacket, 256),
 		readNextCh:    make(chan struct{}, 1),
 		callCh:        make(chan *callV5),
 		callDoneCh:    make(chan *callV5),
@@ -197,7 +197,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
 		sendNoRespCh:  make(chan *sendNoRespRequest),
 		respTimeoutCh: make(chan *callTimeout),
 		unhandled:     cfg.Unhandled,
-		writeCh:       make(chan pendingWrite, 128), // Buffered channel for outgoing packets
+		writeCh:       make(chan pendingWrite, 256), // Buffered channel for outgoing packets
 		// state of dispatch
 		codec:            v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID),
 		activeCallByNode: make(map[enode.ID]*callV5),
@@ -827,21 +827,26 @@ func (t *UDPv5) writeLoop() {
 func (t *UDPv5) readLoop() {
 	defer t.wg.Done()
 
-	buf := make([]byte, maxPacketSize)
+	for {
-	for range t.readNextCh {
+		select {
-		nbytes, from, err := t.conn.ReadFromUDPAddrPort(buf)
+		case <-t.closeCtx.Done():
-		if netutil.IsTemporaryError(err) {
+			t.log.Trace("UDP read loop shutdown")
-			// Ignore temporary read errors.
+		default:
-			t.log.Debug("Temporary UDP read error", "err", err)
+			buf := bufferpool.Get(maxPacketSize)
-			continue
+			nbytes, from, err := t.conn.ReadFromUDPAddrPort(buf)
-		} else if err != nil {
+			if netutil.IsTemporaryError(err) {
-			// Shut down the loop for permanent errors.
+				// Ignore temporary read errors.
-			if !errors.Is(err, io.EOF) {
+				t.log.Debug("Temporary UDP read error", "err", err)
-				t.log.Debug("UDP read error", "err", err)
+				continue
+			} else if err != nil {
+				// Shut down the loop for permanent errors.
+				if !errors.Is(err, io.EOF) {
+					t.log.Debug("UDP read error", "err", err)
+				}
+				return
 			}
-			return
+			t.dispatchReadPacket(from, buf[:nbytes])
 		}
-		t.dispatchReadPacket(from, buf[:nbytes])
 	}
 }
 
@@ -855,6 +860,7 @@ func (t *UDPv5) dispatchReadPacket(from netip.AddrPort, content []byte) bool {
 	case t.packetInCh <- ReadPacket{content, from}:
 		return true
 	case <-t.closeCtx.Done():
+		bufferpool.Put(content)
 		return false
 	}
 }
@@ -864,10 +870,7 @@ func (t *UDPv5) handlePacket(rawpacket []byte, fromAddr netip.AddrPort) error {
 	addr := fromAddr.String()
 	t.log.Trace("<< "+addr, "rawPacket", hexutil.Encode(rawpacket))
 	fromID, fromNode, packet, err := t.codec.Decode(rawpacket, addr)
-	select {
+	bufferpool.Put(rawpacket)
-	case t.readNextCh <- struct{}{}:
-	case <-t.closeCtx.Done():
-	}
 
 	if err != nil {
 		if t.unhandled != nil && v5wire.IsInvalidHeader(err) {