queue.c

   1 #include <sched.h>         // for sched_yield
   2 #include "bits/callback.h" // for Transformer
   3 #include "queue.h"
   4
   5 void queue_ini(Queue *queue)
   6 {
   7         list_ini(&queue->lst);
   8         queue->cnl = queue->fin = 0;
   9         pthread_cond_init(&queue->cnd, NULL);
  10         pthread_mutex_init(&queue->mtx, NULL);
  11 }
  12
  13 void queue_dst(Queue *queue)
  14 {
  15         pthread_cond_destroy(&queue->cnd);
  16         pthread_mutex_destroy(&queue->mtx);
  17 }
  18
  19 void queue_clr(Queue *queue, void *iter, void *arg, void *trans)
  20 {
  21         list_clr(&queue->lst, iter, arg, trans);
  22 }
  23
  24 #define ENQUEUE(queue_fun, list_fun) \
  25         bool queue_fun(Queue *queue, void *dat) \
  26         { \
  27                 bool success = false; \
  28  \
  29                 pthread_mutex_lock(&queue->mtx); \
  30                 if (!queue->fin) { \
  31                         success = true; \
  32                         list_fun(&queue->lst, dat); \
  33                         pthread_cond_signal(&queue->cnd); \
  34                 } \
  35                 pthread_mutex_unlock(&queue->mtx); \
  36  \
  37                 return success; \
  38         }
  39
  40 ENQUEUE(queue_enq, list_apd)
  41 ENQUEUE(queue_ppd, list_ppd)
  42
  43 #undef ENQUEUE
  44
  45 void *queue_deq(Queue *queue, void *trans)
  46 {
  47         void *dat = NULL;
  48
  49         pthread_mutex_lock(&queue->mtx);
  50         while (!queue->cnl && !dat) {
  51                 ListNode **node = &queue->lst.fst;
  52                 if (*node) {
  53                         dat = (*node)->dat;
  54                         list_nrm(&queue->lst, node);
  55
  56                         if (trans)
  57                                 dat = ((Transformer) trans)(dat);
  58                 } else {
  59                         pthread_cond_wait(&queue->cnd, &queue->mtx);
  60                 }
  61         }
  62         pthread_mutex_unlock(&queue->mtx);
  63
  64         return dat;
  65 }
  66
  67 void queue_cnl(Queue *queue)
  68 {
  69         pthread_mutex_lock(&queue->mtx);
  70         queue->cnl = 1;
  71         pthread_cond_broadcast(&queue->cnd);
  72         pthread_mutex_unlock(&queue->mtx);
  73 }
  74
  75 void queue_fin(Queue *queue)
  76 {
  77         pthread_mutex_lock(&queue->mtx);
  78         queue->fin = 1;
  79         pthread_mutex_unlock(&queue->mtx);
  80
  81         for (;;) {
  82                 pthread_mutex_lock(&queue->mtx);
  83                 ListNode *node = queue->lst.fst;
  84                 pthread_mutex_unlock(&queue->mtx);
  85
  86                 if (node)
  87                         sched_yield();
  88                 else
  89                         break;
  90         }
  91 }