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