common/tr_mq.c

   1 /*
   2  * Copyright (c) 2016, JANET(UK)
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  *
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  *
  12  * 2. Redistributions in binary form must reproduce the above copyright
  13  *    notice, this list of conditions and the following disclaimer in the
  14  *    documentation and/or other materials provided with the distribution.
  15  *
  16  * 3. Neither the name of JANET(UK) nor the names of its contributors
  17  *    may be used to endorse or promote products derived from this software
  18  *    without specific prior written permission.
  19  *
  20  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  21  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  22  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
  23  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
  24  * COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
  25  * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  26  * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  27  * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  28  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
  29  * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  30  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
  31  * OF THE POSSIBILITY OF SUCH DAMAGE.
  32  *
  33  */
  34
  35 #include <talloc.h>
  36 #include <pthread.h>
  37
  38 #include <tr_mq.h>
  39 #include <tr_debug.h>
  40
  41 /* Messages */
  42 static int tr_mq_msg_destructor(void *object)
  43 {
  44   TR_MQ_MSG *msg=talloc_get_type_abort(object, TR_MQ_MSG);
  45   if ( (msg->p!=NULL) && (msg->p_free!=NULL))
  46     msg->p_free(msg->p);
  47   return 0;
  48 }
  49
  50 TR_MQ_MSG *tr_mq_msg_new(TALLOC_CTX *mem_ctx, const char *message, TR_MQ_PRIORITY prio)
  51 {
  52   TR_MQ_MSG *msg=talloc(mem_ctx, TR_MQ_MSG);
  53   if (msg!=NULL) {
  54     msg->next=NULL;
  55     msg->prio=prio;
  56     msg->message=talloc_strdup(msg, message);
  57     if (msg->message==NULL) {
  58       talloc_free(msg);
  59       return NULL;
  60     }
  61     msg->p=NULL;
  62     talloc_set_destructor((void *)msg, tr_mq_msg_destructor);
  63   }
  64   return msg;
  65 }
  66
  67 void tr_mq_msg_free(TR_MQ_MSG *msg)
  68 {
  69   if (msg!=NULL)
  70     talloc_free(msg);
  71 }
  72
  73 TR_MQ_PRIORITY tr_mq_msg_get_prio(TR_MQ_MSG *msg)
  74 {
  75   return msg->prio;
  76 }
  77
  78 const char *tr_mq_msg_get_message(TR_MQ_MSG *msg)
  79 {
  80   return msg->message;
  81 }
  82
  83 void *tr_mq_msg_get_payload(TR_MQ_MSG *msg)
  84 {
  85   return msg->p;
  86 }
  87
  88 /* call with a pointer to the payload and a function to free it later */
  89 void tr_mq_msg_set_payload(TR_MQ_MSG *msg, void *p, void (*p_free)(void *))
  90 {
  91   msg->p=p;
  92   msg->p_free=p_free;
  93 }
  94
  95
  96 static TR_MQ_MSG *tr_mq_msg_get_next(TR_MQ_MSG *msg)
  97 {
  98   return msg->next;
  99 }
 100
 101 static void tr_mq_msg_set_next(TR_MQ_MSG *msg, TR_MQ_MSG *next)
 102 {
 103   msg->next=next;
 104 }
 105
 106 /* Message Queues */
 107 TR_MQ *tr_mq_new(TALLOC_CTX *mem_ctx)
 108 {
 109   TR_MQ *mq=talloc(mem_ctx, TR_MQ);
 110   if (mq!=NULL) {
 111     pthread_mutex_init(&(mq->mutex), 0);
 112     mq->head=NULL;
 113     mq->tail=NULL;
 114     mq->last_hi_prio=NULL;
 115   }
 116   return mq;
 117 }
 118
 119 void tr_mq_free(TR_MQ *mq)
 120 {
 121   if (mq!=NULL) {
 122     tr_mq_lock(mq); /* don't pull the rug out from under someone */
 123     talloc_free(mq);
 124   }
 125 }
 126
 127 int tr_mq_lock(TR_MQ *mq)
 128 {
 129   return pthread_mutex_lock(&(mq->mutex));
 130 }
 131
 132 int tr_mq_unlock(TR_MQ *mq)
 133 {
 134   return pthread_mutex_unlock(&(mq->mutex));
 135 }
 136
 137 static TR_MQ_MSG *tr_mq_get_head(TR_MQ *mq)
 138 {
 139   return mq->head;
 140 }
 141
 142 static void tr_mq_set_head(TR_MQ *mq, TR_MQ_MSG *msg)
 143 {
 144   mq->head=msg;
 145 }
 146
 147 static TR_MQ_MSG *tr_mq_get_tail(TR_MQ *mq)
 148 {
 149   return mq->tail;
 150 }
 151
 152 static void tr_mq_set_tail(TR_MQ *mq, TR_MQ_MSG *msg)
 153 {
 154   mq->tail=msg;
 155 }
 156
 157 void tr_mq_set_notify_cb(TR_MQ *mq, TR_MQ_NOTIFY_FN cb, void *arg)
 158 {
 159   mq->notify_cb=cb;
 160   mq->notify_cb_arg=arg;
 161 }
 162
 163 void tr_mq_clear(TR_MQ *mq)
 164 {
 165   TR_MQ_MSG *m=NULL;
 166   TR_MQ_MSG *n=NULL;
 167
 168   tr_mq_lock(mq);
 169   m=tr_mq_get_head(mq);
 170   while (m!=NULL) {
 171     n=tr_mq_msg_get_next(m);
 172     tr_mq_msg_free(m);
 173     m=n;
 174   }
 175   tr_mq_set_head(mq, NULL);
 176   tr_mq_set_tail(mq, NULL);
 177   tr_mq_unlock(mq);
 178 }
 179
 180 static int tr_mq_empty(TR_MQ *mq)
 181 {
 182   return tr_mq_get_head(mq)==NULL;
 183 }
 184
 185 /* puts msg in mq's talloc context */
 186 static void tr_mq_append(TR_MQ *mq, TR_MQ_MSG *msg)
 187 {
 188   if (tr_mq_get_head(mq)==NULL) {
 189     tr_mq_set_head(mq, msg);
 190     tr_mq_set_tail(mq, msg);
 191   } else {
 192     tr_mq_msg_set_next(tr_mq_get_tail(mq), msg); /* add to list */
 193     tr_mq_set_tail(mq, msg); /* update tail of list */
 194   }
 195   talloc_steal(mq, msg);
 196 }
 197
 198 static void tr_mq_append_high_prio(TR_MQ *mq, TR_MQ_MSG *new)
 199 {
 200   if (tr_mq_get_head(mq)==NULL) {
 201     tr_mq_set_head(mq, new);
 202     tr_mq_set_tail(mq, new);
 203   } else if (mq->last_hi_prio==NULL) {
 204     tr_mq_msg_set_next(new, tr_mq_get_head(mq)); /* add to front of list */
 205     tr_mq_set_head(mq, new); /* update head of list */
 206   } else {
 207     tr_mq_msg_set_next(new, tr_mq_msg_get_next(mq->last_hi_prio));
 208     tr_mq_msg_set_next(mq->last_hi_prio, new); /* add to end of hi prio msgs */
 209   }
 210   mq->last_hi_prio=new; /* in any case, this is now the last high priority msg */
 211   talloc_steal(mq,new);
 212 }
 213
 214 #define DEBUG_TR_MQ 0
 215 #if DEBUG_TR_MQ
 216 static void tr_mq_print(TR_MQ *mq)
 217 {
 218   TR_MQ_MSG *m=mq->head;
 219   int ii=0;
 220
 221   tr_debug("tr_mq_print: mq contents:");
 222   while(m!=NULL) {
 223     ii++;
 224     tr_debug("tr_mq_print: Entry %02d: %-15s (prio %d)",
 225              ii, tr_mq_msg_get_message(m), tr_mq_msg_get_prio(m));
 226     m=tr_mq_msg_get_next(m);
 227   }
 228 }
 229 #endif
 230 void tr_mq_add(TR_MQ *mq, TR_MQ_MSG *msg)
 231 {
 232   int was_empty=0;
 233   TR_MQ_NOTIFY_FN notify_cb=NULL;
 234   void *notify_cb_arg=NULL;
 235
 236   tr_mq_lock(mq);
 237
 238   was_empty=tr_mq_empty(mq);
 239   switch (tr_mq_msg_get_prio(msg)) {
 240   case TR_MQ_PRIO_HIGH:
 241     tr_mq_append_high_prio(mq, msg);
 242     break;
 243   default:
 244     tr_mq_append(mq, msg);
 245     break;
 246   }
 247   /* before releasing the mutex, get notify_cb data out of mq */
 248   notify_cb=mq->notify_cb;
 249   notify_cb_arg=mq->notify_cb_arg;
 250
 251 #if DEBUG_TR_MQ
 252   tr_mq_print(mq);
 253 #endif
 254
 255   tr_mq_unlock(mq);
 256
 257   /* see if we need to tell someone we became non-empty */
 258   if (was_empty && (notify_cb!=NULL))
 259     notify_cb(mq, notify_cb_arg);
 260 }
 261
 262 /* caller must free msg via tr_mq_msg_free */
 263 TR_MQ_MSG *tr_mq_pop(TR_MQ *mq)
 264 {
 265   TR_MQ_MSG *popped=NULL;
 266
 267   tr_mq_lock(mq);
 268   if (tr_mq_get_head(mq)!=NULL) {
 269     popped=tr_mq_get_head(mq);
 270     tr_mq_set_head(mq, tr_mq_msg_get_next(popped)); /* popped is the old head */
 271
 272     if (popped==mq->last_hi_prio)
 273       mq->last_hi_prio=NULL;
 274
 275     if (tr_mq_get_head(mq)==NULL)
 276       tr_mq_set_tail(mq, NULL); /* just popped the last element */
 277   }
 278   tr_mq_unlock(mq);
 279   if (popped!=NULL)
 280     tr_mq_msg_set_next(popped, NULL); /* disconnect from list */
 281   return popped;
 282 }
 283