control.go RawHistoryFolder
1package dgram
2
3import (
4 "encoding/binary"
5 "errors"
6 "net"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10 pdgram "github.com/bnema/vev/pkg/dgram"
11)
12
13const (
14 controlQueueSize = 16
15 maxACKDelay = 25 * time.Millisecond
16)
17
18type controlRecord struct {
19 kind byte
20 id uint64
21 result chan<- error
22}
23
24// queueACK retains the greatest cumulative acknowledgement without ever making
25// packet receipt wait for a socket write.
26func (t *Transport) queueACK(seq uint64) {
27 t.controlMu.Lock()
28 if !t.ackQueued || seq > t.ackSeq {
29 t.ackSeq = seq
30 t.ackQueued = true
31 }
32 t.controlMu.Unlock()
33 if t.ackWake == nil {
34 // Small unit-test transports may exercise record handling without the
35 // goroutines installed by NewTransportWithOptions.
36 go t.sendAck(seq)
37 return
38 }
39 select {
40 case t.ackWake <- struct{}{}:
41 t.mu.Lock()
42 hook := t.afterACKWakeAccepted
43 t.mu.Unlock()
44 if hook != nil {
45 hook()
46 }
47 default:
48 }
49}
50
51// queueControl is deliberately lossy when the bounded control queue is full:
52// probes and pongs are advisory and must not stall authenticated receipt.
53func (t *Transport) queueControl(kind byte, id uint64) bool {
54 return t.queueControlRecord(controlRecord{kind: kind, id: id})
55}
56
57func (t *Transport) queueControlRecord(record controlRecord) bool {
58 select {
59 case t.control <- record:
60 return true
61 default:
62 return false
63 }
64}
65
66func (t *Transport) takeACK() (uint64, bool) {
67 t.controlMu.Lock()
68 defer t.controlMu.Unlock()
69 seq, ok := t.ackSeq, t.ackQueued
70 t.ackSeq = 0
71 t.ackQueued = false
72 return seq, ok
73}
74
75func (t *Transport) controlLoop() {
76 go t.ackScheduleLoop()
77 for {
78 select {
79 case record := <-t.control:
80 err := t.sendControl(record.kind, record.id)
81 if record.result != nil {
82 select {
83 case record.result <- err:
84 default:
85 }
86 }
87 case <-t.done:
88 return
89 }
90 }
91}
92
93func (t *Transport) ackScheduleLoop() {
94 var ackTimer ports.Timer
95 var ackTimerC <-chan time.Time
96 defer func() {
97 if ackTimer != nil {
98 ackTimer.Stop()
99 }
100 }()
101 scheduleACK := func() {
102 if ackTimerC == nil {
103 if ackTimer == nil {
104 ackTimer = t.clock.NewTimer(maxACKDelay)
105 } else {
106 ackTimer.Reset(maxACKDelay)
107 }
108 ackTimerC = ackTimer.C()
109 }
110 t.mu.Lock()
111 hook := t.afterACKScheduled
112 t.mu.Unlock()
113 if hook != nil {
114 hook()
115 }
116 }
117 dispatchACK := func() {
118 seq, ok := t.takeACK()
119 if !ok {
120 return
121 }
122 select {
123 case t.ackSend <- seq:
124 t.mu.Lock()
125 hook := t.afterACKDispatched
126 t.mu.Unlock()
127 if hook != nil {
128 hook()
129 }
130 default:
131 // Preserve the cumulative maximum until the bounded ACK sender has
132 // capacity again.
133 t.queueACK(seq)
134 }
135 }
136
137 for {
138 // Check ACK scheduling explicitly so a sustained stream of advisory
139 // controls cannot prevent the bounded coalescing deadline from starting
140 // or expiring. Dispatch itself never writes to the socket.
141 select {
142 case <-ackTimerC:
143 ackTimerC = nil
144 dispatchACK()
145 continue
146 default:
147 }
148 select {
149 case <-t.ackWake:
150 scheduleACK()
151 continue
152 default:
153 }
154
155 select {
156 case <-t.ackWake:
157 scheduleACK()
158 case <-ackTimerC:
159 ackTimerC = nil
160 dispatchACK()
161 case <-t.done:
162 return
163 }
164 }
165}
166
167func (t *Transport) ackSendLoop() {
168 for {
169 select {
170 case seq := <-t.ackSend:
171 _ = t.sendControl(recAck, seq)
172 case <-t.done:
173 return
174 }
175 }
176}
177
178func (t *Transport) sendControl(kind byte, id uint64) error {
179 // Do not take writeMu: control must remain available while data writes are
180 // blocked. The read lock instead keeps the selected connection alive until
181 // every control fragment has been written.
182 t.controlConnMu.RLock()
183 defer t.controlConnMu.RUnlock()
184
185 var payload [9]byte
186 payload[0] = kind
187 binary.BigEndian.PutUint64(payload[1:], id)
188 frags, err := pdgram.FragmentPayload(t.nextCounter(), payload[:], t.mtu-pdgram.HeaderSize-t.codec.Overhead())
189 if err != nil {
190 return err
191 }
192 for _, frag := range frags {
193 raw, err := pdgram.MarshalFragment(frag)
194 if err != nil {
195 return err
196 }
197 pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
198 t.mu.Lock()
199 pc := t.pc
200 peer := t.peer
201 closed := t.closed
202 closeErr := t.closeErr
203 writeTimeout := t.writeTimeout
204 writeDeadlines := t.writeDeadlines
205 t.mu.Unlock()
206 if closed {
207 if closeErr != nil {
208 return closeErr
209 }
210 return net.ErrClosed
211 }
212 if peer == nil {
213 return errors.New("dgram: control peer unavailable")
214 }
215 // PacketConn permits concurrent callers. Control traffic deliberately
216 // bypasses data's write lock so a retransmit cannot starve heartbeats.
217 if err := t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout); err != nil {
218 return err
219 }
220 }
221 return nil
222}
223
224func (t *Transport) sendAck(seq uint64) {
225 _ = t.sendControl(recAck, seq)
226}