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   wq->head   = 0;
  49   wq->tail   = 0;
  50
  51   InitializeCriticalSection(&wq->queueLock);
  52   wq->workAvailable = newSemaphore(0, WORKQUEUE_SIZE);
  53   wq->roomAvailable = newSemaphore(WORKQUEUE_SIZE, WORKQUEUE_SIZE);
  54
  55   /* Fail if we were unable to create any of the sync objects. */
  56   if ( NULL == wq->workAvailable ||
  57        NULL == wq->roomAvailable ) {
  58     FreeWorkQueue(wq);
  59     return NULL;
  60   }
  61
  62   return wq;
  63 }
  64
  65 void
  66 FreeWorkQueue ( WorkQueue* pq )
  67 {
  68   /* Close the semaphores; any threads blocked waiting
  69    * on either will as a result be woken up.
  70    */
  71   if ( pq->workAvailable ) {
  72     CloseHandle(pq->workAvailable);
  73   }
  74   if ( pq->roomAvailable ) {
  75     CloseHandle(pq->workAvailable);
  76   }
  77   free(pq);
  78   return;
  79 }
  80
  81 HANDLE
  82 GetWorkQueueHandle ( WorkQueue* pq )
  83 {
  84   if (!pq) return NULL;
  85
  86   return pq->workAvailable;
  87 }
  88
  89 /*
  90  * Function: GetWork
  91  *
  92  * Fetch a work item from the queue, blocking if none available.
  93  * Return value indicates of FALSE indicates error/fatal condition.
  94  */
  95 BOOL
  96 GetWork ( WorkQueue* pq, void** ppw )
  97 {
  98   DWORD rc;
  99
 100   if (!pq) {
 101     queue_error("GetWork", "NULL WorkQueue object");
 102     return FALSE;
 103   }
 104   if (!ppw) {
 105     queue_error("GetWork", "NULL WorkItem object");
 106     return FALSE;
 107   }
 108
 109   /* Block waiting for work item to become available */
 110   if ( (rc = WaitForSingleObject( pq->workAvailable, INFINITE)) != WAIT_OBJECT_0 ) {
 111     queue_error_rc("GetWork.WaitForSingleObject(workAvailable)",
 112                    ( (WAIT_FAILED == rc) ? GetLastError() : rc));
 113     return FALSE;
 114   }
 115
 116   return FetchWork(pq,ppw);
 117 }
 118
 119 /*
 120  * Function: FetchWork
 121  *
 122  * Fetch a work item from the queue, blocking if none available.
 123  * Return value indicates of FALSE indicates error/fatal condition.
 124  */
 125 BOOL
 126 FetchWork ( WorkQueue* pq, void** ppw )
 127 {
 128   DWORD rc;
 129
 130   if (!pq) {
 131     queue_error("FetchWork", "NULL WorkQueue object");
 132     return FALSE;
 133   }
 134   if (!ppw) {
 135     queue_error("FetchWork", "NULL WorkItem object");
 136     return FALSE;
 137   }
 138
 139   EnterCriticalSection(&pq->queueLock);
 140   *ppw = pq->items[pq->head];
 141   /* For sanity's sake, zero out the pointer. */
 142   pq->items[pq->head] = NULL;
 143   pq->head = (pq->head + 1) % WORKQUEUE_SIZE;
 144   rc = ReleaseSemaphore(pq->roomAvailable,1, NULL);
 145   LeaveCriticalSection(&pq->queueLock);
 146   if ( 0 == rc ) {
 147     queue_error_rc("FetchWork.ReleaseSemaphore()", GetLastError());
 148     return FALSE;
 149   }
 150
 151   return TRUE;
 152 }
 153
 154 /*
 155  * Function: SubmitWork
 156  *
 157  * Add work item to the queue, blocking if no room available.
 158  * Return value indicates of FALSE indicates error/fatal condition.
 159  */
 160 BOOL
 161 SubmitWork ( WorkQueue* pq, void* pw )
 162 {
 163   DWORD rc;
 164
 165   if (!pq) {
 166     queue_error("SubmitWork", "NULL WorkQueue object");
 167     return FALSE;
 168   }
 169   if (!pw) {
 170     queue_error("SubmitWork", "NULL WorkItem object");
 171     return FALSE;
 172   }
 173
 174   /* Block waiting for work item to become available */
 175   if ( (rc = WaitForSingleObject( pq->roomAvailable, INFINITE)) != WAIT_OBJECT_0 ) {
 176     queue_error_rc("SubmitWork.WaitForSingleObject(workAvailable)",
 177                    ( (WAIT_FAILED == rc) ? GetLastError() : rc));
 178
 179     return FALSE;
 180   }
 181
 182   EnterCriticalSection(&pq->queueLock);
 183   pq->items[pq->tail] = pw;
 184   pq->tail = (pq->tail + 1) % WORKQUEUE_SIZE;
 185   rc = ReleaseSemaphore(pq->workAvailable,1, NULL);
 186   LeaveCriticalSection(&pq->queueLock);
 187   if ( 0 == rc ) {
 188     queue_error_rc("SubmitWork.ReleaseSemaphore()", GetLastError());
 189     return FALSE;
 190   }
 191
 192   return TRUE;
 193 }
 194
 195 /* Error handling */
 196
 197 static void
 198 queue_error_rc( char* loc,
 199                 DWORD err)
 200 {
 201   fprintf(stderr, "%s failed: return code = 0x%lx\n", loc, err);
 202   fflush(stderr);
 203   return;
 204 }
 205
 206
 207 static void
 208 queue_error( char* loc,
 209              char* reason)
 210 {
 211   fprintf(stderr, "%s failed: %s\n", loc, reason);
 212   fflush(stderr);
 213   return;
 214 }
 215