make send in separate goroutine

Signed-off-by: Chen Kai <281165273grape@gmail.com>

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

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=

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.