| Strategies for Implementing POSIX Condition Variables on Win32Strategies for Implementing POSIX Condition Variables on Win32Douglas C. Schmidt andIrfan PyaraliDepartment of Computer ScienceWashington University, St. Louis, Missouri1. IntroductionThe threading API provided by the Microsoft Win32[Richter] family of operating systems (i.e.,Windows NT, Windows '95, and Windows CE) provides some of the sameconcurrency constructs defined by the POSIX Pthreads specification[Pthreads]. For instance, they both support mutexes,which serialize access to shared state. However, Win32 lacksfull-fledged condition variables, which are a synchronizationmechanism used by threads to wait until a condition expressioninvolving shared data attains a particular state. The lack of condition variables in Win32 makes it harder to implementcertain concurrency abstractions, such as thread-safe message queuesand thread pools. This article explores various techniques andpatterns for implementing POSIX condition variables correctly and/orfairly on Win32. Section 2 explains what condition variables are andshows how to use them. Secion 3 explains alternative strategies forimplementing POSIX condition variables using Win32 synchronizationprimitives. A subsequent article will describe how the WrapperFacade pattern and various C++ language features can help reducecommon mistakes that occur when programming condition variables. 2. Overview of POSIX Condition VariablesCondition variables are a powerful synchronization mechanism definedin the POSIX Pthreads specification. They provide a different type ofsynchronization than locking mechanisms like mutexes. For instance, amutex is used to cause other threads to wait while the threadholding the mutex executes code in a critical section. In contrast, acondition variable is typically used by a thread to makeitself wait until an expression involving shared data attainsa particular state. In general, condition variables are more appropriate than mutexes forsituations involving complex condition expressions or schedulingbehaviors. For instance, condition variables are often used toimplement thread-safe message queues [Schmidt], whichprovide a ``producer/consumer'' communication mechanism for passingmessages between multiple threads. In this use-case, conditionvariables block producer threads when a message queue is ``full'' andblock consumer threads when the queue is ``empty.'' 2.1. POSIX Condition Variable OperationsThe following are the four primary operations defined on POSIXcondition variables: pthread_cond_wait (pthread_cond_t *cv, pthread_mutex_t*external_mutex) -- This function waits for conditioncv to be notified. When called, it atomically (1)releases the associated external_mutex (which the callermust hold while evaluating the condition expression) and (2) goes tosleep awaiting a subsequent notification from another thread (via thepthread_cond_signal orpthread_cond_broadcast operations described next). Theexternal_mutex will be locked whenpthread_cond_wait returns. pthread_cond_signal (pthread_cond_t *cv) -- Thisfunction notifies one thread waiting on condition variablecv. pthread_cond_broadcast (pthread_cond_t *cv) -- Thisfunction notifies all threads that are current waiting oncondition variable cv. pthread_cond_timedwait (pthread_cond_t *cv, pthread_mutex_t*external_mutex, const struct timespec *abstime) -- Thisfunction is a time-based variant of pthread_cond_wait.It waits up to abstime amount of time for cvto be notified. If abstime elapses beforecv is notified, the function returns back to the callerwith an ETIME result, signifying that a timeout hasoccurred. Even in the case of timeouts, theexternal_mutex will be locked whenpthread_cond_timedwait returns.The Pthreads API defines two other operations,pthread_cond_init and pthread_cond_destroy,which initialize and destroy instances of POSIX condition variables,respectively. 2.2. Using Condition Variables to Implement Counting SemaphoresThe expression waited upon by a condition variable can be arbitrarilycomplex. For instance, a thread may need to block until a certainexpression involving one or more data items shared by other threadsbecomes true. To illustrate how this works, we'll implement acounting semaphore abstraction below using condition variables. Counting semaphores are a synchronization mechanism commonly used toserialize or coordinate multiple threads of control. Conceptually,counting semaphores are non-negative integers that can be incrementedand decremented atomically. If a thread tries to decrement asemaphore whose value equals 0 this thread is suspended waiting foranother thread to increment the semaphore count above 0. Counting semaphores are often used to keep track of changes in thestate of objects shared by multiple threads in a process. Forinstance, they can record the occurrence of a particular event.Unlike condition variables, semaphores maintain state. Therefore,they allow threads to make decisions based upon this state, even ifthe event has occurred in the past. The standard POSIX Pthreads interface doesn't contain a countingsemaphore abstraction. However, it's straightforward to create oneusing mutexes and condition variables, as shown below. First, we define a struct that holds the stateinformation necessary to implement a counting semaphore: struct sema_t{private: u_int count_; // Current count of the semaphore. u_long waiters_count_; // Number of threads that have called <sema_wait>. pthread_mutex_t lock_; // Serialize access to <count_> and <waiters_count_>. pthread_cond_t count_nonzero_; // Condition variable that blocks the <count_> 0.};Instances of type sema_t are created by the followingsema_init factory (error checking is omitted to savespace):intsema_init (sema_t *s, u_int initial_count){ pthread_mutex_init (&s->lock_); pthread_cond_init (&s->count_nonzero_); s->count_ = initial_count; s->waiters_count_ = 0;}The sema_wait function shown below blocks the callingthread until the semaphore count pointed to by s isgreater than 0, at which point the count is atomicallydecremented. int sema_wait (sema_t *s){ // Acquire mutex to enter critical section. pthread_mutex_lock (&s->lock_); // Keep track of the number of waiters so that <sema_post> works correctly. s->waiters_count_++; // Wait until the semaphore count is > 0, then atomically release // <lock_> and wait for <count_nonzero_> to be signaled. while (s->count_ == 0) pthread_cond_wait (&s->count_nonzero_, &s->lock_); // <s->lock_> is now held. // Decrement the waiters count. s->waiters_count_--; // Decrement the semaphore's count. s->count_--; // Release mutex to leave critical section. pthread_mutex_unlock (&s->lock_);}The counting semaphore implementation shown above uses a ``callermanipulates waiter count'' idiom to simplify the synchronizationbetween the sema_wait and sema_postfunctions. This synchronization idiom is used elsewhere in thisarticle. The sema_post function atomically increments thesemaphore count pointed to by s. Only one thread will beunblocked and allowed to continue, regardless of the number of threadsblocked on the semaphore.intsema_post (sema_t *s){ pthread_mutex_lock (&s->lock_); // Always allow one thread to continue if it is waiting. if (s->waiters_count_ > 0) pthread_cond_signal (&s->count_nonzero_); // Increment the semaphore's count. s->count_++; pthread_mutex_unlock (&s->lock_);}To further clarify the behavior of counting semaphores, assume thereare two threads, T1 and T2, both of which arecollaborating via semaphore sema_t s. If T1calls sema_wait when s's semaphore count is0 it will block until thread T2 callssema_post. When T2 subsequently callssema_post, this function detects if the shared datawaiters_count_ is > 0, which signifies that threadT1 is blocked waiting to acquire semaphores. When thread T2 signals the condition variablecount_nonzero_ in sema_post, threadT1 is awakened in pthread_cond_wait. Whenthread T1 is rescheduled and dispatched, it re-evaluatesits condition expression, i.e., s->count_ == 0.If count_ is > 0 the sema_post functionreturns to thread T1. Note that a while loop is used to wait for condition variablecount_nonzero_ in sema_wait. This conditionvariable looping idiom [Lea] is necessary sinceanother thread, e.g., T3, may have acquired thesemaphore first and decremented s->count_ back to 0before thread T1 is awakened. In this example, the whileloop ensures that T1 doesn't return until the semaphore is> 0. In general, the while loop idiom ensures that threads don'treturn from pthread_cond_wait prematurely. 3. Implementing Condition Variables on Win32As shown above, POSIX Pthreads defines condition variables via thepthread_cond_t type and its supporting operations. Incontrast, the Microsoft Win32 interface does not support conditionvariables natively. To workaround this omission, many developerswrite POSIX condition variable emulations using Win32 synchronizationprimitives such as mutexes, critical sections, semaphores, and events,which are outlined in the sidebar. Sidebar: Overview of Win32 Synchronization PrimitivesWin32 contains a range of synchronization primitives. The primitivesused in this article are outlined below. MutexesMutexes are ``binary semaphores'' that serialize thread access tocritical sections. Win32 mutexes are accessed viaHANDLEs, and can serialize threads within one process oracross multiple processes. A single mutex is acquired viaWaitForSingleObject andWaitForMultipleObjects can acquire multiple mutexes. Amutex is released via the ReleaseMutex operation. Critical SectionsCritical sections are ``lightweight'' mutexes that can only serializethreads within a single process. They are more efficient than Win32mutexes, however, and potentially require only a single hardwareinstruction on some platforms. In contrast, Win32 mutexes require asystem call and and is substantially slower. A critical section isacquired via EnterCriticalSection and released viaLeaveCriticalSection.Although they are slower in some situations, someexperiments show that critical sections provide no performancebenefit at all when running on an SMP system and two threads arerunning on different processors and contending for the same criticalsection. Moreover, Win32 mutexes are also functionally richer thancritical sections. For instance, WaitForMultipleObjectscan be used to wait for multiple waitable objects (mutexes,semaphores, threads, processes, etc). By enabling thewaitAll parameter, all waitable objectHANDLEs passed to WaitForMultipleObjects canbe acquired atomically. The waitAll feature supports``all or nothing'' synchronization semantics that are not availablefrom critical sections.SemaphoresWin32 semaphores are more general synchronization objects thatimplement the counting semaphores mechanism described in Section 2.2.Like mutexes, Win32 semaphores can synchronize threads within oneprocess or across multiple processes. A single semaphore is acquiredvia WaitForSingleObject andWaitForMultipleObjects can acquire multiple semaphores.A semaphore is released via the ReleaseSemaphoreoperation, which increments the semaphore value by a user-specifiedamount, which is typically 1.EventsWin32 events come in two types: (1) auto-reset events and (2)manual-reset events. The semantics of the Win32 event operations,e.g., SetEvent, PulseEvent,ResetEvent, differ depending on what type of eventthey're operating on. A single event of either type can be waitedupon via WaitForSingleObject andWaitForMultipleObjects can wait for multiple events. When a manual-reset event is signaled by SetEvent it setsthe event into the signaled state and wakes up all threads waiting onthis event. In contrast, when an auto-reset event is signaled bySetEvent and there are any threads waiting, it wakes upone thread and reset the event to the non-signaled state. If thereare no threads waiting the event remains signaled until a singlewaiting thread waits on it and is released.When a manual-reset event is signaled by PulseEvent itwakes up all waiting threads and atomically resets the event. Incontrast, if an auto-reset event is signaled byPulseEvent and there are any threads waiting, it wakes upone thread. In either case, the auto-reset event is set to thenon-signaled state. Both Win32 events and POSIX condition variables provide similarwaiting, signaling, and broadcastingfeatures. For instance, WaitForMultipleObjects canacquire a mutex and wait on an event simultaneously via thewaitAll flag and SignalObjectAndWait canrelease a mutex and wait on an event atomically. These functionsprovide semantics akin to the pthread_cond_wait andpthread_cond_signal. Thus, there are instances whereeither events and condition variables can be used interchangably. However, extreme care must be taken with Win32 events to ensure thatthere are no race conditions introduced when switching from onemechanism to another. Unfortunately, there's no way to release justone waiting thread with a manual-reset event. Likewise, there's noway to release all waiting threads with an auto-reset event. Thislimitation is a major source of difficulty when implementing conditionvariables, as shown in Section 3. After years of repeatedly seeing Win32 implementations of conditionvariables posted in newsgroups likecomp.programming.threads it became apparent that manyWin32 implementations are either incorrect or contain subtle problemsthat can lead to starvation, unfairness, or race conditions. To helpdevelopers avoid these problems, this article evaluates commonstrategies for implementing POSIX condition variables on Win32,illustrating common traps and pitfalls and ways to avoid them. All these strategies described in this paper are designed for threadsin the same process. Note that we use a C-style of C++ programming tofocus our emphasis on the steps in each solution. 3.1. The PulseEvent SolutionThe first solution we'll examine starts by defining apthread_cond_t data structure that contains a pair ofWin32 events: typedef struct{ enum { SIGNAL = 0, BROADCAST = 1, MAX_EVENTS = 2 }; HANDLE events_[MAX_EVENTS]; // Signal and broadcast event HANDLEs.} pthread_cond_t;In this solution, an auto-reset event (events_[SIGNAL])implements pthread_cond_signal and a manual-reset event(events_[BROADCAST]) implementspthread_cond_broadcast. Using these synchronization mechanisms, thepthread_cond_init initialization function is defined asfollows:int pthread_cond_init (pthread_cond_t *cv, const pthread_condattr_t *){ // Create an auto-reset event. cv->events_[SIGNAL] = CreateEvent (NULL, // no security FALSE, // auto-reset event FALSE, // non-signaled initially NULL); // unnamed // Create a manual-reset event. cv->events_[BROADCAST] = CreateEvent (NULL, // no security TRUE, // manual-reset FALSE, // non-signaled initially NULL); // unnamed}The following three functions implement the core operations of POSIXcondition variables (for simplicity, most error handling is omitted inthis and other solutions). We'll start by defining a mutual exclusiontype for the POSIX pthread_mutex_t using a Win32CRITICAL_SECTION, as follows:typedef CRITICAL_SECTION pthread_mutex_t;The pthread_cond_wait function first releases theassociated external_mutex of typepthread_mutex_t, which must be held when the callerchecks the condition expression. The function then usesWaitForMultipleObjects to wait for either theSIGNAL or BROADCAST event to becomesignaled: int pthread_cond_wait (pthread_cond_t *cv, pthread_mutex_t *external_mutex){ // Release the <external_mutex> here and wait for either event // to become signaled, due to <pthread_cond_signal> being // called or <pthread_cond_broadcast> being called. LeaveCriticalSection (external_mutex); WaitForMultipleObjects (2, // Wait on both <events_> ev->events_, FALSE, // Wait for either event to be signaled INFINITE); // Wait "forever" // Reacquire the mutex before returning. EnterCriticalSection (external_mutex, INFINITE);}Note that the pthread_cond_timedwait function can beimplemented by mapping its struct timespec *abstimeparameter into a non-INFINITE final argument toWaitForMultipleObject.The pthread_cond_signal function tries to wakeup one threadwaiting on the condition variable cv, as follows:int pthread_cond_signal (pthread_cond_t *cv){ // Try to release one waiting thread. PulseEvent (cv->events_[SIGNAL]);}The call to PulseEvent on the auto-reset eventcv->events_[SIGNAL] awakens one waiting thread. TheSIGNAL event is atomically reset to the non-signaledstate, even if no threads are waiting on this event. The pthread_cond_broadcast function is similar, though itwakes up all threads waiting on the condition variablecv:int pthread_cond_broadcast (pthread_cond_t *cv){ // Try to release all waiting threads. PulseEvent (cv->events_[BROADCAST]);}Calling PulseEvent on a manual-reset event releases allthreads that are waiting on cv->events_[BROADCAST]. Aswith pthread_cond_signal, the state of the event isatomically reset to non-signaled, even if no threads are waiting. Evaluating the PulseEvent SolutionAlthough the PulseEvent solution is concise and``intuitive,'' it is also incorrect. The root of the problem is knownas the lost wakeup bug. This bug occurs from the lack ofatomicity between (1) releasing the external mutex and (2) the startof the wait, i.e.: LeaveCriticalSection (external_mutex); // Potential "lost wakeup bug" here! WaitForMultipleObjects (2, ev->events_, FALSE, INFINITE);The lost wakeup bug manifests itself when the calling thread ispreempted after releasing the external_mutex,but before calling WaitForMultipleObjects. Inthis case, a different thread could acquire theexternal_mutex and perform apthread_cond_signal in between the two calls. Becauseour signal implementation pulses the auto-reset event, theevent remain unsignaled. Therefore, when the preempted calling threadresumes and calls WaitForMultipleObjects it may find bothevents non-signaled. In this case, it might wait indefinitely sinceits wakeup signal got ``lost.'' Note that this problem could be trivially solved if Win32 providedsome type of ``SignalObjectAndWaitMultiple'' functionthat released a mutex or critical section and atomically waited for anarray of HANDLEs to be signaled. 3.2. The SetEvent SolutionOne way to avoid the lost wakeup bug is to use theSetEvent Win32 functions instead ofPulseEvent. The second solution, shown below, implementsthis approach using a pthread_cond_t data structurethat's similar to the first one. The difference is that it maintainsa count of the threads waiting on the condition variable and aCRITICAL_SECTION to protect this count, as follows:typedef struct{ u_int waiters_count_; // Count of the number of waiters. CRITICAL_SECTION waiters_count_lock_; // Serialize access to <waiters_count_>. // Same as before...} pthread_cond_t;The pthread_cond_init initialization function is definedas follows:int pthread_cond_init (pthread_cond_t *cv, const pthread_condattr_t *){ // Initialize the count to 0. cv->waiters_count_ = 0; // Create an auto-reset and manual-reset event, as before...}The pthread_cond_wait function waits for a conditioncv and atomically releases the associatedexternal_mutex that it holds while checking the conditionexpression: 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_); // It's ok to release the <external_mutex> here since Win32 // manual-reset events maintain state when used with // <SetEvent>. This avoids the "lost wakeup" bug... LeaveCriticalSection (external_mutex); // Wait for either event to become signaled due to <pthread_cond_signal> // being called or <pthread_cond_broadcast> being called. int result = WaitForMultipleObjects (2, ev->events_, FALSE, INFINITE); EnterCriticalSection (&cv->waiters_count_lock_); cv->waiters_count_--; int last_waiter = result == WAIT_OBJECT_0 + BROADCAST && cv->waiters_count_ == 0; LeaveCriticalSection (&cv->waiters_count_lock_); // Some thread called <pthread_cond_broadcast>. if (last_waiter) // We're the last waiter to be notified or to stop waiting, so // reset the manual event. ResetEvent (cv->events_[BROADCAST]); // Reacquire the <external_mutex>. EnterCriticalSection (external_mutex, INFINITE);}The release of the external_mutex and the subsequent waitfor either of the events_ to become signaled is stillnon-atomic. This implementation avoids the lost wakeup bug, however,by relying on the ``stickiness'' of the manual-reset event, the use ofSetEvent rather than PulseEvent, and thewaiters_count_ count. Note how the last waiter thread inpthread_cond_wait resets the broadcastmanual-reset event to non-signaled before exiting the function. Thisevent is signaled in pthread_cond_broadcast, which wakesup all threads waiting on the condition variablecv, as follows:int pthread_cond_broadcast (pthread_cond_t *cv){ // Avoid race conditions. EnterCriticalSection (&cv->waiters_count_lock_); int have_waiters = cv->waiters_count_ > 0; LeaveCriticalSection (&cv->waiters_count_lock_); if (have_waiters) SetEvent (cv->events_[BROADCAST]);}Calling SetEvent on a manual-reset event sets thecv->events_[BROADCAST] event to the signaled state. Thisreleases all threads until the event is manually reset in thepthread_cond_wait function above. The pthread_cond_signal function wakes up a threadwaiting on the condition variable cv:int pthread_cond_signal (pthread_cond_t *cv){ // Avoid race conditions. EnterCriticalSection (&cv->waiters_count_lock_); int have_waiters = cv->waiters_count_ > 0; LeaveCriticalSection (&cv->waiters_count_lock_); if (have_waiters) SetEvent (cv->events_[SIGNAL]);}Calling SetEvent on an auto-reset event will set thecv->events_[SIGNAL] event to the signaled state. Thisreleases a single thread and atomically resets the event to thenon-signaled state. If there are no waiting threads, theSetEvent function is skipped.Evaluating the SetEvent SolutionAlthough the solution above doesn't suffer from the lost wakeup bugexhibited by the PulseEvent implementation, it does havethe following drawbacks: Unfairness -- The semantics of the POSIXpthread_cond_broadcast function is to wake up all threadscurrently blocked in wait calls on the conditionvariable. The awakened threads then compete for theexternal_mutex. To ensure fairness, all ofthese threads should be released from theirpthread_cond_wait calls and allowed to recheck theircondition expressions before other threads can successfully complete await on the condition variable. Unfortunately, the SetEvent implementation above does notguarantee that all threads sleeping on the condition variablewhen cond_broadcast is called will acquire theexternal_mutex and check their condition expressions.Although the Pthreads specification does not mandate this degree offairness, the lack of fairness can cause starvation. To illustrate the unfairness problem, imagine there are 2 threads,C1 and C2, that are blocked inpthread_cond_wait on condition variablenot_empty_ that is guarding a thread-safe message queue.Another thread, P1 then places two messages onto the queueand calls pthread_cond_broadcast. If C1returns from pthread_cond_wait, dequeues and processesthe message, and immediately waits again then it and only itmay end up acquiring both messages. Thus, C2 will neverget a chance to dequeue a message and run. The following illustrates the sequence of events: Thread C1 attempts to dequeue and waits on CV non_empty_ Thread C2 attempts to dequeue and waits on CV non_empty_ Thread P1 enqueues 2 messages and broadcasts to CV not_empty_ Thread P1 exits Thread C1 wakes up from CV not_empty_, dequeues a message and runs Thread C1 waits again on CV not_empty_, immediately dequeues the 2nd message and runs Thread C1 exits Thread C2 is the only thread left and blocks forever since not_empty_ will never be signaledDepending on the algorithm being implemented, this lack of fairnessmay yield concurrent programs that have subtle bugs. Of course,application developers should not rely on the fairness semantics ofpthread_cond_broadcast. Howver, there are many caseswhere fair implementations of condition variables can simplifyapplication code. Incorrectness -- A variation on the unfairness problemdescribed above occurs when a third consumer thread, C3, isallowed to slip through even though it was not waiting on conditionvariable not_empty_ when a broadcast occurred. To illustrate this, we will use the same scenario as above: 2 threads,C1 and C2, are blocked dequeuing messages fromthe message queue. Another thread, P1 then places twomessages onto the queue and calls pthread_cond_broadcast.C1 returns from pthread_cond_wait, dequeuesand processes the message. At this time, C3 acquires theexternal_mutex, calls pthread_cond_wait andwaits on the events in WaitForMultipleObjects. SinceC2 has not had a chance to run yet, theBROADCAST event is still signaled. C3 thenreturns from WaitForMultipleObjects, and dequeues andprocesses the message in the queue. Thus, C2 will never geta chance to dequeue a message and run. The following illustrates the sequence of events: Thread C1 attempts to dequeue and waits on CV non_empty_ Thread C2 attempts to dequeue and waits on CV non_empty_ Thread P1 enqueues 2 messages and broadcasts to CV not_empty_ Thread P1 exits Thread C1 wakes up from CV not_empty_, dequeues a message and runs Thread C1 exits Thread C3 waits on CV not_empty_, immediately dequeues the 2nd message and runs Thread C3 exits Thread C2 is the only thread left and blocks forever since not_empty_ will never be signaledIn the above case, a thread that was not waiting on the conditionvariable when a broadcast occurred was allowed to proceed. This leadsto incorrect semantics for a condition variable. Increased serialization overhead -- The implementationshown above uses the waiters_count_lock_ critical sectionto protect the waiters_count_ from being corrupted byrace conditions. This additional serialization overhead is notnecessary as long as pthread_cond_signal andpthread_cond_broadcast are always called by a thread thathas locked the same external_mutex used bypthread_cond_wait. For instance, application code thatalways uses the following idiom does not require this additionalserialization: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 optimizationdoes 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_signalafter the pthread_mutex_unlock to minimize thenumber of trips through the condition variable scheduler. The problemwith this optimization, of course, is that race conditions can occuron the waiters_count_ if it was tested/modified withoutbeing protected by the external_mutex. Note that thePulseEvents solution did not have this race conditionsince it did not have any directly mutable internal state. In general, race conditions can occur with condition variableimplementations that use internal state if threads don't acquire theexternal_mutex when callingpthread_cond_signal andpthread_cond_broadcast. The POSIX Pthread specificationallows these functions to be called by a thread whether or not thatthread has currently locked the external_mutex.Therefore, although the use of the internal lock can increaseoverhead, it is necessary to provide robust, standard-compliantcondition variable programming model. 3.3. The Generation Count SolutionThere are several ways to solve the fairness and correctness problems.The following example illustrates one strategy for alleviating theseproblems. We start by creating a more sophisticatedpthread_cond_t struct.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;As shown in the pthread_cond_t struct above,this solution uses just one Win32 event. This manual-reset eventblocks threads in pthread_cond_wait and releases one ormore threads in pthread_cond_signal andpthread_cond_broadcast, respectively. To enhancefairness, this scheme uses (1) a wait_generation_count_data member to track real signals/broadcasts and (2) arelease_count_ to control how many threads should benotified for each signal/broadcast.The following function creates and initializes a condition variableusing the generation count implementation: 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}The pthread_cond_wait implementation shown below waitsfor condition cv and atomically releases the associatedexternal_mutex that must be held while checking thecondition expression: intpthread_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_);}This function loops until the event_ HANDLEis signaled and there are still threads from this ``generation'' thathaven't been released from the wait. Thewait_generation_count_ field is incremented every timethe event_ is signal viapthread_cond_broadcast orpthread_cond_signal. It tries to eliminate the fairnessproblems with the SetEvents solution by not responding tosignal or broadcast notifications that have occurred in a previous``generation,'' i.e., before the current group ofthreads started waiting. The following function notifies a single thread waiting on a conditionvariable: intpthread_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_);}Note that we only signal the event if there are more waiters thanthreads in the generation that is in the midst of being released. The following implementation of pthread_cond_broadcastnotifies all threads waiting on a condition variable: intpthread_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_);}Evaluating the Generation Count SolutionThe unfairness problem of the SetEvent solution isnominally avoided by having each thread test whether there wasactually a signal since it waited or if there are only left-overreleases. This prevents one thread from taking more than one releasewithout waiting in a new ``generation.'' However, the following arethe subtle traps and pitfalls with this approach: Busy-waiting -- This solution can result in busy-waitingif a waiter has the highest priority thread. The problem is that oncepthread_cond_broadcast signals the manual-resetevent_ it remains signaled. Therefore, the highestpriority thread may cycle endlessly through the for loopin pthread_cond_wait and never sleep on the event. Unfairness -- The for loop inpthread_cond_wait leaves the critical section beforecalling WaitForSingleObject. Thus, it's possible foranother thread to acquire the external_mutex and callpthread_cond_signal orpthread_cond_broadcast again during this unprotectedregion. If this situation occurs, thewait_generation_count_ will increase, which may cause thewaiting thread to break out of the loop prematurely. In this case,the waiting thread can steal a release that was intended for anotherthread. Therefore, the Generation Count solution isn't entirely fairafter all! Serialization overhead -- Because there is additionalstate in the pthread_cond_t implementation, the internalserialization logic is more complex. 3.4. The SignalObjectAndWait SolutionA more complete, albeit more complex, way to achieve fairness is touse a broader set of Win32 synchronization mechanisms. The solutionshown below leverages the Windows NT 4.0SignalObjectAndWait function, in conjunction with thefollowing three synchronization mechanisms: A Win32 semaphore (sema_) -- which is used toqueue up threads waiting for the condition to become signaled. A Win32 auto-reset event (waiters_done_) --which is used by the broadcast or signaling thread to wait for all thewaiting thread(s) to wake up and be released from the semaphore. A Win32 Critical Section (waiters_count_lock_) --which serializes access to the count of waiting threads. This approach yields the following struct to implementPOSIX condition variables on Win32: 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;For variety's sake, this solution optimizes the serialization requiredby assuming that the external_mutex is held whenpthread_cond_signal orpthread_cond_broadcast is called. If this optimizationis not possible, an extra mutex can be added topthread_cond_t and held for the entire duration of thepthread_cond_signal, pthread_cond_broadcast,pthread_cond_wait, andpthread_cond_timedwait calls. The remainder of this section illustrates how to implement conditionvariables on Windows NT 4.0 using the SignalObjectAndWaitfunction. This function atomically signals one HANDLEand waits for another HANDLE to become signaled. Sincemutexes are accessed via HANDLEs, whereasCRITICAL_SECTIONS are not, we'll need to typedef a Win32mutex for the POSIX pthread_mutex_t instead of aCRITICAL_SECTION:typedef HANDLE pthread_mutex_t;The pthread_cond_init function is defined as follows: 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}The pthread_cond_wait function uses two steps: It waits for the sema_ semaphore to be signaled,which indicates that a pthread_cond_broadcast orpthread_cond_signal has occurred. It sets the waiters_done_ auto-reset event into thesignaled state when the last waiting thread is about to leave thepthread_cond_wait critical section. Step 2 collaborates with the pthread_cond_broadcastfunction described below to ensure fairness. But first, here's theimplementation of pthread_cond_wait: intpthread_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);}This implementation of pthread_cond_wait ensures that theexternal_mutex is held until all threads waiting on thecv have a chance to wait again on theexternal_mutex before returning to their callers. Thissolution relies on the fact that Windows NT mutex requests are queuedin FIFO order, rather than in, e.g., priority order[MSMutex]. Because the external_mutexqueue is serviced in FIFO order, all waiting threads will acquire theexternal mutex before any of them can reacquire it a second time.This property is essential to ensure fairness. We use the SignalObjectAndWait function to ensureatomicity when signaling one synchronization object(external_mutex) and waiting for another synchronizationobject to become signaled (cv->sema_). This isparticularly important for the last waiter thread, i.e., theone that signals the broadcasting thread when all waiters are done.If we don't leverage the atomicity ofSignalObjectAndWait, our solution is potentially unfairsince the last waiter thread may not get the chance to wait on theexternal_mutex before one of the other waiters gets it,performs some work, releases and then immediately reacquiresexternal_mutex. The pthread_cond_signal function is straightforward sinceit just releases one waiting thread. Therefore, it simply incrementsthe sema_ semaphore by 1:intpthread_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);}The pthread_cond_broadcast function is more complex andrequires two steps: It wakes up all the threads waiting on the sema_semaphore, which can be done atomically by passing thewaiters_count_ to ReleaseSemaphore; It then blocks on the auto-reset waiters_done_ eventuntil the last thread in the group of waiting threads exits thepthread_cond_wait critical section.Here's the code: intpthread_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_);}As mentioned above, this solution requires that thepthread_cond_signal andpthread_cond_broadcast are only called by a thread that'slocked the same external_mutex. Evaluating the SignalObjectAndWait SolutionOur implementation of pthread_cond_wait relied on theSignalObjectAndWait function, which was first releasedwith Windows NT 4.0. Although this solution provides fairness, it hasmore overhead than the earlier solutions. For instance, theexternal_mutex is a mutex rather than a critical section,which increases locking and unlocking overhead. Moreover, thepthread_cond_wait and pthread_cond_broadcastimplementations require additional synchronization operations toensure fairness when broadcasting. Unfortunately, the SignalObjectAndWait function is notavailable in Windows CE, Windows '95, or Windows NT 3.51. Therefore,emulating condition variables can be trickier, less ``fair,'' and morecomputationally expensive on these Win32 platforms. The next articlein this series will show how to develop a portable condition variableimplementation for platforms that lackSignalObjectAndWait. |
|
