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

change streamFramer to pop as many stream frames as possible at once

Browse source 
ref #217
This commit is contained in:
Lucas Clemente 2016-07-25 17:00:28 +02:00
parent 6f657c02e4
commit 7adf760fa4
4 changed files with 164 additions and 259 deletions
Download patch file Download diff file Expand all files Collapse all files

46
packet_packer.go
View file

@ -182,53 +182,23 @@ func (p *packetPacker) composeNextPacket(stopWaitingFrame *frames.StopWaitingFra
return nil, fmt.Errorf("Packet Packer BUG: packet payload (%d) too large (%d)", payloadLength, maxFrameSize)
}
hasStreamFrames := false
// temporarily increase the maxFrameSize by 2 bytes
// this leads to a properly sized packet in all cases, since we do all the packet length calculations with StreamFrames that have the DataLen set
// however, for the last StreamFrame in the packet, we can omit the DataLen, thus saving 2 bytes and yielding a packet of exactly the correct size
maxFrameSize += 2
for {
if payloadLength > maxFrameSize {
return nil, fmt.Errorf("Packet Packer BUG: packet payload (%d) too large (%d)", payloadLength, maxFrameSize)
fs := p.streamFramer.PopStreamFrames(maxFrameSize - payloadLength)
if len(fs) != 0 {
fs[len(fs)-1].DataLenPresent = false
}
frame, err := p.streamFramer.PopStreamFrame(maxFrameSize - payloadLength)
if err != nil {
return nil, err
}
if frame == nil {
break
}
frame.DataLenPresent = true // set the dataLen by default. Remove them later if applicable
frameHeaderLen, _ := frame.MinLength(p.version) // StreamFrame.MinLength *never* returns an error
payloadLength += frameHeaderLen + frame.DataLen()
blockedFrame := p.streamFramer.PopBlockedFrame()
if blockedFrame != nil {
blockedLength, _ := blockedFrame.MinLength(p.version) // BlockedFrame.MinLength *never* returns an error
if payloadLength+blockedLength <= maxFrameSize {
payloadFrames = append(payloadFrames, blockedFrame)
payloadLength += blockedLength
} else {
p.controlFrames = append(p.controlFrames, blockedFrame)
}
// TODO: Simplify
for _, f := range fs {
payloadFrames = append(payloadFrames, f)
}
payloadFrames = append(payloadFrames, frame)
hasStreamFrames = true
}
// remove the dataLen for the last StreamFrame in the packet
if hasStreamFrames {
lastStreamFrame, ok := payloadFrames[len(payloadFrames)-1].(*frames.StreamFrame)
if !ok {
return nil, errors.New("PacketPacker BUG: StreamFrame type assertion failed")
}
lastStreamFrame.DataLenPresent = false
// payloadLength -= 2
for b := p.streamFramer.PopBlockedFrame(); b != nil; b = p.streamFramer.PopBlockedFrame() {
p.controlFrames = append(p.controlFrames, b)
}
return payloadFrames, nil

75
packet_packer_test.go
View file

@ -385,7 +385,20 @@ var _ = Describe("Packet packer", func() {
})
Context("Blocked frames", func() {
It("adds a blocked frame to a packet if there is enough space", func() {
It("queues a BLOCKED frame", func() {
length := 100
streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}}
f := &frames.StreamFrame{
StreamID: 5,
Data: bytes.Repeat([]byte{'f'}, length),
}
streamFramer.AddFrameForRetransmission(f)
_, err := packer.composeNextPacket(nil, publicHeaderLen)
Expect(err).ToNot(HaveOccurred())
Expect(packer.controlFrames[0]).To(Equal(&frames.BlockedFrame{StreamID: 5}))
})
It("removes the dataLen attribute from the last StreamFrame, even if it queued a BLOCKED frame", func() {
length := 100
streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}}
f := &frames.StreamFrame{
@ -395,61 +408,8 @@ var _ = Describe("Packet packer", func() {
streamFramer.AddFrameForRetransmission(f)
p, err := packer.composeNextPacket(nil, publicHeaderLen)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(HaveLen(2))
Expect(p[0]).To(Equal(&frames.BlockedFrame{StreamID: 5}))
})
It("removes the dataLen attribute from the last StreamFrame, even if it inserted a BlockedFrame before", func() {
length := 100
streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}}
f := &frames.StreamFrame{
StreamID: 5,
Data: bytes.Repeat([]byte{'f'}, length),
}
streamFramer.AddFrameForRetransmission(f)
p, err := packer.composeNextPacket(nil, publicHeaderLen)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(HaveLen(2))
Expect(p[1].(*frames.StreamFrame).DataLenPresent).To(BeFalse())
})
It("packs a BlockedFrame in the next packet if the current packet doesn't have enough space", func() {
dataLen := int(protocol.MaxFrameAndPublicHeaderSize-publicHeaderLen) - (1 + 1 + 2) + 1
streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}}
f := &frames.StreamFrame{
StreamID: 5,
Data: bytes.Repeat([]byte{'f'}, dataLen),
}
streamFramer.AddFrameForRetransmission(f)
p, err := packer.composeNextPacket(nil, publicHeaderLen)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(HaveLen(1))
p, err = packer.composeNextPacket(nil, publicHeaderLen)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(HaveLen(1))
Expect(p[0]).To(Equal(&frames.BlockedFrame{StreamID: 5}))
})
It("packs a packet with the maximum size with a BlockedFrame", func() {
blockedFrame := &frames.BlockedFrame{StreamID: 0x1337}
blockedFrameLen, _ := blockedFrame.MinLength(0)
f1 := &frames.StreamFrame{
StreamID: 5,
Offset: 1,
}
streamFrameHeaderLen, _ := f1.MinLength(0)
// this is the maximum dataLen of a StreamFrames that fits into one packet
dataLen := int(protocol.MaxFrameAndPublicHeaderSize - publicHeaderLen - streamFrameHeaderLen - blockedFrameLen)
streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}}
f1.Data = bytes.Repeat([]byte{'f'}, dataLen)
streamFramer.AddFrameForRetransmission(f1)
p, err := packer.PackPacket(nil, []frames.Frame{}, 0)
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
Expect(p.raw).To(HaveLen(int(protocol.MaxPacketSize)))
p, err = packer.PackPacket(nil, []frames.Frame{}, 0)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(BeNil())
Expect(p[0].(*frames.StreamFrame).DataLenPresent).To(BeFalse())
})
It("packs a connection-level BlockedFrame", func() {
@ -459,10 +419,9 @@ var _ = Describe("Packet packer", func() {
Data: []byte("foobar"),
}
streamFramer.AddFrameForRetransmission(f)
p, err := packer.composeNextPacket(nil, publicHeaderLen)
_, err := packer.composeNextPacket(nil, publicHeaderLen)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(HaveLen(2))
Expect(p[0]).To(Equal(&frames.BlockedFrame{StreamID: 0}))
Expect(packer.controlFrames[0]).To(Equal(&frames.BlockedFrame{StreamID: 0}))
})
})
})

64
stream_framer.go
View file

@ -89,11 +89,9 @@ func (f *streamFramer) EstimatedDataLen() protocol.ByteCount {
return l
}
func (f *streamFramer) PopStreamFrame(maxLen protocol.ByteCount) (*frames.StreamFrame, error) {
if frame := f.maybePopFrameForRetransmission(maxLen); frame != nil {
return frame, nil
}
return f.maybePopNormalFrame(maxLen)
func (f *streamFramer) PopStreamFrames(maxLen protocol.ByteCount) []*frames.StreamFrame {
fs, currentLen := f.maybePopFramesForRetransmission(maxLen)
return append(fs, f.maybePopNormalFrames(maxLen-currentLen)...)
}
func (f *streamFramer) PopBlockedFrame() *frames.BlockedFrame {
@ -105,32 +103,39 @@ func (f *streamFramer) PopBlockedFrame() *frames.BlockedFrame {
return frame
}
func (f *streamFramer) maybePopFrameForRetransmission(maxLen protocol.ByteCount) *frames.StreamFrame {
if len(f.retransmissionQueue) == 0 {
return nil
}
func (f *streamFramer) maybePopFramesForRetransmission(maxLen protocol.ByteCount) (res []*frames.StreamFrame, currentLen protocol.ByteCount) {
for len(f.retransmissionQueue) > 0 {
frame := f.retransmissionQueue[0]
frame.DataLenPresent = true
frameHeaderLen, _ := frame.MinLength(protocol.VersionWhatever) // can never error
if maxLen < frameHeaderLen {
return nil
if currentLen+frameHeaderLen > maxLen {
break
}
splitFrame := maybeSplitOffFrame(frame, maxLen-frameHeaderLen)
currentLen += frameHeaderLen
splitFrame := maybeSplitOffFrame(frame, maxLen-currentLen)
if splitFrame != nil { // StreamFrame was split
return splitFrame
res = append(res, splitFrame)
currentLen += splitFrame.DataLen()
break
}
f.retransmissionQueue = f.retransmissionQueue[1:]
return frame
res = append(res, frame)
currentLen += frame.DataLen()
}
return
}
func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames.StreamFrame, error) {
frame := &frames.StreamFrame{DataLenPresent: true}
func (f *streamFramer) maybePopNormalFrames(maxBytes protocol.ByteCount) (res []*frames.StreamFrame) {
f.streamsMutex.RLock()
defer f.streamsMutex.RUnlock()
frame := &frames.StreamFrame{DataLenPresent: true}
var currentLen protocol.ByteCount
for _, s := range *f.streams {
if s == nil {
continue
@ -140,16 +145,13 @@ func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames
// not perfect, but thread-safe since writeOffset is only written when getting data
frame.Offset = s.writeOffset
frameHeaderBytes, _ := frame.MinLength(protocol.VersionWhatever) // can never error
if maxBytes < frameHeaderBytes {
continue
if currentLen+frameHeaderBytes > maxBytes {
return // theoretically, we could find another stream that fits, but this is quite unlikely, so we stop here
}
maxLen := maxBytes - frameHeaderBytes
maxLen := maxBytes - currentLen - frameHeaderBytes
if s.lenOfDataForWriting() != 0 {
fcAllowance, err := f.getFCAllowanceForStream(s)
if err != nil {
return nil, err
}
fcAllowance, _ := f.getFCAllowanceForStream(s) // can never error
maxLen = utils.MinByteCount(maxLen, fcAllowance)
}
@ -162,15 +164,15 @@ func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames
if s.shouldSendFin() {
frame.FinBit = true
s.sentFin()
return frame, nil
res = append(res, frame)
currentLen += frameHeaderBytes + frame.DataLen()
frame = &frames.StreamFrame{DataLenPresent: true}
}
continue
}
frame.Data = data
if err := f.flowControlManager.AddBytesSent(s.streamID, protocol.ByteCount(len(data))); err != nil {
return nil, err
}
f.flowControlManager.AddBytesSent(s.streamID, protocol.ByteCount(len(data)))
// Finally, check if we are now FC blocked and should queue a BLOCKED frame
individualFcOffset, _ := f.flowControlManager.SendWindowSize(s.streamID) // can never error
@ -183,9 +185,11 @@ func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames
f.blockedFrameQueue = append(f.blockedFrameQueue, &frames.BlockedFrame{StreamID: 0})
}
return frame, nil
res = append(res, frame)
currentLen += frameHeaderBytes + frame.DataLen()
frame = &frames.StreamFrame{DataLenPresent: true}
}
return nil, nil
return
}
func (f *streamFramer) getFCAllowanceForStream(s *stream) (protocol.ByteCount, error) {

216
stream_framer_test.go
View file

@ -47,16 +47,16 @@ var _ = Describe("Stream Framer", func() {
It("sets the DataLenPresent for dequeued retransmitted frames", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
f, err := framer.PopStreamFrame(protocol.MaxByteCount)
Expect(err).NotTo(HaveOccurred())
Expect(f.DataLenPresent).To(BeTrue())
fs := framer.PopStreamFrames(protocol.MaxByteCount)
Expect(fs).To(HaveLen(1))
Expect(fs[0].DataLenPresent).To(BeTrue())
})
It("sets the DataLenPresent for dequeued normal frames", func() {
stream1.dataForWriting = []byte("foobar")
f, err := framer.PopStreamFrame(protocol.MaxByteCount)
Expect(err).NotTo(HaveOccurred())
Expect(f.DataLenPresent).To(BeTrue())
fs := framer.PopStreamFrames(protocol.MaxByteCount)
Expect(fs).To(HaveLen(1))
Expect(fs[0].DataLenPresent).To(BeTrue())
})
Context("HasData", func() {
@ -103,9 +103,9 @@ var _ = Describe("Stream Framer", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
stream1.dataForWriting = []byte("foobar")
Expect(framer.EstimatedDataLen()).To(Equal(protocol.ByteCount(2 + 6)))
framer.PopStreamFrame(1000)
framer.PopStreamFrames(8)
Expect(framer.EstimatedDataLen()).To(Equal(protocol.ByteCount(6)))
framer.PopStreamFrame(1000)
framer.PopStreamFrames(1000)
Expect(framer.EstimatedDataLen()).To(BeZero())
})
@ -130,66 +130,52 @@ var _ = Describe("Stream Framer", func() {
Context("Popping", func() {
It("returns nil when popping an empty framer", func() {
Expect(framer.PopStreamFrame(1000)).To(BeNil())
Expect(framer.PopStreamFrames(1000)).To(BeEmpty())
})
It("pops frames for retransmission", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
framer.AddFrameForRetransmission(retransmittedFrame2)
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(retransmittedFrame1))
frame, err = framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(retransmittedFrame2))
frame, err = framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(2))
Expect(fs[0]).To(Equal(retransmittedFrame1))
Expect(fs[1]).To(Equal(retransmittedFrame2))
Expect(framer.PopStreamFrames(1000)).To(BeEmpty())
})
It("returns normal frames", func() {
stream1.dataForWriting = []byte("foobar")
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(stream1.streamID))
Expect(frame.Data).To(Equal([]byte("foobar")))
frame, err = framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].StreamID).To(Equal(stream1.streamID))
Expect(fs[0].Data).To(Equal([]byte("foobar")))
Expect(framer.PopStreamFrames(1000)).To(BeEmpty())
})
It("returns multiple normal frames", func() {
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
frame1, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
frame2, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(2))
// Swap if we dequeued in other order
if frame1.StreamID != stream1.streamID {
frame1, frame2 = frame2, frame1
if fs[0].StreamID != stream1.streamID {
fs[0], fs[1] = fs[1], fs[0]
}
Expect(frame1.StreamID).To(Equal(stream1.streamID))
Expect(frame1.Data).To(Equal([]byte("foobar")))
Expect(frame2.StreamID).To(Equal(stream2.streamID))
Expect(frame2.Data).To(Equal([]byte("foobaz")))
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
Expect(fs[0].StreamID).To(Equal(stream1.streamID))
Expect(fs[0].Data).To(Equal([]byte("foobar")))
Expect(fs[1].StreamID).To(Equal(stream2.streamID))
Expect(fs[1].Data).To(Equal([]byte("foobaz")))
Expect(framer.PopStreamFrames(1000)).To(BeEmpty())
})
It("returns retransmission frames before normal frames", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
stream1.dataForWriting = []byte("foobar")
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(Equal(retransmittedFrame1))
frame, err = framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(stream1.streamID))
frame, err = framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(2))
Expect(fs[0]).To(Equal(retransmittedFrame1))
Expect(fs[1].StreamID).To(Equal(stream1.streamID))
Expect(framer.PopStreamFrames(1000)).To(BeEmpty())
})
Context("splitting of frames", func() {
@ -229,48 +215,46 @@ var _ = Describe("Stream Framer", func() {
It("splits a frame", func() {
framer.AddFrameForRetransmission(retransmittedFrame2)
origlen := retransmittedFrame2.DataLen()
frame, err := framer.PopStreamFrame(6)
Expect(err).ToNot(HaveOccurred())
minLength, _ := frame.MinLength(0)
Expect(minLength + frame.DataLen()).To(Equal(protocol.ByteCount(6)))
Expect(framer.retransmissionQueue[0].Data).To(HaveLen(int(origlen - frame.DataLen())))
Expect(framer.retransmissionQueue[0].Offset).To(Equal(frame.DataLen()))
fs := framer.PopStreamFrames(6)
Expect(fs).To(HaveLen(1))
minLength, _ := fs[0].MinLength(0)
Expect(minLength + fs[0].DataLen()).To(Equal(protocol.ByteCount(6)))
Expect(framer.retransmissionQueue[0].Data).To(HaveLen(int(origlen - fs[0].DataLen())))
Expect(framer.retransmissionQueue[0].Offset).To(Equal(fs[0].DataLen()))
})
It("only removes a frame from the framer after returning all split parts", func() {
framer.AddFrameForRetransmission(retransmittedFrame2)
frame, err := framer.PopStreamFrame(6)
Expect(err).ToNot(HaveOccurred())
Expect(frame).ToNot(BeNil())
fs := framer.PopStreamFrames(6)
Expect(fs).To(HaveLen(1))
Expect(framer.HasData()).To(BeTrue())
frame, err = framer.PopStreamFrame(100)
Expect(err).ToNot(HaveOccurred())
Expect(frame).ToNot(BeNil())
fs = framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(framer.HasData()).To(BeFalse())
})
It("gets the whole data of a frame if it was split", func() {
origdata := []byte("foobar")
stream1.dataForWriting = origdata
frame, err := framer.PopStreamFrame(7)
Expect(err).ToNot(HaveOccurred())
Expect(frame.Data).To(Equal([]byte("foo")))
fs := framer.PopStreamFrames(7)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).To(Equal([]byte("foo")))
var b bytes.Buffer
frame.Write(&b, 0)
fs[0].Write(&b, 0)
Expect(b.Len()).To(Equal(7))
frame, err = framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.Data).To(Equal([]byte("bar")))
fs = framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).To(Equal([]byte("bar")))
})
It("correctly calculates the byte length when returning a split frame", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
framer.AddFrameForRetransmission(retransmittedFrame2)
startByteLength := framer.EstimatedDataLen()
frame, err := framer.PopStreamFrame(6)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(retransmittedFrame1.StreamID)) // make sure the right frame was popped
Expect(framer.EstimatedDataLen()).To(Equal(startByteLength - frame.DataLen()))
fs := framer.PopStreamFrames(6)
Expect(fs).To(HaveLen(1))
Expect(fs[0].StreamID).To(Equal(retransmittedFrame1.StreamID)) // make sure the right frame was popped
Expect(framer.EstimatedDataLen()).To(Equal(startByteLength - fs[0].DataLen()))
})
})
@ -278,12 +262,12 @@ var _ = Describe("Stream Framer", func() {
It("sends FINs when streams are closed", func() {
stream1.writeOffset = 42
stream1.closed = 1
frame, err := framer.PopStreamFrame(1000)
Expect(err).NotTo(HaveOccurred())
Expect(frame.StreamID).To(Equal(stream1.streamID))
Expect(frame.Offset).To(Equal(stream1.writeOffset))
Expect(frame.FinBit).To(BeTrue())
Expect(frame.Data).To(BeEmpty())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].StreamID).To(Equal(stream1.streamID))
Expect(fs[0].Offset).To(Equal(stream1.writeOffset))
Expect(fs[0].FinBit).To(BeTrue())
Expect(fs[0].Data).To(BeEmpty())
})
})
})
@ -291,15 +275,13 @@ var _ = Describe("Stream Framer", func() {
Context("flow control", func() {
It("tells the FlowControlManager how many bytes it sent", func() {
stream1.dataForWriting = []byte("foobar")
_, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
framer.PopStreamFrames(1000)
Expect(fcm.bytesSent).To(Equal(protocol.ByteCount(6)))
})
It("does not count retransmitted frames as sent bytes", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
_, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
framer.PopStreamFrames(1000)
Expect(fcm.bytesSent).To(BeZero())
})
@ -307,62 +289,61 @@ var _ = Describe("Stream Framer", func() {
stream1.writeOffset = 10
stream1.dataForWriting = []byte("foobar")
fcm.sendWindowSizes[stream1.streamID] = 10 + 6
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.DataLen()).To(Equal(protocol.ByteCount(6)))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].DataLen()).To(Equal(protocol.ByteCount(6)))
})
It("returns a smaller frame if the whole frame doesn't fit", func() {
stream1.dataForWriting = []byte("foobar")
fcm.sendWindowSizes[stream1.streamID] = 3
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.Data).To(Equal([]byte("foo")))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).To(Equal([]byte("foo")))
})
It("returns a smaller frame if the whole frame doesn't fit in the stream flow control window, for non-zero StreamFrame offset", func() {
stream1.writeOffset = 1
stream1.dataForWriting = []byte("foobar")
fcm.sendWindowSizes[stream1.StreamID()] = 4
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.Data).To(Equal([]byte("foo")))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).To(Equal([]byte("foo")))
})
It("returns a smaller frame if the whole frame doesn't fit in the connection flow control window", func() {
stream1.dataForWriting = []byte("foobar")
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()}
fcm.remainingConnectionWindowSize = 3
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.Data).To(Equal([]byte("foo")))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).To(Equal([]byte("foo")))
})
It("ignores the connection flow control window for non-contributing streams", func() {
stream1.dataForWriting = []byte("foobar")
fcm.remainingConnectionWindowSize = 0
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.Data).To(Equal([]byte("foobar")))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).To(Equal([]byte("foobar")))
})
It("respects the connection flow control window for contributing streams", func() {
stream1.dataForWriting = []byte("foobar")
fcm.remainingConnectionWindowSize = 0
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()}
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(BeEmpty())
})
It("selects a stream that is not flow control blocked", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 0
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(stream2.StreamID()))
Expect(frame.Data).To(Equal([]byte("foobaz")))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].StreamID).To(Equal(stream2.StreamID()))
Expect(fs[0].Data).To(Equal([]byte("foobaz")))
})
It("chooses a non-contributing stream if the connection is flow control blocked", func() {
@ -370,9 +351,9 @@ var _ = Describe("Stream Framer", func() {
stream2.dataForWriting = []byte("foobaz")
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()}
fcm.remainingConnectionWindowSize = 0
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame.StreamID).To(Equal(stream2.StreamID()))
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].StreamID).To(Equal(stream2.StreamID()))
})
It("returns nil if every stream is individually flow control blocked", func() {
@ -380,9 +361,8 @@ var _ = Describe("Stream Framer", func() {
fcm.sendWindowSizes[stream2.StreamID()] = 0
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(BeEmpty())
})
It("returns nil if every stream is connection flow control blocked", func() {
@ -390,9 +370,8 @@ var _ = Describe("Stream Framer", func() {
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID(), stream2.StreamID()}
frame, err := framer.PopStreamFrame(1000)
Expect(err).ToNot(HaveOccurred())
Expect(frame).To(BeNil())
fs := framer.PopStreamFrames(1000)
Expect(fs).To(BeEmpty())
})
})
@ -408,8 +387,7 @@ var _ = Describe("Stream Framer", func() {
It("queues and pops BLOCKED frames for individually blocked streams", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 3
stream1.dataForWriting = []byte("foo")
_, err := framer.PopStreamFrame(1000)
Expect(err).NotTo(HaveOccurred())
framer.PopStreamFrames(1000)
blockedFrame := framer.PopBlockedFrame()
Expect(blockedFrame).ToNot(BeNil())
Expect(blockedFrame.StreamID).To(Equal(stream1.StreamID()))
@ -420,8 +398,7 @@ var _ = Describe("Stream Framer", func() {
fcm.remainingConnectionWindowSize = 3
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()}
stream1.dataForWriting = []byte("foo")
_, err := framer.PopStreamFrame(1000)
Expect(err).NotTo(HaveOccurred())
framer.PopStreamFrames(1000)
blockedFrame := framer.PopBlockedFrame()
Expect(blockedFrame).ToNot(BeNil())
Expect(blockedFrame.StreamID).To(BeZero())
@ -431,19 +408,14 @@ var _ = Describe("Stream Framer", func() {
It("does not queue BLOCKED frames for non-contributing streams", func() {
fcm.remainingConnectionWindowSize = 3
stream1.dataForWriting = []byte("foo")
_, err := framer.PopStreamFrame(1000)
Expect(err).NotTo(HaveOccurred())
framer.PopStreamFrames(1000)
Expect(framer.PopBlockedFrame()).To(BeNil())
})
It("does not queue BLOCKED frames twice", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 3
stream1.dataForWriting = []byte("foobar")
_, err := framer.PopStreamFrame(1000)
Expect(err).NotTo(HaveOccurred())
frame, err := framer.PopStreamFrame(1000)
Expect(err).NotTo(HaveOccurred())
Expect(frame).To(BeNil())
framer.PopStreamFrames(1000)
blockedFrame := framer.PopBlockedFrame()
Expect(blockedFrame).ToNot(BeNil())
Expect(blockedFrame.StreamID).To(Equal(stream1.StreamID()))