revert "Fix data race in async_scope"

Author: lixin-wei

include/exec/async_scope.hpp

@@ -51,7 +51,6 @@ namespace exec {
       inplace_stop_source __stop_source_{};
       mutable std::mutex __lock_{};
       mutable __std::atomic_ptrdiff_t __active_ = 0;
-      mutable __std::atomic_ptrdiff_t __pending_notifiers_ =
         0; // Track in-flight __complete() calls
       mutable __intrusive_queue<&__task::__next_> __waiters_{};
 
@@ -85,14 +84,8 @@ namespace exec {
           // the waiter is queued but after __active is checked, the waiter will never be notified
           std::unique_lock __guard{this->__scope_->__lock_};
           auto& __active = this->__scope_->__active_;
-          auto& __pending = this->__scope_->__pending_notifiers_;
           auto& __waiters = this->__scope_->__waiters_;
-          // Also check __pending_notifiers_ to avoid race with in-flight __complete() calls.
-          // A __complete() that did fetch_sub but hasn't locked the mutex yet will have
-          // incremented __pending_notifiers_, preventing us from completing immediately.
-          if (
-            __active.load(__std::memory_order_acquire) != 0
-            || __pending.load(__std::memory_order_acquire) != 0) {
+          if (__active.load(__std::memory_order_acquire) != 0) {
             __waiters.push_back(this);
             return;
           }
@@ -165,13 +158,9 @@ namespace exec {
         __nest_op_base<_ReceiverId>* __op_;
 
         static void __complete(const __impl* __scope) noexcept {
-          // Increment pending BEFORE fetch_sub to close race window with on_empty().
-          // This ensures on_empty() sees pending > 0 if we're about to lock the mutex.
-          __scope->__pending_notifiers_.fetch_add(1, __std::memory_order_acquire);
           auto& __active = __scope->__active_;
+          std::unique_lock __guard{__scope->__lock_};
           if (__active.fetch_sub(1, __std::memory_order_acq_rel) == 1) {
-            std::unique_lock __guard{__scope->__lock_};
-            __scope->__pending_notifiers_.fetch_sub(1, __std::memory_order_release);
             auto __local_waiters = std::move(__scope->__waiters_);
             __guard.unlock();
             __scope = nullptr;
@@ -181,8 +170,6 @@ namespace exec {
               __next->__notify_waiter(__next);
               // __scope must be considered deleted
             }
-          } else {
-            __scope->__pending_notifiers_.fetch_sub(1, __std::memory_order_release);
           }
         }