diff --git a/go.mod b/go.mod index 0432acf97a..30e815d4d7 100644 --- a/go.mod +++ b/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 diff --git a/p2p/discover/v5_talk.go b/p2p/discover/v5_talk.go index dca09870d8..3187aafe53 100644 --- a/p2p/discover/v5_talk.go +++ b/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. diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 434962b725..637a71ea52 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, 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. diff --git a/p2p/discover/v5wire/encoding.go b/p2p/discover/v5wire/encoding.go index ec5ef8a261..e1067378ab 100644 --- a/p2p/discover/v5wire/encoding.go +++ b/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 { diff --git a/p2p/discover/v5wire/msg.go b/p2p/discover/v5wire/msg.go index 089fd4ebdc..eb4123bba1 100644 --- a/p2p/discover/v5wire/msg.go +++ b/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.