Class: Oxidized::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/oxidized/worker.rb

Instance Method Summary collapse

Constructor Details

#initialize(nodes) ⇒ Worker

Returns a new instance of Worker.



5
6
7
8
9
10
11
 
# File 'lib/oxidized/worker.rb', line 5

def initialize(nodes)
  @jobs_done  = 0
  @nodes      = nodes
  @jobs       = Jobs.new(Oxidized.config.threads, Oxidized.config.use_max_threads, Oxidized.config.interval, @nodes)
  @nodes.jobs = @jobs
  Thread.abort_on_exception = true
end

Instance Method Details

#process(job) ⇒ Object 



43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/oxidized/worker.rb', line 43

def process(job)
  node = job.node
  node.last = job
  node.stats.add job
  @jobs.duration job.time
  node.running = false
  if job.status == :success
    process_success node, job
  else
    process_failure node, job
  end
rescue NodeNotFound
  Oxidized.logger.warn "#{node.group}/#{node.name} not found, removed while collecting?"
end

#reload ⇒ Object 



58
59
60
 
# File 'lib/oxidized/worker.rb', line 58

def reload
  @nodes.load
end

#work ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/oxidized/worker.rb', line 13

def work
  ended = []
  @jobs.delete_if { |job| ended << job unless job.alive? }
  ended.each      { |job| process job }
  @jobs.work

  while @jobs.size < @jobs.want
    Oxidized.logger.debug "lib/oxidized/worker.rb: Jobs running: #{@jobs.size} of #{@jobs.want} - ended: #{@jobs_done} of #{@nodes.size}"
    # ask for next node in queue non destructive way
    nextnode = @nodes.first
    unless nextnode.last.nil?
      # Set unobtainable value for 'last' if interval checking is disabled
      last = Oxidized.config.interval.zero? ? Time.now.utc + 10 : nextnode.last.end
      break if last + Oxidized.config.interval > Time.now.utc
    end
    # shift nodes and get the next node
    node = @nodes.get
    node.running? ? next : node.running = true

    @jobs.push Job.new node
    Oxidized.logger.debug "lib/oxidized/worker.rb: Added #{node.group}/#{node.name} to the job queue"
  end

  if cycle_finished?
    run_done_hook
    exit 0 if Oxidized.config.run_once
  end
  Oxidized.logger.debug("lib/oxidized/worker.rb: #{@jobs.size} jobs running in parallel") unless @jobs.empty?
end