From bed4d0cc80d86626649d54d3ffe27e25d701ca5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=AC=E9=A3=8E?= <807188583@qq.com> Date: Wed, 29 Jul 2026 11:25:23 +0800 Subject: [PATCH 1/3] bthread: skip signal_task when no worker is waiting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem ready_to_run* unconditionally calls signal_task() on every task-ready event, even when all workers are busy and nobody is parked in wait(). This results in unnecessary futex_wake syscalls that wake no thread, wasting CPU under load. ## Solution Add per-tag waiter counting via _tagged_waiter_num: ParkingLot::wait() tracks waiters entering/leaving (via parking_lot_waiter_add/sub out-of-line forwarders) BEFORE the state check. ready_to_run* checks has_waiting_workers(tag) before calling signal_task — if no worker of the given tag is parked, the wakeup is skipped (BAIDU_UNLIKELY marks the true branch as cold since under load all workers are busy). ## Race-free design The waiter_add() is called BEFORE the get_state() check in wait(). This ensures that any concurrent ready_to_run() that calls has_waiting_workers() will see the waiter even if it races between the state check and futex_wait(). If the state already changed (signal was sent), wait() undoes the tracking and returns immediately without sleeping. This eliminates the lost-wakeup window entirely. ## Why it's safe - Tasks are already enqueued (push_rq) before the signal check, so they will be picked up by the next steal regardless of signaling. - _tagged_waiter_num uses relaxed atomics (it's a hint, not a synchronization primitive); the actual wake ordering is handled by parking_lot's existing futex mechanism. ## Multi-tag benefit Per-tag accounting avoids tag-A's ready_to_run() issuing a signal_task() just because tag-B has parked workers. Tested: bthread_unittest 20/20 on Linux x86 (Ubuntu 24.04/GCC 13) and macOS arm64 (M4). --- src/bthread/parking_lot.h | 47 +++++++++++++++++++++++++++++++++--- src/bthread/task_control.cpp | 15 ++++++++++++ src/bthread/task_control.h | 25 ++++++++++++++++++- src/bthread/task_group.cpp | 16 +++++++++--- 4 files changed, 94 insertions(+), 9 deletions(-) diff --git a/src/bthread/parking_lot.h b/src/bthread/parking_lot.h index bbc9a7c3fd..2e841d98e5 100644 --- a/src/bthread/parking_lot.h +++ b/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); +void parking_lot_waiter_sub(TaskControl* tc, bthread_tag_t tag); + // Park idle workers. class BAIDU_CACHELINE_ALIGNMENT ParkingLot { public: @@ -45,7 +55,15 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot { ParkingLot() : _pending_signal(0), _waiter_num(0) - , _no_signal_when_no_waiter(FLAGS_parking_lot_no_signal_when_no_waiter) {} + , _no_signal_when_no_waiter(FLAGS_parking_lot_no_signal_when_no_waiter) + , _tc(nullptr), _tag(0) {} + + // Set TaskControl pointer + owning tag for waiter tracking (called once + // during TaskControl::init after the ParkingLot array is constructed). + void set_task_control(TaskControl* tc, bthread_tag_t tag) { + _tc = tc; + _tag = tag; + } // Wake up at most `num_task' workers. // Returns #workers woken up. @@ -65,17 +83,34 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot { // Wait for tasks. // If the `expected_state' does not match, wait() may finish directly. void wait(const State& expected_state) { - if (get_state().val != expected_state.val) { - // Fast path, no need to futex_wait. - return; + // Track this waiter BEFORE checking state, so that a concurrent + // ready_to_run() that calls has_waiting_workers() will see us + // even if it races between our state check and futex_wait(). + // This eliminates the lost-wakeup window. + if (_tc) { + parking_lot_waiter_add(_tc, _tag); } if (_no_signal_when_no_waiter) { _waiter_num.fetch_add(1, butil::memory_order_relaxed); } + if (get_state().val != expected_state.val) { + // State changed since caller last checked — signal already sent. + // Undo the tracking and return without sleeping. + if (_no_signal_when_no_waiter) { + _waiter_num.fetch_sub(1, butil::memory_order_relaxed); + } + if (_tc) { + parking_lot_waiter_sub(_tc, _tag); + } + return; + } futex_wait_private(&_pending_signal, expected_state.val, NULL); if (_no_signal_when_no_waiter) { _waiter_num.fetch_sub(1, butil::memory_order_relaxed); } + if (_tc) { + parking_lot_waiter_sub(_tc, _tag); + } } // Wakeup suspended wait() and make them unwaitable ever. @@ -92,6 +127,10 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot { // In busy worker scenarios, signal overhead // can be reduced. bool _no_signal_when_no_waiter; + // TaskControl pointer for waiter tracking (may be NULL before init). + TaskControl* _tc; + // Owning tag, used to update the per-tag waiter counter in TaskControl. + bthread_tag_t _tag; }; } // namespace bthread diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index 3ad3aa508d..b33c50eae8 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -65,6 +65,11 @@ DEFINE_bool(parking_lot_no_signal_when_no_waiter, false, "In busy worker scenarios, signal overhead can be reduced."); DEFINE_bool(enable_bthread_priority_queue, false, "Whether to enable priority queue"); +// Out-of-line forwarders for ParkingLot per-tag waiter accounting, so that +// parking_lot.h can stay free of the TaskControl definition. +void parking_lot_waiter_add(TaskControl* tc, bthread_tag_t tag) { tc->waiter_add(tag); } +void parking_lot_waiter_sub(TaskControl* tc, bthread_tag_t tag) { tc->waiter_sub(tag); } + DECLARE_int32(bthread_concurrency); DECLARE_int32(bthread_min_concurrency); DECLARE_int32(bthread_parking_lot_of_each_tag); @@ -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) {} @@ -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) { + _tagged_pl[i][j].set_task_control(this, (bthread_tag_t)i); + } + } + // Wait for at least one group is added so that choose_one_group() // never returns NULL. // TODO: Handle the case that worker quits before add_group diff --git a/src/bthread/task_control.h b/src/bthread/task_control.h index 1dd3dfc107..e29dd2d5a0 100644 --- a/src/bthread/task_control.h +++ b/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); + } + void print_rq_sizes(std::ostream& os); double get_cumulated_worker_time(); @@ -182,6 +199,12 @@ friend bthread_t init_for_pthread_stack_trace(); size_t _pl_num_of_each_tag; std::vector _tagged_pl; + + // Signal optimization: per-tag count of workers currently parked in + // ParkingLot::wait(). Read (relaxed) by ready_to_run* to decide whether + // signal_task() can be skipped when no worker of that tag is waiting. + std::vector> _tagged_waiter_num; + // Per-tag CPU binding lists. _tag_cpus[tag] is the round-robin list of // CPU IDs to which workers of that tag are bound. Empty means no binding. std::vector> _tag_cpus; diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 72a6b91836..2c3599db97 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -930,7 +930,9 @@ void TaskGroup::ready_to_run(TaskMeta* meta, bool nosignal) { const int additional_signal = _num_nosignal; _num_nosignal = 0; _nsignaled += 1 + additional_signal; - _control->signal_task(1 + additional_signal, _tag); + if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { + _control->signal_task(1 + additional_signal, _tag); + } } } @@ -939,7 +941,9 @@ void TaskGroup::flush_nosignal_tasks() { if (val) { _num_nosignal = 0; _nsignaled += val; - _control->signal_task(val, _tag); + if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { + _control->signal_task(val, _tag); + } } } @@ -963,7 +967,9 @@ void TaskGroup::ready_to_run_remote(TaskMeta* meta, bool nosignal) { _remote_num_nosignal = 0; _remote_nsignaled += 1 + additional_signal; _remote_rq._mutex.unlock(); - _control->signal_task(1 + additional_signal, _tag); + if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { + _control->signal_task(1 + additional_signal, _tag); + } } } @@ -976,7 +982,9 @@ void TaskGroup::flush_nosignal_tasks_remote_locked(butil::Mutex& locked_mutex) { _remote_num_nosignal = 0; _remote_nsignaled += val; locked_mutex.unlock(); - _control->signal_task(val, _tag); + if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { + _control->signal_task(val, _tag); + } } void TaskGroup::ready_to_run_general(TaskMeta* meta, bool nosignal) { From 2535c2cd84b2f46a837fdf0ba27fa12985ef0083 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=AC=E9=A3=8E?= <807188583@qq.com> Date: Thu, 30 Jul 2026 00:29:58 +0800 Subject: [PATCH 2/3] bthread: fix init ordering, zero-init, and bounds checks per review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address Copilot review feedback on PR #3418: 1. Move set_task_control() BEFORE pthread_create() in TaskControl::init() to eliminate a race where workers could enter ParkingLot::wait() with _tc==nullptr, causing has_waiting_workers() to return false and signal_task() to be skipped — permanently sleeping the worker. 2. Explicitly zero-initialize _tagged_waiter_num elements in the TaskControl constructor body. butil::atomic's default constructor does not guarantee a zero value. 3. Add tag >= 0 bounds checks to has_waiting_workers(), waiter_add(), and waiter_sub() to prevent negative-tag (BTHREAD_TAG_INVALID=-1) array indexing (UB). 4. Include bthread/types.h directly in parking_lot.h for bthread_tag_t, making the header self-contained. Tested: bthread_butex_unittest 12/12 on macOS arm64. --- src/bthread/parking_lot.h | 1 + src/bthread/task_control.cpp | 30 ++++++++++++++++++++---------- src/bthread/task_control.h | 14 ++++++++++---- 3 files changed, 31 insertions(+), 14 deletions(-) diff --git a/src/bthread/parking_lot.h b/src/bthread/parking_lot.h index 2e841d98e5..d43a9469c0 100644 --- a/src/bthread/parking_lot.h +++ b/src/bthread/parking_lot.h @@ -24,6 +24,7 @@ #include #include "butil/atomicops.h" +#include "bthread/types.h" // bthread_tag_t #include "bthread/sys_futex.h" namespace bthread { diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index b33c50eae8..92f317bb94 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -226,7 +226,13 @@ TaskControl::TaskControl() , _tagged_waiter_num(FLAGS_task_group_ntags) , _tag_cpus(FLAGS_task_group_ntags) , _tag_next_worker_id(FLAGS_task_group_ntags) -{} +{ + // Explicitly zero-initialize all atomic counters. butil::atomic's + // default constructor does not guarantee a zero value. + for (size_t i = 0; i < _tagged_waiter_num.size(); ++i) { + _tagged_waiter_num[i].store(0, butil::memory_order_relaxed); + } +} int TaskControl::init_ed_priority_queues() { if (!_enable_priority_queue) { @@ -294,6 +300,19 @@ int TaskControl::init(int concurrency) { #endif // BRPC_BTHREAD_TRACER _workers.resize(_concurrency); + + // 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). This MUST be done BEFORE creating worker + // threads, otherwise a worker could enter wait() with _tc == nullptr, + // causing has_waiting_workers() to return false and signal_task() to + // be skipped — permanently sleeping the worker (lost wakeup). + for (size_t i = 0; i < _tagged_pl.size(); ++i) { + for (size_t j = 0; j < _pl_num_of_each_tag; ++j) { + _tagged_pl[i][j].set_task_control(this, (bthread_tag_t)i); + } + } + for (int i = 0; i < _concurrency; ++i) { auto arg = new WorkerThreadArgs(this, i % FLAGS_task_group_ntags); const int rc = pthread_create(&_workers[i], NULL, worker_thread, arg); @@ -308,15 +327,6 @@ 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) { - _tagged_pl[i][j].set_task_control(this, (bthread_tag_t)i); - } - } - // Wait for at least one group is added so that choose_one_group() // never returns NULL. // TODO: Handle the case that worker quits before add_group diff --git a/src/bthread/task_control.h b/src/bthread/task_control.h index e29dd2d5a0..9d7118a6cd 100644 --- a/src/bthread/task_control.h +++ b/src/bthread/task_control.h @@ -79,18 +79,24 @@ friend bthread_t init_for_pthread_stack_trace(); // 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. + // Returns true if the tag is valid AND at least one worker is parked. bool has_waiting_workers(bthread_tag_t tag) const { - return tag < (bthread_tag_t)_tagged_waiter_num.size() + return tag >= 0 + && 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. + // entering/leaving wait() for a given tag. No-op for invalid tags. void waiter_add(bthread_tag_t tag) { - _tagged_waiter_num[tag].fetch_add(1, butil::memory_order_relaxed); + if (tag >= 0 && tag < (bthread_tag_t)_tagged_waiter_num.size()) { + _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); + if (tag >= 0 && tag < (bthread_tag_t)_tagged_waiter_num.size()) { + _tagged_waiter_num[tag].fetch_sub(1, butil::memory_order_relaxed); + } } void print_rq_sizes(std::ostream& os); From 073e14a123f2a2ca3b8833e597566ba44c6f1b51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=AC=E9=A3=8E?= <807188583@qq.com> Date: Thu, 30 Jul 2026 00:44:51 +0800 Subject: [PATCH 3/3] bthread: revert has_waiting_workers gate, keep ParkingLot::signal optimization The per-tag has_waiting_workers() check in ready_to_run*() introduced a lost-wakeup race: 1. worker saves _last_pl_state via get_state() 2. worker scans queues, finds nothing 3. producer enqueues task 4. producer checks has_waiting_workers() -> false (waiter not registered yet) 5. producer skips signal_task() entirely, _pending_signal unchanged 6. worker registers as waiter, re-checks state -> unchanged -> sleeps 7. task stuck indefinitely The existing ParkingLot::signal() already handles this safely: it always updates _pending_signal (so the worker's re-check in step 6 detects the change) and only skips futex_wake when no waiter is present. The --parking_lot_no_signal_when_no_waiter flag already enables this per-lot optimization without the race. Revert all has_waiting_workers() gating in task_group.cpp (4 sites). Remove _tagged_waiter_num, waiter_add/sub, has_waiting_workers, set_task_control, and related forwarders from task_control.{h,cpp} and parking_lot.h. Restore parking_lot.h to its pre-optimization structure (retaining the pre-existing _no_signal_when_no_waiter fast path). Tested: bthread_butex_unittest 12/12 on macOS arm64. --- src/bthread/parking_lot.h | 42 ++---------------------------------- src/bthread/task_control.cpp | 26 +--------------------- src/bthread/task_control.h | 28 ------------------------ src/bthread/task_group.cpp | 16 ++++---------- 4 files changed, 7 insertions(+), 105 deletions(-) diff --git a/src/bthread/parking_lot.h b/src/bthread/parking_lot.h index d43a9469c0..5dcf2b0029 100644 --- a/src/bthread/parking_lot.h +++ b/src/bthread/parking_lot.h @@ -24,23 +24,12 @@ #include #include "butil/atomicops.h" -#include "bthread/types.h" // bthread_tag_t #include "bthread/sys_futex.h" 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); -void parking_lot_waiter_sub(TaskControl* tc, bthread_tag_t tag); - // Park idle workers. class BAIDU_CACHELINE_ALIGNMENT ParkingLot { public: @@ -56,15 +45,7 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot { ParkingLot() : _pending_signal(0), _waiter_num(0) - , _no_signal_when_no_waiter(FLAGS_parking_lot_no_signal_when_no_waiter) - , _tc(nullptr), _tag(0) {} - - // Set TaskControl pointer + owning tag for waiter tracking (called once - // during TaskControl::init after the ParkingLot array is constructed). - void set_task_control(TaskControl* tc, bthread_tag_t tag) { - _tc = tc; - _tag = tag; - } + , _no_signal_when_no_waiter(FLAGS_parking_lot_no_signal_when_no_waiter) {} // Wake up at most `num_task' workers. // Returns #workers woken up. @@ -84,37 +65,22 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot { // Wait for tasks. // If the `expected_state' does not match, wait() may finish directly. void wait(const State& expected_state) { - // Track this waiter BEFORE checking state, so that a concurrent - // ready_to_run() that calls has_waiting_workers() will see us - // even if it races between our state check and futex_wait(). - // This eliminates the lost-wakeup window. - if (_tc) { - parking_lot_waiter_add(_tc, _tag); - } if (_no_signal_when_no_waiter) { _waiter_num.fetch_add(1, butil::memory_order_relaxed); } if (get_state().val != expected_state.val) { - // State changed since caller last checked — signal already sent. - // Undo the tracking and return without sleeping. if (_no_signal_when_no_waiter) { _waiter_num.fetch_sub(1, butil::memory_order_relaxed); } - if (_tc) { - parking_lot_waiter_sub(_tc, _tag); - } return; } futex_wait_private(&_pending_signal, expected_state.val, NULL); if (_no_signal_when_no_waiter) { _waiter_num.fetch_sub(1, butil::memory_order_relaxed); } - if (_tc) { - parking_lot_waiter_sub(_tc, _tag); - } } - // Wakeup suspended wait() and make them unwaitable ever. + // Wakeup suspended wait() and make them unwaitable ever. void stop() { _pending_signal.fetch_or(1); futex_wake_private(&_pending_signal, 10000); @@ -128,10 +94,6 @@ class BAIDU_CACHELINE_ALIGNMENT ParkingLot { // In busy worker scenarios, signal overhead // can be reduced. bool _no_signal_when_no_waiter; - // TaskControl pointer for waiter tracking (may be NULL before init). - TaskControl* _tc; - // Owning tag, used to update the per-tag waiter counter in TaskControl. - bthread_tag_t _tag; }; } // namespace bthread diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index 92f317bb94..d37ed64678 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -65,11 +65,6 @@ DEFINE_bool(parking_lot_no_signal_when_no_waiter, false, "In busy worker scenarios, signal overhead can be reduced."); DEFINE_bool(enable_bthread_priority_queue, false, "Whether to enable priority queue"); -// Out-of-line forwarders for ParkingLot per-tag waiter accounting, so that -// parking_lot.h can stay free of the TaskControl definition. -void parking_lot_waiter_add(TaskControl* tc, bthread_tag_t tag) { tc->waiter_add(tag); } -void parking_lot_waiter_sub(TaskControl* tc, bthread_tag_t tag) { tc->waiter_sub(tag); } - DECLARE_int32(bthread_concurrency); DECLARE_int32(bthread_min_concurrency); DECLARE_int32(bthread_parking_lot_of_each_tag); @@ -223,16 +218,9 @@ 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) -{ - // Explicitly zero-initialize all atomic counters. butil::atomic's - // default constructor does not guarantee a zero value. - for (size_t i = 0; i < _tagged_waiter_num.size(); ++i) { - _tagged_waiter_num[i].store(0, butil::memory_order_relaxed); - } -} +{} int TaskControl::init_ed_priority_queues() { if (!_enable_priority_queue) { @@ -301,18 +289,6 @@ int TaskControl::init(int concurrency) { _workers.resize(_concurrency); - // 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). This MUST be done BEFORE creating worker - // threads, otherwise a worker could enter wait() with _tc == nullptr, - // causing has_waiting_workers() to return false and signal_task() to - // be skipped — permanently sleeping the worker (lost wakeup). - for (size_t i = 0; i < _tagged_pl.size(); ++i) { - for (size_t j = 0; j < _pl_num_of_each_tag; ++j) { - _tagged_pl[i][j].set_task_control(this, (bthread_tag_t)i); - } - } - for (int i = 0; i < _concurrency; ++i) { auto arg = new WorkerThreadArgs(this, i % FLAGS_task_group_ntags); const int rc = pthread_create(&_workers[i], NULL, worker_thread, arg); diff --git a/src/bthread/task_control.h b/src/bthread/task_control.h index 9d7118a6cd..f07663be65 100644 --- a/src/bthread/task_control.h +++ b/src/bthread/task_control.h @@ -76,29 +76,6 @@ friend bthread_t init_for_pthread_stack_trace(); 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. - // Returns true if the tag is valid AND at least one worker is parked. - bool has_waiting_workers(bthread_tag_t tag) const { - return tag >= 0 - && 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. No-op for invalid tags. - void waiter_add(bthread_tag_t tag) { - if (tag >= 0 && tag < (bthread_tag_t)_tagged_waiter_num.size()) { - _tagged_waiter_num[tag].fetch_add(1, butil::memory_order_relaxed); - } - } - void waiter_sub(bthread_tag_t tag) { - if (tag >= 0 && tag < (bthread_tag_t)_tagged_waiter_num.size()) { - _tagged_waiter_num[tag].fetch_sub(1, butil::memory_order_relaxed); - } - } - void print_rq_sizes(std::ostream& os); double get_cumulated_worker_time(); @@ -206,11 +183,6 @@ friend bthread_t init_for_pthread_stack_trace(); size_t _pl_num_of_each_tag; std::vector _tagged_pl; - // Signal optimization: per-tag count of workers currently parked in - // ParkingLot::wait(). Read (relaxed) by ready_to_run* to decide whether - // signal_task() can be skipped when no worker of that tag is waiting. - std::vector> _tagged_waiter_num; - // Per-tag CPU binding lists. _tag_cpus[tag] is the round-robin list of // CPU IDs to which workers of that tag are bound. Empty means no binding. std::vector> _tag_cpus; diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 2c3599db97..72a6b91836 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -930,9 +930,7 @@ void TaskGroup::ready_to_run(TaskMeta* meta, bool nosignal) { const int additional_signal = _num_nosignal; _num_nosignal = 0; _nsignaled += 1 + additional_signal; - if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { - _control->signal_task(1 + additional_signal, _tag); - } + _control->signal_task(1 + additional_signal, _tag); } } @@ -941,9 +939,7 @@ void TaskGroup::flush_nosignal_tasks() { if (val) { _num_nosignal = 0; _nsignaled += val; - if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { - _control->signal_task(val, _tag); - } + _control->signal_task(val, _tag); } } @@ -967,9 +963,7 @@ void TaskGroup::ready_to_run_remote(TaskMeta* meta, bool nosignal) { _remote_num_nosignal = 0; _remote_nsignaled += 1 + additional_signal; _remote_rq._mutex.unlock(); - if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { - _control->signal_task(1 + additional_signal, _tag); - } + _control->signal_task(1 + additional_signal, _tag); } } @@ -982,9 +976,7 @@ void TaskGroup::flush_nosignal_tasks_remote_locked(butil::Mutex& locked_mutex) { _remote_num_nosignal = 0; _remote_nsignaled += val; locked_mutex.unlock(); - if (BAIDU_UNLIKELY(_control->has_waiting_workers(_tag))) { - _control->signal_task(val, _tag); - } + _control->signal_task(val, _tag); } void TaskGroup::ready_to_run_general(TaskMeta* meta, bool nosignal) {