bm_improved_buffered_channel.cr

# Buffered channel, using a queue.
class Channel::Buffered2(T) < Channel(T)
  def initialize(@capacity = 32)
    @queue = Deque(T).new(@capacity)
    super()
  end

  # Send a value to the channel.
  def send(value : T)
    while full?
      raise_if_closed
      @senders << Fiber.current
      Crystal::Scheduler.reschedule
    end

    raise_if_closed

    @queue << value
    if receiver = @receivers.shift?
      Crystal::Scheduler.enqueue receiver
    end
    self
  end

  private def receive_impl
    while empty?
      yield if @closed
      @receivers << Fiber.current
      Crystal::Scheduler.reschedule
    end

    @queue.shift.tap do
      if sender = @senders.shift?
        Crystal::Scheduler.enqueue sender
      end
    end
  end

  def full?
    @queue.size >= @capacity
  end

  def empty?
    @queue.empty?
  end
end

require "benchmark"

ch1 = Channel::Buffered(Nil).new(1)
ch2 = Channel::Buffered2(Nil).new(1)
100.times do
  spawn do
    loop do
      ch1.send nil
    end
  end
end

100.times do
  spawn do
    loop do
      ch2.send nil
    end
  end
end

Benchmark.ips do |x|
  x.report("original") { ch1.receive  }
  x.report("improved") { ch2.receive  }
end