forked from quic-go/quic-go
/
streams_map_incoming_uni.go
192 lines (171 loc) · 5.45 KB
/
streams_map_incoming_uni.go
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
// This file was automatically generated by genny.
// Any changes will be lost if this file is regenerated.
// see https://github.com/cheekybits/genny
package quic
import (
"context"
"sync"
"github.com/IoTPanic/quic-go/internal/protocol"
"github.com/IoTPanic/quic-go/internal/wire"
)
// When a stream is deleted before it was accepted, we can't delete it from the map immediately.
// We need to wait until the application accepts it, and delete it then.
type receiveStreamIEntry struct {
stream receiveStreamI
shouldDelete bool
}
type incomingUniStreamsMap struct {
mutex sync.RWMutex
newStreamChan chan struct{}
streams map[protocol.StreamNum]receiveStreamIEntry
nextStreamToAccept protocol.StreamNum // the next stream that will be returned by AcceptStream()
nextStreamToOpen protocol.StreamNum // the highest stream that the peer opened
maxStream protocol.StreamNum // the highest stream that the peer is allowed to open
maxNumStreams uint64 // maximum number of streams
newStream func(protocol.StreamNum) receiveStreamI
queueMaxStreamID func(*wire.MaxStreamsFrame)
closeErr error
}
func newIncomingUniStreamsMap(
newStream func(protocol.StreamNum) receiveStreamI,
maxStreams uint64,
queueControlFrame func(wire.Frame),
) *incomingUniStreamsMap {
return &incomingUniStreamsMap{
newStreamChan: make(chan struct{}, 1),
streams: make(map[protocol.StreamNum]receiveStreamIEntry),
maxStream: protocol.StreamNum(maxStreams),
maxNumStreams: maxStreams,
newStream: newStream,
nextStreamToOpen: 1,
nextStreamToAccept: 1,
queueMaxStreamID: func(f *wire.MaxStreamsFrame) { queueControlFrame(f) },
}
}
func (m *incomingUniStreamsMap) AcceptStream(ctx context.Context) (receiveStreamI, error) {
// drain the newStreamChan, so we don't check the map twice if the stream doesn't exist
select {
case <-m.newStreamChan:
default:
}
m.mutex.Lock()
var num protocol.StreamNum
var entry receiveStreamIEntry
for {
num = m.nextStreamToAccept
if m.closeErr != nil {
m.mutex.Unlock()
return nil, m.closeErr
}
var ok bool
entry, ok = m.streams[num]
if ok {
break
}
m.mutex.Unlock()
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-m.newStreamChan:
}
m.mutex.Lock()
}
m.nextStreamToAccept++
// If this stream was completed before being accepted, we can delete it now.
if entry.shouldDelete {
if err := m.deleteStream(num); err != nil {
m.mutex.Unlock()
return nil, err
}
}
m.mutex.Unlock()
return entry.stream, nil
}
func (m *incomingUniStreamsMap) GetOrOpenStream(num protocol.StreamNum) (receiveStreamI, error) {
m.mutex.RLock()
if num > m.maxStream {
m.mutex.RUnlock()
return nil, streamError{
message: "peer tried to open stream %d (current limit: %d)",
nums: []protocol.StreamNum{num, m.maxStream},
}
}
// if the num is smaller than the highest we accepted
// * this stream exists in the map, and we can return it, or
// * this stream was already closed, then we can return the nil
if num < m.nextStreamToOpen {
var s receiveStreamI
// If the stream was already queued for deletion, and is just waiting to be accepted, don't return it.
if entry, ok := m.streams[num]; ok && !entry.shouldDelete {
s = entry.stream
}
m.mutex.RUnlock()
return s, nil
}
m.mutex.RUnlock()
m.mutex.Lock()
// no need to check the two error conditions from above again
// * maxStream can only increase, so if the id was valid before, it definitely is valid now
// * highestStream is only modified by this function
for newNum := m.nextStreamToOpen; newNum <= num; newNum++ {
m.streams[newNum] = receiveStreamIEntry{stream: m.newStream(newNum)}
select {
case m.newStreamChan <- struct{}{}:
default:
}
}
m.nextStreamToOpen = num + 1
entry := m.streams[num]
m.mutex.Unlock()
return entry.stream, nil
}
func (m *incomingUniStreamsMap) DeleteStream(num protocol.StreamNum) error {
m.mutex.Lock()
defer m.mutex.Unlock()
return m.deleteStream(num)
}
func (m *incomingUniStreamsMap) deleteStream(num protocol.StreamNum) error {
if _, ok := m.streams[num]; !ok {
return streamError{
message: "tried to delete unknown incoming stream %d",
nums: []protocol.StreamNum{num},
}
}
// Don't delete this stream yet, if it was not yet accepted.
// Just save it to streamsToDelete map, to make sure it is deleted as soon as it gets accepted.
if num >= m.nextStreamToAccept {
entry, ok := m.streams[num]
if ok && entry.shouldDelete {
return streamError{
message: "tried to delete incoming stream %d multiple times",
nums: []protocol.StreamNum{num},
}
}
entry.shouldDelete = true
m.streams[num] = entry // can't assign to struct in map, so we need to reassign
return nil
}
delete(m.streams, num)
// queue a MAX_STREAM_ID frame, giving the peer the option to open a new stream
if m.maxNumStreams > uint64(len(m.streams)) {
maxStream := m.nextStreamToOpen + protocol.StreamNum(m.maxNumStreams-uint64(len(m.streams))) - 1
// Never send a value larger than protocol.MaxStreamCount.
if maxStream <= protocol.MaxStreamCount {
m.maxStream = maxStream
m.queueMaxStreamID(&wire.MaxStreamsFrame{
Type: protocol.StreamTypeUni,
MaxStreamNum: m.maxStream,
})
}
}
return nil
}
func (m *incomingUniStreamsMap) CloseWithError(err error) {
m.mutex.Lock()
m.closeErr = err
for _, entry := range m.streams {
entry.stream.closeForShutdown(err)
}
m.mutex.Unlock()
close(m.newStreamChan)
}