Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
# 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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.