Xchan.rb

About

xchan.rb is an easy to use library for InterProcess Communication (IPC).

The library provides a channel that can help facilitate communication between Ruby processes who have a parent <=> child relationship. A channel lock is provided by lockf(3) and a temporary, unlinked file to protect against race conditions that can happen when multiple processes access the same channel at the same time.

Features

• Minimalist Inter-Process Communication (IPC) for parent <=> child processes.
• Channel-based communication.
• Support for multiple serializers (:marshal, :json, :yaml) and raw string communication (:pure).
• Blocking (#send, #recv) and non-blocking (#send_nonblock, #recv_nonblock) operations.
• Built-in file-based locking (lockf(3)) to prevent race conditions.
• Option to use a null lock for scenarios where locking is not needed.
• Access to underlying UNIX sockets for fine-grained control over socket options.
• Mac, BSD, and Linux support.
• Good docs.

Examples

Serialization

Options

The first argument provided to xchan is the serializer that should be used. A channel that will communicate purely in strings (in other words: without serialization) is available as xchan(:pure) - otherwise a wide range of serializers are available by default: xchan(:marshal), xchan(:json), and xchan(:yaml).

#!/usr/bin/env ruby
require "xchan"

##
# Marshal as the serializer
ch = xchan(:marshal)
Process.wait fork { ch.send(5) }
puts "#{ch.recv} + 7 = 12"
ch.close

##
# 5 + 7 = 12

Read operations

#recv

The ch.recv method performs a blocking read. A read can block when a lock is held by another process, or when a read from Chan::UNIXSocket#r blocks. The example performs a read that blocks until the parent process writes to the channel:

#!/usr/bin/env ruby
require "xchan"

ch = xchan(:marshal)
fork do
  print "Received a random number (child process): ", ch.recv, "\n"
end
sleep(1)
puts "Send a random number (from parent process)"
ch.send(rand(21))
ch.close
Process.wait

##
# Send a random number (from parent process)
# Received random number (child process): XX

#recv_nonblock

The non-blocking counterpart to #recv is #recv_nonblock. The #recv_nonblock method raises Chan::WaitLockable when a read blocks because of a lock held by another process, and the method raises Chan::WaitReadable when a read from Chan::UNIXSocket#r blocks:

#!/usr/bin/env ruby
require "xchan"

def read(ch)
  ch.recv_nonblock
rescue Chan::WaitReadable
  puts "Wait 1 second for channel to be readable"
  ch.wait_readable(1)
  retry
rescue Chan::WaitLockable
  puts "Wait 1 second for channel to be lockable"
  ch.wait_lockable(1)
  retry
end
trap("SIGINT") { exit(1) }
read(xchan(:marshal))

##
# Wait 1 second for channel to be readable
# Wait 1 second for channel to be readable
# ^C

Write operations

#send

The ch.send method performs a blocking write. A write can block when a lock is held by another process, or when a write to Chan::UNIXSocket#w blocks. The example fills the send buffer:

#!/usr/bin/env ruby
require "xchan"

ch = xchan(:marshal, sock: Socket::SOCK_STREAM)
sndbuf = ch.w.getsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF)
while ch.bytes_sent <= sndbuf.int
  ch.send(1)
end

#send_nonblock

The non-blocking counterpart to #send is #send_nonblock. The #send_nonblock method raises Chan::WaitLockable when a write blocks because of a lock held by another process, and the method raises Chan::WaitWritable when a write to Chan::UNIXSocket#w blocks. The example frees space on the send buffer:

#!/usr/bin/env ruby
require "xchan"

def send_nonblock(ch, buf)
  ch.send_nonblock(buf)
rescue Chan::WaitWritable
  puts "Blocked - free send buffer"
  ch.recv
  retry
rescue Chan::WaitLockable
  ch.wait_lockable
  retry
end

ch = xchan(:marshal, sock: Socket::SOCK_STREAM)
sndbuf = ch.w.getsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF)
while ch.bytes_sent <= sndbuf.int
  send_nonblock(ch, 1)
end

##
# Blocked - free send buffer

Lock

File

The default lock for a channel is a file lock. The locking mechanism is implemented with the lockf function from the C standard library. Nothing special has to be done to use it, and it allows a channel to be safely accessed across multiple processes:

#!/usr/bin/env ruby
require "xchan"

##
# 'lock: :file' is added just for the example
# It is the default behavior, and not necessary
ch = xchan(:marshal, lock: :file)
5.times.map do
  fork do
    ch.send(5)
  end
end.each { Process.wait(_1) }

Null

The null lock is the same as using no lock at all. The null lock is implemented as a collection of no-op operations. The null lock is implemented in the Chan::NullLock class, and in certain situations, it can be useful and preferable to using a file lock:

#!/usr/bin/env ruby
require "xchan"

ch = xchan(:marshal, lock: :null)
fork do
  ch.send(5)
end
Process.wait

Socket

Options

A channel has one socket for read operations and another socket for write operations. Chan::UNIXSocket#r returns the socket used for read operations, and Chan::UNIXSocket#w returns the socket used for write operations:

#!/usr/bin/env ruby
require "xchan"
ch = xchan(:marshal)

##
# Print the value of SO_RCVBUF
rcvbuf = ch.r.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF)
print "The read buffer can contain a maximum of: ", rcvbuf.int, " bytes.\n"

##
# Print the value of SO_SNDBUF
sndbuf = ch.w.getsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF)
print "The maximum size of a single message is: ", sndbuf.int, " bytes.\n"

##
# The read buffer can contain a maximum of: 16384 bytes.
# The maximum size of a single message is: 2048 bytes.

Documentation

A complete API reference is available at 0x1eef.github.io/x/xchan.rb

Install

xchan.rb can be installed via rubygems.org:

gem install xchan.rb

Sources

• github.com/@0x1eef
• gitlab.com/@0x1eef

License

BSD Zero Clause <br> See share/xchan.rb/LICENSE