Repository: quic-go/quic-go
Pull request #4223 (2023-12-30 03:37:19)

use a ring buffer for the datagram queue

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/quic-go/quic-go/pull/4223?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`b71a76b`)](https://app.codecov.io/gh/quic-go/quic-go/commit/b71a76bf14a4d3ab843e9c8788bd85e7dc708704?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) 84.01% compared to head [(`d4cd6b6`)](https://app.codecov.io/gh/quic-go/quic-go/pull/4223?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) 84.01%.


<details><summary>Additional details and impacted files</summary>


```diff
@@                    Coverage Diff                    @@
##           datagram-queue-length    #4223      +/-   ##
=========================================================
- Coverage                  84.01%   84.01%   -0.00%     
=========================================================
  Files                        149      149              
  Lines                      15419    15421       +2     
=========================================================
+ Hits                       12954    12955       +1     
- Misses                      1967     1969       +2     
+ Partials                     498      497       -1     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/quic-go/quic-go/pull/4223?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go).

sukunrt: Sorry for the drive by comment. Just to point it out in case it was actually missed. Please ignore if not relevant. 

Is a Peek followed by a Pop threadsafe considering that we only synchronise the calls individually and not the whole Peek + Pop operation? 
I can't see any issues currently because it seems to be called only from the connection run loop go routine. 
marten-seemann: > Sorry for the drive by comment.

Thanks for reviewing!



> Is a Peek followed by a Pop threadsafe considering that we only synchronise the calls individually and not the whole Peek + Pop operation?

The entire `RingBuffer` is not thread-safe (it doesn't try to: it doesn't even contain a mutex). It's the caller's responsibility to synchronize access.

I believe the `datagramQueue` does that: `Peek` will return a non-nil frame iff there's actually a frame queued. While it's possible that new frames are added between the `Peek` and `Pop` call, the only way that frame could disappear via `Pop` call, there's no other way to dequeue / delete a frame from the queue. Do you see any problem with that logic?
sukunrt: This makes sense. 
Only `Add` happens from a different goroutine. `Peek` and `Pop` happen from the same goroutine so we will always `Pop` the element we received from `Peek` since `Add` puts it at the back of the queue. 

Patch:
diff --git a/datagram_queue.go b/datagram_queue.go
index 43026b25..e26285b2 100644
--- a/datagram_queue.go
+++ b/datagram_queue.go
@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/quic-go/quic-go/internal/utils"
+	"github.com/quic-go/quic-go/internal/utils/ringbuffer"
 	"github.com/quic-go/quic-go/internal/wire"
 )
 
@@ -15,8 +16,8 @@ const (
 
 type datagramQueue struct {
 	sendMx    sync.Mutex
-	sendQueue []*wire.DatagramFrame // TODO: this could be a ring buffer
-	sent      chan struct{}         // used to notify Add that a datagram was dequeued
+	sendQueue ringbuffer.RingBuffer[*wire.DatagramFrame]
+	sent      chan struct{} // used to notify Add that a datagram was dequeued
 
 	rcvMx    sync.Mutex
 	rcvQueue [][]byte
@@ -47,8 +48,8 @@ func (h *datagramQueue) Add(f *wire.DatagramFrame) error {
 	h.sendMx.Lock()
 
 	for {
-		if len(h.sendQueue) < maxDatagramSendQueueLen {
-			h.sendQueue = append(h.sendQueue, f)
+		if h.sendQueue.Len() < maxDatagramSendQueueLen {
+			h.sendQueue.PushBack(f)
 			h.sendMx.Unlock()
 			h.hasData()
 			return nil
@@ -72,19 +73,16 @@ func (h *datagramQueue) Add(f *wire.DatagramFrame) error {
 func (h *datagramQueue) Peek() *wire.DatagramFrame {
 	h.sendMx.Lock()
 	defer h.sendMx.Unlock()
-	if len(h.sendQueue) == 0 {
+	if h.sendQueue.Empty() {
 		return nil
 	}
-	return h.sendQueue[0]
+	return h.sendQueue.PeekFront()
 }
 
 func (h *datagramQueue) Pop() {
 	h.sendMx.Lock()
 	defer h.sendMx.Unlock()
-	if len(h.sendQueue) == 0 {
-		panic("datagramQueue BUG: Pop called for nil frame")
-	}
-	h.sendQueue = h.sendQueue[1:]
+	_ = h.sendQueue.PopFront()
 	select {
 	case h.sent <- struct{}{}:
 	default:
diff --git a/internal/utils/ringbuffer/ringbuffer.go b/internal/utils/ringbuffer/ringbuffer.go
index 81a5ad44..f9b2c797 100644
--- a/internal/utils/ringbuffer/ringbuffer.go
+++ b/internal/utils/ringbuffer/ringbuffer.go
@@ -8,7 +8,7 @@ type RingBuffer[T any] struct {
 	full             bool
 }
 
-// Init preallocs a buffer with a certain size.
+// Init preallocates a buffer with a certain size.
 func (r *RingBuffer[T]) Init(size int) {
 	r.ring = make([]T, size)
 }
@@ -62,6 +62,16 @@ func (r *RingBuffer[T]) PopFront() T {
 	return t
 }
 
+// PeekFront returns the next element.
+// It must not be called when the buffer is empty, that means that
+// callers might need to check if there are elements in the buffer first.
+func (r *RingBuffer[T]) PeekFront() T {
+	if r.Empty() {
+		panic("github.com/quic-go/quic-go/internal/utils/ringbuffer: peek from an empty queue")
+	}
+	return r.ring[r.headPos]
+}
+
 // Grow the maximum size of the queue.
 // This method assume the queue is full.
 func (r *RingBuffer[T]) grow() {


Test patch:
diff --git a/internal/utils/ringbuffer/ringbuffer_test.go b/internal/utils/ringbuffer/ringbuffer_test.go
index 13241a30..68f1c7cc 100644
--- a/internal/utils/ringbuffer/ringbuffer_test.go
+++ b/internal/utils/ringbuffer/ringbuffer_test.go
@@ -6,14 +6,17 @@ import (
 )
 
 var _ = Describe("RingBuffer", func() {
-	It("push and pop", func() {
+	It("push, peek and pop", func() {
 		r := RingBuffer[int]{}
 		Expect(len(r.ring)).To(Equal(0))
 		Expect(func() { r.PopFront() }).To(Panic())
 		r.PushBack(1)
 		r.PushBack(2)
 		r.PushBack(3)
+		Expect(r.PeekFront()).To(Equal(1))
+		Expect(r.PeekFront()).To(Equal(1))
 		Expect(r.PopFront()).To(Equal(1))
+		Expect(r.PeekFront()).To(Equal(2))
 		Expect(r.PopFront()).To(Equal(2))
 		r.PushBack(4)
 		r.PushBack(5)
@@ -25,7 +28,16 @@ var _ = Describe("RingBuffer", func() {
 		Expect(r.PopFront()).To(Equal(5))
 		Expect(r.PopFront()).To(Equal(6))
 	})
-	It("clear", func() {
+
+	It("panics when Peek or Pop are called on an empty buffer", func() {
+		r := RingBuffer[string]{}
+		Expect(r.Empty()).To(BeTrue())
+		Expect(r.Len()).To(BeZero())
+		Expect(func() { r.PeekFront() }).To(Panic())
+		Expect(func() { r.PopFront() }).To(Panic())
+	})
+
+	It("clearing", func() {
 		r := RingBuffer[int]{}
 		r.Init(2)
 		r.PushBack(1)