common/tr_mq.c

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