queue: add try_evict_lowest API for priority-based queue eviction - #89
Merged
Conversation
Add eviction support to the priority queue so that when a queue is full, a higher-priority incoming task can evict the lowest-priority queued task. The API is exposed through QueueCore, TaskInjector (priority and outer), and Remote. Only priority queues support eviction; other queue types return None. Signed-off-by: Rishabh Mittal <rishabh_mittal@airbnb.com> Co-Authored-By: Rishabh Mittal <rishabh_mittal@airbnb.com> Signed-off-by: rishabh mittal <mittalrishabh@gmail.com>
|
No actionable comments were generated in the recent review. 🎉 📝 WalkthroughWalkthroughAdds a priority-based eviction API and implementation: tightened generic bounds to require Changes
Sequence DiagramsequenceDiagram
participant Client as Incoming Task
participant Remote as Remote<T>
participant Injector as TaskInjector<T>
participant QueueCore as QueueCore<T>
participant PrioQ as Priority Queue
Client->>Remote: try_evict_lowest(incoming_priority)
Remote->>Injector: try_evict_lowest(incoming_priority)
Injector->>QueueCore: try_evict(incoming_priority)
QueueCore->>PrioQ: inspect back entry (lowest priority)
alt queue empty
PrioQ-->>QueueCore: empty
QueueCore-->>Injector: Ok(None)
else has entries
PrioQ-->>QueueCore: back_priority
alt incoming_priority > back_priority
PrioQ->>PrioQ: remove back entry (race-resilient)
PrioQ-->>QueueCore: Task
QueueCore-->>Injector: Ok(Some(Task))
else incoming_priority <= back_priority
PrioQ-->>QueueCore: not strictly higher
QueueCore-->>Injector: Err(())
end
end
Injector-->>Remote: Option<T>
Remote-->>Client: Option<T>
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/queue.rs (1)
153-157:⚠️ Potential issue | 🟠 MajorPre-existing CI pipeline failure blocks merge
The pipeline reports:
[error] 153-157: This 'impl Default for QueueType' can be derived. Replace with #[derive(Default)] and mark the default variant.This is pre-existing code but is failing CI and blocking the PR. Fix by deriving
Defaultand annotating the unit variant:🛠️ Proposed fix
+#[derive(Default)] pub enum QueueType { + #[default] SingleLevel, Multilevel(multilevel::Builder), Priority(priority::Builder), } - -impl Default for QueueType { - fn default() -> QueueType { - QueueType::SingleLevel - } -}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/queue.rs` around lines 153 - 157, Replace the manual impl Default for QueueType with a derived Default: remove the impl Default block and add #[derive(Default)] to the QueueType enum declaration, and mark the unit variant SingleLevel with #[default] so the compiler can derive the default; this replaces the impl Default for QueueType { fn default() -> QueueType { QueueType::SingleLevel } } with a derived default on the enum.src/pool/spawn.rs (1)
219-222:⚠️ Potential issue | 🟠 MajorPre-existing CI pipeline failures block merge
The pipeline reports:
[error] 219-219: trait AssertSync is never used.
[error] 221-221: trait AssertSend is never used.These are intentional compile-time assertions (the accompanying comment confirms this). Suppress the dead-code lint with
#[allow(dead_code)]:🛠️ Proposed fix
+#[allow(dead_code)] trait AssertSync: Sync {} impl<T: Send> AssertSync for Remote<T> {} +#[allow(dead_code)] trait AssertSend: Send {} impl<T: Send> AssertSend for Remote<T> {}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/pool/spawn.rs` around lines 219 - 222, The unused-trait lint is firing for the intentional compile-time assertion traits AssertSync and AssertSend; add #[allow(dead_code)] above the trait declarations for AssertSync and AssertSend (the impls for Remote<T>) so the dead-code lint is suppressed while preserving the compile-time check of Remote<T> implementing Send/Sync; locate the trait declarations named AssertSync and AssertSend and annotate them with #[allow(dead_code)].
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/queue/priority.rs`:
- Around line 540-594: The stress test test_evict_concurrent_stress only uses
incoming_priority = 0 for all evictors and therefore never exercises the TOCTOU
race where a back entry with priority ≤ incoming_priority can be exposed by a
concurrent evictor; add a complementary test (e.g.,
test_evict_concurrent_mixed_priorities) that pushes tasks with a known priority
(e.g., 30) via MockTask and spawns multiple evictor threads calling try_evict
with different incoming_priority values (e.g., 25 and 28) so evictors can race
on entries around the threshold; in each evictor validate that any returned task
has priority > the caller's incoming_priority (use MockTask::priority or key().0
to check) and also assert total pushed == evicted + remaining to keep
conservation checks (use injector, try_evict, evicted_count, and
injector.queue.pop() like the original test).
- Around line 152-171: The try_evict_for_priority check-then-pop has a TOCTOU
bug: after reading self.pq.back() you must atomically remove that specific entry
instead of calling pop_back() which may remove a different (higher-priority)
entry; change the logic in try_evict_for_priority to call Entry::remove() on the
Entry returned by self.pq.back() (and if it returns true then call
entry.value().take() to extract the task and return Ok(Some(task)), if remove()
returns false return Ok(None)), and stop using pop_back() for the validated
entry so you never accidentally evict a different key than the one you
inspected.
---
Outside diff comments:
In `@src/pool/spawn.rs`:
- Around line 219-222: The unused-trait lint is firing for the intentional
compile-time assertion traits AssertSync and AssertSend; add #[allow(dead_code)]
above the trait declarations for AssertSync and AssertSend (the impls for
Remote<T>) so the dead-code lint is suppressed while preserving the compile-time
check of Remote<T> implementing Send/Sync; locate the trait declarations named
AssertSync and AssertSend and annotate them with #[allow(dead_code)].
In `@src/queue.rs`:
- Around line 153-157: Replace the manual impl Default for QueueType with a
derived Default: remove the impl Default block and add #[derive(Default)] to the
QueueType enum declaration, and mark the unit variant SingleLevel with
#[default] so the compiler can derive the default; this replaces the impl
Default for QueueType { fn default() -> QueueType { QueueType::SingleLevel } }
with a derived default on the enum.
Signed-off-by: rishabh mittal <mittalrishabh@gmail.com>
Signed-off-by: rishabh mittal <mittalrishabh@gmail.com>
There was a problem hiding this comment.
🧹 Nitpick comments (3)
src/queue.rs (1)
49-49: The'staticbound is redundant sinceTaskCell: 'static.
TaskCellalready has a'staticsupertrait bound (line 22). The added+ 'staticon the impl is harmless but unnecessary — the compiler infers it fromT: TaskCell.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/queue.rs` at line 49, The impl for TaskInjector redundantly repeats the 'static bound; update the impl header so the generic constraint relies on the existing TaskCell:'static supertrait instead of duplicating it. Specifically, edit the impl declaration for TaskInjector<T> (currently declared as impl<T: TaskCell + Send + 'static> TaskInjector<T>) to remove the extra + 'static so it becomes impl<T: TaskCell + Send> TaskInjector<T>, leaving the TaskCell trait definition unchanged.src/queue/priority.rs (2)
129-145: Consider defensive handling inpop()to avoid panic if eviction races with consumption.Line 144 uses
.take().unwrap()which will panic iftake()returnsNone. While crossbeam-skiplist should preventpop_front()from returning an already-removed entry, a defensive.take()?(or filtering with.and_then()) would makepop()resilient to any unexpected interaction with the new eviction path.🛡️ Defensive take
pub fn pop(&self) -> Option<Pop<T>> { fn into_pop<T>(mut t: T) -> Pop<T> where T: TaskCell, { let schedule_time = t.mut_extras().schedule_time.unwrap(); Pop { task_cell: t, schedule_time, from_local: false, } } self.pq .pop_front() - .map(|e| into_pop(e.value().take().unwrap())) + .and_then(|e| e.value().take().map(into_pop)) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/queue/priority.rs` around lines 129 - 145, The pop() implementation currently calls .take().unwrap() inside map which can panic if an entry was evicted concurrently; update pop() (and the helper into_pop<T>) to handle a None from take() defensively by using .and_then() / .map(|e| e.value().take()).and_then(...) or using .take()? so that if take() returns None the whole operation returns None rather than panicking; ensure the returned Option<Pop<T>> remains None on a failed take and keep references to Pop, TaskCell, into_pop, and self.pq.pop_front() to locate the change.
43-43: Same note assrc/queue.rs: the'staticbound is already implied byTaskCell: 'static(line 22 ofsrc/queue.rs).Redundant but not harmful. Consistent with the outer
TaskInjectorimpl, so fine to keep for symmetry.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/queue/priority.rs` at line 43, The explicit 'static bound on the generic T (the "T: TaskCell + Send + 'static" in the TaskInjector impl in priority.rs) is redundant because TaskCell already implies 'static; remove the trailing "'static" from that generic bound so it reads "T: TaskCell + Send" to match the outer TaskInjector impl and avoid redundant constraints.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@src/queue.rs`:
- Line 49: The impl for TaskInjector redundantly repeats the 'static bound;
update the impl header so the generic constraint relies on the existing
TaskCell:'static supertrait instead of duplicating it. Specifically, edit the
impl declaration for TaskInjector<T> (currently declared as impl<T: TaskCell +
Send + 'static> TaskInjector<T>) to remove the extra + 'static so it becomes
impl<T: TaskCell + Send> TaskInjector<T>, leaving the TaskCell trait definition
unchanged.
In `@src/queue/priority.rs`:
- Around line 129-145: The pop() implementation currently calls .take().unwrap()
inside map which can panic if an entry was evicted concurrently; update pop()
(and the helper into_pop<T>) to handle a None from take() defensively by using
.and_then() / .map(|e| e.value().take()).and_then(...) or using .take()? so that
if take() returns None the whole operation returns None rather than panicking;
ensure the returned Option<Pop<T>> remains None on a failed take and keep
references to Pop, TaskCell, into_pop, and self.pq.pop_front() to locate the
change.
- Line 43: The explicit 'static bound on the generic T (the "T: TaskCell + Send
+ 'static" in the TaskInjector impl in priority.rs) is redundant because
TaskCell already implies 'static; remove the trailing "'static" from that
generic bound so it reads "T: TaskCell + Send" to match the outer TaskInjector
impl and avoid redundant constraints.
Member
Author
|
@Connor1996 can you review it. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
issue # tikv/tikv#19386
Add eviction support to the priority queue so that when a queue is full, a higher-priority incoming task can evict the lowest-priority queued task.
The API is exposed through QueueCore, TaskInjector (priority and outer), and Remote. Only priority queues support eviction; other queue types return None.
Summary by CodeRabbit
New Features
Tests