queue.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include "queue.h"
   4
   5 Queue *queue_create()
   6 {
   7         Queue *queue = malloc(sizeof(Queue));
   8         queue->cancel = false;
   9         queue->list = list_create(NULL);
  10         pthread_cond_init(&queue->cv, NULL);
  11         pthread_mutex_init(&queue->mtx, NULL);
  12         return queue;
  13 }
  14
  15 void queue_delete(Queue *queue)
  16 {
  17         pthread_cond_destroy(&queue->cv);
  18         pthread_mutex_destroy(&queue->mtx);
  19         list_clear(&queue->list);
  20         free(queue);
  21 }
  22
  23 void queue_enqueue(Queue *queue, void *elem)
  24 {
  25         pthread_mutex_lock(&queue->mtx);
  26         list_put(&queue->list, elem, NULL);
  27         pthread_cond_signal(&queue->cv);
  28         pthread_mutex_unlock(&queue->mtx);
  29 }
  30
  31 void *dequeue(Queue *queue)
  32 {
  33         return queue_dequeue_callback(queue, NULL);
  34 }
  35
  36 void *queue_dequeue_callback(Queue *queue, void (*callback)(void *elem))
  37 {
  38         void *elem = NULL;
  39
  40         while (! queue->cancel && ! elem) {
  41                 pthread_mutex_lock(&queue->mtx);
  42
  43                 ListPair **lptr = &queue->list.first;
  44                 if (*lptr) {
  45                         elem = (*lptr)->key;
  46                         ListPair *next = (*lptr)->next;
  47                         free(*lptr);
  48                         *lptr = next;
  49
  50                         if (callback)
  51                                 callback(elem);
  52                 } else {
  53                         pthread_cond_wait(&queue->cv, &queue->mtx);
  54                 }
  55
  56                 pthread_mutex_unlock(&queue->mtx);
  57         }
  58
  59         return elem;
  60 }
  61
  62 void queue_cancel(Queue *queue)
  63 {
  64         queue->cancel = true;
  65
  66         pthread_mutex_lock(&queue->mtx);
  67         pthread_cond_broadcast(&queue->cv);
  68         pthread_mutex_unlock(&queue->mtx);
  69 }