class Pallets::Manager

Attributes

scheduler[R]
workers[R]

Public Class Methods

new(concurrency: Pallets.configuration.concurrency) click to toggle source
# File lib/pallets/manager.rb, line 5
def initialize(concurrency: Pallets.configuration.concurrency)
  @backend = Pallets.backend
  @workers = concurrency.times.map { Worker.new(self, @backend) }
  @scheduler = Scheduler.new(self, @backend)
  @lock = Mutex.new
  @needs_to_stop = false
end

Public Instance Methods

remove_worker(worker) click to toggle source
# File lib/pallets/manager.rb, line 42
def remove_worker(worker)
  @lock.synchronize { @workers.delete(worker) }
end
replace_worker(worker) click to toggle source
# File lib/pallets/manager.rb, line 46
def replace_worker(worker)
  @lock.synchronize do
    @workers.delete(worker)

    return if @needs_to_stop

    worker = Worker.new(self, @backend)
    @workers << worker
    worker.start
  end
end
shutdown() click to toggle source

Attempt to gracefully shutdown every worker. If any is still busy after the given timeout, hard shutdown it. We don't need to worry about lost jobs caused by the hard shutdown; there is a reliability list that contains all active jobs, which will be automatically requeued upon next start

# File lib/pallets/manager.rb, line 23
def shutdown
  @needs_to_stop = true

  @workers.reverse_each(&:graceful_shutdown)
  @scheduler.shutdown

  Pallets.logger.info 'Waiting for workers to finish their jobs...'
  # Wait for 10 seconds at most
  10.times do
    return if @workers.empty?
    sleep 1
  end

  @workers.reverse_each(&:hard_shutdown)
  # Ensure Pallets::Shutdown got propagated and workers finished; if not,
  # their threads will be killed anyway when the manager quits
  sleep 0.5
end
start() click to toggle source
# File lib/pallets/manager.rb, line 13
def start
  @workers.each(&:start)
  @scheduler.start
end