Class: WorkQueue

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

Overview

WorkQueue

Description

A tunable work queue, designed to coordinate work between a producer and a pool of worker threads.

Usage

wq = WorkQueue.new
wq.enqueue_b { puts "Hello from the WorkQueue" }
wq.join

Constant Summary collapse

VERSION = 
"2.0.1"

Instance Method Summary collapse

Constructor Details

#initialize(max_threads = nil, max_tasks = nil) ⇒ WorkQueue

Creates a new work queue with the desired parameters.

wq = WorkQueue.new(5,10,20)



44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/work_queue.rb', line 44

def initialize(max_threads=nil, max_tasks=nil)
    self.max_threads = max_threads
    self.max_tasks = max_tasks
    @threads = Array.new
    @threads.extend(MonitorMixin)
    @threads_waiting = 0
    @tasks = Array.new
    @tasks.extend(MonitorMixin)
    @task_enqueued = @tasks.new_cond
    @task_completed = @tasks.new_cond
    @cur_tasks = 0
end

Instance Method Details

#cur_tasks ⇒ Object

Returns the current number of active tasks. This value is just a snapshot, and may change immediately upon returning.

wq = WorkQueue.new(1)
wq.enqueue_b { sleep(1) }
wq.cur_tasks		#=> 0
wq.enqueue_b {}
wq.cur_tasks		#=> 1



106
107
108
 
# File 'lib/work_queue.rb', line 106

def cur_tasks
    @cur_tasks
end

#cur_threads ⇒ Object

Returns the current number of worker threads. This value is just a snapshot, and may change immediately upon returning.

wq = WorkQueue.new(10)
wq.cur_threads		#=> 0
wq.enqueue_b {}
wq.cur_threads		#=> 1



79
80
81
 
# File 'lib/work_queue.rb', line 79

def cur_threads
    @threads.size
end

#enqueue_b(*args, &block) ⇒ Object

Schedules the given Block for future execution by a worker thread. If there is no space left in the queue, waits until space becomes available.

wq = WorkQueue.new(1)
wq.enqueue_b {}



128
129
130
 
# File 'lib/work_queue.rb', line 128

def enqueue_b(*args, &block)
   enqueue(block, args) 
end

#enqueue_p(proc, *args) ⇒ Object

Schedules the given Proc for future execution by a worker thread. If there is no space left in the queue, waits until space becomes available.

wq = WorkQueue.new(1)
wq.enqueue_p(Proc.new {})



117
118
119
 
# File 'lib/work_queue.rb', line 117

def enqueue_p(proc, *args)
    enqueue(proc, args) 
end

#join ⇒ Object

Waits until the tasks queue is empty and all worker threads have finished.

wq = WorkQueue.new(1)
wq.enqueue_b { sleep(1) }
wq.join



139
140
141
142
143
 
# File 'lib/work_queue.rb', line 139

def join
    @tasks.synchronize do
        @task_completed.wait_while { cur_tasks > 0 }
    end
end

#kill ⇒ Object

Stops all worker threads immediately, aborting any ongoing tasks.

wq = WorkQueue.new(1)
wq.enqueue_b { sleep(1) }
wq.kill



152
153
154
155
156
157
158
159
160
 
# File 'lib/work_queue.rb', line 152

def kill
    @tasks.synchronize do
        @threads.dup.each { |thread| thread.exit.join }
        @threds.clear
        @threads_waiting = 0
        @tasks.clear
        @cur_tasks = 0
    end
end

#max_tasks ⇒ Object

Returns the maximum number of queued tasks. This value is set upon initialization and cannot be changed afterwards.

wq = WorkQueue.new()
wq.max_tasks		#=> Infinity
wq = WorkQueue.new(nil,1)
wq.max_tasks		#=> 1



92
93
94
 
# File 'lib/work_queue.rb', line 92

def max_tasks
    @max_tasks
end

#max_threads ⇒ Object

Returns the maximum number of worker threads. This value is set upon initialization and cannot be changed afterwards.

wq = WorkQueue.new()
wq.max_threads		#=> Infinity
wq = WorkQueue.new(1)
wq.max_threads		#=> 1



66
67
68
 
# File 'lib/work_queue.rb', line 66

def max_threads
    @max_threads
end