return OpenStreamSync calls in the order they were called

streams_map_outgoing_bidi.go

@@ -13,7 +13,8 @@ import (
 
 type outgoingBidiStreamsMap struct {
 	mutex sync.RWMutex
-	cond  sync.Cond
+
+	openQueue []chan struct{}
 
 	streams map[protocol.StreamNum]streamI
 
@@ -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) {
@@ -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 m.nextStream > m.maxStream {
+			// no stream available. Continue waiting
+			continue
 		}
-		if err != nil && err != errTooManyOpenStreams {
-			return nil, streamOpenErr{err}
-		}
-		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) {
@@ -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) {
@@ -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()
 }

streams_map_outgoing_generic.go

@@ -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
 
@@ -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) {
@@ -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 m.nextStream > m.maxStream {
+			// no stream available. Continue waiting
+			continue
 		}
-		if err != nil && err != errTooManyOpenStreams {
-			return nil, streamOpenErr{err}
-		}
-		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) {
@@ -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) {
@@ -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()
 }

streams_map_outgoing_generic_test.go

@@ -122,6 +122,107 @@ var _ = Describe("Streams Map (outgoing)", func() {
 			Eventually(done).Should(BeClosed())
 		})
 
+		It("opens streams in the right order", func() {
+			mockSender.EXPECT().queueControlFrame(gomock.Any()).AnyTimes()
+			done1 := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				str, err := m.OpenStreamSync()
+				Expect(err).ToNot(HaveOccurred())
+				Expect(str.(*mockGenericStream).num).To(Equal(protocol.StreamNum(1)))
+				close(done1)
+			}()
+			Consistently(done1).ShouldNot(BeClosed())
+			done2 := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				str, err := m.OpenStreamSync()
+				Expect(err).ToNot(HaveOccurred())
+				Expect(str.(*mockGenericStream).num).To(Equal(protocol.StreamNum(2)))
+				close(done2)
+			}()
+			Consistently(done2).ShouldNot(BeClosed())
+
+			m.SetMaxStream(1)
+			Eventually(done1).Should(BeClosed())
+			Consistently(done2).ShouldNot(BeClosed())
+			m.SetMaxStream(2)
+			Eventually(done2).Should(BeClosed())
+		})
+
+		It("unblocks multiple OpenStreamSync calls at the same time", func() {
+			mockSender.EXPECT().queueControlFrame(gomock.Any()).AnyTimes()
+			done := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				_, err := m.OpenStreamSync()
+				Expect(err).ToNot(HaveOccurred())
+				done <- struct{}{}
+			}()
+			go func() {
+				defer GinkgoRecover()
+				_, err := m.OpenStreamSync()
+				Expect(err).ToNot(HaveOccurred())
+				done <- struct{}{}
+			}()
+			Consistently(done).ShouldNot(Receive())
+			go func() {
+				defer GinkgoRecover()
+				_, err := m.OpenStreamSync()
+				Expect(err).To(MatchError("test done"))
+				done <- struct{}{}
+			}()
+			Consistently(done).ShouldNot(Receive())
+
+			m.SetMaxStream(2)
+			Eventually(done).Should(Receive())
+			Eventually(done).Should(Receive())
+			Consistently(done).ShouldNot(Receive())
+
+			m.CloseWithError(errors.New("test done"))
+			Eventually(done).Should(Receive())
+		})
+
+		It("returns an error for OpenStream while an OpenStreamSync call is blocking", func() {
+			mockSender.EXPECT().queueControlFrame(gomock.Any()).MaxTimes(2)
+			openedSync := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				str, err := m.OpenStreamSync()
+				Expect(err).ToNot(HaveOccurred())
+				Expect(str.(*mockGenericStream).num).To(Equal(protocol.StreamNum(1)))
+				close(openedSync)
+			}()
+			Consistently(openedSync).ShouldNot(BeClosed())
+
+			start := make(chan struct{})
+			openend := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				var hasStarted bool
+				for {
+					str, err := m.OpenStream()
+					if err == nil {
+						Expect(str.(*mockGenericStream).num).To(Equal(protocol.StreamNum(2)))
+						close(openend)
+						return
+					}
+					expectTooManyStreamsError(err)
+					if !hasStarted {
+						close(start)
+						hasStarted = true
+					}
+				}
+			}()
+
+			Eventually(start).Should(BeClosed())
+			m.SetMaxStream(1)
+			Eventually(openedSync).Should(BeClosed())
+			Consistently(openend).ShouldNot(BeClosed())
+			m.SetMaxStream(2)
+			Eventually(openend).Should(BeClosed())
+		})
+
 		It("stops opening synchronously when it is closed", func() {
 			mockSender.EXPECT().queueControlFrame(gomock.Any())
 			testErr := errors.New("test error")

streams_map_outgoing_uni.go

@@ -13,7 +13,8 @@ import (
 
 type outgoingUniStreamsMap struct {
 	mutex sync.RWMutex
-	cond  sync.Cond
+
+	openQueue []chan struct{}
 
 	streams map[protocol.StreamNum]sendStreamI
 
@@ -31,15 +32,13 @@ func newOutgoingUniStreamsMap(
 	newStream func(protocol.StreamNum) sendStreamI,
 	queueControlFrame func(wire.Frame),
 ) *outgoingUniStreamsMap {
-	m := &outgoingUniStreamsMap{
+	return &outgoingUniStreamsMap{
 		streams:              make(map[protocol.StreamNum]sendStreamI),
 		maxStream:            protocol.InvalidStreamNum,
 		nextStream:           1,
 		newStream:            newStream,
 		queueStreamIDBlocked: func(f *wire.StreamsBlockedFrame) { queueControlFrame(f) },
 	}
-	m.cond.L = &m.mutex
-	return m
 }
 
 func (m *outgoingUniStreamsMap) OpenStream() (sendStreamI, error) {
@@ -50,51 +49,70 @@ func (m *outgoingUniStreamsMap) OpenStream() (sendStreamI, 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 *outgoingUniStreamsMap) OpenStreamSync() (sendStreamI, 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 m.nextStream > m.maxStream {
+			// no stream available. Continue waiting
+			continue
 		}
-		if err != nil && err != errTooManyOpenStreams {
-			return nil, streamOpenErr{err}
-		}
-		m.cond.Wait()
+		str := m.openStream()
+		m.openQueue = m.openQueue[1:]
+		m.unblockOpenSync()
+		return str, nil
 	}
 }
 
-func (m *outgoingUniStreamsMap) openStreamImpl() (sendStreamI, 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.StreamTypeUni,
-				StreamLimit: streamNum,
-			})
-			m.blockedSent = true
-		}
-		return nil, errTooManyOpenStreams
-	}
+func (m *outgoingUniStreamsMap) openStream() sendStreamI {
 	s := m.newStream(m.nextStream)
 	m.streams[m.nextStream] = s
 	m.nextStream++
-	return s, nil
+	return s
+}
+
+func (m *outgoingUniStreamsMap) maybeSendBlockedFrame() {
+	if m.blockedSent {
+		return
+	}
+
+	var streamNum protocol.StreamNum
+	if m.maxStream != protocol.InvalidStreamNum {
+		streamNum = m.maxStream
+	}
+	m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
+		Type:        protocol.StreamTypeUni,
+		StreamLimit: streamNum,
+	})
+	m.blockedSent = true
 }
 
 func (m *outgoingUniStreamsMap) GetStream(num protocol.StreamNum) (sendStreamI, error) {
@@ -127,12 +145,24 @@ func (m *outgoingUniStreamsMap) DeleteStream(num protocol.StreamNum) error {
 
 func (m *outgoingUniStreamsMap) 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 *outgoingUniStreamsMap) unblockOpenSync() {
+	if len(m.openQueue) == 0 {
+		return
+	}
+	select {
+	case m.openQueue[0] <- struct{}{}:
+	default:
 	}
-	m.mutex.Unlock()
 }
 
 func (m *outgoingUniStreamsMap) CloseWithError(err error) {
@@ -141,6 +171,8 @@ func (m *outgoingUniStreamsMap) CloseWithError(err error) {
 	for _, str := range m.streams {
 		str.closeForShutdown(err)
 	}
-	m.cond.Broadcast()
+	for _, c := range m.openQueue {
+		close(c)
+	}
 	m.mutex.Unlock()
 }