Channel implementation (pipe mvars + thread local storage)

channel_mvar_pipe.rb

require "thread"

module Axon
  class Mvar
    SENTINEL = 'x'.freeze

    def initialize
      @pipe    = IO.pipe
      @value   = nil
      @was_set = false
    end

    def set(value)
      @value = value
      @pipe[1].write(SENTINEL)
    end

    def get
      @pipe[0].read(1)
      @value
    end
  end
end

module Axon
  class Channel

    attr_reader :buffer_size

    def initialize(buffer_size=0)
      @buffer_size  = buffer_size
      @buffer       = []
      @inboxes      = {}
      @chan_lock    = Mutex.new
      @senders      = []
      @receivers    = []
      @mvars        = {}
      @mvar_lock    = Mutex.new
    end

    def send(value)
      receiver = nil
      mvar = nil

      @chan_lock.synchronize do
        if @receivers.size > 0
          receiver = @receivers.shift
        elsif @buffer.size < @buffer_size
          @buffer << value
        else
          mvar = get_thread_mvar()
          @senders << [mvar, value]
        end
      end

      receiver.set(value) if receiver
      mvar.get if mvar

      nil
    end

    def receive
      sender, value, mvar = nil, nil, nil

      @chan_lock.synchronize do
        if @buffer.size > 0
          value = @buffer.shift
          if @senders.size > 0
            sender, new_value = @senders.shift
            @buffer << new_value
          end
        elsif @senders.size > 0
          sender, value = @senders.shift
        else
          mvar = get_thread_mvar()
          @receivers << mvar
        end
      end

      sender.set(nil) if sender
      value = mvar.get if mvar

      value
    end

    def close
    end

    private

    # HACK
    def get_thread_mvar
      mvar = @mvars[Thread.current]
      unless mvar
        @mvars[Thread.current] = Axon::Mvar.new
      end
      @mvars[Thread.current]
    end
  end
end

perf

> ruby example/benchmark.rb 
                                              user     system      total        real
20k msg (buf=  0),   1 snd,   1 rcv       0.350000   0.290000   0.640000 (  0.515689)
20k msg (buf=  1),   1 snd,   1 rcv       0.330000   0.280000   0.610000 (  0.480422)
20k msg (buf= 10),   1 snd,   1 rcv       0.110000   0.030000   0.140000 (  0.124402)
20k msg (buf=  0),  10 snd,   1 rcv       0.340000   0.230000   0.570000 (  0.428301)
20k msg (buf=  1),  10 snd,   1 rcv       0.300000   0.470000   0.770000 (  0.556495)
20k msg (buf= 10),  10 snd,   1 rcv       0.360000   0.270000   0.630000 (  0.467264)
20k msg (buf=  0),   1 snd,  10 rcv       0.370000   0.330000   0.700000 (  0.501613)
20k msg (buf=  1),   1 snd,  10 rcv       0.340000   0.240000   0.580000 (  0.420689)
20k msg (buf= 10),   1 snd,  10 rcv       0.370000   0.370000   0.740000 (  0.547012)
20k msg (buf=  0),  10 snd,  10 rcv       0.360000   0.340000   0.700000 (  0.427516)
20k msg (buf=  1),  10 snd,  10 rcv       0.250000   0.340000   0.590000 (  0.378542)
>