Initial commit.

lib/process_shared.rb

@@ -0,0 +1,6 @@
+require 'ffi'
+
+require 'process_shared/semaphore'
+require 'process_shared/bounded_semaphore'
+require 'process_shared/mutex'
+require 'process_shared/shared_memory'

lib/process_shared/abstract_semaphore.rb

@@ -0,0 +1,50 @@
+require 'process_shared/psem'
+require 'process_shared/with_self'
+
+module ProcessShared
+  class AbstractSemaphore
+    include WithSelf
+    protected
+    include ProcessShared::PSem
+    public
+
+    # Generate a name for a semaphore.
+    def self.gen_name(middle, name = nil)
+      if name
+        name
+      else
+        @count ||= 0
+        @count += 1
+        "ps-#{middle}-#{Process.pid}-#{@count}"
+      end
+    end
+
+    def self.make_finalizer(name)
+      proc { ProcessShared::PSem.psem_unlink(name, nil) }
+    end
+
+    # private_class_method :new
+
+    protected
+
+    attr_reader :sem, :err
+
+    def init(size, middle, name, &block)
+      @sem = FFI::MemoryPointer.new(size)
+      @err = FFI::MemoryPointer.new(:pointer)
+      psem_name = AbstractSemaphore.gen_name(middle, name)
+      block.call(psem_name)
+
+     if name
+        # name explicitly given.  Don't unlink because we might want to share it with another process.
+        # Instead, register a finalizer to unlink.
+        ObjectSpace.define_finalizer(self, self.class.make_finalizer(name))
+      else
+        # On Linux, removes the entry in /dev/shm and prevents other
+        # processes from opening this semaphore unless they inherit it
+        # as forked children.
+        psem_unlink(psem_name, err) unless name
+      end
+    end
+  end
+end

lib/process_shared/bounded_semaphore.rb

@@ -0,0 +1,43 @@
+require 'process_shared/psem'
+require 'process_shared/semaphore'
+
+module ProcessShared
+  class BoundedSemaphore < Semaphore
+    # With no associated block, open is a synonym for
+    # Semaphore.new. If the optional code block is given, it will be
+    # passed `sem` as an argument, and the Semaphore object will
+    # automatically be closed when the block terminates. In this
+    # instance, Semaphore.open returns the value of the block.
+    #
+    # @param [Integer] value the initial semaphore value
+    # @param [String] name not currently supported
+    def self.open(maxvalue, value = 1, name = nil, &block)
+      new(maxvalue, value, name).with_self(&block)
+    end
+
+    # Create a new semaphore with initial value `value`.  After
+    # Kernel#fork, the semaphore will be shared across two (or more)
+    # processes. The semaphore must be closed with #close in each
+    # process that no longer needs the semaphore.
+    #
+    # (An object finalizer is registered that will close the semaphore
+    # to avoid memory leaks, but this should be considered a last
+    # resort).
+    #
+    # @param [Integer] value the initial semaphore value
+    # @param [String] name not currently supported
+    def initialize(maxvalue, value = 1, name = nil)
+      init(PSem.sizeof_bsem_t, 'bsem', name) do |sem_name|
+        bsem_open(sem, sem_name, maxvalue, value, err)
+      end
+    end
+
+    protected
+
+    alias_method :psem_unlink, :bsem_unlink
+    alias_method :psem_close, :bsem_close
+    alias_method :psem_wait, :bsem_wait
+    alias_method :psem_post, :bsem_post
+    alias_method :psem_getvalue, :bsem_getvalue
+  end
+end

lib/process_shared/condition_variable.rb

@@ -0,0 +1,27 @@
+require 'process_shared/semaphore'
+
+module ProcessShared
+  # TODO: implement this
+  class ConditionVariable
+    def initialize
+      @sem = Semaphore.new
+    end
+
+    def broadcast
+      @sem.post
+    end
+
+    def signal
+      @sem.post
+    end
+
+    def wait(mutex, timeout = nil)
+      mutex.unlock
+      begin
+        @sem.wait
+      ensure
+        mutex.lock
+      end
+    end
+  end
+end

lib/process_shared/libc.rb

@@ -0,0 +1,36 @@
+require 'ffi'
+
+require 'process_shared/posix_call'
+require 'process_shared/psem'
+
+module ProcessShared
+  module LibC
+    extend FFI::Library
+    extend PosixCall
+
+    ffi_lib FFI::Library::LIBC
+
+    MAP_FAILED = FFI::Pointer.new(-1)
+    MAP_SHARED = PSem.map_shared
+    MAP_PRIVATE = PSem.map_private
+
+    PROT_READ = PSem.prot_read
+    PROT_WRITE = PSem.prot_write
+    PROT_EXEC = PSem.prot_exec
+    PROT_NONE = PSem.prot_none
+
+    O_RDWR = PSem.o_rdwr
+    O_CREAT = PSem.o_creat
+    O_EXCL = PSem.o_excl
+
+    attach_variable :errno, :int
+
+    attach_function :mmap, [:pointer, :size_t, :int, :int, :int, :off_t], :pointer
+    attach_function :munmap, [:pointer, :size_t], :int
+    attach_function :ftruncate, [:int, :off_t], :int
+    attach_function :close, [:int], :int
+
+    error_check(:mmap) { |v| v == MAP_FAILED }
+    error_check(:munmap, :ftruncate, :close)
+  end
+end

lib/process_shared/mutex.rb

@@ -0,0 +1,103 @@
+require 'process_shared/bounded_semaphore'
+require 'process_shared/with_self'
+require 'process_shared/shared_memory'
+require 'process_shared/process_error'
+
+module ProcessShared
+  class Mutex
+    include WithSelf
+
+    def self.open(&block)
+      new.with_self(&block)
+    end
+
+    def initialize
+      @internal_sem = BoundedSemaphore.new(1)
+      @locked_by = SharedMemory.new(:int)
+
+      @sem = BoundedSemaphore.new(1)
+    end
+
+    # @return [Mutex]
+    def lock
+      @sem.wait
+      self.locked_by = ::Process.pid
+      self
+    end
+
+    # @return [Boolean]
+    def locked?
+      locked_by > 0
+    end
+
+    # Releases the lock and sleeps timeout seconds if it is given and
+    # non-nil or forever.
+    #
+    # @return [Numeric]
+    def sleep(timeout = nil)
+      unlock
+      begin
+        timeout ? sleep(timeout) : sleep
+      ensure
+        lock
+      end
+    end
+
+    # @return [Boolean]
+    def try_lock
+      with_internal_lock do
+        if @locked_by.get_int(0) > 0
+          false                 # was locked
+        else
+          @sem.wait
+          self.locked_by = ::Process.pid
+          true
+        end
+      end
+    end
+
+    # @return [Mutex]
+    def unlock
+      if (p = locked_by) != ::Process.pid
+        raise ProcessError, "lock is held by #{p} not #{::Process.pid}"
+      end
+
+      @sem.post
+      self
+    end
+
+    # Acquire the lock, yield the block, then ensure the lock is
+    # unlocked.
+    def synchronize
+      lock
+      begin
+        yield
+      ensure
+        unlock
+      end
+    end
+
+    private
+
+    def locked_by
+      with_internal_lock do
+        @locked_by.get_int(0)
+      end
+    end
+
+    def locked_by=(val)
+      with_internal_lock do
+        @locked_by.put_int(0, val)
+      end
+    end
+
+    def with_internal_lock
+      @internal_sem.wait
+      begin
+        yield
+      ensure
+        @internal_sem.post
+      end
+    end
+  end
+end

lib/process_shared/posix_call.rb

@@ -0,0 +1,29 @@
+# require 'process_shared/libc' - circular dependency here...
+
+module ProcessShared
+  module PosixCall
+    # Replace methods in `syms` with error checking wrappers that
+    # invoke the original method and raise a SystemCallError with the
+    # current errno if the return value is an error.
+    #
+    # Errors are detected if the block returns true when called with
+    # the original method's return value.
+    def error_check(*syms, &is_err)
+      unless block_given?
+        is_err = lambda { |v| (v == -1) }
+      end
+
+      syms.each do |sym|
+        method = self.method(sym)
+        define_singleton_method(sym) do |*args|
+          ret = method.call(*args)
+          if is_err.call(ret)
+            raise SystemCallError.new("error in #{sym}", LibC.errno)
+          else
+            ret
+          end
+        end
+      end
+    end
+  end
+end

lib/process_shared/process_error.rb

@@ -0,0 +1,3 @@
+module ProcessShared
+  class ProcessError < Exception; end
+end

lib/process_shared/psem.rb

@@ -0,0 +1,109 @@
+require 'ffi'
+
+module ProcessShared
+  module PSem
+    class Error < FFI::Struct
+      layout(:source, :int,
+             :errno, :int)
+    end
+
+    extend FFI::Library
+
+    # Workaround FFI dylib/bundle issue.  See https://github.com/ffi/ffi/issues/42
+    suffix = if FFI::Platform.mac?
+               'bundle'
+             else
+               FFI::Platform::LIBSUFFIX
+             end
+
+    ffi_lib File.join(File.expand_path(File.dirname(__FILE__)),
+                      'libpsem.' + suffix)
+
+    class << self
+      # Replace methods in `syms` with error checking wrappers that
+      # invoke the original psem method and raise an appropriate
+      # error.
+      def psem_error_check(*syms)
+        syms.each do |sym|
+          method = self.method(sym)
+
+          block = lambda do |*args|
+            if method.call(*args) < 0
+              errp = args[-1]
+              unless errp.nil?
+                begin
+                  err = Error.new(errp.get_pointer(0))
+                  if err[:source] == PSem.e_source_system
+                    raise SystemCallError.new("error in #{sym}", err[:errno])
+                  else
+                    raise "error in #{sym}: #{err.get_integer(1)}"
+                  end
+                ensure
+                  psem_error_free(err)
+                end
+              end
+            end
+          end
+
+          define_method(sym, &block)
+          define_singleton_method(sym, &block)
+        end
+      end
+    end
+
+    # Generic constants
+
+    int_consts = [:o_rdwr,
+                  :o_creat,
+                  :o_excl,
+                  
+                  :prot_read,
+                  :prot_write,
+                  :prot_exec,
+                  :prot_none,
+                  
+                  :map_shared,
+                  :map_private]
+    int_consts.each { |sym| attach_variable sym, :int }
+
+    # Other constants, functions
+
+    attach_function :psem_error_free, :error_free, [:pointer], :void
+
+    attach_variable :e_source_system, :E_SOURCE_SYSTEM, :int
+    attach_variable :e_source_psem, :E_SOURCE_PSEM, :int
+
+    attach_variable :e_name_too_long, :E_NAME_TOO_LONG, :int
+
+    attach_variable :sizeof_psem_t, :size_t
+    attach_variable :sizeof_bsem_t, :size_t
+
+    # PSem functions
+
+    attach_function :psem_open, [:pointer, :string, :uint, :pointer], :int
+    attach_function :psem_close, [:pointer, :pointer], :int
+    attach_function :psem_unlink, [:string, :pointer], :int
+    attach_function :psem_post, [:pointer, :pointer], :int
+    attach_function :psem_wait, [:pointer, :pointer], :int
+    attach_function :psem_trywait, [:pointer, :pointer], :int
+    attach_function :psem_timedwait, [:pointer, :pointer, :pointer], :int
+    attach_function :psem_getvalue, [:pointer, :pointer, :pointer], :int
+
+    psem_error_check(:psem_open, :psem_close, :psem_unlink, :psem_post,
+                     :psem_wait, :psem_trywait, :psem_timedwait, :psem_getvalue)
+
+    # BSem functions
+    
+    attach_function :bsem_open, [:pointer, :string, :uint, :uint, :pointer], :int
+    attach_function :bsem_close, [:pointer, :pointer], :int
+    attach_function :bsem_unlink, [:string, :pointer], :int
+    attach_function :bsem_post, [:pointer, :pointer], :int
+    attach_function :bsem_wait, [:pointer, :pointer], :int
+    attach_function :bsem_trywait, [:pointer, :pointer], :int
+    attach_function :bsem_timedwait, [:pointer, :pointer, :pointer], :int
+    attach_function :bsem_getvalue, [:pointer, :pointer, :pointer], :int
+
+    psem_error_check(:bsem_open, :bsem_close, :bsem_unlink, :bsem_post,
+                     :bsem_wait, :bsem_trywait, :bsem_timedwait, :bsem_getvalue)
+  end
+end

lib/process_shared/rt.rb

@@ -0,0 +1,21 @@
+require 'process_shared/posix_call'
+require 'process_shared/psem'
+
+module ProcessShared
+  module RT
+    extend FFI::Library
+    extend PosixCall
+
+    # FIXME: mac and linux OK, but what about everything else?
+    if FFI::Platform.mac?
+      ffi_lib 'c'
+    else
+      ffi_lib 'rt'
+    end
+
+    attach_function :shm_open, [:string, :int, :mode_t], :int
+    attach_function :shm_unlink, [:string], :int
+
+    error_check :shm_open, :shm_unlink
+  end
+end

lib/process_shared/semaphore.rb

@@ -0,0 +1,60 @@
+require 'process_shared/psem'
+require 'process_shared/abstract_semaphore'
+
+module ProcessShared
+  class Semaphore < AbstractSemaphore
+    # With no associated block, open is a synonym for
+    # Semaphore.new. If the optional code block is given, it will be
+    # passed `sem` as an argument, and the Semaphore object will
+    # automatically be closed when the block terminates. In this
+    # instance, Semaphore.open returns the value of the block.
+    #
+    # @param [Integer] value the initial semaphore value
+    # @param [String] name not currently supported
+    def self.open(value = 1, name = nil, &block)
+      new(value, name).with_self(&block)
+    end
+
+    # Create a new semaphore with initial value `value`.  After
+    # Kernel#fork, the semaphore will be shared across two (or more)
+    # processes. The semaphore must be closed with #close in each
+    # process that no longer needs the semaphore.
+    #
+    # (An object finalizer is registered that will close the semaphore
+    # to avoid memory leaks, but this should be considered a last
+    # resort).
+    #
+    # @param [Integer] value the initial semaphore value
+    # @param [String] name not currently supported
+    def initialize(value = 1, name = nil)
+      init(PSem.sizeof_psem_t, 'psem', name) do |sem_name|
+        psem_open(sem, sem_name, value, err)
+      end
+    end
+
+    # Decrement the value of the semaphore.  If the value is zero,
+    # wait until another process increments via #post.
+    def wait
+      psem_wait(sem, err)
+    end
+
+    # Increment the value of the semaphore.  If other processes are
+    # waiting on this semaphore, one will be woken.
+    def post
+      psem_post(sem, err)
+    end
+
+    # Get the current value of the semaphore.
+    #
+    # @return [Integer] the current value of the semaphore.
+    def value
+      int = FFI::MemoryPointer.new(:int)
+      psem_getvalue(sem, int, err)
+      int.get_int(0)
+    end
+
+    def close
+      psem_close(sem, err)
+    end
+  end
+end

lib/process_shared/shared_memory.rb

@@ -0,0 +1,45 @@
+require 'process_shared/rt'
+require 'process_shared/libc'
+require 'process_shared/with_self'
+
+module ProcessShared
+  # Memory block shared across processes. TODO: finalizer that closes...
+  class SharedMemory < FFI::Pointer
+    include WithSelf
+
+    attr_reader :size, :fd
+
+    def self.open(size, &block)
+      new(size).with_self(&block)
+    end
+
+    def initialize(size)
+      @size = case size
+              when Symbol
+                FFI.type_size(size)
+              else
+                size
+              end
+
+      name = "/ps-shm#{rand(10000)}"
+      @fd = RT.shm_open(name,
+                        LibC::O_CREAT | LibC::O_RDWR | LibC::O_EXCL,
+                        0777)
+      RT.shm_unlink(name)
+      
+      LibC.ftruncate(@fd, @size)
+      @pointer = LibC.mmap(nil,
+                           @size,
+                           LibC::PROT_READ | LibC::PROT_WRITE,
+                           LibC::MAP_SHARED,
+                           @fd,
+                           0)
+      super(@pointer)
+    end
+
+    def close
+      LibC.munmap(@pointer, @size)
+      LibC.close(@fd)
+    end
+  end
+end

lib/process_shared/with_self.rb

@@ -0,0 +1,20 @@
+module ProcessShared
+  module WithSelf
+    # With no associated block, return self.  If the optional code
+    # block is given, it will be passed `self` as an argument, and the
+    # self object will automatically be closed (by invoking `close` on
+    # `self`) when the block terminates. In this instance, value of
+    # the block is returned.
+    def with_self
+      if block_given?
+        begin
+          yield self
+        ensure
+          self.close
+        end
+      else
+        self
+      end
+    end
+  end
+end