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

Mock FlowControlManager using gomock

Browse source
This commit is contained in:
Lucas Clemente 2017-06-08 15:55:08 +02:00
parent 94843e1919
commit eeb2d8d821
9 changed files with 325 additions and 228 deletions
Download patch file Download diff file Expand all files Collapse all files

140
internal/mocks/flow_control_manager.go Normal file
View file

@ -0,0 +1,140 @@
// Automatically generated by MockGen. DO NOT EDIT!
// Source: github.com/lucas-clemente/quic-go/flowcontrol (interfaces: FlowControlManager)
package mocks
import (
gomock "github.com/golang/mock/gomock"
flowcontrol "github.com/lucas-clemente/quic-go/flowcontrol"
protocol "github.com/lucas-clemente/quic-go/protocol"
)
// Mock of FlowControlManager interface
type MockFlowControlManager struct {
ctrl *gomock.Controller
recorder *_MockFlowControlManagerRecorder
}
// Recorder for MockFlowControlManager (not exported)
type _MockFlowControlManagerRecorder struct {
mock *MockFlowControlManager
}
func NewMockFlowControlManager(ctrl *gomock.Controller) *MockFlowControlManager {
mock := &MockFlowControlManager{ctrl: ctrl}
mock.recorder = &_MockFlowControlManagerRecorder{mock}
return mock
}
func (_m *MockFlowControlManager) EXPECT() *_MockFlowControlManagerRecorder {
return _m.recorder
}
func (_m *MockFlowControlManager) AddBytesRead(_param0 protocol.StreamID, _param1 protocol.ByteCount) error {
ret := _m.ctrl.Call(_m, "AddBytesRead", _param0, _param1)
ret0, _ := ret[0].(error)
return ret0
}
func (_mr *_MockFlowControlManagerRecorder) AddBytesRead(arg0, arg1 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "AddBytesRead", arg0, arg1)
}
func (_m *MockFlowControlManager) AddBytesSent(_param0 protocol.StreamID, _param1 protocol.ByteCount) error {
ret := _m.ctrl.Call(_m, "AddBytesSent", _param0, _param1)
ret0, _ := ret[0].(error)
return ret0
}
func (_mr *_MockFlowControlManagerRecorder) AddBytesSent(arg0, arg1 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "AddBytesSent", arg0, arg1)
}
func (_m *MockFlowControlManager) GetReceiveWindow(_param0 protocol.StreamID) (protocol.ByteCount, error) {
ret := _m.ctrl.Call(_m, "GetReceiveWindow", _param0)
ret0, _ := ret[0].(protocol.ByteCount)
ret1, _ := ret[1].(error)
return ret0, ret1
}
func (_mr *_MockFlowControlManagerRecorder) GetReceiveWindow(arg0 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "GetReceiveWindow", arg0)
}
func (_m *MockFlowControlManager) GetWindowUpdates() []flowcontrol.WindowUpdate {
ret := _m.ctrl.Call(_m, "GetWindowUpdates")
ret0, _ := ret[0].([]flowcontrol.WindowUpdate)
return ret0
}
func (_mr *_MockFlowControlManagerRecorder) GetWindowUpdates() *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "GetWindowUpdates")
}
func (_m *MockFlowControlManager) NewStream(_param0 protocol.StreamID, _param1 bool) {
_m.ctrl.Call(_m, "NewStream", _param0, _param1)
}
func (_mr *_MockFlowControlManagerRecorder) NewStream(arg0, arg1 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "NewStream", arg0, arg1)
}
func (_m *MockFlowControlManager) RemainingConnectionWindowSize() protocol.ByteCount {
ret := _m.ctrl.Call(_m, "RemainingConnectionWindowSize")
ret0, _ := ret[0].(protocol.ByteCount)
return ret0
}
func (_mr *_MockFlowControlManagerRecorder) RemainingConnectionWindowSize() *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "RemainingConnectionWindowSize")
}
func (_m *MockFlowControlManager) RemoveStream(_param0 protocol.StreamID) {
_m.ctrl.Call(_m, "RemoveStream", _param0)
}
func (_mr *_MockFlowControlManagerRecorder) RemoveStream(arg0 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "RemoveStream", arg0)
}
func (_m *MockFlowControlManager) ResetStream(_param0 protocol.StreamID, _param1 protocol.ByteCount) error {
ret := _m.ctrl.Call(_m, "ResetStream", _param0, _param1)
ret0, _ := ret[0].(error)
return ret0
}
func (_mr *_MockFlowControlManagerRecorder) ResetStream(arg0, arg1 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "ResetStream", arg0, arg1)
}
func (_m *MockFlowControlManager) SendWindowSize(_param0 protocol.StreamID) (protocol.ByteCount, error) {
ret := _m.ctrl.Call(_m, "SendWindowSize", _param0)
ret0, _ := ret[0].(protocol.ByteCount)
ret1, _ := ret[1].(error)
return ret0, ret1
}
func (_mr *_MockFlowControlManagerRecorder) SendWindowSize(arg0 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "SendWindowSize", arg0)
}
func (_m *MockFlowControlManager) UpdateHighestReceived(_param0 protocol.StreamID, _param1 protocol.ByteCount) error {
ret := _m.ctrl.Call(_m, "UpdateHighestReceived", _param0, _param1)
ret0, _ := ret[0].(error)
return ret0
}
func (_mr *_MockFlowControlManagerRecorder) UpdateHighestReceived(arg0, arg1 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "UpdateHighestReceived", arg0, arg1)
}
func (_m *MockFlowControlManager) UpdateWindow(_param0 protocol.StreamID, _param1 protocol.ByteCount) (bool, error) {
ret := _m.ctrl.Call(_m, "UpdateWindow", _param0, _param1)
ret0, _ := ret[0].(bool)
ret1, _ := ret[1].(error)
return ret0, ret1
}
func (_mr *_MockFlowControlManagerRecorder) UpdateWindow(arg0, arg1 interface{}) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "UpdateWindow", arg0, arg1)
}

3
internal/mocks/gen.go Normal file
View file

@ -0,0 +1,3 @@
package mocks
//go:generate mockgen -destination flow_control_manager.go -package mocks github.com/lucas-clemente/quic-go/flowcontrol FlowControlManager

7
packet_packer_test.go
View file

@ -48,13 +48,8 @@ var _ = Describe("Packet packer", func() {
)
BeforeEach(func() {
fcm := newMockFlowControlHandler()
fcm.sendWindowSizes[3] = protocol.MaxByteCount
fcm.sendWindowSizes[5] = protocol.MaxByteCount
fcm.sendWindowSizes[7] = protocol.MaxByteCount
cpm := &mockConnectionParametersManager{}
streamFramer = newStreamFramer(newStreamsMap(nil, protocol.PerspectiveServer, cpm), fcm)
streamFramer = newStreamFramer(newStreamsMap(nil, protocol.PerspectiveServer, cpm), nil)
packer = &packetPacker{
cryptoSetup: &mockCryptoSetup{encLevelSeal: protocol.EncryptionForwardSecure},

2
protocol/protocol.go
View file

@ -31,7 +31,7 @@ type StreamID uint32
type ByteCount uint64
// MaxByteCount is the maximum value of a ByteCount
const MaxByteCount = math.MaxUint64
const MaxByteCount = ByteCount(math.MaxUint64)
// MaxReceivePacketSize maximum packet size of any QUIC packet, based on
// ethernet's max size, minus the IP and UDP headers. IPv6 has a 40 byte header,

11
quic_suite_test.go
View file

@ -1,6 +1,7 @@
package quic
import (
"github.com/golang/mock/gomock"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
@ -11,3 +12,13 @@ func TestQuicGo(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "QUIC Suite")
}
var mockCtrl *gomock.Controller
var _ = BeforeEach(func() {
mockCtrl = gomock.NewController(GinkgoT())
})
var _ = AfterEach(func() {
mockCtrl.Finish()
})

1
session.go
View file

@ -31,7 +31,6 @@ type receivedPacket struct {
var (
errRstStreamOnInvalidStream = errors.New("RST_STREAM received for unknown stream")
errWindowUpdateOnClosedStream = errors.New("WINDOW_UPDATE received for an already closed stream")
errSessionAlreadyClosed = errors.New("cannot close session; it was already closed before")
)
var (

29
session_test.go
View file

@ -17,6 +17,7 @@ import (
"github.com/lucas-clemente/quic-go/crypto"
"github.com/lucas-clemente/quic-go/frames"
"github.com/lucas-clemente/quic-go/handshake"
"github.com/lucas-clemente/quic-go/internal/mocks"
"github.com/lucas-clemente/quic-go/protocol"
"github.com/lucas-clemente/quic-go/qerr"
"github.com/lucas-clemente/quic-go/testdata"
@ -474,21 +475,22 @@ var _ = Describe("Session", func() {
It("passes the byte offset to the flow controller", func() {
sess.streamsMap.GetOrOpenStream(5)
sess.flowControlManager = newMockFlowControlHandler()
fcm := mocks.NewMockFlowControlManager(mockCtrl)
sess.flowControlManager = fcm
fcm.EXPECT().ResetStream(protocol.StreamID(5), protocol.ByteCount(0x1337))
err := sess.handleRstStreamFrame(&frames.RstStreamFrame{
StreamID: 5,
ByteOffset: 0x1337,
})
Expect(err).ToNot(HaveOccurred())
Expect(sess.flowControlManager.(*mockFlowControlHandler).highestReceivedForStream).To(Equal(protocol.StreamID(5)))
Expect(sess.flowControlManager.(*mockFlowControlHandler).highestReceived).To(Equal(protocol.ByteCount(0x1337)))
})
It("returns errors from the flow controller", func() {
sess.streamsMap.GetOrOpenStream(5)
sess.flowControlManager = newMockFlowControlHandler()
testErr := errors.New("flow control violation")
sess.flowControlManager.(*mockFlowControlHandler).flowControlViolation = testErr
sess.streamsMap.GetOrOpenStream(5)
fcm := mocks.NewMockFlowControlManager(mockCtrl)
sess.flowControlManager = fcm
fcm.EXPECT().ResetStream(protocol.StreamID(5), protocol.ByteCount(0x1337)).Return(testErr)
err := sess.handleRstStreamFrame(&frames.RstStreamFrame{
StreamID: 5,
ByteOffset: 0x1337,
@ -1062,9 +1064,11 @@ var _ = Describe("Session", func() {
It("retransmits a WindowUpdates if it hasn't already sent a WindowUpdate with a higher ByteOffset", func() {
_, err := sess.GetOrOpenStream(5)
Expect(err).ToNot(HaveOccurred())
fc := newMockFlowControlHandler()
fc.receiveWindow = 0x1000
sess.flowControlManager = fc
fcm := mocks.NewMockFlowControlManager(mockCtrl)
sess.flowControlManager = fcm
fcm.EXPECT().GetWindowUpdates()
fcm.EXPECT().GetReceiveWindow(protocol.StreamID(5)).Return(protocol.ByteCount(0x1000), nil)
fcm.EXPECT().GetWindowUpdates()
wuf := &frames.WindowUpdateFrame{
StreamID: 5,
ByteOffset: 0x1000,
@ -1082,9 +1086,10 @@ var _ = Describe("Session", func() {
It("doesn't retransmit WindowUpdates if it already sent a WindowUpdate with a higher ByteOffset", func() {
_, err := sess.GetOrOpenStream(5)
Expect(err).ToNot(HaveOccurred())
fc := newMockFlowControlHandler()
fc.receiveWindow = 0x2000
sess.flowControlManager = fc
fcm := mocks.NewMockFlowControlManager(mockCtrl)
sess.flowControlManager = fcm
fcm.EXPECT().GetWindowUpdates()
fcm.EXPECT().GetReceiveWindow(protocol.StreamID(5)).Return(protocol.ByteCount(0x2000), nil)
sph.retransmissionQueue = []*ackhandler.Packet{{
Frames: []frames.Frame{&frames.WindowUpdateFrame{
StreamID: 5,

157
stream_framer_test.go
View file

@ -4,18 +4,24 @@ import (
"bytes"
"github.com/lucas-clemente/quic-go/frames"
"github.com/lucas-clemente/quic-go/internal/mocks"
"github.com/lucas-clemente/quic-go/protocol"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
var _ = Describe("Stream Framer", func() {
const (
id1 = protocol.StreamID(10)
id2 = protocol.StreamID(11)
)
var (
retransmittedFrame1, retransmittedFrame2 *frames.StreamFrame
framer *streamFramer
streamsMap *streamsMap
stream1, stream2 *stream
fcm *mockFlowControlHandler
mockFcm *mocks.MockFlowControlManager
)
BeforeEach(func() {
@ -28,19 +34,15 @@ var _ = Describe("Stream Framer", func() {
Data: []byte{0xDE, 0xCA, 0xFB, 0xAD},
}
stream1 = &stream{streamID: 10}
stream2 = &stream{streamID: 11}
stream1 = &stream{streamID: id1}
stream2 = &stream{streamID: id2}
streamsMap = newStreamsMap(nil, protocol.PerspectiveServer, &mockConnectionParametersManager{})
streamsMap.putStream(stream1)
streamsMap.putStream(stream2)
fcm = newMockFlowControlHandler()
fcm.sendWindowSizes[stream1.streamID] = protocol.MaxByteCount
fcm.sendWindowSizes[stream2.streamID] = protocol.MaxByteCount
fcm.sendWindowSizes[retransmittedFrame1.StreamID] = protocol.MaxByteCount
fcm.sendWindowSizes[retransmittedFrame2.StreamID] = protocol.MaxByteCount
framer = newStreamFramer(streamsMap, fcm)
mockFcm = mocks.NewMockFlowControlManager(mockCtrl)
framer = newStreamFramer(streamsMap, mockFcm)
})
It("says if it has retransmissions", func() {
@ -57,6 +59,9 @@ var _ = Describe("Stream Framer", func() {
})
It("sets the DataLenPresent for dequeued normal frames", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
fs := framer.PopStreamFrames(protocol.MaxByteCount)
Expect(fs).To(HaveLen(1))
@ -79,6 +84,9 @@ var _ = Describe("Stream Framer", func() {
})
It("returns normal frames", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
@ -88,6 +96,12 @@ var _ = Describe("Stream Framer", func() {
})
It("returns multiple normal frames", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
mockFcm.EXPECT().SendWindowSize(id2).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id2, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
fs := framer.PopStreamFrames(1000)
@ -104,6 +118,9 @@ var _ = Describe("Stream Framer", func() {
})
It("returns retransmission frames before normal frames", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
framer.AddFrameForRetransmission(retransmittedFrame1)
stream1.dataForWriting = []byte("foobar")
fs := framer.PopStreamFrames(1000)
@ -114,9 +131,13 @@ var _ = Describe("Stream Framer", func() {
})
It("does not pop empty frames", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
stream1.dataForWriting = []byte("foobar")
fs := framer.PopStreamFrames(4)
Expect(fs).To(HaveLen(0))
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(1))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
fs = framer.PopStreamFrames(5)
Expect(fs).To(HaveLen(1))
Expect(fs[0].Data).ToNot(BeEmpty())
@ -124,6 +145,12 @@ var _ = Describe("Stream Framer", func() {
})
It("uses the round-robin scheduling", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
mockFcm.EXPECT().SendWindowSize(id2).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id2, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = bytes.Repeat([]byte("f"), 100)
stream2.dataForWriting = bytes.Repeat([]byte("e"), 100)
fs := framer.PopStreamFrames(10)
@ -222,6 +249,12 @@ var _ = Describe("Stream Framer", func() {
})
It("gets the whole data of a frame if it was split", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
origdata := []byte("foobar")
stream1.dataForWriting = origdata
fs := framer.PopStreamFrames(7)
@ -238,6 +271,8 @@ var _ = Describe("Stream Framer", func() {
Context("sending FINs", func() {
It("sends FINs when streams are closed", func() {
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(0))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.writeOffset = 42
stream1.finishedWriting.Set(true)
fs := framer.PopStreamFrames(1000)
@ -249,9 +284,10 @@ var _ = Describe("Stream Framer", func() {
})
It("sends FINs when flow-control blocked", func() {
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(0))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.writeOffset = 42
stream1.finishedWriting.Set(true)
fcm.sendWindowSizes[stream1.StreamID()] = 42
fs := framer.PopStreamFrames(1000)
Expect(fs).To(HaveLen(1))
Expect(fs[0].StreamID).To(Equal(stream1.streamID))
@ -261,6 +297,9 @@ var _ = Describe("Stream Framer", func() {
})
It("bundles FINs with data", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
stream1.finishedWriting.Set(true)
fs := framer.PopStreamFrames(1000)
@ -274,70 +313,55 @@ var _ = Describe("Stream Framer", func() {
Context("flow control", func() {
It("tells the FlowControlManager how many bytes it sent", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
framer.PopStreamFrames(1000)
Expect(fcm.bytesSent).To(Equal(protocol.ByteCount(6)))
})
It("does not count retransmitted frames as sent bytes", func() {
framer.AddFrameForRetransmission(retransmittedFrame1)
framer.PopStreamFrames(1000)
Expect(fcm.bytesSent).To(BeZero())
})
It("returns the whole frame if it fits", func() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(10+6), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.writeOffset = 10
stream1.dataForWriting = []byte("foobar")
fcm.sendWindowSizes[stream1.streamID] = 10 + 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() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(3), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
fcm.sendWindowSizes[stream1.streamID] = 3
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() {
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(3), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.writeOffset = 1
stream1.dataForWriting = []byte("foobar")
fcm.sendWindowSizes[stream1.StreamID()] = 3
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
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
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()}
fs := framer.PopStreamFrames(1000)
Expect(fs).To(BeEmpty())
})
It("selects a stream that is not flow control blocked", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 0
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(0), nil)
mockFcm.EXPECT().SendWindowSize(id2).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id2, protocol.ByteCount(6))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
fs := framer.PopStreamFrames(1000)
@ -346,46 +370,25 @@ var _ = Describe("Stream Framer", func() {
Expect(fs[0].Data).To(Equal([]byte("foobaz")))
})
It("chooses a non-contributing stream if the connection is flow control blocked", func() {
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()}
fcm.remainingConnectionWindowSize = 0
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() {
fcm.sendWindowSizes[stream1.StreamID()] = 0
fcm.sendWindowSizes[stream2.StreamID()] = 0
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(0), nil)
mockFcm.EXPECT().SendWindowSize(id2).Return(protocol.ByteCount(0), nil)
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
fs := framer.PopStreamFrames(1000)
Expect(fs).To(BeEmpty())
})
It("returns nil if every stream is connection flow control blocked", func() {
fcm.remainingConnectionWindowSize = 0
stream1.dataForWriting = []byte("foobar")
stream2.dataForWriting = []byte("foobaz")
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID(), stream2.StreamID()}
fs := framer.PopStreamFrames(1000)
Expect(fs).To(BeEmpty())
})
})
Context("BLOCKED frames", func() {
BeforeEach(func() {
fcm.remainingConnectionWindowSize = protocol.MaxByteCount
})
It("Pop returns nil if no frame is queued", func() {
Expect(framer.PopBlockedFrame()).To(BeNil())
})
It("queues and pops BLOCKED frames for individually blocked streams", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 3
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(3), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foo")
frames := framer.PopStreamFrames(1000)
Expect(frames).To(HaveLen(1))
@ -396,7 +399,11 @@ var _ = Describe("Stream Framer", func() {
})
It("does not queue a stream-level BLOCKED frame after sending the FinBit frame", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 5000
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(5000), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(0))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foo")
frames := framer.PopStreamFrames(1000)
Expect(frames).To(HaveLen(1))
@ -411,8 +418,10 @@ var _ = Describe("Stream Framer", func() {
})
It("queues and pops BLOCKED frames for connection blocked streams", func() {
fcm.remainingConnectionWindowSize = 3
fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()}
// FCM already considers the connection window size
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(3), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.ByteCount(0))
stream1.dataForWriting = []byte("foo")
framer.PopStreamFrames(1000)
blockedFrame := framer.PopBlockedFrame()
@ -422,14 +431,18 @@ var _ = Describe("Stream Framer", func() {
})
It("does not queue BLOCKED frames for non-contributing streams", func() {
fcm.remainingConnectionWindowSize = 3
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.MaxByteCount, nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foo")
framer.PopStreamFrames(1000)
Expect(framer.PopBlockedFrame()).To(BeNil())
})
It("does not queue BLOCKED frames twice", func() {
fcm.sendWindowSizes[stream1.StreamID()] = 3
mockFcm.EXPECT().SendWindowSize(id1).Return(protocol.ByteCount(3), nil)
mockFcm.EXPECT().AddBytesSent(id1, protocol.ByteCount(3))
mockFcm.EXPECT().RemainingConnectionWindowSize().Return(protocol.MaxByteCount)
stream1.dataForWriting = []byte("foobar")
framer.PopStreamFrames(1000)
blockedFrame := framer.PopBlockedFrame()

203
stream_test.go
View file

@ -5,114 +5,17 @@ import (
"io"
"time"
"github.com/lucas-clemente/quic-go/congestion"
"github.com/lucas-clemente/quic-go/flowcontrol"
"github.com/lucas-clemente/quic-go/frames"
"github.com/lucas-clemente/quic-go/internal/mocks"
"github.com/lucas-clemente/quic-go/protocol"
"github.com/lucas-clemente/quic-go/utils"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
type mockFlowControlHandler struct {
streamsContributing []protocol.StreamID
sendWindowSizes map[protocol.StreamID]protocol.ByteCount
remainingConnectionWindowSize protocol.ByteCount
bytesReadForStream protocol.StreamID
bytesRead protocol.ByteCount
bytesSent protocol.ByteCount
receiveWindow protocol.ByteCount
highestReceivedForStream protocol.StreamID
highestReceived protocol.ByteCount
flowControlViolation error
triggerStreamWindowUpdate bool
triggerConnectionWindowUpdate bool
}
var _ flowcontrol.FlowControlManager = &mockFlowControlHandler{}
func newMockFlowControlHandler() *mockFlowControlHandler {
return &mockFlowControlHandler{
sendWindowSizes: make(map[protocol.StreamID]protocol.ByteCount),
}
}
func (m *mockFlowControlHandler) NewStream(streamID protocol.StreamID, contributesToConnectionFlow bool) {
panic("not implemented")
}
func (m *mockFlowControlHandler) RemoveStream(streamID protocol.StreamID) {
delete(m.sendWindowSizes, streamID)
}
func (m *mockFlowControlHandler) GetWindowUpdates() (res []flowcontrol.WindowUpdate) {
if m.triggerStreamWindowUpdate {
res = append(res, flowcontrol.WindowUpdate{StreamID: 42, Offset: 0x1337})
}
if m.triggerConnectionWindowUpdate {
res = append(res, flowcontrol.WindowUpdate{StreamID: 0, Offset: 0x1337})
}
return res
}
func (m *mockFlowControlHandler) GetReceiveWindow(protocol.StreamID) (protocol.ByteCount, error) {
return m.receiveWindow, nil
}
func (m *mockFlowControlHandler) AddBytesRead(streamID protocol.StreamID, n protocol.ByteCount) error {
m.bytesReadForStream = streamID
m.bytesRead = n
return nil
}
func (m *mockFlowControlHandler) ResetStream(streamID protocol.StreamID, byteOffset protocol.ByteCount) error {
m.bytesRead = byteOffset
return m.UpdateHighestReceived(streamID, byteOffset)
}
func (m *mockFlowControlHandler) UpdateHighestReceived(streamID protocol.StreamID, byteOffset protocol.ByteCount) error {
if m.flowControlViolation != nil {
return m.flowControlViolation
}
m.highestReceivedForStream = streamID
m.highestReceived = byteOffset
return nil
}
func (m *mockFlowControlHandler) AddBytesSent(streamID protocol.StreamID, n protocol.ByteCount) error {
m.bytesSent += n
m.sendWindowSizes[streamID] -= n
for _, s := range m.streamsContributing {
if s == streamID {
m.remainingConnectionWindowSize -= n
return nil
}
}
return nil
}
func (m *mockFlowControlHandler) SendWindowSize(streamID protocol.StreamID) (protocol.ByteCount, error) {
res := m.sendWindowSizes[streamID]
for _, s := range m.streamsContributing {
if s == streamID {
return utils.MinByteCount(res, m.remainingConnectionWindowSize), nil
}
}
return res, nil
}
func (m *mockFlowControlHandler) RemainingConnectionWindowSize() protocol.ByteCount {
return m.remainingConnectionWindowSize
}
func (m *mockFlowControlHandler) UpdateWindow(streamID protocol.StreamID, offset protocol.ByteCount) (bool, error) {
panic("not implemented")
}
var _ = Describe("Stream", func() {
const streamID protocol.StreamID = 1337
var (
str *stream
onDataCalled bool
@ -120,6 +23,8 @@ var _ = Describe("Stream", func() {
resetCalled bool
resetCalledForStream protocol.StreamID
resetCalledAtOffset protocol.ByteCount
mockFcm *mocks.MockFlowControlManager
)
onData := func() {
@ -135,11 +40,8 @@ var _ = Describe("Stream", func() {
BeforeEach(func() {
onDataCalled = false
resetCalled = false
var streamID protocol.StreamID = 1337
cpm := &mockConnectionParametersManager{}
flowControlManager := flowcontrol.NewFlowControlManager(cpm, &congestion.RTTStats{})
flowControlManager.NewStream(streamID, true)
str = newStream(streamID, onData, onReset, flowControlManager)
mockFcm = mocks.NewMockFlowControlManager(mockCtrl)
str = newStream(streamID, onData, onReset, mockFcm)
})
It("gets stream id", func() {
@ -148,6 +50,8 @@ var _ = Describe("Stream", func() {
Context("reading", func() {
It("reads a single StreamFrame", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -162,6 +66,9 @@ var _ = Describe("Stream", func() {
})
It("reads a single StreamFrame in multiple goes", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -180,6 +87,9 @@ var _ = Describe("Stream", func() {
})
It("reads all data available", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2)).Times(2)
frame1 := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD},
@ -200,6 +110,9 @@ var _ = Describe("Stream", func() {
})
It("assembles multiple StreamFrames", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2)).Times(2)
frame1 := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD},
@ -220,6 +133,8 @@ var _ = Describe("Stream", func() {
})
It("waits until data is available", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2))
go func() {
frame := frames.StreamFrame{
Offset: 0,
@ -236,6 +151,9 @@ var _ = Describe("Stream", func() {
})
It("handles StreamFrames in wrong order", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2)).Times(2)
frame1 := frames.StreamFrame{
Offset: 2,
Data: []byte{0xBE, 0xEF},
@ -256,6 +174,10 @@ var _ = Describe("Stream", func() {
})
It("ignores duplicate StreamFrames", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2)).Times(2)
frame1 := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD},
@ -282,6 +204,10 @@ var _ = Describe("Stream", func() {
})
It("doesn't rejects a StreamFrames with an overlapping data range", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(6))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(4))
frame1 := frames.StreamFrame{
Offset: 0,
Data: []byte("foob"),
@ -293,6 +219,7 @@ var _ = Describe("Stream", func() {
err := str.AddStreamFrame(&frame1)
Expect(err).ToNot(HaveOccurred())
err = str.AddStreamFrame(&frame2)
Expect(err).ToNot(HaveOccurred())
b := make([]byte, 6)
n, err := str.Read(b)
Expect(err).ToNot(HaveOccurred())
@ -301,6 +228,8 @@ var _ = Describe("Stream", func() {
})
It("calls onData", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -315,6 +244,8 @@ var _ = Describe("Stream", func() {
Context("closing", func() {
Context("with FIN bit", func() {
It("returns EOFs", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -332,6 +263,9 @@ var _ = Describe("Stream", func() {
})
It("handles out-of-order frames", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2)).Times(2)
frame1 := frames.StreamFrame{
Offset: 2,
Data: []byte{0xBE, 0xEF},
@ -356,6 +290,8 @@ var _ = Describe("Stream", func() {
})
It("returns EOFs with partial read", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(2))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(2))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD},
@ -371,6 +307,8 @@ var _ = Describe("Stream", func() {
})
It("handles immediate FINs", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(0))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(0))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{},
@ -387,6 +325,8 @@ var _ = Describe("Stream", func() {
Context("when CloseRemote is called", func() {
It("closes", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(0))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(0))
str.CloseRemote(0)
b := make([]byte, 8)
n, err := str.Read(b)
@ -430,6 +370,7 @@ var _ = Describe("Stream", func() {
Context("reset by the peer", func() {
It("continues reading after receiving a remote error", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -443,6 +384,7 @@ var _ = Describe("Stream", func() {
})
It("reads a delayed StreamFrame that arrives after receiving a remote error", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
str.RegisterRemoteError(testErr)
frame := frames.StreamFrame{
Offset: 0,
@ -457,6 +399,7 @@ var _ = Describe("Stream", func() {
})
It("returns the error if reading past the offset of the frame received", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -471,6 +414,7 @@ var _ = Describe("Stream", func() {
})
It("returns an EOF when reading past the offset, if the stream received a finbit", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -486,6 +430,7 @@ var _ = Describe("Stream", func() {
})
It("continues reading in small chunks after receiving a remote error", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
frame := frames.StreamFrame{
Offset: 0,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
@ -505,19 +450,18 @@ var _ = Describe("Stream", func() {
})
It("doesn't inform the flow controller about bytes read after receiving the remote error", func() {
str.flowControlManager = newMockFlowControlHandler()
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(4))
// No AddBytesRead()
frame := frames.StreamFrame{
Offset: 0,
StreamID: 5,
Data: []byte{0xDE, 0xAD, 0xBE, 0xEF},
}
str.AddStreamFrame(&frame)
str.flowControlManager.ResetStream(5, 4)
str.RegisterRemoteError(testErr)
b := make([]byte, 3)
_, err := str.Read(b)
Expect(err).ToNot(HaveOccurred())
Expect(str.flowControlManager.(*mockFlowControlHandler).bytesRead).To(BeEquivalentTo(4))
})
It("stops writing after receiving a remote error", func() {
@ -617,6 +561,7 @@ var _ = Describe("Stream", func() {
})
It("doesn't allow further reads", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(6))
str.AddStreamFrame(&frames.StreamFrame{
Data: []byte("foobar"),
})
@ -801,32 +746,15 @@ var _ = Describe("Stream", func() {
})
})
Context("flow control, for receiving", func() {
BeforeEach(func() {
str.flowControlManager = &mockFlowControlHandler{}
})
It("updates the highestReceived value in the flow controller", func() {
frame := frames.StreamFrame{
Offset: 2,
Data: []byte("foobar"),
}
err := str.AddStreamFrame(&frame)
Expect(err).ToNot(HaveOccurred())
Expect(str.flowControlManager.(*mockFlowControlHandler).highestReceivedForStream).To(Equal(str.streamID))
Expect(str.flowControlManager.(*mockFlowControlHandler).highestReceived).To(Equal(protocol.ByteCount(2 + 6)))
})
It("errors when a StreamFrames causes a flow control violation", func() {
testErr := errors.New("flow control violation")
str.flowControlManager.(*mockFlowControlHandler).flowControlViolation = testErr
frame := frames.StreamFrame{
Offset: 2,
Data: []byte("foobar"),
}
err := str.AddStreamFrame(&frame)
Expect(err).To(MatchError(testErr))
})
It("errors when a StreamFrames causes a flow control violation", func() {
testErr := errors.New("flow control violation")
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(8)).Return(testErr)
frame := frames.StreamFrame{
Offset: 2,
Data: []byte("foobar"),
}
err := str.AddStreamFrame(&frame)
Expect(err).To(MatchError(testErr))
})
Context("closing", func() {
@ -852,6 +780,8 @@ var _ = Describe("Stream", func() {
})
It("is not finished if it is only closed for reading", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(0))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(0))
finishReading()
Expect(str.finished()).To(BeFalse())
})
@ -878,11 +808,12 @@ var _ = Describe("Stream", func() {
})
It("is finished after finishing reading and being locally reset", func() {
mockFcm.EXPECT().UpdateHighestReceived(streamID, protocol.ByteCount(0))
mockFcm.EXPECT().AddBytesRead(streamID, protocol.ByteCount(0))
finishReading()
Expect(str.finished()).To(BeFalse())
str.Reset(testErr)
Expect(str.finished()).To(BeTrue())
})
})
})