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

mock_sender_test.go

@@ -33,6 +33,20 @@ func (m *MockSender) EXPECT() *MockSenderMockRecorder {
 	return m.recorder
 }
 
+// Available mocks base method
+func (m *MockSender) Available() <-chan struct{} {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Available")
+	ret0, _ := ret[0].(<-chan struct{})
+	return ret0
+}
+
+// Available indicates an expected call of Available
+func (mr *MockSenderMockRecorder) Available() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Available", reflect.TypeOf((*MockSender)(nil).Available))
+}
+
 // Close mocks base method
 func (m *MockSender) Close() {
 	m.ctrl.T.Helper()
@@ -70,3 +84,17 @@ func (mr *MockSenderMockRecorder) Send(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockSender)(nil).Send), arg0)
 }
+
+// WouldBlock mocks base method
+func (m *MockSender) WouldBlock() bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "WouldBlock")
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// WouldBlock indicates an expected call of WouldBlock
+func (mr *MockSenderMockRecorder) WouldBlock() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WouldBlock", reflect.TypeOf((*MockSender)(nil).WouldBlock))
+}

send_queue.go

@@ -3,6 +3,8 @@ package quic
 type sender interface {
 	Send(p *packetBuffer)
 	Run() error
+	WouldBlock() bool
+	Available() <-chan struct{}
 	Close()
 }
 
@@ -10,28 +12,44 @@ 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,21 +42,20 @@ var _ = Describe("Send Queue", func() {
 		Eventually(done).Should(BeClosed())
 	})
 
-	It("blocks sending when too many packets are queued", func() {
+	It("panics when Send() is called although there's no space in the queue", func() {
-		q.Send(getPacket([]byte("foobar")))
+		Expect(q.WouldBlock()).To(BeFalse())
+		q.Send(getPacket([]byte("foobar1")))
+		Expect(q.WouldBlock()).To(BeTrue())
+		Expect(func() { q.Send(getPacket([]byte("foobar2"))) }).To(Panic())
+	})
 
-		written := make(chan []byte, 2)
+	It("signals when sending is possible again", func() {
-		c.EXPECT().Write(gomock.Any()).Do(func(p []byte) { written <- p }).Times(2)
+		Expect(q.WouldBlock()).To(BeFalse())
-
+		q.Send(getPacket([]byte("foobar1")))
-		sent := make(chan struct{})
+		Consistently(q.Available()).ShouldNot(Receive())
-		go func() {
-			defer GinkgoRecover()
-			q.Send(getPacket([]byte("raboof")))
-			close(sent)
-		}()
-
-		Consistently(sent).ShouldNot(BeClosed())
 
+		// 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()
@@ -64,8 +63,10 @@ var _ = Describe("Send Queue", func() {
 			close(done)
 		}()
 
-		Eventually(written).Should(Receive(Equal([]byte("foobar"))))
+		Eventually(q.Available()).Should(Receive())
-		Eventually(written).Should(Receive(Equal([]byte("raboof"))))
+		Expect(q.WouldBlock()).To(BeFalse())
+		Expect(func() { q.Send(getPacket([]byte("foobar2"))) }).ToNot(Panic())
+
 		q.Close()
 		Eventually(done).Should(BeClosed())
 	})

session.go

@@ -542,7 +542,10 @@ func (s *session) run() error {
 		}
 	}
 
-	var closeErr closeError
+	var (
+		closeErr           closeError
+		sendQueueAvailable <-chan struct{}
+	)
 
 runLoop:
 	for {
@@ -583,6 +586,7 @@ runLoop:
 			case <-s.sendingScheduled:
 				// We do all the interesting stuff after the switch statement, so
 				// nothing to see here.
+			case <-sendQueueAvailable:
 			case firstPacket := <-s.receivedPackets:
 				s.sentPacketHandler.ReceivedBytes(firstPacket.Size())
 				wasProcessed := s.handlePacketImpl(firstPacket)
@@ -655,9 +659,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)
@@ -1541,6 +1556,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

@@ -1214,9 +1214,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() {
@@ -1256,6 +1257,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()
@@ -1431,6 +1433,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()
@@ -1449,6 +1452,7 @@ var _ = Describe("Session", func() {
 			sph.EXPECT().SendMode().Return(ackhandler.SendAny).Times(3)
 			packer.EXPECT().PackPacket().Return(getPacket(10), nil)
 			packer.EXPECT().PackPacket().Return(nil, nil)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Send(gomock.Any())
 			go func() {
 				defer GinkgoRecover()
@@ -1467,6 +1471,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()
@@ -1493,6 +1498,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()
@@ -1515,6 +1521,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()
@@ -1525,9 +1532,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() {
@@ -1545,6 +1613,7 @@ var _ = Describe("Session", func() {
 
 		BeforeEach(func() {
 			sender = NewMockSender(mockCtrl)
+			sender.EXPECT().WouldBlock().AnyTimes()
 			sender.EXPECT().Run()
 			sess.sendQueue = sender
 			sess.handshakeConfirmed = true