src/queue_mqueue.c

   1 /*======================================================================*
   2  * Copyright (C) 2008 Light Weight Event System                         *
   3  * All rights reserved.                                                 *
   4  *                                                                      *
   5  * This program is free software; you can redistribute it and/or modify *
   6  * it under the terms of the GNU General Public License as published by *
   7  * the Free Software Foundation; either version 2 of the License, or    *
   8  * (at your option) any later version.                                  *
   9  *                                                                      *
  10  * This program is distributed in the hope that it will be useful,      *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
  13  * GNU General Public License for more details.                         *
  14  *                                                                      *
  15  * You should have received a copy of the GNU General Public License    *
  16  * along with this program; if not, write to the Free Software          *
  17  * Foundation, Inc., 51 Franklin Street, Fifth Floor,                   *
  18  * Boston, MA 02110-1301 USA.                                           *
  19  *======================================================================*/
  20
  21 #include "config.h"
  22
  23 #include "queue.h"
  24 #include "queue_mqueue.h"
  25
  26 #include "perror.h"
  27 #include "opt.h"
  28
  29 #if defined(HAVE_MQUEUE_H)
  30
  31 #include <sys/types.h>  /* Must be included before mqueue.h on FreeBSD 4.9. */
  32 #include <mqueue.h>
  33 #include <sys/time.h>
  34 #include <sys/resource.h>
  35 #include <fcntl.h>
  36 #include <limits.h>
  37 #include <stdio.h>
  38 #include <stdlib.h>
  39 #include <string.h>
  40
  41 /* The following is for Solaris. */
  42 #if !defined(MQ_PRIO_MAX) && defined(_POSIX_MQ_PRIO_MAX)
  43 #define MQ_PRIO_MAX _POSIX_MQ_PRIO_MAX
  44 #endif
  45
  46 struct priv {
  47   char*         path;
  48   mqd_t         mq;
  49   size_t        max_sz;
  50   size_t        max_cnt;
  51 };
  52
  53 static void destructor (struct queue* this_queue)
  54 {
  55   struct priv* ppriv = (struct priv*)this_queue->priv;
  56
  57   free (ppriv->path);
  58   free (ppriv);
  59
  60   this_queue->vtbl = 0;
  61   this_queue->priv = 0;
  62 }
  63
  64 static int xopen (struct queue* this_queue, int flags)
  65 {
  66   struct priv* ppriv = (struct priv*)this_queue->priv;
  67
  68   if ( (mqd_t)-1 == ppriv->mq )
  69     {
  70       struct mq_attr attr;
  71       memset(&attr, 0, sizeof(attr));
  72       attr.mq_maxmsg = ppriv->max_cnt;
  73       attr.mq_msgsize = ppriv->max_sz;
  74
  75       errno = 0 ;
  76       /* see http://www.die.net/doc/linux/man/man7/mq_overview.7.html */
  77       if ( (mqd_t)-1 == (ppriv->mq = mq_open (ppriv->path,
  78                                               flags | O_CREAT, 0666, &attr)) )
  79         {
  80           /* Try resetting the mqueue resource parameters before formally
  81            * failing see http://www.die.net/doc/linux/man/man2/getrlimit.2.html
  82            */
  83           struct rlimit rlim;
  84           PERROR("mq_open");
  85           errno = 0;
  86           attr.mq_maxmsg += 1;
  87           rlim.rlim_cur = RLIM_INFINITY; /* attr.mq_maxmsg * sizeof(struct msg_msg *) + attr.mq_maxmsg * attr.mq_msgsize ; */
  88           rlim.rlim_max = RLIM_INFINITY ; /*rlim.rlim_cur ; */
  89           if ( setrlimit (RLIMIT_MSGQUEUE, &rlim) )
  90             {
  91               PERROR("setrlimit");
  92             }
  93           attr.mq_maxmsg -= 1 ;
  94
  95           if ( (mqd_t)-1 == (ppriv->mq = mq_open (ppriv->path,
  96                                                   flags | O_CREAT,
  97                                                   0666, &attr)) )
  98             {
  99               LOG_ER("mq_open(path=\"%s\", flags=0x%04x, 0666, "
 100                      "{attr.mq_maxmsg=%i, attr.mq_msgsize=%i})\n",
 101                      ppriv->path, flags | O_CREAT, attr.mq_maxmsg,
 102                      attr.mq_msgsize);
 103               return -1;
 104             }
 105         }
 106     }
 107
 108   return 0;
 109 }
 110
 111 static int xclose (struct queue* this_queue)
 112 {
 113   struct priv* ppriv = (struct priv*)this_queue->priv;
 114
 115   if ( (mqd_t)-1 != ppriv->mq )
 116     {
 117       if ( (mq_close (ppriv->mq)) < 0 )
 118         {
 119           PERROR("mq_close");
 120           return -1;
 121         }
 122     }
 123
 124   ppriv->mq = (mqd_t)-1;
 125
 126   return 0;
 127 }
 128
 129 static int xread (struct queue* this_queue, void* buf,
 130                   size_t count, int* pending)
 131 {
 132   struct priv* ppriv = (struct priv*)this_queue->priv;
 133   struct mq_attr attr;
 134   int mq_rec_rtrn;
 135   unsigned int pri;
 136
 137   if ( 0 == buf )
 138     {
 139       LOG_ER("queue read with NULL buf pointer\n");
 140       return -1;
 141     }
 142   if ( (mqd_t)-1 == ppriv->mq )
 143     {
 144       LOG_ER("queue read with queue closed.\n");
 145       return -1;
 146     }
 147
 148   LOG_PROG("about to call mq_receive().\n");
 149
 150   if ( (mq_rec_rtrn = mq_receive (ppriv->mq, buf, count, &pri)) < 0 )
 151     {
 152       switch ( errno )
 153         {
 154         default:
 155           PERROR("mq_receive");
 156
 157           /* If we've been interrupted it's likely that we're shutting
 158              down, so no need to print errors. */
 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   return malloc (*newcount = ((struct priv*)this_queue->priv)->max_sz);
 212 }
 213
 214 static void dealloc (struct queue* this_queue, void* buf)
 215 {
 216   (void) this_queue; /* appease -Wall -Werror */
 217   free (buf);
 218 }
 219
 220 int queue_mqueue_ctor (struct queue* this_queue,
 221                        const char*   path,
 222                        size_t        max_sz,
 223                        size_t        max_cnt)
 224 {
 225   static struct queue_vtbl vtbl = {
 226       destructor,
 227       xopen, xclose,
 228       xread, xwrite,
 229       alloc, dealloc
 230   };
 231
 232   struct priv* ppriv;
 233
 234   this_queue->vtbl = 0;
 235   this_queue->priv = 0;
 236
 237   ppriv = (struct priv*)malloc(sizeof(struct priv));
 238   if ( 0 == ppriv )
 239     {
 240       LOG_ER("malloc failed attempting to allocate %d bytes\n",
 241              sizeof(*ppriv));
 242       return -1;
 243     }
 244   memset(ppriv, 0, sizeof(*ppriv));
 245
 246   if ( 0 == (ppriv->path = strdup(path)) )
 247     {
 248       LOG_ER("strdup failed attempting to dup \"%s\"\n",
 249              path);
 250       free(ppriv);
 251       return -1;
 252     }
 253
 254   ppriv->mq = (mqd_t)-1;
 255   ppriv->max_sz = max_sz;
 256   ppriv->max_cnt = max_cnt;
 257
 258   this_queue->vtbl = &vtbl;
 259   this_queue->priv = ppriv;
 260
 261   return 0;
 262 }
 263
 264 #else  /* if defined(HAVE_MQUEUE_H) */
 265
 266 int queue_mqueue_ctor (struct queue* this_queue,
 267                        const char*   path,
 268                        size_t        max_sz,
 269                        size_t        max_cnt)
 270 {
 271   this_queue->vtbl = 0;
 272   this_queue->priv = 0;
 273   (void)path;     /* appease -Wall -Werror */
 274   (void)max_sz;   /* appease -Wall -Werror */
 275   (void)max_cnt;  /* appease -Wall -Werror */
 276
 277   return -1;
 278 }
 279
 280 #endif /* HAVE_MQUEUE_H */