C++20 thread possibly waiting on std::atomic forever

Question

Consider the following example code where thread A pushes functions on a queue and thread B executes those when popping from the queue:

std::atomic<uint32_t> itemCount;

//Executed by thread A
void run(std::function<void()> function) {

    if (queue.push(std::move(function))) {
        itemCount.fetch_add(1, std::memory_order_acq_rel);
        itemCount.notify_one();
    }

}


//Executed by thread B
void threadMain(){

    std::function<void()> function;

    while(true){

        if (queue.pop(function)) {

            itemCount.fetch_sub(1, std::memory_order_acq_rel);
            function();

        }else{
            itemCount.wait(0, std::memory_order_acquire);
        }

    }

}

where queue is a concurrent queue which has a push and a pop function, each returning a bool indicating whether the given operation was successful. So push returns false if it's full and pop returns false if it's empty.

Now I was wondering if the code is thread-safe under all circumstances. Let's suppose thread B's pop fails and is about to invoke std::atomic<T>::wait. At the same time, thread A pushes a new element while thread B checks the initial wait condition. Since itemCount hasn't changed yet, it fails.

Immediately after that, thread A increments the counter and tries to notify one waiting thread (although thread B doesn't wait internally yet). Thread B finally waits on the atomic, causing the thread to never wake up again due to the lost signal despite there being an element in the queue. That only stops as soon as a new element is being pushed on the queue, notifying B to continue execution.

I wasn't able to reproduce this situation manually since the timing is close to impossible to get right.

Is this a serious concern or impossible to happen? What (preferably atomic) alternatives do exist in order to account for such rare situations?

EDIT: Just to mention, the queue is not blocking and only utilizes atomic operations.

---

The reason I'm asking is I don't understand how it's possible to implement an atomic wait operation. Although the standard says the whole operation is atomic (consisting of a load + predicate check + wait), in the implementation I'm using std::atomic<T>::wait is implemented roughly as follows:

void wait(const _TVal _Expected, const memory_order _Order = memory_order_seq_cst) const noexcept {
    _Atomic_wait_direct(this, _Atomic_reinterpret_as<long>(_Expected), _Order);
}

where _Atomic_wait_direct is defined as

template <class _Ty, class _Value_type>
void _Atomic_wait_direct(
    const _Atomic_storage<_Ty>* const _This, _Value_type _Expected_bytes, const memory_order _Order) noexcept {
    const auto _Storage_ptr = _STD addressof(_This->_Storage);
    for (;;) {
        const _Value_type _Observed_bytes = _Atomic_reinterpret_as<_Value_type>(_This->load(_Order));
        if (_Expected_bytes != _Observed_bytes) {
            return;
        }

        __std_atomic_wait_direct(_Storage_ptr, &_Expected_bytes, sizeof(_Value_type), _Atomic_wait_no_timeout);
    }
}

We can clearly see that there is an atomic load with the specified memory order to check the state of the atomic itself. However, I don't see how the whole operation can be considered atomic since there is a comparision right before the call to __std_atomic_wait_direct.

With condition variables, the predicate itself is secured by a mutex but how is the atomic itself secured here?

Answer 1

Here's what the standard has to say:

[intro.races]/4 All modifications to a particular atomic object M occur in some particular total order, called the modification order of M.

[atomics.wait]/4 A call to an atomic waiting operation on an atomic object M is eligible to be unblocked by a call to an atomic notifying operation on M if there exist side effects X and Y on M such that:
(4.1) — the atomic waiting operation has blocked after observing the result of X,
(4.2) — X precedes Y in the modification order of M, and
(4.3) — Y happens before the call to the atomic notifying operation.

You posit the following scenario:

1. The current value of itemCount is zero, either from original initialization or a previous fetch_sub.
2. wait loads itemCount and observes the value of 0.
3. The other thread calls fetch_add and notify_one
4. wait goes blocking, as it believes the now-stale value of 0.

In this scenario, M is itemCount, X is the old fetch_sub that brought the value to 0 (we assume that it happened long ago and is properly visible to all threads) and Y is the fetch_add that changes the value to 1.

The standard says that the wait call (spanning steps 2 and 4) is in fact eligible to be unblocked by notify_one call on step 3. Indeed:

(4.1) - wait has blocked after observing itemCount == 0 (if it has not, then the problem fails to arise).
(4.2) - fetch_sub precedes fetch_add in the modification order of itemCount (by assumption that fetch_sub happened long ago).
(4.3) - fetch_add happens before (in fact, is sequenced before) notify_one; they are called one after the other by the same thread.

Therefore, a conforming implementation must either not allow wait to block in the first place, or have notify_one wake it up; it cannot allow the notification to be missed.

---

The only place where memory orders figure in this discussion is in (4.1), "the atomic waiting operation has blocked after observing the result of X". Perhaps fetch_add actually occurred before wait (by wall clock), but was not visible to wait, so it went blocking anyway. But it doesn't matter to the outcome - either wait observes the result of fetch_add and doesn't block at all; or it observes the result of the old fetch_sub and blocks, but then notfiy_one is required to wake it up.

Answer 2

Just to point out what may have led to the confusion in the first place: the function __std_atomic_wait_direct, all by itself, does the compare and block in an atomic fashion. It boils down to either a call to WaitOnAddress, or else uses a condition variable, where the compare is done while holding the associated mutex.

The if (_Expected_bytes != _Observed_bytes) is merely an optimization. If the variable is already different from the expected value, we can return right away without making the relatively expensive call to an underlying atomic function. It doesn't need to be atomic with the block, and could be removed altogether without affecting the semantics.