Queue Handoff

A queue lets one thread hand values to another thread safely.
queue handoff `Queue#push` and `Queue#pop` coordinate producer and consumer code without manual locking.
                            item_count
                            item_count = 3 (default)item_count = 2item_count = 4
                        

                            queue_handoff.rb
                        
Replay: real traced execution (multi-file project)

                            item_count = 3
queue = Queue.new
received = []

producer = Thread.new do
  item_count.times do |index|
    queue.push("item#{index + 1}")
  end
  queue.push(:done)
end

consumer = Thread.new do
  loop do
    item = queue.pop
    break if item == :done

    received << item
  end
end

producer.join
consumer.join

puts "item_count=#{item_count}"
puts "received=#{received.join("/")}"
puts "complete=#{received.length == item_count}"
item_count = 2
queue = Queue.new
received = []

producer = Thread.new do
  item_count.times do |index|
    queue.push("item#{index + 1}")
  end
  queue.push(:done)
end

consumer = Thread.new do
  loop do
    item = queue.pop
    break if item == :done

    received << item
  end
end

producer.join
consumer.join

puts "item_count=#{item_count}"
puts "received=#{received.join("/")}"
puts "complete=#{received.length == item_count}"
item_count = 4
queue = Queue.new
received = []

producer = Thread.new do
  item_count.times do |index|
    queue.push("item#{index + 1}")
  end
  queue.push(:done)
end

consumer = Thread.new do
  loop do
    item = queue.pop
    break if item == :done

    received << item
  end
end

producer.join
consumer.join

puts "item_count=#{item_count}"
puts "received=#{received.join("/")}"
puts "complete=#{received.length == item_count}"

                        
1item_count ← 3, queue ← ⟨Thread::Queue A⟩, received ← [], producer ← ⟨Thread B /tmp/execution/⟨tmp C⟩.rb:49 run⟩1item_count→ 3 = 3  #@item_count=2, 42queue→ ⟨Thread::Queue A⟩ = Queue.new3received→ [] = []45producer→ ⟨Thread B /tmp/execution/⟨tmp C⟩.rb:49 run⟩ = Thread.new do6  item_count.times do |index|7    queue.push("item#{index + 1}")8  end9  queue.push(:done)10end1112consumer→ ⟨Thread D /tmp/execution/⟨tmp C⟩.rb:85 run⟩ = Thread.new do13  loop do14    item = queue.pop15    break if item == :done1617    received << item18  end19end⋯
2do⋯5producer = Thread.new do6  item_count3.times do |index|7    queue.push("item#{index + 1}")8  end9  queue⟨Thread::Queue A⟩.push(:done)10end⋯
3do |index|pass 1 of 3⋯5producer = Thread.new do6  item_count.times do |index0|7    queue⟨Thread::Queue A⟩.push("item#{index0 + 1}")8  end⋯All 3 passes — pass 1 is the card abovepassindex
10
21
32
4queue.push(:done)⋯5producer = Thread.new do6  item_count3.times do |index|7    queue.push("item#{index + 1}")8  end9  queue⟨Thread::Queue A⟩.push(:done)10end⋯
5item ← item1pass 1 of 4⋯12consumer = Thread.new do13  loop do14    item→ item1 = queue⟨Thread::Queue A⟩.pop15    break if itemitem1 == :done1617    received[] << itemitem118  end⋯All 4 passes — pass 1 is the card abovepassreceiveditem
1[]item1
2["item1"]item2
3["item1", "item2"]item3
4["item1", "item2", "item3"]done
6puts "item_count=#{item_count}"⋯24puts "item_count=#{item_count3}"25puts "received=#{received.join("/")item1/item2/item3}"26puts "complete=#{received.length3 == item_count3}"⋯outputitem_count=3
received=item1/item2/item3
complete=true
1item_count ← 2, queue ← ⟨Thread::Queue A⟩, received ← [], producer ← ⟨Thread B /tmp/execution/⟨tmp C⟩.rb:49 run⟩1item_count→ 2 = 22queue→ ⟨Thread::Queue A⟩ = Queue.new3received→ [] = []45producer→ ⟨Thread B /tmp/execution/⟨tmp C⟩.rb:49 run⟩ = Thread.new do6  item_count.times do |index|7    queue.push("item#{index + 1}")8  end9  queue.push(:done)10end1112consumer→ ⟨Thread D /tmp/execution/⟨tmp C⟩.rb:85 run⟩ = Thread.new do13  loop do14    item = queue.pop15    break if item == :done1617    received << item18  end19end⋯
2do⋯5producer = Thread.new do6  item_count2.times do |index|7    queue.push("item#{index + 1}")8  end9  queue⟨Thread::Queue A⟩.push(:done)10end⋯
3do |index|pass 1 of 2⋯5producer = Thread.new do6  item_count.times do |index0|7    queue⟨Thread::Queue A⟩.push("item#{index0 + 1}")8  end⋯
4do |index|pass 2 of 2⋯5producer = Thread.new do6  item_count.times do |index1|7    queue⟨Thread::Queue A⟩.push("item#{index1 + 1}")8  end⋯
5queue.push(:done)⋯5producer = Thread.new do6  item_count2.times do |index|7    queue.push("item#{index + 1}")8  end9  queue⟨Thread::Queue A⟩.push(:done)10end⋯
6item ← item1pass 1 of 3⋯12consumer = Thread.new do13  loop do14    item→ item1 = queue⟨Thread::Queue A⟩.pop15    break if itemitem1 == :done1617    received[] << itemitem118  end⋯All 3 passes — pass 1 is the card abovepassreceiveditem
1[]item1
2["item1"]item2
3["item1", "item2"]done
7puts "item_count=#{item_count}"⋯24puts "item_count=#{item_count2}"25puts "received=#{received.join("/")item1/item2}"26puts "complete=#{received.length2 == item_count2}"⋯outputitem_count=2
received=item1/item2
complete=true
1item_count ← 4, queue ← ⟨Thread::Queue A⟩, received ← [], producer ← ⟨Thread B /tmp/execution/⟨tmp C⟩.rb:49 run⟩1item_count→ 4 = 42queue→ ⟨Thread::Queue A⟩ = Queue.new3received→ [] = []45producer→ ⟨Thread B /tmp/execution/⟨tmp C⟩.rb:49 run⟩ = Thread.new do6  item_count.times do |index|7    queue.push("item#{index + 1}")8  end9  queue.push(:done)10end1112consumer→ ⟨Thread D /tmp/execution/⟨tmp C⟩.rb:85 run⟩ = Thread.new do13  loop do14    item = queue.pop15    break if item == :done1617    received << item18  end19end⋯
2do⋯5producer = Thread.new do6  item_count4.times do |index|7    queue.push("item#{index + 1}")8  end9  queue⟨Thread::Queue A⟩.push(:done)10end⋯
3do |index|pass 1 of 4⋯5producer = Thread.new do6  item_count.times do |index0|7    queue⟨Thread::Queue A⟩.push("item#{index0 + 1}")8  end⋯All 4 passes — pass 1 is the card abovepassindex
10
21
32
43
4queue.push(:done)⋯5producer = Thread.new do6  item_count4.times do |index|7    queue.push("item#{index + 1}")8  end9  queue⟨Thread::Queue A⟩.push(:done)10end⋯
5item ← item1pass 1 of 5⋯12consumer = Thread.new do13  loop do14    item→ item1 = queue⟨Thread::Queue A⟩.pop15    break if itemitem1 == :done1617    received[] << itemitem118  end⋯All 5 passes — pass 1 is the card abovepassreceiveditem
1[]item1
2["item1"]item2
3["item1", "item2"]item3
4["item1", "item2", "item3"]item4
5["item1", "item2", "item3", "item4"]done
6puts "item_count=#{item_count}"⋯24puts "item_count=#{item_count4}"25puts "received=#{received.join("/")item1/item2/item3/item4}"26puts "complete=#{received.length4 == item_count4}"⋯outputitem_count=4
received=item1/item2/item3/item4
complete=true
pass	`received`	`item`
1	[]	item1
2	["item1"]	item2
3	["item1", "item2"]	item3
4	["item1", "item2", "item3"]	done
All 3 passes — pass 1 is the card above
pass	`index`
1	0
2	1
3	2
All 4 passes — pass 1 is the card above
pass	`index`
1	0
2	1
3	2
4	3