From 0f1f1c8d4192c9ceeffaf269947f11cd435176f9 Mon Sep 17 00:00:00 2001 From: Marten Seemann Date: Wed, 1 Nov 2017 15:29:15 +0700 Subject: [PATCH] implement the MAX_DATA and MAX_STREAM_DATA frames For gQUIC WINDOW_UPDATEs are converted to MAX_DATA and MAX_STREAM_DATA, respectively. --- ackhandler/packet_test.go | 6 +- ackhandler/retransmittable_test.go | 3 +- internal/protocol/version.go | 5 + internal/protocol/version_test.go | 7 ++ internal/wire/max_data_frame.go | 51 +++++++++ internal/wire/max_data_frame_test.go | 56 ++++++++++ internal/wire/max_stream_data_frame.go | 56 ++++++++++ internal/wire/max_stream_data_frame_test.go | 62 ++++++++++ internal/wire/window_update_frame.go | 49 +++----- internal/wire/window_update_frame_test.go | 90 ++++++++++----- internal/wire/wire_suite_test.go | 12 +- packet_packer_test.go | 10 +- packet_unpacker.go | 79 +++++++------ packet_unpacker_test.go | 118 ++++++++++++++++---- quic_suite_test.go | 11 ++ session.go | 36 +++--- session_test.go | 22 ++-- streams_map_test.go | 7 -- 18 files changed, 512 insertions(+), 168 deletions(-) create mode 100644 internal/wire/max_data_frame.go create mode 100644 internal/wire/max_data_frame_test.go create mode 100644 internal/wire/max_stream_data_frame.go create mode 100644 internal/wire/max_stream_data_frame_test.go diff --git a/ackhandler/packet_test.go b/ackhandler/packet_test.go index 4d3ae560..6fecf47c 100644 --- a/ackhandler/packet_test.go +++ b/ackhandler/packet_test.go @@ -10,7 +10,7 @@ var _ = Describe("Packet", func() { Context("getting frames for retransmission", func() { ackFrame := &wire.AckFrame{LargestAcked: 13} stopWaitingFrame := &wire.StopWaitingFrame{LeastUnacked: 7331} - windowUpdateFrame := &wire.WindowUpdateFrame{StreamID: 999} + maxStreamDataFrame := &wire.MaxStreamDataFrame{StreamID: 999} streamFrame := &wire.StreamFrame{ StreamID: 5, @@ -32,7 +32,7 @@ var _ = Describe("Packet", func() { It("returns all retransmittable frames", func() { packet := &Packet{ Frames: []wire.Frame{ - windowUpdateFrame, + maxStreamDataFrame, ackFrame, stopWaitingFrame, streamFrame, @@ -42,7 +42,7 @@ var _ = Describe("Packet", func() { fs := packet.GetFramesForRetransmission() Expect(fs).To(ContainElement(streamFrame)) Expect(fs).To(ContainElement(rstStreamFrame)) - Expect(fs).To(ContainElement(windowUpdateFrame)) + Expect(fs).To(ContainElement(maxStreamDataFrame)) Expect(fs).ToNot(ContainElement(stopWaitingFrame)) Expect(fs).ToNot(ContainElement(ackFrame)) }) diff --git a/ackhandler/retransmittable_test.go b/ackhandler/retransmittable_test.go index aecca065..63d0a589 100644 --- a/ackhandler/retransmittable_test.go +++ b/ackhandler/retransmittable_test.go @@ -18,7 +18,8 @@ var _ = Describe("retransmittable frames", func() { &wire.PingFrame{}: true, &wire.RstStreamFrame{}: true, &wire.StreamFrame{}: true, - &wire.WindowUpdateFrame{}: true, + &wire.MaxDataFrame{}: true, + &wire.MaxStreamDataFrame{}: true, } { f := fl e := el diff --git a/internal/protocol/version.go b/internal/protocol/version.go index bf206c01..c3cec86f 100644 --- a/internal/protocol/version.go +++ b/internal/protocol/version.go @@ -68,6 +68,11 @@ func (vn VersionNumber) CryptoStreamID() StreamID { return 0 } +// UsesMaxDataFrame tells if this version uses MAX_DATA, MAX_STREAM_DATA, BLOCKED and STREAM_BLOCKED instead of WINDOW_UDPATE and BLOCKED frames +func (vn VersionNumber) UsesMaxDataFrame() bool { + return vn.CryptoStreamID() == 0 +} + // StreamContributesToConnectionFlowControl says if a stream contributes to connection-level flow control func (vn VersionNumber) StreamContributesToConnectionFlowControl(id StreamID) bool { if id == vn.CryptoStreamID() { diff --git a/internal/protocol/version_test.go b/internal/protocol/version_test.go index 6d9fff3e..6a25a9a7 100644 --- a/internal/protocol/version_test.go +++ b/internal/protocol/version_test.go @@ -52,6 +52,13 @@ var _ = Describe("Version", func() { Expect(VersionTLS.CryptoStreamID()).To(Equal(StreamID(0))) }) + It("tells if a version uses the MAX_DATA, MAX_STREAM_DATA, BLOCKED and STREAM_BLOCKED frames", func() { + Expect(Version37.UsesMaxDataFrame()).To(BeFalse()) + Expect(Version38.UsesMaxDataFrame()).To(BeFalse()) + Expect(Version39.UsesMaxDataFrame()).To(BeFalse()) + Expect(VersionTLS.UsesMaxDataFrame()).To(BeTrue()) + }) + It("says if a stream contributes to connection-level flowcontrol, for gQUIC", func() { Expect(Version39.StreamContributesToConnectionFlowControl(1)).To(BeFalse()) Expect(Version39.StreamContributesToConnectionFlowControl(2)).To(BeTrue()) diff --git a/internal/wire/max_data_frame.go b/internal/wire/max_data_frame.go new file mode 100644 index 00000000..cd3ff655 --- /dev/null +++ b/internal/wire/max_data_frame.go @@ -0,0 +1,51 @@ +package wire + +import ( + "bytes" + + "github.com/lucas-clemente/quic-go/internal/protocol" + "github.com/lucas-clemente/quic-go/internal/utils" +) + +// A MaxDataFrame carries flow control information for the connection +type MaxDataFrame struct { + ByteOffset protocol.ByteCount +} + +// ParseMaxDataFrame parses a MAX_DATA frame +func ParseMaxDataFrame(r *bytes.Reader, version protocol.VersionNumber) (*MaxDataFrame, error) { + // read the TypeByte + if _, err := r.ReadByte(); err != nil { + return nil, err + } + + frame := &MaxDataFrame{} + byteOffset, err := utils.GetByteOrder(version).ReadUint64(r) + if err != nil { + return nil, err + } + frame.ByteOffset = protocol.ByteCount(byteOffset) + return frame, nil +} + +//Write writes a MAX_STREAM_DATA frame +func (f *MaxDataFrame) Write(b *bytes.Buffer, version protocol.VersionNumber) error { + if !version.UsesMaxDataFrame() { + // write a gQUIC WINDOW_UPDATE frame (with stream ID 0, which means connection-level there) + return (&windowUpdateFrame{ + StreamID: 0, + ByteOffset: f.ByteOffset, + }).Write(b, version) + } + b.WriteByte(0x4) + utils.GetByteOrder(version).WriteUint64(b, uint64(f.ByteOffset)) + return nil +} + +// MinLength of a written frame +func (f *MaxDataFrame) MinLength(version protocol.VersionNumber) (protocol.ByteCount, error) { + if !version.UsesMaxDataFrame() { // writing this frame would result in a gQUIC WINDOW_UPDATE being written, which is longer + return 1 + 4 + 8, nil + } + return 1 + 8, nil +} diff --git a/internal/wire/max_data_frame_test.go b/internal/wire/max_data_frame_test.go new file mode 100644 index 00000000..e4cd75c1 --- /dev/null +++ b/internal/wire/max_data_frame_test.go @@ -0,0 +1,56 @@ +package wire + +import ( + "bytes" + + "github.com/lucas-clemente/quic-go/internal/protocol" + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" +) + +var _ = Describe("MAX_DATA frame", func() { + Context("when parsing", func() { + It("accepts sample frame", func() { + b := bytes.NewReader([]byte{0x4, + 0xde, 0xca, 0xfb, 0xad, 0x11, 0x22, 0x33, 0x44, // byte offset + }) + frame, err := ParseMaxDataFrame(b, versionBigEndian) + Expect(err).ToNot(HaveOccurred()) + Expect(frame.ByteOffset).To(Equal(protocol.ByteCount(0xdecafbad11223344))) + Expect(b.Len()).To(BeZero()) + }) + + It("errors on EOFs", func() { + data := []byte{0x4, + 0x44, 0x33, 0x22, 0x11, 0xad, 0xfb, 0xca, 0xde, // byte offset + } + _, err := ParseMaxDataFrame(bytes.NewReader(data), versionMaxDataFrame) + Expect(err).NotTo(HaveOccurred()) + for i := range data { + _, err := ParseMaxDataFrame(bytes.NewReader(data[0:i]), versionMaxDataFrame) + Expect(err).To(HaveOccurred()) + } + }) + }) + + Context("writing", func() { + It("has proper min length", func() { + f := &MaxDataFrame{ + ByteOffset: 0xdeadbeef, + } + Expect(f.MinLength(versionMaxDataFrame)).To(Equal(protocol.ByteCount(1 + 8))) + }) + + It("writes a MAX_DATA frame", func() { + b := &bytes.Buffer{} + f := &MaxDataFrame{ + ByteOffset: 0xdeadbeefcafe1337, + } + err := f.Write(b, versionMaxDataFrame) + Expect(err).ToNot(HaveOccurred()) + Expect(b.Bytes()).To(Equal([]byte{0x4, + 0xde, 0xad, 0xbe, 0xef, 0xca, 0xfe, 0x13, 0x37, // byte offset + })) + }) + }) +}) diff --git a/internal/wire/max_stream_data_frame.go b/internal/wire/max_stream_data_frame.go new file mode 100644 index 00000000..56c44c9f --- /dev/null +++ b/internal/wire/max_stream_data_frame.go @@ -0,0 +1,56 @@ +package wire + +import ( + "bytes" + + "github.com/lucas-clemente/quic-go/internal/protocol" + "github.com/lucas-clemente/quic-go/internal/utils" +) + +// A MaxStreamDataFrame carries flow control information for a stream +type MaxStreamDataFrame struct { + StreamID protocol.StreamID + ByteOffset protocol.ByteCount +} + +// ParseMaxStreamDataFrame parses a MAX_STREAM_DATA frame +func ParseMaxStreamDataFrame(r *bytes.Reader, version protocol.VersionNumber) (*MaxStreamDataFrame, error) { + frame := &MaxStreamDataFrame{} + + // read the TypeByte + if _, err := r.ReadByte(); err != nil { + return nil, err + } + + sid, err := utils.GetByteOrder(version).ReadUint32(r) + if err != nil { + return nil, err + } + frame.StreamID = protocol.StreamID(sid) + + byteOffset, err := utils.GetByteOrder(version).ReadUint64(r) + if err != nil { + return nil, err + } + frame.ByteOffset = protocol.ByteCount(byteOffset) + return frame, nil +} + +// Write writes a MAX_STREAM_DATA frame +func (f *MaxStreamDataFrame) Write(b *bytes.Buffer, version protocol.VersionNumber) error { + if !version.UsesMaxDataFrame() { + return (&windowUpdateFrame{ + StreamID: f.StreamID, + ByteOffset: f.ByteOffset, + }).Write(b, version) + } + b.WriteByte(0x5) + utils.GetByteOrder(version).WriteUint32(b, uint32(f.StreamID)) + utils.GetByteOrder(version).WriteUint64(b, uint64(f.ByteOffset)) + return nil +} + +// MinLength of a written frame +func (f *MaxStreamDataFrame) MinLength(version protocol.VersionNumber) (protocol.ByteCount, error) { + return 1 + 4 + 8, nil +} diff --git a/internal/wire/max_stream_data_frame_test.go b/internal/wire/max_stream_data_frame_test.go new file mode 100644 index 00000000..6cad6ac1 --- /dev/null +++ b/internal/wire/max_stream_data_frame_test.go @@ -0,0 +1,62 @@ +package wire + +import ( + "bytes" + + "github.com/lucas-clemente/quic-go/internal/protocol" + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" +) + +var _ = Describe("MAX_STREAM_DATA frame", func() { + Context("parsing", func() { + It("accepts sample frame", func() { + b := bytes.NewReader([]byte{0x5, + 0xde, 0xad, 0xbe, 0xef, // stream id + 0xde, 0xca, 0xfb, 0xad, 0x11, 0x22, 0x33, 0x44, // byte offset + }) + frame, err := ParseMaxStreamDataFrame(b, versionMaxDataFrame) + Expect(err).ToNot(HaveOccurred()) + Expect(frame.StreamID).To(Equal(protocol.StreamID(0xdeadbeef))) + Expect(frame.ByteOffset).To(Equal(protocol.ByteCount(0xdecafbad11223344))) + Expect(b.Len()).To(BeZero()) + }) + + It("errors on EOFs", func() { + data := []byte{0x5, + 0xef, 0xbe, 0xad, 0xde, // stream id + 0x44, 0x33, 0x22, 0x11, 0xad, 0xfb, 0xca, 0xde, // byte offset + } + _, err := ParseMaxStreamDataFrame(bytes.NewReader(data), versionMaxDataFrame) + Expect(err).NotTo(HaveOccurred()) + for i := range data { + _, err := ParseMaxStreamDataFrame(bytes.NewReader(data[0:i]), versionMaxDataFrame) + Expect(err).To(HaveOccurred()) + } + }) + }) + + Context("writing", func() { + It("has proper min length", func() { + f := &MaxStreamDataFrame{ + StreamID: 0x1337, + ByteOffset: 0xdeadbeef, + } + Expect(f.MinLength(0)).To(Equal(protocol.ByteCount(13))) + }) + + It("writes a sample frame", func() { + b := &bytes.Buffer{} + f := &MaxStreamDataFrame{ + StreamID: 0xdecafbad, + ByteOffset: 0xdeadbeefcafe1337, + } + err := f.Write(b, versionMaxDataFrame) + Expect(err).ToNot(HaveOccurred()) + Expect(b.Bytes()).To(Equal([]byte{0x5, + 0xde, 0xca, 0xfb, 0xad, // stream id + 0xde, 0xad, 0xbe, 0xef, 0xca, 0xfe, 0x13, 0x37, // byte offset + })) + }) + }) +}) diff --git a/internal/wire/window_update_frame.go b/internal/wire/window_update_frame.go index fe896e7a..20d7b66f 100644 --- a/internal/wire/window_update_frame.go +++ b/internal/wire/window_update_frame.go @@ -7,44 +7,29 @@ import ( "github.com/lucas-clemente/quic-go/internal/utils" ) -// A WindowUpdateFrame in QUIC -type WindowUpdateFrame struct { +type windowUpdateFrame struct { StreamID protocol.StreamID ByteOffset protocol.ByteCount } -//Write writes a RST_STREAM frame -func (f *WindowUpdateFrame) Write(b *bytes.Buffer, version protocol.VersionNumber) error { +// ParseWindowUpdateFrame parses a WINDOW_UPDATE frame +// The frame returned is +// * a MAX_STREAM_DATA frame, if the WINDOW_UPDATE applies to a stream +// * a MAX_DATA frame, if the WINDOW_UPDATE applies to the connection +func ParseWindowUpdateFrame(r *bytes.Reader, version protocol.VersionNumber) (Frame, error) { + f, err := ParseMaxStreamDataFrame(r, version) + if err != nil { + return nil, err + } + if f.StreamID == 0 { + return &MaxDataFrame{ByteOffset: f.ByteOffset}, nil + } + return f, nil +} + +func (f *windowUpdateFrame) Write(b *bytes.Buffer, version protocol.VersionNumber) error { b.WriteByte(0x4) utils.GetByteOrder(version).WriteUint32(b, uint32(f.StreamID)) utils.GetByteOrder(version).WriteUint64(b, uint64(f.ByteOffset)) return nil } - -// MinLength of a written frame -func (f *WindowUpdateFrame) MinLength(version protocol.VersionNumber) (protocol.ByteCount, error) { - return 1 + 4 + 8, nil -} - -// ParseWindowUpdateFrame parses a RST_STREAM frame -func ParseWindowUpdateFrame(r *bytes.Reader, version protocol.VersionNumber) (*WindowUpdateFrame, error) { - frame := &WindowUpdateFrame{} - - // read the TypeByte - if _, err := r.ReadByte(); err != nil { - return nil, err - } - - sid, err := utils.GetByteOrder(version).ReadUint32(r) - if err != nil { - return nil, err - } - frame.StreamID = protocol.StreamID(sid) - - byteOffset, err := utils.GetByteOrder(version).ReadUint64(r) - if err != nil { - return nil, err - } - frame.ByteOffset = protocol.ByteCount(byteOffset) - return frame, nil -} diff --git a/internal/wire/window_update_frame_test.go b/internal/wire/window_update_frame_test.go index a54f004d..5086cd77 100644 --- a/internal/wire/window_update_frame_test.go +++ b/internal/wire/window_update_frame_test.go @@ -8,16 +8,18 @@ import ( . "github.com/onsi/gomega" ) -var _ = Describe("WindowUpdateFrame", func() { - Context("when parsing", func() { +var _ = Describe("WINDOW_UPDATE frame", func() { + Context("parsing", func() { Context("in little endian", func() { It("accepts sample frame", func() { b := bytes.NewReader([]byte{0x4, 0xef, 0xbe, 0xad, 0xde, // stream id 0x44, 0x33, 0x22, 0x11, 0xad, 0xfb, 0xca, 0xde, // byte offset }) - frame, err := ParseWindowUpdateFrame(b, versionLittleEndian) + f, err := ParseWindowUpdateFrame(b, versionLittleEndian) Expect(err).ToNot(HaveOccurred()) + Expect(f).To(BeAssignableToTypeOf(&MaxStreamDataFrame{})) + frame := f.(*MaxStreamDataFrame) Expect(frame.StreamID).To(Equal(protocol.StreamID(0xdeadbeef))) Expect(frame.ByteOffset).To(Equal(protocol.ByteCount(0xdecafbad11223344))) Expect(b.Len()).To(BeZero()) @@ -25,46 +27,65 @@ var _ = Describe("WindowUpdateFrame", func() { }) Context("in big endian", func() { - It("accepts sample frame", func() { + It("parses a stream-level WINDOW_UPDATE", func() { b := bytes.NewReader([]byte{0x4, 0xde, 0xad, 0xbe, 0xef, // stream id 0xde, 0xca, 0xfb, 0xad, 0x11, 0x22, 0x33, 0x44, // byte offset }) - frame, err := ParseWindowUpdateFrame(b, versionBigEndian) + f, err := ParseWindowUpdateFrame(b, versionBigEndian) Expect(err).ToNot(HaveOccurred()) - Expect(frame.StreamID).To(Equal(protocol.StreamID(0xdeadbeef))) + Expect(f).To(BeAssignableToTypeOf(&MaxStreamDataFrame{})) + frame := f.(*MaxStreamDataFrame) Expect(frame.ByteOffset).To(Equal(protocol.ByteCount(0xdecafbad11223344))) - Expect(b.Len()).To(BeZero()) + Expect(frame.StreamID).To(Equal(protocol.StreamID(0xdeadbeef))) }) - }) - It("errors on EOFs", func() { - data := []byte{0x4, - 0xef, 0xbe, 0xad, 0xde, // stream id - 0x44, 0x33, 0x22, 0x11, 0xad, 0xfb, 0xca, 0xde, // byte offset - } - _, err := ParseWindowUpdateFrame(bytes.NewReader(data), protocol.VersionWhatever) - Expect(err).NotTo(HaveOccurred()) - for i := range data { - _, err := ParseWindowUpdateFrame(bytes.NewReader(data[0:i]), protocol.VersionWhatever) - Expect(err).To(HaveOccurred()) - } + It("parses a connection-level WINDOW_UPDATE", func() { + b := bytes.NewReader([]byte{0x4, + 0x0, 0x0, 0x0, 0x0, // stream id + 0xde, 0xca, 0xfb, 0xad, 0x11, 0x22, 0x33, 0x44, // byte offset + }) + f, err := ParseWindowUpdateFrame(b, versionBigEndian) + Expect(err).ToNot(HaveOccurred()) + Expect(f).To(BeAssignableToTypeOf(&MaxDataFrame{})) + frame := f.(*MaxDataFrame) + Expect(frame.ByteOffset).To(Equal(protocol.ByteCount(0xdecafbad11223344))) + }) + + It("errors on EOFs", func() { + data := []byte{0x4, + 0xef, 0xbe, 0xad, 0xde, // stream id + 0x44, 0x33, 0x22, 0x11, 0xad, 0xfb, 0xca, 0xde, // byte offset + } + _, err := ParseWindowUpdateFrame(bytes.NewReader(data), versionBigEndian) + Expect(err).NotTo(HaveOccurred()) + for i := range data { + _, err := ParseWindowUpdateFrame(bytes.NewReader(data[0:i]), versionBigEndian) + Expect(err).To(HaveOccurred()) + } + }) }) }) - Context("when writing", func() { - It("has proper min length", func() { - f := &WindowUpdateFrame{ - StreamID: 0x1337, + Context("writing", func() { + It("has the proper min length for the stream-level WINDOW_UPDATE frame", func() { + f := &MaxDataFrame{ ByteOffset: 0xdeadbeef, } - Expect(f.MinLength(0)).To(Equal(protocol.ByteCount(13))) + Expect(f.MinLength(versionBigEndian)).To(Equal(protocol.ByteCount(1 + 4 + 8))) + }) + + It("has the proper min length for the connection-level WINDOW_UPDATE frame", func() { + f := &MaxDataFrame{ + ByteOffset: 0xdeadbeef, + } + Expect(f.MinLength(versionBigEndian)).To(Equal(protocol.ByteCount(1 + 4 + 8))) }) Context("in little endian", func() { It("writes a sample frame", func() { b := &bytes.Buffer{} - f := &WindowUpdateFrame{ + f := &windowUpdateFrame{ StreamID: 0xdecafbad, ByteOffset: 0xdeadbeefcafe1337, } @@ -78,16 +99,29 @@ var _ = Describe("WindowUpdateFrame", func() { }) Context("in big endian", func() { - It("writes a sample frame", func() { + It("writes a stream-level WINDOW_UPDATE frame", func() { b := &bytes.Buffer{} - f := &WindowUpdateFrame{ + f := &MaxStreamDataFrame{ StreamID: 0xdecafbad, ByteOffset: 0xdeadbeefcafe1337, } err := f.Write(b, versionBigEndian) Expect(err).ToNot(HaveOccurred()) Expect(b.Bytes()).To(Equal([]byte{0x4, - 0xde, 0xca, 0xfb, 0xad, // stream id + 0xde, 0xca, 0xfb, 0xad, // stream ID 0 + 0xde, 0xad, 0xbe, 0xef, 0xca, 0xfe, 0x13, 0x37, // byte offset + })) + }) + + It("writes a connection-level WINDOW_UPDATE frame", func() { + b := &bytes.Buffer{} + f := &MaxDataFrame{ + ByteOffset: 0xdeadbeefcafe1337, + } + err := f.Write(b, versionBigEndian) + Expect(err).ToNot(HaveOccurred()) + Expect(b.Bytes()).To(Equal([]byte{0x4, + 0x0, 0x0, 0x0, 0x0, // stream ID 0 0xde, 0xad, 0xbe, 0xef, 0xca, 0xfe, 0x13, 0x37, // byte offset })) }) diff --git a/internal/wire/wire_suite_test.go b/internal/wire/wire_suite_test.go index 322b9ede..4cd39e6f 100644 --- a/internal/wire/wire_suite_test.go +++ b/internal/wire/wire_suite_test.go @@ -15,11 +15,19 @@ func TestCrypto(t *testing.T) { } const ( - versionLittleEndian = protocol.Version37 // a QUIC version that uses little endian encoding - versionBigEndian = protocol.Version39 // a QUIC version that uses big endian encoding + // a QUIC version that uses little endian encoding + versionLittleEndian = protocol.Version37 + // a QUIC version that uses big endian encoding + versionBigEndian = protocol.Version39 + // a QUIC version that uses the MAX_DATA / MAX_STREAM_DATA and BLOCKED / STREAM_BLOCKED frames + versionMaxDataFrame = protocol.VersionTLS ) var _ = BeforeSuite(func() { Expect(utils.GetByteOrder(versionLittleEndian)).To(Equal(utils.LittleEndian)) Expect(utils.GetByteOrder(versionBigEndian)).To(Equal(utils.BigEndian)) + Expect(utils.GetByteOrder(versionMaxDataFrame)).To(Equal(utils.BigEndian)) + Expect(versionLittleEndian.UsesMaxDataFrame()).To(BeFalse()) + Expect(versionBigEndian.UsesMaxDataFrame()).To(BeFalse()) + Expect(versionMaxDataFrame.UsesMaxDataFrame()).To(BeTrue()) }) diff --git a/packet_packer_test.go b/packet_packer_test.go index d4a62378..100f6c81 100644 --- a/packet_packer_test.go +++ b/packet_packer_test.go @@ -236,7 +236,7 @@ var _ = Describe("Packet packer", func() { ErrorCode: 0x1337, ReasonPhrase: "foobar", } - packer.controlFrames = []wire.Frame{&wire.WindowUpdateFrame{StreamID: 37}} + packer.controlFrames = []wire.Frame{&wire.MaxStreamDataFrame{StreamID: 37}} streamFramer.AddFrameForRetransmission(&wire.StreamFrame{ StreamID: 5, Data: []byte("foobar"), @@ -249,7 +249,7 @@ var _ = Describe("Packet packer", func() { It("packs only control frames", func() { packer.QueueControlFrame(&wire.RstStreamFrame{}) - packer.QueueControlFrame(&wire.WindowUpdateFrame{}) + packer.QueueControlFrame(&wire.MaxDataFrame{}) p, err := packer.PackPacket() Expect(p).ToNot(BeNil()) Expect(err).ToNot(HaveOccurred()) @@ -670,12 +670,12 @@ var _ = Describe("Packet packer", func() { }) It("queues a control frame to be sent in the next packet", func() { - wuf := &wire.WindowUpdateFrame{StreamID: 5} - packer.QueueControlFrame(wuf) + msd := &wire.MaxStreamDataFrame{StreamID: 5} + packer.QueueControlFrame(msd) p, err := packer.PackPacket() Expect(err).NotTo(HaveOccurred()) Expect(p.frames).To(HaveLen(1)) - Expect(p.frames[0]).To(Equal(wuf)) + Expect(p.frames[0]).To(Equal(msd)) }) Context("retransmitting of handshake packets", func() { diff --git a/packet_unpacker.go b/packet_unpacker.go index d31e535f..638ab1a3 100644 --- a/packet_unpacker.go +++ b/packet_unpacker.go @@ -63,43 +63,50 @@ func (u *packetUnpacker) Unpack(headerBinary []byte, hdr *wire.Header, data []by if err != nil { err = qerr.Error(qerr.InvalidAckData, err.Error()) } - } else { - switch typeByte { - case 0x01: - frame, err = wire.ParseRstStreamFrame(r, u.version) - if err != nil { - err = qerr.Error(qerr.InvalidRstStreamData, err.Error()) - } - case 0x02: - frame, err = wire.ParseConnectionCloseFrame(r, u.version) - if err != nil { - err = qerr.Error(qerr.InvalidConnectionCloseData, err.Error()) - } - case 0x03: - frame, err = wire.ParseGoawayFrame(r, u.version) - if err != nil { - err = qerr.Error(qerr.InvalidGoawayData, err.Error()) - } - case 0x04: - frame, err = wire.ParseWindowUpdateFrame(r, u.version) - if err != nil { - err = qerr.Error(qerr.InvalidWindowUpdateData, err.Error()) - } - case 0x05: - frame, err = wire.ParseBlockedFrame(r, u.version) - if err != nil { - err = qerr.Error(qerr.InvalidBlockedData, err.Error()) - } - case 0x06: - frame, err = wire.ParseStopWaitingFrame(r, hdr.PacketNumber, hdr.PacketNumberLen, u.version) - if err != nil { - err = qerr.Error(qerr.InvalidStopWaitingData, err.Error()) - } - case 0x07: - frame, err = wire.ParsePingFrame(r, u.version) - default: - err = qerr.Error(qerr.InvalidFrameData, fmt.Sprintf("unknown type byte 0x%x", typeByte)) + } else if typeByte == 0x01 { + frame, err = wire.ParseRstStreamFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidRstStreamData, err.Error()) } + } else if typeByte == 0x02 { + frame, err = wire.ParseConnectionCloseFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidConnectionCloseData, err.Error()) + } + } else if typeByte == 0x3 { + frame, err = wire.ParseGoawayFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidGoawayData, err.Error()) + } + } else if u.version.UsesMaxDataFrame() && typeByte == 0x4 { // in IETF QUIC, 0x4 is a MAX_DATA frame + frame, err = wire.ParseMaxDataFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidWindowUpdateData, err.Error()) + } + } else if typeByte == 0x4 { // in gQUIC, 0x4 is a WINDOW_UPDATE frame + frame, err = wire.ParseWindowUpdateFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidWindowUpdateData, err.Error()) + } + } else if u.version.UsesMaxDataFrame() && typeByte == 0x5 { // in IETF QUIC, 0x5 is a MAX_STREAM_DATA frame + frame, err = wire.ParseMaxStreamDataFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidWindowUpdateData, err.Error()) + } + } else if typeByte == 0x5 { // in gQUIC, 0x5 is a BLOCKED frame + frame, err = wire.ParseBlockedFrame(r, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidBlockedData, err.Error()) + } + } else if typeByte == 0x6 { + frame, err = wire.ParseStopWaitingFrame(r, hdr.PacketNumber, hdr.PacketNumberLen, u.version) + if err != nil { + err = qerr.Error(qerr.InvalidStopWaitingData, err.Error()) + } + } else if typeByte == 0x7 { + frame, err = wire.ParsePingFrame(r, u.version) + } else { + err = qerr.Error(qerr.InvalidFrameData, fmt.Sprintf("unknown type byte 0x%x", typeByte)) } if err != nil { return nil, err diff --git a/packet_unpacker_test.go b/packet_unpacker_test.go index be8d608d..3e56dd17 100644 --- a/packet_unpacker_test.go +++ b/packet_unpacker_test.go @@ -151,29 +151,103 @@ var _ = Describe("Packet unpacker", func() { Expect(packet.frames).To(Equal([]wire.Frame{f})) }) - It("unpacks WINDOW_UPDATE frames", func() { - f := &wire.WindowUpdateFrame{ - StreamID: 0xDEADBEEF, - ByteOffset: 0xCAFE000000001337, - } - buf := &bytes.Buffer{} - err := f.Write(buf, protocol.VersionWhatever) - Expect(err).ToNot(HaveOccurred()) - setData(buf.Bytes()) - packet, err := unpacker.Unpack(hdrBin, hdr, data) - Expect(err).ToNot(HaveOccurred()) - Expect(packet.frames).To(Equal([]wire.Frame{f})) + Context("flow control frames, when the crypto stream is stream 0", func() { + BeforeEach(func() { + unpacker.version = versionCryptoStream0 + }) + + It("unpacks MAX_DATA frames", func() { + f := &wire.MaxDataFrame{ + ByteOffset: 0xcafe000000001337, + } + buf := &bytes.Buffer{} + err := f.Write(buf, versionCryptoStream0) + Expect(err).ToNot(HaveOccurred()) + setData(buf.Bytes()) + packet, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err).ToNot(HaveOccurred()) + Expect(packet.frames).To(Equal([]wire.Frame{f})) + }) + + It("unpacks MAX_STREAM_DATA frames", func() { + f := &wire.MaxStreamDataFrame{ + StreamID: 0xdeadbeef, + ByteOffset: 0xcafe000000001337, + } + buf := &bytes.Buffer{} + err := f.Write(buf, versionCryptoStream0) + Expect(err).ToNot(HaveOccurred()) + setData(buf.Bytes()) + packet, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err).ToNot(HaveOccurred()) + Expect(packet.frames).To(Equal([]wire.Frame{f})) + }) + + It("errors on invalid frames", func() { + for b, e := range map[byte]qerr.ErrorCode{ + 0x04: qerr.InvalidWindowUpdateData, + 0x05: qerr.InvalidWindowUpdateData, + } { + setData([]byte{b}) + _, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err.(*qerr.QuicError).ErrorCode).To(Equal(e)) + } + }) }) - It("unpakcs BLOCKED frames", func() { - f := &wire.BlockedFrame{StreamID: 0xDEADBEEF} - buf := &bytes.Buffer{} - err := f.Write(buf, protocol.VersionWhatever) - Expect(err).ToNot(HaveOccurred()) - setData(buf.Bytes()) - packet, err := unpacker.Unpack(hdrBin, hdr, data) - Expect(err).ToNot(HaveOccurred()) - Expect(packet.frames).To(Equal([]wire.Frame{f})) + Context("flow control frames, when the crypto stream is stream 1", func() { + BeforeEach(func() { + unpacker.version = versionCryptoStream1 + }) + + It("unpacks a stream-level WINDOW_UPDATE frame", func() { + f := &wire.MaxStreamDataFrame{ + StreamID: 0xdeadbeef, + ByteOffset: 0xcafe000000001337, + } + buf := &bytes.Buffer{} + err := f.Write(buf, versionCryptoStream1) + Expect(err).ToNot(HaveOccurred()) + setData(buf.Bytes()) + packet, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err).ToNot(HaveOccurred()) + Expect(packet.frames).To(Equal([]wire.Frame{f})) + }) + + It("unpacks a connection-level WINDOW_UPDATE frame", func() { + f := &wire.MaxDataFrame{ + ByteOffset: 0xcafe000000001337, + } + buf := &bytes.Buffer{} + err := f.Write(buf, versionCryptoStream1) + Expect(err).ToNot(HaveOccurred()) + setData(buf.Bytes()) + packet, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err).ToNot(HaveOccurred()) + Expect(packet.frames).To(Equal([]wire.Frame{f})) + }) + + It("unpacks BLOCKED frames", func() { + f := &wire.BlockedFrame{StreamID: 0xDEADBEEF} + buf := &bytes.Buffer{} + err := f.Write(buf, protocol.VersionWhatever) + Expect(err).ToNot(HaveOccurred()) + setData(buf.Bytes()) + packet, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err).ToNot(HaveOccurred()) + Expect(packet.frames).To(Equal([]wire.Frame{f})) + }) + + It("errors on invalid frames", func() { + for b, e := range map[byte]qerr.ErrorCode{ + 0x04: qerr.InvalidWindowUpdateData, + 0x05: qerr.InvalidBlockedData, + } { + setData([]byte{b}) + _, err := unpacker.Unpack(hdrBin, hdr, data) + Expect(err.(*qerr.QuicError).ErrorCode).To(Equal(e)) + } + }) }) It("unpacks STOP_WAITING frames", func() { @@ -207,8 +281,6 @@ var _ = Describe("Packet unpacker", func() { 0x01: qerr.InvalidRstStreamData, 0x02: qerr.InvalidConnectionCloseData, 0x03: qerr.InvalidGoawayData, - 0x04: qerr.InvalidWindowUpdateData, - 0x05: qerr.InvalidBlockedData, 0x06: qerr.InvalidStopWaitingData, } { setData([]byte{b}) diff --git a/quic_suite_test.go b/quic_suite_test.go index 7059345f..840e1041 100644 --- a/quic_suite_test.go +++ b/quic_suite_test.go @@ -2,6 +2,7 @@ package quic import ( "github.com/golang/mock/gomock" + "github.com/lucas-clemente/quic-go/internal/protocol" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" @@ -13,8 +14,18 @@ func TestQuicGo(t *testing.T) { RunSpecs(t, "QUIC Suite") } +const ( + versionCryptoStream1 = protocol.Version39 + versionCryptoStream0 = protocol.VersionTLS +) + var mockCtrl *gomock.Controller +var _ = BeforeSuite(func() { + Expect(versionCryptoStream0.CryptoStreamID()).To(Equal(protocol.StreamID(0))) + Expect(versionCryptoStream1.CryptoStreamID()).To(Equal(protocol.StreamID(1))) +}) + var _ = BeforeEach(func() { mockCtrl = gomock.NewController(GinkgoT()) }) diff --git a/session.go b/session.go index 34c820a3..b4586f08 100644 --- a/session.go +++ b/session.go @@ -491,8 +491,10 @@ func (s *session) handleFrames(fs []wire.Frame) error { s.receivedPacketHandler.SetLowerLimit(frame.LeastUnacked - 1) case *wire.RstStreamFrame: err = s.handleRstStreamFrame(frame) - case *wire.WindowUpdateFrame: - err = s.handleWindowUpdateFrame(frame) + case *wire.MaxDataFrame: + s.handleMaxDataFrame(frame) + case *wire.MaxStreamDataFrame: + err = s.handleMaxStreamDataFrame(frame) case *wire.BlockedFrame: case *wire.PingFrame: default: @@ -542,12 +544,11 @@ func (s *session) handleStreamFrame(frame *wire.StreamFrame) error { return str.AddStreamFrame(frame) } -func (s *session) handleWindowUpdateFrame(frame *wire.WindowUpdateFrame) error { - if frame.StreamID == 0 { - s.connFlowController.UpdateSendWindow(frame.ByteOffset) - return nil - } +func (s *session) handleMaxDataFrame(frame *wire.MaxDataFrame) { + s.connFlowController.UpdateSendWindow(frame.ByteOffset) +} +func (s *session) handleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) error { str, err := s.streamsMap.GetOrOpenStream(frame.StreamID) if err != nil { return err @@ -646,11 +647,11 @@ func (s *session) processTransportParameters(params *handshake.TransportParamete func (s *session) sendPacket() error { s.packer.SetLeastUnacked(s.sentPacketHandler.GetLeastUnacked()) - // Get WindowUpdate frames + // Get MAX_DATA and MAX_STREAM_DATA frames // this call triggers the flow controller to increase the flow control windows, if necessary - windowUpdateFrames := s.getWindowUpdateFrames() - for _, wuf := range windowUpdateFrames { - s.packer.QueueControlFrame(wuf) + windowUpdates := s.getWindowUpdates() + for _, f := range windowUpdates { + s.packer.QueueControlFrame(f) } ack := s.receivedPacketHandler.GetAckFrame() @@ -732,10 +733,10 @@ func (s *session) sendPacket() error { } // send every window update twice - for _, f := range windowUpdateFrames { + for _, f := range windowUpdates { s.packer.QueueControlFrame(f) } - windowUpdateFrames = nil + windowUpdates = nil ack = nil } } @@ -872,19 +873,18 @@ func (s *session) tryDecryptingQueuedPackets() { s.undecryptablePackets = s.undecryptablePackets[:0] } -func (s *session) getWindowUpdateFrames() []*wire.WindowUpdateFrame { - var res []*wire.WindowUpdateFrame +func (s *session) getWindowUpdates() []wire.Frame { + var res []wire.Frame s.streamsMap.Range(func(str streamI) { if offset := str.GetWindowUpdate(); offset != 0 { - res = append(res, &wire.WindowUpdateFrame{ + res = append(res, &wire.MaxStreamDataFrame{ StreamID: str.StreamID(), ByteOffset: offset, }) } }) if offset := s.connFlowController.GetWindowUpdate(); offset != 0 { - res = append(res, &wire.WindowUpdateFrame{ - StreamID: 0, + res = append(res, &wire.MaxDataFrame{ ByteOffset: offset, }) } diff --git a/session_test.go b/session_test.go index 93bd39c8..cca47f97 100644 --- a/session_test.go +++ b/session_test.go @@ -385,7 +385,7 @@ var _ = Describe("Session", func() { }) }) - Context("handling WINDOW_UPDATE frames", func() { + Context("handling MAX_DATA and MAX_STREAM_DATA frames", func() { var connFC *mocks.MockConnectionFlowController BeforeEach(func() { @@ -398,7 +398,7 @@ var _ = Describe("Session", func() { str, err := sess.GetOrOpenStream(5) str.(*mocks.MockStreamI).EXPECT().UpdateSendWindow(offset) Expect(err).ToNot(HaveOccurred()) - err = sess.handleWindowUpdateFrame(&wire.WindowUpdateFrame{ + err = sess.handleMaxStreamDataFrame(&wire.MaxStreamDataFrame{ StreamID: 5, ByteOffset: offset, }) @@ -408,14 +408,10 @@ var _ = Describe("Session", func() { It("updates the flow control window of the connection", func() { offset := protocol.ByteCount(0x800000) connFC.EXPECT().UpdateSendWindow(offset) - err := sess.handleWindowUpdateFrame(&wire.WindowUpdateFrame{ - StreamID: 0, - ByteOffset: offset, - }) - Expect(err).ToNot(HaveOccurred()) + sess.handleMaxDataFrame(&wire.MaxDataFrame{ByteOffset: offset}) }) - It("opens a new stream when receiving a WINDOW_UPDATE for an unknown stream", func() { + It("opens a new stream when receiving a MAX_STREAM_DATA frame for an unknown stream", func() { newStreamLambda := sess.streamsMap.newStream sess.streamsMap.newStream = func(id protocol.StreamID) streamI { str := newStreamLambda(id) @@ -424,7 +420,7 @@ var _ = Describe("Session", func() { } return str } - err := sess.handleWindowUpdateFrame(&wire.WindowUpdateFrame{ + err := sess.handleMaxStreamDataFrame(&wire.MaxStreamDataFrame{ StreamID: 5, ByteOffset: 0x1337, }) @@ -434,7 +430,7 @@ var _ = Describe("Session", func() { Expect(str).ToNot(BeNil()) }) - It("ignores WINDOW_UPDATEs for a closed stream", func() { + It("ignores MAX_STREAM_DATA frames for a closed stream", func() { str, err := sess.GetOrOpenStream(3) Expect(err).ToNot(HaveOccurred()) str.(*mocks.MockStreamI).EXPECT().Finished().Return(true) @@ -443,7 +439,7 @@ var _ = Describe("Session", func() { str, err = sess.GetOrOpenStream(3) Expect(err).ToNot(HaveOccurred()) Expect(str).To(BeNil()) - err = sess.handleFrames([]wire.Frame{&wire.WindowUpdateFrame{ + err = sess.handleFrames([]wire.Frame{&wire.MaxStreamDataFrame{ StreamID: 3, ByteOffset: 1337, }}) @@ -813,7 +809,7 @@ var _ = Describe("Session", func() { Expect(sess.sentPacketHandler.(*mockSentPacketHandler).sentPackets[0].Frames).To(ContainElement(&wire.PingFrame{})) }) - It("sends two WINDOW_UPDATE frames", func() { + It("sends two MAX_STREAM_DATA frames", func() { mockFC := mocks.NewMockStreamFlowController(mockCtrl) mockFC.EXPECT().GetWindowUpdate().Return(protocol.ByteCount(0x1000)) mockFC.EXPECT().GetWindowUpdate().Return(protocol.ByteCount(0)).Times(2) @@ -827,7 +823,7 @@ var _ = Describe("Session", func() { err = sess.sendPacket() Expect(err).NotTo(HaveOccurred()) buf := &bytes.Buffer{} - (&wire.WindowUpdateFrame{ + (&wire.MaxStreamDataFrame{ StreamID: 5, ByteOffset: 0x1000, }).Write(buf, sess.version) diff --git a/streams_map_test.go b/streams_map_test.go index b4c83a39..14ec0e6c 100644 --- a/streams_map_test.go +++ b/streams_map_test.go @@ -14,11 +14,6 @@ import ( ) var _ = Describe("Streams Map", func() { - const ( - versionCryptoStream1 = protocol.Version39 - versionCryptoStream0 = protocol.VersionTLS - ) - var ( m *streamsMap finishedStreams map[protocol.StreamID]*gomock.Call @@ -37,8 +32,6 @@ var _ = Describe("Streams Map", func() { } BeforeEach(func() { - Expect(versionCryptoStream0.CryptoStreamID()).To(Equal(protocol.StreamID(0))) - Expect(versionCryptoStream1.CryptoStreamID()).To(Equal(protocol.StreamID(1))) finishedStreams = make(map[protocol.StreamID]*gomock.Call) })