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
// 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 (
"fmt"
"sync"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
)
type incomingBidiStreamsMap struct {
mutex sync.RWMutex
cond sync.Cond
streams map[protocol.StreamID]streamI
nextStream protocol.StreamID // the next stream that will be returned by AcceptStream()
highestStream protocol.StreamID // the highest stream that the peer openend
maxStream protocol.StreamID // the highest stream that the peer is allowed to open
maxNumStreams int // maximum number of streams
newStream func(protocol.StreamID) streamI
queueMaxStreamID func(*wire.MaxStreamIDFrame)
closeErr error
}
func newIncomingBidiStreamsMap(
nextStream protocol.StreamID,
initialMaxStreamID protocol.StreamID,
maxNumStreams int,
queueControlFrame func(wire.Frame),
newStream func(protocol.StreamID) streamI,
) *incomingBidiStreamsMap {
m := &incomingBidiStreamsMap{
streams: make(map[protocol.StreamID]streamI),
nextStream: nextStream,
maxStream: initialMaxStreamID,
maxNumStreams: maxNumStreams,
newStream: newStream,
queueMaxStreamID: func(f *wire.MaxStreamIDFrame) { queueControlFrame(f) },
}
m.cond.L = &m.mutex
return m
}
func (m *incomingBidiStreamsMap) AcceptStream() (streamI, error) {
m.mutex.Lock()
defer m.mutex.Unlock()
var str streamI
for {
var ok bool
if m.closeErr != nil {
return nil, m.closeErr
}
str, ok = m.streams[m.nextStream]
if ok {
break
}
m.cond.Wait()
}
m.nextStream += 4
return str, nil
}
func (m *incomingBidiStreamsMap) GetOrOpenStream(id protocol.StreamID) (streamI, error) {
m.mutex.RLock()
if id > m.maxStream {
m.mutex.RUnlock()
return nil, fmt.Errorf("peer tried to open stream %d (current limit: %d)", id, m.maxStream)
}
// if the id 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 id <= m.highestStream {
s := m.streams[id]
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
var start protocol.StreamID
if m.highestStream == 0 {
start = m.nextStream
} else {
start = m.highestStream + 4
}
for newID := start; newID <= id; newID += 4 {
m.streams[newID] = m.newStream(newID)
m.cond.Signal()
}
m.highestStream = id
s := m.streams[id]
m.mutex.Unlock()
return s, nil
}
func (m *incomingBidiStreamsMap) DeleteStream(id protocol.StreamID) error {
m.mutex.Lock()
defer m.mutex.Unlock()
if _, ok := m.streams[id]; !ok {
return fmt.Errorf("Tried to delete unknown stream %d", id)
}
delete(m.streams, id)
// queue a MAX_STREAM_ID frame, giving the peer the option to open a new stream
if numNewStreams := m.maxNumStreams - len(m.streams); numNewStreams > 0 {
m.maxStream = m.highestStream + protocol.StreamID(numNewStreams*4)
m.queueMaxStreamID(&wire.MaxStreamIDFrame{StreamID: m.maxStream})
}
return nil
}
func (m *incomingBidiStreamsMap) CloseWithError(err error) {
m.mutex.Lock()
m.closeErr = err
m.mutex.Unlock()
m.cond.Broadcast()
}