![]() |
A Windows CE SemaphoreThis sample is of general interest for anyone interested in threads and synchronization. There are three source files listed below:
/* SemphWCE.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. WINDOWS CE: There is a collection of Windows CE functions to simulate semaphores using only a mutex and an event. As Windows CE events cannot be named, these simulated semaphores cannot be named either. 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 Win32 error, that code is unchanged. 3. Notice the new handle type "SYNCHHANDLE" that 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. 4. Mutexes are used for critical sections. These could be replaced with CRITICAL_SECTION objects but then this would give up the time out capability. 5. The implementation shows several interesting aspects of synchronization, some of which are specific to Win32 and some of which are general. These are pointed out in the comments as appropriate. 6. The wait function emulates WaitForSingleObject only. An emulation of WaitForMultipleObjects is much harder to implement outside the kernel, and it is not clear how to handle a mixture of WCE semaphores and normal events and mutexes. */ #include "EvryThng.h" /* This brings in Windows.h, stdio.h, etc. */ #include "Synchize.h" static SYNCHHANDLE CleanUp (SYNCHHANDLE hSynch, DWORD Flags); SYNCHHANDLE CreateSemaphoreCE ( LPSECURITY_ATTRIBUTES lpSemaphoreAttributes, /* pointer to security attributes */ LONG lInitialCount, /* initial count */ LONG lMaximumCount, /* maximum count */ LPCTSTR lpName ) /* Semaphore for use with Windows CE that does not support them directly. Requires a counter, a mutex to protect the counter, and an autoreset event. Here are the rules that must always hold between the autoreset event and the mutex (any violation of these rules by the CE 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 current semaphore count ("CurCount") is greater than zero. 3. The semaphore count is always >= 0 and <= the maximum count */ { SYNCHHANDLE hSynch = NULL; _try { if (lInitialCount > lMaximumCount || lMaximumCount < 0 || lInitialCount < 0) { /* Bad parameters */ SetLastError (SYNCH_ERROR); _leave; } hSynch = HeapAlloc (GetProcessHeap(), HEAP_ZERO_MEMORY, SYNCH_HANDLE_SIZE); if (hSynch == NULL) _leave; hSynch->MaxCount = lMaximumCount; hSynch->CurCount = lInitialCount; hSynch->lpName = lpName; hSynch->hMutex = CreateMutex (lpSemaphoreAttributes, FALSE, NULL); WaitForSingleObject (hSynch->hMutex, INFINITE); /* Create the event. It is initially signaled if and only if the initial count is > 0 */ hSynch->hEvent = CreateEvent (lpSemaphoreAttributes, FALSE, lInitialCount > 0, NULL); ReleaseMutex (hSynch->hMutex); hSynch->hSemph = NULL; } _finally { /* 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, 6 /* An event and a mutex, but no semaphore. */)); } } BOOL ReleaseSemaphoreCE (SYNCHHANDLE hSemCE, LONG cReleaseCount, LPLONG lpPreviousCount) /* Windows CE equivalent to ReleaseSemaphore. */ { BOOL Result = TRUE; /* Gain access to the object to assure that the release count would not cause the total count to exceed the maximum. */ _try { WaitForSingleObject (hSemCE->hMutex, INFINITE); *lpPreviousCount = hSemCE->CurCount; if (hSemCE->CurCount + cReleaseCount > hSemCE->MaxCount || cReleaseCount <= 0) { SetLastError (SYNCH_ERROR); Result = FALSE; _leave; } hSemCE->CurCount += cReleaseCount; /* Set the autoreset event, releasing exactly one waiting thread, now or in the future. */ SetEvent (hSemCE->hEvent); } _finally { ReleaseMutex (hSemCE->hMutex); return Result; } } DWORD WaitForSemaphoreCE (SYNCHHANDLE hSemCE, DWORD dwMilliseconds) /* Windows CE semaphore equivalent of WaitForSingleObject. */ { DWORD WaitResult; WaitResult = WaitForSingleObject (hSemCE->hMutex, dwMilliseconds); if (WaitResult != WAIT_OBJECT_0 && WaitResult != WAIT_ABANDONED_0) return WaitResult; while (hSemCE->CurCount <= 0) { /* The count is 0, and the thread must wait on the event (which, by the rules, is currently reset) for semaphore resources to become available. First, of course, the mutex must be released so that another thread will be capable of setting the event. */ ReleaseMutex (hSemCE->hMutex); /* Wait for the event to be signaled, indicating a semaphore state change. The event is autoreset and signaled with a SetEvent (not PulseEvent) so exactly one waiting thread (whether or not there is currently a waiting thread) is released as a result of the SetEvent. */ WaitResult = WaitForSingleObject (hSemCE->hEvent, dwMilliseconds); if (WaitResult != WAIT_OBJECT_0) return WaitResult; /* This is where the properties of setting of an autoreset event is critical to assure that, even if the semaphore state changes between the preceding Wait and the next, and even if NO threads are waiting on the event at the time of the SetEvent, at least one thread will be released. Pulsing a manual reset event would appear to work, but it would have a defect which could appear if the semaphore state changed between the two waits. */ WaitResult = WaitForSingleObject (hSemCE->hMutex, dwMilliseconds); if (WaitResult != WAIT_OBJECT_0 && WaitResult != WAIT_ABANDONED_0) return WaitResult; } /* The count is not zero and this thread owns the mutex. */ hSemCE->CurCount--; /* The event is now unsignaled, BUT, the semaphore count may not be zero, in which case the event should be signaled again before releasing the mutex. */ if (hSemCE->CurCount > 0) SetEvent (hSemCE->hEvent); ReleaseMutex (hSemCE->hMutex); return WaitResult; } BOOL CloseSynchHandle (SYNCHHANDLE hSynch) /* Close a synchronization handle. Improvement: Test for a valid handle before dereferencing the handle. */ { BOOL Result = TRUE; if (hSynch->hEvent != NULL) Result = Result && CloseHandle (hSynch->hEvent); if (hSynch->hMutex != NULL) Result = Result && CloseHandle (hSynch->hMutex); if (hSynch->hSemph != NULL) Result = Result && CloseHandle (hSynch->hSemph); HeapFree (GetProcessHeap (), 0, hSynch); return (Result); } static SYNCHHANDLE CleanUp (SYNCHHANDLE hSynch, DWORD Flags) { /* Prepare to return from a create of a synchronization handle. If there was any failure, free any allocated resources. "Flags" indicates which Win32 objects are required in the synchronization handle. */ BOOL ok = TRUE; if (hSynch == NULL) return NULL; if (Flags & 4 == 1 && hSynch->hEvent == NULL) ok = FALSE; if (Flags & 2 == 1 && hSynch->hMutex == NULL) ok = FALSE; if (Flags & 1 == 1 && hSynch->hEvent == NULL) ok = FALSE; if (!ok) { CloseSynchHandle (hSynch); return NULL; } /* Everything worked */ return hSynch; } Here is the header file, Synchize.h, required to support any program using this library. /* Synchize.h - header file to go with Synchize.c */ typedef struct _SYNCH_HANDLE_STRUCTURE { HANDLE hEvent; HANDLE hMutex; HANDLE hSemph; LONG MaxCount; volatile LONG CurCount; LPCTSTR lpName; } SYNCH_HANDLE_STRUCTURE, *SYNCHHANDLE; #define SYNCH_HANDLE_SIZE sizeof (SYNCH_HANDLE_STRUCTURE) /* Error codes - all must have bit 29 set */ #define SYNCH_ERROR 0X20000000 /* EXERCISE - REFINE THE ERROR NUMBERS */ SYNCHHANDLE CreateSemaphoreCE (LPSECURITY_ATTRIBUTES, LONG, LONG, LPCTSTR); BOOL ReleaseSemaphoreCE (SYNCHHANDLE, LONG, LPLONG); DWORD WaitForSemaphoreCE (SYNCHHANDLE, DWORD); BOOL CloseSynchHandle (SYNCHHANDLE); TestSych.cHere is a program that you can use to test the semaphore implementation using a mutex and event. This test program also illustrates a number of Win32 synchronization features and uses a console control handler to shut down the system in an orderly fashion. Among other things, it should never deadlock, and the semaphore state should not be corrupted. See comments in the code. NOTE: This test program, because of its use of standard I/O, will not build for operation in the Windows CE emulation environments. That is the next step. /* Test the synchronization compound object: Windows CE semaphores */ #include "EvryThng.h" #include "Synchize.h" static DWORD WINAPI ProducerTh (LPVOID); static DWORD WINAPI ConsumerTh (LPVOID); static DWORD WINAPI MonitorTh (LPVOID); static BOOL WINAPI CtrlcHandler (DWORD); static SYNCHHANDLE hSemCE; static LONG SemMax, SemInitial, NumThreads, TotalExcess = 0; static HANDLE hTestMutex; static volatile BOOL Debug = FALSE; static volatile BOOL Exit = FALSE; static volatile DWORD NumSuccess = 0, NumFailures = 0, NumProduced = 0, FailureCount = 0, NumConsumed = 0; int _tmain (int argc, LPTSTR argv[]) /* Test CE semaphores; that is, semaphores created out of a mutex, a counter, and an autoreset event. */ { LPHANDLE Producer, Consumer; LONG iThread; DWORD ThreadId; HANDLE hMonitor; if (!SetConsoleCtrlHandler (CtrlcHandler, TRUE)) ReportError (_T("Failed to set console control handler"), 1, TRUE); _tprintf (_T("\nEnter SemMax, SemInitial, NumThreads, Debug: ")); _tscanf (_T("%d %d %d %d"), &SemMax, &SemInitial, &NumThreads, &Debug); if (Debug) _tprintf (_T("You entered: %d %d %d \n"), SemMax, SemInitial, NumThreads); fflush (stdout); /* Create a mutex to synchronize various aspects of the test, such as updating statistics. A CS won't work as we need a WaitForMultipleObjects involving this mutex in the monitor thread. */ hTestMutex = CreateMutex (NULL, FALSE, NULL); if (hTestMutex == NULL) ReportError (_T("Could not create test mutex"), 2, TRUE); NumProduced = SemInitial; /* Initialize the usage statistics for the semaphore. */ hSemCE = CreateSemaphoreCE (NULL, SemInitial, SemMax, NULL); /* CE semaphores cannot be named as the required event objects are unnamed. */ if (hSemCE == NULL) { _tprintf (_T("LastError: %x\n"), GetLastError()); ReportError (_T("Failed to create CE semaphore"), 3, TRUE); } if (Debug) _tprintf ("CE Semaphore created successfully\n"); /* Create all the semaphore consuming and releasing threads. */ Producer = malloc (NumThreads * sizeof(HANDLE)); if (Producer == NULL) ReportError (_T("Cannot allocate Producer handles"), 4, FALSE); Consumer = malloc (NumThreads * sizeof(HANDLE)); if (Consumer == NULL) ReportError (_T("Cannot allocate Consumer handles"), 5, FALSE); hMonitor = (HANDLE)_beginthreadex (NULL, 0, MonitorTh, (LPVOID)5000, CREATE_SUSPENDED, &ThreadId); if (hMonitor == NULL) ReportError (_T("Cannot create monitor thread"), 6, TRUE); SetThreadPriority (hMonitor, THREAD_PRIORITY_HIGHEST); for (iThread = 0; iThread < NumThreads; iThread++) { Producer [iThread] = (HANDLE)_beginthreadex (NULL, 0, ProducerTh, (LPVOID)iThread, CREATE_SUSPENDED, &ThreadId); if (Producer[iThread] == NULL) ReportError (_T("Cannot create producer thread"), 3, TRUE); Consumer [iThread] = (HANDLE)_beginthreadex (NULL, 0, ConsumerTh, (LPVOID)iThread, CREATE_SUSPENDED, &ThreadId); if (Consumer[iThread] == NULL) ReportError (_T("Cannot create consumer thread"), 3, TRUE); } WaitForSingleObject (hTestMutex, INFINITE); _tprintf (_T("All threads created successfully.\n")); ReleaseMutex (hTestMutex); for (iThread = 0; iThread < NumThreads; iThread++) { ResumeThread (Producer [iThread]); ResumeThread (Consumer [iThread]); } ResumeThread (hMonitor); WaitForMultipleObjects (NumThreads, Consumer, TRUE, INFINITE); WaitForMultipleObjects (NumThreads, Producer, TRUE, INFINITE); WaitForSingleObject (hMonitor, INFINITE); for (iThread = 0; iThread < NumThreads; iThread++) { CloseHandle (Producer [iThread]); CloseHandle (Consumer [iThread]); } free (Producer); free (Consumer); CloseHandle (hMonitor); if (!CloseSynchHandle (hSemCE)) ReportError (_T("Failed closing synch handle"), 0, TRUE); _tprintf (_T("All threads terminated\n")); return 0; } static DWORD WINAPI ProducerTh (LPVOID ThNumber) /* Producer thread: 1. Compute for a random period of time. 2. Produce a random number of units, uniformly distributed [1, SemMax] The procedure will fail with no units produced if it would cause the semaphore count to exceed the maximum (this is consistent with Win32 semaphore semantics. */ { DWORD Id = (DWORD)ThNumber, Delay, i, k; LONG PrevCount = 0, RelCount = 0; srand (Id); WaitForSingleObject (hTestMutex, INFINITE); if (Debug) _tprintf (_T("Starting producer number %d.\n"), Id); ReleaseMutex (hTestMutex); while (!Exit) { /* Delay without necessarily giving up processor control. Notice how the production rate is the same as the consumption rate so as to keep things balanced. */ Delay = (DWORD)SemMax * rand(); for (i = 0; i < Delay; i++) k = rand()*rand(); /* Waste time. */ if (rand() % 3 == 0) Sleep (0); /* Give up the processor 1/3 of the time just to make the thread interaction more interesting. */ RelCount = (long)(((float)rand()/RAND_MAX) * SemMax) + 1; if (!ReleaseSemaphoreCE (hSemCE, RelCount, &PrevCount)) { WaitForSingleObject (hTestMutex, INFINITE); if (Debug) _tprintf (_T("Producer #: %d Failed. PrevCount = %d\n"), Id, PrevCount); NumFailures++; /* Maintain producer statistics. */ FailureCount += RelCount; ReleaseMutex (hTestMutex); } else { WaitForSingleObject (hTestMutex, INFINITE); if (Debug) _tprintf (_T("Producer #: %d Succeeded. Prev Count = %d\n"), Id, PrevCount); NumSuccess++; NumProduced += RelCount; ReleaseMutex (hTestMutex); } } _endthreadex (0); return 0; } static DWORD WINAPI ConsumerTh (LPVOID ThNumber) { DWORD Id = (DWORD)ThNumber, Delay, i, k; srand (Id); WaitForSingleObject (hTestMutex, INFINITE); if (Debug) _tprintf (_T("Starting consumer number %d.\n"), Id); ReleaseMutex (hTestMutex); while (!Exit) { /* Delay without necessarily giving up processor control. */ Delay = rand(); for (i = 0; i < Delay; i++) k = rand()*rand(); /* Waste time. */; if (rand() % 3 == 0) Sleep (0); /* Give up the processor 1/3 of the time just to make the thread interaction more interesting. */ if (WaitForSemaphoreCE (hSemCE, rand()) != WAIT_OBJECT_0) { if (Debug) { WaitForSingleObject (hTestMutex, INFINITE); _tprintf (_T("Consumer #: %d Timed out\n"), Id); ReleaseMutex (hTestMutex); } } else { /* The semaphore unit was obtained successfully. Update statistics. */ WaitForSingleObject (hTestMutex, INFINITE); if (Debug) _tprintf (_T("Consumer #: %d Obtained one unit\n"), Id); NumConsumed++; ReleaseMutex (hTestMutex); } } _endthreadex (0); return 0; } static DWORD WINAPI MonitorTh (LPVOID Delay) /* Monitor thread - periodically check the statistics for consistency. */ { LONG CurCount = 0, Max = 0, ExcessCount; HANDLE hBoth[2] = {hTestMutex, hSemCE->hMutex}; /* This is cheating to reach inside the opaque handle, but we need to get simultaneous access to both the semaphore state and to the statistics in order to test consistency. Some consistency failures are still possible, however, as the statistics are updated independently of changing the semaphore state, so the semaphore might be changed by another thread before the producer or consumer can update the global statistics. Thus, there can be no absolute consistency between the semaphore state and the global statistics maintained by this test program, and, to make them consistent would require putting a critical section around the producer/consumer loop bodies, which would destroy the asynchronous operation required for testing. But, the general state of the semaphore can still be checked, and we can also be sure that it has not been corrupted. */ while (!Exit) { WaitForMultipleObjects (2, hBoth, TRUE, INFINITE); CurCount = hSemCE->CurCount; Max = hSemCE->MaxCount; ExcessCount = NumProduced - (NumConsumed + CurCount); /* Excess of production over consumption which should average out to 0. */ _tprintf (_T("*Monitor-stats:\nProduced: %d\nConsumed: %d\nCurrent: %d\nMax: %d\n"), NumProduced, NumConsumed, CurCount, Max); /* For Correctness, we must have: Produced == Consumed + CurCount */ if (ExcessCount != 0 || CurCount < 0 || CurCount > SemMax || SemMax != Max) { _tprintf (_T("****Discrepancy: %d\n"), ExcessCount); TotalExcess += ExcessCount; } else _tprintf (_T("****Consistency test passed. Total excess count: %d\n"), TotalExcess); _tprintf (_T("Successful release calls: %d\nFailed release calls: %d\n"), NumSuccess, NumFailures); _tprintf (_T("Units not released: %d\n*************\n"), FailureCount); ReleaseMutex (hTestMutex); ReleaseMutex (hSemCE->hMutex); Sleep ((DWORD)Delay); } _endthreadex(0); return 0; } static BOOL WINAPI CtrlcHandler (DWORD CtrlEvent) { DWORD PrevCount; LONG i; if (CtrlEvent == CTRL_C_EVENT) { WaitForSingleObject (hTestMutex, INFINITE); _tprintf (_T("Control-c received. Shutting down.\n")); ReleaseMutex (hTestMutex); Exit = TRUE; /* Assure that all Consumer threads are not blocked waiting on a semaphore so that they have a chance to shut down. */ for (i = 0; i < NumThreads; i++) { ReleaseSemaphoreCE (hSemCE, 1, &PrevCount); /* Release the CPU so that the consumer can be scheduled. */ Sleep (0); } return TRUE; } else return FALSE; } /* SYNCHHANDLE CreateSemaphoreCE (LPSECURITY_ATTRIBUTES, LONG, LONG, LPCTSTR); BOOL ReleaseSemaphoreCE (SYNCHHANDLE, LONG, LPLONG); DWORD WaitForSemaphoreCE (SYNCHHANDLE, DWORD); BOOL CloseSynchHandle (SYNCHHANDLE); */ |