Merge pull request #1189 from lucas-clemente/fix-1188

fix race conditions in new streams maps

streams_map_incoming_bidi.go

@@ -69,20 +69,25 @@ func (m *incomingBidiStreamsMap) AcceptStream() (streamI, error) {
 }
 
 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 {
-		m.mutex.RLock()
 		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

streams_map_incoming_generic.go

@@ -67,20 +67,25 @@ func (m *incomingItemsMap) AcceptStream() (item, error) {
 }
 
 func (m *incomingItemsMap) GetOrOpenStream(id protocol.StreamID) (item, 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 {
-		m.mutex.RLock()
 		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

streams_map_incoming_uni.go

@@ -69,20 +69,25 @@ func (m *incomingUniStreamsMap) AcceptStream() (receiveStreamI, error) {
 }
 
 func (m *incomingUniStreamsMap) GetOrOpenStream(id protocol.StreamID) (receiveStreamI, 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 {
-		m.mutex.RLock()
 		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

streams_map_outgoing_bidi.go

@@ -85,10 +85,11 @@ func (m *outgoingBidiStreamsMap) openStreamImpl() (streamI, error) {
 }
 
 func (m *outgoingBidiStreamsMap) GetStream(id protocol.StreamID) (streamI, error) {
+	m.mutex.RLock()
 	if id >= m.nextStream {
+		m.mutex.RUnlock()
 		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
 	}
-	m.mutex.RLock()
 	s := m.streams[id]
 	m.mutex.RUnlock()
 	return s, nil

streams_map_outgoing_generic.go

@@ -86,10 +86,11 @@ func (m *outgoingItemsMap) openStreamImpl() (item, error) {
 }
 
 func (m *outgoingItemsMap) GetStream(id protocol.StreamID) (item, error) {
+	m.mutex.RLock()
 	if id >= m.nextStream {
+		m.mutex.RUnlock()
 		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
 	}
-	m.mutex.RLock()
 	s := m.streams[id]
 	m.mutex.RUnlock()
 	return s, nil

streams_map_outgoing_uni.go

@@ -85,10 +85,11 @@ func (m *outgoingUniStreamsMap) openStreamImpl() (sendStreamI, error) {
 }
 
 func (m *outgoingUniStreamsMap) GetStream(id protocol.StreamID) (sendStreamI, error) {
+	m.mutex.RLock()
 	if id >= m.nextStream {
+		m.mutex.RUnlock()
 		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
 	}
-	m.mutex.RLock()
 	s := m.streams[id]
 	m.mutex.RUnlock()
 	return s, nil