Fast Multi-Process Nonfair-Queue via FIFOs, leveraging PIPE_BUF

non-fair-queue

# Fast Multi-Process Nonfair-Queue via FIFOs, leveraging PIPE_BUF
# ===============================================================
#
# Provided `ulimit -a` indicates pipe size is 512 bytes This will repeat a
# message n times and confirm all the messages are received intact after pipe
# traversal (ie PIPE_BUF is respected).
#
# Increasing the message size by 1 (511 -> 512 for a total length of 513) will
# demonstrate how the messages are interleaved.
#
# Note that non-line buffering in the middle commands results in interleaved
# output as well - there are many subtle ways to achieve non-line buffering.
# To demonstrate make the message size small (msgsize=10) and try one of the
# following:
#
# * cat
# * sed
# * ruby -e 'while line = gets; $stdout.print line; end'
#
# Variations that do not buffer:
#
# * sed -l
# * ruby -e '$stdout.sync = true; while line = gets; $stdout.print line; end'
#
# Performance appears to be linearly related to the number of fifos.
#

rm -f {a,b,c,d,e,f,g,h,z}.fifo
mkfifo {a,b,c,d,e,f,g,h,z}.fifo

n=100000
msgsize=511
message=$(LC_CTYPE=C tr -d -c [:alnum:] < /dev/urandom | head -c $msgsize)

yes "$message" | head -n $n | tee {a,b,c,d,e,f,g,h}.fifo >/dev/null &
sed -l < a.fifo > z.fifo &
sed -l < b.fifo > z.fifo &
sed -l < c.fifo > z.fifo &
sed -l < d.fifo > z.fifo &
sed -l < e.fifo > z.fifo &
sed -l < f.fifo > z.fifo &
sed -l < g.fifo > z.fifo &
sed -l < h.fifo > z.fifo &
time cat z.fifo > /dev/null

# 800000 qffoEK4pBLlDFV....
#
# With `> /dev/null` instead of `count -u`
# real  0m4.800s ( > 160k msg/sec )
# user  0m2.626s
# sys 0m6.711s
#

rm -f {a,b,c,d,e,f,g,h,z}.fifo

non-fair-queue.rb

#!/usr/bin/env ruby
# Similar example, in ruby.  Note that w.sync is true by default. 
#
# time ./pipebuff_test > /dev/null
# real  0m3.708s
# user  0m3.619s
# sys 0m2.818s
#
# ./pipebuff_test | uniq -c
# 800000 sgdobseci ...
#

n = 100000
msgsize=511
message = (0...msgsize).map{ ('a'..'z').to_a[rand(26)] }.join
message = message + "\n"

r, w = IO.pipe
8.times do
  fork do
    r.close
    n.times do |i|
      w.write message
    end
  end
end

w.close
while line = r.gets
  if line == message
    $stdout.write line
  end
end