Class: Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/thread.rb

Overview

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

Direct Known Subclasses

SizedQueue

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Queue

Creates a new queue.



144
145
146
147
148
149
150
151
 
# File 'lib/thread.rb', line 144

def initialize
  @que = []
  @waiting = []
  @que.taint          # enable tainted communication
  @waiting.taint
  self.taint
  @mutex = Mutex.new
end

Instance Method Details

#clear ⇒ Object

Removes all objects from the queue.



217
218
219
 
# File 'lib/thread.rb', line 217

def clear
  @que.clear
end

#empty? ⇒ Boolean

Returns true if the queue is empty.

Returns:

  • (Boolean) 


210
211
212
 
# File 'lib/thread.rb', line 210

def empty?
  @que.empty?
end

#length ⇒ Object Also known as: size

Returns the length of the queue.



224
225
226
 
# File 'lib/thread.rb', line 224

def length
  @que.length
end

#num_waiting ⇒ Object

Returns the number of threads waiting on the queue.



236
237
238
 
# File 'lib/thread.rb', line 236

def num_waiting
  @waiting.size
end

#pop(non_block = false) ⇒ Object Also known as: shift, deq

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.



183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/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

#push(obj) ⇒ Object Also known as: <<, enq

Pushes obj to the queue.



156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/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