Channel implementation (mutex mvars + thread local storage)

channel_mvar_cond.rb

require "thread"

module Axon
  class Mvar
    def initialize
      @lock    = Mutex.new
      @cond    = ConditionVariable.new
      @value   = nil
      @was_set = false
    end

    def set(value)
      @lock.lock
      @value = value
      @was_set = true
      @cond.signal
      @lock.unlock
    end

    def get
      @lock.lock
      unless @was_set
        @cond.wait(@lock)
      end
      @was_set = false
      @lock.unlock
      @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

    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.480000   0.190000   0.670000 (  0.545337)
20k msg (buf=  1),   1 snd,   1 rcv       0.280000   0.150000   0.430000 (  0.345259)
20k msg (buf= 10),   1 snd,   1 rcv       0.080000   0.010000   0.090000 (  0.082841)
20k msg (buf=  0),  10 snd,   1 rcv       0.270000   0.270000   0.540000 (  0.427491)
20k msg (buf=  1),  10 snd,   1 rcv       0.360000   0.370000   0.730000 (  0.555719)
20k msg (buf= 10),  10 snd,   1 rcv       0.220000   0.150000   0.370000 (  0.269784)
20k msg (buf=  0),   1 snd,  10 rcv       0.420000   0.380000   0.800000 (  0.599846)
20k msg (buf=  1),   1 snd,  10 rcv       0.440000   0.330000   0.770000 (  0.563077)
20k msg (buf= 10),   1 snd,  10 rcv       0.170000   0.220000   0.390000 (  0.291597)
20k msg (buf=  0),  10 snd,  10 rcv       0.400000   0.710000   1.110000 (  0.531021)
20k msg (buf=  1),  10 snd,  10 rcv       0.310000   0.330000   0.640000 (  0.490750)
>