Module: Contrek::Concurrent::Poolable

Included in:
Finder
Defined in:
lib/contrek/finder/concurrent/poolable.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#number_of_threads ⇒ Object (readonly)

Returns the value of attribute number_of_threads.



6
7
8
 
# File 'lib/contrek/finder/concurrent/poolable.rb', line 6

def number_of_threads
  @number_of_threads
end

Instance Method Details

#enqueue!(**payload, &block) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/contrek/finder/concurrent/poolable.rb', line 20

def enqueue!(**payload, &block)
  if @number_of_threads > 0
    @threads << Thread.new do
      @semaphore.acquire
      begin
        block.call(payload)
      ensure
        @semaphore.release
      end
    end
  else
    block.call(payload)
  end
end

#initialize(number_of_threads: 0, **kwargs) ⇒ Object 



7
8
9
10
11
12
13
14
 
# File 'lib/contrek/finder/concurrent/poolable.rb', line 7

def initialize(number_of_threads: 0, **kwargs)
  @number_of_threads = number_of_threads || ::Concurrent.physical_processor_count
  if @number_of_threads > 0
    @threads = ::Concurrent::Array.new
    @semaphore = ::Concurrent::Semaphore.new(@number_of_threads)
  end
  super(**kwargs)
end

#wait! ⇒ Object 



16
17
18
 
# File 'lib/contrek/finder/concurrent/poolable.rb', line 16

def wait!
  @threads.each(&:join)
end