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

Merge pull request #1112 from lucas-clemente/refactor-packet-sending

Browse Source 
Refactor packet sending
This commit is contained in:
Marten Seemann
2018-01-09 17:05:24 +07:00
committed by GitHub
parent af4c7c2faf 3e2faa45a6
commit 5cd5d5dae9
9 changed files with 454 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
ackhandler/ackhandler_suite_test.go
View File

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

7
ackhandler/sent_packet_handler.go
View File

@@ -365,12 +365,11 @@ func (h *sentPacketHandler) GetStopWaitingFrame(force bool) *wire.StopWaitingFra
}
func (h *sentPacketHandler) SendingAllowed() bool {
congestionLimited := h.bytesInFlight > h.congestion.GetCongestionWindow()
cwnd := h.congestion.GetCongestionWindow()
congestionLimited := h.bytesInFlight > cwnd
maxTrackedLimited := protocol.PacketNumber(len(h.retransmissionQueue)+h.packetHistory.Len()) >= protocol.MaxTrackedSentPackets
if congestionLimited {
utils.Debugf("Congestion limited: bytes in flight %d, window %d",
h.bytesInFlight,
h.congestion.GetCongestionWindow())
utils.Debugf("Congestion limited: bytes in flight %d, window %d", h.bytesInFlight, cwnd)
}
// Workaround for #555:
// Always allow sending of retransmissions. This should probably be limited

115
ackhandler/sent_packet_handler_test.go
View File

@@ -3,60 +3,15 @@ package ackhandler
import (
"time"
"github.com/golang/mock/gomock"
"github.com/lucas-clemente/quic-go/congestion"
"github.com/lucas-clemente/quic-go/internal/mocks"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
type mockCongestion struct {
argsOnPacketSent []interface{}
maybeExitSlowStart bool
onRetransmissionTimeout bool
getCongestionWindow bool
packetsAcked [][]interface{}
packetsLost [][]interface{}
}
func (m *mockCongestion) TimeUntilSend(now time.Time, bytesInFlight protocol.ByteCount) time.Duration {
panic("not implemented")
}
func (m *mockCongestion) OnPacketSent(sentTime time.Time, bytesInFlight protocol.ByteCount, packetNumber protocol.PacketNumber, bytes protocol.ByteCount, isRetransmittable bool) bool {
m.argsOnPacketSent = []interface{}{sentTime, bytesInFlight, packetNumber, bytes, isRetransmittable}
return false
}
func (m *mockCongestion) GetCongestionWindow() protocol.ByteCount {
m.getCongestionWindow = true
return protocol.DefaultTCPMSS
}
func (m *mockCongestion) MaybeExitSlowStart() {
m.maybeExitSlowStart = true
}
func (m *mockCongestion) OnRetransmissionTimeout(packetsRetransmitted bool) {
m.onRetransmissionTimeout = true
}
func (m *mockCongestion) RetransmissionDelay() time.Duration {
return defaultRTOTimeout
}
func (m *mockCongestion) SetNumEmulatedConnections(n int) { panic("not implemented") }
func (m *mockCongestion) OnConnectionMigration() { panic("not implemented") }
func (m *mockCongestion) SetSlowStartLargeReduction(enabled bool) { panic("not implemented") }
func (m *mockCongestion) OnPacketAcked(n protocol.PacketNumber, l protocol.ByteCount, bif protocol.ByteCount) {
m.packetsAcked = append(m.packetsAcked, []interface{}{n, l, bif})
}
func (m *mockCongestion) OnPacketLost(n protocol.PacketNumber, l protocol.ByteCount, bif protocol.ByteCount) {
m.packetsLost = append(m.packetsLost, []interface{}{n, l, bif})
}
func retransmittablePacket(num protocol.PacketNumber) *Packet {
return &Packet{
PacketNumber: num,
@@ -708,15 +663,23 @@ var _ = Describe("SentPacketHandler", func() {
Context("congestion", func() {
var (
cong *mockCongestion
cong *mocks.MockSendAlgorithm
)
BeforeEach(func() {
cong = &mockCongestion{}
cong = mocks.NewMockSendAlgorithm(mockCtrl)
cong.EXPECT().RetransmissionDelay().AnyTimes()
handler.congestion = cong
})
It("should call OnSent", func() {
cong.EXPECT().OnPacketSent(
gomock.Any(),
protocol.ByteCount(42),
protocol.PacketNumber(1),
protocol.ByteCount(42),
true,
)
p := &Packet{
PacketNumber: 1,
Length: 42,
@@ -724,62 +687,60 @@ var _ = Describe("SentPacketHandler", func() {
}
err := handler.SentPacket(p)
Expect(err).NotTo(HaveOccurred())
Expect(cong.argsOnPacketSent[1]).To(Equal(protocol.ByteCount(42)))
Expect(cong.argsOnPacketSent[2]).To(Equal(protocol.PacketNumber(1)))
Expect(cong.argsOnPacketSent[3]).To(Equal(protocol.ByteCount(42)))
Expect(cong.argsOnPacketSent[4]).To(BeTrue())
})
It("should call MaybeExitSlowStart and OnPacketAcked", func() {
cong.EXPECT().OnPacketSent(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(2)
cong.EXPECT().MaybeExitSlowStart()
cong.EXPECT().OnPacketAcked(
protocol.PacketNumber(1),
protocol.ByteCount(1),
protocol.ByteCount(1),
)
handler.SentPacket(retransmittablePacket(1))
handler.SentPacket(retransmittablePacket(2))
err := handler.ReceivedAck(&wire.AckFrame{LargestAcked: 1, LowestAcked: 1}, 1, protocol.EncryptionForwardSecure, time.Now())
Expect(err).NotTo(HaveOccurred())
Expect(cong.maybeExitSlowStart).To(BeTrue())
Expect(cong.packetsAcked).To(BeEquivalentTo([][]interface{}{
{protocol.PacketNumber(1), protocol.ByteCount(1), protocol.ByteCount(1)},
}))
Expect(cong.packetsLost).To(BeEmpty())
})
It("should call MaybeExitSlowStart and OnPacketLost", func() {
cong.EXPECT().OnPacketSent(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(3)
cong.EXPECT().OnRetransmissionTimeout(true).Times(2)
cong.EXPECT().OnPacketLost(
protocol.PacketNumber(1),
protocol.ByteCount(1),
protocol.ByteCount(2),
)
cong.EXPECT().OnPacketLost(
protocol.PacketNumber(2),
protocol.ByteCount(1),
protocol.ByteCount(1),
)
handler.SentPacket(retransmittablePacket(1))
handler.SentPacket(retransmittablePacket(2))
handler.SentPacket(retransmittablePacket(3))
handler.OnAlarm() // RTO, meaning 2 lost packets
Expect(cong.maybeExitSlowStart).To(BeFalse())
Expect(cong.onRetransmissionTimeout).To(BeTrue())
Expect(cong.packetsAcked).To(BeEmpty())
Expect(cong.packetsLost).To(BeEquivalentTo([][]interface{}{
{protocol.PacketNumber(1), protocol.ByteCount(1), protocol.ByteCount(2)},
{protocol.PacketNumber(2), protocol.ByteCount(1), protocol.ByteCount(1)},
}))
})
It("allows or denies sending based on congestion", func() {
Expect(handler.retransmissionQueue).To(BeEmpty())
handler.bytesInFlight = 100
cong.EXPECT().GetCongestionWindow().Return(protocol.MaxByteCount)
Expect(handler.SendingAllowed()).To(BeTrue())
err := handler.SentPacket(&Packet{
PacketNumber: 1,
Frames: []wire.Frame{&wire.PingFrame{}},
Length: protocol.DefaultTCPMSS + 1,
})
Expect(err).NotTo(HaveOccurred())
cong.EXPECT().GetCongestionWindow().Return(protocol.ByteCount(0))
Expect(handler.SendingAllowed()).To(BeFalse())
})
It("allows or denies sending based on the number of tracked packets", func() {
cong.EXPECT().GetCongestionWindow().Return(protocol.MaxByteCount).AnyTimes()
Expect(handler.SendingAllowed()).To(BeTrue())
handler.retransmissionQueue = make([]*Packet, protocol.MaxTrackedSentPackets)
Expect(handler.SendingAllowed()).To(BeFalse())
})
It("allows sending if there are retransmisisons outstanding", func() {
err := handler.SentPacket(&Packet{
PacketNumber: 1,
Frames: []wire.Frame{&wire.PingFrame{}},
Length: protocol.DefaultTCPMSS + 1,
})
Expect(err).NotTo(HaveOccurred())
handler.bytesInFlight = 100
cong.EXPECT().GetCongestionWindow().Return(protocol.ByteCount(0)).AnyTimes()
Expect(handler.SendingAllowed()).To(BeFalse())
handler.retransmissionQueue = []*Packet{nil}
Expect(handler.SendingAllowed()).To(BeTrue())

4
internal/mocks/received_packet_handler.go → internal/mocks/ackhandler/received_packet_handler.go
View File

@@ -1,8 +1,8 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/lucas-clemente/quic-go/ackhandler (interfaces: ReceivedPacketHandler)
// Package mocks is a generated GoMock package.
package mocks
// Package mockackhandler is a generated GoMock package.
package mockackhandler
import (
reflect "reflect"

4
internal/mocks/sent_packet_handler.go → internal/mocks/ackhandler/sent_packet_handler.go
View File

@@ -1,8 +1,8 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/lucas-clemente/quic-go/ackhandler (interfaces: SentPacketHandler)
// Package mocks is a generated GoMock package.
package mocks
// Package mockackhandler is a generated GoMock package.
package mockackhandler
import (
reflect "reflect"

154
internal/mocks/congestion.go Normal file
View File

@@ -0,0 +1,154 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/lucas-clemente/quic-go/congestion (interfaces: SendAlgorithm)
// Package mocks is a generated GoMock package.
package mocks
import (
reflect "reflect"
time "time"
gomock "github.com/golang/mock/gomock"
protocol "github.com/lucas-clemente/quic-go/internal/protocol"
)
// MockSendAlgorithm is a mock of SendAlgorithm interface
type MockSendAlgorithm struct {
ctrl *gomock.Controller
recorder *MockSendAlgorithmMockRecorder
}
// MockSendAlgorithmMockRecorder is the mock recorder for MockSendAlgorithm
type MockSendAlgorithmMockRecorder struct {
mock *MockSendAlgorithm
}
// NewMockSendAlgorithm creates a new mock instance
func NewMockSendAlgorithm(ctrl *gomock.Controller) *MockSendAlgorithm {
mock := &MockSendAlgorithm{ctrl: ctrl}
mock.recorder = &MockSendAlgorithmMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use
func (m *MockSendAlgorithm) EXPECT() *MockSendAlgorithmMockRecorder {
return m.recorder
}
// GetCongestionWindow mocks base method
func (m *MockSendAlgorithm) GetCongestionWindow() protocol.ByteCount {
ret := m.ctrl.Call(m, "GetCongestionWindow")
ret0, _ := ret[0].(protocol.ByteCount)
return ret0
}
// GetCongestionWindow indicates an expected call of GetCongestionWindow
func (mr *MockSendAlgorithmMockRecorder) GetCongestionWindow() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCongestionWindow", reflect.TypeOf((*MockSendAlgorithm)(nil).GetCongestionWindow))
}
// MaybeExitSlowStart mocks base method
func (m *MockSendAlgorithm) MaybeExitSlowStart() {
m.ctrl.Call(m, "MaybeExitSlowStart")
}
// MaybeExitSlowStart indicates an expected call of MaybeExitSlowStart
func (mr *MockSendAlgorithmMockRecorder) MaybeExitSlowStart() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MaybeExitSlowStart", reflect.TypeOf((*MockSendAlgorithm)(nil).MaybeExitSlowStart))
}
// OnConnectionMigration mocks base method
func (m *MockSendAlgorithm) OnConnectionMigration() {
m.ctrl.Call(m, "OnConnectionMigration")
}
// OnConnectionMigration indicates an expected call of OnConnectionMigration
func (mr *MockSendAlgorithmMockRecorder) OnConnectionMigration() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnConnectionMigration", reflect.TypeOf((*MockSendAlgorithm)(nil).OnConnectionMigration))
}
// OnPacketAcked mocks base method
func (m *MockSendAlgorithm) OnPacketAcked(arg0 protocol.PacketNumber, arg1, arg2 protocol.ByteCount) {
m.ctrl.Call(m, "OnPacketAcked", arg0, arg1, arg2)
}
// OnPacketAcked indicates an expected call of OnPacketAcked
func (mr *MockSendAlgorithmMockRecorder) OnPacketAcked(arg0, arg1, arg2 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPacketAcked", reflect.TypeOf((*MockSendAlgorithm)(nil).OnPacketAcked), arg0, arg1, arg2)
}
// OnPacketLost mocks base method
func (m *MockSendAlgorithm) OnPacketLost(arg0 protocol.PacketNumber, arg1, arg2 protocol.ByteCount) {
m.ctrl.Call(m, "OnPacketLost", arg0, arg1, arg2)
}
// OnPacketLost indicates an expected call of OnPacketLost
func (mr *MockSendAlgorithmMockRecorder) OnPacketLost(arg0, arg1, arg2 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPacketLost", reflect.TypeOf((*MockSendAlgorithm)(nil).OnPacketLost), arg0, arg1, arg2)
}
// OnPacketSent mocks base method
func (m *MockSendAlgorithm) OnPacketSent(arg0 time.Time, arg1 protocol.ByteCount, arg2 protocol.PacketNumber, arg3 protocol.ByteCount, arg4 bool) bool {
ret := m.ctrl.Call(m, "OnPacketSent", arg0, arg1, arg2, arg3, arg4)
ret0, _ := ret[0].(bool)
return ret0
}
// OnPacketSent indicates an expected call of OnPacketSent
func (mr *MockSendAlgorithmMockRecorder) OnPacketSent(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPacketSent", reflect.TypeOf((*MockSendAlgorithm)(nil).OnPacketSent), arg0, arg1, arg2, arg3, arg4)
}
// OnRetransmissionTimeout mocks base method
func (m *MockSendAlgorithm) OnRetransmissionTimeout(arg0 bool) {
m.ctrl.Call(m, "OnRetransmissionTimeout", arg0)
}
// OnRetransmissionTimeout indicates an expected call of OnRetransmissionTimeout
func (mr *MockSendAlgorithmMockRecorder) OnRetransmissionTimeout(arg0 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnRetransmissionTimeout", reflect.TypeOf((*MockSendAlgorithm)(nil).OnRetransmissionTimeout), arg0)
}
// RetransmissionDelay mocks base method
func (m *MockSendAlgorithm) RetransmissionDelay() time.Duration {
ret := m.ctrl.Call(m, "RetransmissionDelay")
ret0, _ := ret[0].(time.Duration)
return ret0
}
// RetransmissionDelay indicates an expected call of RetransmissionDelay
func (mr *MockSendAlgorithmMockRecorder) RetransmissionDelay() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RetransmissionDelay", reflect.TypeOf((*MockSendAlgorithm)(nil).RetransmissionDelay))
}
// SetNumEmulatedConnections mocks base method
func (m *MockSendAlgorithm) SetNumEmulatedConnections(arg0 int) {
m.ctrl.Call(m, "SetNumEmulatedConnections", arg0)
}
// SetNumEmulatedConnections indicates an expected call of SetNumEmulatedConnections
func (mr *MockSendAlgorithmMockRecorder) SetNumEmulatedConnections(arg0 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetNumEmulatedConnections", reflect.TypeOf((*MockSendAlgorithm)(nil).SetNumEmulatedConnections), arg0)
}
// SetSlowStartLargeReduction mocks base method
func (m *MockSendAlgorithm) SetSlowStartLargeReduction(arg0 bool) {
m.ctrl.Call(m, "SetSlowStartLargeReduction", arg0)
}
// SetSlowStartLargeReduction indicates an expected call of SetSlowStartLargeReduction
func (mr *MockSendAlgorithmMockRecorder) SetSlowStartLargeReduction(arg0 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetSlowStartLargeReduction", reflect.TypeOf((*MockSendAlgorithm)(nil).SetSlowStartLargeReduction), arg0)
}
// TimeUntilSend mocks base method
func (m *MockSendAlgorithm) TimeUntilSend(arg0 time.Time, arg1 protocol.ByteCount) time.Duration {
ret := m.ctrl.Call(m, "TimeUntilSend", arg0, arg1)
ret0, _ := ret[0].(time.Duration)
return ret0
}
// TimeUntilSend indicates an expected call of TimeUntilSend
func (mr *MockSendAlgorithmMockRecorder) TimeUntilSend(arg0, arg1 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TimeUntilSend", reflect.TypeOf((*MockSendAlgorithm)(nil).TimeUntilSend), arg0, arg1)
}

5
internal/mocks/gen.go
View File

@@ -3,8 +3,9 @@ package mocks
//go:generate sh -c "./mockgen_internal.sh mockhandshake handshake/mint_tls.go github.com/lucas-clemente/quic-go/internal/handshake MintTLS"
//go:generate sh -c "./mockgen_internal.sh mocks tls_extension_handler.go github.com/lucas-clemente/quic-go/internal/handshake TLSExtensionHandler"
//go:generate sh -c "./mockgen_internal.sh mocks stream_flow_controller.go github.com/lucas-clemente/quic-go/internal/flowcontrol StreamFlowController"
//go:generate sh -c "./mockgen_internal.sh mocks sent_packet_handler.go github.com/lucas-clemente/quic-go/ackhandler SentPacketHandler"
//go:generate sh -c "./mockgen_internal.sh mocks received_packet_handler.go github.com/lucas-clemente/quic-go/ackhandler ReceivedPacketHandler"
//go:generate sh -c "./mockgen_internal.sh mockackhandler ackhandler/sent_packet_handler.go github.com/lucas-clemente/quic-go/ackhandler SentPacketHandler"
//go:generate sh -c "./mockgen_internal.sh mockackhandler ackhandler/received_packet_handler.go github.com/lucas-clemente/quic-go/ackhandler ReceivedPacketHandler"
//go:generate sh -c "./mockgen_internal.sh mocks congestion.go github.com/lucas-clemente/quic-go/congestion SendAlgorithm"
//go:generate sh -c "./mockgen_internal.sh mocks connection_flow_controller.go github.com/lucas-clemente/quic-go/internal/flowcontrol ConnectionFlowController"
//go:generate sh -c "./mockgen_internal.sh mockcrypto crypto/aead.go github.com/lucas-clemente/quic-go/internal/crypto AEAD"
//go:generate sh -c "goimports -w ."

156
session.go
View File

@@ -396,9 +396,26 @@ runLoop:
s.keepAlivePingSent = true
}
if err := s.sendPacket(); err != nil {
s.closeLocal(err)
sendingAllowed := s.sentPacketHandler.SendingAllowed()
if !sendingAllowed { // if congestion limited, at least try sending an ACK frame
if err := s.maybeSendAckOnlyPacket(); err != nil {
s.closeLocal(err)
}
} else {
// repeatedly try sending until we don't have any more data, or run out of the congestion window
for sendingAllowed {
sentPacket, err := s.sendPacket()
if err != nil {
s.closeLocal(err)
break
}
if !sentPacket {
break
}
sendingAllowed = s.sentPacketHandler.SendingAllowed()
}
}
if !s.receivedTooManyUndecrytablePacketsTime.IsZero() && s.receivedTooManyUndecrytablePacketsTime.Add(protocol.PublicResetTimeout).Before(now) && len(s.undecryptablePackets) != 0 {
s.closeLocal(qerr.Error(qerr.DecryptionFailure, "too many undecryptable packets received"))
}
@@ -697,7 +714,26 @@ func (s *session) processTransportParameters(params *handshake.TransportParamete
// so we don't need to update stream flow control windows
}
func (s *session) sendPacket() error {
func (s *session) maybeSendAckOnlyPacket() error {
ack := s.receivedPacketHandler.GetAckFrame()
if ack == nil {
return nil
}
s.packer.QueueControlFrame(ack)
if !s.version.UsesIETFFrameFormat() { // for gQUIC, maybe add a STOP_WAITING
if swf := s.sentPacketHandler.GetStopWaitingFrame(false); swf != nil {
s.packer.QueueControlFrame(swf)
}
}
packet, err := s.packer.PackAckPacket()
if err != nil {
return err
}
return s.sendPackedPacket(packet)
}
func (s *session) sendPacket() (bool, error) {
s.packer.SetLeastUnacked(s.sentPacketHandler.GetLeastUnacked())
if offset := s.connFlowController.GetWindowUpdate(); offset != 0 {
@@ -713,83 +749,65 @@ func (s *session) sendPacket() error {
s.packer.QueueControlFrame(ack)
}
// Repeatedly try sending until we don't have any more data, or run out of the congestion window
// check for retransmissions first
for {
if !s.sentPacketHandler.SendingAllowed() {
if ack == nil {
return nil
retransmitPacket := s.sentPacketHandler.DequeuePacketForRetransmission()
if retransmitPacket == nil {
break
}
// retransmit handshake packets
if retransmitPacket.EncryptionLevel != protocol.EncryptionForwardSecure {
if s.handshakeComplete {
// don't retransmit handshake packets when the handshake is complete
continue
}
// If we aren't allowed to send, at least try sending an ACK frame
utils.Debugf("\tDequeueing handshake retransmission for packet 0x%x", retransmitPacket.PacketNumber)
if !s.version.UsesIETFFrameFormat() {
if swf := s.sentPacketHandler.GetStopWaitingFrame(false); swf != nil {
s.packer.QueueControlFrame(swf)
}
s.packer.QueueControlFrame(s.sentPacketHandler.GetStopWaitingFrame(true))
}
packet, err := s.packer.PackAckPacket()
packet, err := s.packer.PackHandshakeRetransmission(retransmitPacket)
if err != nil {
return err
return false, err
}
return s.sendPackedPacket(packet)
if err := s.sendPackedPacket(packet); err != nil {
return false, err
}
return true, nil
}
// check for retransmissions first
for {
retransmitPacket := s.sentPacketHandler.DequeuePacketForRetransmission()
if retransmitPacket == nil {
break
}
if retransmitPacket.EncryptionLevel != protocol.EncryptionForwardSecure {
if s.handshakeComplete {
// Don't retransmit handshake packets when the handshake is complete
continue
}
utils.Debugf("\tDequeueing handshake retransmission for packet 0x%x", retransmitPacket.PacketNumber)
if !s.version.UsesIETFFrameFormat() {
s.packer.QueueControlFrame(s.sentPacketHandler.GetStopWaitingFrame(true))
}
packet, err := s.packer.PackHandshakeRetransmission(retransmitPacket)
if err != nil {
return err
}
if err = s.sendPackedPacket(packet); err != nil {
return err
}
} else {
utils.Debugf("\tDequeueing retransmission for packet 0x%x", retransmitPacket.PacketNumber)
// resend the frames that were in the packet
for _, frame := range retransmitPacket.GetFramesForRetransmission() {
// TODO: only retransmit WINDOW_UPDATEs if they actually enlarge the window
switch f := frame.(type) {
case *wire.StreamFrame:
s.streamFramer.AddFrameForRetransmission(f)
default:
s.packer.QueueControlFrame(frame)
}
}
// queue all retransmittable frames sent in forward-secure packets
utils.Debugf("\tDequeueing retransmission for packet 0x%x", retransmitPacket.PacketNumber)
// resend the frames that were in the packet
for _, frame := range retransmitPacket.GetFramesForRetransmission() {
// TODO: only retransmit WINDOW_UPDATEs if they actually enlarge the window
switch f := frame.(type) {
case *wire.StreamFrame:
s.streamFramer.AddFrameForRetransmission(f)
default:
s.packer.QueueControlFrame(frame)
}
}
hasRetransmission := s.streamFramer.HasFramesForRetransmission()
if !s.version.UsesIETFFrameFormat() && (ack != nil || hasRetransmission) {
if swf := s.sentPacketHandler.GetStopWaitingFrame(hasRetransmission); swf != nil {
s.packer.QueueControlFrame(swf)
}
}
// add a retransmittable frame
if s.sentPacketHandler.ShouldSendRetransmittablePacket() {
s.packer.MakeNextPacketRetransmittable()
}
packet, err := s.packer.PackPacket()
if err != nil || packet == nil {
return err
}
if err = s.sendPackedPacket(packet); err != nil {
return err
}
ack = nil
}
hasRetransmission := s.streamFramer.HasFramesForRetransmission()
if !s.version.UsesIETFFrameFormat() && (ack != nil || hasRetransmission) {
if swf := s.sentPacketHandler.GetStopWaitingFrame(hasRetransmission); swf != nil {
s.packer.QueueControlFrame(swf)
}
}
// add a retransmittable frame
if s.sentPacketHandler.ShouldSendRetransmittablePacket() {
s.packer.MakeNextPacketRetransmittable()
}
packet, err := s.packer.PackPacket()
if err != nil || packet == nil {
return false, err
}
if err := s.sendPackedPacket(packet); err != nil {
return false, err
}
return true, nil
}
func (s *session) sendPackedPacket(packet *packedPacket) error {

244
session_test.go
View File

@@ -19,6 +19,7 @@ import (
"github.com/lucas-clemente/quic-go/internal/crypto"
"github.com/lucas-clemente/quic-go/internal/handshake"
"github.com/lucas-clemente/quic-go/internal/mocks"
"github.com/lucas-clemente/quic-go/internal/mocks/ackhandler"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/testdata"
"github.com/lucas-clemente/quic-go/internal/wire"
@@ -274,7 +275,7 @@ var _ = Describe("Session", func() {
Context("handling ACK frames", func() {
It("informs the SentPacketHandler about ACKs", func() {
f := &wire.AckFrame{LargestAcked: 3, LowestAcked: 2}
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().ReceivedAck(f, protocol.PacketNumber(42), protocol.EncryptionSecure, gomock.Any())
sph.EXPECT().GetLowestPacketNotConfirmedAcked()
sess.sentPacketHandler = sph
@@ -284,11 +285,11 @@ var _ = Describe("Session", func() {
})
It("tells the ReceivedPacketHandler to ignore low ranges", func() {
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().ReceivedAck(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any())
sph.EXPECT().GetLowestPacketNotConfirmedAcked().Return(protocol.PacketNumber(0x42))
sess.sentPacketHandler = sph
rph := mocks.NewMockReceivedPacketHandler(mockCtrl)
rph := mockackhandler.NewMockReceivedPacketHandler(mockCtrl)
rph.EXPECT().IgnoreBelow(protocol.PacketNumber(0x42))
sess.receivedPacketHandler = rph
err := sess.handleAckFrame(&wire.AckFrame{LargestAcked: 3, LowestAcked: 2}, protocol.EncryptionUnencrypted)
@@ -738,60 +739,28 @@ var _ = Describe("Session", func() {
packetNumber := protocol.PacketNumber(0x035e)
err := sess.receivedPacketHandler.ReceivedPacket(packetNumber, true)
Expect(err).ToNot(HaveOccurred())
err = sess.sendPacket()
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
Expect(mconn.written).To(Receive(ContainSubstring(string([]byte{0x03, 0x5e}))))
})
It("sends ACK frames when congestion limited", func() {
swf := &wire.StopWaitingFrame{LeastUnacked: 10}
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(false)
sph.EXPECT().GetStopWaitingFrame(false).Return(swf)
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(HaveLen(2))
Expect(p.Frames[0]).To(BeAssignableToTypeOf(&wire.AckFrame{}))
Expect(p.Frames[1]).To(Equal(swf))
})
sess.sentPacketHandler = sph
sess.packer.packetNumberGenerator.next = 0x1338
sess.receivedPacketHandler.ReceivedPacket(1, true)
err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(mconn.written).To(HaveLen(1))
})
It("doesn't include a STOP_WAITING for an ACK-only packet for IETF QUIC", func() {
sess.version = versionIETFFrames
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(false)
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(HaveLen(1))
Expect(p.Frames[0]).To(BeAssignableToTypeOf(&wire.AckFrame{}))
})
sess.sentPacketHandler = sph
sess.packer.packetNumberGenerator.next = 0x1338
sess.receivedPacketHandler.ReceivedPacket(1, true)
err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(mconn.written).To(HaveLen(1))
})
It("sends a retransmittable packet when required by the SentPacketHandler", func() {
sess.packer.QueueControlFrame(&wire.AckFrame{LargestAcked: 1000})
sph := mocks.NewMockSentPacketHandler(mockCtrl)
ack := &wire.AckFrame{LargestAcked: 1000}
sess.packer.QueueControlFrame(ack)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(true)
sph.EXPECT().SendingAllowed().Return(false)
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket().Return(true)
sph.EXPECT().SentPacket(gomock.Any())
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(HaveLen(2))
Expect(p.Frames).To(ContainElement(ack))
})
sess.sentPacketHandler = sph
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
})
@@ -800,10 +769,8 @@ var _ = Describe("Session", func() {
fc.EXPECT().GetWindowUpdate().Return(protocol.ByteCount(0x1337))
fc.EXPECT().IsNewlyBlocked()
sess.connFlowController = fc
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(true)
sph.EXPECT().SendingAllowed()
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
@@ -812,8 +779,9 @@ var _ = Describe("Session", func() {
}))
})
sess.sentPacketHandler = sph
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
})
It("adds MAX_STREAM_DATA frames", func() {
@@ -821,18 +789,17 @@ var _ = Describe("Session", func() {
StreamID: 2,
ByteOffset: 20,
})
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(true)
sph.EXPECT().SendingAllowed()
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(ContainElement(&wire.MaxStreamDataFrame{StreamID: 2, ByteOffset: 20}))
})
sess.sentPacketHandler = sph
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
})
It("adds a BLOCKED frame when it is connection-level flow control blocked", func() {
@@ -840,10 +807,8 @@ var _ = Describe("Session", func() {
fc.EXPECT().GetWindowUpdate()
fc.EXPECT().IsNewlyBlocked().Return(true, protocol.ByteCount(1337))
sess.connFlowController = fc
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(true)
sph.EXPECT().SendingAllowed()
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
@@ -852,8 +817,36 @@ var _ = Describe("Session", func() {
}))
})
sess.sentPacketHandler = sph
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
})
It("sends multiple packets", func() {
sess.queueControlFrame(&wire.MaxDataFrame{ByteOffset: 1})
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().DequeuePacketForRetransmission().Times(2)
sph.EXPECT().GetAlarmTimeout().AnyTimes()
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().ShouldSendRetransmittablePacket().Times(2)
sph.EXPECT().SentPacket(gomock.Any()).Times(2)
sph.EXPECT().SendingAllowed().Do(func() { // after sending the first packet
// make sure there's something to send
sess.packer.QueueControlFrame(&wire.MaxDataFrame{ByteOffset: 2})
}).Return(true).Times(2) // allow 2 packets...
sph.EXPECT().SendingAllowed() // ...then report that we're congestion limited
sess.sentPacketHandler = sph
done := make(chan struct{})
go func() {
defer GinkgoRecover()
sess.run()
close(done)
}()
sess.scheduleSending()
Eventually(mconn.written).Should(HaveLen(2))
// make the go routine return
sess.Close(nil)
Eventually(done).Should(BeClosed())
})
It("sends public reset", func() {
@@ -869,11 +862,9 @@ var _ = Describe("Session", func() {
Data: []byte("foobar"),
}
var sentPacket *ackhandler.Packet
sph := mocks.NewMockSentPacketHandler(mockCtrl)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().GetStopWaitingFrame(gomock.Any())
sph.EXPECT().SendingAllowed().Return(true)
sph.EXPECT().SendingAllowed().Return(false)
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
@@ -886,8 +877,9 @@ var _ = Describe("Session", func() {
sess.streamFramer.AddFrameForRetransmission(f)
_, err := sess.GetOrOpenStream(5)
Expect(err).ToNot(HaveOccurred())
err = sess.sendPacket()
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
Expect(sentPacket.PacketNumber).To(Equal(protocol.PacketNumber(0x1337 + 9)))
Expect(sentPacket.Frames).To(ContainElement(f))
@@ -896,16 +888,78 @@ var _ = Describe("Session", func() {
})
})
Context("sending ACK only packets", func() {
It("doesn't do anything if there's no ACK to be sent", func() {
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sess.sentPacketHandler = sph
err := sess.maybeSendAckOnlyPacket()
Expect(err).ToNot(HaveOccurred())
Expect(mconn.written).To(BeEmpty())
})
It("sends ACK only packets", func() {
swf := &wire.StopWaitingFrame{LeastUnacked: 10}
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked()
sph.EXPECT().GetAlarmTimeout().AnyTimes()
sph.EXPECT().SendingAllowed()
sph.EXPECT().GetStopWaitingFrame(false).Return(swf)
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(HaveLen(2))
Expect(p.Frames[0]).To(BeAssignableToTypeOf(&wire.AckFrame{}))
Expect(p.Frames[1]).To(Equal(swf))
})
sess.sentPacketHandler = sph
sess.packer.packetNumberGenerator.next = 0x1338
sess.receivedPacketHandler.ReceivedPacket(1, true)
done := make(chan struct{})
go func() {
defer GinkgoRecover()
sess.run()
close(done)
}()
sess.scheduleSending()
Eventually(mconn.written).Should(HaveLen(1))
// make sure that the go routine returns
sess.Close(nil)
Eventually(done).Should(BeClosed())
})
It("doesn't include a STOP_WAITING for an ACK-only packet for IETF QUIC", func() {
sess.version = versionIETFFrames
done := make(chan struct{})
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked()
sph.EXPECT().GetAlarmTimeout().AnyTimes()
sph.EXPECT().SendingAllowed()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(HaveLen(1))
Expect(p.Frames[0]).To(BeAssignableToTypeOf(&wire.AckFrame{}))
})
sess.sentPacketHandler = sph
sess.packer.packetNumberGenerator.next = 0x1338
sess.receivedPacketHandler.ReceivedPacket(1, true)
go func() {
defer GinkgoRecover()
sess.run()
close(done)
}()
sess.scheduleSending()
Eventually(mconn.written).Should(HaveLen(1))
// make sure that the go routine returns
sess.Close(nil)
Eventually(done).Should(BeClosed())
})
})
Context("retransmissions", func() {
var sph *mocks.MockSentPacketHandler
var sph *mockackhandler.MockSentPacketHandler
BeforeEach(func() {
// a STOP_WAITING frame is added, so make sure the packet number of the new package is higher than the packet number of the retransmitted packet
sess.packer.packetNumberGenerator.next = 0x1337 + 10
sess.packer.hasSentPacket = true // make sure this is not the first packet the packer sends
sph = mocks.NewMockSentPacketHandler(mockCtrl)
sph = mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLeastUnacked().AnyTimes()
sph.EXPECT().SendingAllowed().Return(true)
sph.EXPECT().ShouldSendRetransmittablePacket()
sess.sentPacketHandler = sph
sess.packer.cryptoSetup = &mockCryptoSetup{encLevelSeal: protocol.EncryptionForwardSecure}
})
@@ -919,15 +973,14 @@ var _ = Describe("Session", func() {
Frames: []wire.Frame{sf},
EncryptionLevel: protocol.EncryptionUnencrypted,
})
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.EncryptionLevel).To(Equal(protocol.EncryptionUnencrypted))
Expect(p.Frames).To(Equal([]wire.Frame{swf, sf}))
})
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
})
It("retransmits an unencrypted packet, and doesn't add a STOP_WAITING frame (for IETF QUIC)", func() {
@@ -938,13 +991,13 @@ var _ = Describe("Session", func() {
Frames: []wire.Frame{sf},
EncryptionLevel: protocol.EncryptionUnencrypted,
})
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.EncryptionLevel).To(Equal(protocol.EncryptionUnencrypted))
Expect(p.Frames).To(Equal([]wire.Frame{sf}))
})
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
})
@@ -957,8 +1010,10 @@ var _ = Describe("Session", func() {
EncryptionLevel: protocol.EncryptionSecure,
})
sph.EXPECT().DequeuePacketForRetransmission()
err := sess.sendPacket()
Expect(err).ToNot(HaveOccurred())
sph.EXPECT().ShouldSendRetransmittablePacket()
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeFalse())
Expect(mconn.written).To(BeEmpty())
})
})
@@ -977,13 +1032,14 @@ var _ = Describe("Session", func() {
EncryptionLevel: protocol.EncryptionForwardSecure,
})
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(Equal([]wire.Frame{swf, f}))
Expect(p.EncryptionLevel).To(Equal(protocol.EncryptionForwardSecure))
})
sph.EXPECT().SendingAllowed()
err := sess.sendPacket()
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
})
@@ -999,17 +1055,18 @@ var _ = Describe("Session", func() {
EncryptionLevel: protocol.EncryptionForwardSecure,
})
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(Equal([]wire.Frame{f}))
Expect(p.EncryptionLevel).To(Equal(protocol.EncryptionForwardSecure))
})
sph.EXPECT().SendingAllowed()
err := sess.sendPacket()
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
})
It("sends a StreamFrame from a packet queued for retransmission", func() {
It("sends a STREAM frame from a packet queued for retransmission", func() {
f1 := wire.StreamFrame{
StreamID: 0x5,
Data: []byte("foobar"),
@@ -1032,11 +1089,13 @@ var _ = Describe("Session", func() {
sph.EXPECT().DequeuePacketForRetransmission().Return(p2)
sph.EXPECT().DequeuePacketForRetransmission()
sph.EXPECT().GetStopWaitingFrame(true).Return(&wire.StopWaitingFrame{})
sph.EXPECT().SentPacket(gomock.Any())
sph.EXPECT().SendingAllowed()
err := sess.sendPacket()
sph.EXPECT().ShouldSendRetransmittablePacket()
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.Frames).To(HaveLen(3))
})
sent, err := sess.sendPacket()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Expect(mconn.written).To(HaveLen(1))
packet := <-mconn.written
Expect(packet).To(ContainSubstring("foobar"))
@@ -1096,18 +1155,23 @@ var _ = Describe("Session", func() {
})
It("sets the timer to the ack timer", func() {
rph := mocks.NewMockReceivedPacketHandler(mockCtrl)
rph := mockackhandler.NewMockReceivedPacketHandler(mockCtrl)
rph.EXPECT().GetAckFrame().Return(&wire.AckFrame{LargestAcked: 0x1337})
rph.EXPECT().GetAckFrame()
rph.EXPECT().GetAlarmTimeout().Return(time.Now().Add(10 * time.Millisecond)).MinTimes(1)
sess.receivedPacketHandler = rph
done := make(chan struct{})
go func() {
defer GinkgoRecover()
sess.run()
close(done)
}()
defer sess.Close(nil)
time.Sleep(10 * time.Millisecond)
Eventually(func() int { return len(mconn.written) }).ShouldNot(BeZero())
Expect(mconn.written).To(Receive(ContainSubstring(string([]byte{0x13, 0x37}))))
// make sure the go routine returns
sess.Close(nil)
Eventually(done).Should(BeClosed())
})
})