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, const char *message, TR_MQ_PRIORITY prio)
  17 {
  18   TR_MQ_MSG *msg=talloc(mem_ctx, TR_MQ_MSG);
  19   if (msg!=NULL) {
  20     msg->next=NULL;
  21     msg->prio=prio;
  22     msg->message=talloc_strdup(msg, message);
  23     if (msg->message==NULL) {
  24       talloc_free(msg);
  25       return NULL;
  26     }
  27     msg->p=NULL;
  28     talloc_set_destructor((void *)msg, tr_mq_msg_destructor);
  29   }
  30   return msg;
  31 }
  32
  33 void tr_mq_msg_free(TR_MQ_MSG *msg)
  34 {
  35   if (msg!=NULL)
  36     talloc_free(msg);
  37 }
  38
  39 TR_MQ_PRIORITY tr_mq_msg_get_prio(TR_MQ_MSG *msg)
  40 {
  41   return msg->prio;
  42 }
  43
  44 const char *tr_mq_msg_get_message(TR_MQ_MSG *msg)
  45 {
  46   return msg->message;
  47 }
  48
  49 void *tr_mq_msg_get_payload(TR_MQ_MSG *msg)
  50 {
  51   return msg->p;
  52 }
  53
  54 /* call with a pointer to the payload and a function to free it later */
  55 void tr_mq_msg_set_payload(TR_MQ_MSG *msg, void *p, void (*p_free)(void *))
  56 {
  57   msg->p=p;
  58   msg->p_free=p_free;
  59 }
  60
  61
  62 static TR_MQ_MSG *tr_mq_msg_get_next(TR_MQ_MSG *msg)
  63 {
  64   return msg->next;
  65 }
  66
  67 static void tr_mq_msg_set_next(TR_MQ_MSG *msg, TR_MQ_MSG *next)
  68 {
  69   msg->next=next;
  70 }
  71
  72 /* Message Queues */
  73 TR_MQ *tr_mq_new(TALLOC_CTX *mem_ctx)
  74 {
  75   TR_MQ *mq=talloc(mem_ctx, TR_MQ);
  76   if (mq!=NULL) {
  77     pthread_mutex_init(&(mq->mutex), 0);
  78     mq->head=NULL;
  79     mq->tail=NULL;
  80     mq->last_hi_prio=NULL;
  81   }
  82   return mq;
  83 }
  84
  85 void tr_mq_free(TR_MQ *mq)
  86 {
  87   if (mq!=NULL) {
  88     tr_mq_lock(mq); /* don't pull the rug out from under someone */
  89     talloc_free(mq);
  90   }
  91 }
  92
  93 int tr_mq_lock(TR_MQ *mq)
  94 {
  95   return pthread_mutex_lock(&(mq->mutex));
  96 }
  97
  98 int tr_mq_unlock(TR_MQ *mq)
  99 {
 100   return pthread_mutex_unlock(&(mq->mutex));
 101 }
 102
 103 static TR_MQ_MSG *tr_mq_get_head(TR_MQ *mq)
 104 {
 105   return mq->head;
 106 }
 107
 108 static void tr_mq_set_head(TR_MQ *mq, TR_MQ_MSG *msg)
 109 {
 110   mq->head=msg;
 111 }
 112
 113 static TR_MQ_MSG *tr_mq_get_tail(TR_MQ *mq)
 114 {
 115   return mq->tail;
 116 }
 117
 118 static void tr_mq_set_tail(TR_MQ *mq, TR_MQ_MSG *msg)
 119 {
 120   mq->tail=msg;
 121 }
 122
 123 void tr_mq_set_notify_cb(TR_MQ *mq, TR_MQ_NOTIFY_FN cb, void *arg)
 124 {
 125   mq->notify_cb=cb;
 126   mq->notify_cb_arg=arg;
 127 }
 128
 129 void tr_mq_clear(TR_MQ *mq)
 130 {
 131   TR_MQ_MSG *m=NULL;
 132   TR_MQ_MSG *n=NULL;
 133
 134   tr_mq_lock(mq);
 135   m=tr_mq_get_head(mq);
 136   while (m!=NULL) {
 137     n=tr_mq_msg_get_next(m);
 138     tr_mq_msg_free(m);
 139     m=n;
 140   }
 141   tr_mq_set_head(mq, NULL);
 142   tr_mq_set_tail(mq, NULL);
 143   tr_mq_unlock(mq);
 144 }
 145
 146 static int tr_mq_empty(TR_MQ *mq)
 147 {
 148   return tr_mq_get_head(mq)==NULL;
 149 }
 150
 151 /* puts msg in mq's talloc context */
 152 static void tr_mq_append(TR_MQ *mq, TR_MQ_MSG *msg)
 153 {
 154   if (tr_mq_get_head(mq)==NULL) {
 155     tr_mq_set_head(mq, msg);
 156     tr_mq_set_tail(mq, msg);
 157   } else {
 158     tr_mq_msg_set_next(tr_mq_get_tail(mq), msg); /* add to list */
 159     tr_mq_set_tail(mq, msg); /* update tail of list */
 160   }
 161   talloc_steal(mq, msg);
 162 }
 163
 164 static void tr_mq_append_high_prio(TR_MQ *mq, TR_MQ_MSG *new)
 165 {
 166   if (tr_mq_get_head(mq)==NULL) {
 167     tr_mq_set_head(mq, new);
 168     tr_mq_set_tail(mq, new);
 169   } else if (mq->last_hi_prio==NULL) {
 170     tr_mq_msg_set_next(new, tr_mq_get_head(mq)); /* add to front of list */
 171     tr_mq_set_head(mq, new); /* update head of list */
 172   } else {
 173     tr_mq_msg_set_next(new, tr_mq_msg_get_next(mq->last_hi_prio));
 174     tr_mq_msg_set_next(mq->last_hi_prio, new); /* add to end of hi prio msgs */
 175   }
 176   mq->last_hi_prio=new; /* in any case, this is now the last high priority msg */
 177   talloc_steal(mq,new);
 178 }
 179
 180 #define DEBUG_TR_MQ 0
 181 #if DEBUG_TR_MQ
 182 static void tr_mq_print(TR_MQ *mq)
 183 {
 184   TR_MQ_MSG *m=mq->head;
 185   int ii=0;
 186
 187   tr_debug("tr_mq_print: mq contents:");
 188   while(m!=NULL) {
 189     ii++;
 190     tr_debug("tr_mq_print: Entry %02d: %-15s (prio %d)",
 191              ii, tr_mq_msg_get_message(m), tr_mq_msg_get_prio(m));
 192     m=tr_mq_msg_get_next(m);
 193   }
 194 }
 195 #endif
 196 void tr_mq_add(TR_MQ *mq, TR_MQ_MSG *msg)
 197 {
 198   int was_empty=0;
 199   TR_MQ_NOTIFY_FN notify_cb=NULL;
 200   void *notify_cb_arg=NULL;
 201
 202   tr_mq_lock(mq);
 203
 204   was_empty=tr_mq_empty(mq);
 205   switch (tr_mq_msg_get_prio(msg)) {
 206   case TR_MQ_PRIO_HIGH:
 207     tr_mq_append_high_prio(mq, msg);
 208     break;
 209   default:
 210     tr_mq_append(mq, msg);
 211     break;
 212   }
 213   /* before releasing the mutex, get notify_cb data out of mq */
 214   notify_cb=mq->notify_cb;
 215   notify_cb_arg=mq->notify_cb_arg;
 216
 217 #if DEBUG_TR_MQ
 218   tr_mq_print(mq);
 219 #endif
 220
 221   tr_mq_unlock(mq);
 222
 223   /* see if we need to tell someone we became non-empty */
 224   if (was_empty && (notify_cb!=NULL))
 225     notify_cb(mq, notify_cb_arg);
 226 }
 227
 228 /* caller must free msg via tr_mq_msg_free */
 229 TR_MQ_MSG *tr_mq_pop(TR_MQ *mq)
 230 {
 231   TR_MQ_MSG *popped=NULL;
 232
 233   tr_mq_lock(mq);
 234   if (tr_mq_get_head(mq)!=NULL) {
 235     popped=tr_mq_get_head(mq);
 236     tr_mq_set_head(mq, tr_mq_msg_get_next(popped)); /* popped is the old head */
 237
 238     if (popped==mq->last_hi_prio)
 239       mq->last_hi_prio=NULL;
 240
 241     if (tr_mq_get_head(mq)==NULL)
 242       tr_mq_set_tail(mq, NULL); /* just popped the last element */
 243   }
 244   tr_mq_unlock(mq);
 245   if (popped!=NULL)
 246     tr_mq_msg_set_next(popped, NULL); /* disconnect from list */
 247   return popped;
 248 }
 249