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->finish = false;
   9         queue->cancel = false;
  10         queue->list = list_create(NULL);
  11         pthread_cond_init(&queue->cv, NULL);
  12         pthread_mutex_init(&queue->mtx, NULL);
  13         return queue;
  14 }
  15
  16 void queue_delete(Queue *queue)
  17 {
  18         pthread_cond_destroy(&queue->cv);
  19         pthread_mutex_destroy(&queue->mtx);
  20         list_clear(&queue->list);
  21         free(queue);
  22 }
  23
  24 void queue_enqueue(Queue *queue, void *elem)
  25 {
  26         pthread_mutex_lock(&queue->mtx);
  27         if (! queue->finish) {
  28                 list_put(&queue->list, elem, NULL);
  29                 pthread_cond_signal(&queue->cv);
  30         }
  31         pthread_mutex_unlock(&queue->mtx);
  32 }
  33
  34 void *queue_dequeue(Queue *queue)
  35 {
  36         return queue_dequeue_callback(queue, NULL);
  37 }
  38
  39 void *queue_dequeue_callback(Queue *queue, void (*callback)(void *elem))
  40 {
  41         void *elem = NULL;
  42
  43         while (! queue->cancel && ! elem) {
  44                 pthread_mutex_lock(&queue->mtx);
  45
  46                 ListPair **lptr = &queue->list.first;
  47                 if (*lptr) {
  48                         elem = (*lptr)->key;
  49                         ListPair *next = (*lptr)->next;
  50                         free(*lptr);
  51                         *lptr = next;
  52
  53                         if (callback)
  54                                 callback(elem);
  55                 } else {
  56                         pthread_cond_wait(&queue->cv, &queue->mtx);
  57                 }
  58
  59                 pthread_mutex_unlock(&queue->mtx);
  60         }
  61
  62         return elem;
  63 }
  64
  65 void queue_cancel(Queue *queue)
  66 {
  67         queue->cancel = true;
  68
  69         pthread_mutex_lock(&queue->mtx);
  70         pthread_cond_broadcast(&queue->cv);
  71         pthread_mutex_unlock(&queue->mtx);
  72 }
  73
  74 void queue_finish(Queue *queue)
  75 {
  76         pthread_mutex_lock(&queue->mtx);
  77         queue->finish = true;
  78         pthread_mutex_unlock(&queue->mtx);
  79
  80         while (true) {
  81                 pthread_mutex_lock(&queue->mtx);
  82                 ListPair *first = queue->list.first;
  83                 pthread_mutex_unlock(&queue->mtx);
  84
  85                 if (first)
  86                         sched_yield();
  87                 else
  88                         break;
  89         }
  90 }