src/tsq.c

   1 /*
   2 Copyright (C) 2001-2009, Parrot Foundation.
   3 $Id$
   4
   5 =head1 NAME
   6
   7 src/tsq.c - Thread-safe queues
   8
   9 =head1 DESCRIPTION
  10
  11 This file implements thread-safe queues for Parrot.
  12
  13 =head2 Functions
  14
  15 =over 4
  16
  17 =cut
  18
  19 */
  20
  21 #include "parrot/parrot.h"
  22
  23 /* HEADERIZER HFILE: include/parrot/tsq.h */
  24
  25 /*
  26
  27 =item C<QUEUE_ENTRY * pop_entry(QUEUE *queue)>
  28
  29 Does a synchronized removal of the head entry off the queue and returns it.
  30
  31 =cut
  32
  33 */
  34
  35 PARROT_CAN_RETURN_NULL
  36 QUEUE_ENTRY *
  37 pop_entry(ARGMOD(QUEUE *queue))
  38 {
  39     ASSERT_ARGS(pop_entry)
  40     QUEUE_ENTRY *returnval;
  41     queue_lock(queue);
  42     returnval = nosync_pop_entry(queue);
  43     queue_unlock(queue);
  44     return returnval;
  45 }
  46
  47 /*
  48
  49 =item C<QUEUE_ENTRY * peek_entry(const QUEUE *queue)>
  50
  51 This does no locking, so the result might have changed by the time you
  52 get the entry, but a synchronized C<pop_entry()> will check again and
  53 return C<NULL> if the queue is empty.
  54
  55 =cut
  56
  57 */
  58
  59 PARROT_CAN_RETURN_NULL
  60 PARROT_WARN_UNUSED_RESULT
  61 QUEUE_ENTRY *
  62 peek_entry(ARGIN(const QUEUE *queue))
  63 {
  64     ASSERT_ARGS(peek_entry)
  65     return queue->head;
  66 }
  67
  68 /*
  69
  70 =item C<QUEUE_ENTRY * nosync_pop_entry(QUEUE *queue)>
  71
  72 Grab an entry off the queue with no synchronization. Internal only,
  73 because it's darned evil and shouldn't be used outside the module. It's
  74 in here so we don't have to duplicate pop code.
  75
  76 =cut
  77
  78 */
  79
  80 PARROT_CANNOT_RETURN_NULL
  81 QUEUE_ENTRY *
  82 nosync_pop_entry(ARGMOD(QUEUE *queue))
  83 {
  84     ASSERT_ARGS(nosync_pop_entry)
  85     QUEUE_ENTRY *returnval;
  86     if (!queue->head) {
  87         return NULL;
  88     }
  89     returnval = queue->head;
  90     if (queue->head == queue->tail) {
  91         queue->head = NULL;
  92         queue->tail = NULL;
  93     }
  94     else {
  95         queue->head = queue->head->next;
  96     }
  97     returnval->next = NULL;
  98     return returnval;
  99 }
 100
 101 /*
 102
 103 =item C<QUEUE_ENTRY * wait_for_entry(QUEUE *queue)>
 104
 105 Does a synchronized removal of the head entry off the queue, waiting if
 106 necessary until there is an entry, and then returns it.
 107
 108 =cut
 109
 110 */
 111
 112 PARROT_CAN_RETURN_NULL
 113 QUEUE_ENTRY *
 114 wait_for_entry(ARGMOD(QUEUE *queue))
 115 {
 116     ASSERT_ARGS(wait_for_entry)
 117     QUEUE_ENTRY *returnval;
 118
 119     queue_lock(queue);
 120     while (queue->head == NULL) {
 121         queue_wait(queue);
 122     }
 123     returnval = nosync_pop_entry(queue);
 124     queue_unlock(queue);
 125     return returnval;
 126
 127 }
 128
 129 /*
 130
 131 =item C<void push_entry(QUEUE *queue, QUEUE_ENTRY *entry)>
 132
 133 Does a synchronized insertion of C<entry> onto the tail of the queue.
 134
 135 =cut
 136
 137 */
 138
 139 void
 140 push_entry(ARGMOD(QUEUE *queue), ARGIN(QUEUE_ENTRY *entry))
 141 {
 142     ASSERT_ARGS(push_entry)
 143     queue_lock(queue);
 144     /* Is there something in the queue? */
 145     if (queue->tail) {
 146         queue->tail->next = entry;
 147         queue->tail = entry;
 148     }
 149     else {
 150         queue->head = entry;
 151         queue->tail = entry;
 152     }
 153     queue_signal(queue);        /* assumes only one waiter */
 154     queue_unlock(queue);
 155 }
 156
 157 /*
 158
 159 =item C<void unshift_entry(QUEUE *queue, QUEUE_ENTRY *entry)>
 160
 161 Does a synchronized insertion of C<entry> into the head of the queue.
 162
 163 =cut
 164
 165 */
 166
 167 void
 168 unshift_entry(ARGMOD(QUEUE *queue), ARGIN(QUEUE_ENTRY *entry))
 169 {
 170     ASSERT_ARGS(unshift_entry)
 171     QUEUE_ENTRY *cur;
 172
 173     queue_lock(queue);
 174     cur = queue->head;
 175     if (!cur) {
 176         /* empty just set head */
 177         queue->head = entry;
 178         queue->tail = entry;
 179     }
 180     else {
 181         queue->head = entry;
 182         entry->next = cur;
 183     }
 184     queue_signal(queue);
 185     queue_unlock(queue);
 186 }
 187
 188 /*
 189
 190 =item C<void nosync_insert_entry(QUEUE *queue, QUEUE_ENTRY *entry)>
 191
 192 Inserts a timed event according to C<abstime>. The caller has to hold the
 193 queue mutex.
 194
 195 =cut
 196
 197 */
 198
 199 void
 200 nosync_insert_entry(ARGMOD(QUEUE *queue), ARGIN(QUEUE_ENTRY *entry))
 201 {
 202     ASSERT_ARGS(nosync_insert_entry)
 203     QUEUE_ENTRY *cur = queue->head;
 204     QUEUE_ENTRY *prev;
 205     parrot_event *event;
 206     FLOATVAL abs_time;
 207
 208     PARROT_ASSERT(entry->type == QUEUE_ENTRY_TYPE_TIMED_EVENT);
 209     /*
 210      * empty queue - just insert
 211      */
 212     if (!cur) {
 213         queue->head = entry;
 214         queue->tail = entry;
 215         return;
 216     }
 217
 218     prev     = NULL;
 219     event    = (parrot_event *)entry->data;
 220     abs_time = event->u.timer_event.abs_time;
 221
 222     while (cur && cur->type == QUEUE_ENTRY_TYPE_TIMED_EVENT) {
 223         const parrot_event * const cur_event = (parrot_event *)cur->data;
 224         if (abs_time > cur_event->u.timer_event.abs_time) {
 225             prev = cur;
 226             cur = cur->next;
 227         }
 228         else
 229             break;
 230     }
 231     if (!prev)
 232         queue->head = entry;
 233     else {
 234         prev->next = entry;
 235         if (prev == queue->tail)
 236             queue->tail = entry;
 237     }
 238     entry->next = cur;
 239 }
 240
 241 /*
 242
 243 =item C<void insert_entry(QUEUE *queue, QUEUE_ENTRY *entry)>
 244
 245 Does a synchronized insert of C<entry>.
 246
 247 =cut
 248
 249 */
 250
 251 void
 252 insert_entry(ARGMOD(QUEUE *queue), ARGIN(QUEUE_ENTRY *entry))
 253 {
 254     ASSERT_ARGS(insert_entry)
 255     queue_lock(queue);
 256     nosync_insert_entry(queue, entry);
 257     queue_signal(queue);
 258     queue_unlock(queue);
 259 }
 260
 261 /*
 262
 263 =item C<void queue_lock(QUEUE *queue)>
 264
 265 Locks the queue's mutex.
 266
 267 =cut
 268
 269 */
 270
 271 void
 272 queue_lock(ARGMOD(QUEUE *queue))
 273 {
 274     ASSERT_ARGS(queue_lock)
 275     LOCK(queue->queue_mutex);
 276 }
 277
 278 /*
 279
 280 =item C<void queue_unlock(QUEUE *queue)>
 281
 282 Unlocks the queue's mutex.
 283
 284 =cut
 285
 286 */
 287
 288 void
 289 queue_unlock(ARGMOD(QUEUE *queue))
 290 {
 291     ASSERT_ARGS(queue_unlock)
 292     UNLOCK(queue->queue_mutex);
 293 }
 294
 295 /*
 296
 297 =item C<void queue_broadcast(QUEUE *queue)>
 298
 299 This function wakes up I<every> thread waiting on the queue.
 300
 301 =cut
 302
 303 */
 304
 305 void
 306 queue_broadcast(ARGMOD(QUEUE *queue))
 307 {
 308     ASSERT_ARGS(queue_broadcast)
 309     COND_BROADCAST(queue->queue_condition);
 310 }
 311
 312 /*
 313
 314 =item C<void queue_signal(QUEUE *queue)>
 315
 316 =cut
 317
 318 */
 319
 320 void
 321 queue_signal(ARGMOD(QUEUE *queue))
 322 {
 323     ASSERT_ARGS(queue_signal)
 324     COND_SIGNAL(queue->queue_condition);
 325 }
 326
 327 /*
 328
 329 =item C<void queue_wait(QUEUE *queue)>
 330
 331 Instructs the queue to wait.
 332
 333 =cut
 334
 335 */
 336
 337 void
 338 queue_wait(ARGMOD(QUEUE *queue))
 339 {
 340     ASSERT_ARGS(queue_wait)
 341     COND_WAIT(queue->queue_condition, queue->queue_mutex);
 342 }
 343
 344 /*
 345
 346 =item C<void queue_timedwait(QUEUE *queue, const struct timespec *abs_time)>
 347
 348 Instructs the queue to wait for C<abs_time> seconds (?).
 349
 350 =cut
 351
 352 */
 353
 354 void
 355 queue_timedwait(ARGMOD(QUEUE *queue), ARGIN(const struct timespec *abs_time))
 356 {
 357     ASSERT_ARGS(queue_timedwait)
 358     COND_TIMED_WAIT(queue->queue_condition, queue->queue_mutex, abs_time);
 359 }
 360
 361 /*
 362
 363 =item C<QUEUE* queue_init(UINTVAL prio)>
 364
 365 Initializes the queue, setting C<prio> as the queue's priority.
 366
 367 =cut
 368
 369 */
 370
 371 PARROT_CAN_RETURN_NULL
 372 PARROT_MALLOC
 373 QUEUE*
 374 queue_init(UINTVAL prio)
 375 {
 376     ASSERT_ARGS(queue_init)
 377     QUEUE * const queue = mem_allocate_typed(QUEUE);
 378
 379     queue->head = queue->tail = NULL;
 380     queue->max_prio = prio;
 381     COND_INIT(queue->queue_condition);
 382     MUTEX_INIT(queue->queue_mutex);
 383     return queue;
 384 }
 385
 386 /*
 387
 388 =item C<void queue_destroy(QUEUE *queue)>
 389
 390 Destroys the queue, raising an exception if it is not empty.
 391
 392 =cut
 393
 394 */
 395
 396 void
 397 queue_destroy(ARGMOD(QUEUE *queue))
 398 {
 399     ASSERT_ARGS(queue_destroy)
 400     if (peek_entry(queue))
 401         exit_fatal(1, "Queue not empty on destroy");
 402
 403     COND_DESTROY(queue->queue_condition);
 404     MUTEX_DESTROY(queue->queue_mutex);
 405     mem_sys_free(queue);
 406 }
 407
 408 /*
 409
 410 =back
 411
 412 =head1 SEE ALSO
 413
 414 F<include/parrot/tsq.h>.
 415
 416 =cut
 417
 418 */
 419
 420
 421 /*
 422  * Local variables:
 423  *   c-file-style: "parrot"
 424  * End:
 425  * vim: expandtab shiftwidth=4:
 426  */