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   free(pq);
  86   return;
  87 }
  88
  89 HANDLE
  90 GetWorkQueueHandle ( WorkQueue* pq )
  91 {
  92   if (!pq) return NULL;
  93
  94   return pq->workAvailable;
  95 }
  96
  97 /*
  98  * Function: GetWork
  99  *
 100  * Fetch a work item from the queue, blocking if none available.
 101  * Return value indicates of FALSE indicates error/fatal condition.
 102  */
 103 BOOL
 104 GetWork ( WorkQueue* pq, void** ppw )
 105 {
 106   DWORD rc;
 107
 108   if (!pq) {
 109     queue_error("GetWork", "NULL WorkQueue object");
 110     return FALSE;
 111   }
 112   if (!ppw) {
 113     queue_error("GetWork", "NULL WorkItem object");
 114     return FALSE;
 115   }
 116
 117   /* Block waiting for work item to become available */
 118   if ( (rc = WaitForSingleObject( pq->workAvailable, INFINITE)) != WAIT_OBJECT_0 ) {
 119     queue_error_rc("GetWork.WaitForSingleObject(workAvailable)",
 120                    ( (WAIT_FAILED == rc) ? GetLastError() : rc));
 121     return FALSE;
 122   }
 123
 124   return FetchWork(pq,ppw);
 125 }
 126
 127 /*
 128  * Function: FetchWork
 129  *
 130  * Fetch a work item from the queue, blocking if none available.
 131  * Return value indicates of FALSE indicates error/fatal condition.
 132  */
 133 BOOL
 134 FetchWork ( WorkQueue* pq, void** ppw )
 135 {
 136   DWORD rc;
 137
 138   if (!pq) {
 139     queue_error("FetchWork", "NULL WorkQueue object");
 140     return FALSE;
 141   }
 142   if (!ppw) {
 143     queue_error("FetchWork", "NULL WorkItem object");
 144     return FALSE;
 145   }
 146
 147   EnterCriticalSection(&pq->queueLock);
 148   *ppw = pq->items[pq->head];
 149   /* For sanity's sake, zero out the pointer. */
 150   pq->items[pq->head] = NULL;
 151   pq->head = (pq->head + 1) % WORKQUEUE_SIZE;
 152   rc = ReleaseSemaphore(pq->roomAvailable,1, NULL);
 153   LeaveCriticalSection(&pq->queueLock);
 154   if ( 0 == rc ) {
 155     queue_error_rc("FetchWork.ReleaseSemaphore()", GetLastError());
 156     return FALSE;
 157   }
 158
 159   return TRUE;
 160 }
 161
 162 /*
 163  * Function: SubmitWork
 164  *
 165  * Add work item to the queue, blocking if no room available.
 166  * Return value indicates of FALSE indicates error/fatal condition.
 167  */
 168 BOOL
 169 SubmitWork ( WorkQueue* pq, void* pw )
 170 {
 171   DWORD rc;
 172
 173   if (!pq) {
 174     queue_error("SubmitWork", "NULL WorkQueue object");
 175     return FALSE;
 176   }
 177   if (!pw) {
 178     queue_error("SubmitWork", "NULL WorkItem object");
 179     return FALSE;
 180   }
 181
 182   /* Block waiting for work item to become available */
 183   if ( (rc = WaitForSingleObject( pq->roomAvailable, INFINITE)) != WAIT_OBJECT_0 ) {
 184     queue_error_rc("SubmitWork.WaitForSingleObject(workAvailable)",
 185                    ( (WAIT_FAILED == rc) ? GetLastError() : rc));
 186
 187     return FALSE;
 188   }
 189
 190   EnterCriticalSection(&pq->queueLock);
 191   pq->items[pq->tail] = pw;
 192   pq->tail = (pq->tail + 1) % WORKQUEUE_SIZE;
 193   rc = ReleaseSemaphore(pq->workAvailable,1, NULL);
 194   LeaveCriticalSection(&pq->queueLock);
 195   if ( 0 == rc ) {
 196     queue_error_rc("SubmitWork.ReleaseSemaphore()", GetLastError());
 197     return FALSE;
 198   }
 199
 200   return TRUE;
 201 }
 202
 203 /* Error handling */
 204
 205 static void
 206 queue_error_rc( char* loc,
 207                 DWORD err)
 208 {
 209   fprintf(stderr, "%s failed: return code = 0x%lx\n", loc, err);
 210   fflush(stderr);
 211   return;
 212 }
 213
 214
 215 static void
 216 queue_error( char* loc,
 217              char* reason)
 218 {
 219   fprintf(stderr, "%s failed: %s\n", loc, reason);
 220   fflush(stderr);
 221   return;
 222 }
 223