d25aab8 fix(dgram): gate resend until first send completes · vev

M internal/adapters/dgram/output_pacer.go +13 -0


@@ -108,6 +108,7 @@ func (t *Transport) sendQueuedData(q queuedSend) error {
108	108		t.removePending(q.seq, q.reliable)
109	109		return err
110	110		}
	111	+	t.markPendingReady(q.seq, q.reliable)
111	112		return nil
112	113		}
113	114

@@ -120,6 +121,18 @@ func (t *Transport) markPendingSent(seq uint64, reliable bool) {
120	121		if p := t.pending[seq]; p != nil && p.first.IsZero() {
121	122		p.first = now
122	123		p.last = now
	124	+	p.initialInFlight = true
	125	+	}
	126	+	t.mu.Unlock()
	127	+	}
	128	+
	129	+	func (t *Transport) markPendingReady(seq uint64, reliable bool) {
	130	+	if !reliable {
	131	+	return
	132	+	}
	133	+	t.mu.Lock()
	134	+	if p := t.pending[seq]; p != nil {
	135	+	p.initialInFlight = false
123	136		}
124	137		t.mu.Unlock()
125	138		}


@@ -108,6 +108,7 @@ func (t *Transport) sendQueuedData(q queuedSend) error {
108	t.removePending(q.seq, q.reliable)	108		t.removePending(q.seq, q.reliable)
109	return err	109		return err
110	}	110		}
		111	+	t.markPendingReady(q.seq, q.reliable)
111	return nil	112		return nil
112	}	113		}
113		114

@@ -120,6 +121,18 @@ func (t *Transport) markPendingSent(seq uint64, reliable bool) {
120	if p := t.pending[seq]; p != nil && p.first.IsZero() {	121		if p := t.pending[seq]; p != nil && p.first.IsZero() {
121	p.first = now	122		p.first = now
122	p.last = now	123		p.last = now
		124	+	p.initialInFlight = true
		125	+	}
		126	+	t.mu.Unlock()
		127	+	}
		128	+
		129	+	func (t *Transport) markPendingReady(seq uint64, reliable bool) {
		130	+	if !reliable {
		131	+	return
		132	+	}
		133	+	t.mu.Lock()
		134	+	if p := t.pending[seq]; p != nil {
		135	+	p.initialInFlight = false
123	}	136		}
124	t.mu.Unlock()	137		t.mu.Unlock()
125	}	138		}

M internal/adapters/dgram/transport.go +9 -6


@@ -189,11 +189,12 @@ type Transport struct {
189	189		}
190	190
191	191		type pending struct {
192		-	frame ports.Frame
193		-	first time.Time
194		-	last time.Time
195		-	attempts int
196		-	retransmitted bool
	192	+	frame ports.Frame
	193	+	first time.Time
	194	+	last time.Time
	195	+	initialInFlight bool
	196	+	attempts int
	197	+	retransmitted bool
197	198		}
198	199
199	200		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {

@@ -282,6 +283,7 @@ func (t *Transport) Send(f ports.Frame) error {
282	283		t.removePending(seq, reliable)
283	284		return err
284	285		}
	286	+	t.markPendingReady(seq, reliable)
285	287		return nil
286	288		}
287	289		queued := t.flushQueuedOutputLocked()

@@ -298,6 +300,7 @@ func (t *Transport) Send(f ports.Frame) error {
298	300		t.removePending(seq, reliable)
299	301		return err
300	302		}
	303	+	t.markPendingReady(seq, reliable)
301	304		return nil
302	305		}
303	306

@@ -776,7 +779,7 @@ func (t *Transport) resendPending() {
776	779		break
777	780		}
778	781		p := t.pending[seq]
779		-	if p == nil \|\| p.last.IsZero() \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
	782	+	if p == nil \|\| p.last.IsZero() \|\| p.initialInFlight \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
780	783		continue
781	784		}
782	785		p.last = now

 }
 type pending struct {
-	frame         ports.Frame
+	frame           ports.Frame
-	first         time.Time
+	first           time.Time
-	last          time.Time
+	last            time.Time
-	attempts      int
+	initialInFlight bool
-	retransmitted bool
+	attempts        int
+	retransmitted   bool
 }
 func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
 			t.removePending(seq, reliable)
 			return err
 		}
+		t.markPendingReady(seq, reliable)
 		return nil
 	}
 	queued := t.flushQueuedOutputLocked()
 		t.removePending(seq, reliable)
 		return err
 	}
+	t.markPendingReady(seq, reliable)
 	return nil
 }
 			break
 		}
 		p := t.pending[seq]
-		if p == nil || p.last.IsZero() || now.Sub(p.last) < t.resendDelayLocked(p) {
+		if p == nil || p.last.IsZero() || p.initialInFlight || now.Sub(p.last) < t.resendDelayLocked(p) {
 			continue
 		}
 		p.last = now

M internal/adapters/dgram/transport_test.go +41 -30


@@ -171,12 +171,11 @@ func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
171	171		type manualClock struct {
172	172		mu sync.Mutex
173	173		now time.Time
174		-	timers chan *manualTimer
175		-	active int
	174	+	timers map[*manualTimer]struct{}
176	175		}
177	176
178	177		func newManualClock(now time.Time) *manualClock {
179		-	return &manualClock{now: now, timers: make(chan *manualTimer, 16)}
	178	+	return &manualClock{now: now, timers: make(map[*manualTimer]struct{})}
180	179		}
181	180
182	181		func (c *manualClock) Now() time.Time {

@@ -187,11 +186,9 @@ func (c *manualClock) Now() time.Time {
187	186
188	187		func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
189	188		c.mu.Lock()
190		-	deadline := c.now.Add(d)
191		-	c.active++
	189	+	tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
	190	+	c.timers[tm] = struct{}{}
192	191		c.mu.Unlock()
193		-	tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: deadline, active: true}
194		-	c.timers <- tm
195	192		return tm
196	193		}
197	194

@@ -199,14 +196,13 @@ func (c *manualClock) advance(d time.Duration) {
199	196		c.mu.Lock()
200	197		c.now = c.now.Add(d)
201	198		now := c.now
	199	+	timers := make([]*manualTimer, 0, len(c.timers))
	200	+	for tm := range c.timers {
	201	+	timers = append(timers, tm)
	202	+	}
202	203		c.mu.Unlock()
203		-	n := len(c.timers)
204		-	for range n {
205		-	tm := <-c.timers
206		-	if fired, keep := tm.fireIfDue(now); fired \|\| !keep {
207		-	continue
208		-	}
209		-	c.timers <- tm
	204	+	for _, tm := range timers {
	205	+	tm.fireIfDue(now)
210	206		}
211	207		}
212	208

@@ -223,13 +219,8 @@ func (t *manualTimer) Reset(d time.Duration) bool {
223	219		wasActive := t.active
224	220		t.deadline = t.clock.now.Add(d)
225	221		t.active = true
226		-	if !wasActive {
227		-	t.clock.active++
228		-	}
	222	+	t.clock.timers[t] = struct{}{}
229	223		t.clock.mu.Unlock()
230		-	if !wasActive {
231		-	t.clock.timers <- t
232		-	}
233	224		return wasActive
234	225		}
235	226		func (t *manualTimer) Stop() bool {

@@ -237,29 +228,28 @@ func (t *manualTimer) Stop() bool {
237	228		defer t.clock.mu.Unlock()
238	229		wasActive := t.active
239	230		t.active = false
240		-	if wasActive {
241		-	t.clock.active--
242		-	}
	231	+	delete(t.clock.timers, t)
243	232		return wasActive
244	233		}
245		-	func (t *manualTimer) fireIfDue(now time.Time) (fired bool, keep bool) {
	234	+	func (t *manualTimer) fireIfDue(now time.Time) bool {
246	235		t.clock.mu.Lock()
247	236		if !t.active {
	237	+	delete(t.clock.timers, t)
248	238		t.clock.mu.Unlock()
249		-	return false, false
	239	+	return false
250	240		}
251	241		if now.Before(t.deadline) {
252	242		t.clock.mu.Unlock()
253		-	return false, true
	243	+	return false
254	244		}
255	245		t.active = false
256		-	t.clock.active--
	246	+	delete(t.clock.timers, t)
257	247		t.clock.mu.Unlock()
258	248		select {
259	249		case t.c <- now:
260	250		default:
261	251		}
262		-	return true, false
	252	+	return true
263	253		}
264	254
265	255		func waitForManualTimers(t testing.T, clk manualClock, n int) {

@@ -267,7 +257,7 @@ func waitForManualTimers(t testing.T, clk manualClock, n int) {
267	257		deadline := time.Now().Add(time.Second)
268	258		for time.Now().Before(deadline) {
269	259		clk.mu.Lock()
270		-	active := clk.active
	260	+	active := len(clk.timers)
271	261		clk.mu.Unlock()
272	262		if active >= n {
273	263		return

@@ -275,7 +265,7 @@ func waitForManualTimers(t testing.T, clk manualClock, n int) {
275	265		time.Sleep(time.Millisecond)
276	266		}
277	267		clk.mu.Lock()
278		-	active := clk.active
	268	+	active := len(clk.timers)
279	269		clk.mu.Unlock()
280	270		t.Fatalf("manual timers=%d, want at least %d", active, n)
281	271		}

@@ -353,6 +343,27 @@ func TestResendSkipsQueuedOutputBeforeFirstWireAttempt(t *testing.T) {
353	343		}
354	344		}
355	345
	346	+	func TestResendSkipsOutputDuringFirstWireAttempt(t *testing.T) {
	347	+	tr, writes := newResendTestTransport(t, 10)
	348	+	now := tr.clock.Now()
	349	+	tr.pending[1] = &pending{
	350	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("sending")},
	351	+	first: now.Add(-time.Second),
	352	+	last: now.Add(-time.Second),
	353	+	initialInFlight: true,
	354	+	}
	355	+
	356	+	tr.resendPending()
	357	+	if got := writes.Load(); got != 0 {
	358	+	t.Fatalf("writes=%d, want first wire attempt excluded from resend", got)
	359	+	}
	360	+	tr.pending[1].initialInFlight = false
	361	+	tr.resendPending()
	362	+	if got := writes.Load(); got != 1 {
	363	+	t.Fatalf("writes=%d, want resend after first wire attempt completed", got)
	364	+	}
	365	+	}
	366	+
356	367		func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
357	368		tr, _ := newResendTestTransport(t, 10)
358	369		tests := []struct {

 type manualClock struct {
 	mu     sync.Mutex
 	now    time.Time
-	timers chan *manualTimer
+	timers map[*manualTimer]struct{}
-	active int
 }
 func newManualClock(now time.Time) *manualClock {
-	return &manualClock{now: now, timers: make(chan *manualTimer, 16)}
+	return &manualClock{now: now, timers: make(map[*manualTimer]struct{})}
 }
 func (c *manualClock) Now() time.Time {
 func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
 	c.mu.Lock()
-	deadline := c.now.Add(d)
+	tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
-	c.active++
+	c.timers[tm] = struct{}{}
 	c.mu.Unlock()
-	tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: deadline, active: true}
-	c.timers <- tm
 	return tm
 }
 	c.mu.Lock()
 	c.now = c.now.Add(d)
 	now := c.now
+	timers := make([]*manualTimer, 0, len(c.timers))
+	for tm := range c.timers {
+		timers = append(timers, tm)
+	}
 	c.mu.Unlock()
-	n := len(c.timers)
+	for _, tm := range timers {
-	for range n {
+		tm.fireIfDue(now)
-		tm := <-c.timers
-		if fired, keep := tm.fireIfDue(now); fired || !keep {
-			continue
-		}
-		c.timers <- tm
 	}
 }
 	wasActive := t.active
 	t.deadline = t.clock.now.Add(d)
 	t.active = true
-	if !wasActive {
+	t.clock.timers[t] = struct{}{}
-		t.clock.active++
-	}
 	t.clock.mu.Unlock()
-	if !wasActive {
-		t.clock.timers <- t
-	}
 	return wasActive
 }
 func (t *manualTimer) Stop() bool {
 	defer t.clock.mu.Unlock()
 	wasActive := t.active
 	t.active = false
-	if wasActive {
+	delete(t.clock.timers, t)
-		t.clock.active--
-	}
 	return wasActive
 }
-func (t *manualTimer) fireIfDue(now time.Time) (fired bool, keep bool) {
+func (t *manualTimer) fireIfDue(now time.Time) bool {
 	t.clock.mu.Lock()
 	if !t.active {
+		delete(t.clock.timers, t)
 		t.clock.mu.Unlock()
-		return false, false
+		return false
 	}
 	if now.Before(t.deadline) {
 		t.clock.mu.Unlock()
-		return false, true
+		return false
 	}
 	t.active = false
-	t.clock.active--
+	delete(t.clock.timers, t)
 	t.clock.mu.Unlock()
 	select {
 	case t.c <- now:
 	default:
 	}
-	return true, false
+	return true
 }
 func waitForManualTimers(t *testing.T, clk *manualClock, n int) {
 	deadline := time.Now().Add(time.Second)
 	for time.Now().Before(deadline) {
 		clk.mu.Lock()
-		active := clk.active
+		active := len(clk.timers)
 		clk.mu.Unlock()
 		if active >= n {
 			return
 		time.Sleep(time.Millisecond)
 	}
 	clk.mu.Lock()
-	active := clk.active
+	active := len(clk.timers)
 	clk.mu.Unlock()
 	t.Fatalf("manual timers=%d, want at least %d", active, n)
 }
 	}
 }
+func TestResendSkipsOutputDuringFirstWireAttempt(t *testing.T) {
+	tr, writes := newResendTestTransport(t, 10)
+	now := tr.clock.Now()
+	tr.pending[1] = &pending{
+		frame:           ports.Frame{Type: ports.MsgOutput, Payload: []byte("sending")},
+		first:           now.Add(-time.Second),
+		last:            now.Add(-time.Second),
+		initialInFlight: true,
+	}
+	tr.resendPending()
+	if got := writes.Load(); got != 0 {
+		t.Fatalf("writes=%d, want first wire attempt excluded from resend", got)
+	}
+	tr.pending[1].initialInFlight = false
+	tr.resendPending()
+	if got := writes.Load(); got != 1 {
+		t.Fatalf("writes=%d, want resend after first wire attempt completed", got)
+	}
+}
 func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
 	tr, _ := newResendTestTransport(t, 10)
 	tests := []struct {