Class: Async::Background::Runner

Inherits:
Object
  • Object
show all
Includes:
Clock, QueueExecution, Schedule
Defined in:
lib/async/background/runner.rb,
lib/async/background/runner/schedule.rb,
lib/async/background/runner/queue_execution.rb

Defined Under Namespace

Modules: QueueExecution, Schedule

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config_path: nil, job_count: 2, worker_index:, total_workers:, queue_socket_dir: nil, queue_db_path: nil, queue_mmap: true, drain_timeout: DRAIN_GRACE, metrics_shm_path: Metrics.default_shm_path) ⇒ Runner

Returns a new instance of Runner.



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/async/background/runner.rb', line 38

def initialize(
  config_path: nil,
  job_count: 2,
  worker_index:,
  total_workers:,
  queue_socket_dir: nil,
  queue_db_path: nil,
  queue_mmap: true,
  drain_timeout: DRAIN_GRACE,
  metrics_shm_path: Metrics.default_shm_path
)
  @logger = Console.logger
  @worker_index = worker_index
  @total_workers = total_workers
  @running = true
  @drain_timeout = drain_timeout
  @shutdown = Runtime::Notification.new
  @metrics = Metrics.new(worker_index: worker_index, total_workers: total_workers, shm_path: metrics_shm_path)
  logger.info { "Async::Background worker_index=#{worker_index}/#{total_workers}, job_count=#{job_count}" }

  @jobs = Runtime::TaskGroup.new(on_error: error_handler, on_release: method(:job_released))
  @services = Runtime::TaskGroup.new(on_error: error_handler)
  @semaphore = Runtime::Semaphore.new(job_count)
  @heap = config_path.nil? ? MinHeap.new : build_heap(config_path)
  setup_queue(queue_socket_dir, queue_db_path, queue_mmap)
  validate_work_source!(config_path)
end

Instance Attribute Details

#heap ⇒ Object (readonly)

Returns the value of attribute heap.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def heap
  @heap
end

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def jobs
  @jobs
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def logger
  @logger
end

#metrics ⇒ Object (readonly)

Returns the value of attribute metrics.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def metrics
  @metrics
end

#queue_store ⇒ Object (readonly)

Returns the value of attribute queue_store.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def queue_store
  @queue_store
end

#semaphore ⇒ Object (readonly)

Returns the value of attribute semaphore.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def semaphore
  @semaphore
end

#services ⇒ Object (readonly)

Returns the value of attribute services.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def services
  @services
end

#shutdown ⇒ Object (readonly)

Returns the value of attribute shutdown.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def shutdown
  @shutdown
end

#total_workers ⇒ Object (readonly)

Returns the value of attribute total_workers.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def total_workers
  @total_workers
end

#worker_index ⇒ Object (readonly)

Returns the value of attribute worker_index.



27
28
29
 
# File 'lib/async/background/runner.rb', line 27

def worker_index
  @worker_index
end

Instance Method Details

#run ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/async/background/runner.rb', line 66

def run
  Runtime.scheduler!
  warn_unsafe_timeouts

  Runtime.with_error_handler(error_handler) do
    setup_signal_handlers
    start_signal_watcher
    start_queue_listener if @listen_queue

    scheduler_loop
    shutdown_gracefully
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


87
 
# File 'lib/async/background/runner.rb', line 87

def running? = @running

#stop ⇒ Object 



80
81
82
83
84
85
 
# File 'lib/async/background/runner.rb', line 80

def stop
  return unless @running

  @running = false
  wake_signal_watcher
end