diff --git a/src/bthread/parking_lot.h b/src/bthread/parking_lot.h index bbc9a7c3fd..5dcf2b0029 100644 --- a/src/bthread/parking_lot.h +++ b/src/bthread/parking_lot.h @@ -65,20 +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) { - if (get_state().val != expected_state.val) { - // Fast path, no need to futex_wait. - return; - } if (_no_signal_when_no_waiter) { _waiter_num.fetch_add(1, butil::memory_order_relaxed); } + if (get_state().val != expected_state.val) { + if (_no_signal_when_no_waiter) { + _waiter_num.fetch_sub(1, butil::memory_order_relaxed); + } + 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); } } - // 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); diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index 3ad3aa508d..d37ed64678 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -288,6 +288,7 @@ int TaskControl::init(int concurrency) { #endif // BRPC_BTHREAD_TRACER _workers.resize(_concurrency); + 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 1dd3dfc107..f07663be65 100644 --- a/src/bthread/task_control.h +++ b/src/bthread/task_control.h @@ -73,7 +73,7 @@ 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); } void print_rq_sizes(std::ostream& os); @@ -182,6 +182,7 @@ friend bthread_t init_for_pthread_stack_trace(); size_t _pl_num_of_each_tag; std::vector _tagged_pl; + // 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;