ConditionVariable.cpp

struct ConditionVariableFallback {
    inline bool initialize() {
        BOOL r;

        wakeupType_ = NONE;
        waitersCount_ = 0;

        // Initialize a critical section that protects the waiters and release counters.
        InitializeCriticalSection(&stateLock_);

        // Create an auto-reset event for wake signaling.
        signalEvent_ = CreateEventW(NULL, FALSE, FALSE, NULL);
        if (!signalEvent_)
            goto error1;

        // Create a manual-reset event for broadcast signaling.
        broadcastEvent_ = CreateEventW(NULL, TRUE, FALSE, NULL);
        if (!broadcastEvent_)
            goto error2;

        // Create a semaphore that prevents threads from entering a wait state
        // or waking other threads while a wakeup is ongoing.
        wakeupSemaphore_ = CreateSemaphoreW(NULL, 1, 1, NULL);
        if (!wakeupSemaphore_)
           goto error3;

        return true;

      error3:
        r = CloseHandle(broadcastEvent_);
        assert(r);
      error2:
        r = CloseHandle(signalEvent_);
        assert(r);
      error1:
        DeleteCriticalSection(&stateLock_);
        return false;
    }

    inline void destroy() {
        DeleteCriticalSection(&stateLock_);
        BOOL r = CloseHandle(signalEvent_) &&
                 CloseHandle(broadcastEvent_) &&
                 CloseHandle(wakeupSemaphore_);
        assert(r);
    }

    inline void signal() {
        DWORD r;
        bool hasWaiters;
        
        // Ensure that only one thread can be waking other threads at a time.
        r = WaitForSingleObject(wakeupSemaphore_, INFINITE);
        assert(r == WAIT_OBJECT_0);

        // Check if there are any waiters. If so, update the state to SIGNAL.
        EnterCriticalSection(&stateLock_);
        assert(wakeupType_ == NONE);
        hasWaiters = waitersCount_ > 0;
        if (hasWaiters)
            wakeupType_ = SIGNAL;
        LeaveCriticalSection(&stateLock_);

        if (!hasWaiters) {
            // If there are no waiters, release the wakeupSemaphore ourselves.
            bool success = ReleaseSemaphore(wakeupSemaphore_, 1, NULL);
            assert(success);
            return;
        }

        // If there are any waiters, set the signal event. The woken up 
        // (or last) waiter is responsible for releasing the wakeupSemaphore_.
        bool success = SetEvent(signalEvent_);
        assert(success);
    }


    inline void broadcast() {
        DWORD r;
        bool hasWaiters;
        
        // Ensure that only one thread can be waking other threads at a time.
        r = WaitForSingleObject(wakeupSemaphore_, INFINITE);
        assert(r == WAIT_OBJECT_0);

        // Check if there are any waiters. If so, update the state to SIGNAL.
        EnterCriticalSection(&stateLock_);

        hasWaiters = waitersCount_ > 0;
        
        assert(wakeupType_ == NONE);
        if (hasWaiters)
            wakeupType_ = BROADCAST;

        LeaveCriticalSection(&stateLock_);

        if (!hasWaiters) {
            // If there are no waiters, release the wakeupSemaphore ourselves.
            bool success = ReleaseSemaphore(wakeupSemaphore_, 1, NULL);
            assert(success);
            return;
        }

        // If there are any waiters, set the signal event. The woken up 
        // (or last) waiter is responsible for releasing the wakeupSemaphore_.
        bool success = SetEvent(broadcastEvent_);
        assert(success);
    }

    inline bool wait(CRITICAL_SECTION* userLock, DWORD msec) {
        DWORD waitResult;

        // Make sure that we can't enter a wait state while a signal or a
        // broadcast is happening.
        waitResult = WaitForSingleObject(wakeupSemaphore_, INFINITE);
        assert(waitResult == WAIT_OBJECT_0);
        
        // Register ourselves as a waiter.
        EnterCriticalSection(&stateLock_);
        assert(wakeupType_ == NONE);
        waitersCount_++;
        LeaveCriticalSection(&stateLock_);

        // Now that that this thread has been enlisted as a waiters, it is safe
        // for other threads to do a wakeup.
        BOOL success = ReleaseSemaphore(wakeupSemaphore_);
        assert(success);

        // Release the caller's mutex.
        LeaveCriticalSection(userLock);

        // Wait for either event to become signaled, which happens when
        // signal() or broadcast() is called, or for a timeout.
        HANDLE handles[2] = { signalEvent_, broadcastEvent_ };
        DWORD waitResult = WaitForMultipleObjects(2, handles, FALSE, msec);

        bool releaseWakeupSemaphore = false;

        // Decrease the waiters count, and check the wakeup condition, and
        // decide whether we need to release the releaseMutex_;
        EnterCriticalSection(&stateLock_);
        
        --waitersCount_;

        switch (wakeupType_) {
          case NONE:
            // If there isn't a signal or a broadcast going on, the wait could
            // only have returned due to a timeout.
            assert(waitResult == WAIT_TIMEOUT);
            break;

          case SIGNAL:
            assert(waitResult == WAIT_TIMEOUT || waitResult == WAIT_OBJECT_0);
            
            // If we were woken by the signal event, just release the wake
            // semaphore.
            if (waitResult == WAIT_OBJECT_0) {
                wakeupType_ = NONE;
                releaseWakeupSemaphore = true; 
            }

            // In theory our wait could time out just as signal() is setting
            // the signal event to signaled state. This isn't a problem if 
            // there is more than one thread sleeping on this condition 
            // variable; one of the "other" threads can wake on the event.
            // However if we timed out and we are the *only* waiter then then
            // now the signal event is in a signaled state, and the wake lock
            // is held, and we need to fix this up.
            if (waitResult == WAIT_TIMEOUT && waitersCount_ == 0) {
                BOOL success = ResetEvent(signalEvent_);
                assert(success);

                wakeupType_ = NONE;
                releaseWakeupSemaphore = true;
            }
            break;

          case BROADCAST:
            assert(waitResult == WAIT_TIMEOUT || 
                   waitResult == WAIT_OBJECT_0 + 1);

            // Regardless of whether we timed out or the broadcast event being
            // signaled, if we are the last waiter:
            //   * reset the manual-reset broadcast event.
            //   * set the state back to NONE
            //   * release the wake lock so new threads can enter the wait
            //     and other wakes can happen.
            if (waitersCount_ == 0) {
                bool success = ResetEvent(broadcastEvent_);
                assert(success);

                wakeupType_ = NONE;
                releaseWakeupSemaphore = true;
            }
            break;

          default:
            assert(0);
        }

        LeaveCriticalSection(&stateLock_);

        if (releaseWakeupSemaphore) {
            bool success = ReleaseSemaphore(wakeupSemaphore_, 1, NULL);
            assert(success == 0);
        }

        // Reacquire the user mutex.
        EnterCriticalSection(userLock);

        return waitResult != WAIT_TIMEOUT;
    }

   private:
    enum WakeupType {
        NONE = 0,
        SIGNAL,
        BROADCAST
    };

    size_t waitersCount_;
    WakeupType wakeupType_;
    HANDLE wakeupSemaphore_;
    CRITICAL_SECTION stateLock_;
    HANDLE signalEvent_;
    HANDLE broadcastEvent_;
};