read loop

Author: thinkAfCod

go.mod

@@ -46,6 +46,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

go.sum

@@ -238,6 +238,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

@@ -97,7 +97,6 @@ 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
@@ -113,12 +112,6 @@ 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
@@ -163,10 +156,9 @@ func ListenV5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
 		return nil, err
 	}
 	go t.tab.loop()
-	t.wg.Add(3)
+	t.wg.Add(2)
 	go t.readLoop()
 	go t.dispatch()
-	go t.writeLoop()
 	return t, nil
 }
 
@@ -186,14 +178,13 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
 		clock:        cfg.Clock,
 		respTimeout:  cfg.V5RespTimeout,
 		// channels into dispatch
-		packetInCh:    make(chan ReadPacket, 64),
+		packetInCh:    make(chan ReadPacket, 4),
 		callCh:        make(chan *callV5),
 		callDoneCh:    make(chan *callV5),
 		sendCh:        make(chan sendRequest),
 		sendNoRespCh:  make(chan *sendNoRespRequest),
 		respTimeoutCh: make(chan *callTimeout),
 		unhandled:     cfg.Unhandled,
-		writeCh:       make(chan pendingWrite, 32), // Buffered channel for outgoing packets
 		// state of dispatch
 		codec:            v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID),
 		activeCallByNode: make(map[enode.ID]*callV5),
@@ -625,7 +616,6 @@ func (t *UDPv5) dispatch() {
 			t.handlePacket(p.Data, p.Addr)
 
 		case <-t.closeCtx.Done():
-			close(t.writeCh)
 			for id, queue := range t.callQueue {
 				for _, c := range queue {
 					c.err <- errClosed
@@ -766,39 +756,9 @@ func (t *UDPv5) send(toID enode.ID, toAddr netip.AddrPort, packet v5wire.Packet,
 		return nonce, err
 	}
 
+	_, err = t.conn.WriteToUDPAddrPort(enc, toAddr)
 	t.log.Trace(">> "+packet.Name(), t.logcontext...)
-
-	dataForSend := make([]byte, 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():
-		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 netutil.IsTemporaryError(err) {
-			t.log.Debug("Temporary UDP write error", "addr", pw.toAddr, "err", err)
-		} else if err != nil {
-			if !errors.Is(err, net.ErrClosed) || !errors.Is(err, io.EOF) {
-				t.log.Warn("UDP write error", "addr", pw.toAddr, "err", err)
-			}
-			return
-		}
-	}
+	return nonce, err
 }
 
 // readLoop runs in its own goroutine and reads packets from the network.
@@ -819,7 +779,9 @@ func (t *UDPv5) readLoop() {
 			}
 			return
 		}
-		t.dispatchReadPacket(from, buf[:nbytes])
+		content := make([]byte, nbytes)
+		copy(content, buf[:nbytes])
+		t.dispatchReadPacket(from, content)
 	}
 }
 
@@ -829,10 +791,8 @@ func (t *UDPv5) dispatchReadPacket(from netip.AddrPort, content []byte) bool {
 	if from.Addr().Is4In6() {
 		from = netip.AddrPortFrom(netip.AddrFrom4(from.Addr().As4()), from.Port())
 	}
-	data := make([]byte, len(content))
-	copy(data, content)
 	select {
-	case t.packetInCh <- ReadPacket{data, from}:
+	case t.packetInCh <- ReadPacket{content, from}:
 		return true
 	case <-t.closeCtx.Done():
 		return false