perf(streamframe): pool Send completion channels (#312)

* perf(streamframe): pool Send completion channels and track senders with a WaitGroup

Synchronous Send reuses its one-slot completion channel from a sync.Pool;
a channel returns to the pool only after its single result was received,
so a close-abandoned channel the writer may still complete is dropped.
The in-flight sender count becomes a sync.WaitGroup instead of a counter
plus a per-generation done channel; Add still happens under egressMu while
admission is open, before the writer's Wait.

TransportSend: 4 -> 1 allocs/op, 272 -> 32 B/op.
Typed round trip: 86 -> 74 allocs/op, -19% B/op, time unchanged.

* test(broker): advance the idle clock until the pool evicts

TestPoolIdleEviction advanced the manual clock once after seeing any timer.
The watcher creates its timer, then stops and re-arms it on its first loop,
so an Advance landing before the re-arm fired a stopped timer and the test
timed out on loaded CI runners. Advance repeatedly until the eviction is
observed.

commit: 8ca904fc6b70c0610ed82b89845f4b21b81bc163
author: bnema · 18d ago · 2026-10-04 16:34 UTC
committer: GitHub · 18d ago · 2026-10-04 16:34 UTC
parent: d0781e6
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

2 files · +39 -39

M internal/adapters/streamframe/streamframe.go +28 -36


@@ -79,16 +79,17 @@ type Framer struct {
79	79		// header is the length-prefix buffer, guarded by readMu.
80	80		header [4]byte
81	81
82		-	egressMu sync.Mutex
83		-	egressClosing bool
84		-	egressSenders int
85		-	egressSendersDone chan struct{}
86		-	egress chan sendRequest
87		-	queuedBytes uint64
88		-	done chan struct{}
89		-	writerDone chan struct{}
90		-	closeOnce sync.Once
91		-	closeErr error
	82	+	egressMu sync.Mutex
	83	+	egressClosing bool
	84	+	// egressSenders counts synchronous senders between admission and queue
	85	+	// hand-off; the writer drains the queue only after they all leave.
	86	+	egressSenders sync.WaitGroup
	87	+	egress chan sendRequest
	88	+	queuedBytes uint64
	89	+	done chan struct{}
	90	+	writerDone chan struct{}
	91	+	closeOnce sync.Once
	92	+	closeErr error
92	93		}
93	94
94	95		type sendRequest struct {

@@ -139,18 +140,29 @@ func (f *Framer) Send(payload []byte) error {
139	140		if err != nil {
140	141		return err
141	142		}
142		-	result := make(chan error, 1)
	143	+	result, _ := resultPool.Get().(chan error)
	144	+	if result == nil {
	145	+	result = make(chan error, 1)
	146	+	}
143	147		if err := f.enqueueWait(sendRequest{data: data, done: result}); err != nil {
	148	+	resultPool.Put(result)
144	149		return err
145	150		}
146	151		select {
147	152		case err := <-result:
	153	+	resultPool.Put(result)
148	154		return err
149	155		case <-f.done:
	156	+	// The writer still owns result and may complete it later, so it is
	157	+	// not returned to the pool.
150	158		return ErrClosed
151	159		}
152	160		}
153	161
	162	+	// resultPool recycles the one-slot completion channels of synchronous Send.
	163	+	// A channel is pooled only after its single result was received.
	164	+	var resultPool sync.Pool
	165	+
154	166		// SendAsync admits payload for ordered background transmission without
155	167		// waiting for the write. It returns ErrBackpressure when the byte or message
156	168		// budget is exhausted; ownership includes a payload copy.

@@ -272,18 +284,16 @@ func (f *Framer) enqueueWait(req sendRequest) error {
272	284		f.egressMu.Unlock()
273	285		return ErrClosed
274	286		}
275		-	if f.egressSenders == 0 {
276		-	f.egressSendersDone = make(chan struct{})
277		-	}
278		-	f.egressSenders++
	287	+	// Add happens under egressMu while egressClosing is false, so it always
	288	+	// precedes the writer's Wait, which starts only after Close set it.
	289	+	f.egressSenders.Add(1)
279	290		f.egressMu.Unlock()
	291	+	defer f.egressSenders.Done()
280	292
281	293		select {
282	294		case <-f.done:
283		-	f.finishEgressSender()
284	295		return ErrClosed
285	296		case f.egress <- req:
286		-	f.finishEgressSender()
287	297		return nil
288	298		}
289	299		}

@@ -314,24 +324,6 @@ func (f *Framer) enqueueAsync(req sendRequest) error {
314	324		}
315	325		}
316	326
317		-	func (f *Framer) finishEgressSender() {
318		-	f.egressMu.Lock()
319		-	defer f.egressMu.Unlock()
320		-	f.egressSenders--
321		-	if f.egressSenders == 0 {
322		-	close(f.egressSendersDone)
323		-	}
324		-	}
325		-
326		-	func (f *Framer) waitForEgressSenders() {
327		-	f.egressMu.Lock()
328		-	done := f.egressSendersDone
329		-	f.egressMu.Unlock()
330		-	if done != nil {
331		-	<-done
332		-	}
333		-	}
334		-
335	327		func (f *Framer) writeLoop() {
336	328		defer close(f.writerDone)
337	329		for {

@@ -342,7 +334,7 @@ func (f *Framer) writeLoop() {
342	334		f.egressMu.Unlock()
343	335		f.completeWrite(req, writeAll(f.w, req.data))
344	336		case <-f.done:
345		-	f.waitForEgressSenders()
	337	+	f.egressSenders.Wait()
346	338		f.egressMu.Lock()
347	339		for {
348	340		select {

 	// header is the length-prefix buffer, guarded by readMu.
 	header [4]byte
-	egressMu          sync.Mutex
+	egressMu      sync.Mutex
-	egressClosing     bool
+	egressClosing bool
-	egressSenders     int
+	// egressSenders counts synchronous senders between admission and queue
-	egressSendersDone chan struct{}
+	// hand-off; the writer drains the queue only after they all leave.
-	egress            chan sendRequest
+	egressSenders sync.WaitGroup
-	queuedBytes       uint64
+	egress        chan sendRequest
-	done              chan struct{}
+	queuedBytes   uint64
-	writerDone        chan struct{}
+	done          chan struct{}
-	closeOnce         sync.Once
+	writerDone    chan struct{}
-	closeErr          error
+	closeOnce     sync.Once
+	closeErr      error
 }
 type sendRequest struct {
 	if err != nil {
 		return err
 	}
-	result := make(chan error, 1)
+	result, _ := resultPool.Get().(chan error)
+	if result == nil {
+		result = make(chan error, 1)
+	}
 	if err := f.enqueueWait(sendRequest{data: data, done: result}); err != nil {
+		resultPool.Put(result)
 		return err
 	}
 	select {
 	case err := <-result:
+		resultPool.Put(result)
 		return err
 	case <-f.done:
+		// The writer still owns result and may complete it later, so it is
+		// not returned to the pool.
 		return ErrClosed
 	}
 }
+// resultPool recycles the one-slot completion channels of synchronous Send.
+// A channel is pooled only after its single result was received.
+var resultPool sync.Pool
 // SendAsync admits payload for ordered background transmission without
 // waiting for the write. It returns ErrBackpressure when the byte or message
 // budget is exhausted; ownership includes a payload copy.
 		f.egressMu.Unlock()
 		return ErrClosed
 	}
-	if f.egressSenders == 0 {
+	// Add happens under egressMu while egressClosing is false, so it always
-		f.egressSendersDone = make(chan struct{})
+	// precedes the writer's Wait, which starts only after Close set it.
-	}
+	f.egressSenders.Add(1)
-	f.egressSenders++
 	f.egressMu.Unlock()
+	defer f.egressSenders.Done()
 	select {
 	case <-f.done:
-		f.finishEgressSender()
 		return ErrClosed
 	case f.egress <- req:
-		f.finishEgressSender()
 		return nil
 	}
 }
 	}
 }
-func (f *Framer) finishEgressSender() {
-	f.egressMu.Lock()
-	defer f.egressMu.Unlock()
-	f.egressSenders--
-	if f.egressSenders == 0 {
-		close(f.egressSendersDone)
-	}
-}
-func (f *Framer) waitForEgressSenders() {
-	f.egressMu.Lock()
-	done := f.egressSendersDone
-	f.egressMu.Unlock()
-	if done != nil {
-		<-done
-	}
-}
 func (f *Framer) writeLoop() {
 	defer close(f.writerDone)
 	for {
 			f.egressMu.Unlock()
 			f.completeWrite(req, writeAll(f.w, req.data))
 		case <-f.done:
-			f.waitForEgressSenders()
+			f.egressSenders.Wait()
 			f.egressMu.Lock()
 			for {
 				select {

M internal/usecase/broker/pool_test.go +11 -3


@@ -388,9 +388,17 @@ func TestPoolIdleEviction(t *testing.T) {
388	388		require.NoError(t, err)
389	389		f := <-physicals
390	390		require.NoError(t, s.Close())
391		-	require.Eventually(t, func() bool { clock.mu.Lock(); defer clock.mu.Unlock(); return len(clock.timers) > 0 }, time.Second, time.Millisecond)
392		-	clock.Advance(time.Minute)
393		-	await(t, f.done)
	391	+	// The watcher re-arms its idle timer each loop, so an early Advance can
	392	+	// land before the re-arm. Keep advancing until the eviction is observed.
	393	+	require.Eventually(t, func() bool {
	394	+	clock.Advance(time.Minute)
	395	+	select {
	396	+	case <-f.done:
	397	+	return true
	398	+	default:
	399	+	return false
	400	+	}
	401	+	}, 3*time.Second, time.Millisecond)
394	402		require.NoError(t, p.Close())
395	403		}
396	404

 	require.NoError(t, err)
 	f := <-physicals
 	require.NoError(t, s.Close())
-	require.Eventually(t, func() bool { clock.mu.Lock(); defer clock.mu.Unlock(); return len(clock.timers) > 0 }, time.Second, time.Millisecond)
+	// The watcher re-arms its idle timer each loop, so an early Advance can
-	clock.Advance(time.Minute)
+	// land before the re-arm. Keep advancing until the eviction is observed.
-	await(t, f.done)
+	require.Eventually(t, func() bool {
+		clock.Advance(time.Minute)
+		select {
+		case <-f.done:
+			return true
+		default:
+			return false
+		}
+	}, 3*time.Second, time.Millisecond)
 	require.NoError(t, p.Close())
 }