Optimize low-bar network performance

Author: bitprecipice

CODESTYLE.md

@@ -83,6 +83,8 @@ A routing key is a value that decides *which* instance of something a message be
 Send and receive sit on opposite sides of the backpressure contract. **Senders block**: a packet/frame send is allowed to wait (buffer full, contract wait, write timeout) because blocking the producer is how backpressure propagates toward the source. **Receivers buffer and drop**: a packet/frame receive callback must never block, because by the time data reaches the receive side there is no producer left to slow down — there is only the delivery pipeline, and stalling it stalls everyone behind the stall.
 
 - **A receive callback that hands off must use a 0 timeout on the handoff.** Enqueue non-blocking; if the queue is full, drop (and count the drop). Never propagate a downstream wait back into the delivery path.
+- **A bounded carrier adapter may separate admission from forwarding.** The carrier reader must still offer into that adapter with a zero-wait send and drop immediately on refusal. One adapter-owned forwarding worker may wait on its consumer, because the carrier reader is insulated by the queue; the queue must have independent hard message and byte bounds, cancellation must join the worker, and every queued pooled buffer must be returned before lifecycle completion. Do not share that waiting worker across independent carrier readers.
+- **A reliable byte-stream fragment cannot be dropped and skipped.** If a shared receiver multiplexes reliable logical streams and one bounded handoff refuses immediately, close that complete connection/generation so its owner can reconnect. Waiting creates cross-stream head-of-line blocking; continuing after a dropped byte fragment silently corrupts framing.
 - **Do not call a blocking send from inside a receive callback.** A send blocks by design, which makes it exactly the thing a receive callback must not contain. Hand the data to a queue owned by a sender goroutine instead, with a 0-timeout enqueue as above.
 - **Dropping is correct.** The transmission control running on top of the transport (TCP in the tunnel, the transfer protocol's ack/resend) exists to discover the achievable rate; a dropped packet is the signal it feeds on. Buffering "to be safe" hides the signal and converts loss into latency; blocking converts it into a stall.
 - The failure shape when this rule is broken is **head-of-line blocking across unrelated flows**: receive delivery is fanned out from shared pumps (a dispatch shard serves many flows; a client receive loop serves every sequence from a source), so one blocked callback parks every flow sharing the pump. One dead destination whose return send blocks for its full write timeout can starve delivery for all live destinations for that entire window — observed as flows that look dead on arrival while their peer is provably alive.

LOWBAR.md

@@ -21,6 +21,7 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"sync/atomic"
 
 	// "encoding/base64"
 	"bytes"
@@ -51,6 +52,17 @@ type sinkReceive struct {
 	provideMode  protocol.ProvideMode
 }
 
+// Performs the sink's callback-to-printer handoff without stalling the
+// client's shared receive pump.
+func enqueueSinkReceive(receives chan<- *sinkReceive, receive *sinkReceive) bool {
+	select {
+	case receives <- receive:
+		return true
+	default:
+		return false
+	}
+}
+
 // snapshotSinkReceive formats borrowed receive frames before the callback
 // returns. The decoder may immediately clear and reuse the Frame objects.
 func snapshotSinkReceive(
@@ -708,17 +720,31 @@ func sink(opts docopt.Opts) {
 		// go platformTransport.Run(routeManager)
 	}
 
-	receives := make(chan *sinkReceive)
+	const receiveBufferSize = 256
+	receives := make(chan *sinkReceive, receiveBufferSize)
+	var receiveDropCount atomic.Uint64
 
 	client.AddReceiveCallback(func(source connect.TransferPath, frames []*protocol.Frame, peer connect.Peer) {
-		receives <- snapshotSinkReceive(source, frames, peer)
+		if !enqueueSinkReceive(receives, snapshotSinkReceive(source, frames, peer)) {
+			receiveDropCount.Add(1)
+		}
 	})
 
 	// FIXME reassemble the chunks. Only a complete message counts as 1 against the message count
-	for i := 0; messageCount < 0 || i < messageCount; i += 1 {
+	reportDrops := func() {
+		if dropCount := receiveDropCount.Swap(0); 0 < dropCount {
+			Err.Printf("sink receive buffer full; dropped %d callback delivery(s)", dropCount)
+		}
+	}
+	defer reportDrops()
+	for receiveCount := 0; messageCount < 0 || receiveCount < messageCount; {
 		select {
 		case receive := <-receives:
 			fmt.Printf("[%s %s] %s\n", receive.source, receive.provideMode, receive.frameSummary)
+			receiveCount += 1
+			reportDrops()
+		case <-time.After(time.Second):
+			reportDrops()
 		}
 	}
 }

connectctl/main.go

@@ -30,3 +30,21 @@ func TestSnapshotSinkReceiveDoesNotRetainBorrowedFrames(t *testing.T) {
 		t.Fatalf("frame summary changed after borrowed frame reuse: got %q want %q", snapshot.frameSummary, wantSummary)
 	}
 }
+
+// A full printer queue drops immediately instead of blocking the shared
+// client receive pump.
+func TestEnqueueSinkReceiveDropsWhenFull(t *testing.T) {
+	receives := make(chan *sinkReceive, 1)
+	first := &sinkReceive{frameSummary: "first"}
+	second := &sinkReceive{frameSummary: "second"}
+
+	if !enqueueSinkReceive(receives, first) {
+		t.Fatal("first receive was not admitted")
+	}
+	if enqueueSinkReceive(receives, second) {
+		t.Fatal("second receive was admitted to a full queue")
+	}
+	if got := <-receives; got != first {
+		t.Fatalf("queued receive = %p, want %p", got, first)
+	}
+}

connectctl/main_test.go

@@ -8,6 +8,7 @@ package connect
 import (
 	"context"
 	"runtime"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -144,11 +145,7 @@ func BenchmarkMultiClientEgressParallel(b *testing.B) {
 	defer cancel()
 
 	providerClientId := NewId()
-	settings := DefaultClientSettings()
-	settings.SendBufferSettings.SequenceBufferSize = 0
-	settings.SendBufferSettings.AckBufferSize = 0
-	settings.ReceiveBufferSettings.SequenceBufferSize = 0
-	settings.ForwardBufferSettings.SequenceBufferSize = 0
+	settings := DefaultClientSettingsWithBufferSize(256)
 	providerClient := NewClient(ctx, providerClientId, NewNoContractClientOob(), settings)
 	defer providerClient.Cancel()
 
@@ -162,6 +159,7 @@ func BenchmarkMultiClientEgressParallel(b *testing.B) {
 	if err != nil {
 		b.Fatal(err)
 	}
+	defer natClient.Close()
 
 	clientId := NewId()
 	source := SourceId(clientId)
@@ -207,18 +205,55 @@ func BenchmarkMultiClientBidirectional(b *testing.B) {
 	defer cancel()
 
 	providerClientId := NewId()
-	settings := DefaultClientSettings()
-	settings.SendBufferSettings.SequenceBufferSize = 0
-	settings.SendBufferSettings.AckBufferSize = 0
-	settings.ReceiveBufferSettings.SequenceBufferSize = 0
-	settings.ForwardBufferSettings.SequenceBufferSize = 0
+	settings := DefaultClientSettingsWithBufferSize(256)
 	providerClient := NewClient(ctx, providerClientId, NewNoContractClientOob(), settings)
 	defer providerClient.Cancel()
 
+	type providerEcho struct {
+		packet      []byte
+		destination Id
+		transferKey TransferKey
+	}
+	providerEchoes := make(chan providerEcho, 1024)
+	var providerEchoWaitGroup sync.WaitGroup
+	workerCount := min(runtime.GOMAXPROCS(0), 4)
+	for range workerCount {
+		providerEchoWaitGroup.Add(1)
+		go func() {
+			defer providerEchoWaitGroup.Done()
+			for {
+				select {
+				case <-ctx.Done():
+					for {
+						select {
+						case echo := <-providerEchoes:
+							MessagePoolReturn(echo.packet)
+						default:
+							return
+						}
+					}
+				case echo := <-providerEchoes:
+					success, _ := providerClient.sendRawWithTimeoutDetailed(
+						protocol.MessageType_IpIpPacketFromProvider,
+						echo.packet,
+						echo.destination,
+						nil,
+						0,
+						time.Second,
+						echo.transferKey,
+					)
+					if !success {
+						MessagePoolReturn(echo.packet)
+					}
+				}
+			}
+		}()
+	}
 	// the provider echoes each received packet back with the path reversed, so
 	// the echo lands on the originating flow's update (the steady-state ingress
-	// path).
-	providerClient.AddReceiveCallback(func(source TransferPath, frames []*protocol.Frame, peer Peer) {
+	// path). The shared receive callback only performs a bounded zero-wait
+	// handoff; fixed workers own the blocking transfer sends.
+	providerReceiveUnsub := providerClient.AddReceiveCallback(func(source TransferPath, frames []*protocol.Frame, peer Peer) {
 		for _, frame := range frames {
 			packet, err := ipPacketToProviderBytes(frame)
 			if err != nil {
@@ -231,16 +266,23 @@ func BenchmarkMultiClientBidirectional(b *testing.B) {
 			}
 			reversed := ipPath.ReverseValue()
 			echo := ipOosPacket(&reversed, payload)
-			providerClient.sendRawWithTimeoutDetailed(
-				protocol.MessageType_IpIpPacketFromProvider,
-				echo,
-				source.SourceId,
-				nil,
-				0,
-				-1,
-			)
+			select {
+			case providerEchoes <- providerEcho{
+				packet:      echo,
+				destination: source.SourceId,
+				transferKey: peer.TransferKey,
+			}:
+			default:
+				MessagePoolReturn(echo)
+			}
 		}
 	})
+	defer func() {
+		providerReceiveUnsub()
+		cancel()
+		providerClient.Cancel()
+		providerEchoWaitGroup.Wait()
+	}()
 
 	var receiveCount atomic.Int64
 	natClient, err := testingNewMultiClient(
@@ -253,6 +295,7 @@ func BenchmarkMultiClientBidirectional(b *testing.B) {
 	if err != nil {
 		b.Fatal(err)
 	}
+	defer natClient.Close()
 
 	clientId := NewId()
 	source := SourceId(clientId)

contention_bench_test.go

@@ -322,12 +322,14 @@ func marshalSendPackTransferFrame(m *sendPackFrame) []byte {
 // carrying an Ack. The inner ack frame is not session-stamped (wrapping, when
 // it happens, adds the role/companion hint to the outer encrypted frame).
 type sendAckFrame struct {
-	path        TransferPath
-	messageId   Id
-	sequenceId  Id
-	selective   bool
-	tagSendTime uint64
-	tagSet      bool
+	path                    TransferPath
+	messageId               Id
+	sequenceId              Id
+	selective               bool
+	tagSendTime             uint64
+	tagSet                  bool
+	missingContractId       *Id
+	compactContractRecovery bool
 }
 
 func (m *sendAckFrame) sizeAck() int {
@@ -342,6 +344,12 @@ func (m *sendAckFrame) sizeAck() int {
 		tagBody := sizeTagBody(m.tagSendTime)
 		n += protoSizeTag(4) + protoSizeVarint(uint64(tagBody)) + tagBody
 	}
+	if m.missingContractId != nil {
+		n += protoSizeTag(5) + protoSizeVarint(16) + 16
+	}
+	if m.compactContractRecovery {
+		n += protoSizeTag(6) + 1
+	}
 	return n
 }
 
@@ -357,6 +365,13 @@ func (m *sendAckFrame) appendAck(b []byte) []byte {
 		b = protoAppendVarint(b, uint64(sizeTagBody(m.tagSendTime)))
 		b = appendTagBody(b, m.tagSendTime)
 	}
+	if m.missingContractId != nil {
+		b = appendIdField(b, 5, *m.missingContractId)
+	}
+	if m.compactContractRecovery {
+		b = protoAppendTag(b, 6, protoWireVarint)
+		b = append(b, 1)
+	}
 	return b
 }
 
@@ -1286,6 +1301,26 @@ func decodeAck(b []byte) (*protocol.Ack, bool) {
 				return nil, false
 			}
 			ack.Tag = tg
+		case 5: // missing_contract_id (bytes)
+			if typ != protowire.BytesType {
+				return nil, false
+			}
+			v, vn := protowire.ConsumeBytes(b)
+			if vn < 0 {
+				return nil, false
+			}
+			b = b[vn:]
+			ack.MissingContractId = copyProtoBytes(v)
+		case 6: // compact_contract_recovery
+			if typ != protowire.VarintType {
+				return nil, false
+			}
+			v, vn := protowire.ConsumeVarint(b)
+			if vn < 0 {
+				return nil, false
+			}
+			b = b[vn:]
+			ack.CompactContractRecovery = protowire.DecodeBool(v)
 		default:
 			fn := protowire.ConsumeFieldValue(num, typ, b)
 			if fn < 0 {