diff --git a/go.mod b/go.mod index 3115993a4d..cf8f786b53 100644 --- a/go.mod +++ b/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 diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 8394d3162d..b4e4cac6c1 100644 --- a/p2p/discover/v5_udp.go +++ b/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 range t.readNextCh { - nbytes, from, err := t.conn.ReadFromUDPAddrPort(buf) - if netutil.IsTemporaryError(err) { - // Ignore temporary read errors. - t.log.Debug("Temporary 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) + for { + select { + case <-t.closeCtx.Done(): + t.log.Trace("UDP read loop shutdown") + default: + buf := bufferpool.Get(maxPacketSize) + nbytes, from, err := t.conn.ReadFromUDPAddrPort(buf) + if netutil.IsTemporaryError(err) { + // Ignore temporary read errors. + t.log.Debug("Temporary 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 { - case t.readNextCh <- struct{}{}: - case <-t.closeCtx.Done(): - } + bufferpool.Put(rawpacket) if err != nil { if t.unhandled != nil && v5wire.IsInvalidHeader(err) {