Implement pthread_cond_broadcast on Windows

[git.git] / compat / win32 / pthread.c

diff --git a/compat/win32/pthread.c b/compat/win32/pthread.c
index 631c0a46ea3ddc4daaa773128b4a3e7a001653de..5fc1670bee94880b717df2e513a4d15678383ee0 100644 (file)
--- a/compat/win32/pthread.c
+++ b/compat/win32/pthread.c
@@ -52,24 +52,38 @@ int win32_pthread_join(pthread_t *thread, void **value_ptr)
 int pthread_cond_init(pthread_cond_t *cond, const void *unused)
 {
        cond->waiters = 0;
+       cond->was_broadcast = 0;
+       InitializeCriticalSection(&cond->waiters_lock);
 
        cond->sema = CreateSemaphore(NULL, 0, LONG_MAX, NULL);
        if (!cond->sema)
                die("CreateSemaphore() failed");
+
+       cond->continue_broadcast = CreateEvent(NULL,    /* security */
+                               FALSE,                  /* auto-reset */
+                               FALSE,                  /* not signaled */
+                               NULL);                  /* name */
+       if (!cond->continue_broadcast)
+               die("CreateEvent() failed");
+
        return 0;
 }
 
 int pthread_cond_destroy(pthread_cond_t *cond)
 {
        CloseHandle(cond->sema);
-       cond->sema = NULL;
-
+       CloseHandle(cond->continue_broadcast);
+       DeleteCriticalSection(&cond->waiters_lock);
        return 0;
 }
 
 int pthread_cond_wait(pthread_cond_t *cond, CRITICAL_SECTION *mutex)
 {
-       InterlockedIncrement(&cond->waiters);
+       int last_waiter;
+
+       EnterCriticalSection(&cond->waiters_lock);
+       cond->waiters++;
+       LeaveCriticalSection(&cond->waiters_lock);
 
        /*
         * Unlock external mutex and wait for signal.
@@ -82,22 +96,52 @@ int pthread_cond_wait(pthread_cond_t *cond, CRITICAL_SECTION *mutex)
        /* let's wait - ignore return value */
        WaitForSingleObject(cond->sema, INFINITE);
 
-       /* we're done waiting, so make sure we decrease waiters count */
-       InterlockedDecrement(&cond->waiters);
-
+       /*
+        * Decrease waiters count. If we are the last waiter, then we must
+        * notify the broadcasting thread that it can continue.
+        * But if we continued due to cond_signal, we do not have to do that
+        * because the signaling thread knows that only one waiter continued.
+        */
+       EnterCriticalSection(&cond->waiters_lock);
+       cond->waiters--;
+       last_waiter = cond->was_broadcast && cond->waiters == 0;
+       LeaveCriticalSection(&cond->waiters_lock);
+
+       if (last_waiter) {
+               /*
+                * cond_broadcast was issued while mutex was held. This means
+                * that all other waiters have continued, but are contending
+                * for the mutex at the end of this function because the
+                * broadcasting thread did not leave cond_broadcast, yet.
+                * (This is so that it can be sure that each waiter has
+                * consumed exactly one slice of the semaphor.)
+                * The last waiter must tell the broadcasting thread that it
+                * can go on.
+                */
+               SetEvent(cond->continue_broadcast);
+               /*
+                * Now we go on to contend with all other waiters for
+                * the mutex. Auf in den Kampf!
+                */
+       }
        /* lock external mutex again */
        EnterCriticalSection(mutex);
 
        return 0;
 }
 
+/*
+ * IMPORTANT: This implementation requires that pthread_cond_signal
+ * is called while the mutex is held that is used in the corresponding
+ * pthread_cond_wait calls!
+ */
 int pthread_cond_signal(pthread_cond_t *cond)
 {
-       /*
-        * Access to waiters count is atomic; see "Interlocked Variable Access"
-        * http://msdn.microsoft.com/en-us/library/ms684122(VS.85).aspx
-        */
-       int have_waiters = cond->waiters > 0;
+       int have_waiters;
+
+       EnterCriticalSection(&cond->waiters_lock);
+       have_waiters = cond->waiters > 0;
+       LeaveCriticalSection(&cond->waiters_lock);
 
        /*
         * Signal only when there are waiters
@@ -108,3 +152,37 @@ int pthread_cond_signal(pthread_cond_t *cond)
        else
                return 0;
 }
+
+/*
+ * DOUBLY IMPORTANT: This implementation requires that pthread_cond_broadcast
+ * is called while the mutex is held that is used in the corresponding
+ * pthread_cond_wait calls!
+ */
+int pthread_cond_broadcast(pthread_cond_t *cond)
+{
+       EnterCriticalSection(&cond->waiters_lock);
+
+       if ((cond->was_broadcast = cond->waiters > 0)) {
+               /* wake up all waiters */
+               ReleaseSemaphore(cond->sema, cond->waiters, NULL);
+               LeaveCriticalSection(&cond->waiters_lock);
+               /*
+                * At this point all waiters continue. Each one takes its
+                * slice of the semaphor. Now it's our turn to wait: Since
+                * the external mutex is held, no thread can leave cond_wait,
+                * yet. For this reason, we can be sure that no thread gets
+                * a chance to eat *more* than one slice. OTOH, it means
+                * that the last waiter must send us a wake-up.
+                */
+               WaitForSingleObject(cond->continue_broadcast, INFINITE);
+               /*
+                * Since the external mutex is held, no thread can enter
+                * cond_wait, and, hence, it is safe to reset this flag
+                * without cond->waiters_lock held.
+                */
+               cond->was_broadcast = 0;
+       } else {
+               LeaveCriticalSection(&cond->waiters_lock);
+       }
+       return 0;
+}