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

Merge pull request #2095 from lucas-clemente/frame-retransmisions

Browse source 
retransmit frames, not packets
This commit is contained in:
Marten Seemann 2019-08-31 17:34:10 +07:00 • committed by GitHub
commit dd6e8a7424
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 755 additions and 1214 deletions
Download patch file Download diff file Expand all files Collapse all files

8
internal/ackhandler/frame.go Normal file
View file

@ -0,0 +1,8 @@
package ackhandler
import "github.com/lucas-clemente/quic-go/internal/wire"
type Frame struct {
wire.Frame // nil if the frame has already been acknowledged in another packet
OnLost func(wire.Frame)
}

16
internal/ackhandler/interfaces.go
View file

@ -8,11 +8,22 @@ import (
"github.com/lucas-clemente/quic-go/quictrace"
)
// A Packet is a packet
type Packet struct {
PacketNumber protocol.PacketNumber
Frames []Frame
LargestAcked protocol.PacketNumber // InvalidPacketNumber if the packet doesn't contain an ACK
Length protocol.ByteCount
EncryptionLevel protocol.EncryptionLevel
SendTime time.Time
includedInBytesInFlight bool
}
// SentPacketHandler handles ACKs received for outgoing packets
type SentPacketHandler interface {
// SentPacket may modify the packet
SentPacket(packet *Packet)
SentPacketsAsRetransmission(packets []*Packet, retransmissionOf protocol.PacketNumber)
ReceivedAck(ackFrame *wire.AckFrame, withPacketNumber protocol.PacketNumber, encLevel protocol.EncryptionLevel, recvTime time.Time) error
DropPackets(protocol.EncryptionLevel)
ResetForRetry() error
@ -31,8 +42,7 @@ type SentPacketHandler interface {
// only to be called once the handshake is complete
GetLowestPacketNotConfirmedAcked() protocol.PacketNumber
DequeuePacketForRetransmission() *Packet
DequeueProbePacket() (*Packet, error)
QueueProbePacket() bool /* was a packet queued */
PeekPacketNumber(protocol.EncryptionLevel) (protocol.PacketNumber, protocol.PacketNumberLen)
PopPacketNumber(protocol.EncryptionLevel) protocol.PacketNumber

27
internal/ackhandler/packet.go
View file

@ -1,27 +0,0 @@
package ackhandler
import (
"time"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
)
// A Packet is a packet
type Packet struct {
PacketNumber protocol.PacketNumber
Frames []wire.Frame
LargestAcked protocol.PacketNumber // InvalidPacketNumber if the packet doesn't contain an ACK
Length protocol.ByteCount
EncryptionLevel protocol.EncryptionLevel
SendTime time.Time
// There are two reasons why a packet cannot be retransmitted:
// * it was already retransmitted
// * this packet is a retransmission, and we already received an ACK for the original packet
canBeRetransmitted bool
includedInBytesInFlight bool
retransmittedAs []protocol.PacketNumber
isRetransmission bool // we need a separate bool here because 0 is a valid packet number
retransmissionOf protocol.PacketNumber
}

4
internal/ackhandler/send_mode.go
View file

@ -10,8 +10,6 @@ const (
SendNone SendMode = iota
// SendAck means an ACK-only packet should be sent
SendAck
// SendRetransmission means that retransmissions should be sent
SendRetransmission
// SendPTO means that a probe packet should be sent
SendPTO
// SendAny means that any packet should be sent
@ -24,8 +22,6 @@ func (s SendMode) String() string {
return "none"
case SendAck:
return "ack"
case SendRetransmission:
return "retransmission"
case SendPTO:
return "pto"
case SendAny:

1
internal/ackhandler/send_mode_test.go
View file

@ -11,7 +11,6 @@ var _ = Describe("Send Mode", func() {
Expect(SendAny.String()).To(Equal("any"))
Expect(SendAck.String()).To(Equal("ack"))
Expect(SendPTO.String()).To(Equal("pto"))
Expect(SendRetransmission.String()).To(Equal("retransmission"))
Expect(SendMode(123).String()).To(Equal("invalid send mode: 123"))
})
})

169
internal/ackhandler/sent_packet_handler.go
View file

@ -1,7 +1,6 @@
package ackhandler
import (
"errors"
"fmt"
"math"
"time"
@ -56,8 +55,6 @@ type sentPacketHandler struct {
// Only applies to the application-data packet number space.
lowestNotConfirmedAcked protocol.PacketNumber
retransmissionQueue []*Packet
bytesInFlight protocol.ByteCount
congestion congestion.SendAlgorithmWithDebugInfos
@ -112,14 +109,6 @@ func (h *sentPacketHandler) DropPackets(encLevel protocol.EncryptionLevel) {
}
return true, nil
})
// remove packets from the retransmission queue
var queue []*Packet
for _, packet := range h.retransmissionQueue {
if packet.EncryptionLevel != encLevel {
queue = append(queue, packet)
}
}
h.retransmissionQueue = queue
// drop the packet history
switch encLevel {
case protocol.EncryptionInitial:
@ -138,17 +127,6 @@ func (h *sentPacketHandler) SentPacket(packet *Packet) {
}
}
func (h *sentPacketHandler) SentPacketsAsRetransmission(packets []*Packet, retransmissionOf protocol.PacketNumber) {
var p []*Packet
for _, packet := range packets {
if isAckEliciting := h.sentPacketImpl(packet); isAckEliciting {
p = append(p, packet)
}
}
h.getPacketNumberSpace(p[0].EncryptionLevel).history.SentPacketsAsRetransmission(p, retransmissionOf)
h.setLossDetectionTimer()
}
func (h *sentPacketHandler) getPacketNumberSpace(encLevel protocol.EncryptionLevel) *packetNumberSpace {
switch encLevel {
case protocol.EncryptionInitial:
@ -181,7 +159,6 @@ func (h *sentPacketHandler) sentPacketImpl(packet *Packet) bool /* is ack-elicit
h.lastSentAckElicitingPacketTime = packet.SendTime
packet.includedInBytesInFlight = true
h.bytesInFlight += packet.Length
packet.canBeRetransmitted = true
if h.numProbesToSend > 0 {
h.numProbesToSend--
}
@ -346,12 +323,14 @@ func (h *sentPacketHandler) setLossDetectionTimer() {
// Cancel the alarm if no packets are outstanding
if !h.hasOutstandingPackets() {
h.logger.Debugf("setLossDetectionTimer: canceling. Bytes in flight: %d", h.bytesInFlight)
h.alarm = time.Time{}
return
}
// PTO alarm
h.alarm = h.lastSentAckElicitingPacketTime.Add(h.rttStats.PTO() << h.ptoCount)
h.logger.Debugf("setLossDetectionTimer: setting to", h.alarm)
}
func (h *sentPacketHandler) detectLostPackets(
@ -399,27 +378,23 @@ func (h *sentPacketHandler) detectLostPackets(
}
for _, p := range lostPackets {
h.queueFramesForRetransmission(p)
// the bytes in flight need to be reduced no matter if this packet will be retransmitted
if p.includedInBytesInFlight {
h.bytesInFlight -= p.Length
h.congestion.OnPacketLost(p.PacketNumber, p.Length, priorInFlight)
}
if p.canBeRetransmitted {
// queue the packet for retransmission, and report the loss to the congestion controller
if err := h.queuePacketForRetransmission(p, pnSpace); err != nil {
return err
}
}
pnSpace.history.Remove(p.PacketNumber)
if h.traceCallback != nil {
// TODO: trace frames
h.traceCallback(quictrace.Event{
Time: now,
EventType: quictrace.PacketLost,
EncryptionLevel: p.EncryptionLevel,
PacketNumber: p.PacketNumber,
PacketSize: p.Length,
Frames: p.Frames,
TransportState: h.GetStats(),
//Frames: p.Frames,
TransportState: h.GetStats(),
})
}
}
@ -465,97 +440,16 @@ func (h *sentPacketHandler) GetLossDetectionTimeout() time.Time {
func (h *sentPacketHandler) onPacketAcked(p *Packet, rcvTime time.Time) error {
pnSpace := h.getPacketNumberSpace(p.EncryptionLevel)
// This happens if a packet and its retransmissions is acked in the same ACK.
// As soon as we process the first one, this will remove all the retransmissions,
// so we won't find the retransmitted packet number later.
if packet := pnSpace.history.GetPacket(p.PacketNumber); packet == nil {
return nil
}
// only report the acking of this packet to the congestion controller if:
// * it is an ack-eliciting packet
// * this packet wasn't retransmitted yet
if p.isRetransmission {
// that the parent doesn't exist is expected to happen every time the original packet was already acked
if parent := pnSpace.history.GetPacket(p.retransmissionOf); parent != nil {
if len(parent.retransmittedAs) == 1 {
parent.retransmittedAs = nil
} else {
// remove this packet from the slice of retransmission
retransmittedAs := make([]protocol.PacketNumber, 0, len(parent.retransmittedAs)-1)
for _, pn := range parent.retransmittedAs {
if pn != p.PacketNumber {
retransmittedAs = append(retransmittedAs, pn)
}
}
parent.retransmittedAs = retransmittedAs
}
}
}
// this also applies to packets that have been retransmitted as probe packets
if p.includedInBytesInFlight {
h.bytesInFlight -= p.Length
}
if err := h.stopRetransmissionsFor(p, pnSpace); err != nil {
return err
}
return pnSpace.history.Remove(p.PacketNumber)
}
func (h *sentPacketHandler) stopRetransmissionsFor(p *Packet, pnSpace *packetNumberSpace) error {
if err := pnSpace.history.MarkCannotBeRetransmitted(p.PacketNumber); err != nil {
return err
}
for _, r := range p.retransmittedAs {
packet := pnSpace.history.GetPacket(r)
if packet == nil {
return fmt.Errorf("sent packet handler BUG: marking packet as not retransmittable %d (retransmission of %d) not found in history", r, p.PacketNumber)
}
h.stopRetransmissionsFor(packet, pnSpace)
}
return nil
}
func (h *sentPacketHandler) DequeuePacketForRetransmission() *Packet {
if len(h.retransmissionQueue) == 0 {
return nil
}
packet := h.retransmissionQueue[0]
// Shift the slice and don't retain anything that isn't needed.
copy(h.retransmissionQueue, h.retransmissionQueue[1:])
h.retransmissionQueue[len(h.retransmissionQueue)-1] = nil
h.retransmissionQueue = h.retransmissionQueue[:len(h.retransmissionQueue)-1]
return packet
}
func (h *sentPacketHandler) DequeueProbePacket() (*Packet, error) {
if len(h.retransmissionQueue) > 0 {
return h.DequeuePacketForRetransmission(), nil
}
var pnSpace *packetNumberSpace
var p *Packet
if h.initialPackets != nil {
pnSpace = h.initialPackets
p = h.initialPackets.history.FirstOutstanding()
}
if p == nil && h.handshakePackets != nil {
pnSpace = h.handshakePackets
p = h.handshakePackets.history.FirstOutstanding()
}
if p == nil {
pnSpace = h.oneRTTPackets
p = h.oneRTTPackets.history.FirstOutstanding()
}
if p == nil {
return nil, errors.New("cannot dequeue a probe packet. No outstanding packets")
}
if err := h.queuePacketForRetransmission(p, pnSpace); err != nil {
return nil, err
}
return h.DequeuePacketForRetransmission(), nil
}
func (h *sentPacketHandler) PeekPacketNumber(encLevel protocol.EncryptionLevel) (protocol.PacketNumber, protocol.PacketNumberLen) {
pnSpace := h.getPacketNumberSpace(encLevel)
@ -575,7 +469,7 @@ func (h *sentPacketHandler) PopPacketNumber(encLevel protocol.EncryptionLevel) p
}
func (h *sentPacketHandler) SendMode() SendMode {
numTrackedPackets := len(h.retransmissionQueue) + h.oneRTTPackets.history.Len()
numTrackedPackets := h.oneRTTPackets.history.Len()
if h.initialPackets != nil {
numTrackedPackets += h.initialPackets.history.Len()
}
@ -603,10 +497,6 @@ func (h *sentPacketHandler) SendMode() SendMode {
}
return SendAck
}
// Send retransmissions first, if there are any.
if len(h.retransmissionQueue) > 0 {
return SendRetransmission
}
if numTrackedPackets >= protocol.MaxOutstandingSentPackets {
if h.logger.Debug() {
h.logger.Debugf("Max outstanding limited: tracking %d packets, maximum: %d", numTrackedPackets, protocol.MaxOutstandingSentPackets)
@ -632,30 +522,45 @@ func (h *sentPacketHandler) ShouldSendNumPackets() int {
return int(math.Ceil(float64(protocol.MinPacingDelay) / float64(delay)))
}
func (h *sentPacketHandler) queuePacketForRetransmission(p *Packet, pnSpace *packetNumberSpace) error {
if !p.canBeRetransmitted {
return fmt.Errorf("sent packet handler BUG: packet %d already queued for retransmission", p.PacketNumber)
func (h *sentPacketHandler) QueueProbePacket() bool {
var p *Packet
if h.initialPackets != nil {
p = h.initialPackets.history.FirstOutstanding()
}
if err := pnSpace.history.MarkCannotBeRetransmitted(p.PacketNumber); err != nil {
return err
if p == nil && h.handshakePackets != nil {
p = h.handshakePackets.history.FirstOutstanding()
}
if p == nil {
p = h.oneRTTPackets.history.FirstOutstanding()
}
if p == nil {
return false
}
h.queueFramesForRetransmission(p)
// TODO: don't remove the packet here
// Keep track of acknowledged frames instead.
if p.includedInBytesInFlight {
h.bytesInFlight -= p.Length
}
if err := h.getPacketNumberSpace(p.EncryptionLevel).history.Remove(p.PacketNumber); err != nil {
// should never happen. We just got this packet from the history a lines above.
panic(err)
}
return true
}
func (h *sentPacketHandler) queueFramesForRetransmission(p *Packet) {
for _, f := range p.Frames {
f.OnLost(f.Frame)
}
h.retransmissionQueue = append(h.retransmissionQueue, p)
return nil
}
func (h *sentPacketHandler) ResetForRetry() error {
h.bytesInFlight = 0
var packets []*Packet
h.initialPackets.history.Iterate(func(p *Packet) (bool, error) {
if p.canBeRetransmitted {
packets = append(packets, p)
}
h.queueFramesForRetransmission(p)
return true, nil
})
for _, p := range packets {
h.logger.Debugf("Queueing packet %#x for retransmission.", p.PacketNumber)
h.retransmissionQueue = append(h.retransmissionQueue, p)
}
h.initialPackets = newPacketNumberSpace(h.initialPackets.pns.Pop())
h.setLossDetectionTimer()
return nil

387
internal/ackhandler/sent_packet_handler_test.go
View file

@ -14,40 +14,15 @@ import (
. "github.com/onsi/gomega"
)
func ackElicitingPacket(p *Packet) *Packet {
if p.EncryptionLevel == protocol.EncryptionUnspecified {
p.EncryptionLevel = protocol.Encryption1RTT
}
if p.Length == 0 {
p.Length = 1
}
if p.SendTime.IsZero() {
p.SendTime = time.Now()
}
p.Frames = []wire.Frame{&wire.PingFrame{}}
return p
}
func nonAckElicitingPacket(p *Packet) *Packet {
p = ackElicitingPacket(p)
p.Frames = nil
p.LargestAcked = 1
return p
}
func cryptoPacket(p *Packet) *Packet {
p = ackElicitingPacket(p)
p.EncryptionLevel = protocol.EncryptionInitial
return p
}
var _ = Describe("SentPacketHandler", func() {
var (
handler *sentPacketHandler
streamFrame wire.StreamFrame
lostPackets []protocol.PacketNumber
)
BeforeEach(func() {
lostPackets = nil
rttStats := &congestion.RTTStats{}
handler = NewSentPacketHandler(42, rttStats, nil, utils.DefaultLogger).(*sentPacketHandler)
streamFrame = wire.StreamFrame{
@ -63,17 +38,33 @@ var _ = Describe("SentPacketHandler", func() {
return nil
}
losePacket := func(pn protocol.PacketNumber, encLevel protocol.EncryptionLevel) {
p := getPacket(pn, encLevel)
ExpectWithOffset(1, p).ToNot(BeNil())
handler.queuePacketForRetransmission(p, handler.getPacketNumberSpace(encLevel))
if p.includedInBytesInFlight {
p.includedInBytesInFlight = false
handler.bytesInFlight -= p.Length
ackElicitingPacket := func(p *Packet) *Packet {
if p.EncryptionLevel == protocol.EncryptionUnspecified {
p.EncryptionLevel = protocol.Encryption1RTT
}
r := handler.DequeuePacketForRetransmission()
ExpectWithOffset(1, r).ToNot(BeNil())
ExpectWithOffset(1, r.PacketNumber).To(Equal(pn))
if p.Length == 0 {
p.Length = 1
}
if p.SendTime.IsZero() {
p.SendTime = time.Now()
}
p.Frames = []Frame{
{Frame: &wire.PingFrame{}, OnLost: func(wire.Frame) { lostPackets = append(lostPackets, p.PacketNumber) }},
}
return p
}
nonAckElicitingPacket := func(p *Packet) *Packet {
p = ackElicitingPacket(p)
p.Frames = nil
p.LargestAcked = 1
return p
}
cryptoPacket := func(p *Packet) *Packet {
p = ackElicitingPacket(p)
p.EncryptionLevel = protocol.EncryptionInitial
return p
}
expectInPacketHistory := func(expected []protocol.PacketNumber, encLevel protocol.EncryptionLevel) {
@ -174,21 +165,21 @@ var _ = Describe("SentPacketHandler", func() {
})
})
Context("acks and nacks the right packets", func() {
Context("acks the right packets", func() {
expectInPacketHistoryOrLost := func(expected []protocol.PacketNumber, encLevel protocol.EncryptionLevel) {
pnSpace := handler.getPacketNumberSpace(encLevel)
ExpectWithOffset(1, pnSpace.history.Len()+len(handler.retransmissionQueue)).To(Equal(len(expected)))
ExpectWithOffset(1, pnSpace.history.Len()+len(lostPackets)).To(Equal(len(expected)))
expectedLoop:
for _, p := range expected {
if _, ok := pnSpace.history.packetMap[p]; ok {
continue
}
for _, lost := range handler.retransmissionQueue {
if lost.PacketNumber == p {
for _, lostP := range lostPackets {
if lostP == p {
continue expectedLoop
}
}
Fail(fmt.Sprintf("Packet %d neither in packet history nor declared lost.", p))
Fail(fmt.Sprintf("Packet %d not in packet history.", p))
}
}
@ -333,9 +324,26 @@ var _ = Describe("SentPacketHandler", func() {
Context("determining which ACKs we have received an ACK for", func() {
BeforeEach(func() {
morePackets := []*Packet{
{PacketNumber: 13, LargestAcked: 100, Frames: []wire.Frame{&streamFrame}, Length: 1, EncryptionLevel: protocol.Encryption1RTT},
{PacketNumber: 14, LargestAcked: 200, Frames: []wire.Frame{&streamFrame}, Length: 1, EncryptionLevel: protocol.Encryption1RTT},
{PacketNumber: 15, Frames: []wire.Frame{&streamFrame}, Length: 1, EncryptionLevel: protocol.Encryption1RTT},
{
PacketNumber: 13,
LargestAcked: 100,
Frames: []Frame{{Frame: &streamFrame, OnLost: func(wire.Frame) {}}},
Length: 1,
EncryptionLevel: protocol.Encryption1RTT,
},
{
PacketNumber: 14,
LargestAcked: 200,
Frames: []Frame{{Frame: &streamFrame, OnLost: func(wire.Frame) {}}},
Length: 1,
EncryptionLevel: protocol.Encryption1RTT,
},
{
PacketNumber: 15,
Frames: []Frame{{Frame: &streamFrame, OnLost: func(wire.Frame) {}}},
Length: 1,
EncryptionLevel: protocol.Encryption1RTT,
},
}
for _, packet := range morePackets {
handler.SentPacket(packet)
@ -368,55 +376,6 @@ var _ = Describe("SentPacketHandler", func() {
})
})
Context("ACK processing, for retransmitted packets", func() {
It("sends a packet as retransmission", func() {
// packet 5 was retransmitted as packet 6
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5, Length: 10, EncryptionLevel: protocol.Encryption1RTT}))
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(10)))
losePacket(5, protocol.Encryption1RTT)
Expect(handler.bytesInFlight).To(BeZero())
handler.SentPacketsAsRetransmission([]*Packet{ackElicitingPacket(&Packet{PacketNumber: 6, Length: 11})}, 5)
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(11)))
})
It("removes a packet when it is acked", func() {
// packet 5 was retransmitted as packet 6
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5, Length: 10}))
losePacket(5, protocol.Encryption1RTT)
handler.SentPacketsAsRetransmission([]*Packet{ackElicitingPacket(&Packet{PacketNumber: 6, Length: 11})}, 5)
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(11)))
// ack 5
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 5, Largest: 5}}}
err := handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())
Expect(err).ToNot(HaveOccurred())
expectInPacketHistory([]protocol.PacketNumber{6}, protocol.Encryption1RTT)
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(11)))
})
It("handles ACKs that ack the original packet as well as the retransmission", func() {
// packet 5 was retransmitted as packet 7
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5, Length: 10}))
losePacket(5, protocol.Encryption1RTT)
handler.SentPacketsAsRetransmission([]*Packet{ackElicitingPacket(&Packet{PacketNumber: 7, Length: 11})}, 5)
// ack 5 and 7
ack := &wire.AckFrame{
AckRanges: []wire.AckRange{
{Smallest: 7, Largest: 7},
{Smallest: 5, Largest: 5},
},
}
err := handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())
Expect(err).ToNot(HaveOccurred())
Expect(handler.oneRTTPackets.history.Len()).To(BeZero())
Expect(handler.bytesInFlight).To(BeZero())
})
})
It("does not dequeue a packet if no ACK has been received", func() {
handler.SentPacket(&Packet{PacketNumber: 1, EncryptionLevel: protocol.Encryption1RTT, SendTime: time.Now().Add(-time.Hour)})
Expect(handler.DequeuePacketForRetransmission()).To(BeNil())
})
Context("congestion", func() {
var cong *mocks.MockSendAlgorithmWithDebugInfos
@ -437,7 +396,7 @@ var _ = Describe("SentPacketHandler", func() {
handler.SentPacket(&Packet{
PacketNumber: 1,
Length: 42,
Frames: []wire.Frame{&wire.PingFrame{}},
Frames: []Frame{{Frame: &wire.PingFrame{}, OnLost: func(wire.Frame) {}}},
EncryptionLevel: protocol.Encryption1RTT,
})
})
@ -531,32 +490,10 @@ var _ = Describe("SentPacketHandler", func() {
Expect(handler.SendMode()).To(Equal(SendAck))
})
It("doesn't allow retransmissions if congestion limited", func() {
handler.retransmissionQueue = []*Packet{{PacketNumber: 3}}
cong.EXPECT().CanSend(gomock.Any()).Return(false)
Expect(handler.SendMode()).To(Equal(SendAck))
})
It("allows sending retransmissions", func() {
cong.EXPECT().CanSend(gomock.Any()).Return(true)
handler.retransmissionQueue = []*Packet{{PacketNumber: 3}}
Expect(handler.SendMode()).To(Equal(SendRetransmission))
})
It("allows retransmissions, if we're keeping track of between MaxOutstandingSentPackets and MaxTrackedSentPackets packets", func() {
cong.EXPECT().CanSend(gomock.Any()).Return(true)
Expect(protocol.MaxOutstandingSentPackets).To(BeNumerically("<", protocol.MaxTrackedSentPackets))
handler.retransmissionQueue = make([]*Packet, protocol.MaxOutstandingSentPackets+10)
Expect(handler.SendMode()).To(Equal(SendRetransmission))
handler.retransmissionQueue = make([]*Packet, protocol.MaxTrackedSentPackets)
Expect(handler.SendMode()).To(Equal(SendNone))
})
It("allows RTOs, even when congestion limited", func() {
// note that we don't EXPECT a call to GetCongestionWindow
// that means retransmissions are sent without considering the congestion window
handler.numProbesToSend = 1
handler.retransmissionQueue = []*Packet{{PacketNumber: 3}}
Expect(handler.SendMode()).To(Equal(SendPTO))
})
@ -607,6 +544,19 @@ var _ = Describe("SentPacketHandler", func() {
})
Context("probe packets", func() {
It("queues a probe packet", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 10}))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 11}))
queued := handler.QueueProbePacket()
Expect(queued).To(BeTrue())
Expect(lostPackets).To(Equal([]protocol.PacketNumber{10}))
})
It("says when it can't queue a probe packet", func() {
queued := handler.QueueProbePacket()
Expect(queued).To(BeFalse())
})
It("implements exponential backoff", func() {
sendTime := time.Now().Add(-time.Hour)
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 1, SendTime: sendTime}))
@ -620,9 +570,16 @@ var _ = Describe("SentPacketHandler", func() {
Expect(handler.GetLossDetectionTimeout().Sub(sendTime)).To(Equal(4 * timeout))
})
It("sets the TPO send mode until two packets is sent", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
handler.OnLossDetectionTimeout()
It("sets the PTO send mode until two packets is sent", func() {
var lostPackets []protocol.PacketNumber
handler.SentPacket(ackElicitingPacket(&Packet{
PacketNumber: 1,
SendTime: time.Now().Add(-time.Hour),
Frames: []Frame{
{Frame: &wire.PingFrame{}, OnLost: func(wire.Frame) { lostPackets = append(lostPackets, 1) }},
},
}))
Expect(handler.OnLossDetectionTimeout()).To(Succeed())
Expect(handler.SendMode()).To(Equal(SendPTO))
Expect(handler.ShouldSendNumPackets()).To(Equal(2))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 2}))
@ -633,7 +590,7 @@ var _ = Describe("SentPacketHandler", func() {
It("only counts ack-eliciting packets as probe packets", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
handler.OnLossDetectionTimeout()
Expect(handler.OnLossDetectionTimeout()).To(Succeed())
Expect(handler.SendMode()).To(Equal(SendPTO))
Expect(handler.ShouldSendNumPackets()).To(Equal(2))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 2}))
@ -653,118 +610,49 @@ var _ = Describe("SentPacketHandler", func() {
updateRTT(time.Hour)
Expect(handler.oneRTTPackets.lossTime.IsZero()).To(BeTrue())
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // RTO
p, err := handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
Expect(p.PacketNumber).To(Equal(protocol.PacketNumber(1)))
p, err = handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
Expect(p.PacketNumber).To(Equal(protocol.PacketNumber(2)))
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(2)))
Expect(handler.OnLossDetectionTimeout()).To(Succeed()) // TLP
Expect(handler.ptoCount).To(BeEquivalentTo(1))
Expect(handler.SendMode()).To(Equal(SendPTO))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 3}))
Expect(handler.SendMode()).To(Equal(SendPTO))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 4}))
Expect(handler.ptoCount).To(BeEquivalentTo(3))
Expect(handler.OnLossDetectionTimeout()).To(Succeed()) // PTO
Expect(handler.ptoCount).To(BeEquivalentTo(2))
Expect(handler.SendMode()).To(Equal(SendPTO))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5}))
Expect(handler.SendMode()).To(Equal(SendPTO))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 6}))
Expect(handler.SendMode()).To(Equal(SendAny))
})
It("gets two probe packets if RTO expires, for crypto packets", func() {
It("gets two probe packets if PTO expires, for crypto packets", func() {
handler.SentPacket(cryptoPacket(&Packet{PacketNumber: 1}))
handler.SentPacket(cryptoPacket(&Packet{PacketNumber: 2}))
updateRTT(time.Hour)
Expect(handler.initialPackets.lossTime.IsZero()).To(BeTrue())
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // RTO
p, err := handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
Expect(p.PacketNumber).To(Equal(protocol.PacketNumber(1)))
p, err = handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
Expect(p.PacketNumber).To(Equal(protocol.PacketNumber(2)))
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(2)))
Expect(handler.OnLossDetectionTimeout()).To(Succeed())
Expect(handler.SendMode()).To(Equal(SendPTO))
handler.SentPacket(cryptoPacket(&Packet{PacketNumber: 3}))
Expect(handler.SendMode()).To(Equal(SendPTO))
handler.SentPacket(cryptoPacket(&Packet{PacketNumber: 3}))
Expect(handler.ptoCount).To(BeEquivalentTo(3))
})
It("doesn't delete packets transmitted as PTO from the history", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 2, SendTime: time.Now().Add(-time.Hour)}))
handler.rttStats.UpdateRTT(time.Second, 0, time.Now())
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // RTO
_, err := handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
_, err = handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
expectInPacketHistory([]protocol.PacketNumber{1, 2}, protocol.Encryption1RTT)
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(2)))
// Send a probe packet and receive an ACK for it.
// This verifies the RTO.
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 3}))
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 3, Largest: 3}}}
Expect(handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())).To(Succeed())
Expect(err).ToNot(HaveOccurred())
Expect(handler.oneRTTPackets.history.Len()).To(BeZero())
Expect(handler.bytesInFlight).To(BeZero())
Expect(handler.retransmissionQueue).To(BeEmpty()) // 1 and 2 were already sent as probe packets
Expect(handler.SendMode()).To(Equal(SendAny))
})
It("resets the send mode when it receives an acknowledgement after queueing probe packets", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
handler.rttStats.UpdateRTT(time.Second, 0, time.Now())
handler.OnLossDetectionTimeout()
Expect(handler.OnLossDetectionTimeout()).To(Succeed())
Expect(handler.SendMode()).To(Equal(SendPTO))
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 1, Largest: 1}}}
Expect(handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())).To(Succeed())
Expect(handler.SendMode()).To(Equal(SendAny))
})
It("gets packets sent before the probe packet for retransmission", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 2, SendTime: time.Now().Add(-time.Hour)}))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 3, SendTime: time.Now().Add(-time.Hour)}))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 4, SendTime: time.Now().Add(-time.Hour)}))
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5, SendTime: time.Now().Add(-time.Hour)}))
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // RTO
_, err := handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
_, err = handler.DequeueProbePacket()
Expect(err).ToNot(HaveOccurred())
expectInPacketHistory([]protocol.PacketNumber{1, 2, 3, 4, 5}, protocol.Encryption1RTT)
// Send a probe packet and receive an ACK for it.
// This verifies the RTO.
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 6}))
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 6, Largest: 6}}}
err = handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())
Expect(err).ToNot(HaveOccurred())
Expect(handler.oneRTTPackets.history.Len()).To(BeZero())
Expect(handler.bytesInFlight).To(BeZero())
Expect(handler.retransmissionQueue).To(HaveLen(3)) // packets 3, 4, 5
})
It("handles ACKs for the original packet", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5, SendTime: time.Now().Add(-time.Hour)}))
handler.rttStats.UpdateRTT(time.Second, 0, time.Now())
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // TLP
handler.OnLossDetectionTimeout() // RTO
handler.SentPacketsAsRetransmission([]*Packet{ackElicitingPacket(&Packet{PacketNumber: 6})}, 5)
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 5, Largest: 5}}}
err := handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())
Expect(err).ToNot(HaveOccurred())
err = handler.OnLossDetectionTimeout()
Expect(err).ToNot(HaveOccurred())
})
It("handles ACKs for the original packet", func() {
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 5, SendTime: time.Now().Add(-time.Hour)}))
handler.rttStats.UpdateRTT(time.Second, 0, time.Now())
@ -783,11 +671,7 @@ var _ = Describe("SentPacketHandler", func() {
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 6, Largest: 6}}}
Expect(handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, time.Now())).To(Succeed())
expectInPacketHistory([]protocol.PacketNumber{4, 5}, protocol.Encryption1RTT)
for _, p := range []protocol.PacketNumber{1, 2, 3} {
lost := handler.DequeuePacketForRetransmission()
Expect(lost).ToNot(BeNil())
Expect(lost.PacketNumber).To(Equal(p))
}
Expect(lostPackets).To(Equal([]protocol.PacketNumber{1, 2, 3}))
})
})
@ -800,8 +684,6 @@ var _ = Describe("SentPacketHandler", func() {
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 2, Largest: 2}}}
Expect(handler.ReceivedAck(ack, 1, protocol.Encryption1RTT, now)).To(Succeed())
Expect(handler.DequeuePacketForRetransmission()).ToNot(BeNil())
Expect(handler.DequeuePacketForRetransmission()).To(BeNil())
// no need to set an alarm, since packet 1 was already declared lost
Expect(handler.oneRTTPackets.lossTime.IsZero()).To(BeTrue())
Expect(handler.bytesInFlight).To(BeZero())
@ -821,11 +703,6 @@ var _ = Describe("SentPacketHandler", func() {
// Packet 1 should be considered lost (1+1/8) RTTs after it was sent.
Expect(handler.oneRTTPackets.lossTime.IsZero()).To(BeFalse())
Expect(handler.oneRTTPackets.lossTime.Sub(getPacket(1, protocol.Encryption1RTT).SendTime)).To(Equal(time.Second * 9 / 8))
Expect(handler.OnLossDetectionTimeout()).To(Succeed())
Expect(handler.DequeuePacketForRetransmission()).NotTo(BeNil())
// make sure this is not an RTO: only packet 1 is retransmissted
Expect(handler.DequeuePacketForRetransmission()).To(BeNil())
})
It("sets the early retransmit alarm for crypto packets", func() {
@ -842,22 +719,15 @@ var _ = Describe("SentPacketHandler", func() {
// Packet 1 should be considered lost (1+1/8) RTTs after it was sent.
Expect(handler.initialPackets.lossTime.IsZero()).To(BeFalse())
Expect(handler.initialPackets.lossTime.Sub(getPacket(1, protocol.EncryptionInitial).SendTime)).To(Equal(time.Second * 9 / 8))
Expect(handler.OnLossDetectionTimeout()).To(Succeed())
Expect(handler.DequeuePacketForRetransmission()).NotTo(BeNil())
// make sure this is not an RTO: only packet 1 is retransmissted
Expect(handler.DequeuePacketForRetransmission()).To(BeNil())
})
})
Context("crypto packets", func() {
It("rejects an ACK that acks packets with a higher encryption level", func() {
handler.SentPacket(&Packet{
handler.SentPacket(ackElicitingPacket(&Packet{
PacketNumber: 13,
EncryptionLevel: protocol.Encryption1RTT,
Frames: []wire.Frame{&streamFrame},
Length: 1,
})
}))
ack := &wire.AckFrame{AckRanges: []wire.AckRange{{Smallest: 13, Largest: 13}}}
err := handler.ReceivedAck(ack, 1, protocol.EncryptionHandshake, time.Now())
Expect(err).To(MatchError("PROTOCOL_VIOLATION: Received ACK for an unsent packet"))
@ -865,43 +735,43 @@ var _ = Describe("SentPacketHandler", func() {
It("deletes Initial packets", func() {
for i := protocol.PacketNumber(0); i < 6; i++ {
p := ackElicitingPacket(&Packet{PacketNumber: i, EncryptionLevel: protocol.EncryptionInitial})
handler.SentPacket(p)
handler.SentPacket(ackElicitingPacket(&Packet{
PacketNumber: i,
EncryptionLevel: protocol.EncryptionInitial,
}))
}
for i := protocol.PacketNumber(0); i < 10; i++ {
p := ackElicitingPacket(&Packet{PacketNumber: i, EncryptionLevel: protocol.EncryptionHandshake})
handler.SentPacket(p)
handler.SentPacket(ackElicitingPacket(&Packet{
PacketNumber: i,
EncryptionLevel: protocol.EncryptionHandshake,
}))
}
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(16)))
handler.queuePacketForRetransmission(getPacket(1, protocol.EncryptionInitial), handler.getPacketNumberSpace(protocol.EncryptionInitial))
lostPacket := getPacket(3, protocol.EncryptionHandshake)
handler.queuePacketForRetransmission(lostPacket, handler.getPacketNumberSpace(protocol.EncryptionHandshake))
handler.DropPackets(protocol.EncryptionInitial)
Expect(lostPackets).To(BeEmpty()) // frames must not be queued for retransmission
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(10)))
Expect(handler.initialPackets).To(BeNil())
Expect(handler.handshakePackets.history.Len()).ToNot(BeZero())
packet := handler.DequeuePacketForRetransmission()
Expect(packet).To(Equal(lostPacket))
})
It("deletes Handshake packets", func() {
for i := protocol.PacketNumber(0); i < 6; i++ {
p := ackElicitingPacket(&Packet{PacketNumber: i, EncryptionLevel: protocol.EncryptionHandshake})
handler.SentPacket(p)
handler.SentPacket(ackElicitingPacket(&Packet{
PacketNumber: i,
EncryptionLevel: protocol.EncryptionHandshake,
}))
}
for i := protocol.PacketNumber(0); i < 10; i++ {
p := ackElicitingPacket(&Packet{PacketNumber: i, EncryptionLevel: protocol.Encryption1RTT})
handler.SentPacket(p)
handler.SentPacket(ackElicitingPacket(&Packet{
PacketNumber: i,
EncryptionLevel: protocol.Encryption1RTT,
}))
}
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(16)))
handler.queuePacketForRetransmission(getPacket(1, protocol.EncryptionHandshake), handler.getPacketNumberSpace(protocol.EncryptionInitial))
lostPacket := getPacket(3, protocol.Encryption1RTT)
handler.queuePacketForRetransmission(lostPacket, handler.getPacketNumberSpace(protocol.EncryptionHandshake))
handler.DropPackets(protocol.EncryptionHandshake)
Expect(lostPackets).To(BeEmpty()) // frames must not be queued for retransmission
Expect(handler.bytesInFlight).To(Equal(protocol.ByteCount(10)))
Expect(handler.handshakePackets).To(BeNil())
packet := handler.DequeuePacketForRetransmission()
Expect(packet).To(Equal(lostPacket))
})
})
@ -929,25 +799,16 @@ var _ = Describe("SentPacketHandler", func() {
Context("resetting for retry", func() {
It("queues outstanding packets for retransmission and cancels alarms", func() {
packet := &Packet{
PacketNumber: 42,
EncryptionLevel: protocol.EncryptionInitial,
Frames: []wire.Frame{&wire.CryptoFrame{Data: []byte("foobar")}},
Length: 100,
}
handler.SentPacket(packet)
handler.SentPacket(ackElicitingPacket(&Packet{PacketNumber: 42, EncryptionLevel: protocol.EncryptionInitial}))
Expect(handler.GetLossDetectionTimeout()).ToNot(BeZero())
Expect(handler.bytesInFlight).ToNot(BeZero())
Expect(handler.DequeuePacketForRetransmission()).To(BeNil())
Expect(handler.SendMode()).To(Equal(SendAny))
// now receive a Retry
Expect(handler.ResetForRetry()).To(Succeed())
Expect(lostPackets).To(Equal([]protocol.PacketNumber{42}))
Expect(handler.bytesInFlight).To(BeZero())
Expect(handler.GetLossDetectionTimeout()).To(BeZero())
Expect(handler.SendMode()).To(Equal(SendRetransmission))
p := handler.DequeuePacketForRetransmission()
Expect(p.PacketNumber).To(Equal(packet.PacketNumber))
Expect(p.Frames).To(Equal(packet.Frames))
Expect(handler.SendMode()).To(Equal(SendAny))
})
})
})

81
internal/ackhandler/sent_packet_history.go
View file

@ -9,10 +9,6 @@ import (
type sentPacketHistory struct {
packetList *PacketList
packetMap map[protocol.PacketNumber]*PacketElement
numOutstandingPackets int
firstOutstanding *PacketElement
}
func newSentPacketHistory() *sentPacketHistory {
@ -23,40 +19,8 @@ func newSentPacketHistory() *sentPacketHistory {
}
func (h *sentPacketHistory) SentPacket(p *Packet) {
h.sentPacketImpl(p)
}
func (h *sentPacketHistory) sentPacketImpl(p *Packet) *PacketElement {
el := h.packetList.PushBack(*p)
h.packetMap[p.PacketNumber] = el
if h.firstOutstanding == nil {
h.firstOutstanding = el
}
if p.canBeRetransmitted {
h.numOutstandingPackets++
}
return el
}
func (h *sentPacketHistory) SentPacketsAsRetransmission(packets []*Packet, retransmissionOf protocol.PacketNumber) {
retransmission, ok := h.packetMap[retransmissionOf]
// The retransmitted packet is not present anymore.
// This can happen if it was acked in between dequeueing of the retransmission and sending.
// Just treat the retransmissions as normal packets.
// TODO: This won't happen if we clear packets queued for retransmission on new ACKs.
if !ok {
for _, packet := range packets {
h.sentPacketImpl(packet)
}
return
}
retransmission.Value.retransmittedAs = make([]protocol.PacketNumber, len(packets))
for i, packet := range packets {
retransmission.Value.retransmittedAs[i] = packet.PacketNumber
el := h.sentPacketImpl(packet)
el.Value.isRetransmission = true
el.Value.retransmissionOf = retransmissionOf
}
}
func (h *sentPacketHistory) GetPacket(p protocol.PacketNumber) *Packet {
@ -84,40 +48,10 @@ func (h *sentPacketHistory) Iterate(cb func(*Packet) (cont bool, err error)) err
// It must not be modified (e.g. retransmitted).
// Use DequeueFirstPacketForRetransmission() to retransmit it.
func (h *sentPacketHistory) FirstOutstanding() *Packet {
if h.firstOutstanding == nil {
if !h.HasOutstandingPackets() {
return nil
}
return &h.firstOutstanding.Value
}
// QueuePacketForRetransmission marks a packet for retransmission.
// A packet can only be queued once.
func (h *sentPacketHistory) MarkCannotBeRetransmitted(pn protocol.PacketNumber) error {
el, ok := h.packetMap[pn]
if !ok {
return fmt.Errorf("sent packet history: packet %d not found", pn)
}
if el.Value.canBeRetransmitted {
h.numOutstandingPackets--
if h.numOutstandingPackets < 0 {
panic("numOutstandingHandshakePackets negative")
}
}
el.Value.canBeRetransmitted = false
if el == h.firstOutstanding {
h.readjustFirstOutstanding()
}
return nil
}
// readjustFirstOutstanding readjusts the pointer to the first outstanding packet.
// This is necessary every time the first outstanding packet is deleted or retransmitted.
func (h *sentPacketHistory) readjustFirstOutstanding() {
el := h.firstOutstanding.Next()
for el != nil && !el.Value.canBeRetransmitted {
el = el.Next()
}
h.firstOutstanding = el
return &h.packetList.Front().Value
}
func (h *sentPacketHistory) Len() int {
@ -129,20 +63,11 @@ func (h *sentPacketHistory) Remove(p protocol.PacketNumber) error {
if !ok {
return fmt.Errorf("packet %d not found in sent packet history", p)
}
if el == h.firstOutstanding {
h.readjustFirstOutstanding()
}
if el.Value.canBeRetransmitted {
h.numOutstandingPackets--
if h.numOutstandingPackets < 0 {
panic("numOutstandingHandshakePackets negative")
}
}
h.packetList.Remove(el)
delete(h.packetMap, p)
return nil
}
func (h *sentPacketHistory) HasOutstandingPackets() bool {
return h.numOutstandingPackets > 0
return h.packetList.Len() > 0
}

127
internal/ackhandler/sent_packet_history_test.go
View file

@ -15,13 +15,14 @@ var _ = Describe("SentPacketHistory", func() {
ExpectWithOffset(1, hist.packetMap).To(HaveLen(len(packetNumbers)))
ExpectWithOffset(1, hist.packetList.Len()).To(Equal(len(packetNumbers)))
i := 0
hist.Iterate(func(p *Packet) (bool, error) {
err := hist.Iterate(func(p *Packet) (bool, error) {
pn := packetNumbers[i]
ExpectWithOffset(1, p.PacketNumber).To(Equal(pn))
ExpectWithOffset(1, hist.packetMap[pn].Value.PacketNumber).To(Equal(pn))
i++
return true, nil
})
Expect(err).ToNot(HaveOccurred())
}
BeforeEach(func() {
@ -53,45 +54,6 @@ var _ = Describe("SentPacketHistory", func() {
Expect(front).ToNot(BeNil())
Expect(front.PacketNumber).To(Equal(protocol.PacketNumber(2)))
})
It("gets the second packet if the first one is retransmitted", func() {
hist.SentPacket(&Packet{PacketNumber: 1, canBeRetransmitted: true})
hist.SentPacket(&Packet{PacketNumber: 3, canBeRetransmitted: true})
hist.SentPacket(&Packet{PacketNumber: 4, canBeRetransmitted: true})
front := hist.FirstOutstanding()
Expect(front).ToNot(BeNil())
Expect(front.PacketNumber).To(Equal(protocol.PacketNumber(1)))
// Queue the first packet for retransmission.
// The first outstanding packet should now be 3.
err := hist.MarkCannotBeRetransmitted(1)
Expect(err).ToNot(HaveOccurred())
front = hist.FirstOutstanding()
Expect(front).ToNot(BeNil())
Expect(front.PacketNumber).To(Equal(protocol.PacketNumber(3)))
})
It("gets the third packet if the first two are retransmitted", func() {
hist.SentPacket(&Packet{PacketNumber: 1, canBeRetransmitted: true})
hist.SentPacket(&Packet{PacketNumber: 3, canBeRetransmitted: true})
hist.SentPacket(&Packet{PacketNumber: 4, canBeRetransmitted: true})
front := hist.FirstOutstanding()
Expect(front).ToNot(BeNil())
Expect(front.PacketNumber).To(Equal(protocol.PacketNumber(1)))
// Queue the second packet for retransmission.
// The first outstanding packet should still be 3.
err := hist.MarkCannotBeRetransmitted(3)
Expect(err).ToNot(HaveOccurred())
front = hist.FirstOutstanding()
Expect(front).ToNot(BeNil())
Expect(front.PacketNumber).To(Equal(protocol.PacketNumber(1)))
// Queue the first packet for retransmission.
// The first outstanding packet should still be 4.
err = hist.MarkCannotBeRetransmitted(1)
Expect(err).ToNot(HaveOccurred())
front = hist.FirstOutstanding()
Expect(front).ToNot(BeNil())
Expect(front.PacketNumber).To(Equal(protocol.PacketNumber(4)))
})
})
It("gets a packet by packet number", func() {
@ -161,100 +123,35 @@ var _ = Describe("SentPacketHistory", func() {
})
})
Context("retransmissions", func() {
BeforeEach(func() {
for i := protocol.PacketNumber(1); i <= 5; i++ {
hist.SentPacket(&Packet{PacketNumber: i})
}
})
It("errors if the packet doesn't exist", func() {
err := hist.MarkCannotBeRetransmitted(100)
Expect(err).To(MatchError("sent packet history: packet 100 not found"))
})
It("adds a sent packets as a retransmission", func() {
hist.SentPacketsAsRetransmission([]*Packet{{PacketNumber: 13}}, 2)
expectInHistory([]protocol.PacketNumber{1, 2, 3, 4, 5, 13})
Expect(hist.GetPacket(13).isRetransmission).To(BeTrue())
Expect(hist.GetPacket(13).retransmissionOf).To(Equal(protocol.PacketNumber(2)))
Expect(hist.GetPacket(2).retransmittedAs).To(Equal([]protocol.PacketNumber{13}))
})
It("adds multiple packets sent as a retransmission", func() {
hist.SentPacketsAsRetransmission([]*Packet{{PacketNumber: 13}, {PacketNumber: 15}}, 2)
expectInHistory([]protocol.PacketNumber{1, 2, 3, 4, 5, 13, 15})
Expect(hist.GetPacket(13).isRetransmission).To(BeTrue())
Expect(hist.GetPacket(13).retransmissionOf).To(Equal(protocol.PacketNumber(2)))
Expect(hist.GetPacket(15).retransmissionOf).To(Equal(protocol.PacketNumber(2)))
Expect(hist.GetPacket(2).retransmittedAs).To(Equal([]protocol.PacketNumber{13, 15}))
})
It("adds a packet as a normal packet if the retransmitted packet doesn't exist", func() {
hist.SentPacketsAsRetransmission([]*Packet{{PacketNumber: 13}}, 7)
expectInHistory([]protocol.PacketNumber{1, 2, 3, 4, 5, 13})
Expect(hist.GetPacket(13).isRetransmission).To(BeFalse())
Expect(hist.GetPacket(13).retransmissionOf).To(BeZero())
})
})
Context("outstanding packets", func() {
It("says if it has outstanding packets", func() {
Expect(hist.HasOutstandingPackets()).To(BeFalse())
hist.SentPacket(&Packet{
EncryptionLevel: protocol.Encryption1RTT,
canBeRetransmitted: true,
})
hist.SentPacket(&Packet{EncryptionLevel: protocol.Encryption1RTT})
Expect(hist.HasOutstandingPackets()).To(BeTrue())
})
It("doesn't consider non-ack-eliciting packets as outstanding", func() {
hist.SentPacket(&Packet{
EncryptionLevel: protocol.EncryptionInitial,
})
Expect(hist.HasOutstandingPackets()).To(BeFalse())
})
It("accounts for deleted packets", func() {
hist.SentPacket(&Packet{
PacketNumber: 10,
EncryptionLevel: protocol.Encryption1RTT,
canBeRetransmitted: true,
PacketNumber: 10,
EncryptionLevel: protocol.Encryption1RTT,
})
Expect(hist.HasOutstandingPackets()).To(BeTrue())
err := hist.Remove(10)
Expect(err).ToNot(HaveOccurred())
Expect(hist.HasOutstandingPackets()).To(BeFalse())
})
It("doesn't count packets marked as non-ack-eliciting", func() {
hist.SentPacket(&Packet{
PacketNumber: 10,
EncryptionLevel: protocol.Encryption1RTT,
canBeRetransmitted: true,
})
Expect(hist.HasOutstandingPackets()).To(BeTrue())
err := hist.MarkCannotBeRetransmitted(10)
Expect(err).ToNot(HaveOccurred())
Expect(hist.Remove(10)).To(Succeed())
Expect(hist.HasOutstandingPackets()).To(BeFalse())
})
It("counts the number of packets", func() {
hist.SentPacket(&Packet{
PacketNumber: 10,
EncryptionLevel: protocol.Encryption1RTT,
canBeRetransmitted: true,
PacketNumber: 10,
EncryptionLevel: protocol.Encryption1RTT,
})
hist.SentPacket(&Packet{
PacketNumber: 11,
EncryptionLevel: protocol.Encryption1RTT,
canBeRetransmitted: true,
PacketNumber: 11,
EncryptionLevel: protocol.Encryption1RTT,
})
err := hist.Remove(11)
Expect(err).ToNot(HaveOccurred())
Expect(hist.Remove(11)).To(Succeed())
Expect(hist.HasOutstandingPackets()).To(BeTrue())
err = hist.Remove(10)
Expect(err).ToNot(HaveOccurred())
Expect(hist.Remove(10)).To(Succeed())
Expect(hist.HasOutstandingPackets()).To(BeFalse())
})
})

55
internal/mocks/ackhandler/sent_packet_handler.go
View file

@ -38,35 +38,6 @@ func (m *MockSentPacketHandler) EXPECT() *MockSentPacketHandlerMockRecorder {
return m.recorder
}
// DequeuePacketForRetransmission mocks base method
func (m *MockSentPacketHandler) DequeuePacketForRetransmission() *ackhandler.Packet {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DequeuePacketForRetransmission")
ret0, _ := ret[0].(*ackhandler.Packet)
return ret0
}
// DequeuePacketForRetransmission indicates an expected call of DequeuePacketForRetransmission
func (mr *MockSentPacketHandlerMockRecorder) DequeuePacketForRetransmission() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DequeuePacketForRetransmission", reflect.TypeOf((*MockSentPacketHandler)(nil).DequeuePacketForRetransmission))
}
// DequeueProbePacket mocks base method
func (m *MockSentPacketHandler) DequeueProbePacket() (*ackhandler.Packet, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DequeueProbePacket")
ret0, _ := ret[0].(*ackhandler.Packet)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// DequeueProbePacket indicates an expected call of DequeueProbePacket
func (mr *MockSentPacketHandlerMockRecorder) DequeueProbePacket() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DequeueProbePacket", reflect.TypeOf((*MockSentPacketHandler)(nil).DequeueProbePacket))
}
// DropPackets mocks base method
func (m *MockSentPacketHandler) DropPackets(arg0 protocol.EncryptionLevel) {
m.ctrl.T.Helper()
@ -164,6 +135,20 @@ func (mr *MockSentPacketHandlerMockRecorder) PopPacketNumber(arg0 interface{}) *
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PopPacketNumber", reflect.TypeOf((*MockSentPacketHandler)(nil).PopPacketNumber), arg0)
}
// QueueProbePacket mocks base method
func (m *MockSentPacketHandler) QueueProbePacket() bool {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "QueueProbePacket")
ret0, _ := ret[0].(bool)
return ret0
}
// QueueProbePacket indicates an expected call of QueueProbePacket
func (mr *MockSentPacketHandlerMockRecorder) QueueProbePacket() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "QueueProbePacket", reflect.TypeOf((*MockSentPacketHandler)(nil).QueueProbePacket))
}
// ReceivedAck mocks base method
func (m *MockSentPacketHandler) ReceivedAck(arg0 *wire.AckFrame, arg1 protocol.PacketNumber, arg2 protocol.EncryptionLevel, arg3 time.Time) error {
m.ctrl.T.Helper()
@ -218,18 +203,6 @@ func (mr *MockSentPacketHandlerMockRecorder) SentPacket(arg0 interface{}) *gomoc
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SentPacket", reflect.TypeOf((*MockSentPacketHandler)(nil).SentPacket), arg0)
}
// SentPacketsAsRetransmission mocks base method
func (m *MockSentPacketHandler) SentPacketsAsRetransmission(arg0 []*ackhandler.Packet, arg1 protocol.PacketNumber) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "SentPacketsAsRetransmission", arg0, arg1)
}
// SentPacketsAsRetransmission indicates an expected call of SentPacketsAsRetransmission
func (mr *MockSentPacketHandlerMockRecorder) SentPacketsAsRetransmission(arg0, arg1 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SentPacketsAsRetransmission", reflect.TypeOf((*MockSentPacketHandler)(nil).SentPacketsAsRetransmission), arg0, arg1)
}
// ShouldSendNumPackets mocks base method
func (m *MockSentPacketHandler) ShouldSendNumPackets() int {
m.ctrl.T.Helper()

13
internal/wire/stream_frame.go
View file

@ -143,15 +143,18 @@ func (f *StreamFrame) MaxDataLen(maxSize protocol.ByteCount, version protocol.Ve
}
// MaybeSplitOffFrame splits a frame such that it is not bigger than n bytes.
// If n >= len(frame), nil is returned and nothing is modified.
func (f *StreamFrame) MaybeSplitOffFrame(maxSize protocol.ByteCount, version protocol.VersionNumber) (*StreamFrame, error) {
// It returns if the frame was actually split.
// The frame might not be split if:
// * the size is large enough to fit the whole frame
// * the size is too small to fit even a 1-byte frame. In that case, the frame returned is nil.
func (f *StreamFrame) MaybeSplitOffFrame(maxSize protocol.ByteCount, version protocol.VersionNumber) (*StreamFrame, bool /* was splitting required */) {
if maxSize >= f.Length(version) {
return nil, nil
return nil, false
}
n := f.MaxDataLen(maxSize, version)
if n == 0 {
return nil, errors.New("too small")
return nil, true
}
newFrame := &StreamFrame{
FinBit: false,
@ -164,5 +167,5 @@ func (f *StreamFrame) MaybeSplitOffFrame(maxSize protocol.ByteCount, version pro
f.Data = f.Data[n:]
f.Offset += n
return newFrame, nil
return newFrame, true
}

61
internal/wire/stream_frame_test.go
View file

@ -287,12 +287,13 @@ var _ = Describe("STREAM frame", func() {
Offset: 0xdeadbeef,
Data: make([]byte, 100),
}
newFrame, err := f.MaybeSplitOffFrame(f.Length(versionIETFFrames), versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
Expect(newFrame).To(BeNil())
newFrame, err = f.MaybeSplitOffFrame(f.Length(versionIETFFrames)-1, versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
Expect(newFrame).ToNot(BeNil())
frame, needsSplit := f.MaybeSplitOffFrame(f.Length(versionIETFFrames), versionIETFFrames)
Expect(needsSplit).To(BeFalse())
Expect(frame).To(BeNil())
Expect(f.DataLen()).To(BeEquivalentTo(100))
frame, needsSplit = f.MaybeSplitOffFrame(f.Length(versionIETFFrames)-1, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(frame.DataLen()).To(BeEquivalentTo(99))
})
It("keeps the data len", func() {
@ -301,11 +302,11 @@ var _ = Describe("STREAM frame", func() {
DataLenPresent: true,
Data: make([]byte, 100),
}
newFrame, err := f.MaybeSplitOffFrame(66, versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
Expect(newFrame).ToNot(BeNil())
frame, needsSplit := f.MaybeSplitOffFrame(66, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(frame).ToNot(BeNil())
Expect(f.DataLenPresent).To(BeTrue())
Expect(newFrame.DataLenPresent).To(BeTrue())
Expect(frame.DataLenPresent).To(BeTrue())
})
It("adjusts the offset", func() {
@ -314,11 +315,11 @@ var _ = Describe("STREAM frame", func() {
Offset: 0x100,
Data: []byte("foobar"),
}
newFrame, err := f.MaybeSplitOffFrame(f.Length(versionIETFFrames)-3, versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
Expect(newFrame).ToNot(BeNil())
Expect(newFrame.Offset).To(Equal(protocol.ByteCount(0x100)))
Expect(newFrame.Data).To(Equal([]byte("foo")))
frame, needsSplit := f.MaybeSplitOffFrame(f.Length(versionIETFFrames)-3, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(frame).ToNot(BeNil())
Expect(frame.Offset).To(Equal(protocol.ByteCount(0x100)))
Expect(frame.Data).To(Equal([]byte("foo")))
Expect(f.Offset).To(Equal(protocol.ByteCount(0x100 + 3)))
Expect(f.Data).To(Equal([]byte("bar")))
})
@ -330,12 +331,12 @@ var _ = Describe("STREAM frame", func() {
Offset: 0xdeadbeef,
Data: make([]byte, 100),
}
newFrame, err := f.MaybeSplitOffFrame(50, versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
Expect(newFrame).ToNot(BeNil())
Expect(newFrame.Offset).To(BeNumerically("<", f.Offset))
frame, needsSplit := f.MaybeSplitOffFrame(50, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(frame).ToNot(BeNil())
Expect(frame.Offset).To(BeNumerically("<", f.Offset))
Expect(f.FinBit).To(BeTrue())
Expect(newFrame.FinBit).To(BeFalse())
Expect(frame.FinBit).To(BeFalse())
})
It("produces frames of the correct length, without data len", func() {
@ -347,14 +348,15 @@ var _ = Describe("STREAM frame", func() {
}
minFrameSize := f.Length(versionIETFFrames)
for i := protocol.ByteCount(0); i < minFrameSize; i++ {
_, err := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(err).To(HaveOccurred())
f, needsSplit := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(f).To(BeNil())
}
for i := minFrameSize; i < size; i++ {
f.Data = make([]byte, size)
newFrame, err := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
Expect(newFrame.Length(versionIETFFrames)).To(Equal(i))
f, needsSplit := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(f.Length(versionIETFFrames)).To(Equal(i))
}
})
@ -368,14 +370,15 @@ var _ = Describe("STREAM frame", func() {
}
minFrameSize := f.Length(versionIETFFrames)
for i := protocol.ByteCount(0); i < minFrameSize; i++ {
_, err := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(err).To(HaveOccurred())
f, needsSplit := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
Expect(f).To(BeNil())
}
var frameOneByteTooSmallCounter int
for i := minFrameSize; i < size; i++ {
f.Data = make([]byte, size)
newFrame, err := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(err).ToNot(HaveOccurred())
newFrame, needsSplit := f.MaybeSplitOffFrame(i, versionIETFFrames)
Expect(needsSplit).To(BeTrue())
// There's *one* pathological case, where a data length of x can be encoded into 1 byte
// but a data lengths of x+1 needs 2 bytes
// In that case, it's impossible to create a STREAM frame of the desired size

16
mock_packer_test.go
View file

@ -8,7 +8,6 @@ import (
reflect "reflect"
gomock "github.com/golang/mock/gomock"
ackhandler "github.com/lucas-clemente/quic-go/internal/ackhandler"
handshake "github.com/lucas-clemente/quic-go/internal/handshake"
protocol "github.com/lucas-clemente/quic-go/internal/protocol"
wire "github.com/lucas-clemente/quic-go/internal/wire"
@ -106,21 +105,6 @@ func (mr *MockPackerMockRecorder) PackPacket() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PackPacket", reflect.TypeOf((*MockPacker)(nil).PackPacket))
}
// PackRetransmission mocks base method
func (m *MockPacker) PackRetransmission(arg0 *ackhandler.Packet) ([]*packedPacket, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "PackRetransmission", arg0)
ret0, _ := ret[0].([]*packedPacket)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// PackRetransmission indicates an expected call of PackRetransmission
func (mr *MockPackerMockRecorder) PackRetransmission(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PackRetransmission", reflect.TypeOf((*MockPacker)(nil).PackRetransmission), arg0)
}
// SetToken mocks base method
func (m *MockPacker) SetToken(arg0 []byte) {
m.ctrl.T.Helper()

177
packet_packer.go
View file

@ -17,7 +17,6 @@ import (
type packer interface {
PackPacket() (*packedPacket, error)
MaybePackAckPacket() (*packedPacket, error)
PackRetransmission(packet *ackhandler.Packet) ([]*packedPacket, error)
PackConnectionClose(*wire.ConnectionCloseFrame) (*packedPacket, error)
HandleTransportParameters(*handshake.TransportParameters)
@ -62,17 +61,31 @@ func (p *packedPacket) IsAckEliciting() bool {
return ackhandler.HasAckElicitingFrames(p.frames)
}
func (p *packedPacket) ToAckHandlerPacket() *ackhandler.Packet {
func (p *packedPacket) ToAckHandlerPacket(q *retransmissionQueue) *ackhandler.Packet {
largestAcked := protocol.InvalidPacketNumber
if p.ack != nil {
largestAcked = p.ack.LargestAcked()
}
encLevel := p.EncryptionLevel()
frames := make([]ackhandler.Frame, len(p.frames))
for i, f := range p.frames {
frame := f
frames[i].Frame = frame
switch encLevel {
case protocol.EncryptionInitial:
frames[i].OnLost = q.AddInitial
case protocol.EncryptionHandshake:
frames[i].OnLost = q.AddHandshake
case protocol.Encryption1RTT:
frames[i].OnLost = q.AddAppData
}
}
return &ackhandler.Packet{
PacketNumber: p.header.PacketNumber,
LargestAcked: largestAcked,
Frames: p.frames,
Frames: frames,
Length: protocol.ByteCount(len(p.raw)),
EncryptionLevel: p.EncryptionLevel(),
EncryptionLevel: encLevel,
SendTime: time.Now(),
}
}
@ -130,9 +143,10 @@ type packetPacker struct {
token []byte
pnManager packetNumberManager
framer frameSource
acks ackFrameSource
pnManager packetNumberManager
framer frameSource
acks ackFrameSource
retransmissionQueue *retransmissionQueue
maxPacketSize protocol.ByteCount
numNonAckElicitingAcks int
@ -146,6 +160,7 @@ func newPacketPacker(
initialStream cryptoStream,
handshakeStream cryptoStream,
packetNumberManager packetNumberManager,
retransmissionQueue *retransmissionQueue,
remoteAddr net.Addr, // only used for determining the max packet size
cryptoSetup sealingManager,
framer frameSource,
@ -154,17 +169,18 @@ func newPacketPacker(
version protocol.VersionNumber,
) *packetPacker {
return &packetPacker{
cryptoSetup: cryptoSetup,
destConnID: destConnID,
srcConnID: srcConnID,
initialStream: initialStream,
handshakeStream: handshakeStream,
perspective: perspective,
version: version,
framer: framer,
acks: acks,
pnManager: packetNumberManager,
maxPacketSize: getMaxPacketSize(remoteAddr),
cryptoSetup: cryptoSetup,
destConnID: destConnID,
srcConnID: srcConnID,
initialStream: initialStream,
handshakeStream: handshakeStream,
retransmissionQueue: retransmissionQueue,
perspective: perspective,
version: version,
framer: framer,
acks: acks,
pnManager: packetNumberManager,
maxPacketSize: getMaxPacketSize(remoteAddr),
}
}
@ -237,80 +253,6 @@ func (p *packetPacker) MaybePackAckPacket() (*packedPacket, error) {
return p.writeAndSealPacket(hdr, payload, encLevel, sealer)
}
// PackRetransmission packs a retransmission
// For packets sent after completion of the handshake, it might happen that 2 packets have to be sent.
// This can happen e.g. when a longer packet number is used in the header.
func (p *packetPacker) PackRetransmission(packet *ackhandler.Packet) ([]*packedPacket, error) {
var controlFrames []wire.Frame
var streamFrames []*wire.StreamFrame
for _, f := range packet.Frames {
// CRYPTO frames are treated as control frames here.
// Since we're making sure that the header can never be larger for a retransmission,
// we never have to split CRYPTO frames.
if sf, ok := f.(*wire.StreamFrame); ok {
sf.DataLenPresent = true
streamFrames = append(streamFrames, sf)
} else {
controlFrames = append(controlFrames, f)
}
}
var packets []*packedPacket
for len(controlFrames) > 0 || len(streamFrames) > 0 {
var frames []wire.Frame
var length protocol.ByteCount
sealer, hdr, err := p.getSealerAndHeader(packet.EncryptionLevel)
if err != nil {
return nil, err
}
hdrLen := hdr.GetLength(p.version)
maxSize := p.maxPacketSize - protocol.ByteCount(sealer.Overhead()) - hdrLen
for len(controlFrames) > 0 {
frame := controlFrames[0]
frameLen := frame.Length(p.version)
if length+frameLen > maxSize {
break
}
length += frameLen
frames = append(frames, frame)
controlFrames = controlFrames[1:]
}
for len(streamFrames) > 0 && length+protocol.MinStreamFrameSize < maxSize {
frame := streamFrames[0]
frame.DataLenPresent = false
frameToAdd := frame
sf, err := frame.MaybeSplitOffFrame(maxSize-length, p.version)
if err != nil {
return nil, err
}
if sf != nil {
frameToAdd = sf
} else {
streamFrames = streamFrames[1:]
}
frame.DataLenPresent = true
length += frameToAdd.Length(p.version)
frames = append(frames, frameToAdd)
}
if sf, ok := frames[len(frames)-1].(*wire.StreamFrame); ok {
sfLen := sf.Length(p.version)
sf.DataLenPresent = false
length += sf.Length(p.version) - sfLen
}
p, err := p.writeAndSealPacket(hdr, payload{frames: frames, length: length}, packet.EncryptionLevel, sealer)
if err != nil {
return nil, err
}
packets = append(packets, p)
}
return packets, nil
}
// PackPacket packs a new packet
// the other controlFrames are sent in the next packet, but might be queued and sent in the next packet if the packet would overflow MaxPacketSize otherwise
func (p *packetPacker) PackPacket() (*packedPacket, error) {
@ -371,9 +313,10 @@ func (p *packetPacker) maybePackCryptoPacket() (*packedPacket, error) {
}
hasData := p.initialStream.HasData()
hasRetransmission := p.retransmissionQueue.HasInitialData()
ack := p.acks.GetAckFrame(protocol.EncryptionInitial)
var sealer handshake.LongHeaderSealer
if hasData || ack != nil {
if hasData || hasRetransmission || ack != nil {
s = p.initialStream
encLevel = protocol.EncryptionInitial
sealer = initialSealer
@ -382,8 +325,9 @@ func (p *packetPacker) maybePackCryptoPacket() (*packedPacket, error) {
}
} else {
hasData = p.handshakeStream.HasData()
hasRetransmission = p.retransmissionQueue.HasHandshakeData()
ack = p.acks.GetAckFrame(protocol.EncryptionHandshake)
if hasData || ack != nil {
if hasData || hasRetransmission || ack != nil {
s = p.handshakeStream
encLevel = protocol.EncryptionHandshake
sealer = handshakeSealer
@ -403,7 +347,24 @@ func (p *packetPacker) maybePackCryptoPacket() (*packedPacket, error) {
}
hdr := p.getLongHeader(encLevel)
hdrLen := hdr.GetLength(p.version)
if hasData {
if hasRetransmission {
for {
var f wire.Frame
switch encLevel {
case protocol.EncryptionInitial:
remainingLen := protocol.MinInitialPacketSize - hdrLen - protocol.ByteCount(sealer.Overhead()) - payload.length
f = p.retransmissionQueue.GetInitialFrame(remainingLen)
case protocol.EncryptionHandshake:
remainingLen := p.maxPacketSize - hdrLen - protocol.ByteCount(sealer.Overhead()) - payload.length
f = p.retransmissionQueue.GetHandshakeFrame(remainingLen)
}
if f == nil {
break
}
payload.frames = append(payload.frames, f)
payload.length += f.Length(p.version)
}
} else if hasData {
cf := s.PopCryptoFrame(p.maxPacketSize - hdrLen - protocol.ByteCount(sealer.Overhead()) - payload.length)
payload.frames = []wire.Frame{cf}
payload.length += cf.Length(p.version)
@ -419,14 +380,25 @@ func (p *packetPacker) composeNextPacket(maxFrameSize protocol.ByteCount) (paylo
payload.length += ack.Length(p.version)
}
frames, lengthAdded := p.framer.AppendControlFrames(payload.frames, maxFrameSize-payload.length)
for {
remainingLen := maxFrameSize - payload.length
if remainingLen < protocol.MinStreamFrameSize {
break
}
f := p.retransmissionQueue.GetAppDataFrame(remainingLen)
if f == nil {
break
}
payload.frames = append(payload.frames, f)
payload.length += f.Length(p.version)
}
var lengthAdded protocol.ByteCount
payload.frames, lengthAdded = p.framer.AppendControlFrames(payload.frames, maxFrameSize-payload.length)
payload.length += lengthAdded
frames, lengthAdded = p.framer.AppendStreamFrames(frames, maxFrameSize-payload.length)
if len(frames) > 0 {
payload.frames = append(payload.frames, frames...)
payload.length += lengthAdded
}
payload.frames, lengthAdded = p.framer.AppendStreamFrames(payload.frames, maxFrameSize-payload.length)
payload.length += lengthAdded
return payload, nil
}
@ -478,6 +450,7 @@ func (p *packetPacker) getLongHeader(encLevel protocol.EncryptionLevel) *wire.Ex
switch encLevel {
case protocol.EncryptionInitial:
hdr.Type = protocol.PacketTypeInitial
hdr.Token = p.token
case protocol.EncryptionHandshake:
hdr.Type = protocol.PacketTypeHandshake
}
@ -507,7 +480,6 @@ func (p *packetPacker) writeAndSealPacket(
if encLevel != protocol.Encryption1RTT {
if p.perspective == protocol.PerspectiveClient && header.Type == protocol.PacketTypeInitial {
header.Token = p.token
headerLen := header.GetLength(p.version)
header.Length = pnLen + protocol.MinInitialPacketSize - headerLen
paddingLen = protocol.ByteCount(protocol.MinInitialPacketSize-sealer.Overhead()) - headerLen - payload.length
@ -550,6 +522,7 @@ func (p *packetPacker) writeAndSealPacketWithPadding(
}
if payloadSize := protocol.ByteCount(buffer.Len()-payloadOffset) - paddingLen; payloadSize != payload.length {
fmt.Printf("%#v\n", payload)
return nil, fmt.Errorf("PacketPacker BUG: payload size inconsistent (expected %d, got %d bytes)", payload.length, payloadSize)
}
if size := protocol.ByteCount(buffer.Len() + sealer.Overhead()); size > p.maxPacketSize {

376
packet_packer_test.go
View file

@ -8,7 +8,6 @@ import (
"time"
"github.com/golang/mock/gomock"
"github.com/lucas-clemente/quic-go/internal/ackhandler"
"github.com/lucas-clemente/quic-go/internal/handshake"
"github.com/lucas-clemente/quic-go/internal/mocks"
mockackhandler "github.com/lucas-clemente/quic-go/internal/mocks/ackhandler"
@ -20,21 +19,24 @@ import (
var _ = Describe("Packet packer", func() {
const maxPacketSize protocol.ByteCount = 1357
const version = protocol.VersionTLS
var (
packer *packetPacker
framer *MockFrameSource
ackFramer *MockAckFrameSource
initialStream *MockCryptoStream
handshakeStream *MockCryptoStream
sealingManager *MockSealingManager
pnManager *mockackhandler.MockSentPacketHandler
packer *packetPacker
retransmissionQueue *retransmissionQueue
framer *MockFrameSource
ackFramer *MockAckFrameSource
initialStream *MockCryptoStream
handshakeStream *MockCryptoStream
sealingManager *MockSealingManager
pnManager *mockackhandler.MockSentPacketHandler
)
checkLength := func(data []byte) {
hdr, _, _, err := wire.ParsePacket(data, 0)
Expect(err).ToNot(HaveOccurred())
r := bytes.NewReader(data)
extHdr, err := hdr.ParseExtended(r, protocol.VersionWhatever)
extHdr, err := hdr.ParseExtended(r, version)
Expect(err).ToNot(HaveOccurred())
ExpectWithOffset(1, extHdr.Length).To(BeEquivalentTo(r.Len() + int(extHdr.PacketNumberLen)))
}
@ -61,7 +63,7 @@ var _ = Describe("Packet packer", func() {
BeforeEach(func() {
rand.Seed(GinkgoRandomSeed())
version := protocol.VersionTLS
retransmissionQueue = newRetransmissionQueue(version)
mockSender := NewMockStreamSender(mockCtrl)
mockSender.EXPECT().onHasStreamData(gomock.Any()).AnyTimes()
initialStream = NewMockCryptoStream(mockCtrl)
@ -77,6 +79,7 @@ var _ = Describe("Packet packer", func() {
initialStream,
handshakeStream,
pnManager,
retransmissionQueue,
&net.TCPAddr{},
sealingManager,
framer,
@ -408,6 +411,60 @@ var _ = Describe("Packet packer", func() {
Expect(r.Len()).To(BeZero())
})
It("packs multiple small STREAM frames into single packet", func() {
f1 := &wire.StreamFrame{
StreamID: 5,
Data: []byte("frame 1"),
DataLenPresent: true,
}
f2 := &wire.StreamFrame{
StreamID: 5,
Data: []byte("frame 2"),
DataLenPresent: true,
}
f3 := &wire.StreamFrame{
StreamID: 3,
Data: []byte("frame 3"),
DataLenPresent: true,
}
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil)
ackFramer.EXPECT().GetAckFrame(protocol.Encryption1RTT)
expectAppendControlFrames()
expectAppendStreamFrames(f1, f2, f3)
p, err := packer.PackPacket()
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(p.frames).To(HaveLen(3))
Expect(p.frames[0].(*wire.StreamFrame).Data).To(Equal([]byte("frame 1")))
Expect(p.frames[1].(*wire.StreamFrame).Data).To(Equal([]byte("frame 2")))
Expect(p.frames[2].(*wire.StreamFrame).Data).To(Equal([]byte("frame 3")))
})
It("adds retransmissions", func() {
f1 := &wire.StreamFrame{Data: []byte("frame 1")}
cf := &wire.MaxDataFrame{ByteOffset: 0x42}
retransmissionQueue.AddAppData(f1)
retransmissionQueue.AddAppData(cf)
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil)
ackFramer.EXPECT().GetAckFrame(protocol.Encryption1RTT)
expectAppendControlFrames()
f2 := &wire.StreamFrame{Data: []byte("frame 2")}
expectAppendStreamFrames(f2)
p, err := packer.PackPacket()
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(p.frames).To(HaveLen(3))
Expect(p.frames[0]).To(BeAssignableToTypeOf(&wire.StreamFrame{}))
Expect(p.frames[0].(*wire.StreamFrame).Data).To(Equal([]byte("frame 1")))
Expect(p.frames[1]).To(Equal(cf))
Expect(p.frames[2]).To(BeAssignableToTypeOf(&wire.StreamFrame{}))
Expect(p.frames[2].(*wire.StreamFrame).Data).To(Equal([]byte("frame 2")))
})
Context("making ACK packets ack-eliciting", func() {
sendMaxNumNonAckElicitingAcks := func() {
for i := 0; i < protocol.MaxNonAckElicitingAcks; i++ {
@ -491,230 +548,6 @@ var _ = Describe("Packet packer", func() {
})
})
Context("STREAM frame handling", func() {
It("does not split a STREAM frame with maximum size", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
ackFramer.EXPECT().GetAckFrame(protocol.Encryption1RTT)
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil)
expectAppendControlFrames()
sf := &wire.StreamFrame{
Offset: 1,
StreamID: 5,
}
framer.EXPECT().AppendStreamFrames(gomock.Any(), gomock.Any()).DoAndReturn(func(_ []wire.Frame, maxSize protocol.ByteCount) ([]wire.Frame, protocol.ByteCount) {
sf.Data = bytes.Repeat([]byte{'f'}, int(maxSize-sf.Length(packer.version)))
return []wire.Frame{sf}, sf.Length(packer.version)
})
p, err := packer.PackPacket()
Expect(err).ToNot(HaveOccurred())
Expect(p.frames).To(HaveLen(1))
Expect(p.raw).To(HaveLen(int(maxPacketSize)))
Expect(p.frames[0].(*wire.StreamFrame).Data).To(HaveLen(len(sf.Data)))
Expect(p.frames[0].(*wire.StreamFrame).DataLenPresent).To(BeFalse())
})
It("packs multiple small STREAM frames into single packet", func() {
f1 := &wire.StreamFrame{
StreamID: 5,
Data: []byte("frame 1"),
DataLenPresent: true,
}
f2 := &wire.StreamFrame{
StreamID: 5,
Data: []byte("frame 2"),
DataLenPresent: true,
}
f3 := &wire.StreamFrame{
StreamID: 3,
Data: []byte("frame 3"),
DataLenPresent: true,
}
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil)
ackFramer.EXPECT().GetAckFrame(protocol.Encryption1RTT)
expectAppendControlFrames()
expectAppendStreamFrames(f1, f2, f3)
p, err := packer.PackPacket()
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(p.frames).To(HaveLen(3))
Expect(p.frames[0].(*wire.StreamFrame).Data).To(Equal([]byte("frame 1")))
Expect(p.frames[1].(*wire.StreamFrame).Data).To(Equal([]byte("frame 2")))
Expect(p.frames[2].(*wire.StreamFrame).Data).To(Equal([]byte("frame 3")))
})
})
Context("retransmissions", func() {
It("retransmits a small packet", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil)
frames := []wire.Frame{
&wire.MaxDataFrame{ByteOffset: 0x1234},
&wire.StreamFrame{StreamID: 42, Data: []byte("foobar")},
}
packets, err := packer.PackRetransmission(&ackhandler.Packet{
EncryptionLevel: protocol.Encryption1RTT,
Frames: frames,
})
Expect(err).ToNot(HaveOccurred())
Expect(packets).To(HaveLen(1))
p := packets[0]
Expect(p.EncryptionLevel()).To(Equal(protocol.Encryption1RTT))
Expect(p.frames).To(Equal(frames))
})
It("packs two packets for retransmission if the original packet contained many control frames", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2).Times(2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42)).Times(2)
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil).Times(2)
var frames []wire.Frame
var totalLen protocol.ByteCount
// pack a bunch of control frames, such that the packet is way bigger than a single packet
for i := 0; totalLen < maxPacketSize*3/2; i++ {
f := &wire.MaxStreamDataFrame{
StreamID: protocol.StreamID(i),
ByteOffset: protocol.ByteCount(i),
}
frames = append(frames, f)
totalLen += f.Length(packer.version)
}
packets, err := packer.PackRetransmission(&ackhandler.Packet{
EncryptionLevel: protocol.Encryption1RTT,
Frames: frames,
})
Expect(err).ToNot(HaveOccurred())
Expect(packets).To(HaveLen(2))
Expect(len(packets[0].frames) + len(packets[1].frames)).To(Equal(len(frames)))
Expect(packets[1].frames).To(Equal(frames[len(packets[0].frames):]))
// check that the first packet was filled up as far as possible:
// if the first frame (after the STOP_WAITING) was packed into the first packet, it would have overflown the MaxPacketSize
Expect(len(packets[0].raw) + int(packets[1].frames[1].Length(packer.version))).To(BeNumerically(">", maxPacketSize))
})
It("splits a STREAM frame that doesn't fit", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2).Times(2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42)).Times(2)
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil).Times(2)
packets, err := packer.PackRetransmission(&ackhandler.Packet{
EncryptionLevel: protocol.Encryption1RTT,
Frames: []wire.Frame{&wire.StreamFrame{
StreamID: 42,
Offset: 1337,
Data: bytes.Repeat([]byte{'a'}, int(maxPacketSize)*3/2),
}},
})
Expect(err).ToNot(HaveOccurred())
Expect(packets).To(HaveLen(2))
Expect(packets[0].frames[0]).To(BeAssignableToTypeOf(&wire.StreamFrame{}))
Expect(packets[1].frames[0]).To(BeAssignableToTypeOf(&wire.StreamFrame{}))
sf1 := packets[0].frames[0].(*wire.StreamFrame)
sf2 := packets[1].frames[0].(*wire.StreamFrame)
Expect(sf1.StreamID).To(Equal(protocol.StreamID(42)))
Expect(sf1.Offset).To(Equal(protocol.ByteCount(1337)))
Expect(sf1.DataLenPresent).To(BeFalse())
Expect(sf2.StreamID).To(Equal(protocol.StreamID(42)))
Expect(sf2.Offset).To(Equal(protocol.ByteCount(1337) + sf1.DataLen()))
Expect(sf2.DataLenPresent).To(BeFalse())
Expect(sf1.DataLen() + sf2.DataLen()).To(Equal(maxPacketSize * 3 / 2))
Expect(packets[0].raw).To(HaveLen(int(maxPacketSize)))
})
It("splits STREAM frames, if necessary", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2).AnyTimes()
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42)).AnyTimes()
for i := 0; i < 100; i++ {
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil).MaxTimes(2)
sf1 := &wire.StreamFrame{
StreamID: 42,
Offset: 1337,
Data: bytes.Repeat([]byte{'a'}, 1+int(rand.Int31n(int32(maxPacketSize*4/5)))),
}
sf2 := &wire.StreamFrame{
StreamID: 2,
Offset: 42,
Data: bytes.Repeat([]byte{'b'}, 1+int(rand.Int31n(int32(maxPacketSize*4/5)))),
}
expectedDataLen := sf1.DataLen() + sf2.DataLen()
frames := []wire.Frame{sf1, sf2}
packets, err := packer.PackRetransmission(&ackhandler.Packet{
EncryptionLevel: protocol.Encryption1RTT,
Frames: frames,
})
Expect(err).ToNot(HaveOccurred())
if len(packets) > 1 {
Expect(packets[0].raw).To(HaveLen(int(maxPacketSize)))
}
var dataLen protocol.ByteCount
for _, p := range packets {
for _, f := range p.frames {
dataLen += f.(*wire.StreamFrame).DataLen()
}
}
Expect(dataLen).To(Equal(expectedDataLen))
}
})
It("packs two packets for retransmission if the original packet contained many STREAM frames", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2).Times(2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42)).Times(2)
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil).Times(2)
var frames []wire.Frame
var totalLen protocol.ByteCount
// pack a bunch of control frames, such that the packet is way bigger than a single packet
for i := 0; totalLen < maxPacketSize*3/2; i++ {
f := &wire.StreamFrame{
StreamID: protocol.StreamID(i),
Data: []byte("foobar"),
DataLenPresent: true,
}
frames = append(frames, f)
totalLen += f.Length(packer.version)
}
packets, err := packer.PackRetransmission(&ackhandler.Packet{
EncryptionLevel: protocol.Encryption1RTT,
Frames: frames,
})
Expect(err).ToNot(HaveOccurred())
Expect(packets).To(HaveLen(2))
Expect(len(packets[0].frames) + len(packets[1].frames)).To(Equal(len(frames))) // all frames
Expect(packets[1].frames).To(Equal(frames[len(packets[0].frames):]))
// check that the first packet was filled up as far as possible:
// if the first frame was packed into the first packet, it would have overflown the MaxPacketSize
Expect(len(packets[0].raw) + int(packets[1].frames[1].Length(packer.version))).To(BeNumerically(">", maxPacketSize-protocol.MinStreamFrameSize))
})
It("correctly sets the DataLenPresent on STREAM frames", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().Get1RTTSealer().Return(sealer, nil)
frames := []wire.Frame{
&wire.StreamFrame{StreamID: 4, Data: []byte("foobar"), DataLenPresent: true},
&wire.StreamFrame{StreamID: 5, Data: []byte("barfoo")},
}
packets, err := packer.PackRetransmission(&ackhandler.Packet{
EncryptionLevel: protocol.Encryption1RTT,
Frames: frames,
})
Expect(err).ToNot(HaveOccurred())
Expect(packets).To(HaveLen(1))
p := packets[0]
Expect(p.frames).To(HaveLen(2))
Expect(p.frames[0]).To(BeAssignableToTypeOf(&wire.StreamFrame{}))
Expect(p.frames[1]).To(BeAssignableToTypeOf(&wire.StreamFrame{}))
sf1 := p.frames[0].(*wire.StreamFrame)
sf2 := p.frames[1].(*wire.StreamFrame)
Expect(sf1.StreamID).To(Equal(protocol.StreamID(4)))
Expect(sf1.DataLenPresent).To(BeTrue())
Expect(sf2.StreamID).To(Equal(protocol.StreamID(5)))
Expect(sf2.DataLenPresent).To(BeFalse())
})
})
Context("max packet size", func() {
It("sets the maximum packet size", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2).Times(2)
@ -805,8 +638,25 @@ var _ = Describe("Packet packer", func() {
p, err := packer.PackPacket()
Expect(err).ToNot(HaveOccurred())
Expect(p.frames).To(HaveLen(1))
expectedPacketLen := packer.maxPacketSize
Expect(p.raw).To(HaveLen(int(expectedPacketLen)))
Expect(p.raw).To(HaveLen(int(packer.maxPacketSize)))
Expect(p.header.IsLongHeader).To(BeTrue())
checkLength(p.raw)
})
It("adds retransmissions", func() {
f := &wire.CryptoFrame{Data: []byte("Initial")}
retransmissionQueue.AddInitial(f)
retransmissionQueue.AddHandshake(&wire.CryptoFrame{Data: []byte("Handshake")})
pnManager.EXPECT().PeekPacketNumber(protocol.EncryptionInitial).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.EncryptionInitial).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().GetInitialSealer().Return(sealer, nil)
sealingManager.EXPECT().GetHandshakeSealer().Return(nil, errors.New("no sealer"))
ackFramer.EXPECT().GetAckFrame(protocol.EncryptionInitial)
initialStream.EXPECT().HasData()
p, err := packer.PackPacket()
Expect(err).ToNot(HaveOccurred())
Expect(p.EncryptionLevel()).To(Equal(protocol.EncryptionInitial))
Expect(p.frames).To(Equal([]wire.Frame{f}))
Expect(p.header.IsLongHeader).To(BeTrue())
checkLength(p.raw)
})
@ -923,63 +773,25 @@ var _ = Describe("Packet packer", func() {
Expect(err).ToNot(HaveOccurred())
Expect(packet).ToNot(BeNil())
})
Context("retransmitions", func() {
cf := &wire.CryptoFrame{Data: []byte("foo")}
It("packs a retransmission with the right encryption level", func() {
pnManager.EXPECT().PeekPacketNumber(protocol.EncryptionInitial).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.EncryptionInitial).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().GetInitialSealer().Return(sealer, nil)
packet := &ackhandler.Packet{
EncryptionLevel: protocol.EncryptionInitial,
Frames: []wire.Frame{cf},
}
p, err := packer.PackRetransmission(packet)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(HaveLen(1))
Expect(p[0].header.Type).To(Equal(protocol.PacketTypeInitial))
Expect(p[0].frames).To(Equal([]wire.Frame{cf}))
Expect(p[0].EncryptionLevel()).To(Equal(protocol.EncryptionInitial))
})
It("packs a retransmission for an Initial packet", func() {
token := []byte("initial token")
packer.SetToken(token)
pnManager.EXPECT().PeekPacketNumber(protocol.EncryptionInitial).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
pnManager.EXPECT().PopPacketNumber(protocol.EncryptionInitial).Return(protocol.PacketNumber(0x42))
sealingManager.EXPECT().GetInitialSealer().Return(sealer, nil)
packer.perspective = protocol.PerspectiveClient
packet := &ackhandler.Packet{
EncryptionLevel: protocol.EncryptionInitial,
Frames: []wire.Frame{cf},
}
packets, err := packer.PackRetransmission(packet)
Expect(err).ToNot(HaveOccurred())
Expect(packets).To(HaveLen(1))
p := packets[0]
Expect(p.frames).To(Equal([]wire.Frame{cf}))
Expect(p.EncryptionLevel()).To(Equal(protocol.EncryptionInitial))
Expect(p.header.Type).To(Equal(protocol.PacketTypeInitial))
Expect(p.header.Token).To(Equal(token))
Expect(p.raw).To(HaveLen(protocol.MinInitialPacketSize))
})
})
})
})
})
var _ = Describe("Converting to AckHandler packets", func() {
It("convert a packet", func() {
f1 := &wire.MaxDataFrame{}
f2 := &wire.PingFrame{}
packet := &packedPacket{
header: &wire.ExtendedHeader{Header: wire.Header{}},
frames: []wire.Frame{&wire.MaxDataFrame{}, &wire.PingFrame{}},
frames: []wire.Frame{f1, f2},
ack: &wire.AckFrame{AckRanges: []wire.AckRange{{Largest: 100, Smallest: 80}}},
raw: []byte("foobar"),
}
p := packet.ToAckHandlerPacket()
p := packet.ToAckHandlerPacket(nil)
Expect(p.Length).To(Equal(protocol.ByteCount(6)))
Expect(p.Frames).To(Equal(packet.frames))
Expect(p.Frames).To(HaveLen(2))
Expect(p.Frames[0].Frame).To(Equal(f1))
Expect(p.Frames[1].Frame).To(Equal(f2))
Expect(p.LargestAcked).To(Equal(protocol.PacketNumber(100)))
Expect(p.SendTime).To(BeTemporally("~", time.Now(), 50*time.Millisecond))
})
@ -990,7 +802,7 @@ var _ = Describe("Converting to AckHandler packets", func() {
frames: []wire.Frame{&wire.MaxDataFrame{}, &wire.PingFrame{}},
raw: []byte("foobar"),
}
p := packet.ToAckHandlerPacket()
p := packet.ToAckHandlerPacket(nil)
Expect(p.LargestAcked).To(Equal(protocol.InvalidPacketNumber))
})
})

132
retransmission_queue.go Normal file
View file

@ -0,0 +1,132 @@
package quic
import (
"fmt"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
)
type retransmissionQueue struct {
initial []wire.Frame
initialCryptoData []*wire.CryptoFrame
handshake []wire.Frame
handshakeCryptoData []*wire.CryptoFrame
appData []wire.Frame
streamData []*wire.StreamFrame
version protocol.VersionNumber
}
func newRetransmissionQueue(ver protocol.VersionNumber) *retransmissionQueue {
return &retransmissionQueue{version: ver}
}
func (q *retransmissionQueue) AddInitial(f wire.Frame) {
if cf, ok := f.(*wire.CryptoFrame); ok {
q.initialCryptoData = append(q.initialCryptoData, cf)
return
}
q.initial = append(q.initial, f)
}
func (q *retransmissionQueue) AddHandshake(f wire.Frame) {
if cf, ok := f.(*wire.CryptoFrame); ok {
q.handshakeCryptoData = append(q.handshakeCryptoData, cf)
return
}
q.handshake = append(q.handshake, f)
}
func (q *retransmissionQueue) HasInitialData() bool {
return len(q.initialCryptoData) > 0 || len(q.initial) > 0
}
func (q *retransmissionQueue) HasHandshakeData() bool {
return len(q.handshakeCryptoData) > 0 || len(q.handshake) > 0
}
func (q *retransmissionQueue) AddAppData(f wire.Frame) {
if sf, ok := f.(*wire.StreamFrame); ok {
sf.DataLenPresent = true
q.streamData = append(q.streamData, sf)
return
}
q.appData = append(q.appData, f)
}
func (q *retransmissionQueue) GetInitialFrame(maxLen protocol.ByteCount) wire.Frame {
if len(q.initialCryptoData) > 0 {
if f := q.initialCryptoData[0]; f.Length(q.version) <= maxLen {
q.initialCryptoData = q.initialCryptoData[1:]
return f
}
}
if len(q.initial) == 0 {
return nil
}
f := q.initial[0]
if f.Length(q.version) > maxLen {
return nil
}
q.initial = q.initial[1:]
return f
}
func (q *retransmissionQueue) GetHandshakeFrame(maxLen protocol.ByteCount) wire.Frame {
if len(q.handshakeCryptoData) > 0 {
if f := q.handshakeCryptoData[0]; f.Length(q.version) <= maxLen {
q.handshakeCryptoData = q.handshakeCryptoData[1:]
return f
}
}
if len(q.handshake) == 0 {
return nil
}
f := q.handshake[0]
if f.Length(q.version) > maxLen {
return nil
}
q.handshake = q.handshake[1:]
return f
}
func (q *retransmissionQueue) GetAppDataFrame(maxLen protocol.ByteCount) wire.Frame {
if len(q.streamData) > 0 {
f := q.streamData[0]
if f.Length(q.version) <= maxLen {
q.streamData = q.streamData[1:]
return f
}
if maxLen >= protocol.MinStreamFrameSize {
newFrame, needsSplit := f.MaybeSplitOffFrame(maxLen, q.version)
if needsSplit && newFrame != nil {
return newFrame
}
}
}
if len(q.appData) == 0 {
return nil
}
f := q.appData[0]
if f.Length(q.version) > maxLen {
return nil
}
q.appData = q.appData[1:]
return f
}
func (q *retransmissionQueue) DropPackets(encLevel protocol.EncryptionLevel) {
switch encLevel {
case protocol.EncryptionInitial:
q.initial = nil
q.initialCryptoData = nil
case protocol.EncryptionHandshake:
q.handshake = nil
q.handshakeCryptoData = nil
default:
panic(fmt.Sprintf("unexpected encryption level: %s", encLevel))
}
}

158
retransmission_queue_test.go Normal file
View file

@ -0,0 +1,158 @@
package quic
import (
"math/rand"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
var _ = Describe("Retransmission queue", func() {
const version = protocol.VersionTLS
var q *retransmissionQueue
BeforeEach(func() {
q = newRetransmissionQueue(version)
})
Context("Initial data", func() {
It("doesn't dequeue anything when it's empty", func() {
Expect(q.HasInitialData()).To(BeFalse())
Expect(q.GetInitialFrame(protocol.MaxByteCount)).To(BeNil())
})
It("queues and retrieves a control frame", func() {
f := &wire.MaxDataFrame{ByteOffset: 0x42}
q.AddInitial(f)
Expect(q.HasInitialData()).To(BeTrue())
Expect(q.GetInitialFrame(f.Length(version) - 1)).To(BeNil())
Expect(q.GetInitialFrame(f.Length(version))).To(Equal(f))
Expect(q.HasInitialData()).To(BeFalse())
})
It("queues and retrieves a CRYPTO frame", func() {
f := &wire.CryptoFrame{Data: []byte("foobar")}
q.AddInitial(f)
Expect(q.HasInitialData()).To(BeTrue())
Expect(q.GetInitialFrame(f.Length(version) - 1)).To(BeNil())
Expect(q.GetInitialFrame(f.Length(version))).To(Equal(f))
Expect(q.HasInitialData()).To(BeFalse())
})
It("retrieves both a CRYPTO frame and a control frame", func() {
cf := &wire.MaxDataFrame{ByteOffset: 0x42}
f := &wire.CryptoFrame{Data: []byte("foobar")}
q.AddInitial(f)
q.AddInitial(cf)
Expect(q.HasInitialData()).To(BeTrue())
Expect(q.GetInitialFrame(protocol.MaxByteCount)).To(Equal(f))
Expect(q.GetInitialFrame(protocol.MaxByteCount)).To(Equal(cf))
Expect(q.HasInitialData()).To(BeFalse())
})
It("drops all Initial frames", func() {
q.AddInitial(&wire.CryptoFrame{Data: []byte("foobar")})
q.AddInitial(&wire.MaxDataFrame{ByteOffset: 0x42})
q.DropPackets(protocol.EncryptionInitial)
Expect(q.HasInitialData()).To(BeFalse())
Expect(q.GetInitialFrame(protocol.MaxByteCount)).To(BeNil())
})
})
Context("Handshake data", func() {
It("doesn't dequeue anything when it's empty", func() {
Expect(q.HasHandshakeData()).To(BeFalse())
Expect(q.GetHandshakeFrame(protocol.MaxByteCount)).To(BeNil())
})
It("queues and retrieves a control frame", func() {
f := &wire.MaxDataFrame{ByteOffset: 0x42}
q.AddHandshake(f)
Expect(q.HasHandshakeData()).To(BeTrue())
Expect(q.GetHandshakeFrame(f.Length(version) - 1)).To(BeNil())
Expect(q.GetHandshakeFrame(f.Length(version))).To(Equal(f))
Expect(q.HasHandshakeData()).To(BeFalse())
})
It("queues and retrieves a CRYPTO frame", func() {
f := &wire.CryptoFrame{Data: []byte("foobar")}
q.AddHandshake(f)
Expect(q.HasHandshakeData()).To(BeTrue())
Expect(q.GetHandshakeFrame(f.Length(version) - 1)).To(BeNil())
Expect(q.GetHandshakeFrame(f.Length(version))).To(Equal(f))
Expect(q.HasHandshakeData()).To(BeFalse())
})
It("retrieves both a CRYPTO frame and a control frame", func() {
cf := &wire.MaxDataFrame{ByteOffset: 0x42}
f := &wire.CryptoFrame{Data: []byte("foobar")}
q.AddHandshake(f)
q.AddHandshake(cf)
Expect(q.HasHandshakeData()).To(BeTrue())
Expect(q.GetHandshakeFrame(protocol.MaxByteCount)).To(Equal(f))
Expect(q.GetHandshakeFrame(protocol.MaxByteCount)).To(Equal(cf))
Expect(q.HasHandshakeData()).To(BeFalse())
})
It("drops all Initial frames", func() {
q.AddHandshake(&wire.CryptoFrame{Data: []byte("foobar")})
q.AddHandshake(&wire.MaxDataFrame{ByteOffset: 0x42})
q.DropPackets(protocol.EncryptionHandshake)
Expect(q.HasHandshakeData()).To(BeFalse())
Expect(q.GetHandshakeFrame(protocol.MaxByteCount)).To(BeNil())
})
})
Context("Application data", func() {
It("doesn't dequeue anything when it's empty", func() {
Expect(q.GetAppDataFrame(protocol.MaxByteCount)).To(BeNil())
})
It("queues and retrieves a control frame", func() {
f := &wire.MaxDataFrame{ByteOffset: 0x42}
q.AddAppData(f)
Expect(q.GetAppDataFrame(f.Length(version) - 1)).To(BeNil())
Expect(q.GetAppDataFrame(f.Length(version))).To(Equal(f))
})
It("queues and retrieves a STREAM frame", func() {
f := &wire.StreamFrame{Data: []byte("foobar")}
q.AddAppData(f)
Expect(q.GetAppDataFrame(f.Length(version) - 1)).To(BeNil())
Expect(q.GetAppDataFrame(f.Length(version))).To(Equal(f))
})
It("splits STREAM frames larger than MinStreamFrameSize", func() {
data := make([]byte, 1000)
rand.Read(data)
f := &wire.StreamFrame{
Data: data,
FinBit: true,
}
q.AddAppData(f)
Expect(q.GetAppDataFrame(protocol.MinStreamFrameSize - 1)).To(BeNil())
f1 := q.GetAppDataFrame(protocol.MinStreamFrameSize).(*wire.StreamFrame)
Expect(f1).ToNot(BeNil())
Expect(f1.Length(version)).To(Equal(protocol.MinStreamFrameSize))
Expect(f1.FinBit).To(BeFalse())
Expect(f1.Data).To(Equal(data[:f1.DataLen()]))
f2 := q.GetAppDataFrame(protocol.MaxByteCount).(*wire.StreamFrame)
Expect(f2).ToNot(BeNil())
Expect(f2.FinBit).To(BeTrue())
Expect(f1.DataLen() + f2.DataLen()).To(BeEquivalentTo(1000))
Expect(f2.Data).To(Equal(data[f1.DataLen():]))
Expect(q.GetAppDataFrame(protocol.MaxByteCount)).To(BeNil())
})
It("returns a control frame if it doesn't split a STREAM frame", func() {
cf := &wire.MaxDataFrame{ByteOffset: 0x42}
q.AddAppData(&wire.StreamFrame{Data: make([]byte, 1000)})
q.AddAppData(cf)
Expect(q.GetAppDataFrame(protocol.MinStreamFrameSize - 1)).To(Equal(cf))
})
})
})

76
session.go
View file

@ -115,6 +115,7 @@ type session struct {
cryptoStreamManager *cryptoStreamManager
sentPacketHandler ackhandler.SentPacketHandler
receivedPacketHandler ackhandler.ReceivedPacketHandler
retransmissionQueue *retransmissionQueue
framer framer
windowUpdateQueue *windowUpdateQueue
connFlowController flowcontrol.ConnectionFlowController
@ -242,6 +243,7 @@ var newSession = func(
initialStream,
handshakeStream,
s.sentPacketHandler,
s.retransmissionQueue,
s.RemoteAddr(),
cs,
s.framer,
@ -328,6 +330,7 @@ var newClientSession = func(
initialStream,
handshakeStream,
s.sentPacketHandler,
s.retransmissionQueue,
s.RemoteAddr(),
cs,
s.framer,
@ -350,6 +353,7 @@ var newClientSession = func(
func (s *session) preSetup() {
s.sendQueue = newSendQueue(s.conn)
s.retransmissionQueue = newRetransmissionQueue(s.version)
s.frameParser = wire.NewFrameParser(s.version)
s.rttStats = &congestion.RTTStats{}
s.receivedPacketHandler = ackhandler.NewReceivedPacketHandler(s.rttStats, s.logger, s.version)
@ -1109,16 +1113,6 @@ sendLoop:
return err
}
numPacketsSent++
case ackhandler.SendRetransmission:
sentPacket, err := s.maybeSendRetransmission()
if err != nil {
return err
}
if sentPacket {
numPacketsSent++
// This can happen if a retransmission queued, but it wasn't necessary to send it.
// e.g. when an Initial is queued, but we already received a packet from the server.
}
case ackhandler.SendAny:
sentPacket, err := s.sendPacket()
if err != nil {
@ -1152,55 +1146,29 @@ func (s *session) maybeSendAckOnlyPacket() error {
if packet == nil {
return nil
}
s.sentPacketHandler.SentPacket(packet.ToAckHandlerPacket())
s.sentPacketHandler.SentPacket(packet.ToAckHandlerPacket(s.retransmissionQueue))
s.sendQueue.Send(packet)
return nil
}
// maybeSendRetransmission sends retransmissions for at most one packet.
// It takes care that Initials aren't retransmitted, if a packet from the server was already received.
func (s *session) maybeSendRetransmission() (bool, error) {
retransmitPacket := s.sentPacketHandler.DequeuePacketForRetransmission()
if retransmitPacket == nil {
return false, nil
}
s.logger.Debugf("Dequeueing retransmission for packet 0x%x (%s)", retransmitPacket.PacketNumber, retransmitPacket.EncryptionLevel)
packets, err := s.packer.PackRetransmission(retransmitPacket)
if err != nil {
return false, err
}
ackhandlerPackets := make([]*ackhandler.Packet, len(packets))
for i, packet := range packets {
ackhandlerPackets[i] = packet.ToAckHandlerPacket()
}
s.sentPacketHandler.SentPacketsAsRetransmission(ackhandlerPackets, retransmitPacket.PacketNumber)
for _, packet := range packets {
s.sendPackedPacket(packet)
}
return true, nil
}
func (s *session) sendProbePacket() error {
p, err := s.sentPacketHandler.DequeueProbePacket()
if err != nil {
return err
// Queue probe packets until we actually send out a packet.
for {
if wasQueued := s.sentPacketHandler.QueueProbePacket(); !wasQueued {
break
}
sent, err := s.sendPacket()
if err != nil {
return err
}
if sent {
return nil
}
}
s.logger.Debugf("Sending a retransmission for %#x as a probe packet.", p.PacketNumber)
packets, err := s.packer.PackRetransmission(p)
if err != nil {
return err
}
ackhandlerPackets := make([]*ackhandler.Packet, len(packets))
for i, packet := range packets {
ackhandlerPackets[i] = packet.ToAckHandlerPacket()
}
s.sentPacketHandler.SentPacketsAsRetransmission(ackhandlerPackets, p.PacketNumber)
for _, packet := range packets {
s.sendPackedPacket(packet)
}
return nil
// If there is nothing else to queue, make sure we send out something.
s.framer.QueueControlFrame(&wire.PingFrame{})
_, err := s.sendPacket()
return err
}
func (s *session) sendPacket() (bool, error) {
@ -1213,7 +1181,7 @@ func (s *session) sendPacket() (bool, error) {
if err != nil || packet == nil {
return false, err
}
s.sentPacketHandler.SentPacket(packet.ToAckHandlerPacket())
s.sentPacketHandler.SentPacket(packet.ToAckHandlerPacket(s.retransmissionQueue))
s.sendPackedPacket(packet)
return true, nil
}

85
session_test.go
View file

@ -892,78 +892,40 @@ var _ = Describe("Session", func() {
Expect(frames).To(Equal([]wire.Frame{&wire.DataBlockedFrame{DataLimit: 1337}}))
})
It("sends a retransmission and a regular packet in the same run", func() {
packetToRetransmit := &ackhandler.Packet{PacketNumber: 10}
retransmittedPacket := getPacket(123)
newPacket := getPacket(234)
sess.windowUpdateQueue.callback(&wire.MaxDataFrame{})
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLossDetectionTimeout().AnyTimes()
sph.EXPECT().DequeuePacketForRetransmission().Return(packetToRetransmit)
sph.EXPECT().SendMode().Return(ackhandler.SendRetransmission)
sph.EXPECT().SendMode().Return(ackhandler.SendAny)
sph.EXPECT().ShouldSendNumPackets().Return(2)
sph.EXPECT().TimeUntilSend()
gomock.InOrder(
packer.EXPECT().PackRetransmission(packetToRetransmit).Return([]*packedPacket{retransmittedPacket}, nil),
sph.EXPECT().SentPacketsAsRetransmission(gomock.Any(), protocol.PacketNumber(10)).Do(func(packets []*ackhandler.Packet, _ protocol.PacketNumber) {
Expect(packets).To(HaveLen(1))
Expect(packets[0].PacketNumber).To(Equal(protocol.PacketNumber(123)))
}),
packer.EXPECT().PackPacket().Return(newPacket, nil),
sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
Expect(p.PacketNumber).To(Equal(protocol.PacketNumber(234)))
}),
)
sess.sentPacketHandler = sph
Expect(sess.sendPackets()).To(Succeed())
Eventually(mconn.written).Should(HaveLen(2))
})
It("sends multiple packets, if the retransmission is split", func() {
packet := &ackhandler.Packet{
PacketNumber: 42,
Frames: []wire.Frame{&wire.StreamFrame{
StreamID: 0x5,
Data: []byte("foobar"),
}},
EncryptionLevel: protocol.Encryption1RTT,
}
retransmissions := []*packedPacket{getPacket(1337), getPacket(1338)}
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLossDetectionTimeout().AnyTimes()
sph.EXPECT().DequeuePacketForRetransmission().Return(packet)
packer.EXPECT().PackRetransmission(packet).Return(retransmissions, nil)
sph.EXPECT().SentPacketsAsRetransmission(gomock.Any(), protocol.PacketNumber(42)).Do(func(packets []*ackhandler.Packet, _ protocol.PacketNumber) {
Expect(packets).To(HaveLen(2))
Expect(packets[0].PacketNumber).To(Equal(protocol.PacketNumber(1337)))
Expect(packets[1].PacketNumber).To(Equal(protocol.PacketNumber(1338)))
})
sess.sentPacketHandler = sph
sent, err := sess.maybeSendRetransmission()
Expect(err).NotTo(HaveOccurred())
Expect(sent).To(BeTrue())
Eventually(mconn.written).Should(HaveLen(2))
})
It("sends a probe packet", func() {
packetToRetransmit := &ackhandler.Packet{PacketNumber: 0x42}
retransmittedPacket := getPacket(123)
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLossDetectionTimeout().AnyTimes()
sph.EXPECT().TimeUntilSend()
sph.EXPECT().SendMode().Return(ackhandler.SendPTO)
sph.EXPECT().ShouldSendNumPackets().Return(1)
sph.EXPECT().DequeueProbePacket().Return(packetToRetransmit, nil)
packer.EXPECT().PackRetransmission(packetToRetransmit).Return([]*packedPacket{retransmittedPacket}, nil)
sph.EXPECT().SentPacketsAsRetransmission(gomock.Any(), protocol.PacketNumber(0x42)).Do(func(packets []*ackhandler.Packet, _ protocol.PacketNumber) {
Expect(packets).To(HaveLen(1))
Expect(packets[0].PacketNumber).To(Equal(protocol.PacketNumber(123)))
sph.EXPECT().QueueProbePacket()
packer.EXPECT().PackPacket().Return(getPacket(123), nil)
sph.EXPECT().SentPacket(gomock.Any()).Do(func(packet *ackhandler.Packet) {
Expect(packet.PacketNumber).To(Equal(protocol.PacketNumber(123)))
})
sess.sentPacketHandler = sph
Expect(sess.sendPackets()).To(Succeed())
})
It("sends a PING as a probe packet", func() {
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLossDetectionTimeout().AnyTimes()
sph.EXPECT().TimeUntilSend()
sph.EXPECT().SendMode().Return(ackhandler.SendPTO)
sph.EXPECT().ShouldSendNumPackets().Return(1)
sph.EXPECT().QueueProbePacket().Return(false)
packer.EXPECT().PackPacket().Return(getPacket(123), nil)
sph.EXPECT().SentPacket(gomock.Any()).Do(func(packet *ackhandler.Packet) {
Expect(packet.PacketNumber).To(Equal(protocol.PacketNumber(123)))
})
sess.sentPacketHandler = sph
Expect(sess.sendPackets()).To(Succeed())
// We're using a mock packet packer in this test.
// We therefore need to test separately that the PING was actually queued.
frames, _ := sess.framer.AppendControlFrames(nil, protocol.MaxByteCount)
Expect(frames).To(Equal([]wire.Frame{&wire.PingFrame{}}))
})
It("doesn't send when the SentPacketHandler doesn't allow it", func() {
sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLossDetectionTimeout().AnyTimes()
@ -980,7 +942,6 @@ var _ = Describe("Session", func() {
BeforeEach(func() {
sph = mockackhandler.NewMockSentPacketHandler(mockCtrl)
sph.EXPECT().GetLossDetectionTimeout().AnyTimes()
sph.EXPECT().DequeuePacketForRetransmission().AnyTimes()
sess.sentPacketHandler = sph
streamManager.EXPECT().CloseWithError(gomock.Any())
})