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 bool queue_enq(Queue *queue, void *dat)
  25 {
  26         bool success = false;
  27
  28         pthread_mutex_lock(&queue->mtx);
  29         if (!queue->fin) {
  30                 success = true;
  31                 list_apd(&queue->lst, dat);
  32                 pthread_cond_signal(&queue->cnd);
  33         }
  34         pthread_mutex_unlock(&queue->mtx);
  35
  36         return success;
  37 }
  38
  39 void *queue_deq(Queue *queue, void *trans)
  40 {
  41         void *dat = NULL;
  42
  43         pthread_mutex_lock(&queue->mtx);
  44         while (!queue->cnl && !dat) {
  45                 ListNode **node = &queue->lst.fst;
  46                 if (*node) {
  47                         dat = (*node)->dat;
  48                         list_nrm(&queue->lst, node);
  49
  50                         if (trans)
  51                                 dat = ((Transformer) trans)(dat);
  52                 } else {
  53                         pthread_cond_wait(&queue->cnd, &queue->mtx);
  54                 }
  55         }
  56         pthread_mutex_unlock(&queue->mtx);
  57
  58         return dat;
  59 }
  60
  61 void queue_cnl(Queue *queue)
  62 {
  63         pthread_mutex_lock(&queue->mtx);
  64         queue->cnl = 1;
  65         pthread_cond_broadcast(&queue->cnd);
  66         pthread_mutex_unlock(&queue->mtx);
  67 }
  68
  69 void queue_fin(Queue *queue)
  70 {
  71         pthread_mutex_lock(&queue->mtx);
  72         queue->fin = 1;
  73         pthread_mutex_unlock(&queue->mtx);
  74
  75         for (;;) {
  76                 pthread_mutex_lock(&queue->mtx);
  77                 ListNode *node = queue->lst.fst;
  78                 pthread_mutex_unlock(&queue->mtx);
  79
  80                 if (node)
  81                         sched_yield();
  82                 else
  83                         break;
  84         }
  85 }