Module: Polyrun::Queue::WorkerLoop

Defined in:
lib/polyrun/queue/worker_loop.rb

Overview

Claim / run / ack loop for polyrun run-queue workers (forked or direct).

Class Method Summary collapse

Class Method Details

.run(store:, worker_id:, batch:, cmd:, on_failure:) ⇒ Object 



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/polyrun/queue/worker_loop.rb', line 7

def run(store:, worker_id:, batch:, cmd:, on_failure:)
  batches_ok = 0
  batches_fail = 0
  loop do
    claim = store.claim!(worker_id: worker_id, batch_size: batch)
    paths = claim["paths"] || []
    break if paths.empty?

    code = run_batch(cmd, paths)
    if code == 0
      store.ack!(lease_id: claim["lease_id"], worker_id: worker_id)
      batches_ok += 1
    elsif on_failure.to_s == "requeue"
      store.reclaim_lease!(claim["lease_id"])
      batches_fail += 1
      return {ok: batches_ok, fail: batches_fail, exit_code: 1}
    else
      batches_fail += 1
      return {ok: batches_ok, fail: batches_fail, exit_code: code.zero? ? 1 : code}
    end
  end
  {ok: batches_ok, fail: batches_fail, exit_code: 0}
rescue Polyrun::Error => e
  Polyrun::Log.warn "polyrun run-queue worker #{worker_id}: #{e.message}"
  {ok: batches_ok, fail: batches_fail, exit_code: 2}
end

.run_batch(cmd, paths) ⇒ Object 



34
35
36
 
# File 'lib/polyrun/queue/worker_loop.rb', line 34

def run_batch(cmd, paths)
  system(*cmd, *paths) ? 0 : ($?.exitstatus || 1)
end