Class: Concurrent::Actor::Utils::Pool

Inherits:
RestartingContext show all
Defined in:
lib/concurrent-ruby-edge/concurrent/actor/utils/pool.rb

Overview

Allows to create a pool of workers and distribute work between them

Examples:

class Worker < Concurrent::Actor::RestartingContext
  def on_message(message)
    p message * 5
  end
end

pool = Concurrent::Actor::Utils::Pool.spawn! 'pool', 5 do |index|
  Worker.spawn name: "worker-#{index}", supervise: true, args: []
end

pool << 'asd' << 2
# prints:
# "asdasdasdasdasd"
# 10

Yields:

  • (balancer, index) —

    a block spawning an worker instance. called +size+ times. The worker should be descendant of AbstractWorker and supervised, see example.

Yield Parameters:

  • balancer (Balancer) —

    to pass to the worker

  • index (Integer) —

    of the worker, usually used in its name

Yield Returns:

  • (Reference) —

    the reference of newly created worker

Instance Method Summary collapse

Constructor Details

#initialize(size, &worker_initializer) ⇒ Pool

Returns a new instance of Pool.



31
32
33
34
35
36
37
38
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/pool.rb', line 31

def initialize(size, &worker_initializer)
  @balancer = Balancer.spawn name: :balancer, supervise: true
  @workers  = ::Array.new(size, &worker_initializer)
  @workers.each do |worker|
    Type! worker, Reference
    @balancer << [:subscribe, worker]
  end
end

Instance Method Details

#on_message(message) ⇒ undocumented 



40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/pool.rb', line 40

def on_message(message)
  command, _ = message
  return if [:restarted, :reset, :resumed, :terminated].include? command # ignore events from supervised actors

  envelope_to_redirect = if envelope.future
                           envelope
                         else
                           Envelope.new(envelope.message, Promises.resolvable_future, envelope.sender, envelope.address)
                         end
  envelope_to_redirect.future.on_fulfillment! { @balancer << :subscribe } # TODO check safety of @balancer reading
  redirect @balancer, envelope_to_redirect
end