ext/posix_mq/posix_mq.c

   1 #define _XOPEN_SOURCE 600
   2 #ifdef HAVE_SYS_SELECT_H
   3 #  include <sys/select.h>
   4 #endif
   5 #ifdef HAVE_SIGNAL_H
   6 #  include <signal.h>
   7 #endif
   8 #include <ruby.h>
   9
  10 #include <time.h>
  11 #include <mqueue.h>
  12 #include <fcntl.h>
  13 #include <sys/stat.h>
  14 #include <errno.h>
  15 #include <assert.h>
  16 #include <unistd.h>
  17
  18 #if defined(__linux__)
  19 #  define MQD_TO_FD(mqd) (int)(mqd)
  20 #elif defined(HAVE___MQ_OSHANDLE) /* FreeBSD */
  21 #  define MQD_TO_FD(mqd) __mq_oshandle(mqd)
  22 #else
  23 #  warning mqd_t is not select()-able on your OS
  24 #  define MQ_IO_MARK(mq) ((void)(0))
  25 #  define MQ_IO_SET(mq,val) ((void)(0))
  26 #endif
  27
  28 #ifdef MQD_TO_FD
  29 #  define MQ_IO_MARK(mq) rb_gc_mark((mq)->io)
  30 #  define MQ_IO_SET(mq,val) do { (mq)->io = (val); } while (0)
  31 #endif
  32
  33 struct posix_mq {
  34         mqd_t des;
  35         long msgsize;
  36         VALUE name;
  37 #ifdef MQD_TO_FD
  38         VALUE io;
  39 #endif
  40 };
  41
  42 static VALUE cPOSIX_MQ, cAttr;
  43 static ID id_new;
  44 static ID sym_r, sym_w, sym_rw;
  45 static const mqd_t MQD_INVALID = (mqd_t)-1;
  46
  47 /* Ruby 1.8.6+ macros (for compatibility with Ruby 1.9) */
  48 #ifndef RSTRING_PTR
  49 #  define RSTRING_PTR(s) (RSTRING(s)->ptr)
  50 #endif
  51 #ifndef RSTRING_LEN
  52 #  define RSTRING_LEN(s) (RSTRING(s)->len)
  53 #endif
  54 #ifndef RSTRUCT_PTR
  55 #  define RSTRUCT_PTR(s) (RSTRUCT(s)->ptr)
  56 #endif
  57 #ifndef RSTRUCT_LEN
  58 #  define RSTRUCT_LEN(s) (RSTRUCT(s)->len)
  59 #endif
  60
  61 #ifndef HAVE_RB_STR_SET_LEN
  62 #  ifdef RUBINIUS
  63 #    define rb_str_set_len(str,len) rb_str_resize(str,len)
  64 #  else /* 1.8.6 optimized version */
  65 /* this is taken from Ruby 1.8.7, 1.8.6 may not have it */
  66 static void rb_18_str_set_len(VALUE str, long len)
  67 {
  68         RSTRING(str)->len = len;
  69         RSTRING(str)->ptr[len] = '\0';
  70         rb_str_flush(str);
  71 }
  72 #    define rb_str_set_len(str,len) rb_18_str_set_len(str,len)
  73 #  endif /* ! RUBINIUS */
  74 #endif /* !defined(HAVE_RB_STR_SET_LEN) */
  75
  76 #ifndef HAVE_RB_STRUCT_ALLOC_NOINIT
  77 static VALUE rb_struct_alloc_noinit(VALUE class)
  78 {
  79         return rb_funcall(class, id_new, 0, 0);
  80 }
  81 #endif /* !defined(HAVE_RB_STRUCT_ALLOC_NOINIT) */
  82
  83 /* partial emulation of the 1.9 rb_thread_blocking_region under 1.8 */
  84 #ifndef HAVE_RB_THREAD_BLOCKING_REGION
  85 #  include <rubysig.h>
  86 #  define RUBY_UBF_IO ((rb_unblock_function_t *)-1)
  87 typedef void rb_unblock_function_t(void *);
  88 typedef VALUE rb_blocking_function_t(void *);
  89 static VALUE
  90 rb_thread_blocking_region(
  91         rb_blocking_function_t *func, void *data1,
  92         rb_unblock_function_t *ubf, void *data2)
  93 {
  94         VALUE rv;
  95
  96         assert(RUBY_UBF_IO == ubf && "RUBY_UBF_IO required for emulation");
  97
  98         TRAP_BEG;
  99         rv = func(data1);
 100         TRAP_END;
 101
 102         return rv;
 103 }
 104 #endif /* ! HAVE_RB_THREAD_BLOCKING_REGION */
 105
 106 /* used to pass arguments to mq_open inside blocking region */
 107 struct open_args {
 108         int argc;
 109         const char *name;
 110         int oflags;
 111         mode_t mode;
 112         struct mq_attr attr;
 113 };
 114
 115 /* used to pass arguments to mq_send/mq_receive inside blocking region */
 116 struct rw_args {
 117         mqd_t des;
 118         char *msg_ptr;
 119         size_t msg_len;
 120         unsigned msg_prio;
 121         struct timespec *timeout;
 122 };
 123
 124 /* hope it's there..., TODO: a better version that works in rbx */
 125 struct timeval rb_time_interval(VALUE);
 126
 127 static struct timespec *convert_timeout(struct timespec *dest, VALUE time)
 128 {
 129         struct timeval tv, now;
 130
 131         if (NIL_P(time))
 132                 return NULL;
 133
 134         tv = rb_time_interval(time); /* aggregate return :( */
 135         gettimeofday(&now, NULL);
 136         dest->tv_sec = now.tv_sec + tv.tv_sec;
 137         dest->tv_nsec = (now.tv_usec + tv.tv_usec) * 1000;
 138
 139         if (dest->tv_nsec > 1000000000) {
 140                 dest->tv_nsec -= 1000000000;
 141                 dest->tv_sec++;
 142         }
 143
 144         return dest;
 145 }
 146
 147 /* runs without GVL */
 148 static VALUE xopen(void *ptr)
 149 {
 150         struct open_args *x = ptr;
 151         mqd_t rv;
 152
 153         switch (x->argc) {
 154         case 2: rv = mq_open(x->name, x->oflags); break;
 155         case 3: rv = mq_open(x->name, x->oflags, x->mode, NULL); break;
 156         case 4: rv = mq_open(x->name, x->oflags, x->mode, &x->attr); break;
 157         default: rv = MQD_INVALID;
 158         }
 159
 160         return (VALUE)rv;
 161 }
 162
 163 /* runs without GVL */
 164 static VALUE xsend(void *ptr)
 165 {
 166         struct rw_args *x = ptr;
 167
 168         if (x->timeout)
 169                 return (VALUE)mq_timedsend(x->des, x->msg_ptr, x->msg_len,
 170                                            x->msg_prio, x->timeout);
 171
 172         return (VALUE)mq_send(x->des, x->msg_ptr, x->msg_len, x->msg_prio);
 173 }
 174
 175 /* runs without GVL */
 176 static VALUE xrecv(void *ptr)
 177 {
 178         struct rw_args *x = ptr;
 179
 180         if (x->timeout)
 181                 return (VALUE)mq_timedreceive(x->des, x->msg_ptr, x->msg_len,
 182                                               &x->msg_prio, x->timeout);
 183
 184         return (VALUE)mq_receive(x->des, x->msg_ptr, x->msg_len, &x->msg_prio);
 185 }
 186
 187 /* runs without GVL, path resolution may be slow */
 188 static VALUE xunlink(void *ptr)
 189 {
 190         VALUE name = (VALUE)ptr;
 191
 192         return (VALUE)mq_unlink(RSTRING_PTR(name));
 193 }
 194
 195 /* called by GC */
 196 static void mark(void *ptr)
 197 {
 198         struct posix_mq *mq = ptr;
 199
 200         rb_gc_mark(mq->name);
 201         MQ_IO_MARK(mq);
 202 }
 203
 204 /* called by GC */
 205 static void _free(void *ptr)
 206 {
 207         struct posix_mq *mq = ptr;
 208
 209         if (mq->des != MQD_INVALID) {
 210                 /* we ignore errors when gc-ing */
 211                 int saved_errno = errno;
 212
 213                 mq_close(mq->des);
 214                 errno = saved_errno;
 215                 mq->des = MQD_INVALID;
 216         }
 217 }
 218
 219 /* automatically called at creation (before initialize) */
 220 static VALUE alloc(VALUE klass)
 221 {
 222         struct posix_mq *mq;
 223         VALUE rv = Data_Make_Struct(klass, struct posix_mq, mark, _free, mq);
 224
 225         mq->des = MQD_INVALID;
 226         mq->msgsize = -1;
 227         mq->name = Qnil;
 228         MQ_IO_SET(mq, Qnil);
 229
 230         return rv;
 231 }
 232
 233 /* unwraps the posix_mq struct from self */
 234 static struct posix_mq *get(VALUE self, int need_valid)
 235 {
 236         struct posix_mq *mq;
 237
 238         Data_Get_Struct(self, struct posix_mq, mq);
 239
 240         if (need_valid && mq->des == MQD_INVALID)
 241                 rb_raise(rb_eIOError, "closed queue descriptor");
 242
 243         return mq;
 244 }
 245
 246 /* converts the POSIX_MQ::Attr astruct into a struct mq_attr attr */
 247 static void attr_from_struct(struct mq_attr *attr, VALUE astruct, int all)
 248 {
 249         VALUE *ptr;
 250
 251         if (CLASS_OF(astruct) != cAttr)
 252                 rb_raise(rb_eArgError, "not a POSIX_MQ::Attr: %s",
 253                          RSTRING_PTR(rb_inspect(astruct)));
 254
 255         ptr = RSTRUCT_PTR(astruct);
 256
 257         attr->mq_flags = NUM2LONG(ptr[0]);
 258
 259         if (all || !NIL_P(ptr[1]))
 260                 attr->mq_maxmsg = NUM2LONG(ptr[1]);
 261         if (all || !NIL_P(ptr[2]))
 262                 attr->mq_msgsize = NUM2LONG(ptr[2]);
 263         if (!NIL_P(ptr[3]))
 264                 attr->mq_curmsgs = NUM2LONG(ptr[3]);
 265 }
 266
 267 /*
 268  * call-seq:
 269  *      POSIX_MQ.new(name [, flags [, mode [, mq_attr]])        => mq
 270  *
 271  * Opens a POSIX message queue given by +name+.  +name+ should start
 272  * with a slash ("/") for portable applications.
 273  *
 274  * If a Symbol is given in place of integer +flags+, then:
 275  *
 276  * * +:r+ is equivalent to IO::RDONLY
 277  * * +:w+ is equivalent to IO::CREAT|IO::WRONLY
 278  * * +:rw+ is equivalent to IO::CREAT|IO::RDWR
 279  *
 280  * +mode+ is an integer and only used when IO::CREAT is used.
 281  * +mq_attr+ is a POSIX_MQ::Attr and only used if IO::CREAT is used.
 282  * If +mq_attr+ is not specified when creating a queue, then the
 283  * system defaults will be used.
 284  *
 285  * See the manpage for mq_open(3) for more details on this function.
 286  */
 287 static VALUE init(int argc, VALUE *argv, VALUE self)
 288 {
 289         struct posix_mq *mq = get(self, 0);
 290         struct open_args x;
 291         VALUE name, oflags, mode, attr;
 292
 293         rb_scan_args(argc, argv, "13", &name, &oflags, &mode, &attr);
 294
 295         if (TYPE(name) != T_STRING)
 296                 rb_raise(rb_eArgError, "name must be a string");
 297
 298         switch (TYPE(oflags)) {
 299         case T_NIL:
 300                 x.oflags = O_RDONLY;
 301                 break;
 302         case T_SYMBOL:
 303                 if (oflags == sym_r)
 304                         x.oflags = O_RDONLY;
 305                 else if (oflags == sym_w)
 306                         x.oflags = O_CREAT|O_WRONLY;
 307                 else if (oflags == sym_rw)
 308                         x.oflags = O_CREAT|O_RDWR;
 309                 else
 310                         rb_raise(rb_eArgError,
 311                                  "symbol must be :r, :w, or :rw: %s",
 312                                  RSTRING_PTR(rb_inspect(oflags)));
 313                 break;
 314         case T_BIGNUM:
 315         case T_FIXNUM:
 316                 x.oflags = NUM2INT(oflags);
 317                 break;
 318         default:
 319                 rb_raise(rb_eArgError, "flags must be an int, :r, :w, or :wr");
 320         }
 321
 322         x.name = RSTRING_PTR(name);
 323         x.argc = 2;
 324
 325         switch (TYPE(mode)) {
 326         case T_FIXNUM:
 327                 x.argc = 3;
 328                 x.mode = NUM2INT(mode);
 329                 break;
 330         case T_NIL:
 331                 if (x.oflags & O_CREAT) {
 332                         x.argc = 3;
 333                         x.mode = 0666;
 334                 }
 335                 break;
 336         default:
 337                 rb_raise(rb_eArgError, "mode not an integer");
 338         }
 339
 340         switch (TYPE(attr)) {
 341         case T_STRUCT:
 342                 x.argc = 4;
 343                 attr_from_struct(&x.attr, attr, 1);
 344
 345                 /* principle of least surprise */
 346                 if (x.attr.mq_flags & O_NONBLOCK)
 347                         x.oflags |= O_NONBLOCK;
 348                 break;
 349         case T_NIL:
 350                 break;
 351         default:
 352                 rb_raise(rb_eArgError, "attr must be a POSIX_MQ::Attr: %s",
 353                          RSTRING_PTR(rb_inspect(attr)));
 354         }
 355
 356         mq->des = (mqd_t)rb_thread_blocking_region(xopen, &x, RUBY_UBF_IO, 0);
 357         if (mq->des == MQD_INVALID)
 358                 rb_sys_fail("mq_open");
 359
 360         mq->name = rb_str_dup(name);
 361
 362         return self;
 363 }
 364
 365 /*
 366  * call-seq:
 367  *      POSIX_MQ.unlink(name) =>        1
 368  *
 369  * Unlinks the message queue given by +name+.  The queue will be destroyed
 370  * when the last process with the queue open closes its queue descriptors.
 371  */
 372 static VALUE s_unlink(VALUE self, VALUE name)
 373 {
 374         mqd_t rv;
 375         void *ptr = (void *)name;
 376
 377         if (TYPE(name) != T_STRING)
 378                 rb_raise(rb_eArgError, "argument must be a string");
 379
 380         rv = (mqd_t)rb_thread_blocking_region(xunlink, ptr, RUBY_UBF_IO, 0);
 381         if (rv == MQD_INVALID)
 382                 rb_sys_fail("mq_unlink");
 383
 384         return INT2NUM(1);
 385 }
 386
 387 /*
 388  * call-seq:
 389  *      mq.unlink =>    mq
 390  *
 391  * Unlinks the message queue to prevent other processes from accessing it.
 392  * All existing queue descriptors to this queue including those opened by
 393  * other processes are unaffected.  The queue will only be destroyed
 394  * when the last process with open descriptors to this queue closes
 395  * the descriptors.
 396  */
 397 static VALUE _unlink(VALUE self)
 398 {
 399         struct posix_mq *mq = get(self, 0);
 400         mqd_t rv;
 401         void *ptr = (void *)mq->name;
 402
 403         assert(TYPE(mq->name) == T_STRING && "mq->name is not a string");
 404
 405         rv = (mqd_t)rb_thread_blocking_region(xunlink, ptr, RUBY_UBF_IO, 0);
 406         if (rv == MQD_INVALID)
 407                 rb_sys_fail("mq_unlink");
 408
 409         return self;
 410 }
 411
 412 static void setup_send_buffer(struct rw_args *x, VALUE buffer)
 413 {
 414         buffer = rb_obj_as_string(buffer);
 415         x->msg_ptr = RSTRING_PTR(buffer);
 416         x->msg_len = (size_t)RSTRING_LEN(buffer);
 417 }
 418
 419 /*
 420  * call-seq:
 421  *      mq.send(string [,priority[, timeout]])  => nil
 422  *
 423  * Inserts the given +string+ into the message queue with an optional,
 424  * unsigned integer +priority+.  If the optional +timeout+ is specified,
 425  * then Errno::ETIMEDOUT will be raised if the operation cannot complete
 426  * before +timeout+ seconds has elapsed.  Without +timeout+, this method
 427  * may block until the queue is writable.
 428  */
 429 static VALUE _send(int argc, VALUE *argv, VALUE self)
 430 {
 431         struct posix_mq *mq = get(self, 1);
 432         struct rw_args x;
 433         VALUE buffer, prio, timeout;
 434         mqd_t rv;
 435         struct timespec expire;
 436
 437         rb_scan_args(argc, argv, "12", &buffer, &prio, &timeout);
 438
 439         setup_send_buffer(&x, buffer);
 440         x.des = mq->des;
 441         x.timeout = convert_timeout(&expire, timeout);
 442         x.msg_prio = NIL_P(prio) ? 0 : NUM2UINT(prio);
 443
 444         rv = (mqd_t)rb_thread_blocking_region(xsend, &x, RUBY_UBF_IO, 0);
 445         if (rv == MQD_INVALID)
 446                 rb_sys_fail("mq_send");
 447
 448         return Qnil;
 449 }
 450
 451 /*
 452  * call-seq:
 453  *      mq << string    => mq
 454  *
 455  * Inserts the given +string+ into the message queue with a
 456  * default priority of 0 and no timeout.
 457  */
 458 static VALUE send0(VALUE self, VALUE buffer)
 459 {
 460         struct posix_mq *mq = get(self, 1);
 461         struct rw_args x;
 462         mqd_t rv;
 463
 464         setup_send_buffer(&x, buffer);
 465         x.des = mq->des;
 466         x.timeout = NULL;
 467         x.msg_prio = 0;
 468
 469         rv = (mqd_t)rb_thread_blocking_region(xsend, &x, RUBY_UBF_IO, 0);
 470         if (rv == MQD_INVALID)
 471                 rb_sys_fail("mq_send");
 472
 473         return self;
 474 }
 475
 476 #ifdef MQD_TO_FD
 477 /*
 478  * call-seq:
 479  *      mq.to_io        => IO
 480  *
 481  * Returns an IO.select-able +IO+ object.  This method is only available
 482  * under Linux and is not intended to be portable.
 483  */
 484 static VALUE to_io(VALUE self)
 485 {
 486         struct posix_mq *mq = get(self, 1);
 487         int fd = MQD_TO_FD(mq->des);
 488
 489         if (NIL_P(mq->io))
 490                 mq->io = rb_funcall(rb_cIO, id_new, 1, INT2NUM(fd));
 491
 492         return mq->io;
 493 }
 494 #endif
 495
 496 static void get_msgsize(struct posix_mq *mq)
 497 {
 498         struct mq_attr attr;
 499
 500         if (mq_getattr(mq->des, &attr) == MQD_INVALID)
 501                 rb_sys_fail("mq_getattr");
 502
 503         mq->msgsize = attr.mq_msgsize;
 504 }
 505
 506 /*
 507  * call-seq:
 508  *      mq.receive([buffer, [timeout]])         => [ message, priority ]
 509  *
 510  * Takes the highest priority message off the queue and returns
 511  * an array containing the message as a String and the Integer
 512  * priority of the message.
 513  *
 514  * If the optional +buffer+ is present, then it must be a String
 515  * which will receive the data.
 516  *
 517  * If the optional +timeout+ is present, then it may be a Float
 518  * or Integer specifying the timeout in seconds.  Errno::ETIMEDOUT
 519  * will be raised if +timeout+ has elapsed and there are no messages
 520  * in the queue.
 521  */
 522 static VALUE receive(int argc, VALUE *argv, VALUE self)
 523 {
 524         struct posix_mq *mq = get(self, 1);
 525         struct rw_args x;
 526         VALUE buffer, timeout;
 527         ssize_t r;
 528         struct timespec expire;
 529
 530         if (mq->msgsize < 0)
 531                 get_msgsize(mq);
 532
 533         rb_scan_args(argc, argv, "02", &buffer, &timeout);
 534         x.timeout = convert_timeout(&expire, timeout);
 535
 536         if (NIL_P(buffer)) {
 537                 buffer = rb_str_new(0, mq->msgsize);
 538         } else {
 539                 StringValue(buffer);
 540                 rb_str_modify(buffer);
 541                 rb_str_resize(buffer, mq->msgsize);
 542         }
 543         OBJ_TAINT(buffer);
 544         x.msg_ptr = RSTRING_PTR(buffer);
 545         x.msg_len = (size_t)mq->msgsize;
 546         x.des = mq->des;
 547
 548         r = (ssize_t)rb_thread_blocking_region(xrecv, &x, RUBY_UBF_IO, 0);
 549         if (r < 0)
 550                 rb_sys_fail("mq_receive");
 551
 552         rb_str_set_len(buffer, r);
 553
 554         return rb_ary_new3(2, buffer, UINT2NUM(x.msg_prio));
 555 }
 556
 557 /*
 558  * call-seq:
 559  *      mq.attr =>       mq_attr
 560  *
 561  * Returns a POSIX_MQ::Attr struct containing the attributes
 562  * of the message queue.  See the mq_getattr(3) manpage for
 563  * more details.
 564  */
 565 static VALUE getattr(VALUE self)
 566 {
 567         struct posix_mq *mq = get(self, 1);
 568         struct mq_attr attr;
 569         VALUE astruct;
 570         VALUE *ptr;
 571
 572         if (mq_getattr(mq->des, &attr) == MQD_INVALID)
 573                 rb_sys_fail("mq_getattr");
 574
 575         astruct = rb_struct_alloc_noinit(cAttr);
 576         ptr = RSTRUCT_PTR(astruct);
 577         ptr[0] = LONG2NUM(attr.mq_flags);
 578         ptr[1] = LONG2NUM(attr.mq_maxmsg);
 579         ptr[2] = LONG2NUM(attr.mq_msgsize);
 580         ptr[3] = LONG2NUM(attr.mq_curmsgs);
 581
 582         return astruct;
 583 }
 584
 585 /*
 586  * call-seq:
 587  *      mq.attr = POSIX_MQ::Attr(IO::NONBLOCK)          => mq_attr
 588  *
 589  * Only the IO::NONBLOCK flag may be set or unset (zero) in this manner.
 590  * See the mq_setattr(3) manpage for more details.
 591  *
 592  * Consider using the POSIX_MQ#nonblock= method as it is easier and
 593  * more natural to use.
 594  */
 595 static VALUE setattr(VALUE self, VALUE astruct)
 596 {
 597         struct posix_mq *mq = get(self, 1);
 598         struct mq_attr newattr;
 599
 600         attr_from_struct(&newattr, astruct, 0);
 601
 602         if (mq_setattr(mq->des, &newattr, NULL) == MQD_INVALID)
 603                 rb_sys_fail("mq_setattr");
 604
 605         return astruct;
 606 }
 607
 608 /*
 609  * call-seq:
 610  *      mq.close        => nil
 611  *
 612  * Closes the underlying message queue descriptor.
 613  * If this descriptor had a registered notification request, the request
 614  * will be removed so another descriptor or process may register a
 615  * notification request.  Message queue descriptors are automatically
 616  * closed by garbage collection.
 617  */
 618 static VALUE _close(VALUE self)
 619 {
 620         struct posix_mq *mq = get(self, 1);
 621
 622         if (mq_close(mq->des) == MQD_INVALID)
 623                 rb_sys_fail("mq_close");
 624
 625         mq->des = MQD_INVALID;
 626         MQ_IO_SET(mq, Qnil);
 627
 628         return Qnil;
 629 }
 630
 631 /*
 632  *  call-seq:
 633  *      mq.closed?      => true or false
 634  *
 635  *  Returns +true+ if the message queue descriptor is closed and therefore
 636  *  unusable, otherwise +false+
 637  */
 638 static VALUE closed(VALUE self)
 639 {
 640         struct posix_mq *mq = get(self, 0);
 641
 642         return mq->des == MQD_INVALID ? Qtrue : Qfalse;
 643 }
 644
 645 /*
 646  * call-seq:
 647  *      mq.name         => string
 648  *
 649  * Returns the string name of message queue associated with +mq+
 650  */
 651 static VALUE name(VALUE self)
 652 {
 653         struct posix_mq *mq = get(self, 0);
 654
 655         return mq->name;
 656 }
 657
 658 static int lookup_sig(VALUE sig)
 659 {
 660         static VALUE list;
 661         const char *ptr;
 662         long len;
 663
 664         sig = rb_obj_as_string(sig);
 665         len = RSTRING_LEN(sig);
 666         ptr = RSTRING_PTR(sig);
 667
 668         if (len > 3 && !memcmp("SIG", ptr, 3))
 669                 sig = rb_str_new(ptr + 3, len - 3);
 670
 671         if (!list) {
 672                 VALUE mSignal = rb_define_module("Signal"""); /* avoid RDoc */
 673
 674                 list = rb_funcall(mSignal, rb_intern("list"), 0, 0);
 675                 rb_global_variable(&list);
 676         }
 677
 678         sig = rb_hash_aref(list, sig);
 679         if (NIL_P(sig))
 680                 rb_raise(rb_eArgError, "invalid signal: %s\n",
 681                          RSTRING_PTR(rb_inspect(sig)));
 682
 683         return NUM2INT(sig);
 684 }
 685
 686 /*
 687  * call-seq:
 688  *      mq.notify = signal      => signal
 689  *
 690  * Registers the notification request to deliver a given +signal+
 691  * to the current process when message is received.
 692  * If +signal+ is +nil+, it will unregister and disable the notification
 693  * request to allow other processes to register a request.
 694  * If +signal+ is +false+, it will register a no-op notification request
 695  * which will prevent other processes from registering a notification.
 696  * Only one process may have a notification request for a queue
 697  * at a time, Errno::EBUSY will be raised if there is already
 698  * a notification request registration for the queue.
 699  *
 700  * For readers of the mq_notify(3) manpage, passing +false+
 701  * is equivalent to SIGEV_NONE, and passing +nil+ is equivalent
 702  * of passing a NULL notification pointer to mq_notify(3).
 703  */
 704 static VALUE setnotify(VALUE self, VALUE arg)
 705 {
 706         struct posix_mq *mq = get(self, 1);
 707         struct sigevent not;
 708         struct sigevent * notification = &not;
 709         VALUE rv = arg;
 710
 711         not.sigev_notify = SIGEV_SIGNAL;
 712
 713         switch (TYPE(arg)) {
 714         case T_FALSE:
 715                 not.sigev_notify = SIGEV_NONE;
 716                 break;
 717         case T_NIL:
 718                 notification = NULL;
 719                 break;
 720         case T_FIXNUM:
 721                 not.sigev_signo = NUM2INT(arg);
 722                 break;
 723         case T_SYMBOL:
 724         case T_STRING:
 725                 not.sigev_signo = lookup_sig(arg);
 726                 rv = INT2NUM(not.sigev_signo);
 727                 break;
 728         default:
 729                 /* maybe support Proc+thread via sigev_notify_function.. */
 730                 rb_raise(rb_eArgError, "must be a signal or nil");
 731         }
 732
 733         if (mq_notify(mq->des, notification) == MQD_INVALID)
 734                 rb_sys_fail("mq_notify");
 735
 736         return rv;
 737 }
 738
 739 /*
 740  * call-seq:
 741  *      mq.nonblock?    => true or false
 742  *
 743  * Returns the current non-blocking state of the message queue descriptor.
 744  */
 745 static VALUE getnonblock(VALUE self)
 746 {
 747         struct mq_attr attr;
 748         struct posix_mq *mq = get(self, 1);
 749
 750         if (mq_getattr(mq->des, &attr) == MQD_INVALID)
 751                 rb_sys_fail("mq_getattr");
 752
 753         mq->msgsize = attr.mq_msgsize; /* optimization */
 754
 755         return attr.mq_flags & O_NONBLOCK ? Qtrue : Qfalse;
 756 }
 757
 758 /*
 759  * call-seq:
 760  *      mq.nonblock = boolean   => boolean
 761  *
 762  * Enables or disables non-blocking operation for the message queue
 763  * descriptor.  Errno::EAGAIN will be raised in situations where
 764  * the queue would block.  This is not compatible with +timeout+
 765  * arguments to POSIX_MQ#send and POSIX_MQ#receive.
 766  */
 767 static VALUE setnonblock(VALUE self, VALUE nb)
 768 {
 769         struct mq_attr newattr, oldattr;
 770         struct posix_mq *mq = get(self, 1);
 771
 772         if (nb == Qtrue)
 773                 newattr.mq_flags = O_NONBLOCK;
 774         else if (nb == Qfalse)
 775                 newattr.mq_flags = 0;
 776         else
 777                 rb_raise(rb_eArgError, "must be true or false");
 778
 779         if (mq_setattr(mq->des, &newattr, &oldattr) == MQD_INVALID)
 780                 rb_sys_fail("mq_setattr");
 781
 782         mq->msgsize = oldattr.mq_msgsize; /* optimization */
 783
 784         return nb;
 785 }
 786
 787 void Init_posix_mq_ext(void)
 788 {
 789         cPOSIX_MQ = rb_define_class("POSIX_MQ", rb_cObject);
 790         rb_define_alloc_func(cPOSIX_MQ, alloc);
 791         cAttr = rb_const_get(cPOSIX_MQ, rb_intern("Attr"));
 792
 793         /*
 794          * The maximum number of open message descriptors supported
 795          * by the system.  This may be -1, in which case it is dynamically
 796          * set at runtime.  Consult your operating system documentation
 797          * for system-specific information about this.
 798          */
 799         rb_define_const(cPOSIX_MQ, "OPEN_MAX",
 800                         LONG2NUM(sysconf(_SC_MQ_OPEN_MAX)));
 801
 802         /*
 803          * The maximum priority that may be specified for POSIX_MQ#send
 804          * On POSIX-compliant systems, this is at least 31, but some
 805          * systems allow higher limits.
 806          * The minimum priority is always zero.
 807          */
 808         rb_define_const(cPOSIX_MQ, "PRIO_MAX",
 809                         LONG2NUM(sysconf(_SC_MQ_PRIO_MAX)));
 810
 811         rb_define_singleton_method(cPOSIX_MQ, "unlink", s_unlink, 1);
 812
 813         rb_define_method(cPOSIX_MQ, "initialize", init, -1);
 814         rb_define_method(cPOSIX_MQ, "send", _send, -1);
 815         rb_define_method(cPOSIX_MQ, "<<", send0, 1);
 816         rb_define_method(cPOSIX_MQ, "receive", receive, -1);
 817         rb_define_method(cPOSIX_MQ, "attr", getattr, 0);
 818         rb_define_method(cPOSIX_MQ, "attr=", setattr, 1);
 819         rb_define_method(cPOSIX_MQ, "close", _close, 0);
 820         rb_define_method(cPOSIX_MQ, "closed?", closed, 0);
 821         rb_define_method(cPOSIX_MQ, "unlink", _unlink, 0);
 822         rb_define_method(cPOSIX_MQ, "name", name, 0);
 823         rb_define_method(cPOSIX_MQ, "notify=", setnotify, 1);
 824         rb_define_method(cPOSIX_MQ, "nonblock=", setnonblock, 1);
 825         rb_define_method(cPOSIX_MQ, "nonblock?", getnonblock, 0);
 826 #ifdef MQD_TO_FD
 827         rb_define_method(cPOSIX_MQ, "to_io", to_io, 0);
 828 #endif
 829
 830         id_new = rb_intern("new");
 831         sym_r = ID2SYM(rb_intern("r"));
 832         sym_w = ID2SYM(rb_intern("w"));
 833         sym_rw = ID2SYM(rb_intern("rw"));
 834 }