src/queue_msg.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_msg.h"
  25
  26 #include "perror.h"
  27
  28 #if HAVE_SYS_MSG_H
  29
  30 #include <fcntl.h>
  31 #include <stdio.h>
  32 #include <stdlib.h>
  33 #include <string.h>
  34
  35 #include <sys/types.h>
  36 #include <sys/ipc.h>
  37 #include <sys/msg.h>
  38
  39 struct priv {
  40   char*         path;
  41   key_t         key;
  42   int           mq;
  43   size_t        max_sz;
  44   size_t        max_cnt;
  45   int           flags;
  46 };
  47
  48 #define MSG_TYPE 1  /* Must be positive non-zero. */
  49
  50 struct local_msgbuf {
  51   long mtype;    /* Type of received/sent message. */
  52   char mtext[1]; /* Text of the message. */
  53 };
  54
  55
  56 static void destructor (struct queue* this_queue)
  57 {
  58   struct priv* ppriv = (struct priv*)this_queue->priv;
  59
  60   if ( -1 != ppriv->mq )
  61     {
  62       msgctl (ppriv->mq, IPC_RMID, NULL); /* We ignore errors at this point. */
  63     }
  64
  65   free (ppriv->path);
  66   free (ppriv);
  67
  68   this_queue->vtbl = 0;
  69   this_queue->priv = 0;
  70 }
  71
  72 /* This function will increase the maximum message size limit on Linux
  73  * systems.  It will silently do nothing on other systems.
  74  */
  75
  76 static void set_max_msg_size (size_t max_sz)
  77 {
  78   size_t old_max = 0;
  79
  80   FILE* fp = fopen("/proc/sys/kernel/msgmax", "r+");
  81   if ( NULL == fp )
  82     {
  83       return;
  84     }
  85
  86   fscanf (fp, "%ld", &old_max);
  87
  88   if ( max_sz > old_max ) {
  89       rewind (fp);
  90       fprintf (fp, "%ld", max_sz);
  91   }
  92
  93   fclose (fp);
  94 }
  95
  96 static int xopen (struct queue* this_queue, int flags)
  97 {
  98   struct priv* ppriv = (struct priv*)this_queue->priv;
  99   struct msqid_ds ds;
 100   int mq;
 101
 102   if ( -1 != ppriv->mq )
 103     {
 104       LOG_ER("Queue already open.\n");
 105       return -1;
 106     }
 107
 108   /* Get a handle to our message queue. */
 109   mq = msgget (ppriv->key, 0666 | IPC_CREAT);
 110   if ( -1 == mq )
 111     {
 112       PERROR("msgget()");
 113       return -1;
 114     }
 115
 116   /* Set the queue size in bytes. */
 117   set_max_msg_size (ppriv->max_sz);
 118
 119   memset (&ds, 0, sizeof(ds));
 120   ds.msg_qbytes = ppriv->max_sz * ppriv->max_cnt;
 121   if ( msgctl (mq, IPC_SET, &ds) )
 122     {
 123       if ( EPERM == errno )
 124         {
 125           LOG_ER("Permission denied attempting to set the queue "
 126                  "size using msgctl() -- you may need to be root.\n");
 127         }
 128       else
 129         {
 130           PERROR("msgctl()");
 131           return -1;
 132         }
 133     }
 134
 135   ppriv->mq = mq;
 136   ppriv->flags = flags;
 137
 138   return 0;
 139 }
 140
 141 static int xclose (struct queue* this_queue)
 142 {
 143   struct priv* ppriv = (struct priv*)this_queue->priv;
 144
 145   ppriv->mq = -1;  /* No additional close operation required. */
 146   return 0;
 147 }
 148
 149 static int xread (struct queue* this_queue, void* buf,
 150                   size_t count, int* pending)
 151 {
 152   struct priv* ppriv = (struct priv*)this_queue->priv;
 153
 154   struct local_msgbuf* mp;
 155   struct msqid_ds ds;
 156   int msgrcv_ret;
 157
 158   if ( O_WRONLY == ppriv->flags )
 159     {
 160       LOG_ER("Queue read attempted without appropriate permission.\n");
 161       return -1;
 162     }
 163   if ( 0 == buf )
 164     {
 165       LOG_ER("Queue read with NULL buf pointer\n");
 166       return -1;
 167     }
 168   if ( -1 == ppriv->mq )
 169     {
 170       LOG_ER("Queue read with queue closed.\n");
 171       return -1;
 172     }
 173   if ( count > ppriv->max_sz )
 174     {
 175       LOG_ER("Queue read with  count (%d) > max_sz (%d).\n",
 176              count, ppriv->max_sz);
 177       return -1;
 178     }
 179
 180   mp = (struct local_msgbuf*)(((char*)buf) - sizeof(mp->mtype));
 181
 182   /* Receive bytes. */
 183   if ( (msgrcv_ret = msgrcv(ppriv->mq, mp, count, MSG_TYPE, 0)) < 0 )
 184     {
 185       switch ( errno )
 186         {
 187         default:
 188           PERROR("msgrcv");
 189
 190           /* If we've been interrupted or if the message queue was
 191            * removed, it's likely that we're shutting down, so no need to
 192            * print errors. */
 193         case EIDRM:
 194         case EINTR:
 195           return msgrcv_ret;
 196         }
 197     }
 198
 199   if ( MSG_TYPE != mp->mtype )
 200     {
 201       LOG_WARN("Unrecognized message type %d.\n", mp->mtype);
 202     }
 203
 204   if ( msgctl(ppriv->mq, IPC_STAT, &ds) )
 205     {
 206       PERROR("msgctl(ppriv->mq, IPC_GET, &ds)");
 207       return -1;
 208     }
 209
 210   *pending =  ds.msg_qnum;
 211
 212   return msgrcv_ret;
 213 }
 214
 215 static int xwrite (struct queue* this_queue, const void* buf, size_t count)
 216 {
 217   int retry = 99;
 218   struct priv* ppriv = (struct priv*)this_queue->priv;
 219
 220   struct local_msgbuf* mp;
 221   int msgsnd_ret;
 222
 223   if ( O_RDONLY == ppriv->flags )
 224     {
 225       LOG_ER("Queue write attempted without appropriate permission.\n");
 226       return -1;
 227     }
 228   if ( 0 == buf )
 229     {
 230       LOG_ER("Queue write with NULL buf pointer.\n");
 231       return -1;
 232     }
 233   if ( -1 == ppriv->mq )
 234     {
 235       LOG_ER("Queue write with queue closed.\n");
 236       return -1;
 237     }
 238   if ( count > ppriv->max_sz )
 239     {
 240       LOG_ER("Queue write with count (%d) > max_sz (%d).\n",
 241              count, ppriv->max_sz);
 242       return -1;
 243     }
 244
 245   mp = (struct local_msgbuf*)(((char*)buf) - sizeof(mp->mtype));
 246
 247   mp->mtype = MSG_TYPE;  /* Must be positive non-zero. */
 248
 249   while ( ((msgsnd_ret = msgsnd(ppriv->mq, mp, count, IPC_NOWAIT)) < 0)
 250           && retry )
 251     {
 252       /* We'll try msgsnd a few times, since with IPC_NOWAIT we might
 253        * get a transient error if the queue is full.
 254        */
 255       switch (errno )
 256         {
 257         case EAGAIN:
 258           --retry;
 259           LOG_ER("Queue full, will now retry msgsnd.\n");
 260           break;
 261
 262         default:
 263           PERROR("msgsnd");
 264           return msgsnd_ret;
 265         }
 266     }
 267
 268   return msgsnd_ret;
 269 }
 270
 271 static void* alloc (struct queue* this_queue, size_t* newcount)
 272 {
 273   struct local_msgbuf* mp;
 274   struct priv* ppriv = (struct priv*)this_queue->priv;
 275   char* ret = (char*)malloc(ppriv->max_sz + sizeof(mp->mtype));
 276
 277   if ( ret )
 278     {
 279       *newcount = ppriv->max_sz;
 280       return ret + sizeof(mp->mtype);
 281     }
 282
 283   return 0;
 284 }
 285
 286 static void dealloc (struct queue* this_queue, void* buf)
 287 {
 288   struct local_msgbuf* mp =
 289     (struct local_msgbuf*)((char*)buf - sizeof(mp->mtype));
 290   (void)this_queue; /* appease -Wall -Werror */
 291   free(mp);
 292 }
 293
 294 int queue_msg_ctor (struct queue* this_queue,
 295                     const char*   path,
 296                     size_t        max_sz,
 297                     size_t        max_cnt)
 298 {
 299   static struct queue_vtbl vtbl = {
 300       destructor,
 301       xopen, xclose,
 302       xread, xwrite,
 303       alloc, dealloc
 304   };
 305
 306   struct priv* ppriv;
 307
 308   this_queue->vtbl = 0;
 309   this_queue->priv = 0;
 310
 311   ppriv = (struct priv*)malloc (sizeof(struct priv));
 312   if ( 0 == ppriv )
 313     {
 314       LOG_ER("Failed to allocate %d bytes for queue data.\n", sizeof(*ppriv));
 315       return -1;
 316     }
 317   memset(ppriv, 0, sizeof(*ppriv));
 318
 319   if ( 0 == (ppriv->path = strdup(path)) )
 320     {
 321       LOG_ER("Strdup failed attempting to dup \"%s\".\n", path);
 322       free(ppriv);
 323       return -1;
 324     }
 325
 326   /* We try to convert the path into an appropriate numeric key to
 327      identify this queue. */
 328   if ( sscanf(path, "%d", (int*)&ppriv->key) != 1 )
 329     {
 330       /* A default key is provided. */
 331       ppriv->key = 1;
 332     }
 333
 334   ppriv->mq = -1;
 335   ppriv->max_sz = max_sz;
 336   ppriv->max_cnt = max_cnt;
 337
 338   this_queue->vtbl = &vtbl;
 339   this_queue->priv = ppriv;
 340
 341   return 0;
 342 }
 343
 344 #else  /* if HAVE_SYS_MSG_H */
 345
 346 int queue_msg_ctor (struct queue* this_queue,
 347                     const char*   path,
 348                     size_t        max_sz,
 349                     size_t        max_cnt)
 350 {
 351   this_queue->vtbl = 0;
 352   this_queue->priv = 0;
 353   return -1;
 354 }
 355
 356 #endif /* HAVE_SYS_MSG_H */