1package dgram
2
3import (
4 "errors"
5 "net"
6 "time"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11func (t *Transport) Send(f ports.Frame) error {
12 end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
13 err := t.send(f, false)
14 end(err == nil)
15 return err
16}
17
18func (t *Transport) SendAsync(f ports.Frame) error {
19 end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
20 err := t.send(f, true)
21 end(err == nil)
22 return err
23}
24
25// SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
26// and close cancellation. Callers must not layer a shorter preflight timer over it.
27func (t *Transport) SendSynchronous(f ports.Frame) error { return t.Send(f) }
28
29func (t *Transport) send(f ports.Frame, async bool) error {
30 reliable := true
31 if err := t.lockOutboundSlot(reliable); err != nil {
32 return err
33 }
34 unlockOutbound := true
35 defer func() {
36 if unlockOutbound {
37 t.outboundMu.Unlock()
38 }
39 }()
40 // lockOutboundSlot returns with both outboundMu and mu held.
41
42 t.seq++
43 seq := t.seq
44 if reliable {
45 now := t.clock.Now()
46 if len(t.pending) == 0 {
47 t.health.pendingStarted(now)
48 }
49 t.pending[seq] = &pending{frame: f, enqueued: now}
50 }
51 var done chan error
52 if !async {
53 done = make(chan error, 1)
54 }
55 job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
56 if shouldPaceOutput(f) || len(t.outputQueue) > 0 {
57 wasEmpty := len(t.outputQueue) == 0
58 t.outputQueue = append(t.outputQueue, queuedSend(job))
59 if wasEmpty {
60 t.notifyOutputPacerLocked()
61 }
62 t.mu.Unlock()
63 t.outboundMu.Unlock()
64 unlockOutbound = false
65 if async {
66 return nil
67 }
68 return t.waitQueuedSend(done)
69 }
70 t.mu.Unlock()
71 if err := t.queueDataJob(job); err != nil {
72 t.removePending(seq, reliable)
73 return err
74 }
75 t.outboundMu.Unlock()
76 unlockOutbound = false
77 if async {
78 return nil
79 }
80 return t.waitQueuedSend(done)
81}
82
83func (t *Transport) queueDataJob(job dataSendJob) error {
84 select {
85 case t.dataSend <- job:
86 return nil
87 case <-t.done:
88 return t.closedError()
89 }
90}
91
92func (t *Transport) waitQueuedSend(done <-chan error) error {
93 select {
94 case err := <-done:
95 return err
96 case <-t.done:
97 t.mu.Lock()
98 err := t.closeErr
99 t.mu.Unlock()
100 if err != nil {
101 return err
102 }
103 return errors.New("dgram: closed")
104 }
105}
106
107func (t *Transport) lockOutboundSlot(reliable bool) error {
108 var pendingWaitStarted time.Time
109 for {
110 t.outboundMu.Lock()
111 t.mu.Lock()
112 if t.closed {
113 err := t.closeErr
114 t.mu.Unlock()
115 t.outboundMu.Unlock()
116 if err != nil {
117 return err
118 }
119 return errors.New("dgram: closed")
120 }
121 if !reliable || len(t.pending) < t.maxPending {
122 return nil
123 }
124 if t.linkState != ports.LinkStateConnected {
125 t.mu.Unlock()
126 t.outboundMu.Unlock()
127 return ErrPendingFull
128 }
129 if pendingWaitStarted.IsZero() {
130 pendingWaitStarted = t.clock.Now()
131 }
132 remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
133 if remaining <= 0 {
134 t.mu.Unlock()
135 t.outboundMu.Unlock()
136 return ErrPendingFull
137 }
138 wake := t.sendWake
139 timer := t.clock.NewTimer(remaining)
140 t.mu.Unlock()
141 t.outboundMu.Unlock()
142 select {
143 case <-timer.C():
144 case <-wake:
145 case <-t.done:
146 }
147 timer.Stop()
148 }
149}
150
151func (t *Transport) removePending(seq uint64, reliable bool) {
152 if !reliable {
153 return
154 }
155 t.mu.Lock()
156 removed := t.removePendingLocked(seq)
157 if removed {
158 t.notifySendWaitersLocked()
159 }
160 t.mu.Unlock()
161}
162
163func (t *Transport) removeQueuedPending(queued []queuedSend, err error) {
164 t.mu.Lock()
165 removed := false
166 for _, q := range queued {
167 if q.reliable {
168 removed = t.removePendingLocked(q.seq) || removed
169 }
170 if q.done != nil {
171 q.done <- err
172 close(q.done)
173 }
174 }
175 if removed {
176 t.notifySendWaitersLocked()
177 }
178 t.mu.Unlock()
179}
180
181func (t *Transport) removePendingLocked(seq uint64) bool {
182 if _, ok := t.pending[seq]; !ok {
183 return false
184 }
185 delete(t.pending, seq)
186 if len(t.pending) == 0 {
187 t.health.pendingCleared()
188 }
189 return true
190}
191
192func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
193 t.writeMu.Lock()
194 defer t.writeMu.Unlock()
195
196 t.mu.Lock()
197 pc := t.pc
198 writeTimeout := t.writeTimeout
199 writeDeadlines := t.writeDeadlines
200 t.mu.Unlock()
201
202 return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
203}
204
205func (t *Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines *writeDeadlineState, timeout time.Duration) error {
206 if timeout <= 0 {
207 _, err := pc.WriteTo(pkt, peer)
208 return err
209 }
210 ownDeadline := t.clock.Now().Add(timeout)
211 finish, err := deadlines.begin(ownDeadline)
212 if err != nil {
213 return err
214 }
215 defer finish()
216 for {
217 _, err = pc.WriteTo(pkt, peer)
218 if err == nil {
219 return nil
220 }
221 var timeoutErr interface{ Timeout() bool }
222 now := t.clock.Now()
223 if !errors.As(err, &timeoutErr) || !timeoutErr.Timeout() || !now.Before(ownDeadline) {
224 return err
225 }
226 deadlines.expire(now)
227 }
228}
229
230func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
231
232func (t *Transport) notifySendWaitersLocked() {
233 if t.sendWake == nil {
234 return
235 }
236 close(t.sendWake)
237 t.sendWake = make(chan struct{})
238}