From a7e07541f42480fc1ca716b3d142f507a796ff10 Mon Sep 17 00:00:00 2001 From: YaoZengzeng Date: Fri, 10 Jul 2026 06:35:49 +0000 Subject: [PATCH 1/5] kthena-router: boost newest same-session request to queue head In session-boost mode, boosted requests were dequeued FIFO among themselves. Change the ordering so the most-recently-arrived boosted request wins (LIFO): a freshly boosted follow-up always jumps to the head of the queue so it reuses the still-warm prefix cache before earlier boosted requests. Non-boosted requests keep FIFO ordering, and boosted requests still outrank non-boosted ones. Also remove the grace-period 'head already boosted' fast path so the freed slot is held for the full grace window, allowing a still-newer same-session follow-up to arrive and take the head. Docs updated accordingly. Signed-off-by: YaoZengzeng --- docs/kthena/docs/user-guide/session-boost.md | 21 +++++----- docs/proposal/session-boost-strategy.md | 41 ++++++++++--------- pkg/kthena-router/datastore/fairness_queue.go | 8 +++- .../datastore/session_boost_queue.go | 26 +++--------- .../datastore/session_boost_queue_test.go | 11 ++--- 5 files changed, 52 insertions(+), 55 deletions(-) diff --git a/docs/kthena/docs/user-guide/session-boost.md b/docs/kthena/docs/user-guide/session-boost.md index d12b2dc44..f7738659e 100644 --- a/docs/kthena/docs/user-guide/session-boost.md +++ b/docs/kthena/docs/user-guide/session-boost.md @@ -169,7 +169,8 @@ This design is intentional: inference engines such as vLLM evict their prefix (K The session boost queue uses a simple two-level priority: 1. **Boosted requests** (session tracked in the LRU cache) are always dequeued before non-boosted requests. -2. **Within the same boost level**, requests are served in FIFO order (earliest arrival first). +2. **Within the boosted group**, the most-recently-arrived boosted request wins (LIFO): a freshly boosted follow-up always jumps to the head of the queue rather than joining the back of the boosted group. This keeps the newest same-session request first so it reuses the still-warm prefix cache. +3. **Among non-boosted requests**, ordering is FIFO (earliest arrival first). ### Backpressure Control @@ -197,13 +198,13 @@ If a request waits in the queue longer than `SESSION_BOOST_TIMEOUT`, the router Session boost and user fairness are two mutually exclusive scheduling strategies for the per-model request queue; they configure that queue for different goals: -| Aspect | Session Boost | User Fairness | -| ---------------- | ---------------------------------- | --------------------------------- | -| Goal | Maximize prefix cache hits | Equitable resource allocation | -| Activation | `ENABLE_SESSION_BOOST=true` | `ENABLE_FAIRNESS_SCHEDULING=true` | -| Requires user ID | No | Yes | -| Priority logic | Boosted > non-boosted, FIFO within | Lower recent usage wins | -| Best for | Multi-turn latency optimization | Multi-tenant contention | +| Aspect | Session Boost | User Fairness | +| ---------------- | ----------------------------------------- | --------------------------------- | +| Goal | Maximize prefix cache hits | Equitable resource allocation | +| Activation | `ENABLE_SESSION_BOOST=true` | `ENABLE_FAIRNESS_SCHEDULING=true` | +| Requires user ID | No | Yes | +| Priority logic | Boosted > non-boosted; newest boost first | Lower recent usage wins | +| Best for | Multi-turn latency optimization | Multi-tenant contention | The two strategies are **mutually exclusive**; enabling both is a configuration error. Enable one or the other based on your primary scheduling concern. @@ -286,9 +287,9 @@ The grace period is an **advanced, scenario-specific** tuning knob. It is **disa By default, when a request completes the queue immediately dequeues the next request. The grace period (`SESSION_BOOST_GRACE_PERIOD`) instead makes the queue **briefly hold the freed dequeue slot** after a completion, waiting for a follow-up from the *same* session to arrive so it can ride the still-warm prefix cache: -- If a boosted (same-session) request arrives during the grace window, it is admitted first when the window ends, because boosted requests outrank others in the queue. +- If a boosted (same-session) request arrives during the grace window, it is admitted first when the window ends, because boosted requests outrank others in the queue—and among boosted requests the most-recently-arrived one is at the head. - If no boosted request arrives before the window expires, the next non-boosted request proceeds as normal. -- If the head of the queue is already a boosted request, the grace period is skipped entirely. +- The queue always waits for the full grace window even when a boosted request is already at the head, because a still-newer same-session follow-up may yet arrive and should be the one to ride the warm prefix cache. ### When it might help diff --git a/docs/proposal/session-boost-strategy.md b/docs/proposal/session-boost-strategy.md index 05b9b12eb..9432ec077 100644 --- a/docs/proposal/session-boost-strategy.md +++ b/docs/proposal/session-boost-strategy.md @@ -98,9 +98,9 @@ Session boost is a priority strategy of the shared per-model request priority qu │ │ │ │ │ │ Ordering: │ │ │ │ 1. SessionBoost=true > false │ │ -│ │ 2. Within same boost: FIFO │ │ +│ │ 2. Within boosted: newest first │ │ │ │ │ │ -│ │ [Boosted-1] [Boosted-2] [Normal-1] │ │ +│ │ [Boosted-new] [Boosted-old] [Normal]│ │ │ └──────────────────────────────────────┘ │ │ │ │ │ ▼ │ @@ -159,14 +159,14 @@ Timeline: ▼ ▼ ▼ ─────┼──┼──────────────────────────────────┼───── │ │ │ - │ │ Case A: Head already boosted → │ - │ │ skip grace, dequeue immediately │ - │ │ │ - │ │ Case B: Otherwise hold the slot │ - │ │ for the full grace period, then │ - │ │ dequeue the head (boosted ranks │ - │ │ first, so a follow-up that │ - │ │ arrived wins automatically) │ + │ │ Hold the slot for the full │ + │ │ grace period, then dequeue the │ + │ │ head. Boosted requests rank │ + │ │ first and, among boosted, the │ + │ │ newest-arrived wins, so a │ + │ │ same-session follow-up that │ + │ │ arrived during the window is │ + │ │ dequeued first automatically. │ │ └────────────────────────────────┐ ``` @@ -178,16 +178,17 @@ It is important that grace is **triggered only by release events**, because a re 1. A request finishes → `Release()` runs → `inflightCount` is decremented → a signal is sent on `releaseCh`. 2. The freed slot would normally be claimed immediately by the current heap head (which may be an unrelated, non-boosted request). The grace wait instead **holds that just-freed slot for up to `SessionBoostGracePeriod`**, betting that a same-session follow-up is about to arrive and reuse the warm prefix cache. -3. When the wait resolves, `tryBackpressureDequeue` runs and re-checks **both** capacity gates before admitting anyone. A boosted follow-up that arrived during grace is admitted first because boosted requests outrank others in the heap, and only when `inflight < MaxConcurrent` **and** at least one backend pod reports capacity. +3. When the wait resolves, `tryBackpressureDequeue` runs and re-checks **both** capacity gates before admitting anyone. A boosted follow-up that arrived during grace is admitted first because boosted requests outrank others in the heap and, among boosted requests, the most-recently-arrived one is at the head — and only when `inflight < MaxConcurrent` **and** at least one backend pod reports capacity. + +The grace wait always runs for the full window (subject to shutdown), and in every case the two capacity gates are the final arbiter: -The grace wait can resolve in three ways, and in every case the two capacity gates are the final arbiter: +| Outcome | Trigger | What happens next | +| ----------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| **Grace expires** | The timer fires (`timer.C`). | Stop holding the slot and fall through to the capacity gates, which admit the heap head. If a same-session follow-up arrived during the wait it is now boosted and sits at the head (newest boosted first), so it wins automatically (subject to inflight + backend capacity). | -| Outcome | Trigger | What happens next | -| -------------------------- | --------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Skip grace (fast path)** | The heap head is *already* session-boosted when the release fires (`isHeadSessionBoosted()`). | No wait at all — go straight to the capacity gates and admit if they pass. There is nothing to wait for. | -| **Grace expires** | The timer fires (`timer.C`). | Stop holding the slot and fall through to the capacity gates, which admit the heap head. If a same-session follow-up arrived during the wait it is now boosted and sits at the head, so it wins automatically (subject to inflight + backend capacity). | +> There is no "head already boosted" fast path. The queue holds the freed slot for the full grace window even when a boosted request is already at the head, because a still-newer same-session follow-up may yet arrive during the window and should be the one to ride the warm prefix cache. -This ordering matters because the three checks answer different questions and run in sequence: +This ordering matters because the timing layer and the two gates answer different questions and run in sequence: ``` Release frees a slot @@ -215,7 +216,7 @@ Two interactions are worth calling out explicitly: - **Grace never overrides backpressure.** If the grace window ends but the inflight limit is already reached or every backend pod is busy, the request is **not** admitted — `tryBackpressureDequeue` simply holds (and drains any cancelled requests from the heap) until the next release or new arrival reopens the gates. Grace only chooses *who* tries next; the capacity gates decide *whether* anyone runs. - **Fresh arrivals on an idle queue bypass grace.** Grace is tied to `releaseCh` (a freed slot), not to `notifyCh` (a new arrival). When a request lands on an otherwise idle queue with no pending release, it goes straight to the capacity gates with no grace delay, so enabling grace adds no admission latency to first turns. The only place a new arrival waits is *inside* an already-running grace window, where it is precisely the boosted follow-up the window exists to catch. When both a release and a new arrival are pending at once, the release is preferred so the freed slot is the one held for the grace window. -The net effect is a strict precedence: **grace timing → inflight gate → backend gate**. The grace layer is purely additive and optional (`SessionBoostGracePeriod = 0` removes it entirely, taking the no-grace fast path), and it can only ever *delay* an admission to favor a session-boosted follow-up — it can never admit a request that the inflight or backend gates would otherwise reject. +The net effect is a strict precedence: **grace timing → inflight gate → backend gate**. The grace layer is purely additive and optional (`SessionBoostGracePeriod = 0` removes it entirely, dequeuing immediately on each release), and it can only ever *delay* an admission to favor a session-boosted follow-up — it can never admit a request that the inflight or backend gates would otherwise reject. #### Configuration @@ -266,7 +267,9 @@ type RequestPriorityQueue struct { // Session-boost ordering (RequestPriorityQueue.Less when sessionBoost == true): // 1. SessionBoost=true comes before SessionBoost=false -// 2. Within same boost status: earlier RequestTime comes first (FIFO) +// 2. Within boosted requests: later RequestTime comes first (newest-first / LIFO), +// so a freshly boosted follow-up jumps to the head +// 3. Within non-boosted requests: earlier RequestTime comes first (FIFO) ``` #### Session Identification diff --git a/pkg/kthena-router/datastore/fairness_queue.go b/pkg/kthena-router/datastore/fairness_queue.go index 099887fc2..f0ff232ac 100644 --- a/pkg/kthena-router/datastore/fairness_queue.go +++ b/pkg/kthena-router/datastore/fairness_queue.go @@ -273,11 +273,17 @@ func NewRequestPriorityQueueWithConfig(metricsInstance *metrics.Metrics, cfg Fai func (pq *RequestPriorityQueue) Len() int { return len(pq.heap) } func (pq *RequestPriorityQueue) Less(i, j int) bool { - // Session-boost mode: boosted requests outrank others; ties broken FIFO. + // Session-boost mode: boosted requests outrank others. Among boosted requests + // the most-recently-arrived one wins (LIFO): a freshly boosted follow-up always + // jumps to the head so it can reuse the still-warm prefix cache before earlier + // boosted requests. Non-boosted requests keep FIFO ordering. if pq.sessionBoost { if pq.heap[i].SessionBoost != pq.heap[j].SessionBoost { return pq.heap[i].SessionBoost } + if pq.heap[i].SessionBoost { + return pq.heap[i].RequestTime.After(pq.heap[j].RequestTime) + } return pq.heap[i].RequestTime.Before(pq.heap[j].RequestTime) } // same user, FIFO diff --git a/pkg/kthena-router/datastore/session_boost_queue.go b/pkg/kthena-router/datastore/session_boost_queue.go index 95ee4d5dd..54e31aa9c 100644 --- a/pkg/kthena-router/datastore/session_boost_queue.go +++ b/pkg/kthena-router/datastore/session_boost_queue.go @@ -233,30 +233,16 @@ func (pq *RequestPriorityQueue) drainPendingRelease() bool { } } -// isHeadSessionBoosted checks if the highest-priority request in the queue has a session boost. -func (pq *RequestPriorityQueue) isHeadSessionBoosted() bool { - pq.mu.RLock() - defer pq.mu.RUnlock() - if len(pq.heap) == 0 { - return false - } - return pq.heap[0].SessionBoost -} - // waitGraceAndDequeue holds a just-freed slot for up to SessionBoostGracePeriod // before dispatching, giving a same-session follow-up time to arrive. It does not // need to watch for the follow-up itself: boosted requests outrank others in the -// heap, so once the timer fires tryBackpressureDequeue admits the boosted request -// first if one showed up. The wait stays responsive to shutdown via ctx/stopCh. +// heap and, among boosted requests, the most-recently-arrived one wins, so once +// the timer fires tryBackpressureDequeue admits the newest boosted follow-up first +// if one showed up. The wait always runs for the full window even when the head is +// already boosted, because a still-newer same-session follow-up may yet arrive and +// should be the one to ride the warm prefix cache. The wait stays responsive to +// shutdown via ctx/stopCh. func (pq *RequestPriorityQueue) waitGraceAndDequeue(ctx context.Context) { - // Fast path: a boosted follow-up is already at the head, so there is nothing - // to wait for. - if pq.isHeadSessionBoosted() { - klog.V(4).Info("[SessionBoost] grace: head already boosted, skipping wait") - pq.tryBackpressureDequeue(ctx) - return - } - klog.V(4).Infof("[SessionBoost] grace: holding freed slot for %v", pq.config.SessionBoostGracePeriod) timer := time.NewTimer(pq.config.SessionBoostGracePeriod) defer timer.Stop() diff --git a/pkg/kthena-router/datastore/session_boost_queue_test.go b/pkg/kthena-router/datastore/session_boost_queue_test.go index 30859b3e2..e99235298 100644 --- a/pkg/kthena-router/datastore/session_boost_queue_test.go +++ b/pkg/kthena-router/datastore/session_boost_queue_test.go @@ -217,12 +217,13 @@ func TestSessionBoostQueue_MultipleSessions(t *testing.T) { t.Errorf("Last two should not be boosted: third=%v fourth=%v", third.SessionBoost, fourth.SessionBoost) } - // Among boosted: FIFO order (boost-A arrived before boost-B) - if first.UserID != "u2" { - t.Errorf("Expected boost-A first (earlier arrival), got %s", first.UserID) + // Among boosted: newest arrival wins (LIFO), so boost-B (later arrival) is + // dequeued before boost-A. + if first.UserID != "u3" { + t.Errorf("Expected boost-B first (latest arrival), got %s", first.UserID) } - if second.UserID != "u3" { - t.Errorf("Expected boost-B second, got %s", second.UserID) + if second.UserID != "u2" { + t.Errorf("Expected boost-A second, got %s", second.UserID) } // Among normal: FIFO order From 3d3ce0c912380f8594ccd1149a37593730f1033b Mon Sep 17 00:00:00 2001 From: YaoZengzeng Date: Fri, 10 Jul 2026 07:42:40 +0000 Subject: [PATCH 2/5] kthena-router: order boosted requests by session completion time Replace the LIFO-by-arrival ordering of boosted requests with an ordering based on each session's completion time: the session whose previous turn completed most recently is served first, because its prefix (KV) cache is the most likely to still be warm on the backend. Unlike plain LIFO, a session's completion time is fixed once its turn finishes, so a waiting boosted request cannot be starved by a stream of newer arrivals. Boosted requests still always outrank non-boosted ones, and non-boosted requests keep FIFO ordering. SessionTracker now stores the completion timestamp per session (LRU value) and exposes CompletionTime; PushRequest captures it into the new Request.SessionCompletedAt field used by Less. Signed-off-by: YaoZengzeng --- docs/kthena/docs/user-guide/session-boost.md | 20 ++++---- docs/proposal/session-boost-strategy.md | 24 ++++----- pkg/kthena-router/datastore/fairness_queue.go | 35 ++++++++----- .../datastore/session_boost_queue.go | 41 ++++++++++------ .../datastore/session_boost_queue_test.go | 49 +++++++++++++++++-- 5 files changed, 120 insertions(+), 49 deletions(-) diff --git a/docs/kthena/docs/user-guide/session-boost.md b/docs/kthena/docs/user-guide/session-boost.md index f7738659e..77a520f46 100644 --- a/docs/kthena/docs/user-guide/session-boost.md +++ b/docs/kthena/docs/user-guide/session-boost.md @@ -169,9 +169,11 @@ This design is intentional: inference engines such as vLLM evict their prefix (K The session boost queue uses a simple two-level priority: 1. **Boosted requests** (session tracked in the LRU cache) are always dequeued before non-boosted requests. -2. **Within the boosted group**, the most-recently-arrived boosted request wins (LIFO): a freshly boosted follow-up always jumps to the head of the queue rather than joining the back of the boosted group. This keeps the newest same-session request first so it reuses the still-warm prefix cache. +2. **Within the boosted group**, requests are ordered by their session's **completion time**: the session whose previous turn completed most recently is served first. Its prefix (KV) cache is the most likely to still be warm on the backend, so serving it first maximizes cache reuse. This ordering does not depend on when the follow-up request arrived, only on how recently the session's last turn finished; ties are broken FIFO by arrival time. 3. **Among non-boosted requests**, ordering is FIFO (earliest arrival first). +Ordering by completion time (rather than a plain LIFO on arrival) keeps the queue fair: a boosted request cannot be starved by newer arrivals, because a session's completion time is fixed once its turn finishes and does not keep getting pushed back. Every boosted request also still leapfrogs the entire non-boosted tier, so a follow-up turn is always prioritized over unrelated traffic. + ### Backpressure Control The queue uses two-level admission control to avoid flooding backends: @@ -198,13 +200,13 @@ If a request waits in the queue longer than `SESSION_BOOST_TIMEOUT`, the router Session boost and user fairness are two mutually exclusive scheduling strategies for the per-model request queue; they configure that queue for different goals: -| Aspect | Session Boost | User Fairness | -| ---------------- | ----------------------------------------- | --------------------------------- | -| Goal | Maximize prefix cache hits | Equitable resource allocation | -| Activation | `ENABLE_SESSION_BOOST=true` | `ENABLE_FAIRNESS_SCHEDULING=true` | -| Requires user ID | No | Yes | -| Priority logic | Boosted > non-boosted; newest boost first | Lower recent usage wins | -| Best for | Multi-turn latency optimization | Multi-tenant contention | +| Aspect | Session Boost | User Fairness | +| ---------------- | ------------------------------------------------------------ | --------------------------------- | +| Goal | Maximize prefix cache hits | Equitable resource allocation | +| Activation | `ENABLE_SESSION_BOOST=true` | `ENABLE_FAIRNESS_SCHEDULING=true` | +| Requires user ID | No | Yes | +| Priority logic | Boosted > non-boosted; most-recently-completed session first | Lower recent usage wins | +| Best for | Multi-turn latency optimization | Multi-tenant contention | The two strategies are **mutually exclusive**; enabling both is a configuration error. Enable one or the other based on your primary scheduling concern. @@ -287,7 +289,7 @@ The grace period is an **advanced, scenario-specific** tuning knob. It is **disa By default, when a request completes the queue immediately dequeues the next request. The grace period (`SESSION_BOOST_GRACE_PERIOD`) instead makes the queue **briefly hold the freed dequeue slot** after a completion, waiting for a follow-up from the *same* session to arrive so it can ride the still-warm prefix cache: -- If a boosted (same-session) request arrives during the grace window, it is admitted first when the window ends, because boosted requests outrank others in the queue—and among boosted requests the most-recently-arrived one is at the head. +- If a boosted (same-session) request arrives during the grace window, it is admitted first when the window ends, because boosted requests outrank others in the queue—and among boosted requests the one whose session completed most recently is at the head. - If no boosted request arrives before the window expires, the next non-boosted request proceeds as normal. - The queue always waits for the full grace window even when a boosted request is already at the head, because a still-newer same-session follow-up may yet arrive and should be the one to ride the warm prefix cache. diff --git a/docs/proposal/session-boost-strategy.md b/docs/proposal/session-boost-strategy.md index 9432ec077..fc9173215 100644 --- a/docs/proposal/session-boost-strategy.md +++ b/docs/proposal/session-boost-strategy.md @@ -79,11 +79,12 @@ Session boost is a priority strategy of the shared per-model request priority qu │ ┌──────────────────┐ ┌─────────────────────────┐ │ │ │ SessionTracker │<────│ MarkSessionRequest │ │ │ │ (bounded LRU) │ │ Completed() │ │ -│ │ keys: sessionID │ │ (after response sent) │ │ -│ │ cap: 4096 default│ └─────────────────────────┘ │ +│ │ key: sessionID │ │ (after response sent) │ │ +│ │ value: completeAt│ └─────────────────────────┘ │ +│ │ cap: 4096 default│ │ │ └────────┬─────────┘ │ │ │ │ -│ │ HasRecentCompletion(sessionID)? │ +│ │ CompletionTime(sessionID)? │ │ ▼ │ │ ┌──────────────────┐ │ │ │ PushRequest() │ │ @@ -98,9 +99,10 @@ Session boost is a priority strategy of the shared per-model request priority qu │ │ │ │ │ │ Ordering: │ │ │ │ 1. SessionBoost=true > false │ │ -│ │ 2. Within boosted: newest first │ │ +│ │ 2. Within boosted: most recently │ │ +│ │ completed session first │ │ │ │ │ │ -│ │ [Boosted-new] [Boosted-old] [Normal]│ │ +│ │ [Boosted-warm][Boosted-cooler][Norm]│ │ │ └──────────────────────────────────────┘ │ │ │ │ │ ▼ │ @@ -178,13 +180,13 @@ It is important that grace is **triggered only by release events**, because a re 1. A request finishes → `Release()` runs → `inflightCount` is decremented → a signal is sent on `releaseCh`. 2. The freed slot would normally be claimed immediately by the current heap head (which may be an unrelated, non-boosted request). The grace wait instead **holds that just-freed slot for up to `SessionBoostGracePeriod`**, betting that a same-session follow-up is about to arrive and reuse the warm prefix cache. -3. When the wait resolves, `tryBackpressureDequeue` runs and re-checks **both** capacity gates before admitting anyone. A boosted follow-up that arrived during grace is admitted first because boosted requests outrank others in the heap and, among boosted requests, the most-recently-arrived one is at the head — and only when `inflight < MaxConcurrent` **and** at least one backend pod reports capacity. +3. When the wait resolves, `tryBackpressureDequeue` runs and re-checks **both** capacity gates before admitting anyone. A boosted follow-up that arrived during grace is admitted first because boosted requests outrank others in the heap and, among boosted requests, the one whose session completed most recently is at the head — and only when `inflight < MaxConcurrent` **and** at least one backend pod reports capacity. The grace wait always runs for the full window (subject to shutdown), and in every case the two capacity gates are the final arbiter: -| Outcome | Trigger | What happens next | -| ----------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| **Grace expires** | The timer fires (`timer.C`). | Stop holding the slot and fall through to the capacity gates, which admit the heap head. If a same-session follow-up arrived during the wait it is now boosted and sits at the head (newest boosted first), so it wins automatically (subject to inflight + backend capacity). | +| Outcome | Trigger | What happens next | +| ----------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **Grace expires** | The timer fires (`timer.C`). | Stop holding the slot and fall through to the capacity gates, which admit the heap head. If a same-session follow-up arrived during the wait it is now boosted and sits at the head (session with the most recent completion first), so it wins automatically (subject to inflight + backend capacity). | > There is no "head already boosted" fast path. The queue holds the freed slot for the full grace window even when a boosted request is already at the head, because a still-newer same-session follow-up may yet arrive during the window and should be the one to ride the warm prefix cache. @@ -267,8 +269,8 @@ type RequestPriorityQueue struct { // Session-boost ordering (RequestPriorityQueue.Less when sessionBoost == true): // 1. SessionBoost=true comes before SessionBoost=false -// 2. Within boosted requests: later RequestTime comes first (newest-first / LIFO), -// so a freshly boosted follow-up jumps to the head +// 2. Within boosted requests: the session whose previous turn completed most +// recently comes first (warmest prefix cache); ties broken FIFO by RequestTime // 3. Within non-boosted requests: earlier RequestTime comes first (FIFO) ``` diff --git a/pkg/kthena-router/datastore/fairness_queue.go b/pkg/kthena-router/datastore/fairness_queue.go index f0ff232ac..a6a4e28a4 100644 --- a/pkg/kthena-router/datastore/fairness_queue.go +++ b/pkg/kthena-router/datastore/fairness_queue.go @@ -148,10 +148,14 @@ type Request struct { SessionID string // Session identifier for multi-turn conversations Priority float64 // Priority (lower value means higher priority) SessionBoost bool // Whether this request has session priority boost (recently completed session) - RequestTime time.Time - NotifyChan chan struct{} - CancelCh <-chan struct{} // Request-scoped cancellation signal - Release func() // Set by the queue when a permit is acquired + // SessionCompletedAt is the time the session's previous turn completed, captured + // when the request is boosted. Boosted requests are ordered by this timestamp + // (most recent first) so the session with the warmest prefix cache runs first. + SessionCompletedAt time.Time + RequestTime time.Time + NotifyChan chan struct{} + CancelCh <-chan struct{} // Request-scoped cancellation signal + Release func() // Set by the queue when a permit is acquired // admitMu serializes admission (by the dequeue loop) against abandonment (by // the waiting caller on timeout/cancel), closing the race where the loop has @@ -273,16 +277,19 @@ func NewRequestPriorityQueueWithConfig(metricsInstance *metrics.Metrics, cfg Fai func (pq *RequestPriorityQueue) Len() int { return len(pq.heap) } func (pq *RequestPriorityQueue) Less(i, j int) bool { - // Session-boost mode: boosted requests outrank others. Among boosted requests - // the most-recently-arrived one wins (LIFO): a freshly boosted follow-up always - // jumps to the head so it can reuse the still-warm prefix cache before earlier - // boosted requests. Non-boosted requests keep FIFO ordering. + // Session-boost mode: boosted requests outrank others. Among boosted requests, + // the session whose previous turn completed most recently wins, because its + // prefix cache is the most likely to still be warm on the backend; ties are + // broken FIFO by arrival time. Non-boosted requests keep FIFO ordering. if pq.sessionBoost { if pq.heap[i].SessionBoost != pq.heap[j].SessionBoost { return pq.heap[i].SessionBoost } if pq.heap[i].SessionBoost { - return pq.heap[i].RequestTime.After(pq.heap[j].RequestTime) + if !pq.heap[i].SessionCompletedAt.Equal(pq.heap[j].SessionCompletedAt) { + return pq.heap[i].SessionCompletedAt.After(pq.heap[j].SessionCompletedAt) + } + return pq.heap[i].RequestTime.Before(pq.heap[j].RequestTime) } return pq.heap[i].RequestTime.Before(pq.heap[j].RequestTime) } @@ -323,9 +330,13 @@ func (pq *RequestPriorityQueue) PushRequest(r *Request) error { defer pq.mu.Unlock() // In session-boost mode, promote requests whose session recently completed. - if pq.sessionBoost && pq.sessionTracker != nil && r.SessionID != "" && - pq.sessionTracker.HasRecentCompletion(r.SessionID) { - r.SessionBoost = true + // Capture the session's completion time so boosted requests can be ordered by + // prefix-cache warmth (most recently completed first). + if pq.sessionBoost && pq.sessionTracker != nil && r.SessionID != "" { + if completedAt, ok := pq.sessionTracker.CompletionTime(r.SessionID); ok { + r.SessionBoost = true + r.SessionCompletedAt = completedAt + } } heap.Push(pq, r) diff --git a/pkg/kthena-router/datastore/session_boost_queue.go b/pkg/kthena-router/datastore/session_boost_queue.go index 54e31aa9c..33cdbfc39 100644 --- a/pkg/kthena-router/datastore/session_boost_queue.go +++ b/pkg/kthena-router/datastore/session_boost_queue.go @@ -56,10 +56,13 @@ type PodCounter func() int // keeping a few extra entries is harmless because boosting only matters when the // queue is contended. type SessionTracker struct { - // cache maps a session ID to a presence marker, ordered by recency. The - // underlying hashicorp/golang-lru cache is safe for concurrent use and evicts - // the least-recently-used session once capacity is exceeded. - cache *lru.Cache[string, struct{}] + // cache maps a session ID to the time its most recent request completed, + // ordered by recency. The underlying hashicorp/golang-lru cache is safe for + // concurrent use and evicts the least-recently-used session once capacity is + // exceeded. The stored timestamp is used to order boosted requests so that the + // session whose previous turn completed most recently—and is therefore most + // likely to still have a warm prefix cache—is served first. + cache *lru.Cache[string, time.Time] } // NewSessionTracker creates a new session tracker that remembers up to capacity @@ -69,7 +72,7 @@ func NewSessionTracker(capacity int) *SessionTracker { if capacity <= 0 { capacity = defaultSessionBoostMaxSessions } - cache, err := lru.NewWithEvict(capacity, func(sessionID string, _ struct{}) { + cache, err := lru.NewWithEvict(capacity, func(sessionID string, _ time.Time) { klog.V(4).Infof("[SessionTracker] evicted LRU session %q", sessionID) }) if err != nil { @@ -80,13 +83,13 @@ func NewSessionTracker(capacity int) *SessionTracker { } // MarkRequestCompleted records that a request from the given session has completed, -// promoting it to the most-recently-used position. When the cache exceeds its -// capacity, the least-recently-used session is evicted. +// storing the completion time and promoting it to the most-recently-used position. +// When the cache exceeds its capacity, the least-recently-used session is evicted. func (st *SessionTracker) MarkRequestCompleted(sessionID string) { if sessionID == "" { return } - st.cache.Add(sessionID, struct{}{}) + st.cache.Add(sessionID, time.Now()) } // HasRecentCompletion reports whether the given session ID is currently tracked @@ -99,6 +102,16 @@ func (st *SessionTracker) HasRecentCompletion(sessionID string) bool { return st.cache.Contains(sessionID) } +// CompletionTime returns the time the given session's most recent request completed +// and whether the session is currently tracked. It is a pure read (via Peek) and +// does not change recency ordering. +func (st *SessionTracker) CompletionTime(sessionID string) (time.Time, bool) { + if sessionID == "" { + return time.Time{}, false + } + return st.cache.Peek(sessionID) +} + // ActiveSessions returns the number of sessions currently tracked. func (st *SessionTracker) ActiveSessions() int { return st.cache.Len() @@ -236,12 +249,12 @@ func (pq *RequestPriorityQueue) drainPendingRelease() bool { // waitGraceAndDequeue holds a just-freed slot for up to SessionBoostGracePeriod // before dispatching, giving a same-session follow-up time to arrive. It does not // need to watch for the follow-up itself: boosted requests outrank others in the -// heap and, among boosted requests, the most-recently-arrived one wins, so once -// the timer fires tryBackpressureDequeue admits the newest boosted follow-up first -// if one showed up. The wait always runs for the full window even when the head is -// already boosted, because a still-newer same-session follow-up may yet arrive and -// should be the one to ride the warm prefix cache. The wait stays responsive to -// shutdown via ctx/stopCh. +// heap and, among boosted requests, the one whose session completed most recently +// wins, so once the timer fires tryBackpressureDequeue admits the warmest boosted +// follow-up first if one showed up. The wait always runs for the full window even +// when the head is already boosted, because a follow-up from an even-more-recently +// completed session may yet arrive and should be the one to ride the warm prefix +// cache. The wait stays responsive to shutdown via ctx/stopCh. func (pq *RequestPriorityQueue) waitGraceAndDequeue(ctx context.Context) { klog.V(4).Infof("[SessionBoost] grace: holding freed slot for %v", pq.config.SessionBoostGracePeriod) timer := time.NewTimer(pq.config.SessionBoostGracePeriod) diff --git a/pkg/kthena-router/datastore/session_boost_queue_test.go b/pkg/kthena-router/datastore/session_boost_queue_test.go index e99235298..d496094c4 100644 --- a/pkg/kthena-router/datastore/session_boost_queue_test.go +++ b/pkg/kthena-router/datastore/session_boost_queue_test.go @@ -217,10 +217,10 @@ func TestSessionBoostQueue_MultipleSessions(t *testing.T) { t.Errorf("Last two should not be boosted: third=%v fourth=%v", third.SessionBoost, fourth.SessionBoost) } - // Among boosted: newest arrival wins (LIFO), so boost-B (later arrival) is - // dequeued before boost-A. + // Among boosted: the session that completed most recently wins. conv-B was + // marked completed after conv-A, so boost-B (u3) is dequeued before boost-A (u2). if first.UserID != "u3" { - t.Errorf("Expected boost-B first (latest arrival), got %s", first.UserID) + t.Errorf("Expected boost-B first (most recently completed session), got %s", first.UserID) } if second.UserID != "u2" { t.Errorf("Expected boost-A second, got %s", second.UserID) @@ -235,6 +235,49 @@ func TestSessionBoostQueue_MultipleSessions(t *testing.T) { } } +// TestSessionBoostQueue_OrderedByCompletionTime verifies that boosted requests are +// dequeued by their session's completion time (most recently completed first), +// independent of the order in which the requests arrived in the queue. +func TestSessionBoostQueue_OrderedByCompletionTime(t *testing.T) { + cfg := sessionBoostConfig() + q := newSessionBoostQueue(cfg, nil) + defer q.Close() + + // Completion order: conv-A, then conv-B, then conv-C (conv-C most recent). + q.MarkSessionRequestCompleted("conv-A") + q.MarkSessionRequestCompleted("conv-B") + q.MarkSessionRequestCompleted("conv-C") + + now := time.Now() + // Arrival order deliberately differs from completion order: B arrives first, + // then C, then A. This distinguishes completion-time ordering from both FIFO + // (which would yield B, C, A) and LIFO (which would yield A, C, B). + requests := []*Request{ + {UserID: "u-B", ModelName: "m", SessionID: "conv-B", RequestTime: now}, + {UserID: "u-C", ModelName: "m", SessionID: "conv-C", RequestTime: now.Add(time.Millisecond)}, + {UserID: "u-A", ModelName: "m", SessionID: "conv-A", RequestTime: now.Add(2 * time.Millisecond)}, + } + for _, r := range requests { + if err := q.PushRequest(r); err != nil { + t.Fatalf("PushRequest failed: %v", err) + } + } + + // Expect dequeue order by most-recent completion: C, then B, then A. + for i, expected := range []string{"u-C", "u-B", "u-A"} { + got, err := q.popWhenAvailable(context.Background()) + if err != nil { + t.Fatalf("Pop %d failed: %v", i, err) + } + if !got.SessionBoost { + t.Errorf("Position %d: expected boosted request, got non-boosted %s", i, got.UserID) + } + if got.UserID != expected { + t.Errorf("Position %d: expected %s (by completion time), got %s", i, expected, got.UserID) + } + } +} + func TestSessionBoostQueue_BackpressureDrainsCancelledWhenBusy(t *testing.T) { // Backends are always busy, so tryBackpressureDequeue never pops. Cancelled // requests must still be drained from the heap so the queue size does not stay From 57cd2d3daf78b1ad72fd07de55d998a8372ed98e Mon Sep 17 00:00:00 2001 From: YaoZengzeng Date: Fri, 10 Jul 2026 09:34:44 +0000 Subject: [PATCH 3/5] kthena-router: rename SessionCompletedAt to LastTurnCompletedAt Address review feedback: SessionCompletedAt could be read as the whole session ending. The field is the completion time of the session's previous turn, so rename it to LastTurnCompletedAt for clarity. Signed-off-by: YaoZengzeng --- pkg/kthena-router/datastore/fairness_queue.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/pkg/kthena-router/datastore/fairness_queue.go b/pkg/kthena-router/datastore/fairness_queue.go index a6a4e28a4..ac66e10ee 100644 --- a/pkg/kthena-router/datastore/fairness_queue.go +++ b/pkg/kthena-router/datastore/fairness_queue.go @@ -148,11 +148,11 @@ type Request struct { SessionID string // Session identifier for multi-turn conversations Priority float64 // Priority (lower value means higher priority) SessionBoost bool // Whether this request has session priority boost (recently completed session) - // SessionCompletedAt is the time the session's previous turn completed, captured + // LastTurnCompletedAt is the time the session's previous turn completed, captured // when the request is boosted. Boosted requests are ordered by this timestamp // (most recent first) so the session with the warmest prefix cache runs first. - SessionCompletedAt time.Time - RequestTime time.Time + LastTurnCompletedAt time.Time + RequestTime time.Time NotifyChan chan struct{} CancelCh <-chan struct{} // Request-scoped cancellation signal Release func() // Set by the queue when a permit is acquired @@ -286,8 +286,8 @@ func (pq *RequestPriorityQueue) Less(i, j int) bool { return pq.heap[i].SessionBoost } if pq.heap[i].SessionBoost { - if !pq.heap[i].SessionCompletedAt.Equal(pq.heap[j].SessionCompletedAt) { - return pq.heap[i].SessionCompletedAt.After(pq.heap[j].SessionCompletedAt) + if !pq.heap[i].LastTurnCompletedAt.Equal(pq.heap[j].LastTurnCompletedAt) { + return pq.heap[i].LastTurnCompletedAt.After(pq.heap[j].LastTurnCompletedAt) } return pq.heap[i].RequestTime.Before(pq.heap[j].RequestTime) } @@ -335,7 +335,7 @@ func (pq *RequestPriorityQueue) PushRequest(r *Request) error { if pq.sessionBoost && pq.sessionTracker != nil && r.SessionID != "" { if completedAt, ok := pq.sessionTracker.CompletionTime(r.SessionID); ok { r.SessionBoost = true - r.SessionCompletedAt = completedAt + r.LastTurnCompletedAt = completedAt } } From c36017c9efc0a1902b2fdaaaf6dc2caa958dc457 Mon Sep 17 00:00:00 2001 From: YaoZengzeng Date: Fri, 10 Jul 2026 09:41:28 +0000 Subject: [PATCH 4/5] kthena-router: make session-boost completion-time tests deterministic The completion-time ordering tests relied on consecutive MarkRequestCompleted calls producing strictly increasing time.Now() values. On a coarse clock or under fast execution two completions could share an instant, making the boosted ordering fall back to RequestTime and the expected order nondeterministic. Inject an overridable clock into SessionTracker (defaults to time.Now) and use it in the tests to assign strictly distinct completion timestamps, removing the flake without arbitrary sleeps. Signed-off-by: YaoZengzeng --- .../datastore/session_boost_queue.go | 9 ++++-- .../datastore/session_boost_queue_test.go | 29 +++++++++++++++---- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/pkg/kthena-router/datastore/session_boost_queue.go b/pkg/kthena-router/datastore/session_boost_queue.go index 33cdbfc39..f9e917b04 100644 --- a/pkg/kthena-router/datastore/session_boost_queue.go +++ b/pkg/kthena-router/datastore/session_boost_queue.go @@ -63,6 +63,11 @@ type SessionTracker struct { // session whose previous turn completed most recently—and is therefore most // likely to still have a warm prefix cache—is served first. cache *lru.Cache[string, time.Time] + + // now returns the current time when a completion is recorded. It defaults to + // time.Now and is only overridden in tests to make completion timestamps + // deterministic. + now func() time.Time } // NewSessionTracker creates a new session tracker that remembers up to capacity @@ -79,7 +84,7 @@ func NewSessionTracker(capacity int) *SessionTracker { // capacity is guaranteed positive above, so this is unreachable in practice. klog.Errorf("[SessionTracker] failed to create LRU cache (capacity=%d): %v", capacity, err) } - return &SessionTracker{cache: cache} + return &SessionTracker{cache: cache, now: time.Now} } // MarkRequestCompleted records that a request from the given session has completed, @@ -89,7 +94,7 @@ func (st *SessionTracker) MarkRequestCompleted(sessionID string) { if sessionID == "" { return } - st.cache.Add(sessionID, time.Now()) + st.cache.Add(sessionID, st.now()) } // HasRecentCompletion reports whether the given session ID is currently tracked diff --git a/pkg/kthena-router/datastore/session_boost_queue_test.go b/pkg/kthena-router/datastore/session_boost_queue_test.go index d496094c4..ab8b6d33b 100644 --- a/pkg/kthena-router/datastore/session_boost_queue_test.go +++ b/pkg/kthena-router/datastore/session_boost_queue_test.go @@ -187,8 +187,16 @@ func TestSessionBoostQueue_MultipleSessions(t *testing.T) { q := newSessionBoostQueue(cfg, nil) defer q.Close() - q.MarkSessionRequestCompleted("conv-A") - q.MarkSessionRequestCompleted("conv-B") + // Use a controllable clock so conv-B's completion is strictly after conv-A's, + // keeping the completion-time ordering below deterministic. + st := q.GetSessionTracker() + base := time.Now() + markCompletedAt := func(sessionID string, completedAt time.Time) { + st.now = func() time.Time { return completedAt } + q.MarkSessionRequestCompleted(sessionID) + } + markCompletedAt("conv-A", base) + markCompletedAt("conv-B", base.Add(time.Second)) now := time.Now() requests := []*Request{ @@ -243,10 +251,21 @@ func TestSessionBoostQueue_OrderedByCompletionTime(t *testing.T) { q := newSessionBoostQueue(cfg, nil) defer q.Close() + // Drive the tracker with a controllable clock so each completion gets a + // strictly increasing, distinct timestamp regardless of the wall clock's + // resolution or how fast the test executes. This keeps the completion-time + // ordering deterministic instead of falling back to the RequestTime tiebreak. + st := q.GetSessionTracker() + base := time.Now() + markCompletedAt := func(sessionID string, completedAt time.Time) { + st.now = func() time.Time { return completedAt } + q.MarkSessionRequestCompleted(sessionID) + } + // Completion order: conv-A, then conv-B, then conv-C (conv-C most recent). - q.MarkSessionRequestCompleted("conv-A") - q.MarkSessionRequestCompleted("conv-B") - q.MarkSessionRequestCompleted("conv-C") + markCompletedAt("conv-A", base) + markCompletedAt("conv-B", base.Add(time.Second)) + markCompletedAt("conv-C", base.Add(2*time.Second)) now := time.Now() // Arrival order deliberately differs from completion order: B arrives first, From b1c76c375e04fc22b06cd8d6f54db7253dcffc14 Mon Sep 17 00:00:00 2001 From: YaoZengzeng Date: Fri, 10 Jul 2026 09:47:51 +0000 Subject: [PATCH 5/5] kthena-router: fix gofmt alignment in Request struct Signed-off-by: YaoZengzeng --- pkg/kthena-router/datastore/fairness_queue.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/kthena-router/datastore/fairness_queue.go b/pkg/kthena-router/datastore/fairness_queue.go index ac66e10ee..073dc0fdb 100644 --- a/pkg/kthena-router/datastore/fairness_queue.go +++ b/pkg/kthena-router/datastore/fairness_queue.go @@ -153,9 +153,9 @@ type Request struct { // (most recent first) so the session with the warmest prefix cache runs first. LastTurnCompletedAt time.Time RequestTime time.Time - NotifyChan chan struct{} - CancelCh <-chan struct{} // Request-scoped cancellation signal - Release func() // Set by the queue when a permit is acquired + NotifyChan chan struct{} + CancelCh <-chan struct{} // Request-scoped cancellation signal + Release func() // Set by the queue when a permit is acquired // admitMu serializes admission (by the dequeue loop) against abandonment (by // the waiting caller on timeout/cancel), closing the race where the loop has