Skip to content

Commit 44c78d5

Browse files
Haily Nguyenhanguyen-nuro
Haily Nguyen
authored and
hanguyen-nuro
committed
Add loss and delay filter to vnet
Implement basic DelayFilter and LossFilter functionality for network simulation in the vnet package.
1 parent d458a44 commit 44c78d5

4 files changed

Lines changed: 495 additions & 114 deletions

File tree

‎vnet/delay_filter.go‎

Lines changed: 82 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -4,77 +4,126 @@
44
package vnet
55

66
import (
7-
"context"
7+
"sync"
8+
"sync/atomic"
89
"time"
910
)
1011

11-
// DelayFilter delays outgoing packets by the given delay. Run must be called
12-
// before any packets will be forwarded.
12+
// DelayFilter delays inbound packets by the given delay. Automatically starts
13+
// processing when created and runs until Close() is called.
1314
type DelayFilter struct {
1415
NIC
15-
delay time.Duration
16+
delay atomic.Int64 // atomic field - stores time.Duration as int64
1617
push chan struct{}
1718
queue *chunkQueue
19+
done chan struct{}
20+
wg sync.WaitGroup
1821
}
1922

2023
type timedChunk struct {
2124
Chunk
2225
deadline time.Time
2326
}
2427

25-
// NewDelayFilter creates a new DelayFilter with the given nic and delay.
28+
// NewDelayFilter creates and starts a new DelayFilter with the given nic and delay.
2629
func NewDelayFilter(nic NIC, delay time.Duration) (*DelayFilter, error) {
27-
return &DelayFilter{
30+
f := &DelayFilter{
2831
NIC: nic,
29-
delay: delay,
3032
push: make(chan struct{}),
3133
queue: newChunkQueue(0, 0),
32-
}, nil
34+
done: make(chan struct{}),
35+
}
36+
37+
f.delay.Store(int64(delay))
38+
39+
// Start processing automatically
40+
f.wg.Add(1)
41+
go f.run()
42+
43+
return f, nil
44+
}
45+
46+
// SetDelay atomically updates the delay
47+
func (f *DelayFilter) SetDelay(newDelay time.Duration) {
48+
f.delay.Store(int64(newDelay))
49+
}
50+
51+
func (f *DelayFilter) getDelay() time.Duration {
52+
return time.Duration(f.delay.Load())
3353
}
3454

3555
func (f *DelayFilter) onInboundChunk(c Chunk) {
3656
f.queue.push(timedChunk{
3757
Chunk: c,
38-
deadline: time.Now().Add(f.delay),
58+
deadline: time.Now().Add(f.getDelay()),
3959
})
4060
f.push <- struct{}{}
4161
}
4262

43-
// Run starts forwarding of packets. Packets will be forwarded if they spent
44-
// >delay time in the internal queue. Must be called before any packet will be
45-
// forwarded.
46-
func (f *DelayFilter) Run(ctx context.Context) { //nolint:cyclop
63+
// run processes the delayed packets queue until Close() is called.
64+
func (f *DelayFilter) run() {
65+
defer f.wg.Done()
66+
4767
timer := time.NewTimer(0)
68+
defer timer.Stop()
69+
4870
for {
4971
select {
50-
case <-ctx.Done():
72+
case <-f.done:
73+
// Drain remaining packets immediately on shutdown
74+
for {
75+
next, ok := f.queue.pop()
76+
if !ok {
77+
break
78+
}
79+
if timedChunk, ok := next.(timedChunk); ok {
80+
f.NIC.onInboundChunk(timedChunk.Chunk)
81+
}
82+
}
5183
return
84+
5285
case <-f.push:
53-
next := f.queue.peek().(timedChunk) //nolint:forcetypeassert
54-
if !timer.Stop() {
55-
<-timer.C
86+
// New packet arrived, update timer for next deadline
87+
next := f.queue.peek()
88+
if next != nil {
89+
if timedChunk, ok := next.(timedChunk); ok {
90+
if !timer.Stop() {
91+
<-timer.C
92+
}
93+
timer.Reset(time.Until(timedChunk.deadline))
94+
}
5695
}
57-
timer.Reset(time.Until(next.deadline))
96+
5897
case now := <-timer.C:
59-
next := f.queue.peek()
60-
if next == nil {
61-
timer.Reset(time.Minute)
98+
// Process all ready packets
99+
for {
100+
next := f.queue.peek()
101+
if next == nil {
102+
break
103+
}
62104

63-
continue
64-
}
65-
if n, ok := next.(timedChunk); ok && n.deadline.Before(now) {
66-
f.queue.pop() // ignore result because we already got and casted it from peek
67-
f.NIC.onInboundChunk(n.Chunk)
105+
if timedChunk, ok := next.(timedChunk); ok && timedChunk.deadline.Before(now) {
106+
_, _ = f.queue.pop() // We already have the item from peek()
107+
f.NIC.onInboundChunk(timedChunk.Chunk)
108+
} else {
109+
break
110+
}
68111
}
69-
next = f.queue.peek()
70-
if next == nil {
71-
timer.Reset(time.Minute)
72112

73-
continue
74-
}
75-
if n, ok := next.(timedChunk); ok {
76-
timer.Reset(time.Until(n.deadline))
113+
// Schedule timer for next packet
114+
next := f.queue.peek()
115+
if next == nil {
116+
timer.Reset(time.Minute) // Long timeout when queue is empty
117+
} else if timedChunk, ok := next.(timedChunk); ok {
118+
timer.Reset(time.Until(timedChunk.deadline))
77119
}
78120
}
79121
}
80122
}
123+
124+
// Close stops the DelayFilter and waits for graceful shutdown.
125+
func (f *DelayFilter) Close() error {
126+
close(f.done)
127+
f.wg.Wait()
128+
return nil
129+
}

‎vnet/delay_filter_test.go‎

Lines changed: 126 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -4,103 +4,157 @@
44
package vnet
55

66
import (
7-
"context"
87
"testing"
98
"time"
109

1110
"github.com/stretchr/testify/assert"
1211
)
1312

14-
func TestDelayFilter(t *testing.T) {
15-
t.Run("schedulesOnePacketAtATime", func(t *testing.T) {
16-
nic := newMockNIC(t)
17-
df, err := NewDelayFilter(nic, 10*time.Millisecond)
18-
if !assert.NoError(t, err, "should succeed") {
19-
return
13+
type TimestampedChunk struct {
14+
ts time.Time
15+
c Chunk
16+
}
17+
18+
func initTest(t *testing.T) (*DelayFilter, chan TimestampedChunk) {
19+
t.Helper()
20+
nic := newMockNIC(t)
21+
df, err := NewDelayFilter(nic, 0)
22+
if !assert.NoError(t, err, "should succeed") {
23+
return nil, nil
24+
}
25+
26+
receiveCh := make(chan TimestampedChunk)
27+
28+
nic.mockOnInboundChunk = func(c Chunk) {
29+
receivedAt := time.Now()
30+
receiveCh <- TimestampedChunk{
31+
ts: receivedAt,
32+
c: c,
2033
}
34+
}
35+
36+
return df, receiveCh
37+
}
2138

22-
ctx, cancel := context.WithCancel(context.Background())
23-
defer cancel()
24-
go df.Run(ctx)
39+
func scheduleOnePacketAtATime(t *testing.T, df *DelayFilter, receiveCh chan TimestampedChunk, delay time.Duration, nrPackets int) {
40+
t.Helper()
41+
df.SetDelay(delay)
42+
lastNr := -1
43+
for i := 0; i < nrPackets; i++ {
44+
sent := time.Now()
45+
df.onInboundChunk(&chunkUDP{
46+
chunkIP: chunkIP{timestamp: sent},
47+
userData: []byte{byte(i)},
48+
})
49+
50+
select {
51+
case c := <-receiveCh:
52+
nr := int(c.c.UserData()[0])
53+
54+
assert.Greater(t, nr, lastNr)
55+
lastNr = nr
2556

26-
type TimestampedChunk struct {
27-
ts time.Time
28-
c Chunk
57+
assert.Greater(t, c.ts.Sub(sent), delay)
58+
assert.Less(t, c.ts.Sub(sent), delay+10*time.Millisecond)
59+
case <-time.After(time.Second):
60+
assert.Fail(t, "expected to receive next chunk")
2961
}
30-
receiveCh := make(chan TimestampedChunk)
31-
nic.mockOnInboundChunk = func(c Chunk) {
32-
receivedAt := time.Now()
33-
receiveCh <- TimestampedChunk{
34-
ts: receivedAt,
35-
c: c,
36-
}
62+
}
63+
}
64+
65+
func scheduleManyPackets(t *testing.T, df *DelayFilter, receiveCh chan TimestampedChunk, delay time.Duration, nrPackets int) {
66+
t.Helper()
67+
df.SetDelay(delay)
68+
sent := time.Now()
69+
70+
for i := 0; i < nrPackets; i++ {
71+
df.onInboundChunk(&chunkUDP{
72+
chunkIP: chunkIP{timestamp: sent},
73+
userData: []byte{byte(i)},
74+
})
75+
}
76+
77+
// receive nrPackets chunks with a minimum delay
78+
for i := 0; i < nrPackets; i++ {
79+
select {
80+
case c := <-receiveCh:
81+
nr := int(c.c.UserData()[0])
82+
assert.Equal(t, i, nr)
83+
assert.Greater(t, c.ts.Sub(sent), delay)
84+
assert.Less(t, c.ts.Sub(sent), delay+10*time.Millisecond)
85+
case <-time.After(time.Second):
86+
assert.Fail(t, "expected to receive next chunk")
3787
}
88+
}
89+
}
3890

39-
lastNr := -1
40-
for i := 0; i < 100; i++ {
41-
sent := time.Now()
42-
df.onInboundChunk(&chunkUDP{
43-
chunkIP: chunkIP{timestamp: sent},
44-
userData: []byte{byte(i)},
45-
})
46-
47-
select {
48-
case c := <-receiveCh:
49-
nr := int(c.c.UserData()[0])
50-
51-
assert.Greater(t, nr, lastNr)
52-
lastNr = nr
53-
54-
assert.Greater(t, c.ts.Sub(sent), 10*time.Millisecond)
55-
case <-time.After(time.Second):
56-
assert.Fail(t, "expected to receive next chunk")
57-
}
91+
func TestDelayFilter(t *testing.T) {
92+
t.Run("schedulesOnePacketAtATime", func(t *testing.T) {
93+
df, receiveCh := initTest(t)
94+
if df == nil {
95+
return
5896
}
97+
98+
scheduleOnePacketAtATime(t, df, receiveCh, 10*time.Millisecond, 100)
99+
assert.NoError(t, df.Close())
59100
})
60101

61102
t.Run("schedulesSubsequentManyPackets", func(t *testing.T) {
62-
nic := newMockNIC(t)
63-
df, err := NewDelayFilter(nic, 10*time.Millisecond)
64-
if !assert.NoError(t, err, "should succeed") {
103+
df, receiveCh := initTest(t)
104+
if df == nil {
65105
return
66106
}
67107

68-
ctx, cancel := context.WithCancel(context.Background())
69-
defer cancel()
70-
go df.Run(ctx)
108+
scheduleManyPackets(t, df, receiveCh, 10*time.Millisecond, 100)
109+
assert.NoError(t, df.Close())
110+
})
71111

72-
type TimestampedChunk struct {
73-
ts time.Time
74-
c Chunk
112+
t.Run("scheduleIncreasingDelayOnePacketAtATime", func(t *testing.T) {
113+
df, receiveCh := initTest(t)
114+
if df == nil {
115+
return
75116
}
76-
receiveCh := make(chan TimestampedChunk)
77-
nic.mockOnInboundChunk = func(c Chunk) {
78-
receivedAt := time.Now()
79-
receiveCh <- TimestampedChunk{
80-
ts: receivedAt,
81-
c: c,
82-
}
117+
118+
scheduleOnePacketAtATime(t, df, receiveCh, 10*time.Millisecond, 10)
119+
scheduleOnePacketAtATime(t, df, receiveCh, 50*time.Millisecond, 10)
120+
scheduleOnePacketAtATime(t, df, receiveCh, 100*time.Millisecond, 10)
121+
assert.NoError(t, df.Close())
122+
})
123+
124+
t.Run("scheduleDecreasingDelayOnePacketAtATime", func(t *testing.T) {
125+
df, receiveCh := initTest(t)
126+
if df == nil {
127+
return
83128
}
84129

85-
// schedule 100 chunks
86-
sent := time.Now()
87-
for i := 0; i < 100; i++ {
88-
df.onInboundChunk(&chunkUDP{
89-
chunkIP: chunkIP{timestamp: sent},
90-
userData: []byte{byte(i)},
91-
})
130+
scheduleOnePacketAtATime(t, df, receiveCh, 100*time.Millisecond, 10)
131+
scheduleOnePacketAtATime(t, df, receiveCh, 50*time.Millisecond, 10)
132+
scheduleOnePacketAtATime(t, df, receiveCh, 10*time.Millisecond, 10)
133+
assert.NoError(t, df.Close())
134+
})
135+
136+
t.Run("scheduleIncreasingDelayManyPackets", func(t *testing.T) {
137+
df, receiveCh := initTest(t)
138+
if df == nil {
139+
return
92140
}
93141

94-
// receive 100 chunks with delay>10ms
95-
for i := 0; i < 100; i++ {
96-
select {
97-
case c := <-receiveCh:
98-
nr := int(c.c.UserData()[0])
99-
assert.Equal(t, i, nr)
100-
assert.Greater(t, c.ts.Sub(sent), 10*time.Millisecond)
101-
case <-time.After(time.Second):
102-
assert.Fail(t, "expected to receive next chunk")
103-
}
142+
scheduleManyPackets(t, df, receiveCh, 10*time.Millisecond, 100)
143+
scheduleManyPackets(t, df, receiveCh, 50*time.Millisecond, 100)
144+
scheduleManyPackets(t, df, receiveCh, 100*time.Millisecond, 100)
145+
assert.NoError(t, df.Close())
146+
})
147+
148+
t.Run("scheduleDecreasingDelayManyPackets", func(t *testing.T) {
149+
df, receiveCh := initTest(t)
150+
if df == nil {
151+
return
104152
}
153+
154+
scheduleManyPackets(t, df, receiveCh, 100*time.Millisecond, 100)
155+
scheduleManyPackets(t, df, receiveCh, 50*time.Millisecond, 100)
156+
scheduleManyPackets(t, df, receiveCh, 10*time.Millisecond, 100)
157+
assert.NoError(t, df.Close())
105158
})
159+
106160
}

0 commit comments

Comments
 (0)