bthread: skip signal_task when no worker is waiting

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);

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);

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<TaggedParkingLot> _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<std::vector<unsigned>> _tag_cpus;