1package dgram
2
3import (
4 "encoding/binary"
5 "errors"
6 "net"
7 "sync"
8 "sync/atomic"
9 "testing"
10 "time"
11
12 "github.com/bnema/vev/internal/ports"
13 pdgram "github.com/bnema/vev/pkg/dgram"
14)
15
16type timeoutErr struct{}
17
18func (timeoutErr) Error() string { return "i/o timeout" }
19func (timeoutErr) Timeout() bool { return true }
20
21type blockingWritePC struct {
22 addr net.Addr
23
24 mu sync.Mutex
25 closed bool
26 deadline time.Time
27 done chan struct{}
28}
29
30func newBlockingWritePC() *blockingWritePC {
31 return &blockingWritePC{addr: testAddr("blocked"), done: make(chan struct{})}
32}
33
34func (p *blockingWritePC) ReadFrom([]byte) (int, net.Addr, error) {
35 <-p.done
36 return 0, nil, errors.New("closed")
37}
38
39func (p *blockingWritePC) WriteTo([]byte, net.Addr) (int, error) {
40 for {
41 p.mu.Lock()
42 closed := p.closed
43 deadline := p.deadline
44 p.mu.Unlock()
45 if closed {
46 return 0, errors.New("closed")
47 }
48 if !deadline.IsZero() {
49 wait := time.Until(deadline)
50 if wait > 0 {
51 timer := time.NewTimer(wait)
52 select {
53 case <-timer.C:
54 case <-p.done:
55 timer.Stop()
56 return 0, errors.New("closed")
57 }
58 }
59 return 0, timeoutErr{}
60 }
61 select {
62 case <-time.After(time.Millisecond):
63 case <-p.done:
64 return 0, errors.New("closed")
65 }
66 }
67}
68
69func (p *blockingWritePC) Close() error {
70 p.mu.Lock()
71 defer p.mu.Unlock()
72 if !p.closed {
73 p.closed = true
74 close(p.done)
75 }
76 return nil
77}
78
79func (p *blockingWritePC) LocalAddr() net.Addr { return p.addr }
80func (p *blockingWritePC) SetDeadline(t time.Time) error {
81 p.mu.Lock()
82 p.deadline = t
83 p.mu.Unlock()
84 return nil
85}
86func (p *blockingWritePC) SetReadDeadline(time.Time) error { return nil }
87func (p *blockingWritePC) SetWriteDeadline(t time.Time) error {
88 p.mu.Lock()
89 p.deadline = t
90 p.mu.Unlock()
91 return nil
92}
93
94type fixedClock struct{ now time.Time }
95
96func (c fixedClock) Now() time.Time { return c.now }
97func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
98 return realTimer{t: time.NewTimer(d)}
99}
100
101type manualClock struct {
102 mu sync.Mutex
103 now time.Time
104 timers map[*manualTimer]struct{}
105 timerCreated chan struct{}
106}
107
108func newManualClock(now time.Time) *manualClock {
109 return &manualClock{
110 now: now,
111 timers: make(map[*manualTimer]struct{}),
112 timerCreated: make(chan struct{}, 64),
113 }
114}
115
116func (c *manualClock) Now() time.Time {
117 c.mu.Lock()
118 defer c.mu.Unlock()
119 return c.now
120}
121
122func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
123 c.mu.Lock()
124 tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
125 c.timers[tm] = struct{}{}
126 c.mu.Unlock()
127 select {
128 case c.timerCreated <- struct{}{}:
129 default:
130 }
131 return tm
132}
133
134func (c *manualClock) advance(d time.Duration) {
135 c.mu.Lock()
136 c.now = c.now.Add(d)
137 now := c.now
138 timers := make([]*manualTimer, 0, len(c.timers))
139 for tm := range c.timers {
140 timers = append(timers, tm)
141 }
142 c.mu.Unlock()
143 for _, tm := range timers {
144 tm.fireIfDue(now)
145 }
146}
147
148type manualTimer struct {
149 clock *manualClock
150 c chan time.Time
151 deadline time.Time
152 active bool
153}
154
155func (t *manualTimer) C() <-chan time.Time { return t.c }
156func (t *manualTimer) Reset(d time.Duration) bool {
157 t.clock.mu.Lock()
158 wasActive := t.active
159 t.deadline = t.clock.now.Add(d)
160 t.active = true
161 t.clock.timers[t] = struct{}{}
162 t.clock.mu.Unlock()
163 return wasActive
164}
165func (t *manualTimer) Stop() bool {
166 t.clock.mu.Lock()
167 defer t.clock.mu.Unlock()
168 wasActive := t.active
169 t.active = false
170 delete(t.clock.timers, t)
171 return wasActive
172}
173func (t *manualTimer) fireIfDue(now time.Time) bool {
174 t.clock.mu.Lock()
175 if !t.active {
176 delete(t.clock.timers, t)
177 t.clock.mu.Unlock()
178 return false
179 }
180 if now.Before(t.deadline) {
181 t.clock.mu.Unlock()
182 return false
183 }
184 t.active = false
185 delete(t.clock.timers, t)
186 t.clock.mu.Unlock()
187 select {
188 case t.c <- now:
189 default:
190 }
191 return true
192}
193
194func waitForManualTimers(t *testing.T, clk *manualClock, n int) {
195 t.Helper()
196 deadline := time.Now().Add(time.Second)
197 for time.Now().Before(deadline) {
198 clk.mu.Lock()
199 active := len(clk.timers)
200 clk.mu.Unlock()
201 if active >= n {
202 return
203 }
204 time.Sleep(time.Millisecond)
205 }
206 clk.mu.Lock()
207 active := len(clk.timers)
208 clk.mu.Unlock()
209 t.Fatalf("manual timers=%d, want at least %d", active, n)
210}
211
212type deadlineCapturePC struct {
213 addr net.Addr
214 deadline atomic.Value
215 done chan struct{}
216}
217
218func (p *deadlineCapturePC) ReadFrom([]byte) (int, net.Addr, error) {
219 <-p.done
220 return 0, nil, errors.New("closed")
221}
222func (p *deadlineCapturePC) WriteTo(b []byte, _ net.Addr) (int, error) { return len(b), nil }
223func (p *deadlineCapturePC) Close() error {
224 select {
225 case <-p.done:
226 default:
227 close(p.done)
228 }
229 return nil
230}
231func (p *deadlineCapturePC) LocalAddr() net.Addr { return p.addr }
232func (p *deadlineCapturePC) SetDeadline(time.Time) error { return nil }
233func (p *deadlineCapturePC) SetReadDeadline(time.Time) error { return nil }
234func (p *deadlineCapturePC) SetWriteDeadline(t time.Time) error {
235 if !t.IsZero() {
236 p.deadline.Store(t)
237 }
238 return nil
239}
240
241func eventuallyPacketCount(t *testing.T, pc *fakePC, want int) {
242 t.Helper()
243 deadline := time.Now().Add(time.Second)
244 for time.Now().Before(deadline) {
245 if len(pc.in) >= want {
246 return
247 }
248 time.Sleep(time.Millisecond)
249 }
250 t.Fatalf("datagrams=%d, want at least %d", len(pc.in), want)
251}
252
253func eventually(t *testing.T, timeout time.Duration, condition func() bool) {
254 t.Helper()
255 deadline := time.Now().Add(timeout)
256 for time.Now().Before(deadline) {
257 if condition() {
258 return
259 }
260 time.Sleep(time.Millisecond)
261 }
262 t.Fatal("condition was not met before timeout")
263}
264
265func newResendTestTransport(t *testing.T, maxResendPerTick int) (*Transport, *atomic.Int32) {
266 t.Helper()
267 aPC, _ := newPair()
268 codec, err := pdgram.NewCodec(key())
269 if err != nil {
270 t.Fatal(err)
271 }
272 var writes atomic.Int32
273 aPC.drop = func(_ []byte, addr net.Addr) bool {
274 if addr.String() == "b" {
275 writes.Add(1)
276 return true
277 }
278 return false
279 }
280 return &Transport{
281 pc: aPC,
282 peer: testAddr("b"),
283 codec: codec,
284 sendDir: 1,
285 mtu: pdgram.DefaultMTU,
286 pending: make(map[uint64]*pending),
287 resendAfter: 100 * time.Millisecond,
288 maxResendAfter: 500 * time.Millisecond,
289 maxResendPerTick: maxResendPerTick,
290 rto: 100 * time.Millisecond,
291 clock: realClock{},
292 congestion: newCongestionController(pdgram.DefaultMTU),
293 done: make(chan struct{}),
294 }, &writes
295}
296
297func ackRecord(seq uint64) []byte {
298 var b [9]byte
299 b[0] = recAck
300 binary.BigEndian.PutUint64(b[1:], seq)
301 return b[:]
302}
303
304func probeRecord(kind byte, id uint64) []byte {
305 var b [9]byte
306 b[0] = kind
307 binary.BigEndian.PutUint64(b[1:], id)
308 return b[:]
309}
310
311func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
312 tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
313 tr.deliverCond = sync.NewCond(&tr.deliverMu)
314 for i := range maxRecvBuffer + 100 {
315 tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
316 }
317 if got := len(tr.recvBuf); got != maxRecvBuffer {
318 t.Fatalf("recvBuf len=%d, want %d", got, maxRecvBuffer)
319 }
320 tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
321 if _, ok := tr.recvBuf[1]; !ok {
322 t.Fatalf("next expected sequence was dropped when far-future buffer was full")
323 }
324 if got := len(tr.recvBuf); got > maxRecvBuffer+1 {
325 t.Fatalf("recvBuf len=%d, want <= %d", got, maxRecvBuffer+1)
326 }
327}
328
329func TestOutputPrerequisitesRemainInFullReceiveBuffer(t *testing.T) {
330 tr := &Transport{
331 nextRecvSeq: 1,
332 recvBuf: map[uint64]ports.Frame{2: {Type: ports.MsgOutput, Payload: []byte("stale")}},
333 maxRecvBuffer: 1,
334 done: make(chan struct{}),
335 }
336 tr.deliverCond = sync.NewCond(&tr.deliverMu)
337
338 ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("replacement")})
339 if ack || queued || ackSeq != 0 {
340 t.Fatalf("replacement ackSeq=%d ack=%v queued=%v, want output rejected while prerequisite buffer is full", ackSeq, ack, queued)
341 }
342 if _, ok := tr.recvBuf[2]; !ok {
343 t.Fatal("prerequisite output was discarded from full receive buffer")
344 }
345 if _, ok := tr.recvBuf[3]; ok {
346 t.Fatal("far-future replacement bypassed full receive buffer")
347 }
348}
349
350func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
351 aPC, bPC := newPair()
352 codec, err := pdgram.NewCodec(key())
353 if err != nil {
354 t.Fatal(err)
355 }
356 tr := &Transport{pc: aPC, peer: testAddr("b"), codec: codec, sendDir: 1, mtu: pdgram.DefaultMTU, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, clock: realClock{}, done: make(chan struct{})}
357 tr.deliverCond = sync.NewCond(&tr.deliverMu)
358 defer func() { _ = aPC.Close() }()
359 defer func() { _ = bPC.Close() }()
360
361 for i := range maxRecvBuffer {
362 tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
363 }
364 tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))
365 select {
366 case pkt := <-bPC.in:
367 t.Fatalf("unexpected ACK packet for dropped frame: %x", pkt.b)
368 case <-time.After(25 * time.Millisecond):
369 }
370
371 tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}))
372 select {
373 case <-bPC.in:
374 case <-time.After(time.Second):
375 t.Fatal("expected ACK for buffered contiguous frame")
376 }
377}
378
379func recvMaybe(tr *Transport, d time.Duration) (ports.Frame, bool) {
380 ch := make(chan ports.Frame, 1)
381 go func() { f, _ := tr.Recv(); ch <- f }()
382 select {
383 case f := <-ch:
384 return f, true
385 case <-time.After(d):
386 return ports.Frame{}, false
387 }
388}
389
390func recvWithin(t *testing.T, tr *Transport, d time.Duration) ports.Frame {
391 t.Helper()
392 ch := make(chan ports.Frame, 1)
393 go func() { f, _ := tr.Recv(); ch <- f }()
394 select {
395 case f := <-ch:
396 return f
397 case <-time.After(d):
398 t.Fatal("timeout")
399 return ports.Frame{}
400 }
401}
402
403func waitPeer(t *testing.T, tr *Transport, want string) {
404 t.Helper()
405 deadline := time.Now().Add(time.Second)
406 for time.Now().Before(deadline) {
407 if tr.Peer().String() == want {
408 return
409 }
410 time.Sleep(time.Millisecond)
411 }
412 t.Fatalf("peer=%v, want %s", tr.Peer(), want)
413}
414
415func recvErrWithin(t *testing.T, tr *Transport, d time.Duration) (ports.Frame, error) {
416 t.Helper()
417 type result struct {
418 f ports.Frame
419 err error
420 }
421 ch := make(chan result, 1)
422 go func() {
423 f, err := tr.Recv()
424 ch <- result{f: f, err: err}
425 }()
426 select {
427 case r := <-ch:
428 return r.f, r.err
429 case <-time.After(d):
430 t.Fatal("timeout")
431 return ports.Frame{}, nil
432 }
433}