stage writes in a buffer; write at once on the transport.

p2p/security/noise/rw.go

@@ -11,12 +11,16 @@ import (
 
 // MaxTransportMsgLength is the Noise-imposed maximum transport message length,
 // inclusive of the MAC size (16 bytes, Poly1305 for noise-libp2p).
-const MaxTransportMsgLength = 65535
+const MaxTransportMsgLength = 0xffff
 
 // MaxPlaintextLength is the maximum payload size. It is MaxTransportMsgLength
 // minus the MAC size. Payloads over this size will be automatically chunked.
 const MaxPlaintextLength = MaxTransportMsgLength - poly1305.TagSize
 
+// LengthPrefixLength is the length of the length prefix itself, which precedes
+// all transport messages in order to delimit them. In bytes.
+const LengthPrefixLength = 2
+
 // Read reads from the secure connection, returning plaintext data in `buf`.
 //
 // Honours io.Reader in terms of behaviour.
@@ -134,12 +138,19 @@ func (s *secureSession) readMsgInsecure() ([]byte, error) {
 
 // writeMsgInsecure writes to the insecure conn.
 // data will be prefixed with its length in bytes, written as a 16-bit uint in network order.
-func (s *secureSession) writeMsgInsecure(data []byte) (n int, err error) {
+func (s *secureSession) writeMsgInsecure(data []byte) (int, error) {
-	binary.BigEndian.PutUint16(s.wlen, uint16(len(data)))
+	// we rather stage the length-prefixed write in a buffer to then call Write
-	n, err = s.insecure.Write(s.wlen)
+	// on the underlying transport at once, rather than Write twice and likely
-	if err != nil {
+	// induce transport-level fragmentation.
-		return n, fmt.Errorf("error writing length prefix: %w", err)
+	l := len(data)
+	buf := pool.Get(LengthPrefixLength + l)
+	defer pool.Put(buf)
+
+	// length-prefix || data
+	binary.BigEndian.PutUint16(buf, uint16(l))
+	n := copy(buf[LengthPrefixLength:], data)
+	if n != l {
+		return 0, fmt.Errorf("assertion failed during noise secure channel write; expected to copy %d bytes, copied: %d", l, n)
 	}
-	n, err = s.insecure.Write(data)
+	return s.insecure.Write(buf)
-	return n + 2, err // +2 for length prefix.
 }

p2p/security/noise/session.go

@@ -23,14 +23,11 @@ type secureSession struct {
 	readLock  sync.Mutex
 	writeLock sync.Mutex
 	insecure  net.Conn
-	// queued bytes seek value.
+
-	qseek int
+	qseek int    // queued bytes seek value.
-	// queued bytes remaining; saves us from computing over and over.
+	qrem  int    // queued bytes remaining; saves us from computing over and over.
-	qrem int
+	qbuf  []byte // queued bytes buffer.
-	// queued bytes buffer.
+	rlen  []byte // work buffer to read in the incoming message length (2 bytes).
-	qbuf []byte
-	// work buffers for read and write message lengths (16 bits each); guarded by locks.
-	rlen, wlen []byte
 
 	enc *noise.CipherState
 	dec *noise.CipherState
@@ -46,7 +43,6 @@ func newSecureSession(tpt *Transport, ctx context.Context, insecure net.Conn, re
 		localKey:  tpt.privateKey,
 		remoteID:  remote,
 		rlen:      make([]byte, 2),
-		wlen:      make([]byte, 2),
 	}
 
 	err := s.runHandshake(ctx)