src/tsq.c

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