This is an automated email from the ASF dual-hosted git repository.
Similarityoung pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go-pixiu.git
The following commit(s) were added to refs/heads/develop by this push:
new 55d841c94 perf(llm): make cooldown store eviction O(1) with
container/list LRU (#966)
55d841c94 is described below
commit 55d841c94d2af8e9ae55a02430cf5b29ec056189
Author: 承潜 <[email protected]>
AuthorDate: Sat Aug 15 14:08:30 2026 +0800
perf(llm): make cooldown store eviction O(1) with container/list LRU (#966)
* perf(llm): make cooldown store eviction O(1) with container/list LRU
The bounded cooldown store evicted its oldest entry by scanning the
whole map to find the minimum lastFailure, so eviction on the failure
path was O(N) in the number of tracked endpoints.
Back the store with a container/list ordered oldest-at-front,
newest-at-back, and key the map to *list.Element. markFailure now moves
a refreshed entry to the back and inserts new entries at the back;
eviction pops the front in O(1). Each cooldownEntry carries its own key
so a front pop can delete the matching map entry without a reverse
lookup. A shared removeLocked helper keeps the map and list in sync on
every delete, sweep, and eviction, so no stale list element survives.
Caller-visible cooldown behavior (key, TTL, idempotent replay, lazy
sweep) is unchanged. No new dependency, no background goroutine.
New tests cover the LRU-specific invariants: refresh updates recency,
the TTL sweep clears both map and list state, and deleting an expired
cooldown leaves no stale list element.
Closes #943
* refactor(llm): drop dead eviction guard in cooldown store
evictOldestIfFullLocked is only called for brand-new keys, so the
oldestKey == current guard can never fire. Worse, if it ever did it
would skip eviction and let markFailure push the store past capacity.
Remove the guard and the now-unused current parameter.
* perf(llm): inject nowFn into cooldownStore to fix timestamp/list ordering
under concurrency
markFailure previously sampled time.Now() before acquiring the mutex, so
concurrent callers could record a newer timestamp but insert earlier in the
recency list, causing eviction to drop a still-active cooldown entry.
Fix: add nowFn func() time.Time to cooldownStore (default time.Now), called
inside the mutex so timestamp and PushBack order are always consistent.
markFailure signature drops the lastFailure parameter; callers no longer
supply a timestamp. Tests inject a controlled clock via store.nowFn.
---
pkg/filter/llm/proxy/filter.go | 93 ++++++++++++++++-----------
pkg/filter/llm/proxy/filter_test.go | 121 +++++++++++++++++++++++++++++++++---
2 files changed, 168 insertions(+), 46 deletions(-)
diff --git a/pkg/filter/llm/proxy/filter.go b/pkg/filter/llm/proxy/filter.go
index 5118a67b4..7783c2f6b 100644
--- a/pkg/filter/llm/proxy/filter.go
+++ b/pkg/filter/llm/proxy/filter.go
@@ -19,6 +19,7 @@ package proxy
import (
"bytes"
+ "container/list"
"crypto/sha256"
"errors"
"fmt"
@@ -127,12 +128,24 @@ type (
}
cooldownStore struct {
- mu sync.Mutex
- lastFailureByEndpoint map[cooldownKey]cooldownEntry
- lastSweep time.Time
+ mu sync.Mutex
+ // lastFailureByEndpoint indexes each tracked endpoint to its
node in
+ // recencyOrder, so lookups stay O(1) and the node can be
repositioned
+ // or removed without scanning.
+ lastFailureByEndpoint map[cooldownKey]*list.Element
+ // recencyOrder holds *cooldownEntry values ordered
oldest-at-front,
+ // newest-at-back. Eviction pops the front in O(1); refreshing
an entry
+ // moves its node to the back.
+ recencyOrder *list.List
+ lastSweep time.Time
+ // nowFn is called inside the mutex so the recorded timestamp
and the
+ // recency-list insertion order are always consistent, even
under
+ // concurrent writes. Overridable in tests.
+ nowFn func() time.Time
}
cooldownEntry struct {
+ key cooldownKey
lastFailure time.Time
ttl time.Duration
}
@@ -459,12 +472,14 @@ func (executor *RequestExecutor)
markEndpointCooldown(endpoint *model.Endpoint)
if store == nil || endpoint == nil {
return
}
- store.markFailure(executor.clusterName, endpoint, time.Now())
+ store.markFailure(executor.clusterName, endpoint)
}
func newCooldownStore() *cooldownStore {
return &cooldownStore{
- lastFailureByEndpoint: map[cooldownKey]cooldownEntry{},
+ lastFailureByEndpoint: map[cooldownKey]*list.Element{},
+ recencyOrder: list.New(),
+ nowFn: time.Now,
}
}
@@ -477,33 +492,41 @@ func (s *cooldownStore)
lastFailureWithCurrentTTL(clusterName string, endpoint *
key := newCooldownKey(clusterName, endpoint)
now := time.Now()
s.sweepExpiredIfNeededLocked(now, key)
- entry, ok := s.lastFailureByEndpoint[key]
+ element, ok := s.lastFailureByEndpoint[key]
if !ok {
return time.Time{}, 0, false
}
+ entry := element.Value.(*cooldownEntry)
currentTTL := endpointCooldownInterval(endpoint)
if entry.ttl != currentTTL {
entry.ttl = currentTTL
- s.lastFailureByEndpoint[key] = entry
}
return entry.lastFailure, entry.ttl, true
}
-func (s *cooldownStore) markFailure(clusterName string, endpoint
*model.Endpoint, lastFailure time.Time) {
+func (s *cooldownStore) markFailure(clusterName string, endpoint
*model.Endpoint) {
if s == nil || endpoint == nil {
return
}
s.mu.Lock()
defer s.mu.Unlock()
+ lastFailure := s.nowFn()
key := newCooldownKey(clusterName, endpoint)
s.sweepExpiredIfNeededLocked(time.Now(), key)
- if _, ok := s.lastFailureByEndpoint[key]; !ok {
- s.evictOldestIfFullLocked(key)
+ ttl := endpointCooldownInterval(endpoint)
+ if element, ok := s.lastFailureByEndpoint[key]; ok {
+ entry := element.Value.(*cooldownEntry)
+ entry.lastFailure = lastFailure
+ entry.ttl = ttl
+ s.recencyOrder.MoveToBack(element)
+ return
}
- s.lastFailureByEndpoint[key] = cooldownEntry{
+ s.evictOldestIfFullLocked()
+ s.lastFailureByEndpoint[key] = s.recencyOrder.PushBack(&cooldownEntry{
+ key: key,
lastFailure: lastFailure,
- ttl: endpointCooldownInterval(endpoint),
- }
+ ttl: ttl,
+ })
}
func (s *cooldownStore) deleteLastFailureIfMatches(clusterName string,
endpoint *model.Endpoint, expected time.Time) bool {
@@ -513,14 +536,21 @@ func (s *cooldownStore)
deleteLastFailureIfMatches(clusterName string, endpoint
s.mu.Lock()
defer s.mu.Unlock()
key := newCooldownKey(clusterName, endpoint)
- current, ok := s.lastFailureByEndpoint[key]
- if !ok || current.lastFailure != expected {
+ element, ok := s.lastFailureByEndpoint[key]
+ if !ok || element.Value.(*cooldownEntry).lastFailure != expected {
return false
}
- delete(s.lastFailureByEndpoint, key)
+ s.removeLocked(key, element)
return true
}
+// removeLocked drops one entry from both the map and the recency list,
+// keeping the two structures in sync. Callers must hold s.mu.
+func (s *cooldownStore) removeLocked(key cooldownKey, element *list.Element) {
+ delete(s.lastFailureByEndpoint, key)
+ s.recencyOrder.Remove(element)
+}
+
func (s *cooldownStore) sweepExpiredIfNeededLocked(now time.Time, current
cooldownKey) {
if len(s.lastFailureByEndpoint) < cooldownStoreSweepAt &&
!s.lastSweep.IsZero() &&
@@ -532,39 +562,28 @@ func (s *cooldownStore) sweepExpiredIfNeededLocked(now
time.Time, current cooldo
}
func (s *cooldownStore) sweepExpiredExceptLocked(now time.Time, current
cooldownKey) {
- for key, entry := range s.lastFailureByEndpoint {
+ for key, element := range s.lastFailureByEndpoint {
if key == current {
continue
}
+ entry := element.Value.(*cooldownEntry)
if now.Sub(entry.lastFailure) >= entry.ttl {
- delete(s.lastFailureByEndpoint, key)
+ s.removeLocked(key, element)
}
}
}
-func (s *cooldownStore) evictOldestIfFullLocked(current cooldownKey) {
+// evictOldestIfFullLocked removes the least-recently-failed entry when the
+// store is at capacity, in O(1) via the front of the recency list.
+func (s *cooldownStore) evictOldestIfFullLocked() {
if len(s.lastFailureByEndpoint) < maxCooldownStoreEntries {
return
}
-
- var (
- oldestKey cooldownKey
- oldestEntry cooldownEntry
- found bool
- )
- for key, entry := range s.lastFailureByEndpoint {
- if key == current {
- continue
- }
- if !found || entry.lastFailure.Before(oldestEntry.lastFailure) {
- oldestKey = key
- oldestEntry = entry
- found = true
- }
- }
- if found {
- delete(s.lastFailureByEndpoint, oldestKey)
+ oldest := s.recencyOrder.Front()
+ if oldest == nil {
+ return
}
+ s.removeLocked(oldest.Value.(*cooldownEntry).key, oldest)
}
func newCooldownKey(clusterName string, endpoint *model.Endpoint) cooldownKey {
diff --git a/pkg/filter/llm/proxy/filter_test.go
b/pkg/filter/llm/proxy/filter_test.go
index deaca8492..dc4d04e1f 100644
--- a/pkg/filter/llm/proxy/filter_test.go
+++ b/pkg/filter/llm/proxy/filter_test.go
@@ -188,7 +188,9 @@ func
TestRequestExecutorEndpointInCooldownClearsExpiredCooldownFromProxyStore(t
clusterName: clusterName,
cooldowns: store,
}
- store.markFailure(clusterName, endpoint, time.Now().Add(-time.Hour))
+ store.nowFn = func() time.Time { return time.Now().Add(-time.Hour) }
+ store.markFailure(clusterName, endpoint)
+ store.nowFn = time.Now
assert.False(t, executor.endpointInCooldown(endpoint))
@@ -269,7 +271,9 @@ func
TestRequestExecutorCooldownSurvivesNonIdentityLLMConfigChanges(t *testing.T
clusterName: clusterName,
cooldowns: store,
}
- store.markFailure(clusterName, oldEndpoint,
time.Now().Add(-50*time.Millisecond))
+ store.nowFn = func() time.Time { return time.Now().Add(-50 *
time.Millisecond) }
+ store.markFailure(clusterName, oldEndpoint)
+ store.nowFn = time.Now
assert.True(t, executor.endpointInCooldown(replacement))
}
@@ -288,7 +292,9 @@ func
TestCooldownStoreLazySweepKeepsEntryAfterEndpointIntervalExtends(t *testing
clusterName: clusterName,
cooldowns: store,
}
- store.markFailure(clusterName, oldEndpoint,
time.Now().Add(-50*time.Millisecond))
+ store.nowFn = func() time.Time { return time.Now().Add(-50 *
time.Millisecond) }
+ store.markFailure(clusterName, oldEndpoint)
+ store.nowFn = time.Now
assert.True(t, executor.endpointInCooldown(replacement))
_, _, _ = store.lastFailureWithCurrentTTL(clusterName, activeEndpoint)
@@ -310,8 +316,10 @@ func
TestCooldownStoreLazySweepRemovesExpiredChurnedEndpointEntry(t *testing.T)
activeEndpoint := testLLMEndpoint("ep-2", 18086)
store := newCooldownStore()
- store.markFailure(clusterName, oldEndpoint, time.Now().Add(-time.Hour))
- store.markFailure(clusterName, movedEndpoint, time.Now())
+ store.nowFn = func() time.Time { return time.Now().Add(-time.Hour) }
+ store.markFailure(clusterName, oldEndpoint)
+ store.nowFn = time.Now
+ store.markFailure(clusterName, movedEndpoint)
store.mu.Lock()
_, oldExistsAfterMove :=
store.lastFailureByEndpoint[newCooldownKey(clusterName, oldEndpoint)]
@@ -336,7 +344,9 @@ func
TestCooldownStoreLazySweepRemovesExpiredEndpointFromDifferentCluster(t *tes
activeEndpoint := testLLMEndpoint("ep-2", 18093)
store := newCooldownStore()
- store.markFailure("old-cluster", expiredEndpoint,
time.Now().Add(-time.Hour))
+ store.nowFn = func() time.Time { return time.Now().Add(-time.Hour) }
+ store.markFailure("old-cluster", expiredEndpoint)
+ store.nowFn = time.Now
store.mu.Lock()
store.lastSweep = time.Now().Add(-cooldownStoreSweepAfter -
time.Millisecond)
store.mu.Unlock()
@@ -351,6 +361,8 @@ func
TestCooldownStoreLazySweepRemovesExpiredEndpointFromDifferentCluster(t *tes
func TestCooldownStoreEvictsOldestEntryWhenCapacityExceeded(t *testing.T) {
store := newCooldownStore()
now := time.Now()
+ current := now
+ store.nowFn = func() time.Time { return current }
oldestEndpoint := testLLMEndpoint("ep-0", 19000)
for i := 0; i < maxCooldownStoreEntries; i++ {
@@ -358,11 +370,13 @@ func
TestCooldownStoreEvictsOldestEntryWhenCapacityExceeded(t *testing.T) {
if i == 0 {
oldestEndpoint = endpoint
}
- store.markFailure("capacity-cluster", endpoint,
now.Add(time.Duration(i)*time.Millisecond))
+ current = now.Add(time.Duration(i) * time.Millisecond)
+ store.markFailure("capacity-cluster", endpoint)
}
newestEndpoint := testLLMEndpoint("ep-new", 21000)
- store.markFailure("capacity-cluster", newestEndpoint,
now.Add(time.Hour))
+ current = now.Add(time.Hour)
+ store.markFailure("capacity-cluster", newestEndpoint)
store.mu.Lock()
_, oldestExists :=
store.lastFailureByEndpoint[newCooldownKey("capacity-cluster", oldestEndpoint)]
@@ -375,6 +389,95 @@ func
TestCooldownStoreEvictsOldestEntryWhenCapacityExceeded(t *testing.T) {
assert.Equal(t, maxCooldownStoreEntries, entryCount)
}
+// TestCooldownStoreRefreshUpdatesRecency verifies that re-failing an existing
+// endpoint moves it to the newest side of the LRU, so a later eviction drops a
+// genuinely older entry instead of the just-refreshed one.
+func TestCooldownStoreRefreshUpdatesRecency(t *testing.T) {
+ store := newCooldownStore()
+ now := time.Now()
+ current := now
+ store.nowFn = func() time.Time { return current }
+
+ first := testLLMEndpoint("ep-first", 19000)
+ second := testLLMEndpoint("ep-second", 19001)
+ store.markFailure("recency-cluster", first)
+ current = now.Add(time.Millisecond)
+ store.markFailure("recency-cluster", second)
+
+ // Refresh the first endpoint so it becomes the most recently failed.
+ current = now.Add(2 * time.Millisecond)
+ store.markFailure("recency-cluster", first)
+
+ store.mu.Lock()
+ front := store.recencyOrder.Front().Value.(*cooldownEntry)
+ back := store.recencyOrder.Back().Value.(*cooldownEntry)
+ listLen := store.recencyOrder.Len()
+ mapLen := len(store.lastFailureByEndpoint)
+ store.mu.Unlock()
+
+ assert.Equal(t, newCooldownKey("recency-cluster", second), front.key,
"least-recent should be the un-refreshed entry")
+ assert.Equal(t, newCooldownKey("recency-cluster", first), back.key,
"most-recent should be the refreshed entry")
+ assert.Equal(t, 2, listLen)
+ assert.Equal(t, mapLen, listLen, "map and recency list must stay in
sync")
+}
+
+// TestCooldownStoreSweepRemovesMapAndListState verifies the lazy TTL sweep
+// drops an expired entry from both the map and the recency list, leaving no
+// stale list element behind.
+func TestCooldownStoreSweepRemovesMapAndListState(t *testing.T) {
+ clusterName := "sweep-cluster"
+ expiredEndpoint := testLLMEndpoint("ep-expired", 19000)
+ activeEndpoint := testLLMEndpoint("ep-active", 19001)
+ store := newCooldownStore()
+
+ store.nowFn = func() time.Time { return time.Now().Add(-time.Hour) }
+ store.markFailure(clusterName, expiredEndpoint)
+ store.nowFn = time.Now
+ store.mu.Lock()
+ store.lastSweep = time.Now().Add(-cooldownStoreSweepAfter -
time.Millisecond)
+ store.mu.Unlock()
+
+ // Touching an unrelated endpoint triggers the lazy sweep.
+ store.lastFailureWithCurrentTTL(clusterName, activeEndpoint)
+
+ store.mu.Lock()
+ _, expiredExists :=
store.lastFailureByEndpoint[newCooldownKey(clusterName, expiredEndpoint)]
+ listLen := store.recencyOrder.Len()
+ mapLen := len(store.lastFailureByEndpoint)
+ store.mu.Unlock()
+
+ assert.False(t, expiredExists)
+ assert.Equal(t, 0, listLen, "expired entry must be removed from the
recency list, not just the map")
+ assert.Equal(t, mapLen, listLen, "map and recency list must stay in
sync")
+}
+
+// TestCooldownStoreDeleteExpiredLeavesNoStaleListElement verifies that
clearing
+// an expired cooldown via the request path removes both the map entry and its
+// recency list element.
+func TestCooldownStoreDeleteExpiredLeavesNoStaleListElement(t *testing.T) {
+ clusterName := "delete-expired-cluster"
+ endpoint := testLLMEndpoint("ep-1", 19000)
+ store := newCooldownStore()
+ executor := &RequestExecutor{
+ clusterName: clusterName,
+ cooldowns: store,
+ }
+ store.nowFn = func() time.Time { return time.Now().Add(-time.Hour) }
+ store.markFailure(clusterName, endpoint)
+ store.nowFn = time.Now
+
+ // endpointInCooldown observes the entry as expired and deletes it.
+ assert.False(t, executor.endpointInCooldown(endpoint))
+
+ store.mu.Lock()
+ mapLen := len(store.lastFailureByEndpoint)
+ listLen := store.recencyOrder.Len()
+ store.mu.Unlock()
+
+ assert.Equal(t, 0, mapLen)
+ assert.Equal(t, 0, listLen, "deleting an expired cooldown must not
leave a stale list element")
+}
+
func TestStrategyExecuteIgnoresUnhealthyPreferredEndpoint(t *testing.T) {
clusterName := "llm-preferred-health"
healthyEndpoint := testLLMEndpoint("ep-1", 18086)
@@ -468,7 +571,7 @@ func BenchmarkCooldown_EndpointInCooldown(b *testing.B) {
endpoint.LLMMeta.APIKey = fmt.Sprintf("api-key-%d", i)
endpoint.LLMMeta.HealthCheckInterval = cooldownTTLMillis
endpoints[i] = endpoint
- store.markFailure(clusterName, endpoint, time.Now())
+ store.markFailure(clusterName, endpoint)
}
b.ReportAllocs()
