Skip to content

Commit d8a0c46

Browse files
LukeAVanDrieLukeAVanDrie
authored
refactor(flowcontrol): unify on a single priority queue (#1831)
With the dispatch queue single-ended, the FCFS-only ListQueue and the capability-negotiation machinery that selected between queue types no longer earn their complexity. Collapse to one queue. - Delete ListQueue; route all flows through the priority queue. FCFS orders by its existing enqueue-time comparator, yielding exact logical-arrival ordering (the heap honors the policy, unlike ListQueue's physical-insertion approximation). - Remove the QueueCapability system end to end: Capabilities() from the queue contract and implementation, RequiredQueueCapabilities() from OrderingPolicy and all ordering policies, and the capabilityChecker validation in the registry config. - Queue selection collapses to always using the priority queue. The registry no longer validates a band's queue name at config construction; an unknown name now surfaces when the flow is first provisioned. The Queue config field is retained here and removed in a follow-up. Signed-off-by: Luke Van Drie <lukevandrie@google.com>
1 parent add7644 commit d8a0c46

23 files changed

Lines changed: 89 additions & 542 deletions

File tree

‎pkg/epp/config/loader/flowcontrol_test.go‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,6 @@ func newFlowControlTestHandle(t *testing.T) fwkplugin.Handle {
6565
Type: edf.EDFOrderingPolicyType,
6666
Name: edf.EDFOrderingPolicyType,
6767
},
68-
RequiredQueueCapabilitiesV: []fwkfc.QueueCapability{fwkfc.CapabilityPriorityConfigurable},
6968
})
7069
handle.AddPlugin(usagelimits.StaticUsageLimitPolicyType, usagelimits.DefaultPolicy())
7170
return handle

‎pkg/epp/flowcontrol/contracts/errors.go‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,6 @@ var (
2828
// ErrPriorityBandNotFound indicates that a requested priority band does not exist in the registry configuration.
2929
ErrPriorityBandNotFound = errors.New("priority band not found")
3030

31-
// ErrPolicyQueueIncompatible indicates that a selected policy is not compatible with the capabilities of the queue.
32-
ErrPolicyQueueIncompatible = errors.New("policy is not compatible with queue capabilities")
33-
3431
// ErrInvalidQueueItemHandle indicates that a QueueItemHandle provided to a SafeQueue operation (e.g.,
3532
// SafeQueue.Remove()) is not valid for that queue, has been invalidated, or does not correspond to an actual item in
3633
// the queue.

‎pkg/epp/flowcontrol/contracts/mocks/mocks.go‎

Lines changed: 13 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -159,21 +159,19 @@ func (m *MockEndpointCandidates) Locate(ctx context.Context, requestMetadata map
159159
// It is used for tests that need to control the exact return values of a queue's methods without simulating the queue's
160160
// internal logic or state.
161161
type MockSafeQueue struct {
162-
NameV string
163-
CapabilitiesV []flowcontrol.QueueCapability
164-
LenV int
165-
ByteSizeV uint64
166-
PeekV flowcontrol.QueueItemAccessor
167-
AddFunc func(item flowcontrol.QueueItemAccessor)
168-
RemoveFunc func(handle flowcontrol.QueueItemHandle) (flowcontrol.QueueItemAccessor, error)
169-
CleanupFunc func(predicate contracts.PredicateFunc) []flowcontrol.QueueItemAccessor
170-
DrainFunc func() []flowcontrol.QueueItemAccessor
162+
NameV string
163+
LenV int
164+
ByteSizeV uint64
165+
PeekV flowcontrol.QueueItemAccessor
166+
AddFunc func(item flowcontrol.QueueItemAccessor)
167+
RemoveFunc func(handle flowcontrol.QueueItemHandle) (flowcontrol.QueueItemAccessor, error)
168+
CleanupFunc func(predicate contracts.PredicateFunc) []flowcontrol.QueueItemAccessor
169+
DrainFunc func() []flowcontrol.QueueItemAccessor
171170
}
172171

173-
func (m *MockSafeQueue) Name() string { return m.NameV }
174-
func (m *MockSafeQueue) Capabilities() []flowcontrol.QueueCapability { return m.CapabilitiesV }
175-
func (m *MockSafeQueue) Len() int { return m.LenV }
176-
func (m *MockSafeQueue) ByteSize() uint64 { return m.ByteSizeV }
172+
func (m *MockSafeQueue) Name() string { return m.NameV }
173+
func (m *MockSafeQueue) Len() int { return m.LenV }
174+
func (m *MockSafeQueue) ByteSize() uint64 { return m.ByteSizeV }
177175

178176
func (m *MockSafeQueue) Peek() flowcontrol.QueueItemAccessor {
179177
return m.PeekV
@@ -331,9 +329,8 @@ func (m *MockManagedQueue) Drain() []flowcontrol.QueueItemAccessor {
331329
return drained
332330
}
333331

334-
func (m *MockManagedQueue) FlowKey() flowcontrol.FlowKey { return m.FlowKeyV }
335-
func (m *MockManagedQueue) Name() string { return "" }
336-
func (m *MockManagedQueue) Capabilities() []flowcontrol.QueueCapability { return nil }
332+
func (m *MockManagedQueue) FlowKey() flowcontrol.FlowKey { return m.FlowKeyV }
333+
func (m *MockManagedQueue) Name() string { return "" }
337334
func (m *MockManagedQueue) OrderingPolicy() flowcontrol.OrderingPolicy {
338335
if m.OrderingPolicyFunc != nil {
339336
return m.OrderingPolicyFunc()

‎pkg/epp/flowcontrol/framework/plugins/queue/factory.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,8 @@ func MustRegisterQueue(name RegisteredQueueName, constructor QueueConstructor) {
5050
RegisteredQueues[name] = constructor
5151
}
5252

53-
// NewQueueFromName creates a new SafeQueue given its registered name and the OrderingPolicy that will be optionally
54-
// used to configure the queue (provided it declares CapabilityPriorityConfigurable).
53+
// NewQueueFromName creates a new SafeQueue given its registered name and the OrderingPolicy used to
54+
// configure the queue's ordering.
5555
// This is called by the FlowRegistry during initialization of a flow's ManagedQueue.
5656
func NewQueueFromName(name RegisteredQueueName, policy flowcontrol.OrderingPolicy) (contracts.SafeQueue, error) {
5757
mu.RLock()

‎pkg/epp/flowcontrol/framework/plugins/queue/functional_test.go‎

Lines changed: 29 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ package queue
1818

1919
import (
2020
"fmt"
21-
"slices"
2221
"sort"
2322
"sync"
2423
"sync/atomic"
@@ -170,8 +169,6 @@ func TestQueueConformance(t *testing.T) {
170169
assert.Zero(t, q.Len(), "A new queue should have a length of 0")
171170
assert.Zero(t, q.ByteSize(), "A new queue should have a byte size of 0")
172171
assert.Equal(t, string(queueName), q.Name(), "Name() should return the registered name of the queue")
173-
assert.NotNil(t, q.Capabilities(), "Capabilities() should not return a nil slice")
174-
assert.NotEmpty(t, q.Capabilities(), "Capabilities() should return at least one capability")
175172
})
176173

177174
t.Run("LifecycleAndOrdering_DefaultFIFO", func(t *testing.T) {
@@ -192,38 +189,35 @@ func TestQueueConformance(t *testing.T) {
192189
testLifecycleAndOrdering(t, q, itemsInFIFOOrder, "DefaultFIFO")
193190
})
194191

195-
qForCapCheck, err := constructor(enqueueTimePolicy)
196-
if err == nil && slices.Contains(qForCapCheck.Capabilities(), flowcontrol.CapabilityPriorityConfigurable) {
197-
t.Run("LifecycleAndOrdering_PriorityConfigurable_ByteSize", func(t *testing.T) {
198-
t.Parallel()
199-
q, err := constructor(byteSizePolicy)
200-
require.NoError(t, err, "Setup: creating queue with byteSizePolicy should not fail")
201-
202-
itemLarge := mocks.NewMockQueueItemAccessor(100, "itemLarge_prio", flowKey)
203-
itemSmall := mocks.NewMockQueueItemAccessor(20, "itemSmall_prio", flowKey)
204-
itemMedium := mocks.NewMockQueueItemAccessor(50, "itemMedium_prio", flowKey)
205-
206-
itemsInByteSizeOrder := []*mocks.MockQueueItemAccessor{itemSmall, itemMedium, itemLarge}
207-
testLifecycleAndOrdering(t, q, itemsInByteSizeOrder, "PriorityByteSize")
208-
})
209-
210-
t.Run("LifecycleAndOrdering_PriorityConfigurable_LIFO", func(t *testing.T) {
211-
t.Parallel()
212-
q, err := constructor(reverseEnqueueTimePolicy)
213-
require.NoError(t, err, "Setup: creating queue with reverseEnqueueTimePolicy should not fail")
214-
215-
now := time.Now()
216-
item1 := mocks.NewMockQueueItemAccessor(100, "item1_lifo", flowKey)
217-
item1.EnqueueTimeV = now.Add(-2 * time.Second) // Earliest
218-
item2 := mocks.NewMockQueueItemAccessor(50, "item2_lifo", flowKey)
219-
item2.EnqueueTimeV = now.Add(-1 * time.Second) // Middle
220-
item3 := mocks.NewMockQueueItemAccessor(20, "item3_lifo", flowKey)
221-
item3.EnqueueTimeV = now // Latest
222-
223-
itemsInLIFOOrder := []*mocks.MockQueueItemAccessor{item3, item2, item1}
224-
testLifecycleAndOrdering(t, q, itemsInLIFOOrder, "PriorityLIFO")
225-
})
226-
}
192+
t.Run("LifecycleAndOrdering_PriorityConfigurable_ByteSize", func(t *testing.T) {
193+
t.Parallel()
194+
q, err := constructor(byteSizePolicy)
195+
require.NoError(t, err, "Setup: creating queue with byteSizePolicy should not fail")
196+
197+
itemLarge := mocks.NewMockQueueItemAccessor(100, "itemLarge_prio", flowKey)
198+
itemSmall := mocks.NewMockQueueItemAccessor(20, "itemSmall_prio", flowKey)
199+
itemMedium := mocks.NewMockQueueItemAccessor(50, "itemMedium_prio", flowKey)
200+
201+
itemsInByteSizeOrder := []*mocks.MockQueueItemAccessor{itemSmall, itemMedium, itemLarge}
202+
testLifecycleAndOrdering(t, q, itemsInByteSizeOrder, "PriorityByteSize")
203+
})
204+
205+
t.Run("LifecycleAndOrdering_PriorityConfigurable_LIFO", func(t *testing.T) {
206+
t.Parallel()
207+
q, err := constructor(reverseEnqueueTimePolicy)
208+
require.NoError(t, err, "Setup: creating queue with reverseEnqueueTimePolicy should not fail")
209+
210+
now := time.Now()
211+
item1 := mocks.NewMockQueueItemAccessor(100, "item1_lifo", flowKey)
212+
item1.EnqueueTimeV = now.Add(-2 * time.Second) // Earliest
213+
item2 := mocks.NewMockQueueItemAccessor(50, "item2_lifo", flowKey)
214+
item2.EnqueueTimeV = now.Add(-1 * time.Second) // Middle
215+
item3 := mocks.NewMockQueueItemAccessor(20, "item3_lifo", flowKey)
216+
item3.EnqueueTimeV = now // Latest
217+
218+
itemsInLIFOOrder := []*mocks.MockQueueItemAccessor{item3, item2, item1}
219+
testLifecycleAndOrdering(t, q, itemsInLIFOOrder, "PriorityLIFO")
220+
})
227221

228222
t.Run("Remove_InvalidHandle", func(t *testing.T) {
229223
t.Parallel()

‎pkg/epp/flowcontrol/framework/plugins/queue/listqueue.go‎

Lines changed: 0 additions & 215 deletions
This file was deleted.

‎pkg/epp/flowcontrol/framework/plugins/queue/priorityqueue.go‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ import (
2929
//
3030
// This queue provides a concurrent-safe priority queue whose ordering is maintained by an internal
3131
// container/heap. Items are ordered by the configured OrderingPolicy, with the highest-priority
32-
// item (per the policy) at the head. It advertises the CapabilityPriorityConfigurable capability.
32+
// item (per the policy) at the head.
3333
//
3434
// Each item's position in the heap is tracked on its handle, enabling O(log n) targeted removal.
3535
const PriorityQueueName = "PriorityQueue"
@@ -125,11 +125,6 @@ func (pq *priorityQueue) Name() string {
125125
return PriorityQueueName
126126
}
127127

128-
// Capabilities returns the capabilities of the queue.
129-
func (pq *priorityQueue) Capabilities() []flowcontrol.QueueCapability {
130-
return []flowcontrol.QueueCapability{flowcontrol.CapabilityPriorityConfigurable}
131-
}
132-
133128
// Len returns the number of items in the queue.
134129
func (pq *priorityQueue) Len() int {
135130
pq.mu.RLock()

0 commit comments

Comments
 (0)