Class: Infrastruct::ThreadPool

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

Instance Method Summary collapse

Constructor Details

#initialize(runner, threads:) ⇒ ThreadPool

Returns a new instance of ThreadPool.



3
4
5
6
7
8
9
 
# File 'lib/infrastruct/thread_pool.rb', line 3

def initialize(runner, threads:)
  @runner = runner
  @number_of_threads = threads
  @queue = Infrastruct::BlockingQueue.new
  @results = Array.new
  @threads = []
end

Instance Method Details

#enqueue(args) ⇒ Object 



11
12
13
 
# File 'lib/infrastruct/thread_pool.rb', line 11

def enqueue(args)
  @queue.push(args)
end

#finalize ⇒ Object 



15
16
17
18
19
20
 
# File 'lib/infrastruct/thread_pool.rb', line 15

def finalize
  @queue.unblock!
  @threads.map(&:join)

  @runner.collect(@results)
end

#run ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/infrastruct/thread_pool.rb', line 22

def run
  mutex = Mutex.new

  @threads = @number_of_threads.times.map do |n|
    Thread.new do
      begin
        while args = @queue.pop do
          result = @runner.perform(args)

          mutex.synchronize do
            @results << result
          end
        end
      rescue ThreadError => error
        raise error unless error.message == 'queue empty'
      end
    end
  end
end