Copilot commented on code in PR #3418:
URL: https://github.com/apache/brpc/pull/3418#discussion_r3671468983
##########
src/bthread/task_control.cpp:
##########
@@ -218,6 +223,7 @@ TaskControl::TaskControl()
FLAGS_task_group_ntags * FLAGS_event_dispatcher_num)
, _pl_num_of_each_tag(FLAGS_bthread_parking_lot_of_each_tag)
, _tagged_pl(FLAGS_task_group_ntags)
+ , _tagged_waiter_num(FLAGS_task_group_ntags)
, _tag_cpus(FLAGS_task_group_ntags)
, _tag_next_worker_id(FLAGS_task_group_ntags)
{}
Review Comment:
`_tagged_waiter_num` elements are default-constructed via
`butil::atomic<size_t>::atomic() {}` which does not value-initialize the
underlying atomic. This can leave the per-tag waiter counts indeterminate,
breaking `has_waiting_workers()` (e.g., always signaling or underflowing/never
returning to zero). Explicitly initialize all counters to 0.
##########
src/bthread/task_control.h:
##########
@@ -73,9 +73,26 @@ friend bthread_t init_for_pthread_stack_trace();
int concurrency() const
{ return _concurrency.load(butil::memory_order_acquire); }
- int concurrency(bthread_tag_t tag) const
+ int concurrency(bthread_tag_t tag) const
{ return _tagged_ngroup[tag].load(butil::memory_order_acquire); }
+ // Check if there are workers parked in wait() for the given tag.
+ // Per-tag accounting avoids a tag-A ready_to_run() needlessly issuing a
+ // signal_task() just because workers of a *different* tag B are parked.
+ bool has_waiting_workers(bthread_tag_t tag) const {
+ return tag < (bthread_tag_t)_tagged_waiter_num.size()
+ && _tagged_waiter_num[tag].load(butil::memory_order_relaxed) >
0;
+ }
Review Comment:
`has_waiting_workers()` does not guard against `BTHREAD_TAG_INVALID` (=-1).
Since `bthread_tag_t` is `int`, a negative tag satisfies `tag < size` and will
index `_tagged_waiter_num[tag]` with a negative index (UB). Add `tag >= 0` to
the bounds check.
##########
src/bthread/task_control.cpp:
##########
@@ -302,6 +308,15 @@ int TaskControl::init(int concurrency) {
_signal_per_second.expose("bthread_signal_second");
_status.expose("bthread_group_status");
+ // Wire each ParkingLot back to this TaskControl so that ParkingLot::wait()
+ // can maintain the per-tag _tagged_waiter_num counter consulted by
+ // has_waiting_workers(tag). The outer index `i' is the owning tag.
+ for (size_t i = 0; i < _tagged_pl.size(); ++i) {
+ for (size_t j = 0; j < _pl_num_of_each_tag; ++j) {
Review Comment:
`set_task_control()` wiring is done *after* worker threads are created.
Workers can enter `ParkingLot::wait()` before `_tc/_tag` is set, so they won't
be counted in `_tagged_waiter_num`. After this PR, `ready_to_run*()` may then
skip `signal_task()` (because `has_waiting_workers()` reads 0) and leave
already-parked workers sleeping indefinitely. Wire parking lots before starting
worker threads (or ensure the selected ParkingLot is wired before any wait).
##########
src/bthread/task_control.h:
##########
@@ -73,9 +73,26 @@ friend bthread_t init_for_pthread_stack_trace();
int concurrency() const
{ return _concurrency.load(butil::memory_order_acquire); }
- int concurrency(bthread_tag_t tag) const
+ int concurrency(bthread_tag_t tag) const
{ return _tagged_ngroup[tag].load(butil::memory_order_acquire); }
+ // Check if there are workers parked in wait() for the given tag.
+ // Per-tag accounting avoids a tag-A ready_to_run() needlessly issuing a
+ // signal_task() just because workers of a *different* tag B are parked.
+ bool has_waiting_workers(bthread_tag_t tag) const {
+ return tag < (bthread_tag_t)_tagged_waiter_num.size()
+ && _tagged_waiter_num[tag].load(butil::memory_order_relaxed) >
0;
+ }
+
+ // Called by ParkingLot (via parking_lot_waiter_add/sub) to track waiters
+ // entering/leaving wait() for a given tag.
+ void waiter_add(bthread_tag_t tag) {
+ _tagged_waiter_num[tag].fetch_add(1, butil::memory_order_relaxed);
+ }
+ void waiter_sub(bthread_tag_t tag) {
+ _tagged_waiter_num[tag].fetch_sub(1, butil::memory_order_relaxed);
+ }
Review Comment:
`waiter_add()` / `waiter_sub()` index `_tagged_waiter_num[tag]` without
validating `tag`. Even if tags are expected to be valid, these functions are
reachable via `ParkingLot` and can be called with `BTHREAD_TAG_INVALID` if a
caller ever misconfigures a tag, causing UB. Add a non-negative + upper-bound
guard (and optionally DCHECK).
##########
src/bthread/parking_lot.h:
##########
@@ -30,6 +30,16 @@ namespace bthread {
DECLARE_bool(parking_lot_no_signal_when_no_waiter);
+// Forward declaration
+class TaskControl;
+
+// Out-of-line helpers that forward to TaskControl's per-tag waiter accounting.
+// Defined in task_control.cpp (where TaskControl is complete), so that
+// parking_lot.h can keep TaskControl as an incomplete type and avoid a
+// circular include with task_control.h.
+void parking_lot_waiter_add(TaskControl* tc, bthread_tag_t tag);
Review Comment:
`parking_lot.h` now uses `bthread_tag_t` but does not include the header
that defines it (`bthread/types.h`), relying on indirect includes. This makes
the header non-self-contained and can break if include order changes. Include
`bthread/types.h` directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
