AI Code

Ruby Language Queue


  • q =
  • q.push object
  • q << object # same as #push
  • q.pop #=> object

Multiple Workers One Sink

We want to gather data created by multiple Workers.

First we create a Queue:

sink =

Then 16 workers all generating a random number and pushing it into sink:

(1..16) do do
    sink << rand(1..100)

And to get the data, convert a Queue to an Array:

data = [].tap { |a| a << sink.pop until sink.empty? }

One Source Multiple Workers

We want to process data in parallel.

Let's populate source with some data:

source =
data = (1..100)
data.each { |e| source << e }

Then create some workers to process data:

(1..16) do do
    until source.empty?
      item = source.pop
      sleep 0.5
      puts "Processed: #{item}"

One Source - Pipeline of Work - One Sink

We want to process data in parallel and push it down the line to be processed by other workers.

Since Workers both consume and produce data we have to create two queues:

first_input_source =
first_output_sink  =
100.times { |i| first_input_source << i }

First wave of workers read an item from first_input_source, process the item, and write results in first_output_sink:

(1..16) do do
    loop do
      item = first_input_source.pop
      first_output_source << item ** 2
      first_output_source << item ** 3

Second wave of workers uses first_output_sink as its input source and reads, process then writes to another output sink:

second_input_source = first_output_sink
second_output_sink  =

(1..32) do do
    loop do
      item = second_input_source.pop
      second_output_sink << item * 2
      second_output_sink << item * 3

Now second_output_sink is the sink, let's convert it to an array:

sleep 5 # workaround in place of synchronization
sink = second_output_sink
[].tap { |a| a << sink.pop until sink.empty? }

Pushing Data into a Queue - #push

q =
q << "any object including another queue"
# or
q.push :data
  • There is no high water mark, queues can infinitely grow.
  • #push never blocks

Pulling Data from a Queue - #pop

q =
q << :data
q.pop #=> :data
  • #pop will block until there is some data available.
  • #pop can be used for synchronization.

Synchronization - After a Point in Time

syncer =

a = do
  puts "this happens at end"

b = do
  puts "this happens first"
  syncer << :ok

[a, b].map(&:join)

Converting a Queue into an Array

q =
q << 1
q << 2

a =
a << q.pop until q.empty?

Or a one liner:

[].tap { |array| array < queue.pop until queue.empty? }

Merging Two Queues

  • To avoid infinitely blocking, reading from queues shouldn't happen on the thread merge is happening on.
  • To avoid synchronization or infinitely waiting for one of queues while other has data, reading from queues shouldn't happen on same thread.

Let's start by defining and populating two queues:

q1 =
q2 =
(1..100).each { |e| q1 << e }
(101..200).each { |e| q2 << e }

We should create another queue and push data from other threads into it:

merged =

[q1, q2].map do |q| do
    loop do
      merged << q.pop

If you know you can completely consume both queues (consumption speed is higher than production, you won't run out of RAM) there is a simpler approach:

merged =
merged << q1.pop until q1.empty?
merged << q2.pop until q2.empty?

Got any Ruby Language Question?