Class: FutureProof::ThreadPool

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

Overview

ThreadPool could be used to schedule and group threads.

Instance Method Summary collapse

Constructor Details

#initialize(size) ⇒ ThreadPool

Initializes a new thread pool.



10
11
12
13
14
15
 
# File 'lib/future_proof/thread_pool.rb', line 10

def initialize(size)
  @size    = size
  @threads = []
  @queue   = Queue.new
  @values  = FutureProof::FutureQueue.new
end

Instance Method Details

#finalize ⇒ Object

Flags that after all pool jobs are processed thread pool should stop the reactor.



50
51
52
 
# File 'lib/future_proof/thread_pool.rb', line 50

def finalize
  @size.times { @queue.push :END_OF_WORK }
end

#finalize! ⇒ Object

Commands to remove all pool tasks and finishes the execution after all running tasks are completed.



70
71
72
73
 
# File 'lib/future_proof/thread_pool.rb', line 70

def finalize!
  @queue.clear
  finalize
end

#perform ⇒ Object

Note:

Can be restarted after finalization.

Starts execution of the thread pool.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/future_proof/thread_pool.rb', line 32

def perform
  unless @threads.any? { |t| t.alive? }
    @values.start!
    @size.times do
      @threads << Thread.new do
        while job = @queue.pop
          if job == :END_OF_WORK
            break
          else
            @values.push *job[1], &job[0]
          end
        end
      end
    end
  end
end

#submit(*args, &block) ⇒ Object

Note:

Does not start the execution until #perform is called.

Submits a task to a thread pool.

Examples:

thread_pool.submit(25, 2) { |a, b| a ** b }

Parameters:

  • *args (Array<Object>) —

    job arguments.



25
26
27
 
# File 'lib/future_proof/thread_pool.rb', line 25

def submit(*args, &block)
  @queue.push [block, args]
end

#values ⇒ FutureProof::FutureArray

Calls #wait and returns array of all calculated values.

Returns:



63
64
65
66
67
 
# File 'lib/future_proof/thread_pool.rb', line 63

def values
  wait
  @values.stop!
  @values.values
end

#wait ⇒ Object

Calls #finalize and blocks programm flow until all jobs are processed.



55
56
57
58
 
# File 'lib/future_proof/thread_pool.rb', line 55

def wait
  finalize
  @threads.map &:join
end