POSIX_MQ#notify block execution on message received

This is implementation uses both a short-lived POSIX thread and
a pre-spawned Ruby Thread in a manner that works properly under
both Ruby 1.8 (green threads) and 1.9 (where Ruby Threads are
POSIX threads).

The short-lived POSIX thread will write a single "\0" byte to
a pipe the Ruby Thread waits on.  This operation is atomic
on all platforms.  Once the Ruby Thread is woken up from the
pipe, it will execute th block given to it.

This dual-thread implementation is inspired by the way glibc
implements mq_notify(3) + SIGEV_THREAD under Linux where the
kernel itself cannot directly spawn POSIX threads.

1 files changed, 31 insertions, 0 deletions

diff --git a/lib/posix_mq.rb b/lib/posix_mq.rb
index 5e660ec..91f4140 100644
--- a/lib/posix_mq.rb
+++ b/lib/posix_mq.rb
@@ -26,7 +26,38 @@ class POSIX_MQ
         mq.close unless mq.closed?
       end
     end
+  end
 
+  # Executes the given block upon reception of the next message in an
+  # empty queue.  If the message queue is not empty, then this block
+  # will only be fired after the queue is emptied and repopulated with
+  # one message.
+  #
+  # This block will only be executed upon the arrival of the
+  # first message and must be reset/reenabled for subsequent
+  # notifications.  This block will execute in a separate Ruby
+  # Thread (and thus will safely have the GVL by default).
+  def notify(&block)
+    block.arity == 1 or
+      raise ArgumentError, "arity of notify block must be 1"
+    r, w = IO.pipe
+    thr = Thread.new(r, w, self) do |r, w, mq|
+      begin
+        begin
+          r.read(1) or raise Errno::EINTR
+        rescue Errno::EINTR, Errno::EAGAIN
+          retry
+        end
+        block.call(mq)
+      ensure
+        mq.notify_thread = nil
+        r.close rescue nil
+        w.close rescue nil
+      end
+    end
+    self.notify = w
+    self.notify_thread = thr
+    nil
   end
 
 end