mirrors/uquic
1
0
Fork 0
mirror of https://github.com/refraction-networking/uquic.git synced 2025-04-07 06:07:36 +03:00
Code Issues Projects Releases Packages Wiki Activity

remove inactive streams from the window update queue

Browse source
This commit is contained in:
Marten Seemann 2020-01-29 11:14:29 +07:00
parent 63c9272bf4
commit ddc886be7a
2 changed files with 25 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

2
window_update_queue.go
View file

@ -53,6 +53,7 @@ func (q *windowUpdateQueue) QueueAll() {
} }
// queue all stream-level window updates // queue all stream-level window updates
for id := range q.queue { for id := range q.queue {
delete(q.queue, id)
str, err := q.streamGetter.GetOrOpenReceiveStream(id) str, err := q.streamGetter.GetOrOpenReceiveStream(id)
if err != nil || str == nil { // the stream can be nil if it was completed before dequeing the window update if err != nil || str == nil { // the stream can be nil if it was completed before dequeing the window update
continue continue
@ -65,7 +66,6 @@ func (q *windowUpdateQueue) QueueAll() {
StreamID: id, StreamID: id,
ByteOffset: offset, ByteOffset: offset,
}) })
delete(q.queue, id)
} }
q.mutex.Unlock() q.mutex.Unlock()
} }

26
window_update_queue_test.go
View file

@ -52,8 +52,18 @@ var _ = Describe("Window Update Queue", func() {
}) })
It("doesn't queue a MAX_STREAM_DATA for a closed stream", func() { It("doesn't queue a MAX_STREAM_DATA for a closed stream", func() {
streamGetter.EXPECT().GetOrOpenReceiveStream(protocol.StreamID(12)).Return(nil, nil)
q.AddStream(12) q.AddStream(12)
streamGetter.EXPECT().GetOrOpenReceiveStream(protocol.StreamID(12)).Return(nil, nil)
q.QueueAll()
Expect(queuedFrames).To(BeEmpty())
})
It("removes closed streams from the queue", func() {
q.AddStream(12)
streamGetter.EXPECT().GetOrOpenReceiveStream(protocol.StreamID(12)).Return(nil, nil)
q.QueueAll()
Expect(queuedFrames).To(BeEmpty())
// don't EXPECT any further calls to GetOrOpenReceiveStream
q.QueueAll() q.QueueAll()
Expect(queuedFrames).To(BeEmpty()) Expect(queuedFrames).To(BeEmpty())
}) })
@ -61,8 +71,20 @@ var _ = Describe("Window Update Queue", func() {
It("doesn't queue a MAX_STREAM_DATA if the flow controller returns an offset of 0", func() { It("doesn't queue a MAX_STREAM_DATA if the flow controller returns an offset of 0", func() {
stream5 := NewMockStreamI(mockCtrl) stream5 := NewMockStreamI(mockCtrl)
stream5.EXPECT().getWindowUpdate().Return(protocol.ByteCount(0)) stream5.EXPECT().getWindowUpdate().Return(protocol.ByteCount(0))
streamGetter.EXPECT().GetOrOpenReceiveStream(protocol.StreamID(5)).Return(stream5, nil)
q.AddStream(5) q.AddStream(5)
streamGetter.EXPECT().GetOrOpenReceiveStream(protocol.StreamID(5)).Return(stream5, nil)
q.QueueAll()
Expect(queuedFrames).To(BeEmpty())
})
It("removes streams for which the flow controller returns an offset of 0 from the queue", func() {
stream5 := NewMockStreamI(mockCtrl)
stream5.EXPECT().getWindowUpdate().Return(protocol.ByteCount(0))
q.AddStream(5)
streamGetter.EXPECT().GetOrOpenReceiveStream(protocol.StreamID(5)).Return(stream5, nil)
q.QueueAll()
Expect(queuedFrames).To(BeEmpty())
// don't EXPECT any further calls to GetOrOpenReveiveStream and to getWindowUpdate
q.QueueAll() q.QueueAll()
Expect(queuedFrames).To(BeEmpty()) Expect(queuedFrames).To(BeEmpty())
}) })