Skip to content

Commit fe41c1b

Browse files
committed
fix(flowcontrol): call up() after down() in maxMinHeap.Remove
Remove swaps position i with the last element then calls down() but not up(). When the last element comes from a different subtree, the ancestor relationship can break: Before: After Remove(index=3, time=1): 0 0 / \ / \ 5 50 5 50 / \ \ / \ 1 4 40 40 4 Last element (index 5, time=40) swaps into index 3. down(3): no children, no-op. Node 1 (min-level, time=5) must be lowest-priority in {5, 40, 4}. time=40 has lower priority than time=5 -- heap property violated. Add the missing up(i) call after down(i). If down moved the element, up is a no-op at the new position (and vice versa). Signed-off-by: RishabhSaini <rishabhsaini01@gmail.com>
1 parent bdbf7cb commit fe41c1b

2 files changed

Lines changed: 77 additions & 2 deletions

File tree

pkg/epp/flowcontrol/framework/plugins/queue/maxminheap.go

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -285,9 +285,12 @@ func (h *maxMinHeap) Remove(handle flowcontrol.QueueItemHandle) (flowcontrol.Que
285285
delete(h.handles, handle)
286286
h.byteSize.Add(^item.OriginalRequest().ByteSize() + 1) // Atomic subtraction
287287

288-
// The swapped item at index i might violate the heap property, so we must restore it
289-
// by bubbling the item down the heap.
288+
// The swapped item at index i might violate the heap property in either direction:
289+
// - downward: it is less extreme than its descendants -> down() fixes this.
290+
// - upward: it is more extreme than its ancestors -> up() fixes this.
291+
// Both calls are needed because the replacement can come from a different subtree.
290292
h.down(i)
293+
h.up(i)
291294
} else {
292295
// It's the last item, just remove it.
293296
h.items = h.items[:n]

pkg/epp/flowcontrol/framework/plugins/queue/maxminheap_test.go

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ limitations under the License.
1717
package queue
1818

1919
import (
20+
"fmt"
2021
"math"
2122
"testing"
2223
"time"
@@ -61,6 +62,77 @@ func TestMaxMinHeap_InternalProperty(t *testing.T) {
6162
}
6263
}
6364

65+
// TestMaxMinHeap_Remove_CrossSubtreeSwap demonstrates that Remove corrupts the heap when the replacement element
66+
// (swapped from the end of the array) belongs to a different subtree than the removed element.
67+
// The root cause is that Remove calls down() but not up(), so ancestor-level violations go unfixed.
68+
func TestMaxMinHeap_Remove_CrossSubtreeSwap(t *testing.T) {
69+
t.Parallel()
70+
71+
// 6-element max-min heap (FCFS: earlier time = higher priority):
72+
//
73+
// 0 L0 (max) -- highest priority
74+
// / \
75+
// 5 50 L1 (min) -- lowest priority in subtrees
76+
// / \ \
77+
// 1 4 40 L2 (max) -- leaves
78+
//
79+
// Left subtree of root: {5, 1, 4} (index 1, bounded by min-ancestor time=5)
80+
// Right subtree of root: {50, 40} (index 2, bounded by min-ancestor time=50)
81+
times := []int{0, 5, 50, 1, 4, 40}
82+
83+
q := newMaxMinHeap(enqueueTimePolicy)
84+
flowKey := flowcontrol.FlowKey{ID: "f", Priority: 0}
85+
base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
86+
for i, sec := range times {
87+
item := mocks.NewMockQueueItemAccessor(10, fmt.Sprintf("t%d", sec), flowKey)
88+
item.EnqueueTimeV = base.Add(time.Duration(sec) * time.Second)
89+
hi := &heapItem{item: item, index: i}
90+
item.SetHandle(hi)
91+
q.items = append(q.items, item)
92+
q.handles[item.Handle()] = hi
93+
}
94+
q.byteSize.Store(60)
95+
96+
// Remove index 3 (time=1, left subtree leaf). Last element (index 5, time=40, right subtree) swaps in.
97+
// down(3) has no children -> no-op. Heap becomes [0, 5, 50, 40, 4].
98+
// Node 1 (min, time=5) must be lowest priority in its subtree {5, 40, 4}.
99+
// But time=40 has lower priority than time=5 -> VIOLATION.
100+
_, err := q.Remove(q.items[3].Handle())
101+
require.NoError(t, err)
102+
103+
// Verify every node is the max (or min) of its entire subtree -- not just its direct children.
104+
for i := range q.items {
105+
isMin := isMinLevel(i)
106+
for d := range q.items {
107+
if !isAncestor(i, d) {
108+
continue
109+
}
110+
if isMin {
111+
require.Falsef(t, q.policy.Less(q.items[i], q.items[d]),
112+
"min-level node %d (t=%v) has higher priority than descendant %d (t=%v)",
113+
i, q.items[i].EnqueueTime().Sub(base)/time.Second,
114+
d, q.items[d].EnqueueTime().Sub(base)/time.Second)
115+
} else {
116+
require.Falsef(t, q.policy.Less(q.items[d], q.items[i]),
117+
"max-level node %d (t=%v) has lower priority than descendant %d (t=%v)",
118+
i, q.items[i].EnqueueTime().Sub(base)/time.Second,
119+
d, q.items[d].EnqueueTime().Sub(base)/time.Second)
120+
}
121+
}
122+
}
123+
}
124+
125+
// isAncestor reports whether a is a strict ancestor of d in a binary heap.
126+
func isAncestor(a, d int) bool {
127+
if d <= a {
128+
return false
129+
}
130+
for d > a {
131+
d = (d - 1) / 2
132+
}
133+
return d == a
134+
}
135+
64136
// assertHeapProperty checks if the slice of items satisfies the max-min heap property.
65137
func assertHeapProperty(t *testing.T, h *maxMinHeap, msgAndArgs ...any) {
66138
t.Helper()

0 commit comments

Comments
 (0)