From 9e11b37b2820708998f43ddc12a5e5405709b868 Mon Sep 17 00:00:00 2001 From: dajiaohuang Date: Thu, 22 Oct 2026 23:06:23 +0800 Subject: [PATCH] fix(priority): release popped element references --- blocking_memory_test.go | 53 +++++++++++++++++++ blocking_test.go | 111 --------------------------------------- circular_test.go | 52 ------------------ example_blocking_test.go | 23 ++------ priority.go | 7 ++- priority_memory_test.go | 26 +++++++++ priority_test.go | 56 -------------------- queue_memory_test.go | 49 +++++++++++++++++ 8 files changed, 137 insertions(+), 240 deletions(-) create mode 100644 blocking_memory_test.go create mode 100644 priority_memory_test.go create mode 100644 queue_memory_test.go diff --git a/blocking_memory_test.go b/blocking_memory_test.go new file mode 100644 index 0000000..0241b6f --- /dev/null +++ b/blocking_memory_test.go @@ -0,0 +1,53 @@ +package queue + +import "testing" + +func TestBlockingGetZerosPoppedSlot(t *testing.T) { + type payload struct{ id int } + + q := NewBlocking[*payload](nil) + if err := q.Offer(&payload{id: 1}); err != nil { + t.Fatalf("offer: %v", err) + } + + backing := q.elems[:cap(q.elems)] + got, err := q.Get() + if err != nil { + t.Fatalf("get: %v", err) + } + if got == nil || got.id != 1 { + t.Fatalf("unexpected element: %v", got) + } + if len(q.elems) != 0 { + t.Fatalf("expected empty queue after Get, got size %d", len(q.elems)) + } + if backing[0] != nil { + t.Fatal("backing slot still retains the popped payload") + } +} + +func TestBlockingClearZerosBackingSlots(t *testing.T) { + type payload struct{ id int } + + q := NewBlocking[*payload](nil) + for i := 0; i < 3; i++ { + if err := q.Offer(&payload{id: i + 1}); err != nil { + t.Fatalf("offer: %v", err) + } + } + + backing := q.elems[:cap(q.elems)] + removed := q.Clear() + if len(removed) != 3 { + t.Fatalf("expected 3 removed elements, got %d", len(removed)) + } + if q.Size() != 0 { + t.Fatalf("expected empty queue after clear, got size %d", q.Size()) + } + + for i, elem := range backing { + if elem != nil { + t.Fatalf("backing slot %d still retains a payload after Clear", i) + } + } +} diff --git a/blocking_test.go b/blocking_test.go index 701ccaa..062bcf5 100644 --- a/blocking_test.go +++ b/blocking_test.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "reflect" - "runtime" "sort" "sync" "testing" @@ -33,8 +32,6 @@ func TestBlocking(t *testing.T) { t.Run("CondWaitWithCapacity", testBlockingCondWaitWithCapacity) t.Run("MarshalJSON", testBlockingMarshalJSON) t.Run("NewDoesNotAliasCallerSlice", testBlockingNewDoesNotAliasCallerSlice) - t.Run("GetReleasesReference", testBlockingGetReleasesReference) - t.Run("ClearReleasesReferences", testBlockingClearReleasesReferences) t.Run("OfferBroadcastsToAllWaiters", testBlockingOfferBroadcastsToAllWaiters) } @@ -92,114 +89,6 @@ func testBlockingOfferBroadcastsToAllWaiters(t *testing.T) { } } -func testBlockingGetReleasesReference(t *testing.T) { - t.Parallel() - - type payload struct{ id int } - - var bq *queue.Blocking[*payload] - - finalized := make(chan struct{}, 1) - - // Create the payload in a nested scope so p/got don't keep it alive - // after the offer/get cycle. - func() { - p := &payload{id: 42} - runtime.SetFinalizer(p, func(*payload) { - finalized <- struct{}{} - }) - - bq = queue.NewBlocking[*payload](nil) - - if err := bq.Offer(p); err != nil { - t.Fatalf("offer: %v", err) - } - - got, err := bq.Get() - if err != nil { - t.Fatalf("get: %v", err) - } - - if got != p { - t.Fatalf("unexpected element %v", got) - } - }() - - // Queue is still live. After a successful Get, the queue must not - // retain the popped element. Nudge GC until the finalizer fires or - // we give up. - deadline := time.After(time.Second) - - for { - runtime.GC() //nolint:revive // explicit GC needed to drive finalizer - - select { - case <-finalized: - runtime.KeepAlive(bq) - return - case <-deadline: - runtime.KeepAlive(bq) - t.Fatal("popped element not finalized; backing array still holds it") - default: - time.Sleep(10 * time.Millisecond) - } - } -} - -func testBlockingClearReleasesReferences(t *testing.T) { - t.Parallel() - - type payload struct{ id int } - - var bq *queue.Blocking[*payload] - - finalized := make(chan struct{}, 3) - - func() { - items := []*payload{{id: 1}, {id: 2}, {id: 3}} - - for _, it := range items { - runtime.SetFinalizer(it, func(*payload) { - finalized <- struct{}{} - }) - } - - bq = queue.NewBlocking[*payload](nil) - - for _, it := range items { - if err := bq.Offer(it); err != nil { - t.Fatalf("offer: %v", err) - } - } - - cleared := bq.Clear() - if len(cleared) != len(items) { - t.Fatalf("expected %d cleared, got %d", len(items), len(cleared)) - } - }() - - // All items should be eligible for finalization now — the queue is - // empty and the temporary `cleared` slice has gone out of scope. - deadline := time.After(time.Second) - - count := 0 - for count < 3 { - runtime.GC() //nolint:revive // explicit GC needed to drive finalizer - - select { - case <-finalized: - count++ - case <-deadline: - runtime.KeepAlive(bq) - t.Fatalf("only %d/3 payloads finalized; Clear kept references", count) - default: - time.Sleep(10 * time.Millisecond) - } - } - - runtime.KeepAlive(bq) -} - func testBlockingNewDoesNotAliasCallerSlice(t *testing.T) { t.Parallel() diff --git a/circular_test.go b/circular_test.go index 35a3825..e50da92 100644 --- a/circular_test.go +++ b/circular_test.go @@ -5,9 +5,7 @@ import ( "encoding/json" "errors" "reflect" - "runtime" "testing" - "time" "github.com/adrianbrad/queue" ) @@ -27,56 +25,6 @@ func TestCircular(t *testing.T) { t.Run("Iterator", testCircularIterator) t.Run("MarshalJSON", testCircularMarshalJSON) t.Run("NonPositiveCapacity", testCircularNonPositiveCapacity) - t.Run("ResetReleasesBeyondInitial", testCircularResetReleasesBeyondInitial) -} - -func testCircularResetReleasesBeyondInitial(t *testing.T) { - t.Parallel() - - type payload struct{ id int } - - var cq *queue.Circular[*payload] - - finalized := make(chan struct{}, 3) - - func() { - // Capacity 5, 2 initial elems, then offer 3 more. Reset must - // release the 3 extras; on unfixed code, slots 2..4 still hold - // them via the backing array. - cq = queue.NewCircular([]*payload{{id: 1}, {id: 2}}, 5) - - for i := 3; i <= 5; i++ { - p := &payload{id: i} - runtime.SetFinalizer(p, func(*payload) { - finalized <- struct{}{} - }) - - if err := cq.Offer(p); err != nil { - t.Fatalf("offer: %v", err) - } - } - - cq.Reset() - }() - - deadline := time.After(time.Second) - - count := 0 - for count < 3 { - runtime.GC() //nolint:revive // explicit GC needed to drive finalizer - - select { - case <-finalized: - count++ - case <-deadline: - runtime.KeepAlive(cq) - t.Fatalf("only %d/3 extra payloads finalized after Reset", count) - default: - time.Sleep(10 * time.Millisecond) - } - } - - runtime.KeepAlive(cq) } func testCircularNonPositiveCapacity(t *testing.T) { diff --git a/example_blocking_test.go b/example_blocking_test.go index 66ff3d5..a5641c0 100644 --- a/example_blocking_test.go +++ b/example_blocking_test.go @@ -2,7 +2,6 @@ package queue_test import ( "fmt" - "time" "github.com/adrianbrad/queue" ) @@ -27,25 +26,15 @@ func ExampleBlocking() { empty = blockingQueue.IsEmpty() fmt.Println("Empty after clear:", empty) - var ( - elem int - after time.Duration - ) - + var elem int done := make(chan struct{}) - start := time.Now() - - // this function waits for a new element to be available in the queue. + // This function waits for a new element to be available in the queue. go func() { defer close(done) - elem = blockingQueue.GetWait() - after = time.Since(start) }() - time.Sleep(time.Millisecond) - // insert a new element into the queue. if err := blockingQueue.Offer(4); err != nil { fmt.Println("Offer err:", err) @@ -62,11 +51,7 @@ func ExampleBlocking() { <-done - fmt.Printf( - "Elem %d received after %s", - elem, - after.Round(time.Millisecond), - ) + fmt.Println("Elem received:", elem) // Output: // Contains 3: true @@ -75,5 +60,5 @@ func ExampleBlocking() { // Clear: [1 2 3] // Empty after clear: true // Peeked elem: 4 - // Elem 4 received after 1ms + // Elem received: 4 } diff --git a/priority.go b/priority.go index 2b51202..bbee094 100644 --- a/priority.go +++ b/priority.go @@ -46,9 +46,12 @@ func (h *priorityHeap[T]) Push(elem any) { func (h *priorityHeap[T]) Pop() any { n := len(h.elems) - elem := (h.elems)[n-1] + elem := h.elems[n-1] + // Clear the removed slot so pointer elements are not retained by the backing array. + var zero T + h.elems[n-1] = zero - h.elems = (h.elems)[0 : n-1] + h.elems = h.elems[:n-1] return elem } diff --git a/priority_memory_test.go b/priority_memory_test.go new file mode 100644 index 0000000..189bfcb --- /dev/null +++ b/priority_memory_test.go @@ -0,0 +1,26 @@ +package queue + +import "testing" + +func TestPriorityGetZerosPoppedSlot(t *testing.T) { + type payload struct{ id int } + + q := NewPriority([]*payload{{id: 1}}, func(a, b *payload) bool { + return a.id < b.id + }) + backing := q.elements.elems[:cap(q.elements.elems)] + + got, err := q.Get() + if err != nil { + t.Fatalf("get: %v", err) + } + if got == nil || got.id != 1 { + t.Fatalf("unexpected element: %v", got) + } + if len(q.elements.elems) != 0 { + t.Fatalf("expected empty queue after Get, got size %d", len(q.elements.elems)) + } + if backing[0] != nil { + t.Fatal("backing slot still retains the popped payload") + } +} diff --git a/priority_test.go b/priority_test.go index 05277ee..8a7fe95 100644 --- a/priority_test.go +++ b/priority_test.go @@ -5,10 +5,8 @@ import ( "encoding/json" "errors" "reflect" - "runtime" "sort" "testing" - "time" "github.com/adrianbrad/queue" ) @@ -32,60 +30,6 @@ func TestPriority(t *testing.T) { t.Run("Reset", testPriorityReset) t.Run("MarshalJSON", testPriorityMarshalJSON) t.Run("NegativeCapacity", testPriorityNegativeCapacity) - t.Run("ResetReleasesExtras", testPriorityResetReleasesExtras) -} - -func testPriorityResetReleasesExtras(t *testing.T) { - t.Parallel() - - type payload struct{ id int } - - var pq *queue.Priority[*payload] - - finalized := make(chan struct{}, 5) - - lessByID := func(a, b *payload) bool { return a.id < b.id } - - func() { - // 2 initial elems, then Offer 5 more with finalizers. Reset - // should make the extras reclaimable. - pq = queue.NewPriority( - []*payload{{id: 1}, {id: 2}}, - lessByID, - ) - - for i := 3; i <= 7; i++ { - p := &payload{id: i} - runtime.SetFinalizer(p, func(*payload) { - finalized <- struct{}{} - }) - - if err := pq.Offer(p); err != nil { - t.Fatalf("offer: %v", err) - } - } - - pq.Reset() - }() - - deadline := time.After(time.Second) - - count := 0 - for count < 5 { - runtime.GC() //nolint:revive // explicit GC needed to drive finalizer - - select { - case <-finalized: - count++ - case <-deadline: - runtime.KeepAlive(pq) - t.Fatalf("only %d/5 extra payloads finalized after Reset", count) - default: - time.Sleep(10 * time.Millisecond) - } - } - - runtime.KeepAlive(pq) } func testPriorityNegativeCapacity(t *testing.T) { diff --git a/queue_memory_test.go b/queue_memory_test.go new file mode 100644 index 0000000..b9093ee --- /dev/null +++ b/queue_memory_test.go @@ -0,0 +1,49 @@ +package queue + +import "testing" + +func TestCircularResetZerosSlotsBeyondInitialElements(t *testing.T) { + type payload struct{ id int } + + q := NewCircular([]*payload{{id: 1}, {id: 2}}, 5) + for i := 3; i <= 5; i++ { + if err := q.Offer(&payload{id: i}); err != nil { + t.Fatalf("offer: %v", err) + } + } + + q.Reset() + if q.size != 2 { + t.Fatalf("expected reset size 2, got %d", q.size) + } + for i := 2; i < len(q.elems); i++ { + if q.elems[i] != nil { + t.Fatalf("slot %d still retains an extra payload after Reset", i) + } + } +} + +func TestPriorityResetDropsOversizedBacking(t *testing.T) { + type payload struct{ id int } + + lessByID := func(a, b *payload) bool { return a.id < b.id } + q := NewPriority([]*payload{{id: 1}, {id: 2}}, lessByID) + for i := 3; i <= 7; i++ { + if err := q.Offer(&payload{id: i}); err != nil { + t.Fatalf("offer: %v", err) + } + } + if cap(q.elements.elems) <= 2 { + t.Fatalf("expected offered items to grow the backing array, got capacity %d", cap(q.elements.elems)) + } + + q.Reset() + if len(q.elements.elems) != 2 || cap(q.elements.elems) != 2 { + t.Fatalf("expected reset to retain only the two initial slots, got len=%d cap=%d", len(q.elements.elems), cap(q.elements.elems)) + } + for _, elem := range q.elements.elems { + if elem == nil || elem.id < 1 || elem.id > 2 { + t.Fatalf("unexpected element after Reset: %v", elem) + } + } +}