From 3b223b26dcdc92e678387ff7c201e9f773234506 Mon Sep 17 00:00:00 2001 From: Chen Kai <281165273grape@gmail.com> Date: Sun, 11 May 2025 16:53:48 +0800 Subject: [PATCH] make send in separate goroutine Signed-off-by: Chen Kai <281165273grape@gmail.com> --- go.mod | 1 + go.sum | 2 ++ p2p/discover/v5_udp.go | 62 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 62 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 924f0d2642..3115993a4d 100644 --- a/go.mod +++ b/go.mod @@ -119,6 +119,7 @@ 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/go.sum b/go.sum index 5d35a7a0e1..8ac5154557 100644 --- a/go.sum +++ b/go.sum @@ -240,6 +240,8 @@ github.com/labstack/echo/v4 v4.2.1/go.mod h1:AA49e0DZ8kk5jTOOCKNuPR6oTnBS0dYiM4F github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k= github.com/leanovate/gopter v0.2.11 h1:vRjThO1EKPb/1NsDXuDrzldR28RLkBflWYcU9CvzWu4= github.com/leanovate/gopter v0.2.11/go.mod h1:aK3tzZP/C+p1m3SPRE4SYZFGP7jjkuSI4f7Xvpt0S9c= +github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8= +github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg= github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= github.com/matryer/moq v0.0.0-20190312154309-6cfb0558e1bd/go.mod h1:9ELz6aaclSIGnZBoaSLZ3NAl1VTufbOrXBPvtcy6WiQ= diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 23801b923a..8394d3162d 100644 --- a/p2p/discover/v5_udp.go +++ b/p2p/discover/v5_udp.go @@ -37,6 +37,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/p2p/netutil" + bufferpool "github.com/libp2p/go-buffer-pool" ) const ( @@ -99,6 +100,7 @@ type UDPv5 struct { sendCh chan sendRequest sendNoRespCh chan *sendNoRespRequest unhandled chan<- ReadPacket + writeCh chan pendingWrite // New channel for outgoing packets // state of dispatch codec codecV5 @@ -114,6 +116,12 @@ type UDPv5 struct { wg sync.WaitGroup } +// pendingWrite holds data for a packet to be sent by the writeLoop. +type pendingWrite struct { + toAddr netip.AddrPort + data []byte +} + type sendRequest struct { destID enode.ID destAddr netip.AddrPort @@ -158,9 +166,10 @@ func ListenV5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) { return nil, err } go t.tab.loop() - t.wg.Add(2) + t.wg.Add(3) go t.readLoop() go t.dispatch() + go t.writeLoop() return t, nil } @@ -188,6 +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 // state of dispatch codec: v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID), activeCallByNode: make(map[enode.ID]*callV5), @@ -622,6 +632,7 @@ func (t *UDPv5) dispatch() { t.handlePacket(p.Data, p.Addr) case <-t.closeCtx.Done(): close(t.readNextCh) + close(t.writeCh) for id, queue := range t.callQueue { for _, c := range queue { c.err <- errClosed @@ -761,10 +772,55 @@ func (t *UDPv5) send(toID enode.ID, toAddr netip.AddrPort, packet v5wire.Packet, t.log.Warn(">> "+packet.Name(), t.logcontext...) return nonce, err } + t.logcontext = append(t.logcontext, "rawpacket", hexutil.Encode(enc)) - _, err = t.conn.WriteToUDPAddrPort(enc, toAddr) t.log.Trace(">> "+packet.Name(), t.logcontext...) - return nonce, err + + dataForSend := bufferpool.Get(len(enc)) + copy(dataForSend, enc) + + pw := pendingWrite{ + toAddr: toAddr, + data: dataForSend, // codec.Encode should return a new slice, safe to pass directly. + } + + select { + case t.writeCh <- pw: + // Packet successfully queued. + return nonce, nil + case <-t.closeCtx.Done(): + bufferpool.Put(dataForSend) + return nonce, errClosed + } +} + +// writeLoop runs in its own goroutine and sends packets from the writeCh to the network. +func (t *UDPv5) writeLoop() { + defer t.wg.Done() + for pw := range t.writeCh { // Loop continues until writeCh is closed and empty. + _, err := t.conn.WriteToUDPAddrPort(pw.data, pw.toAddr) + if err != nil { + // Generic error logging, as we don't have packetName or rich context here. + select { + case <-t.closeCtx.Done(): + // Log trace level if error occurs during or after shutdown initiation. + t.log.Trace("UDP write error during/after shutdown", "addr", pw.toAddr, "err", err) + default: + // Not closing, so it's a more unexpected error. + if netutil.IsTemporaryError(err) { + t.log.Debug("Temporary UDP write error", "addr", pw.toAddr, "err", err) + } else if !errors.Is(err, net.ErrClosed) && !errors.Is(err, io.EOF) { // Avoid logging common "closed" errors if not caught by closeCtx. + t.log.Warn("UDP write error", "addr", pw.toAddr, "err", err) + } + } + } else { + // Minimal trace log confirming the actual send. + // The detailed packet-specific trace was done in the 'send' method. + t.log.Trace("UDP packet data sent", "addr", pw.toAddr, "len", len(pw.data)) + } + + bufferpool.Put(pw.data) + } } // readLoop runs in its own goroutine and reads packets from the network.