1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
// Copyright 2020 The gVisor Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Package fifo provides the implementation of FIFO queuing discipline that
// queues all outbound packets and asynchronously dispatches them to the
// lower link endpoint in the order that they were queued.
package fifo
import (
"gvisor.dev/gvisor/pkg/atomicbitops"
"gvisor.dev/gvisor/pkg/sleep"
"gvisor.dev/gvisor/pkg/sync"
"gvisor.dev/gvisor/pkg/tcpip"
"gvisor.dev/gvisor/pkg/tcpip/stack"
)
var _ stack.QueueingDiscipline = (*discipline)(nil)
const (
// BatchSize is the number of packets to write in each syscall. It is 47
// because when GVisorGSO is in use then a single 65KB TCP segment can get
// split into 46 segments of 1420 bytes and a single 216 byte segment.
BatchSize = 47
qDiscClosed = 1
)
// discipline represents a QueueingDiscipline which implements a FIFO queue for
// all outgoing packets. discipline can have 1 or more underlying
// queueDispatchers. All outgoing packets are consistently hashed to a single
// underlying queue using the PacketBuffer.Hash if set, otherwise all packets
// are queued to the first queue to avoid reordering in case of missing hash.
type discipline struct {
wg sync.WaitGroup
dispatchers []queueDispatcher
closed atomicbitops.Int32
}
// queueDispatcher is responsible for dispatching all outbound packets in its
// queue. It will also smartly batch packets when possible and write them
// through the lower LinkWriter.
type queueDispatcher struct {
lower stack.LinkWriter
mu sync.Mutex
// +checklocks:mu
queue packetBufferCircularList
newPacketWaker sleep.Waker
closeWaker sleep.Waker
}
// New creates a new fifo queuing discipline with the n queues with maximum
// capacity of queueLen.
//
// +checklocksignore: we don't have to hold locks during initialization.
func New(lower stack.LinkWriter, n int, queueLen int) stack.QueueingDiscipline {
d := &discipline{
dispatchers: make([]queueDispatcher, n),
}
// Create the required dispatchers
for i := range d.dispatchers {
qd := &d.dispatchers[i]
qd.lower = lower
qd.queue.init(queueLen)
d.wg.Add(1)
go func() {
defer d.wg.Done()
qd.dispatchLoop()
}()
}
return d
}
func (qd *queueDispatcher) dispatchLoop() {
s := sleep.Sleeper{}
s.AddWaker(&qd.newPacketWaker)
s.AddWaker(&qd.closeWaker)
defer s.Done()
var batch stack.PacketBufferList
for {
switch w := s.Fetch(true); w {
case &qd.newPacketWaker:
case &qd.closeWaker:
qd.mu.Lock()
for p := qd.queue.removeFront(); p != nil; p = qd.queue.removeFront() {
p.DecRef()
}
qd.queue.decRef()
qd.mu.Unlock()
return
default:
panic("unknown waker")
}
qd.mu.Lock()
for pkt := qd.queue.removeFront(); pkt != nil; pkt = qd.queue.removeFront() {
batch.PushBack(pkt)
if batch.Len() < BatchSize && !qd.queue.isEmpty() {
continue
}
qd.mu.Unlock()
_, _ = qd.lower.WritePackets(batch)
batch.Reset()
qd.mu.Lock()
}
qd.mu.Unlock()
}
}
// WritePacket implements stack.QueueingDiscipline.WritePacket.
//
// The packet must have the following fields populated:
// - pkt.EgressRoute
// - pkt.GSOOptions
// - pkt.NetworkProtocolNumber
func (d *discipline) WritePacket(pkt *stack.PacketBuffer) tcpip.Error {
if d.closed.Load() == qDiscClosed {
return &tcpip.ErrClosedForSend{}
}
qd := &d.dispatchers[int(pkt.Hash)%len(d.dispatchers)]
qd.mu.Lock()
haveSpace := qd.queue.hasSpace()
if haveSpace {
qd.queue.pushBack(pkt.IncRef())
}
qd.mu.Unlock()
if !haveSpace {
return &tcpip.ErrNoBufferSpace{}
}
qd.newPacketWaker.Assert()
return nil
}
func (d *discipline) Close() {
d.closed.Store(qDiscClosed)
for i := range d.dispatchers {
d.dispatchers[i].closeWaker.Assert()
}
d.wg.Wait()
}
|
