Conversation
📝 WalkthroughWalkthroughFiberQueue now stores and runs ChangesFiberQueue Tasklet Migration
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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 |
PR Summary by QodoFiberQueue: inline task callables and reduce producer/consumer contention
AI Description
Diagram
High-Level Assessment
Files changed (2)
|
There was a problem hiding this comment.
🧹 Nitpick comments (1)
util/fibers/fiberqueue_threadpool.h (1)
73-77: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winMake the
blocked_submitters_accounting exception/early-return safe.If
push_ec_.wait(...)unwinds (e.g., fiber cancellation) theblocked_submitters_--on Line 77 is skipped, permanently leaking the counter. Pair the inc/dec with RAII so it is balanced on every exit path.♻️ Suggested RAII fix
- result = true; - blocked_submitters_++; - push_ec_.wait(key.epoch()); - blocked_submitters_--; + result = true; + blocked_submitters_++; + absl::Cleanup dec = [] { blocked_submitters_--; }; + push_ec_.wait(key.epoch());As per coding guidelines: "Use smart pointers (std::unique_ptr) and RAII (absl::Cleanup) for resource management".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@util/fibers/fiberqueue_threadpool.h` around lines 73 - 77, The blocked_submitters_ counter in the fiber queue wait path is not exception/early-return safe because the decrement after push_ec_.wait(key.epoch()) can be skipped if the wait unwinds. Update the submit-wait logic in fiberqueue_threadpool.h to use RAII (for example, absl::Cleanup) around the blocked_submitters_ increment so the decrement is guaranteed on every exit path, and keep the change localized to the push_ec_.wait block in the queue submit/await flow.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@util/fibers/fiberqueue_threadpool.h`:
- Around line 73-77: The blocked_submitters_ counter in the fiber queue wait
path is not exception/early-return safe because the decrement after
push_ec_.wait(key.epoch()) can be skipped if the wait unwinds. Update the
submit-wait logic in fiberqueue_threadpool.h to use RAII (for example,
absl::Cleanup) around the blocked_submitters_ increment so the decrement is
guaranteed on every exit path, and keep the change localized to the
push_ec_.wait block in the queue submit/await flow.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 53c59093-f322-4c85-b43a-cf610949199d
📒 Files selected for processing (2)
util/fibers/fiberqueue_threadpool.ccutil/fibers/fiberqueue_threadpool.h
Code Review by Qodo
Context used✅ Compliance rules (platform):
11 rules✅ Cross-repo context Explored:
repo: dragonflydb/dragonfly (sha: 5dcbdf55) 1. Unused blocked counter
|
1. Replace the std::function<void()> callback (CbFunc) with Tasklet, a fixed-capacity fu2::function_base (capacity_fixed<32,8>, non-copyable, non-throwing). Callbacks up to 32 bytes are stored inline with no heap allocation. Ported from #595. 2. FiberQueue: put pull_ec_ and push_ec_ on separate cache lines pull_ec_ is written by every producer (notify() after each commit); push_ec_ is written by the consumer (notifyAll() per drained batch) and by blocked producers. Adjacent in memory they share a cache line and ping-pong between producer- and consumer-side wakeup traffic. alignas(64) each onto its own line. 3. Add blocked_submitters_ metric Signed-off-by: Roman Gershman <romange@gmail.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@util/fibers/fiberqueue_threadpool.cc`:
- Around line 53-55: The worker loop in FiberQueueThreadPool currently calls
batch[i]() without any local exception handling, so any throwing Tasklet escapes
the worker thread and loses per-task context. Update the loop in the
worker/dispatch path to either make Tasklet explicitly noexcept if callbacks
must not throw, or keep a thin try/catch around batch[i]() that logs the failure
with the task/queue context before terminating or handling it.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b2766a83-988d-457a-8697-c7adbced1e3a
📒 Files selected for processing (2)
util/fibers/fiberqueue_threadpool.ccutil/fibers/fiberqueue_threadpool.h
🚧 Files skipped from review as they are similar to previous changes (1)
- util/fibers/fiberqueue_threadpool.h
| // enqueue_pos_/dequeue_pos_ internally. | ||
| alignas(64) EventCount pull_ec_; | ||
| alignas(64) EventCount push_ec_; | ||
| std::atomic_bool is_closed_{false}; |
There was a problem hiding this comment.
What about is_closed_, it is in the same cache line as push_ec_ so if we read it we get push_ec_
There was a problem hiding this comment.
I checked it. Since is_closed_ changes only once, it does not really matter. As long as variable does not change it does not affect cache-lines.
| push_ec_.notifyAll(); | ||
|
|
||
| for (unsigned i = 0; i < count; ++i) { | ||
| try { |
There was a problem hiding this comment.
I think it is wrong to remove the try/catch.
It will still throw even with Tasklet - the isThrowing just mark how to act on a specific case:
See function2.hpp:
/// \tparam IsThrowing Defines whether the function throws an exception on
/// empty function call, `std::abort` is called otherwise.
The signature doesn't have noexcept at the end, so it can still throw. If you remove this try/catch the LOG(FATAL) won't log the error and any exception will be treated as an uncaught exception.
You can return this block back or add noexcept at the end of fu2Fun:
fu2::function_base<true, false, fu2::capacity_fixed<32, 8>,
false, false, void() noexcept>;
|
|
||
| void Run(); | ||
|
|
||
| unsigned blocked_submitters() const { |
There was a problem hiding this comment.
- Nit: This is dead code, why do you add it if no one uses it? I assume it is for future use?
- If you decide to keep this function, I think you should document it (add a comment above) because the name + non-static signature give the impression a per-queue counter, which it is not (it can be confusing) .
- Also, if it stays consider making it static, since it reads thread-global state and has nothing to do with a specific instance - a non-static accessor makes caller think they can read it as per-queue.
| } | ||
|
|
||
| result = true; | ||
| blocked_submitters_++; |
There was a problem hiding this comment.
Shouldn't a "blocked submitter" be accounted also in line 176?
I don't see blocked_submitters_ being updated there.
| using Fu2Fun::operator=; | ||
| }; | ||
|
|
||
| static_assert(sizeof(Tasklet) == 48); |
There was a problem hiding this comment.
nit: consider adding a message here in case compilation fail.
e.g:
static_assert(sizeof(Tasklet) == 48,
"Tasklet size changed: 32B inline buffer + fu2 dispatch overhead. ");
| friend class FiberQueueThreadPool; | ||
|
|
||
| // Fixed-capacity, non-copyable, non-throwing callable. Callbacks up to 32 bytes are stored | ||
| // inline (no heap allocation), unlike std::function. Keeps queue cells trivially relocatable. |
There was a problem hiding this comment.
nit:
I think that "Keeps queue cells trivially relocatable." is not correct.
From my understanding "trivially relocatable" is true if we can move the object to a new address using memcpy - but this is not what i see in function2.hpp. e.g:
void move(vtable& to_table, data_accessor* from, std::size_t from_capacity,
data_accessor* to,
std::size_t to_capacity) noexcept(HasStrongExceptGuarantee) {
cmd_(&to_table, opcode::op_move, from, from_capacity, to, to_capacity);
....
and after that for inline storage that op_move path move-constructs the callable and destroys the source:
// The object is allocated inplace
else {
construct(std::true_type{}, std::move(*box), to_table, to,
to_capacity);
box->~T();
}
I would just deleted "Keeps queue cells trivially relocatable."
Summary
Replaces
FiberQueue's callback storage with a fixed-capacityfu2::function(Tasklet), separatesthe producer/consumer
EventCounts onto distinct cache lines, and adds ablocked_submitters_metric.Changes
std::function<void()>(CbFunc) withTasklet, a fixed-capacity (32 bytes inline, 8-bytealignment), non-copyable, non-throwing
fu2::function_base. Callbacks up to 32 bytes avoid heapallocation. Ported from Epoll: use busy poll cycle limit #595.
alignas(64)pull_ec_andpush_ec_separately — previously adjacent, they shared a cache lineand ping-ponged between producer-side (
notify()per commit) and consumer-side (notifyAll()perdrained batch) wakeup traffic.
blocked_submitters_thread-local counter, incremented/decremented aroundpush_ec_.wait()when a submitter blocks on a full queue.Summary by CodeRabbit
blocked_submitters()metric to monitor per-thread queue contention levels.