Skip to content

Commit

Permalink
Merge pull request quic-go#1950 from lucas-clemente/openstreamsync-in…
Browse files Browse the repository at this point in the history 
…-order

return OpenStreamSync calls in the order they were called
  • Loading branch information
@marten-seemann
marten-seemann authored Jun 7, 2019
2 parents e7f20d4 + 061b92e commit d0b5cc1
Show file tree
Hide file tree
Showing 4 changed files with 308 additions and 111 deletions.
106 changes: 69 additions & 37 deletions streams_map_outgoing_bidi.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,8 @@ import (

type outgoingBidiStreamsMap struct {
mutex sync.RWMutex
cond sync.Cond

openQueue []chan struct{}

streams map[protocol.StreamNum]streamI

Expand All @@ -31,15 +32,13 @@ func newOutgoingBidiStreamsMap(
newStream func(protocol.StreamNum) streamI,
queueControlFrame func(wire.Frame),
) *outgoingBidiStreamsMap {
m := &outgoingBidiStreamsMap{
return &outgoingBidiStreamsMap{
streams: make(map[protocol.StreamNum]streamI),
maxStream: protocol.InvalidStreamNum,
nextStream: 1,
newStream: newStream,
queueStreamIDBlocked: func(f *wire.StreamsBlockedFrame) { queueControlFrame(f) },
}
m.cond.L = &m.mutex
return m
}

func (m *outgoingBidiStreamsMap) OpenStream() (streamI, error) {
Expand All @@ -50,51 +49,70 @@ func (m *outgoingBidiStreamsMap) OpenStream() (streamI, error) {
return nil, m.closeErr
}

str, err := m.openStreamImpl()
if err != nil {
return nil, streamOpenErr{err}
// if there are OpenStreamSync calls waiting, return an error here
if len(m.openQueue) > 0 || m.nextStream > m.maxStream {
m.maybeSendBlockedFrame()
return nil, streamOpenErr{errTooManyOpenStreams}
}
return str, nil
return m.openStream(), nil
}

func (m *outgoingBidiStreamsMap) OpenStreamSync() (streamI, error) {
m.mutex.Lock()
defer m.mutex.Unlock()

if m.closeErr != nil {
return nil, m.closeErr
}

if len(m.openQueue) == 0 && m.nextStream <= m.maxStream {
return m.openStream(), nil
}

waitChan := make(chan struct{}, 1)
m.openQueue = append(m.openQueue, waitChan)
m.maybeSendBlockedFrame()

for {
m.mutex.Unlock()
<-waitChan
m.mutex.Lock()

if m.closeErr != nil {
return nil, m.closeErr
}
str, err := m.openStreamImpl()
if err == nil {
return str, nil
}
if err != nil && err != errTooManyOpenStreams {
return nil, streamOpenErr{err}
if m.nextStream > m.maxStream {
// no stream available. Continue waiting
continue
}
m.cond.Wait()
str := m.openStream()
m.openQueue = m.openQueue[1:]
m.unblockOpenSync()
return str, nil
}
}

func (m *outgoingBidiStreamsMap) openStreamImpl() (streamI, error) {
if m.nextStream > m.maxStream {
if !m.blockedSent {
var streamNum protocol.StreamNum
if m.maxStream != protocol.InvalidStreamNum {
streamNum = m.maxStream
}
m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
Type: protocol.StreamTypeBidi,
StreamLimit: streamNum,
})
m.blockedSent = true
}
return nil, errTooManyOpenStreams
}
func (m *outgoingBidiStreamsMap) openStream() streamI {
s := m.newStream(m.nextStream)
m.streams[m.nextStream] = s
m.nextStream++
return s, nil
return s
}

func (m *outgoingBidiStreamsMap) maybeSendBlockedFrame() {
if m.blockedSent {
return
}

var streamNum protocol.StreamNum
if m.maxStream != protocol.InvalidStreamNum {
streamNum = m.maxStream
}
m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
Type: protocol.StreamTypeBidi,
StreamLimit: streamNum,
})
m.blockedSent = true
}

func (m *outgoingBidiStreamsMap) GetStream(num protocol.StreamNum) (streamI, error) {
Expand Down Expand Up @@ -127,12 +145,24 @@ func (m *outgoingBidiStreamsMap) DeleteStream(num protocol.StreamNum) error {

func (m *outgoingBidiStreamsMap) SetMaxStream(num protocol.StreamNum) {
m.mutex.Lock()
if num > m.maxStream {
m.maxStream = num
m.blockedSent = false
m.cond.Broadcast()
defer m.mutex.Unlock()

if num <= m.maxStream {
return
}
m.maxStream = num
m.blockedSent = false
m.unblockOpenSync()
}

func (m *outgoingBidiStreamsMap) unblockOpenSync() {
if len(m.openQueue) == 0 {
return
}
select {
case m.openQueue[0] <- struct{}{}:
default:
}
m.mutex.Unlock()
}

func (m *outgoingBidiStreamsMap) CloseWithError(err error) {
Expand All @@ -141,6 +171,8 @@ func (m *outgoingBidiStreamsMap) CloseWithError(err error) {
for _, str := range m.streams {
str.closeForShutdown(err)
}
m.cond.Broadcast()
for _, c := range m.openQueue {
close(c)
}
m.mutex.Unlock()
}
106 changes: 69 additions & 37 deletions streams_map_outgoing_generic.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@ import (
//go:generate genny -in $GOFILE -out streams_map_outgoing_uni.go gen "item=sendStreamI Item=UniStream streamTypeGeneric=protocol.StreamTypeUni"
type outgoingItemsMap struct {
mutex sync.RWMutex
cond sync.Cond

openQueue []chan struct{}

streams map[protocol.StreamNum]item

Expand All @@ -29,15 +30,13 @@ func newOutgoingItemsMap(
newStream func(protocol.StreamNum) item,
queueControlFrame func(wire.Frame),
) *outgoingItemsMap {
m := &outgoingItemsMap{
return &outgoingItemsMap{
streams: make(map[protocol.StreamNum]item),
maxStream: protocol.InvalidStreamNum,
nextStream: 1,
newStream: newStream,
queueStreamIDBlocked: func(f *wire.StreamsBlockedFrame) { queueControlFrame(f) },
}
m.cond.L = &m.mutex
return m
}

func (m *outgoingItemsMap) OpenStream() (item, error) {
Expand All @@ -48,51 +47,70 @@ func (m *outgoingItemsMap) OpenStream() (item, error) {
return nil, m.closeErr
}

str, err := m.openStreamImpl()
if err != nil {
return nil, streamOpenErr{err}
// if there are OpenStreamSync calls waiting, return an error here
if len(m.openQueue) > 0 || m.nextStream > m.maxStream {
m.maybeSendBlockedFrame()
return nil, streamOpenErr{errTooManyOpenStreams}
}
return str, nil
return m.openStream(), nil
}

func (m *outgoingItemsMap) OpenStreamSync() (item, error) {
m.mutex.Lock()
defer m.mutex.Unlock()

if m.closeErr != nil {
return nil, m.closeErr
}

if len(m.openQueue) == 0 && m.nextStream <= m.maxStream {
return m.openStream(), nil
}

waitChan := make(chan struct{}, 1)
m.openQueue = append(m.openQueue, waitChan)
m.maybeSendBlockedFrame()

for {
m.mutex.Unlock()
<-waitChan
m.mutex.Lock()

if m.closeErr != nil {
return nil, m.closeErr
}
str, err := m.openStreamImpl()
if err == nil {
return str, nil
}
if err != nil && err != errTooManyOpenStreams {
return nil, streamOpenErr{err}
if m.nextStream > m.maxStream {
// no stream available. Continue waiting
continue
}
m.cond.Wait()
str := m.openStream()
m.openQueue = m.openQueue[1:]
m.unblockOpenSync()
return str, nil
}
}

func (m *outgoingItemsMap) openStreamImpl() (item, error) {
if m.nextStream > m.maxStream {
if !m.blockedSent {
var streamNum protocol.StreamNum
if m.maxStream != protocol.InvalidStreamNum {
streamNum = m.maxStream
}
m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
Type: streamTypeGeneric,
StreamLimit: streamNum,
})
m.blockedSent = true
}
return nil, errTooManyOpenStreams
}
func (m *outgoingItemsMap) openStream() item {
s := m.newStream(m.nextStream)
m.streams[m.nextStream] = s
m.nextStream++
return s, nil
return s
}

func (m *outgoingItemsMap) maybeSendBlockedFrame() {
if m.blockedSent {
return
}

var streamNum protocol.StreamNum
if m.maxStream != protocol.InvalidStreamNum {
streamNum = m.maxStream
}
m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
Type: streamTypeGeneric,
StreamLimit: streamNum,
})
m.blockedSent = true
}

func (m *outgoingItemsMap) GetStream(num protocol.StreamNum) (item, error) {
Expand Down Expand Up @@ -125,12 +143,24 @@ func (m *outgoingItemsMap) DeleteStream(num protocol.StreamNum) error {

func (m *outgoingItemsMap) SetMaxStream(num protocol.StreamNum) {
m.mutex.Lock()
if num > m.maxStream {
m.maxStream = num
m.blockedSent = false
m.cond.Broadcast()
defer m.mutex.Unlock()

if num <= m.maxStream {
return
}
m.maxStream = num
m.blockedSent = false
m.unblockOpenSync()
}

func (m *outgoingItemsMap) unblockOpenSync() {
if len(m.openQueue) == 0 {
return
}
select {
case m.openQueue[0] <- struct{}{}:
default:
}
m.mutex.Unlock()
}

func (m *outgoingItemsMap) CloseWithError(err error) {
Expand All @@ -139,6 +169,8 @@ func (m *outgoingItemsMap) CloseWithError(err error) {
for _, str := range m.streams {
str.closeForShutdown(err)
}
m.cond.Broadcast()
for _, c := range m.openQueue {
close(c)
}
m.mutex.Unlock()
}
Loading

0 comments on commit d0b5cc1

Please sign in to comment.