read buffer pool

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
@@ -118,7 +119,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_talk.go

@@ -28,7 +28,7 @@ import (
 )
 
 // This is a limit for the number of concurrent talk requests.
-const maxActiveTalkRequests = 1024
+const maxActiveTalkRequests = 2048
 
 // This is the timeout for acquiring a handler execution slot for a talk request.
 // The timeout should be short enough to fit within the request timeout.

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, 1024),
 		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),
@@ -594,9 +594,6 @@ func (t *UDPv5) callDone(c *callV5) {
 func (t *UDPv5) dispatch() {
 	defer t.wg.Done()
 
-	// Arm first read.
-	t.readNextCh <- struct{}{}
-
 	for {
 		select {
 		case c := <-t.callCh:
@@ -629,14 +626,6 @@ func (t *UDPv5) dispatch() {
 			t.send(r.destID, r.destAddr, r.msg, nil)
 
 		case p := <-t.packetInCh:
-			// Arm next read immediately to allow pipelining.
-			// The readLoop can start reading the next packet while this one is being handled.
-			// Backpressure is still maintained by packetInCh (buffer 1) and readNextCh (buffer 1).
-			select {
-			case t.readNextCh <- struct{}{}:
-			case <-t.closeCtx.Done(): // Avoid blocking on send if closing
-				return
-			}
 			t.handlePacket(p.Data, p.Addr)
 
 		case <-t.closeCtx.Done():
@@ -836,21 +825,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])
 	}
 }
 
@@ -864,6 +858,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
 	}
 }
@@ -873,6 +868,8 @@ 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)
+	bufferpool.Put(rawpacket)
+
 	if err != nil {
 		if t.unhandled != nil && v5wire.IsInvalidHeader(err) {
 			// The packet seems unrelated to discv5, send it to the next protocol.

p2p/discover/v5wire/encoding.go

@@ -191,8 +191,15 @@ func (c *Codec) Encode(id enode.ID, addr string, packet Packet, challenge *Whoar
 	case packet.Kind() == WhoareyouPacket:
 		// just send the WHOAREYOU packet raw again, rather than the re-encoded challenge data
 		w := packet.(*Whoareyou)
-		if len(w.Encoded) > 0 {
-			return w.Encoded, w.Nonce, nil
+		if len(w.ChallengeData) > 0 {
+			// This WHOAREYOU packet was encoded before, so it's a resend.
+			// The unmasked packet content is stored in w.ChallengeData.
+			// Just apply the masking again to finish encoding.
+			c.buf.Reset()
+			c.buf.Write(w.ChallengeData)
+			copy(head.IV[:], w.ChallengeData)
+			enc := applyMasking(id, head.IV, c.buf.Bytes())
+			return enc, w.Nonce, nil
 		}
 		head, err = c.encodeWhoareyou(id, packet.(*Whoareyou))
 	case challenge != nil:
@@ -227,7 +234,6 @@ func (c *Codec) Encode(id enode.ID, addr string, packet Packet, challenge *Whoar
 		if err != nil {
 			return nil, Nonce{}, err
 		}
-		challenge.Encoded = bytes.Clone(enc)
 		c.sc.storeSentHandshake(id, addr, challenge)
 		return enc, head.Nonce, err
 	}
@@ -245,13 +251,9 @@ func (c *Codec) Encode(id enode.ID, addr string, packet Packet, challenge *Whoar
 
 // EncodeRaw encodes a packet with the given header.
 func (c *Codec) EncodeRaw(id enode.ID, head Header, msgdata []byte) ([]byte, error) {
+	// header
 	c.writeHeaders(&head)
-
-	// Apply masking.
-	masked := c.buf.Bytes()[sizeofMaskingIV:]
-	mask := head.mask(id)
-	mask.XORKeyStream(masked[:], masked[:])
-
+	applyMasking(id, head.IV, c.buf.Bytes())
 	// Write message data.
 	c.buf.Write(msgdata)
 	return c.buf.Bytes(), nil
@@ -463,7 +465,7 @@ func (c *Codec) Decode(inputData []byte, addr string) (src enode.ID, n *enode.No
 	// Unmask the static header.
 	var head Header
 	copy(head.IV[:], input[:sizeofMaskingIV])
-	mask := head.mask(c.localnode.ID())
+	mask := createMask(c.localnode.ID(), head.IV)
 	staticHeader := input[sizeofMaskingIV:sizeofStaticPacketData]
 	mask.XORKeyStream(staticHeader, staticHeader)
 
@@ -678,13 +680,20 @@ func (h *StaticHeader) checkValid(packetLen int, protocolID [6]byte) error {
 	return nil
 }
 
-// mask returns a cipher for 'masking' / 'unmasking' packet headers.
-func (h *Header) mask(destID enode.ID) cipher.Stream {
+// createMask returns a cipher for 'masking' / 'unmasking' packet headers.
+func createMask(destID enode.ID, iv [16]byte) cipher.Stream {
 	block, err := aes.NewCipher(destID[:16])
 	if err != nil {
 		panic("can't create cipher")
 	}
-	return cipher.NewCTR(block, h.IV[:])
+	return cipher.NewCTR(block, iv[:])
+}
+
+func applyMasking(destID enode.ID, iv [16]byte, packet []byte) []byte {
+	masked := packet[sizeofMaskingIV:]
+	mask := createMask(destID, iv)
+	mask.XORKeyStream(masked[:], masked[:])
+	return packet
 }
 
 func bytesCopy(r *bytes.Buffer) []byte {

p2p/discover/v5wire/msg.go

@@ -63,19 +63,19 @@ type (
 
 	// WHOAREYOU contains the handshake challenge.
 	Whoareyou struct {
-		ChallengeData []byte   // Encoded challenge
-		Nonce         Nonce    // Nonce of request packet
-		IDNonce       [16]byte // Identity proof data
-		RecordSeq     uint64   // ENR sequence number of recipient
+		Nonce     Nonce    // Nonce of request packet
+		IDNonce   [16]byte // Identity proof data
+		RecordSeq uint64   // ENR sequence number of recipient
 
 		// Node is the locally known node record of recipient.
 		// This must be set by the caller of Encode.
-		Node *enode.Node
+		Node *enode.Node `rlp:"-"`
+		// ChallengeData stores the unmasked encoding of the whole packet. This is the
+		// input data for verification. It is assigned by both Encode and Decode
+		// operations.
+		ChallengeData []byte `rlp:"-"`
 
 		sent mclock.AbsTime // for handshake GC.
-
-		// Encoded is packet raw data for sending out, but should not be include in the RLP encoding.
-		Encoded []byte `rlp:"-"`
 	}
 
 	// PING is sent during liveness checks.