gr1ffon/quic-go
1
0
Fork 0
forked from quic-go/quic-go
Code Pull Requests Activity

implement round-robin scheduling in StreamFrameQueue

Browse Source 
fixes #148
This commit is contained in:
Marten Seemann
2016-05-27 13:38:39 +07:00
parent cdcd120d49
commit 26d18976ff
4 changed files with 221 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
stream_frame_queue_test.go
View File

@@ -9,11 +9,11 @@ import (
var _ = Describe("streamFrameQueue", func() {
var prioFrame1, prioFrame2 *frames.StreamFrame
var frame1, frame2 *frames.StreamFrame
var frame1, frame2, frame3 *frames.StreamFrame
var queue *streamFrameQueue
BeforeEach(func() {
queue = &streamFrameQueue{}
queue = newStreamFrameQueue()
prioFrame1 = &frames.StreamFrame{
StreamID: 5,
Data: []byte{0x13, 0x37},
@@ -30,13 +30,17 @@ var _ = Describe("streamFrameQueue", func() {
StreamID: 11,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF, 0x37},
}
frame3 = &frames.StreamFrame{
StreamID: 11,
Data: []byte{0xBE, 0xEF},
}
})
It("sets the DataLenPresent on all StreamFrames", func() {
queue.Push(frame1, false)
queue.Push(prioFrame1, true)
Expect(queue.frames[0].DataLenPresent).To(BeTrue())
Expect(queue.prioFrames[0].DataLenPresent).To(BeTrue())
Expect(queue.frameMap[frame1.StreamID][0].DataLenPresent).To(BeTrue())
})
Context("Queue Length", func() {
@@ -61,14 +65,6 @@ var _ = Describe("streamFrameQueue", func() {
queue.Pop(1000)
Expect(queue.Len()).To(Equal(0))
})
It("does not change the length when using front()", func() {
queue.Push(prioFrame1, true)
queue.Push(frame1, false)
Expect(queue.Len()).To(Equal(2))
queue.front()
Expect(queue.Len()).To(Equal(2))
})
})
Context("Queue Byte Length", func() {
@@ -92,15 +88,66 @@ var _ = Describe("streamFrameQueue", func() {
queue.Pop(1000)
Expect(queue.ByteLen()).To(Equal(protocol.ByteCount(0)))
})
})
It("does not change the byte length when using front()", func() {
Context("Pushing", func() {
It("adds the streams to the map", func() {
queue.Push(frame1, false)
Expect(queue.frameMap).To(HaveKey(frame1.StreamID))
Expect(queue.frameMap[frame1.StreamID][0]).To(Equal(frame1))
})
It("only adds a StreamID once to the active stream list", func() {
queue.Push(frame1, false)
queue.Push(frame1, false)
Expect(queue.frameMap).To(HaveKey(frame1.StreamID))
Expect(queue.frameMap[frame1.StreamID]).To(HaveLen(2))
Expect(queue.activeStreams).To(HaveLen(1))
Expect(queue.activeStreams[0]).To(Equal(frame1.StreamID))
})
})
Context("getNextStream", func() {
It("returns 0 for an empty queue", func() {
streamID, err := queue.getNextStream()
Expect(err).ToNot(HaveOccurred())
Expect(streamID).To(BeZero())
})
It("does not change the byte length when using getNextStream()", func() {
queue.Push(prioFrame1, true)
queue.Push(frame1, false)
length := protocol.ByteCount(len(prioFrame1.Data) + len(frame1.Data))
Expect(queue.ByteLen()).To(Equal(length))
queue.front()
_, err := queue.getNextStream()
Expect(err).ToNot(HaveOccurred())
Expect(queue.ByteLen()).To(Equal(length))
})
It("does not change the length when using front()", func() {
queue.Push(prioFrame1, true)
queue.Push(frame1, false)
Expect(queue.Len()).To(Equal(2))
_, err := queue.getNextStream()
Expect(err).ToNot(HaveOccurred())
Expect(queue.Len()).To(Equal(2))
})
It("returns normal frames if no prio frames are available", func() {
queue.Push(frame1, false)
queue.Push(frame2, false)
streamID, err := queue.getNextStream()
Expect(err).ToNot(HaveOccurred())
Expect(streamID).To(Equal(frame1.StreamID))
})
It("gets the frame inserted at first at first", func() {
queue.Push(frame2, false)
queue.Push(frame1, false)
streamID, err := queue.getNextStream()
Expect(err).ToNot(HaveOccurred())
Expect(streamID).To(Equal(frame2.StreamID))
})
})
Context("Popping", func() {
@@ -110,15 +157,23 @@ var _ = Describe("streamFrameQueue", func() {
It("deletes elements once they are popped", func() {
queue.Push(frame1, false)
Expect(queue.Pop(1000)).To(Equal(frame1))
Expect(queue.Pop(1000)).To(BeNil())
frame, err := queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame1))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
})
It("returns normal frames if no prio frames are available", func() {
queue.Push(frame1, false)
queue.Push(frame2, false)
Expect(queue.Pop(1000)).To(Equal(frame1))
Expect(queue.Pop(1000)).To(Equal(frame2))
frame, err := queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame1))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame2))
})
It("returns prio frames first", func() {
@@ -126,14 +181,63 @@ var _ = Describe("streamFrameQueue", func() {
queue.Push(frame1, false)
queue.Push(frame2, false)
queue.Push(prioFrame2, true)
frame := queue.Pop(1000)
frame, err := queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(prioFrame1))
frame = queue.Pop(1000)
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(prioFrame2))
frame = queue.Pop(1000)
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame1))
})
Context("scheduling", func() {
It("goes around", func() {
queue.Push(frame2, false) // StreamID: 11
queue.Push(frame3, false) // StreamID: 11
queue.Push(frame1, false) // StreamID: 10
frame, err := queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame2))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame1))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame3))
})
It("starts with the frame inserted first", func() {
queue.Push(frame1, false) // StreamID: 10
queue.Push(frame2, false) // StreamID: 11
queue.Push(frame3, false) // StreamID: 11
frame, err := queue.Pop(1000)
Expect(frame).To(Equal(frame1))
Expect(err).ToNot(HaveOccurred())
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame2))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame3))
})
It("goes around, also when frame have to be split", func() {
queue.Push(frame2, false) // StreamID: 11
queue.Push(frame1, false) // StreamID: 10
frame, err := queue.Pop(5)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(frame2.StreamID))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(frame1))
frame, err = queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(frame2.StreamID))
})
})
Context("splitting of frames", func() {
It("splits off nothing", func() {
f := &frames.StreamFrame{
@@ -172,20 +276,23 @@ var _ = Describe("streamFrameQueue", func() {
It("splits a frame", func() {
queue.Push(frame1, false)
origlen := len(frame1.Data)
frame := queue.Pop(6)
frame, err := queue.Pop(6)
Expect(err).ToNot(HaveOccurred())
minLength, _ := frame.MinLength()
Expect(int(minLength) - 1 + len(frame.Data)).To(Equal(6))
Expect(queue.frames[0].Data).To(HaveLen(origlen - len(frame.Data)))
Expect(queue.frames[0].Offset).To(Equal(protocol.ByteCount(len(frame.Data))))
Expect(queue.frameMap[frame1.StreamID][0].Data).To(HaveLen(origlen - len(frame.Data)))
Expect(queue.frameMap[frame1.StreamID][0].Offset).To(Equal(protocol.ByteCount(len(frame.Data))))
})
It("only removes a frame from the queue after return all split parts", func() {
queue.Push(frame1, false)
Expect(queue.Len()).To(Equal(1))
frame := queue.Pop(6)
frame, err := queue.Pop(6)
Expect(err).ToNot(HaveOccurred())
Expect(frame).ToNot(BeNil())
Expect(queue.Len()).To(Equal(1))
frame = queue.Pop(100)
frame, err = queue.Pop(100)
Expect(err).ToNot(HaveOccurred())
Expect(frame).ToNot(BeNil())
Expect(queue.Len()).To(BeZero())
})
@@ -195,8 +302,10 @@ var _ = Describe("streamFrameQueue", func() {
origdata := make([]byte, length)
copy(origdata, frame1.Data)
queue.Push(frame1, false)
frame := queue.Pop(6)
nextframe := queue.Pop(1000)
frame, err := queue.Pop(6)
Expect(err).ToNot(HaveOccurred())
nextframe, err := queue.Pop(1000)
Expect(err).ToNot(HaveOccurred())
Expect(len(frame.Data) + len(nextframe.Data)).To(Equal(length))
data := make([]byte, length)
copy(data, frame.Data)
@@ -208,7 +317,8 @@ var _ = Describe("streamFrameQueue", func() {
queue.Push(frame1, false)
queue.Push(frame2, false)
startByteLength := queue.ByteLen()
frame := queue.Pop(6)
frame, err := queue.Pop(6)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(frame1.StreamID)) // make sure the right frame was popped
Expect(queue.ByteLen()).To(Equal(startByteLength - protocol.ByteCount(len(frame.Data))))
})
@@ -216,37 +326,11 @@ var _ = Describe("streamFrameQueue", func() {
It("does not change the length of the queue when returning a split frame", func() {
queue.Push(frame1, false)
queue.Push(frame2, false)
frame := queue.Pop(6)
frame, err := queue.Pop(6)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(frame1.StreamID)) // make sure the right frame was popped
Expect(queue.Len()).To(Equal(2))
})
})
})
Context("Front", func() {
It("returns nil for an empty queue", func() {
Expect(queue.front()).To(BeNil())
})
It("returns normal frames if no prio frames are available", func() {
queue.Push(frame1, false)
queue.Push(frame2, false)
frame, isPrioFrame := queue.front()
Expect(isPrioFrame).To(BeFalse())
Expect(frame).To(Equal(frame1))
frame, isPrioFrame = queue.front()
Expect(isPrioFrame).To(BeFalse())
Expect(frame).To(Equal(frame1))
})
It("returns prio frames first", func() {
queue.Push(prioFrame1, true)
queue.Push(frame1, false)
queue.Push(frame2, false)
queue.Push(prioFrame2, true)
frame, isPrioFrame := queue.front()
Expect(isPrioFrame).To(BeTrue())
Expect(frame).To(Equal(prioFrame1))
})
})
})