Patterns for POSIX Condition Variables on Win32

1. Introduction

2. The Wrapper Facade Pattern

Intent

Context

Problem

• Developers must have intimate knowledge of many OS platforms -- Implementing concurrent applications using system-level APIs forces developers to deal with the non-portable, tedious, and error-prone OS idiosyncrasies such as Pthreads. Moreover, these APIs are not portable across OS platforms. For example, Win32 lacks Pthreads.

• Increased maintenance effort -- Many application handle portability variations of multi-threading via explicit conditional compilation directives in application source code. Using conditional compilation to address platform-specific variations at all points of use increases the complexity of the source code. It is hard to maintain and extend such applications since platform-specific details are scattered throughout the implementation.

• Inconsistent programming paradigms -- System mechanisms are accessed through C-style function calls, which cause an ``impedance mismatch'' with the OO programming style supported by C++, the language used to implement applications.

Solution

3. Applying the Wrapper Facade Pattern to Condition Variables and Mutexes

class Mutex
{
public:
  int acquire (void);
  int release (void);
  // ...

private:
  pthread_mutex_t mutex_;
};

class Condition
{
public:
  Condition (Mutex &m);
  ~Condition (void);
  int wait (Mutex &m);
  int wait (void);
  int signal (void);
  int broadcast (void);

private:
  Mutex &mutex_;
  pthread_cond_t cond_;
};

4. Win32 Condition Variable Implementations Revised

4.1. Modifying the SignalObjectAndWait Solution for Windows '95 and Windows NT 3.51

typedef CRITICAL_SECTION pthread_mutex_t;

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

  // The following two function calls are used in place of
  // <SignalObjectAndWait> for versions of Win32 that lack this
  // function.

  // Keep the lock held just long enough to increment the count of
  // waiters by one.  We can't keep it held across the call to
  // <WaitForSingleObject> since that will deadlock other calls to
  // <pthread_cond_signal> and <pthread_cond_broadcast>.
  ExitCriticalSection (external_mutex);  

  // Wait to be awakened by a <pthread_cond_signal> or
  // <pthread_cond_broadcast>. 
  WaitForSingleObject (cv->sema_, INFINITE);

  // 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 a
  // <pthread_cond_broadcast> call. 
  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)
    SetEvent (cv->waiters_done_);

  // Always regain the external mutex since that's the guarantee that
  // we give to our callers.
  EnterCriticalSection (external_mutex);
}

Evaluating the Modified Solution

4.2. The Specific Notification Solution

This approach uses the Specific Notification pattern [Cargill].

typedef struct
{
  std::deque waiters_count_;
  // FIFO queue of threads waiting on the condition variable.

  CRITICAL_SECTION waiters_count_lock_;
  // Protect the <waiters_count_> state.
} pthread_cond_t;

int
pthread_cond_init (pthread_cond_t *cv, 
                   const pthread_condattr_t *)
{
  broadcast_ = CreateSemaphore (NULL,  // no security
                                0,     // initially 0
                                0x7fffffff, // max
                                NULL); // unnamed
  mutex_ = CreateMutex (NULL,  // no security
                        FALSE, // unsignaled initially
                        NULL);  // unnamed
}

int
pthread_cond_destroy (pthread_cond_t *cv)
{
}

int 
pthread_cond_wait (pthread_cond_t *cv,
                   pthread_mutex_t *external_mutex)
{
}

int 
pthread_cond_signal (pthread_cond_t *cv)
{
}

int 
pthread_cond_broadcast (pthread_cond_t *cv)

{
}

Evaluating the Specific Notification Solution

5. Implementing Events on non-Win32 Platforms

• event_wait (event_t *event) -- This function waits for event to be signaled. This function emulates a call to WaitForSingleObject where the event is the waitable object and a timeout of INFINITE is specified.

• event_timedwait (event_t *event, const timespec *timeout) -- This function is the same as event_wait, except that the caller waits upto timeout in WaitForSingleObject for the event to become signaled.

• event_signal (event_t *event) -- If the event is a manual-reset event, it remains signaled until it is set explicitly to the nonsignaled state by the event_reset function. Any number of waiting threads, or threads that subsequently begin wait operations for the specified event object by calling one of the wait functions, can be released while the event's state is signaled.

  If the event is an auto-reset event, it remains signaled until a single waiting thread is released, at which time the system automatically sets the state to nonsignaled. If no threads are waiting, the event's state remains signaled.

• event_pulse (event_t *event) -- If event is a manual-reset event, all waiting threads that can be released immediately are released. The function then resets the event's state to nonsignaled and returns.

  If event is an auto-reset event, the function resets the state to nonsignaled and returns after releasing a single waiting thread, even if multiple threads are waiting.

  If no threads are waiting, or if no thread can be released immediately, event_pulse simply sets the event's state to nonsignaled and returns.

• event_reset (event_t *event) -- event_reset function is used primarily for manual-reset events, which must be set explicitly to the nonsignaled state.

Implementation

struct event_t
{
  mutex_t lock_;
  // Protect critical section.

  cond_t condition_;
  // Keeps track of waiters.

  int manual_reset_;
  // Specifies if this is an auto- or manual-reset event.

  int is_signaled_;
  // "True" if signaled.

  u_long waiting_threads_;
  // Number of waiting threads.
};

void
event_init (event_t *event,
            int manual_reset,
            int initial_state,
            int type,
            char *name,
            void *arg)
{
  event->manual_reset_ = manual_reset;
  event->is_signaled_ = initial_state;
  event->waiting_threads_ = 0;
  
  cond_init (&event->condition_,
             type,
             name,
             arg);
  
  mutex_init (&event->lock_,
              type,
              name,
              arg);
}

void
event_destroy (event_t *event)
{
  mutex_destroy (&event->lock_);
  cond_destroy (&event->condition_);
}

void
event_wait (event_t *event)
{
  // grab the lock first
  mutex_lock (&event->lock_);

  if (event->is_signaled_ == 1)
    // Event is currently signaled.
    {
      if (event->manual_reset_ == 0)
        // AUTO: reset state
        event->is_signaled_ = 0;
    }
  else
    // event is currently not signaled
    {
      event->waiting_threads_++;

      cond_wait (&event->condition_,
                 &event->lock_);

      event->waiting_threads_--;
    }

  // Now we can let go of the lock.
  mutex_unlock (&event->lock_);
}

void
event_signal (event_t *event)
{
  // grab the lock first
  mutex_lock (&event->lock_);

  // Manual-reset event.
  if (event->manual_reset_ == 1)
    {
      // signal event
      event->is_signaled_ = 1;

      // wakeup all
      cond_broadcast (&event->condition_);
    }
  // Auto-reset event
  else
    {
      if (event->waiting_threads_ == 0)
        // No waiters: signal event.
        event->is_signaled_ = 1;
      else 
        // Waiters: wakeup one waiter.
        cond_signal (&event->condition_);
    }
  
  // Now we can let go of the lock.
  mutex_unlock (&event->lock_);
}

void
event_pulse (event_t *event)
{
  // grab the lock first
  mutex_lock (&event->lock_);

  // Manual-reset event.
  if (event->manual_reset_ == 1)
    {
      // Wakeup all waiters.
      cond_broadcast (&event->condition_);
    }

  // Auto-reset event: wakeup one waiter.
  else 
    {
      cond_signal (&event->condition_);

      // Reset event.
      event->is_signaled_ = 0;
    }

  // Now we can let go of the lock.
  mutex_unlock (&event->lock_);
}

void
event_reset (event_t *event)
{
  // Grab the lock first.
  mutex_lock (&event->lock_);

  // Reset event.
  event->is_signaled_ = 0;
      
  // Now we can let go of the lock.
  mutex_unlock (&event->lock_);
}

6. Concluding Remarks

Acknowledgements

References

[GoF] Erich Gamma, Richard Helm, Ralph Johnson, and John Vlissides, Design Patterns: Elements of Reusable Software Architecture, Addison-Wesley, 1995.

[Pthreads] ``Threads Extension for Portable Operating Systems,'' IEEE P1003.4a, February, 1996.

[Richter] Jeffrey Richter, Advanced Windows, 1997, Microsoft Press, Redmond, WA.