fix(dgram): move UDP control writes off receive loop

commit: ba81cef3e8a28345bfa33eb363703d43415c844c
author: brice · 3mo ago · 2026-07-10 00:31 UTC
parent: eac5808
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +139 -27

A internal/adapters/dgram/control.go +112 -0

@@ -0,0 +1,112 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"encoding/binary"
	5	+	"runtime"
	6	+
	7	+	pdgram "github.com/bnema/vev/pkg/dgram"
	8	+	)
	9	+
	10	+	const controlQueueSize = 16
	11	+
	12	+	type controlRecord struct {
	13	+	kind byte
	14	+	id uint64
	15	+	}
	16	+
	17	+	// queueACK retains the greatest cumulative acknowledgement without ever making
	18	+	// packet receipt wait for a socket write.
	19	+	func (t *Transport) queueACK(seq uint64) {
	20	+	t.controlMu.Lock()
	21	+	if !t.ackQueued \|\| seq > t.ackSeq {
	22	+	t.ackSeq = seq
	23	+	t.ackQueued = true
	24	+	}
	25	+	t.controlMu.Unlock()
	26	+	if t.ackWake == nil {
	27	+	// Small unit-test transports may exercise record handling without the
	28	+	// goroutines installed by NewTransportWithOptions.
	29	+	go t.sendAck(seq)
	30	+	return
	31	+	}
	32	+	select {
	33	+	case t.ackWake <- struct{}{}:
	34	+	default:
	35	+	}
	36	+	// Let the dedicated writer claim queued control before delivery wakes a
	37	+	// consumer; this remains a non-blocking receive-path operation.
	38	+	runtime.Gosched()
	39	+	}
	40	+
	41	+	// queueControl is deliberately lossy when the bounded control queue is full:
	42	+	// probes and pongs are advisory and must not stall authenticated receipt.
	43	+	func (t *Transport) queueControl(kind byte, id uint64) bool {
	44	+	select {
	45	+	case t.control <- controlRecord{kind: kind, id: id}:
	46	+	return true
	47	+	default:
	48	+	return false
	49	+	}
	50	+	}
	51	+
	52	+	func (t *Transport) takeACK() (uint64, bool) {
	53	+	t.controlMu.Lock()
	54	+	defer t.controlMu.Unlock()
	55	+	seq, ok := t.ackSeq, t.ackQueued
	56	+	t.ackSeq = 0
	57	+	t.ackQueued = false
	58	+	return seq, ok
	59	+	}
	60	+
	61	+	func (t *Transport) controlLoop() {
	62	+	for {
	63	+	// Probes and pongs always win over an ACK that is ready to send.
	64	+	select {
	65	+	case record := <-t.control:
	66	+	t.sendControl(record.kind, record.id)
	67	+	continue
	68	+	default:
	69	+	}
	70	+
	71	+	select {
	72	+	case record := <-t.control:
	73	+	t.sendControl(record.kind, record.id)
	74	+	case <-t.ackWake:
	75	+	if seq, ok := t.takeACK(); ok {
	76	+	t.sendControl(recAck, seq)
	77	+	}
	78	+	case <-t.done:
	79	+	return
	80	+	}
	81	+	}
	82	+	}
	83	+
	84	+	func (t *Transport) sendControl(kind byte, id uint64) {
	85	+	var payload [9]byte
	86	+	payload[0] = kind
	87	+	binary.BigEndian.PutUint64(payload[1:], id)
	88	+	frags, err := pdgram.FragmentPayload(t.nextCounter(), payload[:], t.mtu-pdgram.HeaderSize-t.codec.Overhead())
	89	+	if err != nil {
	90	+	return
	91	+	}
	92	+	for _, frag := range frags {
	93	+	raw, err := pdgram.MarshalFragment(frag)
	94	+	if err != nil {
	95	+	return
	96	+	}
	97	+	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
	98	+	t.mu.Lock()
	99	+	pc := t.pc
	100	+	peer := t.peer
	101	+	closed := t.closed
	102	+	t.mu.Unlock()
	103	+	if closed \|\| peer == nil {
	104	+	return
	105	+	}
	106	+	// PacketConn permits concurrent callers. Control traffic deliberately
	107	+	// bypasses data's write lock so a retransmit cannot starve heartbeats.
	108	+	if _, err := pc.WriteTo(pkt, peer); err != nil {
	109	+	return
	110	+	}
	111	+	}
	112	+	}

@@ -0,0 +1,112 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"encoding/binary"
	5	+	"runtime"
	6	+
	7	+	pdgram "github.com/bnema/vev/pkg/dgram"
	8	+	)
	9	+
	10	+	const controlQueueSize = 16
	11	+
	12	+	type controlRecord struct {
	13	+	kind byte
	14	+	id uint64
	15	+	}
	16	+
	17	+	// queueACK retains the greatest cumulative acknowledgement without ever making
	18	+	// packet receipt wait for a socket write.
	19	+	func (t *Transport) queueACK(seq uint64) {
	20	+	t.controlMu.Lock()
	21	+	if !t.ackQueued \|\| seq > t.ackSeq {
	22	+	t.ackSeq = seq
	23	+	t.ackQueued = true
	24	+	}
	25	+	t.controlMu.Unlock()
	26	+	if t.ackWake == nil {
	27	+	// Small unit-test transports may exercise record handling without the
	28	+	// goroutines installed by NewTransportWithOptions.
	29	+	go t.sendAck(seq)
	30	+	return
	31	+	}
	32	+	select {
	33	+	case t.ackWake <- struct{}{}:
	34	+	default:
	35	+	}
	36	+	// Let the dedicated writer claim queued control before delivery wakes a
	37	+	// consumer; this remains a non-blocking receive-path operation.
	38	+	runtime.Gosched()
	39	+	}
	40	+
	41	+	// queueControl is deliberately lossy when the bounded control queue is full:
	42	+	// probes and pongs are advisory and must not stall authenticated receipt.
	43	+	func (t *Transport) queueControl(kind byte, id uint64) bool {
	44	+	select {
	45	+	case t.control <- controlRecord{kind: kind, id: id}:
	46	+	return true
	47	+	default:
	48	+	return false
	49	+	}
	50	+	}
	51	+
	52	+	func (t *Transport) takeACK() (uint64, bool) {
	53	+	t.controlMu.Lock()
	54	+	defer t.controlMu.Unlock()
	55	+	seq, ok := t.ackSeq, t.ackQueued
	56	+	t.ackSeq = 0
	57	+	t.ackQueued = false
	58	+	return seq, ok
	59	+	}
	60	+
	61	+	func (t *Transport) controlLoop() {
	62	+	for {
	63	+	// Probes and pongs always win over an ACK that is ready to send.
	64	+	select {
	65	+	case record := <-t.control:
	66	+	t.sendControl(record.kind, record.id)
	67	+	continue
	68	+	default:
	69	+	}
	70	+
	71	+	select {
	72	+	case record := <-t.control:
	73	+	t.sendControl(record.kind, record.id)
	74	+	case <-t.ackWake:
	75	+	if seq, ok := t.takeACK(); ok {
	76	+	t.sendControl(recAck, seq)
	77	+	}
	78	+	case <-t.done:
	79	+	return
	80	+	}
	81	+	}
	82	+	}
	83	+
	84	+	func (t *Transport) sendControl(kind byte, id uint64) {
	85	+	var payload [9]byte
	86	+	payload[0] = kind
	87	+	binary.BigEndian.PutUint64(payload[1:], id)
	88	+	frags, err := pdgram.FragmentPayload(t.nextCounter(), payload[:], t.mtu-pdgram.HeaderSize-t.codec.Overhead())
	89	+	if err != nil {
	90	+	return
	91	+	}
	92	+	for _, frag := range frags {
	93	+	raw, err := pdgram.MarshalFragment(frag)
	94	+	if err != nil {
	95	+	return
	96	+	}
	97	+	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
	98	+	t.mu.Lock()
	99	+	pc := t.pc
	100	+	peer := t.peer
	101	+	closed := t.closed
	102	+	t.mu.Unlock()
	103	+	if closed \|\| peer == nil {
	104	+	return
	105	+	}
	106	+	// PacketConn permits concurrent callers. Control traffic deliberately
	107	+	// bypasses data's write lock so a retransmit cannot starve heartbeats.
	108	+	if _, err := pc.WriteTo(pkt, peer); err != nil {
	109	+	return
	110	+	}
	111	+	}
	112	+	}

M internal/adapters/dgram/transport.go +24 -25


@@ -41,7 +41,7 @@ const (
41	41		dataRecordHeaderSize = 12
42	42		maxRecvBuffer = defaultMaxRecvBuf
43	43		linkEventBufferSize = 16
44		-	probeReplyBufferSize = 16
	44	+	probeReplyBufferSize = controlQueueSize // legacy test burst bound
45	45		)
46	46
47	47		var (

@@ -181,7 +181,11 @@ type Transport struct {
181	181		linkEvents chan ports.LinkEvent
182	182		probeWait map[uint64]chan struct{}
183	183		sendWake chan struct{}
184		-	probeReply chan uint64
	184	+	control chan controlRecord
	185	+	ackWake chan struct{}
	186	+	controlMu sync.Mutex
	187	+	ackSeq uint64
	188	+	ackQueued bool
185	189		rebind func(net.PacketConn) (net.PacketConn, error)
186	190		hoppedOffline bool
187	191		outputQueue []queuedSend

@@ -269,17 +273,19 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
269	273		recvBuf: make(map[uint64]ports.Frame),
270	274		}
271	275		t.sendWake = make(chan struct{})
	276	+	t.control = make(chan controlRecord, controlQueueSize)
	277	+	t.ackWake = make(chan struct{}, 1)
272	278		t.outputWake = make(chan struct{})
273	279		if t.observe != nil {
274	280		t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
275	281		go t.diagnosticLoop()
276	282		}
277		-	t.probeReply = make(chan uint64, probeReplyBufferSize)
278	283		t.deliverCond = sync.NewCond(&t.deliverMu)
279	284		go t.readLoop(pc)
280	285		go t.resendLoop()
	286	+	go t.heartbeatLoop()
281	287		go t.deliveryLoop()
282		-	go t.probeReplyLoop()
	288	+	go t.controlLoop()
283	289		go t.outputPaceLoop()
284	290		t.emitDiagnostic()
285	291		return t, nil

@@ -777,10 +783,7 @@ func (t *Transport) handleRecord(p []byte) {
777	783		if len(p) != 9 {
778	784		return
779	785		}
780		-	select {
781		-	case t.probeReply <- binary.BigEndian.Uint64(p[1:]):
782		-	default:
783		-	}
	786	+	t.queueControl(recPong, binary.BigEndian.Uint64(p[1:]))
784	787		case recPong:
785	788		if len(p) != 9 {
786	789		return

@@ -801,7 +804,7 @@ func (t *Transport) handleRecord(p []byte) {
801	804		if reliable {
802	805		ackSeq, ack, queued := t.enqueueReliable(seq, f)
803	806		if ack {
804		-	t.sendAck(ackSeq)
	807	+	t.queueACK(ackSeq)
805	808		}
806	809		if queued {
807	810		t.signalDelivery()

@@ -878,17 +881,6 @@ func (t *Transport) deliver(f ports.Frame) {
878	881		}
879	882		}
880	883
881		-	func (t *Transport) probeReplyLoop() {
882		-	for {
883		-	select {
884		-	case id := <-t.probeReply:
885		-	_ = t.sendProbe(recPong, id)
886		-	case <-t.done:
887		-	return
888		-	}
889		-	}
890		-	}
891		-
892	884		func (t *Transport) notifySendWaitersLocked() {
893	885		if t.sendWake == nil {
894	886		return

@@ -900,8 +892,6 @@ func (t *Transport) notifySendWaitersLocked() {
900	892		func (t *Transport) resendLoop() {
901	893		resend := t.clock.NewTimer(t.resendAfter)
902	894		defer resend.Stop()
903		-	heartbeat := t.clock.NewTimer(t.heartbeat)
904		-	defer heartbeat.Stop()
905	895		for {
906	896		select {
907	897		case <-resend.C():

@@ -909,10 +899,19 @@ func (t *Transport) resendLoop() {
909	899		t.checkSilence()
910	900		t.checkSendStall()
911	901		resend.Reset(t.resendAfter)
	902	+	case <-t.done:
	903	+	return
	904	+	}
	905	+	}
	906	+	}
	907	+
	908	+	func (t *Transport) heartbeatLoop() {
	909	+	heartbeat := t.clock.NewTimer(t.heartbeat)
	910	+	defer heartbeat.Stop()
	911	+	for {
	912	+	select {
912	913		case <-heartbeat.C():
913		-	_ = t.sendProbe(recProbe, 0)
914		-	t.checkSilence()
915		-	t.checkSendStall()
	914	+	t.queueControl(recProbe, 0)
916	915		heartbeat.Reset(t.heartbeat)
917	916		case <-t.done:
918	917		return

 	dataRecordHeaderSize        = 12
 	maxRecvBuffer               = defaultMaxRecvBuf
 	linkEventBufferSize         = 16
-	probeReplyBufferSize        = 16
+	probeReplyBufferSize        = controlQueueSize // legacy test burst bound
 )
 var (
 	linkEvents             chan ports.LinkEvent
 	probeWait              map[uint64]chan struct{}
 	sendWake               chan struct{}
-	probeReply             chan uint64
+	control                chan controlRecord
+	ackWake                chan struct{}
+	controlMu              sync.Mutex
+	ackSeq                 uint64
+	ackQueued              bool
 	rebind                 func(net.PacketConn) (net.PacketConn, error)
 	hoppedOffline          bool
 	outputQueue            []queuedSend
 		recvBuf:                 make(map[uint64]ports.Frame),
 	}
 	t.sendWake = make(chan struct{})
+	t.control = make(chan controlRecord, controlQueueSize)
+	t.ackWake = make(chan struct{}, 1)
 	t.outputWake = make(chan struct{})
 	if t.observe != nil {
 		t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
 		go t.diagnosticLoop()
 	}
-	t.probeReply = make(chan uint64, probeReplyBufferSize)
 	t.deliverCond = sync.NewCond(&t.deliverMu)
 	go t.readLoop(pc)
 	go t.resendLoop()
+	go t.heartbeatLoop()
 	go t.deliveryLoop()
-	go t.probeReplyLoop()
+	go t.controlLoop()
 	go t.outputPaceLoop()
 	t.emitDiagnostic()
 	return t, nil
 		if len(p) != 9 {
 			return
 		}
-		select {
+		t.queueControl(recPong, binary.BigEndian.Uint64(p[1:]))
-		case t.probeReply <- binary.BigEndian.Uint64(p[1:]):
-		default:
-		}
 	case recPong:
 		if len(p) != 9 {
 			return
 		if reliable {
 			ackSeq, ack, queued := t.enqueueReliable(seq, f)
 			if ack {
-				t.sendAck(ackSeq)
+				t.queueACK(ackSeq)
 			}
 			if queued {
 				t.signalDelivery()
 	}
 }
-func (t *Transport) probeReplyLoop() {
-	for {
-		select {
-		case id := <-t.probeReply:
-			_ = t.sendProbe(recPong, id)
-		case <-t.done:
-			return
-		}
-	}
-}
 func (t *Transport) notifySendWaitersLocked() {
 	if t.sendWake == nil {
 		return
 func (t *Transport) resendLoop() {
 	resend := t.clock.NewTimer(t.resendAfter)
 	defer resend.Stop()
-	heartbeat := t.clock.NewTimer(t.heartbeat)
-	defer heartbeat.Stop()
 	for {
 		select {
 		case <-resend.C():
 			t.checkSilence()
 			t.checkSendStall()
 			resend.Reset(t.resendAfter)
+		case <-t.done:
+			return
+		}
+	}
+}
+func (t *Transport) heartbeatLoop() {
+	heartbeat := t.clock.NewTimer(t.heartbeat)
+	defer heartbeat.Stop()
+	for {
+		select {
 		case <-heartbeat.C():
-			_ = t.sendProbe(recProbe, 0)
+			t.queueControl(recProbe, 0)
-			t.checkSilence()
-			t.checkSendStall()
 			heartbeat.Reset(t.heartbeat)
 		case <-t.done:
 			return

M internal/adapters/dgram/transport_test.go +3 -2


@@ -790,9 +790,10 @@ func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {
790	790		t.Fatal(err)
791	791		}
792	792		got := recvWithin(t, b, time.Second)
793		-	if got.Type != ports.MsgInput \|\| string(got.Payload) != "typed" \|\| !droppedAck.Load() {
794		-	t.Fatalf("got=%+v droppedAck=%v", got, droppedAck.Load())
	793	+	if got.Type != ports.MsgInput \|\| string(got.Payload) != "typed" {
	794	+	t.Fatalf("got=%+v", got)
795	795		}
	796	+	eventually(t, time.Second, droppedAck.Load)
796	797		if got, ok := recvMaybe(b, 3*defaultResend); ok {
797	798		t.Fatalf("duplicate reliable frame delivered: %+v", got)
798	799		}

 		t.Fatal(err)
 	}
 	got := recvWithin(t, b, time.Second)
-	if got.Type != ports.MsgInput || string(got.Payload) != "typed" || !droppedAck.Load() {
+	if got.Type != ports.MsgInput || string(got.Payload) != "typed" {
-		t.Fatalf("got=%+v droppedAck=%v", got, droppedAck.Load())
+		t.Fatalf("got=%+v", got)
 	}
+	eventually(t, time.Second, droppedAck.Load)
 	if got, ok := recvMaybe(b, 3*defaultResend); ok {
 		t.Fatalf("duplicate reliable frame delivered: %+v", got)
 	}