Multiple Wait Semaphore

Sept. 8, 2000: Changed to use the broadcase condition variable model. (See Chapter 10 in the Second Edition.)

Page 235 in the book mentioned that you could atomically wait for multiple semaphore units by using a mutex to guard the semaphore. This is correct, but limited. I call this the "First Come, First Served" multiple wait semantics as a large request will block all subsequent requests (regardless of thread priority or request size) until the initial request is satisfied. This might be the semantics you require, but, you might actually require "First Satisfiable" semantics where a small request will be satisfied before a large request.

This small, integrated collection of functions implements the required routines. Here are some features:

1. The CreateSemaphoreMW function has a flag to specify "First Satisfiable" or "First Come, First Served" semantics.
2. You can also name the MW semaphore so that it can be shared between processes.
3. The WaitForSingleObjectMW function has a parameter to specify the number of required units.
4. The CloseSynchHandle function plays the same role as CloseHandle. It is intended to be general purpose so that new objects can be implemented over time.

How would you use such a capability? The First Edition Chapter 11's server showed one example, where different threads require different amounts of a limited serving capability. Here is another example, which is fairly realistic. Chapter 10 in the Second Edition takes the idea of a semaphore a step further by implementing a message queueing system.

Suppose you have a "Monitor" thread that is not to start (or get past a certain point) until ALL worker threads are running or have reached some specified point. Here is what to do, using a "First Satisfiable" multiple wait semaphore

1. Initialize the semaphore to its maximum count (at creation time), where the maximum count is the number of worker threads.
2. Each worker thread releases one unit when it starts.
3. The monitor thread waits atomically for the maximum count. Therefore, it will not start until all threads are running, preventing, perhaps, a race condition.

Here is an exercise involving the above idea and an extra event together with a "First Come, First Served" semaphore. Construct a "thread rendezvous" where all worker threads block at a certain point and they are not released until the monitor finds that some number of them have blocked at the rendezvous. Can you construct a rendezvous object that does not require the monitor thread?

Here is another exercise. Construct a "reverse semaphore" object that is signaled if and only if the count is equal to the maximum value. A waiting thread that is released decreases the semaphore by the count specified in the wait and the reverse semaphore becomes unsignaled. You can do this with some minor modifications to the functions here. See http://www.microsoft.com/msj/0797/win320797top.htm for a different solution, although, in that article, the semaphore is signaled only when it is 0, and there is no multiple wait.

The implementation has some other interesting features:

1. A synchronization handle structure is defined in the header file. It requires an event and two mutexes. You will see why all three are required. Notice that this means that it is possible to implement a semaphore with the other two objects.
2. The WaitForSingleObjectMW has at least one none-obvious step; take a look at the code. Also, this function is efficient; there is no busy waiting.. Again, take a look at the code to see how this is achieved. This solution is an example of the "condition variable model" described in Chapter 10 (Second Edition).
3. The choice of pulsing a manual reset event is essential; again, take a look at the comments.
4. Interprocess synchronization requires that there be synchronization handle structures in shared memory to maintain the object state. However, there must also be a "shadow" structure in the process-private memory containing handles to the mutexes and event. Notice how this works and how shared memory objects are released for reuse after the last open handle is released.
5. The test program shows a multiple producer/consumer model. When you bring it up, you are prompted for the number of producer threads (in this test process) and the number of consumer threads, as well as the object name. To test the interprocess synchronization, bring up two separate processes running the same test program. Use the same name for each, and have only producer threads in one (zero consumer threads) and only consumer threads in the other.

COMMENTS ON TESTING, DEBUGGING, AND VALIDATING

Threaded, synchronized systems present significant challenges not only in the design and implementation, but also in all stages of quality assurance. There are many traps for the unwary. However, writing good code need not be that difficult, and it can be fun and rewarding. Here are some hints as well as some personal opinions. Some of the personal opinions are not widely shared.

1. Testing is necessary, but NOT sufficient, to assure correct operation. This can be said of any program, but it is particularly so when synchronization issues are involved. There can be any number of subtle race conditions, deadlocks, and so on that may never appear in testing, and, if they do, they may not be repeatable or easy to diagnose.
2. The same can be said of debugging. I never used the debugger on these examples despite the fact that Visual C++ has excellent debugging capabilities. Debugging also changes the timing of threads and can mask the defects you are trying to find. Finally, the VC++ debugger actually changes the behavior of functions such as PulseEvent; Microsoft has a discussion of this at http://support.microsoft.com/support/kb/articles/Q173/2/60.ASP.
3. It is absolutely necessary to design and implement the system correctly.
4. Furthermore, you must carefully read and analyze your synchronization code and consider all the possibilities. For example, after every wait or release call, ask what would happen if the thread were preempted at that instant and another thread were to execute the same code. Furthermore, ask if all the conditions that you expect to hold at this point can be assured to hold. For example, consider the potential race condition in Chapter 10's sortMT program. My initial implementations of the program here and of sortMT program had defects that never showed up in tests but became apparent when I went through these thought experiments.
5. Ultimately, you must prove to yourself (and, perhaps, to your colleagues) that the code is correct. The proof may not be formal (although formal proofs can play a role), but it is worthwhile going through the motions.
6. In my personal experience, even the best programmers engage in extreme forms of arm waving when discussing synchronization, and the buzz words fly furiously. Try not to engage in such activity, and be suspicious of anyone who does.
7. Haste is the biggest enemy. Take the time to assure yourself that the code is correct.

I have seen some on-line discussion that emphasizes this point. Under the debugger, PulseEvent had a different effect than without the debugger. Apparently, when pulsing a manual reset event, the signal could be lost under certain circumstances when using the debugger.

I plan to add some material regarding informal proofs of correctness for dealing with synchronized threads.

Now that I have said that, I fully expect to be humbled when someone points out an obvious (or not so obvious) defect in my code. Please take the challenge to do so. If you succeed, I will still be grateful and will make the excuse that I was in a hurry (see the last bullet) even though I did perform all the recommended steps. Jan 5, 2000: I just found a bug after 18 months using this code. It is now fixed in the code below.

This is the function library.

___________________________________________________________________________________________________________

/* SynchObj.c

Copyright (c) 1998, Johnson M. Hart
Permission is granted for any and all use providing that this copyright is
properly acknowledged.
There are no assurances of suitability for any use whatsoever.

Library of synchronization "objects" to supplement the standard Windows
events and mutexes.

For simplicity, this implementation shares a header file (synchize.h) with
another library of synchronization objects (EvMxSem.c) that is designed
to operate under Windows CE, which does not have semaphores.

Implementation notes:

1. All required internal data structures are allocated on the process's heap
2. Where appropriate, a new error code is returned (see the header
   file), or, if the error is a Windows error, that code is unchanged.
3. Notice the new handle type, "SYNCHHANDLE" which has handles, counters,
   and other information. This structure will grow as new objects are added
   to this set; some members are specific to only one or two of the objects;
   in particular, the structure is more general than is required just for
   some of the specific examples.
4. Mutexes are used for critical sections. These could be replaced with
   CRITICAL_SECTION objects if the objects are used totally within
   a process. Even better, if the object is not named, then you
   know that it will be used only within the process and you could
   make the decision at create time. HOWEVER, you will lose the timeout
   option if you do so, AND you will also lose the naming and interprocess
   capabilities.With Windows NT4.0 and later, SignalObjectAndWait is a possibility.
5. These simulated semaphores can be named and hence can be shared
   between processes.
6. No Windows semaphores are used, so this implementation (with null names)
   will operate under Windows CE.
7. The implementation shows several interesting aspect of synchronization, some
   of which are specific to Windows and some of which are general. There are pointed
   out in the comments as appropriate.
8. These objects have a WaitForSingleObject equivalent. There is, however, no
   equivalent to WaitForMultipleObjects as this is very difficult, if not impossible
   to do efficiently outside of the kernel.

*/

#include "EvryThng.h"
#include "Synchize.h"

static SYNCHHANDLE CleanUp (SYNCHHANDLE);
static SYNCHHANDLE AllocSynchHandle (LPCTSTR, LPTSTR, LPTSTR, LPTSTR, LPBOOL);
/*********************************************
MULTIPLE WAIT SEMAPHORE function family.
These semaphores allow a thread to wait atomically for several "units"
rather than just one. The DEFAULT wait semantics are considered to be
"First Satisfiable". That is, a thread with a satisfiable request
will be released even if a thread (regardless of priority) has an
outstanding request that is not currently satisfiable.

Optional semantics, specified at create time, are
"First Come, First Served". A thread with request larger than the
number of units currently available will block all subsequent
threads, even those with satisfiable requests or those of
higher priority. The solution on p. 235 is a First Come, First Served
solution, but here we implement it within the general framework.
*/

SYNCHHANDLE CreateSemaphoreMW (

LPSECURITY_ATTRIBUTES lpSemaphoreAttributes, /* pointer to security attributes */
LONG lInitialCount, /* initial count */
LONG lMaximumCount, /* maximum count */
BOOL fFirstComeFirstServed, /* First Satisfiable is the default */
LPCTSTR lpName )

/* Multiple wait semaphore creation.
Requires a counter, a mutex to protect the semaphore state, and a
manual-reset event.

Here are the rules that must always hold between the manual-reset event
and the mutex (any violation of these rules by the multiple wait semaphore
functions will, in all likelihood, result in a defect):

1. No thread can set, pulse, or reset the event,
nor can it access any part of the SYNCHHANDLE structure,
without first gaining ownership of the mutex.
BUT, a thread can wait on the event without owning the mutex
(this is clearly necessary or else the event could never be set).
2. The event is in a signaled state if and only if the count has just been
changed so that it is greater than zero. To assure this property, the count should
be checked after every semaphore decrease.
3. The semaphore count is always >= 0 and <= the maximum count
*/
{
SYNCHHANDLE hSynch = NULL, hShare = NULL;
TCHAR MutexName [MAX_PATH] = _T(""), EventName [MAX_PATH] = _T(""),
MutexaName[MAX_PATH] = _T("");
BOOL NewObject;

if (lInitialCount > lMaximumCount || lMaximumCount < 0 || lInitialCount < 0) {
/* Bad parameters */
SetLastError (SYNCH_ERROR);
return NULL;
}

hSynch = AllocSynchHandle (lpName, MutexName, EventName, MutexaName, &NewObject);
if (hSynch == NULL) {
SetLastError (SYNCH_ERROR);
return NULL;
}

/* Create the object handles. These are always created in the process's
local handle. */

hSynch->hMutex = CreateMutex (lpSemaphoreAttributes, FALSE, (LPCTSTR)MutexName);

/* Create the event. It is initially signaled if and only if the
initial count is > 0 */
hSynch->hEvent = CreateEvent (lpSemaphoreAttributes, TRUE /* manual reset */,
lInitialCount > 0, (LPCTSTR)EventName);

hSynch->hMutexa = NULL;
hSynch->dwFlags = 6; /* An event and a mutex, but no secondary mutex
unless it is a First Come, First Served Multiple Wait semaphore */
if (fFirstComeFirstServed) {
hSynch->hMutexa = CreateMutex (lpSemaphoreAttributes, FALSE, (LPCTSTR)MutexaName);
hSynch->dwFlags = 7; /* All three objects were created */
}

/* Set the object state, always in the local handle (for quick reference)
and in the shared handle if there is one. */

hSynch->MaxCount = lMaximumCount;
hSynch->CurCount = lInitialCount; /* The local value is not maintained. */
hSynch->fFirstComeFirstServed = fFirstComeFirstServed;
_tcscpy (hSynch->lpName, lpName);

hShare = hSynch->SharedHandle;
if (NewObject && hShare != NULL ) {
/* There is a new shared handle. Set the state if it is new */
hShare->MaxCount = lMaximumCount;
hShare->CurCount = lInitialCount; /* The local value is not maintained. */
hShare->fFirstComeFirstServed = fFirstComeFirstServed;
_tcscpy (hShare->lpName, lpName);
}

/* Return with the handle, or, if there was any error, return
a null after closing any open handles and freeing any allocated memory */

return CleanUp (hSynch);
}

BOOL ReleaseSemaphoreMW (SYNCHHANDLE hSemMW, LONG cReleaseCount, LPLONG lpPreviousCount)
/* Multiple wait equivalent to ReleaseSemaphore */
{
BOOL Result = TRUE;
SYNCHHANDLE hState;

/* Gain access to the object to assure that the release count
would not cause the total count to exceed the maximum */