1package dgram
2
3import (
4 "errors"
5 "net"
6 "sync"
7
8 "github.com/bnema/vev/internal/ports"
9 pdgram "github.com/bnema/vev/pkg/dgram"
10)
11
12func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
13 return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
14}
15
16func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
17 c, err := pdgram.NewCodec(key)
18 if err != nil {
19 return nil, err
20 }
21 opts = normalizeOptions(opts)
22 t := &Transport{
23 pc: pc,
24 codec: c,
25 sendDir: sendDir,
26 recvDir: recvDir,
27 mtu: opts.MTU,
28 peer: peer,
29 pending: make(map[uint64]*pending),
30 replay: pdgram.NewReplayWindow(),
31 reasm: pdgram.NewReassembler(),
32 in: make(chan ports.Frame, 32),
33 done: make(chan struct{}),
34 health: newHealthTracker(opts.Clock.Now()),
35 lastAuthenticatedPacket: opts.Clock.Now(),
36 lastCompleteRecord: opts.Clock.Now(),
37 lastACKProgress: opts.Clock.Now(),
38 observe: opts.Observe,
39 heartbeat: opts.Heartbeat,
40 resendAfter: opts.ResendAfter,
41 maxResendAfter: opts.MaxResendAfter,
42 maxResendPerTick: opts.MaxResendPerTick,
43 rto: opts.ResendAfter,
44 writeTimeout: opts.WriteTimeout,
45 degradedAfter: opts.DegradedAfter,
46 probeAfter: opts.ProbeAfter,
47 offlineAfter: opts.OfflineAfter,
48 deadAfter: opts.DeadAfter,
49 maxPending: opts.MaxPending,
50 maxPendingWait: opts.MaxPendingWait,
51 maxRecvBuffer: opts.MaxRecvBuffer,
52 clock: opts.Clock,
53 congestion: newCongestionController(opts.MTU),
54 linkState: ports.LinkStateConnected,
55 linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
56 probeWait: make(map[uint64]chan struct{}),
57 rebind: opts.RebindPacketConn,
58 nextRecvSeq: 1,
59 recvBuf: make(map[uint64]ports.Frame),
60 }
61 for _, opt := range runtimeOpts {
62 if opt != nil {
63 opt(t)
64 }
65 }
66 t.sendWake = make(chan struct{})
67 t.control = make(chan controlRecord, controlQueueSize)
68 t.ackWake = make(chan struct{}, 1)
69 t.ackSend = make(chan uint64, 1)
70 t.outputWake = make(chan struct{})
71 t.dataSend = make(chan dataSendJob, opts.MaxPending)
72 t.writeDeadlines = newWriteDeadlineState(pc)
73 // One queued batch bounds retransmit work while its sole sender is paced.
74 t.retransmitWork = make(chan []retransmitRecord, 1)
75 if t.observe != nil {
76 t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
77 go t.diagnosticLoop()
78 }
79 t.deliverCond = sync.NewCond(&t.deliverMu)
80 go t.readLoop(pc)
81 go t.retransmitLoop()
82 go t.heartbeatLoop()
83 go t.healthLoop()
84 go t.dataSendLoop()
85 go t.deliveryLoop()
86 go t.controlLoop()
87 go t.ackSendLoop()
88 go t.outputPaceLoop()
89 t.emitDiagnostic()
90 return t, nil
91}
92
93func (t *Transport) Recv() (ports.Frame, error) {
94 end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
95 select {
96 case f := <-t.in:
97 end(true)
98 return f, nil
99 case <-t.done:
100 t.mu.Lock()
101 err := t.closeErr
102 t.mu.Unlock()
103 if err != nil {
104 end(false)
105 return ports.Frame{}, err
106 }
107 err = errors.New("dgram: closed")
108 end(false)
109 return ports.Frame{}, err
110 }
111}
112
113func (t *Transport) Close() error {
114 done := t.beginShutdown()
115 t.closeWithError(errors.New("dgram: closed"))
116 t.mu.Lock()
117 pc := t.pc
118 t.mu.Unlock()
119 err := pc.Close()
120 if done != nil {
121 <-done
122 }
123 return err
124}
125
126func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }
127
128func (t *Transport) LinkState() ports.LinkState {
129 t.mu.Lock()
130 defer t.mu.Unlock()
131 return t.linkState
132}
133
134func (t *Transport) LinkEvents() <-chan ports.LinkEvent { return t.linkEvents }
135
136func (t *Transport) closedError() error {
137 t.mu.Lock()
138 defer t.mu.Unlock()
139 if t.closeErr != nil {
140 return t.closeErr
141 }
142 return errors.New("dgram: closed")
143}
144
145func (t *Transport) closeWithError(err error) {
146 t.mu.Lock()
147 closed := t.closeWithErrorLocked(err)
148 t.mu.Unlock()
149 if closed {
150 t.broadcastClosed()
151 }
152}
153
154func (t *Transport) closeWithErrorLocked(err error) bool {
155 if t.closed {
156 return false
157 }
158 t.closed = true
159 t.closeErr = err
160 for _, q := range t.outputQueue {
161 if q.done != nil {
162 q.done <- err
163 close(q.done)
164 }
165 }
166 t.outputQueue = nil
167 t.probeWait = make(map[uint64]chan struct{})
168 close(t.done)
169 t.notifySendWaitersLocked()
170 return true
171}
172
173func (t *Transport) broadcastClosed() {
174 t.deliverMu.Lock()
175 t.deliverCond.Broadcast()
176 t.deliverMu.Unlock()
177}