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 #include <time.h>
  38 #include <errno.h>
  39
  40 #include <tr_mq.h>
  41 #include <tr_debug.h>
  42
  43 /* Messages */
  44 static int tr_mq_msg_destructor(void *object)
  45 {
  46   TR_MQ_MSG *msg=talloc_get_type_abort(object, TR_MQ_MSG);
  47   if ( (msg->p!=NULL) && (msg->p_free!=NULL))
  48     msg->p_free(msg->p);
  49   return 0;
  50 }
  51
  52 TR_MQ_MSG *tr_mq_msg_new(TALLOC_CTX *mem_ctx, const char *message, TR_MQ_PRIORITY prio)
  53 {
  54   TR_MQ_MSG *msg=talloc(mem_ctx, TR_MQ_MSG);
  55   if (msg!=NULL) {
  56     msg->next=NULL;
  57     msg->prio=prio;
  58     msg->message=talloc_strdup(msg, message);
  59     if (msg->message==NULL) {
  60       talloc_free(msg);
  61       return NULL;
  62     }
  63     msg->p=NULL;
  64     talloc_set_destructor((void *)msg, tr_mq_msg_destructor);
  65   }
  66   return msg;
  67 }
  68
  69 void tr_mq_msg_free(TR_MQ_MSG *msg)
  70 {
  71   if (msg!=NULL)
  72     talloc_free(msg);
  73 }
  74
  75 TR_MQ_PRIORITY tr_mq_msg_get_prio(TR_MQ_MSG *msg)
  76 {
  77   return msg->prio;
  78 }
  79
  80 const char *tr_mq_msg_get_message(TR_MQ_MSG *msg)
  81 {
  82   return msg->message;
  83 }
  84
  85 void *tr_mq_msg_get_payload(TR_MQ_MSG *msg)
  86 {
  87   return msg->p;
  88 }
  89
  90 /* call with a pointer to the payload and a function to free it later */
  91 void tr_mq_msg_set_payload(TR_MQ_MSG *msg, void *p, void (*p_free)(void *))
  92 {
  93   msg->p=p;
  94   msg->p_free=p_free;
  95 }
  96
  97
  98 static TR_MQ_MSG *tr_mq_msg_get_next(TR_MQ_MSG *msg)
  99 {
 100   return msg->next;
 101 }
 102
 103 static void tr_mq_msg_set_next(TR_MQ_MSG *msg, TR_MQ_MSG *next)
 104 {
 105   msg->next=next;
 106 }
 107
 108 /* Message Queues */
 109 TR_MQ *tr_mq_new(TALLOC_CTX *mem_ctx)
 110 {
 111   TR_MQ *mq=talloc(mem_ctx, TR_MQ);
 112   pthread_condattr_t cattr;
 113
 114   if (mq!=NULL) {
 115     pthread_mutex_init(&(mq->mutex), 0);
 116     pthread_condattr_init(&cattr);
 117
 118     pthread_condattr_setclock(&cattr, CLOCK_MONOTONIC); /* use the monotonic clock for timeouts */
 119     pthread_cond_init(&(mq->have_msg_cond), &cattr);
 120     pthread_condattr_destroy(&cattr);
 121
 122     mq->head=NULL;
 123     mq->tail=NULL;
 124     mq->last_hi_prio=NULL;
 125
 126     mq->notify_cb=NULL;
 127     mq->notify_cb_arg=NULL;
 128   }
 129   return mq;
 130 }
 131
 132 void tr_mq_free(TR_MQ *mq)
 133 {
 134   if (mq!=NULL) {
 135     tr_mq_lock(mq); /* don't pull the rug out from under someone */
 136     talloc_free(mq);
 137   }
 138 }
 139
 140 int tr_mq_lock(TR_MQ *mq)
 141 {
 142   return pthread_mutex_lock(&(mq->mutex));
 143 }
 144
 145 int tr_mq_unlock(TR_MQ *mq)
 146 {
 147   return pthread_mutex_unlock(&(mq->mutex));
 148 }
 149
 150 static TR_MQ_MSG *tr_mq_get_head(TR_MQ *mq)
 151 {
 152   return mq->head;
 153 }
 154
 155 static void tr_mq_set_head(TR_MQ *mq, TR_MQ_MSG *msg)
 156 {
 157   mq->head=msg;
 158 }
 159
 160 static TR_MQ_MSG *tr_mq_get_tail(TR_MQ *mq)
 161 {
 162   return mq->tail;
 163 }
 164
 165 static void tr_mq_set_tail(TR_MQ *mq, TR_MQ_MSG *msg)
 166 {
 167   mq->tail=msg;
 168 }
 169
 170 void tr_mq_set_notify_cb(TR_MQ *mq, TR_MQ_NOTIFY_FN cb, void *arg)
 171 {
 172   mq->notify_cb=cb;
 173   mq->notify_cb_arg=arg;
 174 }
 175
 176 void tr_mq_clear(TR_MQ *mq)
 177 {
 178   TR_MQ_MSG *m=NULL;
 179   TR_MQ_MSG *n=NULL;
 180
 181   tr_mq_lock(mq);
 182   m=tr_mq_get_head(mq);
 183   while (m!=NULL) {
 184     n=tr_mq_msg_get_next(m);
 185     tr_mq_msg_free(m);
 186     m=n;
 187   }
 188   tr_mq_set_head(mq, NULL);
 189   tr_mq_set_tail(mq, NULL);
 190   tr_mq_unlock(mq);
 191 }
 192
 193 static int tr_mq_empty(TR_MQ *mq)
 194 {
 195   return tr_mq_get_head(mq)==NULL;
 196 }
 197
 198 /* puts msg in mq's talloc context */
 199 static void tr_mq_append(TR_MQ *mq, TR_MQ_MSG *msg)
 200 {
 201   if (tr_mq_get_head(mq)==NULL) {
 202     tr_mq_set_head(mq, msg);
 203     tr_mq_set_tail(mq, msg);
 204   } else {
 205     tr_mq_msg_set_next(tr_mq_get_tail(mq), msg); /* add to list */
 206     tr_mq_set_tail(mq, msg); /* update tail of list */
 207   }
 208   talloc_steal(mq, msg);
 209 }
 210
 211 static void tr_mq_append_high_prio(TR_MQ *mq, TR_MQ_MSG *new)
 212 {
 213   if (tr_mq_get_head(mq)==NULL) {
 214     tr_mq_set_head(mq, new);
 215     tr_mq_set_tail(mq, new);
 216   } else if (mq->last_hi_prio==NULL) {
 217     tr_mq_msg_set_next(new, tr_mq_get_head(mq)); /* add to front of list */
 218     tr_mq_set_head(mq, new); /* update head of list */
 219   } else {
 220     tr_mq_msg_set_next(new, tr_mq_msg_get_next(mq->last_hi_prio));
 221     tr_mq_msg_set_next(mq->last_hi_prio, new); /* add to end of hi prio msgs */
 222   }
 223   mq->last_hi_prio=new; /* in any case, this is now the last high priority msg */
 224   talloc_steal(mq,new);
 225 }
 226
 227 #define DEBUG_TR_MQ 0
 228 #if DEBUG_TR_MQ
 229 static void tr_mq_print(TR_MQ *mq)
 230 {
 231   TR_MQ_MSG *m=mq->head;
 232   int ii=0;
 233
 234   tr_debug("tr_mq_print: mq contents:");
 235   while(m!=NULL) {
 236     ii++;
 237     tr_debug("tr_mq_print: Entry %02d: %-15s (prio %d)",
 238              ii, tr_mq_msg_get_message(m), tr_mq_msg_get_prio(m));
 239     m=tr_mq_msg_get_next(m);
 240   }
 241 }
 242 #endif
 243 void tr_mq_add(TR_MQ *mq, TR_MQ_MSG *msg)
 244 {
 245   int was_empty=0;
 246   TR_MQ_NOTIFY_FN notify_cb=NULL;
 247   void *notify_cb_arg=NULL;
 248
 249   tr_mq_lock(mq);
 250
 251   was_empty=tr_mq_empty(mq);
 252   switch (tr_mq_msg_get_prio(msg)) {
 253   case TR_MQ_PRIO_HIGH:
 254     tr_mq_append_high_prio(mq, msg);
 255     break;
 256   default:
 257     tr_mq_append(mq, msg);
 258     break;
 259   }
 260   /* before releasing the mutex, get notify_cb data out of mq */
 261   notify_cb=mq->notify_cb;
 262   notify_cb_arg=mq->notify_cb_arg;
 263
 264 #if DEBUG_TR_MQ
 265   tr_mq_print(mq);
 266 #endif
 267
 268   /* Before releasing the lock, signal any waiting threads that there's now
 269    * something in the queue. Used for blocking tr_mq_pop() call. */
 270
 271   if (was_empty)
 272     pthread_cond_broadcast(&(mq->have_msg_cond));
 273
 274   tr_mq_unlock(mq);
 275
 276   /* see if we need to tell someone we became non-empty */
 277   if (was_empty && (notify_cb!=NULL))
 278     notify_cb(mq, notify_cb_arg);
 279 }
 280
 281 /* Compute an absolute time from a desired timeout interval for use with tr_mq_pop().
 282  * Fills in *ts and returns 0 on success. */
 283 int tr_mq_pop_timeout(time_t seconds, struct timespec *ts)
 284 {
 285   if (0!=clock_gettime(CLOCK_MONOTONIC, ts))
 286     return -1;
 287
 288   ts->tv_sec+=seconds;
 289   return 0;
 290 }
 291
 292 /* Retrieves a message from the queue, waiting until absolute
 293  * time ts_abort before giving up (using CLOCK_MONOTONIC). If ts_abort
 294  * has passed, returns an existing message but will not wait if one is
 295  * not already available. If ts_abort is null, no blocking.  Not
 296  * guaranteed to wait if an error occurs - immediately returns without
 297  * a message. Use tr_mq_pop_timeout() to get an absolute time that
 298  * is guaranteed compatible with this function.
 299  *
 300  * Caller should free msg via tr_mq_msg_free when done with it. It stays
 301  * in the TR_MQ's context, though, so use talloc_steal() if you want to do
 302  * something clever with it. */
 303 TR_MQ_MSG *tr_mq_pop(TR_MQ *mq, struct timespec *ts_abort)
 304 {
 305   TR_MQ_MSG *popped=NULL;
 306   int wait_err=0;
 307
 308   tr_mq_lock(mq);
 309   if ((tr_mq_get_head(mq)==NULL) && (ts_abort!=NULL)) {
 310     /* No msgs yet, and blocking was requested */
 311     while ((wait_err==0) && (NULL==tr_mq_get_head(mq)))
 312       wait_err=pthread_cond_timedwait(&(mq->have_msg_cond),
 313                                      &(mq->mutex),
 314                                      ts_abort);
 315
 316     if ((wait_err!=0) && (wait_err!=ETIMEDOUT)) {
 317       tr_notice("tr_mq_pop: error waiting for message.");
 318       return NULL;
 319     }
 320     /* if it timed out, ok to go ahead and check once more for a message, so no special exit */
 321   }
 322
 323   if (tr_mq_get_head(mq)!=NULL) {
 324     popped=tr_mq_get_head(mq);
 325     tr_mq_set_head(mq, tr_mq_msg_get_next(popped)); /* popped is the old head */
 326
 327     if (popped==mq->last_hi_prio)
 328       mq->last_hi_prio=NULL;
 329
 330     if (tr_mq_get_head(mq)==NULL)
 331       tr_mq_set_tail(mq, NULL); /* just popped the last element */
 332   }
 333   tr_mq_unlock(mq);
 334   if (popped!=NULL)
 335     tr_mq_msg_set_next(popped, NULL); /* disconnect from list */
 336   return popped;
 337 }
 338