1//go:build linux
2
3package main
4
5import (
6 "container/heap"
7 "context"
8 "errors"
9 "fmt"
10 "net"
11 "os"
12 "strconv"
13 "strings"
14 "sync"
15 "sync/atomic"
16 "time"
17)
18
19// udpNetem is a harness-owned, user-space UDP relay. It sits between the
20// public dgram client and vev's public _udp-proxy readiness address, so fixture
21// impairment is real network carriage rather than an ignored vev option.
22type udpNetem interface {
23 Port() int
24 Close() error
25}
26
27type udpNetemFactory func(udpNetemConfig) (udpNetem, error)
28
29type udpNetemConfig struct {
30 // RTT is split equally across both relay directions.
31 RTT time.Duration
32 LossPercent int
33 TargetPath string
34}
35
36// udpNetemQueueCapacity bounds both packets waiting to be scheduled and
37// packets waiting for their delivery deadline. When it is full, the relay
38// drops the arriving packet and records it in queueOverflowDrops.
39const udpNetemQueueCapacity = 256
40
41type scheduledPacket struct {
42 packet []byte
43 to net.Addr
44 due time.Time
45}
46
47type scheduledPackets []scheduledPacket
48
49func (p scheduledPackets) Len() int { return len(p) }
50func (p scheduledPackets) Less(i, j int) bool { return p[i].due.Before(p[j].due) }
51func (p scheduledPackets) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
52func (p *scheduledPackets) Push(x any) {
53 packet, ok := x.(scheduledPacket)
54 if !ok {
55 panic(fmt.Sprintf("scheduled packet type %T", x))
56 }
57 *p = append(*p, packet)
58}
59func (p *scheduledPackets) Pop() any {
60 old := *p
61 n := len(old)
62 item := old[n-1]
63 *p = old[:n-1]
64 return item
65}
66
67type udpNetemRelay struct {
68 config udpNetemConfig
69 conn net.PacketConn
70 done chan struct{}
71 wg sync.WaitGroup
72 once sync.Once
73 closeErr error
74
75 // slots bounds queued and deadline-pending packets together. The reader and
76 // the sole scheduler are the relay's only goroutines.
77 slots chan struct{}
78 queue chan scheduledPacket
79 queueOverflowDrops atomic.Uint64
80 deliveryErrors atomic.Uint64
81
82 mu sync.Mutex
83 client, dst net.Addr
84 packet uint64
85}
86
87func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
88 if config.RTT < 0 || config.LossPercent < 0 || config.LossPercent > 100 || config.TargetPath == "" {
89 return nil, fmt.Errorf("invalid UDP netem fixture: rtt=%s loss=%d target=%q", config.RTT, config.LossPercent, config.TargetPath)
90 }
91 var lc net.ListenConfig
92 conn, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
93 if err != nil {
94 return nil, err
95 }
96 r := &udpNetemRelay{
97 config: config,
98 conn: conn,
99 done: make(chan struct{}),
100 slots: make(chan struct{}, udpNetemQueueCapacity),
101 queue: make(chan scheduledPacket, udpNetemQueueCapacity),
102 }
103 r.wg.Add(2)
104 go r.run()
105 go r.schedule()
106 return r, nil
107}
108
109func (r *udpNetemRelay) Port() int {
110 addr, ok := r.conn.LocalAddr().(*net.UDPAddr)
111 if !ok {
112 return 0
113 }
114 return addr.Port
115}
116
117func (r *udpNetemRelay) Close() error {
118 r.once.Do(func() {
119 close(r.done)
120 r.closeErr = r.conn.Close()
121 r.wg.Wait()
122 if drops := r.queueOverflowDrops.Load(); drops != 0 {
123 r.closeErr = errors.Join(r.closeErr, fmt.Errorf("udp netem queue overflow drops: %d", drops))
124 }
125 if writes := r.deliveryErrors.Load(); writes != 0 {
126 r.closeErr = errors.Join(r.closeErr, fmt.Errorf("udp netem delivery errors: %d", writes))
127 }
128 })
129 return r.closeErr
130}
131
132func (r *udpNetemRelay) run() {
133 defer r.wg.Done()
134 buf := make([]byte, 64*1024)
135 for {
136 n, from, err := r.conn.ReadFrom(buf)
137 if err != nil {
138 return
139 }
140 to, ok := r.destination(from)
141 if !ok || r.drop() {
142 continue
143 }
144 // Copy because PacketConn reuses buf on its next read.
145 r.enqueue(scheduledPacket{packet: append([]byte(nil), buf[:n]...), to: to, due: time.Now().Add(r.config.RTT / 2)})
146 }
147}
148
149// enqueue implements the explicit overflow policy: never block the reader;
150// drop a packet if the bounded scheduler capacity has been reached.
151func (r *udpNetemRelay) enqueue(packet scheduledPacket) bool {
152 select {
153 case r.slots <- struct{}{}:
154 case <-r.done:
155 return false
156 default:
157 r.queueOverflowDrops.Add(1)
158 return false
159 }
160 select {
161 case r.queue <- packet:
162 return true
163 case <-r.done:
164 <-r.slots
165 return false
166 }
167}
168
169func (r *udpNetemRelay) schedule() {
170 defer r.wg.Done()
171 pending := scheduledPackets{}
172 heap.Init(&pending)
173 timer := time.NewTimer(time.Hour)
174 if !timer.Stop() {
175 <-timer.C
176 }
177 defer timer.Stop()
178 var timerC <-chan time.Time
179 resetTimer := func() {
180 if timerC != nil && !timer.Stop() {
181 select {
182 case <-timer.C:
183 default:
184 }
185 }
186 if pending.Len() == 0 {
187 timerC = nil
188 return
189 }
190 delay := time.Until(pending[0].due)
191 if delay < 0 {
192 delay = 0
193 }
194 timer.Reset(delay)
195 timerC = timer.C
196 }
197 for {
198 resetTimer()
199 select {
200 case <-r.done:
201 return
202 case packet := <-r.queue:
203 heap.Push(&pending, packet)
204 case <-timerC:
205 now := time.Now()
206 for pending.Len() > 0 && !pending[0].due.After(now) {
207 packet, ok := heap.Pop(&pending).(scheduledPacket)
208 if !ok {
209 panic("scheduled packet heap contained unexpected type")
210 }
211 select {
212 case <-r.done:
213 return
214 default:
215 if _, err := r.conn.WriteTo(packet.packet, packet.to); err != nil {
216 select {
217 case <-r.done:
218 default:
219 r.deliveryErrors.Add(1)
220 }
221 }
222 <-r.slots
223 }
224 }
225 }
226 }
227}
228
229func (r *udpNetemRelay) destination(from net.Addr) (net.Addr, bool) {
230 r.mu.Lock()
231 defer r.mu.Unlock()
232 if r.dst == nil {
233 dst, err := udpNetemTarget(r.config.TargetPath)
234 if err != nil {
235 return nil, false
236 }
237 r.dst = dst
238 }
239 if sameUDPAddress(from, r.dst) {
240 return r.client, r.client != nil
241 }
242 r.client = from
243 return r.dst, true
244}
245
246func (r *udpNetemRelay) drop() bool {
247 r.mu.Lock()
248 defer r.mu.Unlock()
249 r.packet++
250 // The deterministic cycle makes the configured percentage auditable: a 1%
251 // fixture drops exactly one carriage packet in every 100 attempts.
252 return r.config.LossPercent > 0 && int((r.packet-1)%100) < r.config.LossPercent
253}
254
255func udpNetemTarget(path string) (net.Addr, error) {
256 line, err := os.ReadFile(path)
257 if err != nil {
258 return nil, err
259 }
260 fields := strings.Fields(string(line))
261 if len(fields) != 3 || fields[0] != "VEV-UDP" {
262 return nil, errors.New("invalid UDP proxy readiness")
263 }
264 port, err := strconv.Atoi(fields[1])
265 if err != nil || port < 1 || port > 65535 || fields[2] == "" {
266 return nil, errors.New("invalid UDP proxy readiness")
267 }
268 return &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, nil
269}
270
271func sameUDPAddress(a, b net.Addr) bool {
272 ua, aOK := a.(*net.UDPAddr)
273 ub, bOK := b.(*net.UDPAddr)
274 return aOK && bOK && ua.Port == ub.Port && ua.IP.Equal(ub.IP)
275}