1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
/**
* pthread_cond API for Win32
*
* ACE(TM), TAO(TM), CIAO(TM), DAnCE>(TM), and CoSMIC(TM) (henceforth
* referred to as "DOC software") are copyrighted by Douglas C. Schmidt
* and his research group at Washington University, University of California,
* Irvine, and Vanderbilt University, Copyright (c) 1993-2009, all rights
* reserved.
*
* Since DOC software is open-source, freely available software, you are free
* to use, modify, copy, and distribute--perpetually and irrevocably--the DOC
* software source code and object code produced from the source, as well as
* copy and distribute modified versions of this software. You must, however,
* include this copyright statement along with any code built using DOC
* software that you release.
*
* No copyright statement needs to be provided if you just ship binary
* executables of your software products.
*
* See "Strategies for Implementing POSIX Condition Variables on Win32" at
* http://www.cs.wustl.edu/~schmidt/win32-cv-1.html
*/
#include <windows.h>
#include "win32cond.h"
int win32_cond_init(win32_cond_t *cv)
{
cv->waiters_count_ = 0;
cv->was_broadcast_ = 0;
cv->sema_ = CreateSemaphore (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
return 0;
}
int win32_cond_destroy(win32_cond_t *cv)
{
CloseHandle(cv->waiters_done_);
DeleteCriticalSection(&cv->waiters_count_lock_);
CloseHandle(cv->sema_);
return 0;
}
int win32_cond_wait(win32_cond_t *cv, HANDLE *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, INFINITE);
return 0;
}
int win32_cond_signal(win32_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);
return 0;
}
int win32_cond_broadcast(win32_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_);
return 0;
}
|
