Evaluating the SetEvent Solution

• Unfairness -- The semantics of the POSIX pthread_cond_broadcast function is to wake up all threads currently blocked in wait calls on the condition variable. The awakened threads then compete for the external_mutex. To ensure fairness, all of these threads should be released from their pthread_cond_wait calls and allowed to recheck their condition expressions before other threads can successfully complete a wait on the condition variable.

  Unfortunately, the SetEvent implementation above does not guarantee that all threads sleeping on the condition variable when cond_broadcast is called will acquire the external_mutex and check their condition expressions. Although the Pthreads specification does not mandate this degree of fairness, the lack of fairness can cause starvation.

  To illustrate the unfairness problem, imagine there are 2 threads, C₁ and C₂, that are blocked in pthread_cond_wait on condition variable not_empty_ that is guarding a thread-safe message queue. Another thread, P₁ then places two messages onto the queue and calls pthread_cond_broadcast. If C₁ returns from pthread_cond_wait, dequeues and processes the message, and immediately waits again then it and only it may end up acquiring both messages. Thus, C₂ will never get a chance to dequeue a message and run.

  The following illustrates the sequence of events:

  1. Thread C1 attempts to dequeue and waits on CV non_empty_
  2. Thread C2 attempts to dequeue and waits on CV non_empty_
  3. Thread P1 enqueues 2 messages and broadcasts to CV not_empty_
  4. Thread P1 exits
  5. Thread C1 wakes up from CV not_empty_, dequeues a message and runs
  6. Thread C1 waits again on CV not_empty_, immediately dequeues the 2nd message and runs
  7. Thread C1 exits
  8. Thread C2 is the only thread left and blocks forever since not_empty_ will never be signaled

  Depending on the algorithm being implemented, this lack of fairness may yield concurrent programs that have subtle bugs. Of course, application developers should not rely on the fairness semantics of pthread_cond_broadcast. Howver, there are many cases where fair implementations of condition variables can simplify application code.

• Incorrectness -- A variation on the unfairness problem described above occurs when a third consumer thread, C₃, is allowed to slip through even though it was not waiting on condition variable not_empty_ when a broadcast occurred.

  To illustrate this, we will use the same scenario as above: 2 threads, C₁ and C₂, are blocked dequeuing messages from the message queue. Another thread, P₁ then places two messages onto the queue and calls pthread_cond_broadcast. C₁ returns from pthread_cond_wait, dequeues and processes the message. At this time, C₃ acquires the external_mutex, calls pthread_cond_wait and waits on the events in WaitForMultipleObjects. Since C₂ has not had a chance to run yet, the BROADCAST event is still signaled. C₃ then returns from WaitForMultipleObjects, and dequeues and processes the message in the queue. Thus, C₂ will never get a chance to dequeue a message and run.

  The following illustrates the sequence of events:

  1. Thread C1 attempts to dequeue and waits on CV non_empty_
  2. Thread C2 attempts to dequeue and waits on CV non_empty_
  3. Thread P1 enqueues 2 messages and broadcasts to CV not_empty_
  4. Thread P1 exits
  5. Thread C1 wakes up from CV not_empty_, dequeues a message and runs
  6. Thread C1 exits
  7. Thread C3 waits on CV not_empty_, immediately dequeues the 2nd message and runs
  8. Thread C3 exits
  9. Thread C2 is the only thread left and blocks forever since not_empty_ will never be signaled

  In the above case, a thread that was not waiting on the condition variable when a broadcast occurred was allowed to proceed. This leads to incorrect semantics for a condition variable.

• Increased serialization overhead -- The implementation shown above uses the waiters_count_lock_ critical section to protect the waiters_count_ from being corrupted by race conditions. This additional serialization overhead is not necessary as long as pthread_cond_signal and pthread_cond_broadcast are always called by a thread that has locked the same external_mutex used by pthread_cond_wait. For instance, application code that always uses the following idiom does not require this additional serialization:

  
  
  pthread_mutex_t external_mutex;
  pthread_cond_t cv;
  
  void 
  release_resources (int resources_released) 
  {
    // Acquire the lock.
    pthread_mutex_lock (&external_mutex);
  
    // Atomically modify shared state here...
  
    // Could also use <pthread_cond_broadcast>.
    pthread_cond_signal (&cv);
  
    // Release the lock.
    pthread_mutex_unlock (&external_mutex);
  }
  

  In contrast, application code that uses the following optimization does require the additional serialization:

  
  void 
  release_resources (int resources_released) 
  {
    // Acquire the lock.
    pthread_mutex_lock (&external_mutex);
    int can_signal = 0;
  
    // Atomically modify shared state here...
  
    // Keep track of whether we can signal or not
    // as a result of the updated shared state.
    can_signal = 1;
  
    // Release the lock.
    pthread_mutex_unlock (&external_mutex);
  
    if (can_signal)
      // Could also use <pthread_cond_broadcast>.
      pthread_cond_signal (&cv); 
  }
  

  The code shown above puts the pthread_cond_signal after the pthread_mutex_unlock to minimize the number of trips through the condition variable scheduler. The problem with this optimization, of course, is that race conditions can occur on the waiters_count_ if it was tested/modified without being protected by the external_mutex. Note that the PulseEvents solution did not have this race condition since it did not have any directly mutable internal state.

  In general, race conditions can occur with condition variable implementations that use internal state if threads don't acquire the external_mutex when calling pthread_cond_signal and pthread_cond_broadcast. The POSIX Pthread specification allows these functions to be called by a thread whether or not that thread has currently locked the external_mutex. Therefore, although the use of the internal lock can increase overhead, it is necessary to provide robust, standard-compliant condition variable programming model.

typedef struct
{
  int waiters_count_;
  // Count of the number of waiters.

  CRITICAL_SECTION waiters_count_lock_;
  // Serialize access to <waiters_count_>.

  int release_count_;
  // Number of threads to release via a <pthread_cond_broadcast> or a
  // <pthread_cond_signal>. 
  
  int wait_generation_count_;
  // Keeps track of the current "generation" so that we don't allow
  // one thread to steal all the "releases" from the broadcast.

  HANDLE event_;
  // A manual-reset event that's used to block and release waiting
  // threads. 
} pthread_cond_t;

int 
pthread_cond_init (pthread_cond_t *cv, 
                   const pthread_condattr_t *);
{
  cv->waiters_count_ = 0;
  cv->wait_generation_count_ = 0;
  cv->release_count_ = 0;

  // Create a manual-reset event.
  cv->event_ = CreateEvent (NULL,  // no security
                            TRUE,  // manual-reset
                            FALSE, // non-signaled initially
                            NULL); // unnamed
}

int
pthread_cond_wait (pthread_cond_t *cv,
                   pthread_mutex_t *external_mutex)
{
  // Avoid race conditions.
  EnterCriticalSection (&cv->waiters_count_lock_);

  // Increment count of waiters.
  cv->waiters_count_++;

  // Store current generation in our activation record.
  int my_generation = cv->wait_generation_count_;

  LeaveCriticalSection (&cv->waiters_count_lock_);
  LeaveCriticalSection (external_mutex);

  for (;;) {
    // Wait until the event is signaled.
    WaitForSingleObject (cv->event_, INFINITE);

    EnterCriticalSection (&cv->waiters_count_lock_);
    // Exit the loop when the <cv->event_> is signaled and
    // there are still waiting threads from this <wait_generation>
    // that haven't been released from this wait yet.
    int wait_done = cv->release_count_ > 0
                    && cv->wait_generation_count_ != my_generation;
    LeaveCriticalSection (&cv->waiters_count_lock_);

    if (wait_done)
      break;
  }

  EnterCriticalSection (external_mutex);
  EnterCriticalSection (&cv->waiters_count_lock_);
  cv->waiters_count_--;
  cv->release_count_--;
  int last_waiter = cv->release_count_ == 0;
  LeaveCriticalSection (&cv->waiters_count_lock_);

  if (last_waiter)
    // We're the last waiter to be notified, so reset the manual event.
    ResetEvent (cv->event_);
}

int
pthread_cond_signal (pthread_cond_t *cv)
{
  EnterCriticalSection (&cv->waiters_count_lock_);
  if (cv->waiters_count_ > cv->release_count_) {
    SetEvent (cv->event_); // Signal the manual-reset event.
    cv->release_count_++;
    cv->wait_generation_count++;
  }
  LeaveCriticalSection (&cv->waiters_count_lock_);
}

int
pthread_cond_broadcast (pthread_cond_t *cv)
{
  EnterCriticalSection (&cv->waiters_count_lock_);
  if (cv->waiters_count_ > 0) {  
    SetEvent (cv->event_);
    // Release all the threads in this generation.
    cv->release_count_ = cv->waiters_count_;

    // Start a new generation.
    cv->wait_generation_count_++;
  }
  LeaveCriticalSection (&cv->waiters_count_lock_);
}

typedef struct
{
  int waiters_count_;
  // Number of waiting threads.

  CRITICAL_SECTION waiters_count_lock_;
  // Serialize access to <waiters_count_>.

  HANDLE sema_;
  // Semaphore used to queue up threads waiting for the condition to
  // become signaled. 

  HANDLE waiters_done_;
  // An auto-reset event used by the broadcast/signal thread to wait
  // for all the waiting thread(s) to wake up and be released from the
  // semaphore. 

  size_t was_broadcast_;
  // Keeps track of whether we were broadcasting or signaling.  This
  // allows us to optimize the code if we're just signaling.
} pthread_cond_t;

typedef HANDLE pthread_mutex_t;

int 
pthread_cond_init (pthread_cond_t *cv,
                   const pthread_condattr_t *)
{
  cv->waiters_count_ = 0;
  cv->was_broadcast_ = 0;
  cv->sema_ = CreatedSemaphore (NULL,       // no security
                                0,          // initially 0
                                0x7fffffff, // max count
                                NULL);      // unnamed 
  InitializeCriticalSection (&cv->waiters_count_lock_);
  cv->waiters_done_ = CreateEvent (NULL,  // no security
                                   FALSE, // auto-reset
                                   FALSE, // non-signaled initially
                                   NULL); // unnamed
}

int
pthread_cond_wait (pthread_cond_t *cv, 
                   pthread_mutex_t *external_mutex)
{
  // Avoid race conditions.
  EnterCriticalSection (&cv->waiters_count_lock_);
  cv->waiters_count_++;
  LeaveCriticalSection (&cv->waiters_count_lock_);

  // This call atomically releases the mutex and waits on the
  // semaphore until <pthread_cond_signal> or <pthread_cond_broadcast>
  // are called by another thread.
  SignalObjectAndWait (*external_mutex, cv->sema_, INFINITE, FALSE);

  // Reacquire lock to avoid race conditions.
  EnterCriticalSection (&cv->waiters_count_lock_);

  // We're no longer waiting...
  cv->waiters_count_--;

  // Check to see if we're the last waiter after <pthread_cond_broadcast>.
  int last_waiter = cv->was_broadcast_ && cv->waiters_count_ == 0;

  LeaveCriticalSection (&cv->waiters_count_lock_);

  // If we're the last waiter thread during this particular broadcast
  // then let all the other threads proceed.
  if (last_waiter)
    // This call atomically signals the <waiters_done_> event and waits until
    // it can acquire the <external_mutex>.  This is required to ensure fairness. 
    SignalObjectAndWait (cv->waiters_done_, *external_mutex, INFINITE, FALSE);
  else
    // Always regain the external mutex since that's the guarantee we
    // give to our callers. 
    WaitForSingleObject (*external_mutex);
}

int
pthread_cond_signal (pthread_cond_t *cv)
{
  EnterCriticalSection (&cv->waiters_count_lock_);
  int have_waiters = cv->waiters_count_ > 0;
  LeaveCriticalSection (&cv->waiters_count_lock_);

  // If there aren't any waiters, then this is a no-op.  
  if (have_waiters)
    ReleaseSemaphore (cv->sema_, 1, 0);
}

int
pthread_cond_broadcast (pthread_cond_t *cv)
{
  // This is needed to ensure that <waiters_count_> and <was_broadcast_> are
  // consistent relative to each other.
  EnterCriticalSection (&cv->waiters_count_lock_);
  int have_waiters = 0;

  if (cv->waiters_count_ > 0) {
    // We are broadcasting, even if there is just one waiter...
    // Record that we are broadcasting, which helps optimize
    // <pthread_cond_wait> for the non-broadcast case.
    cv->was_broadcast_ = 1;
    have_waiters = 1;
  }

  if (have_waiters) {
    // Wake up all the waiters atomically.
    ReleaseSemaphore (cv->sema_, cv->waiters_count_, 0);

    LeaveCriticalSection (&cv->waiters_count_lock_);

    // Wait for all the awakened threads to acquire the counting
    // semaphore. 
    WaitForSingleObject (cv->waiters_done_, INFINITE);
    // This assignment is okay, even without the <waiters_count_lock_> held 
    // because no other waiter threads can wake up to access it.
    cv->was_broadcast_ = 0;
  }
  else
    LeaveCriticalSection (&cv->waiters_count_lock_);
}