Ruby Language Queue 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? }