src/queue_mqueue.c

   1 /*======================================================================*
   2  * Copyright (c) 2008, Yahoo! Inc. All rights reserved.                 *
   3  * Copyright (c) 2010-2016, OpenX Inc.   All rights reserved.           *
   4  *                                                                      *
   5  * Licensed under the New BSD License (the "License"); you may not use  *
   6  * this file except in compliance with the License.  Unless required    *
   7  * by applicable law or agreed to in writing, software distributed      *
   8  * under the License is distributed on an "AS IS" BASIS, WITHOUT        *
   9  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.     *
  10  * See the License for the specific language governing permissions and  *
  11  * limitations under the License. See accompanying LICENSE file.        *
  12  *======================================================================*/
  13
  14 #include "config.h"
  15
  16 #include "queue.h"
  17 #include "queue_mqueue.h"
  18
  19 #include "perror.h"
  20 #include "opt.h"
  21 #include "sig.h"
  22
  23 #if defined(HAVE_MQUEUE_H)
  24
  25 #include <sys/types.h>  /* Must be included before mqueue.h on FreeBSD 4.9. */
  26 #include <mqueue.h>
  27 #include <sys/time.h>
  28 #include <sys/resource.h>
  29 #include <fcntl.h>
  30 #include <limits.h>
  31 #include <stdio.h>
  32 #include <stdlib.h>
  33 #include <string.h>
  34 #include <time.h>
  35
  36 /* The following is for Solaris. */
  37 #if !defined(MQ_PRIO_MAX) && defined(_POSIX_MQ_PRIO_MAX)
  38 #define MQ_PRIO_MAX _POSIX_MQ_PRIO_MAX
  39 #endif
  40
  41 struct priv {
  42   char*         path;
  43   mqd_t         mq;
  44   size_t        max_sz;
  45   size_t        max_cnt;
  46 };
  47
  48 static void destructor (struct queue* this_queue)
  49 {
  50   struct priv* ppriv = (struct priv*)this_queue->priv;
  51
  52   free (ppriv->path);
  53   free (ppriv);
  54
  55   this_queue->vtbl = 0;
  56   this_queue->priv = 0;
  57 }
  58
  59 static int xopen (struct queue* this_queue, int flags)
  60 {
  61   struct priv* ppriv = (struct priv*)this_queue->priv;
  62
  63   if ( (mqd_t)-1 == ppriv->mq )
  64     {
  65       struct mq_attr attr;
  66       memset(&attr, 0, sizeof(attr));
  67       attr.mq_maxmsg = ppriv->max_cnt;
  68       attr.mq_msgsize = ppriv->max_sz;
  69
  70       errno = 0 ;
  71       /* see http://www.die.net/doc/linux/man/man7/mq_overview.7.html */
  72       if ( (mqd_t)-1 == (ppriv->mq = mq_open (ppriv->path,
  73                                               flags | O_CREAT, 0666, &attr)) )
  74         {
  75           /* Try resetting the mqueue resource parameters before formally
  76            * failing see http://www.die.net/doc/linux/man/man2/getrlimit.2.html
  77            */
  78           struct rlimit rlim;
  79           PERROR("mq_open");
  80           errno = 0;
  81           attr.mq_maxmsg += 1;
  82           rlim.rlim_cur = RLIM_INFINITY; /* attr.mq_maxmsg * sizeof(struct msg_msg *) + attr.mq_maxmsg * attr.mq_msgsize ; */
  83           rlim.rlim_max = RLIM_INFINITY ; /*rlim.rlim_cur ; */
  84           if ( setrlimit (RLIMIT_MSGQUEUE, &rlim) )
  85             {
  86               PERROR("setrlimit");
  87             }
  88           attr.mq_maxmsg -= 1 ;
  89
  90           if ( (mqd_t)-1 == (ppriv->mq = mq_open (ppriv->path,
  91                                                   flags | O_CREAT,
  92                                                   0666, &attr)) )
  93             {
  94               LOG_ER("mq_open(path=\"%s\", flags=0x%04x, 0666, "
  95                      "{attr.mq_maxmsg=%i, attr.mq_msgsize=%i})\n",
  96                      ppriv->path, flags | O_CREAT, attr.mq_maxmsg,
  97                      attr.mq_msgsize);
  98               return -1;
  99             }
 100         }
 101     }
 102   return 0;
 103 }
 104
 105 static int xclose (struct queue* this_queue)
 106 {
 107   struct priv* ppriv = (struct priv*)this_queue->priv;
 108
 109   if ( (mqd_t)-1 != ppriv->mq )
 110     {
 111       if ( (mq_close (ppriv->mq)) < 0 )
 112         {
 113           PERROR("mq_close");
 114           return -1;
 115         }
 116     }
 117
 118   ppriv->mq = (mqd_t)-1;
 119
 120   return 0;
 121 }
 122
 123 static int xread (struct queue* this_queue, void* buf,
 124                   size_t count, int* pending)
 125 {
 126   struct priv* ppriv = (struct priv*)this_queue->priv;
 127   struct mq_attr attr;
 128   int mq_rec_rtrn;
 129   unsigned int pri;
 130
 131   if ( 0 == buf )
 132     {
 133       LOG_ER("queue read with NULL buf pointer\n");
 134       return -1;
 135     }
 136   if ( (mqd_t)-1 == ppriv->mq )
 137     {
 138       LOG_ER("queue read with queue closed.\n");
 139       return -1;
 140     }
 141
 142   LOG_PROG("about to call mq_receive().  gbl_done=%d\n", gbl_done);
 143
 144   /* use blocking reads unless we're shutting down. */
 145   struct timespec time_buf = { time(NULL), arg_wakup_interval_ms * 1000000 };
 146   mq_rec_rtrn = mq_timedreceive (ppriv->mq, buf, count, &pri, &time_buf);
 147
 148   if (mq_rec_rtrn < 0 )
 149     {
 150       LOG_PROG("errno: %d %s\n", errno, strerror(errno));
 151       switch ( errno )
 152         {
 153         default:
 154           PERROR("mq_receive");
 155
 156           /* If we've been interrupted it's likely that we're shutting
 157              down, so no need to print errors. */
 158         case ETIMEDOUT:
 159         case EINTR:
 160           return mq_rec_rtrn;
 161         }
 162     }
 163
 164   if ( MQ_PRIO_MAX - 1 != pri )
 165     {
 166       LOG_ER("queue read returned message with unrecognized "
 167              "priority (pri=%d).\n", pri);
 168     }
 169
 170   LOG_PROG("mq_receive() returned %d.\n", mq_rec_rtrn);
 171
 172   if ( mq_getattr (ppriv->mq, &attr) < 0 )
 173     {
 174       PERROR("mq_getattr");
 175       return -1;
 176     }
 177
 178   *pending = attr.mq_curmsgs;
 179
 180   return mq_rec_rtrn;
 181 }
 182
 183 static int xwrite (struct queue* this_queue, const void* buf, size_t count)
 184 {
 185   struct priv* ppriv = (struct priv*)this_queue->priv;
 186
 187   if ( 0 == buf )
 188     {
 189       LOG_ER("queue write with NULL buf pointer\n");
 190       return -1;
 191     }
 192   if ( (mqd_t)-1 == ppriv->mq )
 193     {
 194       LOG_ER("queue write with queue closed.\n");
 195       return -1;
 196     }
 197
 198   LOG_PROG("about to call mq_send().\n");
 199
 200   if ( 0 != mq_send (ppriv->mq, buf, count, MQ_PRIO_MAX-1) )
 201     {
 202       PERROR("mq_send");
 203       return -1;
 204     }
 205
 206   return 0;
 207 }
 208
 209 static void* alloc (struct queue* this_queue, size_t* newcount)
 210 {
 211   void *data = malloc (*newcount = ((struct priv*)this_queue->priv)->max_sz);
 212   memset(data, 0, *newcount);
 213   return data;
 214 }
 215
 216 static void dealloc (struct queue* this_queue, void* buf)
 217 {
 218   (void) this_queue; /* appease -Wall -Werror */
 219   free (buf);
 220 }
 221
 222 int queue_mqueue_ctor (struct queue* this_queue,
 223                        const char*   path,
 224                        size_t        max_sz,
 225                        size_t        max_cnt)
 226 {
 227   static struct queue_vtbl vtbl = {
 228       destructor,
 229       xopen, xclose,
 230       xread, xwrite,
 231       alloc, dealloc
 232   };
 233
 234   struct priv* ppriv;
 235
 236   this_queue->vtbl = 0;
 237   this_queue->priv = 0;
 238
 239   ppriv = (struct priv*)malloc(sizeof(struct priv));
 240   if ( 0 == ppriv )
 241     {
 242       LOG_ER("malloc failed attempting to allocate %d bytes\n",
 243              sizeof(*ppriv));
 244       return -1;
 245     }
 246   memset(ppriv, 0, sizeof(*ppriv));
 247
 248   if ( 0 == (ppriv->path = strdup(path)) )
 249     {
 250       LOG_ER("strdup failed attempting to dup \"%s\"\n",
 251              path);
 252       free(ppriv);
 253       return -1;
 254     }
 255
 256   ppriv->mq = (mqd_t)-1;
 257   ppriv->max_sz = max_sz;
 258   ppriv->max_cnt = max_cnt;
 259
 260   this_queue->vtbl = &vtbl;
 261   this_queue->priv = ppriv;
 262
 263   return 0;
 264 }
 265
 266 #else  /* if defined(HAVE_MQUEUE_H) */
 267
 268 int queue_mqueue_ctor (struct queue* this_queue,
 269                        const char*   path,
 270                        size_t        max_sz,
 271                        size_t        max_cnt)
 272 {
 273   this_queue->vtbl = 0;
 274   this_queue->priv = 0;
 275   (void)path;     /* appease -Wall -Werror */
 276   (void)max_sz;   /* appease -Wall -Werror */
 277   (void)max_cnt;  /* appease -Wall -Werror */
 278
 279   return -1;
 280 }
 281
 282 #endif /* HAVE_MQUEUE_H */