common/tr_mq.c

   1 #include <talloc.h>
   2 #include <pthread.h>
   3
   4 #include <tr_mq.h>
   5 #include <tr_debug.h>
   6
   7 /* Messages */
   8 static int tr_mq_msg_destructor(void *object)
   9 {
  10   TR_MQ_MSG *msg=talloc_get_type_abort(object, TR_MQ_MSG);
  11   if ( (msg->p!=NULL) && (msg->p_free!=NULL))
  12     msg->p_free(msg->p);
  13   return 0;
  14 }
  15
  16 TR_MQ_MSG *tr_mq_msg_new(TALLOC_CTX *mem_ctx)
  17 {
  18   TR_MQ_MSG *msg=talloc(mem_ctx, TR_MQ_MSG);
  19   if (msg!=NULL) {
  20     msg->next=NULL;
  21     msg->p=NULL;
  22     talloc_set_destructor((void *)msg, tr_mq_msg_destructor);
  23   }
  24   return msg;
  25 }
  26
  27 void tr_mq_msg_free(TR_MQ_MSG *msg)
  28 {
  29   if (msg!=NULL)
  30     talloc_free(msg);
  31 }
  32
  33 void *tr_mq_msg_get_payload(TR_MQ_MSG *msg)
  34 {
  35   return msg->p;
  36 }
  37
  38 /* call with a pointer to the payload and a function to free it later */
  39 void tr_mq_msg_set_payload(TR_MQ_MSG *msg, void *p, void (*p_free)(void *))
  40 {
  41   msg->p=p;
  42   msg->p_free=p_free;
  43 }
  44
  45
  46 static TR_MQ_MSG *tr_mq_msg_get_next(TR_MQ_MSG *msg)
  47 {
  48   return msg->next;
  49 }
  50
  51 static void tr_mq_msg_set_next(TR_MQ_MSG *msg, TR_MQ_MSG *next)
  52 {
  53   msg->next=next;
  54 }
  55
  56 /* Message Queues */
  57 TR_MQ *tr_mq_new(TALLOC_CTX *mem_ctx)
  58 {
  59   TR_MQ *mq=talloc(mem_ctx, TR_MQ);
  60   if (mq!=NULL) {
  61     pthread_mutex_init(&(mq->mutex), 0);
  62     mq->head=NULL;
  63     mq->tail=NULL;
  64   }
  65   return mq;
  66 }
  67
  68 void tr_mq_free(TR_MQ *mq)
  69 {
  70   if (mq!=NULL) {
  71     tr_mq_lock(mq); /* don't pull the rug out from under someone */
  72     talloc_free(mq);
  73   }
  74 }
  75
  76 int tr_mq_lock(TR_MQ *mq)
  77 {
  78   return pthread_mutex_lock(&(mq->mutex));
  79 }
  80
  81 int tr_mq_unlock(TR_MQ *mq)
  82 {
  83   return pthread_mutex_unlock(&(mq->mutex));
  84 }
  85
  86 static TR_MQ_MSG *tr_mq_get_head(TR_MQ *mq)
  87 {
  88   return mq->head;
  89 }
  90
  91 static void tr_mq_set_head(TR_MQ *mq, TR_MQ_MSG *msg)
  92 {
  93   mq->head=msg;
  94 }
  95
  96 static TR_MQ_MSG *tr_mq_get_tail(TR_MQ *mq)
  97 {
  98   return mq->tail;
  99 }
 100
 101 static void tr_mq_set_tail(TR_MQ *mq, TR_MQ_MSG *msg)
 102 {
 103   mq->tail=msg;
 104 }
 105
 106 void tr_mq_set_notify_cb(TR_MQ *mq, TR_MQ_NOTIFY_FN cb, void *arg)
 107 {
 108   mq->notify_cb=cb;
 109   mq->notify_cb_arg=arg;
 110 }
 111
 112
 113 /* puts msg in mq's talloc context */
 114 void tr_mq_append(TR_MQ *mq, TR_MQ_MSG *msg)
 115 {
 116   int was_empty=0;
 117   TR_MQ_NOTIFY_FN notify_cb=NULL;
 118   void *notify_cb_arg=NULL;
 119
 120   tr_mq_lock(mq);
 121   if (tr_mq_get_head(mq)==NULL) {
 122     was_empty=1;
 123     tr_mq_set_head(mq, msg);
 124     tr_mq_set_tail(mq, msg);
 125   } else {
 126     tr_mq_msg_set_next(tr_mq_get_tail(mq), msg); /* add to list */
 127     tr_mq_set_tail(mq, msg); /* update tail of list */
 128   }
 129   talloc_steal(mq, msg);
 130   /* before releasing the mutex, get notify_cb data out of mq */
 131   notify_cb=mq->notify_cb;
 132   notify_cb_arg=mq->notify_cb_arg;
 133   tr_mq_unlock(mq);
 134
 135   /* see if we need to tell someone we became non-empty */
 136   if (was_empty && (notify_cb!=NULL))
 137     notify_cb(mq, notify_cb_arg);
 138 }
 139
 140 /* caller must free msg via tr_mq_msg_free */
 141 TR_MQ_MSG *tr_mq_pop(TR_MQ *mq)
 142 {
 143   TR_MQ_MSG *popped=NULL;
 144
 145   tr_mq_lock(mq);
 146   if (tr_mq_get_head(mq)!=NULL) {
 147     popped=tr_mq_get_head(mq);
 148     tr_mq_set_head(mq, tr_mq_msg_get_next(popped)); /* popped is the old head */
 149     if (tr_mq_get_head(mq)==NULL)
 150       tr_mq_set_tail(mq, NULL); /* just popped the last element */
 151   }
 152   tr_mq_unlock(mq);
 153   if (popped!=NULL)
 154     tr_mq_msg_set_next(popped, NULL); /* disconnect from list */
 155   return popped;
 156 }