rts/win32/WorkQueue.c

   1 /*
   2  * A fixed-size queue; MT-friendly.
   3  *
   4  * (c) sof, 2002-2003.
   5  */
   6 #include "WorkQueue.h"
   7 #include <stdio.h>
   8 #include <stdlib.h>
   9 #include <string.h>
  10
  11 static void queue_error_rc( char* loc, DWORD err);
  12 static void queue_error( char* loc, char* reason);
  13
  14
  15 /* Wrapper around OS call to create semaphore */
  16 static Semaphore
  17 newSemaphore(int initCount, int max)
  18 {
  19   Semaphore s;
  20   s = CreateSemaphore ( NULL,       /* LPSECURITY_ATTRIBUTES (default) */
  21                         initCount,  /* LONG lInitialCount */
  22                         max,        /* LONG lMaxCount */
  23                         NULL);      /* LPCTSTR (anonymous / no object name) */
  24   if ( NULL == s) {
  25     queue_error_rc("newSemaphore", GetLastError());
  26     return NULL;
  27   }
  28   return s;
  29 }
  30
  31 /*
  32  * Function: NewWorkQueue
  33  *
  34  * The queue constructor - semaphores are initialised to match
  35  * max number of queue entries.
  36  *
  37  */
  38 WorkQueue*
  39 NewWorkQueue()
  40 {
  41   WorkQueue* wq = (WorkQueue*)malloc(sizeof(WorkQueue));
  42
  43   if (!wq) {
  44     queue_error("NewWorkQueue", "malloc() failed");
  45     return wq;
  46   }
  47
  48   memset(wq, 0, sizeof *wq);
  49
  50   InitializeCriticalSection(&wq->queueLock);
  51   wq->workAvailable = newSemaphore(0, WORKQUEUE_SIZE);
  52   wq->roomAvailable = newSemaphore(WORKQUEUE_SIZE, WORKQUEUE_SIZE);
  53
  54   /* Fail if we were unable to create any of the sync objects. */
  55   if ( NULL == wq->workAvailable ||
  56        NULL == wq->roomAvailable ) {
  57     FreeWorkQueue(wq);
  58     return NULL;
  59   }
  60
  61   return wq;
  62 }
  63
  64 void
  65 FreeWorkQueue ( WorkQueue* pq )
  66 {
  67   int i;
  68
  69   /* Free any remaining work items. */
  70   for (i = 0; i < WORKQUEUE_SIZE; i++) {
  71     if (pq->items[i] != NULL) {
  72       free(pq->items[i]);
  73     }
  74   }
  75
  76   /* Close the semaphores; any threads blocked waiting
  77    * on either will as a result be woken up.
  78    */
  79   if ( pq->workAvailable ) {
  80     CloseHandle(pq->workAvailable);
  81   }
  82   if ( pq->roomAvailable ) {
  83     CloseHandle(pq->roomAvailable);
  84   }
  85   DeleteCriticalSection(&pq->queueLock);
  86   free(pq);
  87   return;
  88 }
  89
  90 HANDLE
  91 GetWorkQueueHandle ( WorkQueue* pq )
  92 {
  93   if (!pq) return NULL;
  94
  95   return pq->workAvailable;
  96 }
  97
  98 /*
  99  * Function: GetWork
 100  *
 101  * Fetch a work item from the queue, blocking if none available.
 102  * Return value indicates of FALSE indicates error/fatal condition.
 103  */
 104 BOOL
 105 GetWork ( WorkQueue* pq, void** ppw )
 106 {
 107   DWORD rc;
 108
 109   if (!pq) {
 110     queue_error("GetWork", "NULL WorkQueue object");
 111     return FALSE;
 112   }
 113   if (!ppw) {
 114     queue_error("GetWork", "NULL WorkItem object");
 115     return FALSE;
 116   }
 117
 118   /* Block waiting for work item to become available */
 119   if ( (rc = WaitForSingleObject( pq->workAvailable, INFINITE)) != WAIT_OBJECT_0 ) {
 120     queue_error_rc("GetWork.WaitForSingleObject(workAvailable)",
 121                    ( (WAIT_FAILED == rc) ? GetLastError() : rc));
 122     return FALSE;
 123   }
 124
 125   return FetchWork(pq,ppw);
 126 }
 127
 128 /*
 129  * Function: FetchWork
 130  *
 131  * Fetch a work item from the queue, blocking if none available.
 132  * Return value indicates of FALSE indicates error/fatal condition.
 133  */
 134 BOOL
 135 FetchWork ( WorkQueue* pq, void** ppw )
 136 {
 137   DWORD rc;
 138
 139   if (!pq) {
 140     queue_error("FetchWork", "NULL WorkQueue object");
 141     return FALSE;
 142   }
 143   if (!ppw) {
 144     queue_error("FetchWork", "NULL WorkItem object");
 145     return FALSE;
 146   }
 147
 148   EnterCriticalSection(&pq->queueLock);
 149   *ppw = pq->items[pq->head];
 150   /* For sanity's sake, zero out the pointer. */
 151   pq->items[pq->head] = NULL;
 152   pq->head = (pq->head + 1) % WORKQUEUE_SIZE;
 153   rc = ReleaseSemaphore(pq->roomAvailable,1, NULL);
 154   LeaveCriticalSection(&pq->queueLock);
 155   if ( 0 == rc ) {
 156     queue_error_rc("FetchWork.ReleaseSemaphore()", GetLastError());
 157     return FALSE;
 158   }
 159
 160   return TRUE;
 161 }
 162
 163 /*
 164  * Function: SubmitWork
 165  *
 166  * Add work item to the queue, blocking if no room available.
 167  * Return value indicates of FALSE indicates error/fatal condition.
 168  */
 169 BOOL
 170 SubmitWork ( WorkQueue* pq, void* pw )
 171 {
 172   DWORD rc;
 173
 174   if (!pq) {
 175     queue_error("SubmitWork", "NULL WorkQueue object");
 176     return FALSE;
 177   }
 178   if (!pw) {
 179     queue_error("SubmitWork", "NULL WorkItem object");
 180     return FALSE;
 181   }
 182
 183   /* Block waiting for work item to become available */
 184   if ( (rc = WaitForSingleObject( pq->roomAvailable, INFINITE)) != WAIT_OBJECT_0 ) {
 185     queue_error_rc("SubmitWork.WaitForSingleObject(workAvailable)",
 186                    ( (WAIT_FAILED == rc) ? GetLastError() : rc));
 187
 188     return FALSE;
 189   }
 190
 191   EnterCriticalSection(&pq->queueLock);
 192   pq->items[pq->tail] = pw;
 193   pq->tail = (pq->tail + 1) % WORKQUEUE_SIZE;
 194   rc = ReleaseSemaphore(pq->workAvailable,1, NULL);
 195   LeaveCriticalSection(&pq->queueLock);
 196   if ( 0 == rc ) {
 197     queue_error_rc("SubmitWork.ReleaseSemaphore()", GetLastError());
 198     return FALSE;
 199   }
 200
 201   return TRUE;
 202 }
 203
 204 /* Error handling */
 205
 206 static void
 207 queue_error_rc( char* loc,
 208                 DWORD err)
 209 {
 210   fprintf(stderr, "%s failed: return code = 0x%lx\n", loc, err);
 211   fflush(stderr);
 212   return;
 213 }
 214
 215
 216 static void
 217 queue_error( char* loc,
 218              char* reason)
 219 {
 220   fprintf(stderr, "%s failed: %s\n", loc, reason);
 221   fflush(stderr);
 222   return;
 223 }
 224