don't generate new packets when the send queue is full

send_queue.go

@@ -3,35 +3,53 @@ package quic
 type sender interface {
 	Send(p *packetBuffer)
 	Run() error
+	WouldBlock() bool
+	Available() <-chan struct{}
 	Close()
 }
 
 type sendQueue struct {
 	queue       chan *packetBuffer
 	closeCalled chan struct{} // runStopped when Close() is called
 	runStopped  chan struct{} // runStopped when the run loop returns
+	available   chan struct{}
 	conn        sendConn
 }
 
 var _ sender = &sendQueue{}
 
+const sendQueueCapacity = 1
+
 func newSendQueue(conn sendConn) sender {
-	s := &sendQueue{
+	return &sendQueue{
 		conn:        conn,
 		runStopped:  make(chan struct{}),
 		closeCalled: make(chan struct{}),
-		queue:       make(chan *packetBuffer, 1),
+		available:   make(chan struct{}, 1),
+		queue:       make(chan *packetBuffer, sendQueueCapacity),
 	}
-	return s
 }
 
+// Send sends out a packet. It's guaranteed to not block.
+// Callers need to make sure that there's actually space in the send queue by calling WouldBlock.
+// Otherwise Send will panic.
 func (h *sendQueue) Send(p *packetBuffer) {
 	select {
 	case h.queue <- p:
 	case <-h.runStopped:
+	default:
+		panic("sendQueue.Send would have blocked")
 	}
 }
 
+func (h *sendQueue) WouldBlock() bool {
+	return len(h.queue) == sendQueueCapacity
+}
+
+func (h *sendQueue) Available() <-chan struct{} {
+	return h.available
+}
+
 func (h *sendQueue) Run() error {
 	defer close(h.runStopped)
 	var shouldClose bool
@@ -49,6 +67,10 @@ func (h *sendQueue) Run() error {
 				return err
 			}
 			p.Release()
+			select {
+			case h.available <- struct{}{}:
+			default:
+			}
 		}
 	}
 }

send_queue_test.go

@@ -42,30 +42,31 @@ var _ = Describe("Send Queue", func() {
 		Eventually(done).Should(BeClosed())
 	})
 
-	It("blocks sending when too many packets are queued", func() {
-		q.Send(getPacket([]byte("foobar")))
-
-		written := make(chan []byte, 2)
-		c.EXPECT().Write(gomock.Any()).Do(func(p []byte) { written <- p }).Times(2)
-
-		sent := make(chan struct{})
-		go func() {
-			defer GinkgoRecover()
-			q.Send(getPacket([]byte("raboof")))
-			close(sent)
-		}()
+	It("panics when Send() is called although there's no space in the queue", func() {
+		Expect(q.WouldBlock()).To(BeFalse())
+		q.Send(getPacket([]byte("foobar1")))
+		Expect(q.WouldBlock()).To(BeTrue())
+		Expect(func() { q.Send(getPacket([]byte("foobar2"))) }).To(Panic())
+	})
 
-		Consistently(sent).ShouldNot(BeClosed())
+	It("signals when sending is possible again", func() {
+		Expect(q.WouldBlock()).To(BeFalse())
+		q.Send(getPacket([]byte("foobar1")))
+		Consistently(q.Available()).ShouldNot(Receive())
 
+		// now start sending out packets. This should free up queue space.
+		c.EXPECT().Write(gomock.Any()).MinTimes(1).MaxTimes(2)
 		done := make(chan struct{})
 		go func() {
 			defer GinkgoRecover()
 			q.Run()
 			close(done)
 		}()
 
-		Eventually(written).Should(Receive(Equal([]byte("foobar"))))
-		Eventually(written).Should(Receive(Equal([]byte("raboof"))))
+		Eventually(q.Available()).Should(Receive())
+		Expect(q.WouldBlock()).To(BeFalse())
+		Expect(func() { q.Send(getPacket([]byte("foobar2"))) }).ToNot(Panic())
+
 		q.Close()
 		Eventually(done).Should(BeClosed())
 	})

session.go

@@ -540,7 +540,10 @@ func (s *session) run() error {
 		}
 	}
 
-	var closeErr closeError
+	var (
+		closeErr           closeError
+		sendQueueAvailable <-chan struct{}
+	)
 
 runLoop:
 	for {
@@ -563,8 +566,9 @@ runLoop:
 			// We do all the interesting stuff after the switch statement, so
 			// nothing to see here.
 		case <-s.sendingScheduled:
-			// We do all the interesting stuff after the switch statement, so
-			// nothing to see here.
+		// We do all the interesting stuff after the switch statement, so
+		// nothing to see here.
+		case <-sendQueueAvailable:
 		case p := <-s.receivedPackets:
 			// Only reset the timers if this packet was actually processed.
 			// This avoids modifying any state when handling undecryptable packets,
@@ -614,9 +618,20 @@ runLoop:
 			}
 		}
 
+		if s.sendQueue.WouldBlock() {
+			// The send queue is still busy sending out packets.
+			// Wait until there's space to enqueue new packets.
+			sendQueueAvailable = s.sendQueue.Available()
+			continue
+		}
 		if err := s.sendPackets(); err != nil {
 			s.closeLocal(err)
 		}
+		if s.sendQueue.WouldBlock() {
+			sendQueueAvailable = s.sendQueue.Available()
+		} else {
+			sendQueueAvailable = nil
+		}
 	}
 
 	s.handleCloseError(closeErr)
@@ -1495,6 +1510,9 @@ func (s *session) sendPackets() error {
 		default:
 			return fmt.Errorf("BUG: invalid send mode %d", sendMode)
 		}
+		if s.sendQueue.WouldBlock() {
+			return nil
+		}
 	}
 }
 

session_test.go

@@ -1169,9 +1169,10 @@ var _ = Describe("Session", func() {
 
 		BeforeEach(func() {
 			sender = NewMockSender(mockCtrl)
+			sender.EXPECT().Run()
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sess.sendQueue = sender
 			sessionDone = make(chan struct{})
-			sender.EXPECT().Run()
 		})
 
 		AfterEach(func() {
@@ -1211,6 +1212,7 @@ var _ = Describe("Session", func() {
 			packer.EXPECT().PackPacket().Return(p, nil)
 			packer.EXPECT().PackPacket().Return(nil, nil).AnyTimes()
 			sent := make(chan struct{})
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Send(gomock.Any()).Do(func(packet *packetBuffer) { close(sent) })
 			tracer.EXPECT().SentPacket(p.header, p.buffer.Len(), nil, []logging.Frame{})
 			sess.scheduleSending()
@@ -1386,6 +1388,7 @@ var _ = Describe("Session", func() {
 			sph.EXPECT().SendMode().Return(ackhandler.SendAny).Times(3)
 			packer.EXPECT().PackPacket().Return(getPacket(10), nil)
 			packer.EXPECT().PackPacket().Return(getPacket(11), nil)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Send(gomock.Any()).Times(2)
 			go func() {
 				defer GinkgoRecover()
@@ -1404,6 +1407,7 @@ var _ = Describe("Session", func() {
 			sph.EXPECT().SendMode().Return(ackhandler.SendAny)
 			sph.EXPECT().SendMode().Return(ackhandler.SendAck)
 			packer.EXPECT().PackPacket().Return(getPacket(100), nil)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Send(gomock.Any())
 			go func() {
 				defer GinkgoRecover()
@@ -1430,6 +1434,7 @@ var _ = Describe("Session", func() {
 				sph.EXPECT().TimeUntilSend().Return(time.Now().Add(time.Hour)),
 			)
 			written := make(chan struct{}, 2)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Send(gomock.Any()).DoAndReturn(func(p *packetBuffer) { written <- struct{}{} }).Times(2)
 			go func() {
 				defer GinkgoRecover()
@@ -1452,6 +1457,7 @@ var _ = Describe("Session", func() {
 			packer.EXPECT().PackPacket().Return(getPacket(1001), nil)
 			packer.EXPECT().PackPacket().Return(getPacket(1002), nil)
 			written := make(chan struct{}, 3)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Send(gomock.Any()).DoAndReturn(func(p *packetBuffer) { written <- struct{}{} }).Times(3)
 			go func() {
 				defer GinkgoRecover()
@@ -1462,9 +1468,70 @@ var _ = Describe("Session", func() {
 			Eventually(written).Should(HaveLen(3))
 		})
 
+		It("doesn't try to send if the send queue is full", func() {
+			available := make(chan struct{}, 1)
+			sender.EXPECT().WouldBlock().Return(true)
+			sender.EXPECT().Available().Return(available)
+			go func() {
+				defer GinkgoRecover()
+				cryptoSetup.EXPECT().RunHandshake().MaxTimes(1)
+				sess.run()
+			}()
+			sess.scheduleSending()
+			time.Sleep(scaleDuration(50 * time.Millisecond))
+
+			written := make(chan struct{})
+			sender.EXPECT().WouldBlock().AnyTimes()
+			sph.EXPECT().SentPacket(gomock.Any())
+			sph.EXPECT().HasPacingBudget().Return(true).AnyTimes()
+			sph.EXPECT().SendMode().Return(ackhandler.SendAny).AnyTimes()
+			packer.EXPECT().PackPacket().Return(getPacket(1000), nil)
+			packer.EXPECT().PackPacket().Return(nil, nil)
+			sender.EXPECT().Send(gomock.Any()).DoAndReturn(func(p *packetBuffer) { close(written) })
+			available <- struct{}{}
+			Eventually(written).Should(BeClosed())
+		})
+
+		It("stops sending when the send queue is full", func() {
+			sph.EXPECT().SentPacket(gomock.Any())
+			sph.EXPECT().HasPacingBudget().Return(true).AnyTimes()
+			sph.EXPECT().SendMode().Return(ackhandler.SendAny)
+			packer.EXPECT().PackPacket().Return(getPacket(1000), nil)
+			written := make(chan struct{}, 1)
+			sender.EXPECT().WouldBlock()
+			sender.EXPECT().WouldBlock().Return(true).Times(2)
+			sender.EXPECT().Send(gomock.Any()).DoAndReturn(func(p *packetBuffer) { written <- struct{}{} })
+			go func() {
+				defer GinkgoRecover()
+				cryptoSetup.EXPECT().RunHandshake().MaxTimes(1)
+				sess.run()
+			}()
+			available := make(chan struct{}, 1)
+			sender.EXPECT().Available().Return(available)
+			sess.scheduleSending()
+			Eventually(written).Should(Receive())
+			time.Sleep(scaleDuration(50 * time.Millisecond))
+
+			// now make room in the send queue
+			sph.EXPECT().SentPacket(gomock.Any())
+			sph.EXPECT().HasPacingBudget().Return(true).AnyTimes()
+			sph.EXPECT().SendMode().Return(ackhandler.SendAny).AnyTimes()
+			sender.EXPECT().WouldBlock().AnyTimes()
+			packer.EXPECT().PackPacket().Return(getPacket(1001), nil)
+			packer.EXPECT().PackPacket().Return(nil, nil)
+			sender.EXPECT().Send(gomock.Any()).DoAndReturn(func(p *packetBuffer) { written <- struct{}{} })
+			available <- struct{}{}
+			Eventually(written).Should(Receive())
+
+			// The send queue is not full any more. Sending on the available channel should have no effect.
+			available <- struct{}{}
+			time.Sleep(scaleDuration(50 * time.Millisecond))
+		})
+
 		It("doesn't set a pacing timer when there is no data to send", func() {
 			sph.EXPECT().HasPacingBudget().Return(true)
 			sph.EXPECT().SendMode().Return(ackhandler.SendAny).AnyTimes()
+			sender.EXPECT().WouldBlock().AnyTimes()
 			packer.EXPECT().PackPacket()
 			// don't EXPECT any calls to mconn.Write()
 			go func() {
@@ -1482,6 +1549,7 @@ var _ = Describe("Session", func() {
 
 		BeforeEach(func() {
 			sender = NewMockSender(mockCtrl)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Run()
 			sess.sendQueue = sender
 			sess.handshakeConfirmed = true