output_pacer.go RawHistoryFolder

1package dgram
2
3import "github.com/bnema/vev/internal/ports"
4
5type dataSendJob struct {
6	seq      uint64
7	reliable bool
8	frame    ports.Frame
9	done     chan error
10}
11
12type queuedSend = dataSendJob
13
14func (t *Transport) outputPaceLoop() {
15	for {
16		t.outboundMu.Lock()
17		t.mu.Lock()
18		if len(t.outputQueue) == 0 && !t.closed {
19			wake := t.outputWake
20			t.mu.Unlock()
21			t.outboundMu.Unlock()
22			select {
23			case <-wake:
24			case <-t.done:
25				return
26			}
27			continue
28		}
29		if t.closed {
30			t.mu.Unlock()
31			t.outboundMu.Unlock()
32			return
33		}
34		q := t.outputQueue[0]
35		copy(t.outputQueue, t.outputQueue[1:])
36		t.outputQueue = t.outputQueue[:len(t.outputQueue)-1]
37		t.mu.Unlock()
38		if err := t.queueDataJob(q); err != nil {
39			t.removeQueuedPending([]queuedSend{q}, err)
40			t.outboundMu.Unlock()
41			return
42		}
43		t.outboundMu.Unlock()
44	}
45}
46
47func (t *Transport) notifyOutputPacerLocked() {
48	if t.outputWake == nil {
49		return
50	}
51	close(t.outputWake)
52	t.outputWake = make(chan struct{})
53}
54
55func shouldPaceOutput(f ports.Frame) bool {
56	return f.Type == ports.MsgOutput
57}
58
59func (t *Transport) markPendingSent(seq uint64, reliable bool) {
60	if !reliable {
61		return
62	}
63	t.mu.Lock()
64	if p := t.pending[seq]; p != nil {
65		p.initialInFlight = true
66	}
67	t.mu.Unlock()
68}
69
70func (t *Transport) markPendingReady(seq uint64, reliable bool) {
71	if !reliable {
72		return
73	}
74	t.mu.Lock()
75	if p := t.pending[seq]; p != nil {
76		p.initialInFlight = false
77	}
78	t.mu.Unlock()
79}
80
81func (t *Transport) markPendingFinalWrite(seq uint64) {
82	now := t.clock.Now()
83	t.mu.Lock()
84	if p := t.pending[seq]; p != nil {
85		if p.first.IsZero() {
86			p.first = now
87		}
88		p.last = now
89	}
90	t.mu.Unlock()
91}