Class: Chef::Util::ThreadedJobQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/chef/util/threaded_job_queue.rb

Overview

A simple threaded job queue

Create a queue:

queue = ThreadedJobQueue.new

Add jobs:

queue << lambda { |lock| foo.the_bar }

A job is a callable that optionally takes a Mutex instance as its only parameter.

Then start processing jobs with +n+ threads:

queue.process(n)

Instance Method Summary collapse

Constructor Details

#initializeThreadedJobQueue

Returns a new instance of ThreadedJobQueue


38
39
40
41
# File 'lib/chef/util/threaded_job_queue.rb', line 38

def initialize
  @queue = Queue.new
  @lock = Mutex.new
end

Instance Method Details

#<<(job) ⇒ Object


43
44
45
# File 'lib/chef/util/threaded_job_queue.rb', line 43

def <<(job)
  @queue << job
end

#process(concurrency = 10) ⇒ Object


47
48
49
50
51
52
53
54
55
56
57
58
# File 'lib/chef/util/threaded_job_queue.rb', line 47

def process(concurrency = 10)
  workers = (1..concurrency).map do
    Thread.new do
      loop do
        fn = @queue.pop
        fn.arity == 1 ? fn.call(@lock) : fn.call
      end
    end
  end
  workers.each { |worker| self << Thread.method(:exit) }
  workers.each { |worker| worker.join }
end