Queue

This class provides a way to synchronize communication between threads.

Example:

require 'thread'
queue = Queue.new
producer = Thread.new do
 5.times do |i|
 sleep rand(i) # simulate expense
 queue << i
 puts "#{i} produced"
 end
end
consumer = Thread.new do
 5.times do |i|
 value = queue.pop
 sleep rand(i/2) # simulate expense
 puts "consumed #{value}"
 end
end
consumer.join

Public Class Methods

new() click to toggle source

Creates a new queue.

 
 # File thread.rb, line 144
def initialize
 @que = []
 @waiting = []
 @que.taint # enable tainted communication
 @waiting.taint
 self.taint
 @mutex = Mutex.new
end

Public Instance Methods

<<(obj) click to toggle source

Alias of push

Alias for: push

clear() click to toggle source

Removes all objects from the queue.

 
 # File thread.rb, line 217
def clear
 @que.clear
end

deq(non_block=false) click to toggle source

Alias of pop

Alias for: pop

empty?() click to toggle source

Returns true if the queue is empty.

 
 # File thread.rb, line 210
def empty?
 @que.empty?
end

enq(obj) click to toggle source

Alias of push

Alias for: push

length() click to toggle source

Returns the length of the queue.

 
 # File thread.rb, line 224
def length
 @que.length
end

Also aliased as: size

num_waiting() click to toggle source

Returns the number of threads waiting on the queue.

 
 # File thread.rb, line 236
def num_waiting
 @waiting.size
end

pop(non_block=false) click to toggle source

Retrieves data from the queue. If the queue is empty, the calling thread is suspended until data is pushed onto the queue. If non_block is true, the thread isn't suspended, and an exception is raised.

 
 # File thread.rb, line 183
def pop(non_block=false)
 @mutex.synchronize{
 while true
 if @que.empty?
 raise ThreadError, "queue empty" if non_block
 @waiting.push Thread.current
 @mutex.sleep
 else
 return @que.shift
 end
 end
 }
end

Also aliased as: shift, deq

push(obj) click to toggle source

Pushes obj to the queue.

 
 # File thread.rb, line 156
def push(obj)
 @mutex.synchronize{
 @que.push obj
 begin
 t = @waiting.shift
 t.wakeup if t
 rescue ThreadError
 retry
 end
 }
end

Also aliased as: <<, enq

shift(non_block=false) click to toggle source

Alias of pop

Alias for: pop

size() click to toggle source

Alias of length.

Alias for: length