Merge pull request #1285 from lucas-clemente/rto-probe-packets

only allow sending of retransmissions if these are RTO probe packets

internal/ackhandler/send_mode.go

@@ -12,6 +12,8 @@ const (
 	SendAck
 	// SendRetransmission means that retransmissions should be sent
 	SendRetransmission
+	// SendRTO means that an RTO probe packet should be sent
+	SendRTO
 	// SendAny packet should be sent
 	SendAny
 )
@@ -24,6 +26,8 @@ func (s SendMode) String() string {
 		return "ack"
 	case SendRetransmission:
 		return "retransmission"
+	case SendRTO:
+		return "rto"
 	case SendAny:
 		return "any"
 	default:

internal/ackhandler/send_mode_test.go

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

internal/ackhandler/sent_packet_handler.go

@@ -61,6 +61,8 @@ type sentPacketHandler struct {
 
 	// The number of times an RTO has been sent without receiving an ack.
 	rtoCount uint32
+	// The number of RTO probe packets that should be sent.
+	numRTOs int
 
 	// The time at which the next packet will be considered lost based on early transmit or exceeding the reordering window in time.
 	lossTime time.Time
@@ -163,6 +165,9 @@ func (h *sentPacketHandler) sentPacketImpl(packet *Packet) bool /* isRetransmitt
 		packet.includedInBytesInFlight = true
 		h.bytesInFlight += packet.Length
 		packet.canBeRetransmitted = true
+		if h.numRTOs > 0 {
+			h.numRTOs--
+		}
 	}
 	h.congestion.OnPacketSent(packet.SendTime, h.bytesInFlight, packet.PacketNumber, packet.Length, isRetransmittable)
 
@@ -346,6 +351,7 @@ func (h *sentPacketHandler) OnAlarm() error {
 	} else {
 		// RTO
 		h.rtoCount++
+		h.numRTOs += 2
 		err = h.queueRTOs()
 	}
 	if err != nil {
@@ -460,15 +466,18 @@ func (h *sentPacketHandler) SendMode() SendMode {
 		h.logger.Debugf("Limited by the number of tracked packets: tracking %d packets, maximum %d", numTrackedPackets, protocol.MaxTrackedSentPackets)
 		return SendNone
 	}
-	// Send retransmissions first, if there are any.
-	if len(h.retransmissionQueue) > 0 {
-		return SendRetransmission
+	if h.numRTOs > 0 {
+		return SendRTO
 	}
 	// Only send ACKs if we're congestion limited.
 	if cwnd := h.congestion.GetCongestionWindow(); h.bytesInFlight > cwnd {
 		h.logger.Debugf("Congestion limited: bytes in flight %d, window %d", h.bytesInFlight, cwnd)
 		return SendAck
 	}
+	// Send retransmissions first, if there are any.
+	if len(h.retransmissionQueue) > 0 {
+		return SendRetransmission
+	}
 	if numTrackedPackets >= protocol.MaxOutstandingSentPackets {
 		h.logger.Debugf("Max outstanding limited: tracking %d packets, maximum: %d", numTrackedPackets, protocol.MaxOutstandingSentPackets)
 		return SendAck
@@ -481,6 +490,10 @@ func (h *sentPacketHandler) TimeUntilSend() time.Time {
 }
 
 func (h *sentPacketHandler) ShouldSendNumPackets() int {
+	if h.numRTOs > 0 {
+		// RTO probes should not be paced, but must be sent immediately.
+		return h.numRTOs
+	}
 	delay := h.congestion.TimeUntilSend(h.bytesInFlight)
 	if delay == 0 || delay > protocol.MinPacingDelay {
 		return 1

internal/ackhandler/sent_packet_handler_test.go

@@ -668,14 +668,21 @@ var _ = Describe("SentPacketHandler", func() {
 			Expect(handler.SendMode()).To(Equal(SendAck))
 		})
 
+		It("doesn't allow retransmission if congestion limited", func() {
+			handler.bytesInFlight = 100
+			handler.retransmissionQueue = []*Packet{{PacketNumber: 3}}
+			cong.EXPECT().GetCongestionWindow().Return(protocol.ByteCount(50))
+			Expect(handler.SendMode()).To(Equal(SendAck))
+		})
+
 		It("allows sending retransmissions", func() {
-			// note that we don't EXPECT a call to GetCongestionWindow
-			// that means retransmissions are sent without considering the congestion window
+			cong.EXPECT().GetCongestionWindow().Return(protocol.MaxByteCount)
 			handler.retransmissionQueue = []*Packet{{PacketNumber: 3}}
 			Expect(handler.SendMode()).To(Equal(SendRetransmission))
 		})
 
 		It("allow retransmissions, if we're keeping track of between MaxOutstandingSentPackets and MaxTrackedSentPackets packets", func() {
+			cong.EXPECT().GetCongestionWindow().Return(protocol.MaxByteCount)
 			Expect(protocol.MaxOutstandingSentPackets).To(BeNumerically("<", protocol.MaxTrackedSentPackets))
 			handler.retransmissionQueue = make([]*Packet, protocol.MaxOutstandingSentPackets+10)
 			Expect(handler.SendMode()).To(Equal(SendRetransmission))
@@ -683,6 +690,14 @@ var _ = Describe("SentPacketHandler", func() {
 			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.numRTOs = 1
+			handler.retransmissionQueue = []*Packet{{PacketNumber: 3}}
+			Expect(handler.SendMode()).To(Equal(SendRTO))
+		})
+
 		It("gets the pacing delay", func() {
 			sendTime := time.Now().Add(-time.Minute)
 			handler.bytesInFlight = 100
@@ -692,6 +707,11 @@ var _ = Describe("SentPacketHandler", func() {
 			Expect(handler.TimeUntilSend()).To(Equal(sendTime.Add(time.Hour)))
 		})
 
+		It("allows sending of all RTO probe packets", func() {
+			handler.numRTOs = 5
+			Expect(handler.ShouldSendNumPackets()).To(Equal(5))
+		})
+
 		It("allows sending of one packet, if it should be sent immediately", func() {
 			cong.EXPECT().TimeUntilSend(gomock.Any()).Return(time.Duration(0))
 			Expect(handler.ShouldSendNumPackets()).To(Equal(1))
@@ -786,6 +806,16 @@ var _ = Describe("SentPacketHandler", func() {
 			Expect(handler.retransmissionQueue).To(BeEmpty()) // 1 and 2 were already sent as probe packets
 		})
 
+		It("allows sending of two probe packets", func() {
+			handler.SentPacket(retransmittablePacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
+			handler.OnAlarm()
+			Expect(handler.SendMode()).To(Equal(SendRTO))
+			handler.SentPacket(retransmittablePacket(&Packet{PacketNumber: 2}))
+			Expect(handler.SendMode()).To(Equal(SendRTO))
+			handler.SentPacket(retransmittablePacket(&Packet{PacketNumber: 3}))
+			Expect(handler.SendMode()).ToNot(Equal(SendRTO))
+		})
+
 		It("queues packets sent before the probe packet for retransmission", func() {
 			handler.SentPacket(retransmittablePacket(&Packet{PacketNumber: 1, SendTime: time.Now().Add(-time.Hour)}))
 			handler.SentPacket(retransmittablePacket(&Packet{PacketNumber: 2, SendTime: time.Now().Add(-time.Hour)}))

session.go

@@ -859,6 +859,25 @@ sendLoop:
 			// There will only be a new ACK after receiving new packets.
 			// SendAck is only returned when we're congestion limited, so we don't need to set the pacingt timer.
 			return s.maybeSendAckOnlyPacket()
+		case ackhandler.SendRTO:
+			// try to send a retransmission first
+			sentPacket, err := s.maybeSendRetransmission()
+			if err != nil {
+				return err
+			}
+			if !sentPacket {
+				// In RTO mode, a probe packet has to be sent.
+				// Add a PING frame to make sure a (retransmittable) packet will be sent.
+				s.queueControlFrame(&wire.PingFrame{})
+				sentPacket, err := s.sendPacket()
+				if err != nil {
+					return err
+				}
+				if !sentPacket {
+					return errors.New("session BUG: expected a packet to be sent in RTO mode")
+				}
+			}
+			numPacketsSent++
 		case ackhandler.SendRetransmission:
 			sentPacket, err := s.maybeSendRetransmission()
 			if err != nil {

session_test.go

@@ -791,6 +791,51 @@ var _ = Describe("Session", func() {
 			Expect(err).ToNot(HaveOccurred())
 		})
 
+		It("sends an RTO probe packets", func() {
+			sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
+			sph.EXPECT().GetPacketNumberLen(gomock.Any()).Return(protocol.PacketNumberLen2).AnyTimes()
+			sph.EXPECT().TimeUntilSend()
+			sph.EXPECT().DequeuePacketForRetransmission().Return(&ackhandler.Packet{
+				PacketNumber: 10,
+			})
+			sph.EXPECT().DequeuePacketForRetransmission().Return(&ackhandler.Packet{
+				PacketNumber: 11,
+			})
+			sph.EXPECT().SendMode().Return(ackhandler.SendRTO).Times(2)
+			sph.EXPECT().ShouldSendNumPackets().Return(2)
+			sph.EXPECT().GetStopWaitingFrame(gomock.Any()).Return(&wire.StopWaitingFrame{}).Times(2)
+			gomock.InOrder(
+				sph.EXPECT().SentPacketsAsRetransmission(gomock.Any(), protocol.PacketNumber(10)),
+				sph.EXPECT().SentPacketsAsRetransmission(gomock.Any(), protocol.PacketNumber(11)),
+			)
+			sess.sentPacketHandler = sph
+			err := sess.sendPackets()
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("sends RTO probe packets with new data, if no retransmission is available", func() {
+			sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
+			sph.EXPECT().GetPacketNumberLen(gomock.Any()).Return(protocol.PacketNumberLen2).AnyTimes()
+			sph.EXPECT().TimeUntilSend()
+			sph.EXPECT().DequeuePacketForRetransmission().Return(&ackhandler.Packet{
+				PacketNumber: 10,
+			})
+			sph.EXPECT().DequeuePacketForRetransmission()
+			sph.EXPECT().SendMode().Return(ackhandler.SendRTO).Times(2)
+			sph.EXPECT().ShouldSendNumPackets().Return(2)
+			sph.EXPECT().GetStopWaitingFrame(gomock.Any()).Return(&wire.StopWaitingFrame{})
+			gomock.InOrder(
+				sph.EXPECT().SentPacketsAsRetransmission(gomock.Any(), protocol.PacketNumber(10)),
+				sph.EXPECT().SentPacket(gomock.Any()).Do(func(p *ackhandler.Packet) {
+					Expect(p.Frames).To(HaveLen(1))
+					Expect(p.Frames[0]).To(BeAssignableToTypeOf(&wire.PingFrame{}))
+				}),
+			)
+			sess.sentPacketHandler = sph
+			err := sess.sendPackets()
+			Expect(err).ToNot(HaveOccurred())
+		})
+
 		It("doesn't send when the SentPacketHandler doesn't allow it", func() {
 			sph := mockackhandler.NewMockSentPacketHandler(mockCtrl)
 			sph.EXPECT().SendMode().Return(ackhandler.SendNone)