package ackhandler
import (
"errors"
"fmt"
"time"
"github.com/quic-go/quic-go/internal/congestion"
"github.com/quic-go/quic-go/internal/monotime"
"github.com/quic-go/quic-go/internal/protocol"
"github.com/quic-go/quic-go/internal/qerr"
"github.com/quic-go/quic-go/internal/utils"
"github.com/quic-go/quic-go/internal/wire"
"github.com/quic-go/quic-go/qlog"
"github.com/quic-go/quic-go/qlogwriter"
)
const (
timeThreshold = 9.0 / 8
packetThreshold = 3
amplificationFactor = 3
minRTTAfterRetry = 5 * time .Millisecond
maxPTODuration = 60 * time .Second
)
const pathProbePacketLossTimeout = time .Second
type packetNumberSpace struct {
history sentPacketHistory
pns packetNumberGenerator
lossTime monotime .Time
lastAckElicitingPacketTime monotime .Time
largestAcked protocol .PacketNumber
largestSent protocol .PacketNumber
}
func newPacketNumberSpace(initialPN protocol .PacketNumber , isAppData bool ) *packetNumberSpace {
var pns packetNumberGenerator
if isAppData {
pns = newSkippingPacketNumberGenerator (initialPN , protocol .SkipPacketInitialPeriod , protocol .SkipPacketMaxPeriod )
} else {
pns = newSequentialPacketNumberGenerator (initialPN )
}
return &packetNumberSpace {
history : *newSentPacketHistory (isAppData ),
pns : pns ,
largestSent : protocol .InvalidPacketNumber ,
largestAcked : protocol .InvalidPacketNumber ,
}
}
type alarmTimer struct {
Time monotime .Time
TimerType qlog .TimerType
EncryptionLevel protocol .EncryptionLevel
}
type sentPacketHandler struct {
initialPackets *packetNumberSpace
handshakePackets *packetNumberSpace
appDataPackets *packetNumberSpace
lostPackets lostPacketTracker
largestAckedTime monotime .Time
peerCompletedAddressValidation bool
bytesReceived protocol .ByteCount
bytesSent protocol .ByteCount
peerAddressValidated bool
handshakeConfirmed bool
ignorePacketsBelow func (protocol .PacketNumber )
ackedPackets []packetWithPacketNumber
bytesInFlight protocol .ByteCount
congestion congestion .SendAlgorithmWithDebugInfos
rttStats *utils .RTTStats
connStats *utils .ConnectionStats
ptoCount uint32
ptoMode SendMode
numProbesToSend int
alarm alarmTimer
enableECN bool
ecnTracker ecnHandler
perspective protocol .Perspective
qlogger qlogwriter .Recorder
lastMetrics qlog .MetricsUpdated
logger utils .Logger
}
var _ SentPacketHandler = &sentPacketHandler {}
func NewSentPacketHandler (
initialPN protocol .PacketNumber ,
initialMaxDatagramSize protocol .ByteCount ,
rttStats *utils .RTTStats ,
connStats *utils .ConnectionStats ,
clientAddressValidated bool ,
enableECN bool ,
ignorePacketsBelow func (protocol .PacketNumber ),
pers protocol .Perspective ,
qlogger qlogwriter .Recorder ,
logger utils .Logger ,
) SentPacketHandler {
congestion := congestion .NewCubicSender (
congestion .DefaultClock {},
rttStats ,
connStats ,
initialMaxDatagramSize ,
true ,
qlogger ,
)
h := &sentPacketHandler {
peerCompletedAddressValidation : pers == protocol .PerspectiveServer ,
peerAddressValidated : pers == protocol .PerspectiveClient || clientAddressValidated ,
initialPackets : newPacketNumberSpace (initialPN , false ),
handshakePackets : newPacketNumberSpace (0 , false ),
appDataPackets : newPacketNumberSpace (0 , true ),
lostPackets : *newLostPacketTracker (64 ),
rttStats : rttStats ,
connStats : connStats ,
congestion : congestion ,
ignorePacketsBelow : ignorePacketsBelow ,
perspective : pers ,
qlogger : qlogger ,
logger : logger ,
}
if enableECN {
h .enableECN = true
h .ecnTracker = newECNTracker (logger , qlogger )
}
return h
}
func (h *sentPacketHandler ) removeFromBytesInFlight (p *packet ) {
if p .includedInBytesInFlight {
if p .Length > h .bytesInFlight {
panic ("negative bytes_in_flight" )
}
h .bytesInFlight -= p .Length
p .includedInBytesInFlight = false
}
}
func (h *sentPacketHandler ) DropPackets (encLevel protocol .EncryptionLevel , now monotime .Time ) {
if h .perspective == protocol .PerspectiveClient && encLevel == protocol .EncryptionHandshake {
h .peerCompletedAddressValidation = true
}
if encLevel == protocol .EncryptionInitial || encLevel == protocol .EncryptionHandshake {
pnSpace := h .getPacketNumberSpace (encLevel )
if pnSpace == nil {
return
}
for _ , p := range pnSpace .history .Packets () {
h .removeFromBytesInFlight (p )
}
}
switch encLevel {
case protocol .EncryptionInitial :
h .initialPackets = nil
case protocol .EncryptionHandshake :
h .handshakeConfirmed = true
h .handshakePackets = nil
case protocol .Encryption0RTT :
for pn , p := range h .appDataPackets .history .Packets () {
if p .EncryptionLevel != protocol .Encryption0RTT {
break
}
h .removeFromBytesInFlight (p )
h .appDataPackets .history .Remove (pn )
}
default :
panic (fmt .Sprintf ("Cannot drop keys for encryption level %s" , encLevel ))
}
if h .qlogger != nil && h .ptoCount != 0 {
h .qlogger .RecordEvent (qlog .PTOCountUpdated {PTOCount : 0 })
}
h .ptoCount = 0
h .numProbesToSend = 0
h .ptoMode = SendNone
h .setLossDetectionTimer (now )
}
func (h *sentPacketHandler ) ReceivedBytes (n protocol .ByteCount , t monotime .Time ) {
h .connStats .BytesReceived .Add (uint64 (n ))
wasAmplificationLimit := h .isAmplificationLimited ()
h .bytesReceived += n
if wasAmplificationLimit && !h .isAmplificationLimited () {
h .setLossDetectionTimer (t )
}
}
func (h *sentPacketHandler ) ReceivedPacket (l protocol .EncryptionLevel , t monotime .Time ) {
h .connStats .PacketsReceived .Add (1 )
if h .perspective == protocol .PerspectiveServer && l == protocol .EncryptionHandshake && !h .peerAddressValidated {
h .peerAddressValidated = true
h .setLossDetectionTimer (t )
}
}
func (h *sentPacketHandler ) packetsInFlight () int {
packetsInFlight := h .appDataPackets .history .NumOutstanding ()
if h .handshakePackets != nil {
packetsInFlight += h .handshakePackets .history .NumOutstanding ()
}
if h .initialPackets != nil {
packetsInFlight += h .initialPackets .history .NumOutstanding ()
}
return packetsInFlight
}
func (h *sentPacketHandler ) SentPacket (
t monotime .Time ,
pn , largestAcked protocol .PacketNumber ,
streamFrames []StreamFrame ,
frames []Frame ,
encLevel protocol .EncryptionLevel ,
ecn protocol .ECN ,
size protocol .ByteCount ,
isPathMTUProbePacket bool ,
isPathProbePacket bool ,
) {
h .bytesSent += size
h .connStats .BytesSent .Add (uint64 (size ))
h .connStats .PacketsSent .Add (1 )
pnSpace := h .getPacketNumberSpace (encLevel )
if h .logger .Debug () && (pnSpace .history .HasOutstandingPackets () || pnSpace .history .HasOutstandingPathProbes ()) {
for p := max (0 , pnSpace .largestSent +1 ); p < pn ; p ++ {
h .logger .Debugf ("Skipping packet number %d" , p )
}
}
pnSpace .largestSent = pn
p := getPacket ()
p .SendTime = t
p .EncryptionLevel = encLevel
p .Length = size
p .Frames = frames
p .LargestAcked = largestAcked
p .StreamFrames = streamFrames
p .IsPathMTUProbePacket = isPathMTUProbePacket
p .isPathProbePacket = isPathProbePacket
isAckEliciting := p .IsAckEliciting ()
if isPathProbePacket {
pnSpace .history .SentPathProbePacket (pn , p )
h .setLossDetectionTimer (t )
return
}
if isAckEliciting {
pnSpace .lastAckElicitingPacketTime = t
h .bytesInFlight += size
p .includedInBytesInFlight = true
if h .numProbesToSend > 0 {
h .numProbesToSend --
}
}
h .congestion .OnPacketSent (t , h .bytesInFlight , pn , size , isAckEliciting )
if encLevel == protocol .Encryption1RTT && h .ecnTracker != nil {
h .ecnTracker .SentPacket (pn , ecn )
}
pnSpace .history .SentPacket (pn , p )
if !isAckEliciting {
if !h .peerCompletedAddressValidation {
h .setLossDetectionTimer (t )
}
return
}
if h .qlogger != nil {
h .qlogMetricsUpdated ()
}
h .setLossDetectionTimer (t )
}
func (h *sentPacketHandler ) qlogMetricsUpdated () {
var metricsUpdatedEvent qlog .MetricsUpdated
var updated bool
if h .rttStats .HasMeasurement () {
if h .lastMetrics .MinRTT != h .rttStats .MinRTT () {
metricsUpdatedEvent .MinRTT = h .rttStats .MinRTT ()
h .lastMetrics .MinRTT = metricsUpdatedEvent .MinRTT
updated = true
}
if h .lastMetrics .SmoothedRTT != h .rttStats .SmoothedRTT () {
metricsUpdatedEvent .SmoothedRTT = h .rttStats .SmoothedRTT ()
h .lastMetrics .SmoothedRTT = metricsUpdatedEvent .SmoothedRTT
updated = true
}
if h .lastMetrics .LatestRTT != h .rttStats .LatestRTT () {
metricsUpdatedEvent .LatestRTT = h .rttStats .LatestRTT ()
h .lastMetrics .LatestRTT = metricsUpdatedEvent .LatestRTT
updated = true
}
if h .lastMetrics .RTTVariance != h .rttStats .MeanDeviation () {
metricsUpdatedEvent .RTTVariance = h .rttStats .MeanDeviation ()
h .lastMetrics .RTTVariance = metricsUpdatedEvent .RTTVariance
updated = true
}
}
if cwnd := h .congestion .GetCongestionWindow (); h .lastMetrics .CongestionWindow != int (cwnd ) {
metricsUpdatedEvent .CongestionWindow = int (cwnd )
h .lastMetrics .CongestionWindow = metricsUpdatedEvent .CongestionWindow
updated = true
}
if h .lastMetrics .BytesInFlight != int (h .bytesInFlight ) {
metricsUpdatedEvent .BytesInFlight = int (h .bytesInFlight )
h .lastMetrics .BytesInFlight = metricsUpdatedEvent .BytesInFlight
updated = true
}
packetsInFlight := h .packetsInFlight ()
if h .lastMetrics .PacketsInFlight != packetsInFlight {
metricsUpdatedEvent .PacketsInFlight = packetsInFlight
h .lastMetrics .PacketsInFlight = metricsUpdatedEvent .PacketsInFlight
updated = true
}
if updated {
h .qlogger .RecordEvent (metricsUpdatedEvent )
}
}
func (h *sentPacketHandler ) getPacketNumberSpace (encLevel protocol .EncryptionLevel ) *packetNumberSpace {
switch encLevel {
case protocol .EncryptionInitial :
return h .initialPackets
case protocol .EncryptionHandshake :
return h .handshakePackets
case protocol .Encryption0RTT , protocol .Encryption1RTT :
return h .appDataPackets
default :
panic ("invalid packet number space" )
}
}
func (h *sentPacketHandler ) ReceivedAck (ack *wire .AckFrame , encLevel protocol .EncryptionLevel , rcvTime monotime .Time ) (bool , error ) {
pnSpace := h .getPacketNumberSpace (encLevel )
largestAcked := ack .LargestAcked ()
if largestAcked > pnSpace .largestSent {
return false , &qerr .TransportError {
ErrorCode : qerr .ProtocolViolation ,
ErrorMessage : "received ACK for an unsent packet" ,
}
}
if h .perspective == protocol .PerspectiveClient && !h .peerCompletedAddressValidation &&
(encLevel == protocol .EncryptionHandshake || encLevel == protocol .Encryption1RTT ) {
h .peerCompletedAddressValidation = true
h .logger .Debugf ("Peer doesn't await address validation any longer." )
h .setLossDetectionTimer (rcvTime )
}
priorInFlight := h .bytesInFlight
ackedPackets , hasAckEliciting , err := h .detectAndRemoveAckedPackets (ack , encLevel )
if err != nil || len (ackedPackets ) == 0 {
return false , err
}
if len (ackedPackets ) > 0 {
if p := ackedPackets [len (ackedPackets )-1 ]; p .PacketNumber == ack .LargestAcked () && !p .isPathProbePacket && hasAckEliciting {
var ackDelay time .Duration
if encLevel == protocol .Encryption1RTT {
ackDelay = min (ack .DelayTime , h .rttStats .MaxAckDelay ())
}
if h .largestAckedTime .IsZero () || !p .SendTime .Before (h .largestAckedTime ) {
h .rttStats .UpdateRTT (rcvTime .Sub (p .SendTime ), ackDelay )
if h .logger .Debug () {
h .logger .Debugf ("\tupdated RTT: %s (σ: %s)" , h .rttStats .SmoothedRTT (), h .rttStats .MeanDeviation ())
}
h .largestAckedTime = p .SendTime
}
h .congestion .MaybeExitSlowStart ()
}
}
if encLevel == protocol .Encryption1RTT && h .ecnTracker != nil && largestAcked > pnSpace .largestAcked {
congested := h .ecnTracker .HandleNewlyAcked (ackedPackets , int64 (ack .ECT0 ), int64 (ack .ECT1 ), int64 (ack .ECNCE ))
if congested {
h .congestion .OnCongestionEvent (largestAcked , 0 , priorInFlight )
}
}
pnSpace .largestAcked = max (pnSpace .largestAcked , largestAcked )
h .detectLostPackets (rcvTime , encLevel )
if encLevel == protocol .Encryption1RTT {
h .detectLostPathProbes (rcvTime )
}
var acked1RTTPacket bool
for _ , p := range ackedPackets {
if p .includedInBytesInFlight {
h .congestion .OnPacketAcked (p .PacketNumber , p .Length , priorInFlight , rcvTime )
}
if p .EncryptionLevel == protocol .Encryption1RTT {
acked1RTTPacket = true
}
h .removeFromBytesInFlight (p .packet )
if !p .isPathProbePacket {
putPacket (p .packet )
}
}
if encLevel == protocol .Encryption1RTT && largestAcked == pnSpace .largestAcked {
h .detectSpuriousLosses (
ack ,
rcvTime .Add (-min (ack .DelayTime , h .rttStats .MaxAckDelay ())),
)
h .lostPackets .DeleteBefore (rcvTime .Add (-3 * h .rttStats .PTO (false )))
}
ackedPackets = nil
clear (h .ackedPackets )
h .ackedPackets = h .ackedPackets [:0 ]
if h .peerCompletedAddressValidation {
if h .qlogger != nil && h .ptoCount != 0 {
h .qlogger .RecordEvent (qlog .PTOCountUpdated {PTOCount : 0 })
}
h .ptoCount = 0
}
h .numProbesToSend = 0
if h .qlogger != nil {
h .qlogMetricsUpdated ()
}
h .setLossDetectionTimer (rcvTime )
return acked1RTTPacket , nil
}
func (h *sentPacketHandler ) detectSpuriousLosses (ack *wire .AckFrame , ackTime monotime .Time ) {
var maxPacketReordering protocol .PacketNumber
var maxTimeReordering time .Duration
ackRangeIdx := len (ack .AckRanges ) - 1
var spuriousLosses []protocol .PacketNumber
for pn , sendTime := range h .lostPackets .All () {
ackRange := ack .AckRanges [ackRangeIdx ]
for pn > ackRange .Largest {
if ackRangeIdx == 0 {
break
}
ackRangeIdx --
ackRange = ack .AckRanges [ackRangeIdx ]
}
if pn < ackRange .Smallest {
continue
}
if pn <= ackRange .Largest {
packetReordering := h .appDataPackets .history .Difference (ack .LargestAcked (), pn )
timeReordering := ackTime .Sub (sendTime )
maxPacketReordering = max (maxPacketReordering , packetReordering )
maxTimeReordering = max (maxTimeReordering , timeReordering )
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .SpuriousLoss {
EncryptionLevel : protocol .Encryption1RTT ,
PacketNumber : pn ,
PacketReordering : uint64 (packetReordering ),
TimeReordering : timeReordering ,
})
}
spuriousLosses = append (spuriousLosses , pn )
}
}
for _ , pn := range spuriousLosses {
h .lostPackets .Delete (pn )
}
}
func (h *sentPacketHandler ) detectAndRemoveAckedPackets (
ack *wire .AckFrame ,
encLevel protocol .EncryptionLevel ,
) (_ []packetWithPacketNumber , hasAckEliciting bool , _ error ) {
if len (h .ackedPackets ) > 0 {
return nil , false , errors .New ("ackhandler BUG: ackedPackets slice not empty" )
}
pnSpace := h .getPacketNumberSpace (encLevel )
if encLevel == protocol .Encryption1RTT {
for p := range pnSpace .history .SkippedPackets () {
if ack .AcksPacket (p ) {
return nil , false , &qerr .TransportError {
ErrorCode : qerr .ProtocolViolation ,
ErrorMessage : fmt .Sprintf ("received an ACK for skipped packet number: %d (%s)" , p , encLevel ),
}
}
}
}
var ackRangeIndex int
lowestAcked := ack .LowestAcked ()
largestAcked := ack .LargestAcked ()
for pn , p := range pnSpace .history .Packets () {
if pn < lowestAcked {
continue
}
if pn > largestAcked {
break
}
if ack .HasMissingRanges () {
ackRange := ack .AckRanges [len (ack .AckRanges )-1 -ackRangeIndex ]
for pn > ackRange .Largest && ackRangeIndex < len (ack .AckRanges )-1 {
ackRangeIndex ++
ackRange = ack .AckRanges [len (ack .AckRanges )-1 -ackRangeIndex ]
}
if pn < ackRange .Smallest {
continue
}
if pn > ackRange .Largest {
return nil , false , fmt .Errorf ("BUG: ackhandler would have acked wrong packet %d, while evaluating range %d -> %d" , pn , ackRange .Smallest , ackRange .Largest )
}
}
if p .isPathProbePacket {
probePacket := pnSpace .history .RemovePathProbe (pn )
if probePacket != nil {
h .ackedPackets = append (h .ackedPackets , packetWithPacketNumber {PacketNumber : pn , packet : probePacket })
}
continue
}
if p .IsAckEliciting () {
hasAckEliciting = true
}
h .ackedPackets = append (h .ackedPackets , packetWithPacketNumber {PacketNumber : pn , packet : p })
}
if h .logger .Debug () && len (h .ackedPackets ) > 0 {
pns := make ([]protocol .PacketNumber , len (h .ackedPackets ))
for i , p := range h .ackedPackets {
pns [i ] = p .PacketNumber
}
h .logger .Debugf ("\tnewly acked packets (%d): %d" , len (pns ), pns )
}
for _ , p := range h .ackedPackets {
if p .LargestAcked != protocol .InvalidPacketNumber && encLevel == protocol .Encryption1RTT && h .ignorePacketsBelow != nil {
h .ignorePacketsBelow (p .LargestAcked + 1 )
}
for _ , f := range p .Frames {
if f .Handler != nil {
f .Handler .OnAcked (f .Frame )
}
}
for _ , f := range p .StreamFrames {
if f .Handler != nil {
f .Handler .OnAcked (f .Frame )
}
}
if err := pnSpace .history .Remove (p .PacketNumber ); err != nil {
return nil , false , err
}
}
return h .ackedPackets , hasAckEliciting , nil
}
func (h *sentPacketHandler ) getLossTimeAndSpace () (monotime .Time , protocol .EncryptionLevel ) {
var encLevel protocol .EncryptionLevel
var lossTime monotime .Time
if h .initialPackets != nil {
lossTime = h .initialPackets .lossTime
encLevel = protocol .EncryptionInitial
}
if h .handshakePackets != nil && (lossTime .IsZero () || (!h .handshakePackets .lossTime .IsZero () && h .handshakePackets .lossTime .Before (lossTime ))) {
lossTime = h .handshakePackets .lossTime
encLevel = protocol .EncryptionHandshake
}
if lossTime .IsZero () || (!h .appDataPackets .lossTime .IsZero () && h .appDataPackets .lossTime .Before (lossTime )) {
lossTime = h .appDataPackets .lossTime
encLevel = protocol .Encryption1RTT
}
return lossTime , encLevel
}
func (h *sentPacketHandler ) getScaledPTO (includeMaxAckDelay bool ) time .Duration {
pto := h .rttStats .PTO (includeMaxAckDelay ) << h .ptoCount
if pto > maxPTODuration || pto <= 0 {
return maxPTODuration
}
return pto
}
func (h *sentPacketHandler ) getPTOTimeAndSpace (now monotime .Time ) (pto monotime .Time , encLevel protocol .EncryptionLevel ) {
if !h .handshakeConfirmed && !h .hasOutstandingCryptoPackets () {
if h .peerCompletedAddressValidation {
return
}
t := now .Add (h .getScaledPTO (false ))
if h .initialPackets != nil {
return t , protocol .EncryptionInitial
}
return t , protocol .EncryptionHandshake
}
if h .initialPackets != nil && h .initialPackets .history .HasOutstandingPackets () &&
!h .initialPackets .lastAckElicitingPacketTime .IsZero () {
encLevel = protocol .EncryptionInitial
if t := h .initialPackets .lastAckElicitingPacketTime ; !t .IsZero () {
pto = t .Add (h .getScaledPTO (false ))
}
}
if h .handshakePackets != nil && h .handshakePackets .history .HasOutstandingPackets () &&
!h .handshakePackets .lastAckElicitingPacketTime .IsZero () {
t := h .handshakePackets .lastAckElicitingPacketTime .Add (h .getScaledPTO (false ))
if pto .IsZero () || (!t .IsZero () && t .Before (pto )) {
pto = t
encLevel = protocol .EncryptionHandshake
}
}
if h .handshakeConfirmed && h .appDataPackets .history .HasOutstandingPackets () &&
!h .appDataPackets .lastAckElicitingPacketTime .IsZero () {
t := h .appDataPackets .lastAckElicitingPacketTime .Add (h .getScaledPTO (true ))
if pto .IsZero () || (!t .IsZero () && t .Before (pto )) {
pto = t
encLevel = protocol .Encryption1RTT
}
}
return pto , encLevel
}
func (h *sentPacketHandler ) hasOutstandingCryptoPackets () bool {
if h .initialPackets != nil && h .initialPackets .history .HasOutstandingPackets () {
return true
}
if h .handshakePackets != nil && h .handshakePackets .history .HasOutstandingPackets () {
return true
}
return false
}
func (h *sentPacketHandler ) setLossDetectionTimer (now monotime .Time ) {
oldAlarm := h .alarm
newAlarm := h .lossDetectionTime (now )
h .alarm = newAlarm
hasAlarm := !newAlarm .Time .IsZero ()
if !hasAlarm && !oldAlarm .Time .IsZero () {
h .logger .Debugf ("Canceling loss detection timer." )
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .LossTimerUpdated {
Type : qlog .LossTimerUpdateTypeCancelled ,
})
}
}
if h .qlogger != nil && hasAlarm && newAlarm != oldAlarm {
h .qlogger .RecordEvent (qlog .LossTimerUpdated {
Type : qlog .LossTimerUpdateTypeSet ,
TimerType : newAlarm .TimerType ,
EncLevel : newAlarm .EncryptionLevel ,
Time : newAlarm .Time .ToTime (),
})
}
}
func (h *sentPacketHandler ) lossDetectionTime (now monotime .Time ) alarmTimer {
if h .peerCompletedAddressValidation && !h .hasOutstandingCryptoPackets () &&
!h .appDataPackets .history .HasOutstandingPackets () && !h .appDataPackets .history .HasOutstandingPathProbes () {
return alarmTimer {}
}
if h .isAmplificationLimited () {
return alarmTimer {}
}
var pathProbeLossTime monotime .Time
if h .appDataPackets .history .HasOutstandingPathProbes () {
if _ , p := h .appDataPackets .history .FirstOutstandingPathProbe (); p != nil {
pathProbeLossTime = p .SendTime .Add (pathProbePacketLossTimeout )
}
}
lossTime , encLevel := h .getLossTimeAndSpace ()
if !lossTime .IsZero () && (pathProbeLossTime .IsZero () || lossTime .Before (pathProbeLossTime )) {
return alarmTimer {
Time : lossTime ,
TimerType : qlog .TimerTypeACK ,
EncryptionLevel : encLevel ,
}
}
ptoTime , encLevel := h .getPTOTimeAndSpace (now )
if !ptoTime .IsZero () && (pathProbeLossTime .IsZero () || ptoTime .Before (pathProbeLossTime )) {
return alarmTimer {
Time : ptoTime ,
TimerType : qlog .TimerTypePTO ,
EncryptionLevel : encLevel ,
}
}
if !pathProbeLossTime .IsZero () {
return alarmTimer {
Time : pathProbeLossTime ,
TimerType : qlog .TimerTypePathProbe ,
EncryptionLevel : protocol .Encryption1RTT ,
}
}
return alarmTimer {}
}
func (h *sentPacketHandler ) detectLostPathProbes (now monotime .Time ) {
if !h .appDataPackets .history .HasOutstandingPathProbes () {
return
}
lossTime := now .Add (-pathProbePacketLossTimeout )
var lostPathProbes []packetWithPacketNumber
for pn , p := range h .appDataPackets .history .PathProbes () {
if !p .SendTime .After (lossTime ) {
lostPathProbes = append (lostPathProbes , packetWithPacketNumber {PacketNumber : pn , packet : p })
}
}
for _ , p := range lostPathProbes {
for _ , f := range p .Frames {
f .Handler .OnLost (f .Frame )
}
h .appDataPackets .history .RemovePathProbe (p .PacketNumber )
}
}
func (h *sentPacketHandler ) detectLostPackets (now monotime .Time , encLevel protocol .EncryptionLevel ) {
pnSpace := h .getPacketNumberSpace (encLevel )
pnSpace .lossTime = 0
maxRTT := float64 (max (h .rttStats .LatestRTT (), h .rttStats .SmoothedRTT ()))
lossDelay := time .Duration (timeThreshold * maxRTT )
lossDelay = max (lossDelay , protocol .TimerGranularity )
lostSendTime := now .Add (-lossDelay )
priorInFlight := h .bytesInFlight
for pn , p := range pnSpace .history .Packets () {
if pn > pnSpace .largestAcked {
break
}
var packetLost bool
if !p .SendTime .After (lostSendTime ) {
packetLost = true
if !p .isPathProbePacket && p .IsAckEliciting () {
if h .logger .Debug () {
h .logger .Debugf ("\tlost packet %d (time threshold)" , pn )
}
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .PacketLost {
Header : qlog .PacketHeader {
PacketType : qlog .EncryptionLevelToPacketType (p .EncryptionLevel ),
PacketNumber : pn ,
},
Trigger : qlog .PacketLossTimeThreshold ,
})
}
}
} else if pnSpace .history .Difference (pnSpace .largestAcked , pn ) >= packetThreshold {
packetLost = true
if !p .isPathProbePacket && p .IsAckEliciting () {
if h .logger .Debug () {
h .logger .Debugf ("\tlost packet %d (reordering threshold)" , pn )
}
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .PacketLost {
Header : qlog .PacketHeader {
PacketType : qlog .EncryptionLevelToPacketType (p .EncryptionLevel ),
PacketNumber : pn ,
},
Trigger : qlog .PacketLossReorderingThreshold ,
})
}
}
} else if pnSpace .lossTime .IsZero () {
lossTime := p .SendTime .Add (lossDelay )
if h .logger .Debug () {
h .logger .Debugf ("\tsetting loss timer for packet %d (%s) to %s (in %s)" , pn , encLevel , lossDelay , lossTime )
}
pnSpace .lossTime = lossTime
}
if packetLost {
if encLevel == protocol .Encryption0RTT || encLevel == protocol .Encryption1RTT {
h .lostPackets .Add (pn , p .SendTime )
}
pnSpace .history .DeclareLost (pn )
if !p .isPathProbePacket && p .IsAckEliciting () {
h .removeFromBytesInFlight (p )
h .queueFramesForRetransmission (p )
if !p .IsPathMTUProbePacket {
h .congestion .OnCongestionEvent (pn , p .Length , priorInFlight )
}
if encLevel == protocol .Encryption1RTT && h .ecnTracker != nil {
h .ecnTracker .LostPacket (pn )
}
}
}
}
}
func (h *sentPacketHandler ) OnLossDetectionTimeout (now monotime .Time ) error {
defer h .setLossDetectionTimer (now )
if h .handshakeConfirmed {
h .detectLostPathProbes (now )
}
earliestLossTime , encLevel := h .getLossTimeAndSpace ()
if !earliestLossTime .IsZero () {
if h .logger .Debug () {
h .logger .Debugf ("Loss detection alarm fired in loss timer mode. Loss time: %s" , earliestLossTime )
}
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .LossTimerUpdated {
Type : qlog .LossTimerUpdateTypeExpired ,
TimerType : qlog .TimerTypeACK ,
EncLevel : encLevel ,
})
}
h .detectLostPackets (now , encLevel )
return nil
}
if h .bytesInFlight == 0 && !h .peerCompletedAddressValidation {
h .ptoCount ++
h .numProbesToSend ++
if h .initialPackets != nil {
h .ptoMode = SendPTOInitial
} else if h .handshakePackets != nil {
h .ptoMode = SendPTOHandshake
} else {
return errors .New ("sentPacketHandler BUG: PTO fired, but bytes_in_flight is 0 and Initial and Handshake already dropped" )
}
return nil
}
ptoTime , encLevel := h .getPTOTimeAndSpace (now )
if ptoTime .IsZero () {
return nil
}
ps := h .getPacketNumberSpace (encLevel )
if !ps .history .HasOutstandingPackets () && !ps .history .HasOutstandingPathProbes () && !h .peerCompletedAddressValidation {
return nil
}
h .ptoCount ++
if h .logger .Debug () {
h .logger .Debugf ("Loss detection alarm for %s fired in PTO mode. PTO count: %d" , encLevel , h .ptoCount )
}
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .LossTimerUpdated {
Type : qlog .LossTimerUpdateTypeExpired ,
TimerType : qlog .TimerTypePTO ,
EncLevel : encLevel ,
})
h .qlogger .RecordEvent (qlog .PTOCountUpdated {PTOCount : h .ptoCount })
}
h .numProbesToSend += 2
switch encLevel {
case protocol .EncryptionInitial :
h .ptoMode = SendPTOInitial
case protocol .EncryptionHandshake :
h .ptoMode = SendPTOHandshake
case protocol .Encryption1RTT :
pn := h .PopPacketNumber (protocol .Encryption1RTT )
h .getPacketNumberSpace (protocol .Encryption1RTT ).history .SkippedPacket (pn )
h .ptoMode = SendPTOAppData
default :
return fmt .Errorf ("PTO timer in unexpected encryption level: %s" , encLevel )
}
return nil
}
func (h *sentPacketHandler ) GetLossDetectionTimeout () monotime .Time {
return h .alarm .Time
}
func (h *sentPacketHandler ) ECNMode (isShortHeaderPacket bool ) protocol .ECN {
if !h .enableECN {
return protocol .ECNUnsupported
}
if !isShortHeaderPacket {
return protocol .ECNNon
}
return h .ecnTracker .Mode ()
}
func (h *sentPacketHandler ) PeekPacketNumber (encLevel protocol .EncryptionLevel ) (protocol .PacketNumber , protocol .PacketNumberLen ) {
pnSpace := h .getPacketNumberSpace (encLevel )
pn := pnSpace .pns .Peek ()
return pn , protocol .PacketNumberLengthForHeader (pn , pnSpace .largestAcked )
}
func (h *sentPacketHandler ) PopPacketNumber (encLevel protocol .EncryptionLevel ) protocol .PacketNumber {
pnSpace := h .getPacketNumberSpace (encLevel )
skipped , pn := pnSpace .pns .Pop ()
if skipped {
skippedPN := pn - 1
pnSpace .history .SkippedPacket (skippedPN )
if h .logger .Debug () {
h .logger .Debugf ("Skipping packet number %d" , skippedPN )
}
}
return pn
}
func (h *sentPacketHandler ) SendMode (now monotime .Time ) SendMode {
numTrackedPackets := h .appDataPackets .history .Len ()
if h .initialPackets != nil {
numTrackedPackets += h .initialPackets .history .Len ()
}
if h .handshakePackets != nil {
numTrackedPackets += h .handshakePackets .history .Len ()
}
if h .isAmplificationLimited () {
h .logger .Debugf ("Amplification window limited. Received %d bytes, already sent out %d bytes" , h .bytesReceived , h .bytesSent )
return SendNone
}
if numTrackedPackets >= protocol .MaxTrackedSentPackets {
if h .logger .Debug () {
h .logger .Debugf ("Limited by the number of tracked packets: tracking %d packets, maximum %d" , numTrackedPackets , protocol .MaxTrackedSentPackets )
}
return SendNone
}
if h .numProbesToSend > 0 {
return h .ptoMode
}
if !h .congestion .CanSend (h .bytesInFlight ) {
if h .logger .Debug () {
h .logger .Debugf ("Congestion limited: bytes in flight %d, window %d" , h .bytesInFlight , h .congestion .GetCongestionWindow ())
}
return SendAck
}
if numTrackedPackets >= protocol .MaxOutstandingSentPackets {
if h .logger .Debug () {
h .logger .Debugf ("Max outstanding limited: tracking %d packets, maximum: %d" , numTrackedPackets , protocol .MaxOutstandingSentPackets )
}
return SendAck
}
if !h .congestion .HasPacingBudget (now ) {
return SendPacingLimited
}
return SendAny
}
func (h *sentPacketHandler ) TimeUntilSend () monotime .Time {
return h .congestion .TimeUntilSend (h .bytesInFlight )
}
func (h *sentPacketHandler ) SetMaxDatagramSize (s protocol .ByteCount ) {
h .congestion .SetMaxDatagramSize (s )
}
func (h *sentPacketHandler ) isAmplificationLimited () bool {
if h .peerAddressValidated {
return false
}
return h .bytesSent >= amplificationFactor *h .bytesReceived
}
func (h *sentPacketHandler ) QueueProbePacket (encLevel protocol .EncryptionLevel ) bool {
pnSpace := h .getPacketNumberSpace (encLevel )
pn , p := pnSpace .history .FirstOutstanding ()
if p == nil {
return false
}
pnSpace .history .DeclareLost (pn )
h .removeFromBytesInFlight (p )
h .queueFramesForRetransmission (p )
return true
}
func (h *sentPacketHandler ) queueFramesForRetransmission (p *packet ) {
if len (p .Frames ) == 0 && len (p .StreamFrames ) == 0 {
panic ("no frames" )
}
for _ , f := range p .Frames {
if f .Handler != nil {
f .Handler .OnLost (f .Frame )
}
}
for _ , f := range p .StreamFrames {
if f .Handler != nil {
f .Handler .OnLost (f .Frame )
}
}
p .StreamFrames = nil
p .Frames = nil
}
func (h *sentPacketHandler ) ResetForRetry (now monotime .Time ) {
h .bytesInFlight = 0
var firstPacketSendTime monotime .Time
for _ , p := range h .initialPackets .history .Packets () {
if firstPacketSendTime .IsZero () {
firstPacketSendTime = p .SendTime
}
if p .IsAckEliciting () {
h .queueFramesForRetransmission (p )
}
}
for _ , p := range h .appDataPackets .history .Packets () {
if p .IsAckEliciting () {
h .queueFramesForRetransmission (p )
}
}
if h .ptoCount == 0 {
h .rttStats .UpdateRTT (max (minRTTAfterRetry , now .Sub (firstPacketSendTime )), 0 )
if h .logger .Debug () {
h .logger .Debugf ("\tupdated RTT: %s (σ: %s)" , h .rttStats .SmoothedRTT (), h .rttStats .MeanDeviation ())
}
if h .qlogger != nil {
h .qlogMetricsUpdated ()
}
}
h .initialPackets = newPacketNumberSpace (h .initialPackets .pns .Peek (), false )
h .appDataPackets = newPacketNumberSpace (h .appDataPackets .pns .Peek (), true )
oldAlarm := h .alarm
h .alarm = alarmTimer {}
if h .qlogger != nil {
h .qlogger .RecordEvent (qlog .PTOCountUpdated {PTOCount : 0 })
if !oldAlarm .Time .IsZero () {
h .qlogger .RecordEvent (qlog .LossTimerUpdated {
Type : qlog .LossTimerUpdateTypeCancelled ,
})
}
}
h .ptoCount = 0
}
func (h *sentPacketHandler ) MigratedPath (now monotime .Time , initialMaxDatagramSize protocol .ByteCount ) {
h .rttStats .ResetForPathMigration ()
for pn , p := range h .appDataPackets .history .Packets () {
h .appDataPackets .history .DeclareLost (pn )
if !p .isPathProbePacket {
h .removeFromBytesInFlight (p )
if p .IsAckEliciting () {
h .queueFramesForRetransmission (p )
}
}
}
for pn := range h .appDataPackets .history .PathProbes () {
h .appDataPackets .history .RemovePathProbe (pn )
}
h .congestion = congestion .NewCubicSender (
congestion .DefaultClock {},
h .rttStats ,
h .connStats ,
initialMaxDatagramSize ,
true ,
h .qlogger ,
)
h .setLossDetectionTimer (now )
}
The pages are generated with Golds v0.8.4 . (GOOS=linux GOARCH=amd64)
Golds is a Go 101 project developed by Tapir Liu .
PR and bug reports are welcome and can be submitted to the issue list .
Please follow @zigo_101 (reachable from the left QR code) to get the latest news of Golds .