1package dgram
2
3import (
4 "errors"
5 "net"
6 "slices"
7 "time"
8
9 pdgram "github.com/bnema/vev/pkg/dgram"
10)
11
12func (t *Transport) retransmitLoop() {
13 resend := t.clock.NewTimer(t.resendAfter)
14 defer resend.Stop()
15 for {
16 select {
17 case <-resend.C():
18 t.queueRetransmits()
19 resend.Reset(t.resendAfter)
20 t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
21 case <-t.done:
22 return
23 }
24 }
25}
26
27func (t *Transport) dataSendLoop() {
28 pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
29 t.mu.Lock()
30 hook := t.afterDataPaceWait
31 t.mu.Unlock()
32 if hook != nil {
33 hook(deadline)
34 }
35 }}
36 for {
37 select {
38 case job := <-t.dataSend:
39 t.runDataSendJob(&pacer, job, true)
40 case batch := <-t.retransmitWork:
41 t.runRetransmitBatch(&pacer, batch)
42 case <-t.done:
43 return
44 }
45 }
46}
47
48func (t *Transport) runDataSendJob(pacer *bytePacer, job dataSendJob, initial bool) {
49 if initial {
50 t.markPendingSent(job.seq, job.reliable)
51 }
52 _, err := t.writeDataJob(pacer, job, nil)
53 if err != nil {
54 if initial {
55 t.removePending(job.seq, job.reliable)
56 } else {
57 t.markPendingReady(job.seq, job.reliable)
58 }
59 if !errors.Is(err, errPacerClosed) {
60 t.closeWithError(err)
61 }
62 } else {
63 t.markPendingReady(job.seq, job.reliable)
64 }
65 t.mu.Lock()
66 afterDataJob := t.afterDataJob
67 t.mu.Unlock()
68 if afterDataJob != nil {
69 afterDataJob()
70 }
71 if job.done != nil {
72 if errors.Is(err, errPacerClosed) {
73 err = t.closedError()
74 }
75 job.done <- err
76 close(job.done)
77 }
78}
79
80func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
81 if job.reliable && !t.pendingExists(job.seq) {
82 return false, nil
83 }
84 payload := encodeData(job.seq, job.reliable, job.frame)
85 frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
86 if err != nil {
87 return false, err
88 }
89 t.mu.Lock()
90 beforeDataPace := t.beforeDataPace
91 t.mu.Unlock()
92 if beforeDataPace != nil {
93 beforeDataPace()
94 }
95 limits := func() (int, int) {
96 t.mu.Lock()
97 defer t.mu.Unlock()
98 return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
99 }
100 emitted := false
101 wireBytes := 0
102 for i, frag := range frags {
103 if job.reliable && !t.pendingExists(job.seq) {
104 return emitted, nil
105 }
106 raw, err := pdgram.MarshalFragment(frag)
107 if err != nil {
108 return emitted, err
109 }
110 pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
111 wireBytes += len(pkt)
112 if job.reliable {
113 t.mu.Lock()
114 if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
115 pending.wireBytes = wireBytes
116 }
117 t.mu.Unlock()
118 }
119 if err := pacer.wait(t.done, len(pkt), limits); err != nil {
120 return emitted, err
121 }
122 if job.reliable && !t.pendingExists(job.seq) {
123 return emitted, nil
124 }
125 t.mu.Lock()
126 peer := t.peer
127 closed := t.closed
128 closeErr := t.closeErr
129 t.mu.Unlock()
130 if closed {
131 if closeErr != nil {
132 return emitted, closeErr
133 }
134 return emitted, net.ErrClosed
135 }
136 if peer == nil {
137 return emitted, errors.New("dgram: no peer")
138 }
139 if i == len(frags)-1 {
140 t.markPendingFinalWrite(job.seq)
141 }
142 if err := t.writePacket(pkt, peer); err != nil {
143 return emitted, err
144 }
145 if !emitted {
146 emitted = true
147 if onFirstWrite != nil {
148 onFirstWrite()
149 }
150 }
151 }
152 return emitted, nil
153}
154
155func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
156 defer func() {
157 t.mu.Lock()
158 hook := t.afterDataJob
159 t.mu.Unlock()
160 if hook != nil {
161 hook()
162 }
163 }()
164 lossApplied := false
165 onFirstWrite := func() {
166 if lossApplied {
167 return
168 }
169 t.mu.Lock()
170 t.congestion.onLoss()
171 t.mu.Unlock()
172 lossApplied = true
173 }
174 for _, r := range batch {
175 job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
176 _, err := t.writeDataJob(pacer, job, onFirstWrite)
177 if err == nil {
178 t.markPendingReady(r.seq, true)
179 continue
180 }
181 t.markPendingReady(r.seq, true)
182 return
183 }
184}
185
186func (t *Transport) queueRetransmits() {
187 // Do not mark records in-flight while the one-slot sender queue is full.
188 // There is one producer and the sender only consumes, so capacity cannot be
189 // lost after this check.
190 if len(t.retransmitWork) == cap(t.retransmitWork) {
191 return
192 }
193
194 now := t.clock.Now()
195 t.mu.Lock()
196 resend, emitDiagnostic := t.selectRetransmitsLocked(now)
197 t.mu.Unlock()
198
199 if len(resend) > 0 {
200 select {
201 case t.retransmitWork <- resend:
202 case <-t.done:
203 return
204 }
205 }
206 if emitDiagnostic {
207 t.emitDiagnostic()
208 }
209}
210
211// selectRetransmitsLocked deterministically reserves one bounded retransmit
212// turn. The caller must hold t.mu and performs all pacing and writes after
213// releasing it.
214func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
215 resend := make([]retransmitRecord, 0, t.maxResendPerTick)
216 seqs := make([]uint64, 0, len(t.pending))
217 for seq := range t.pending {
218 seqs = append(seqs, seq)
219 }
220 slices.Sort(seqs)
221 limit := t.maxResendPerTick
222 emitDiagnostic := false
223 for _, seq := range seqs {
224 if limit <= 0 {
225 break
226 }
227 p := t.pending[seq]
228 if p == nil || p.last.IsZero() || p.initialInFlight || now.Sub(p.last) < t.resendDelayLocked(p) {
229 continue
230 }
231 p.initialInFlight = true
232 p.attempts++
233 p.retransmitted = true
234 t.retransmits++
235 emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
236 resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
237 limit--
238 }
239 return resend, emitDiagnostic
240}
241
242// resendPending remains a test hook for retransmission selection. Constructed
243// transports transfer its batch to the sole data sender; small unit fixtures
244// without transport goroutines execute the selected writes inline.
245func (t *Transport) resendPending() {
246 now := t.clock.Now()
247 t.mu.Lock()
248 resend, _ := t.selectRetransmitsLocked(now)
249 t.mu.Unlock()
250 if t.retransmitWork != nil {
251 select {
252 case t.retransmitWork <- resend:
253 case <-t.done:
254 }
255 return
256 }
257 pacer := bytePacer{clk: t.clock}
258 t.runRetransmitBatch(&pacer, resend)
259}
260
261func (t *Transport) resendDelayLocked(p *pending) time.Duration {
262 d := t.rto
263 if d <= 0 {
264 d = t.resendAfter
265 }
266 for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
267 d *= 2
268 if d > t.maxResendAfter {
269 return t.maxResendAfter
270 }
271 }
272 return d
273}
274
275func (t *Transport) updateRTTLocked(sample time.Duration) {
276 if sample <= 0 {
277 return
278 }
279 if t.srtt <= 0 {
280 t.srtt = sample
281 t.rttvar = sample / 2
282 } else {
283 delta := t.srtt - sample
284 if delta < 0 {
285 delta = -delta
286 }
287 t.rttvar = (3*t.rttvar + delta) / 4
288 t.srtt = (7*t.srtt + sample) / 8
289 }
290 variation := 4 * t.rttvar
291 if variation < time.Millisecond {
292 variation = time.Millisecond
293 }
294 rto := t.srtt + variation
295 if rto < t.resendAfter {
296 rto = t.resendAfter
297 }
298 if rto > t.maxResendAfter {
299 rto = t.maxResendAfter
300 }
301 t.rto = rto
302 t.congestion.onRTT(t.srtt)
303}