Class: CloudCrowd::Worker

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

Overview

The Worker, forked off from the Node when a new WorkUnit is received, launches an Action for processing. Workers will only ever receive WorkUnits that they are able to handle (for which they have a corresponding action in their actions directory). If communication with the central server is interrupted, the Worker will repeatedly attempt to complete its unit – every Worker::RETRY_WAIT seconds. Any exceptions that take place during the course of the Action will cause the Worker to mark the WorkUnit as having failed. When finished, the Worker's process exits, minimizing the potential for memory leaks.

Constant Summary

RETRY_WAIT =

Wait five seconds to retry, after internal communcication errors.

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(node, unit) ⇒ Worker

A new Worker customizes itself to its WorkUnit at instantiation.



20
21
22
23
24
25
26
27
28
# File 'lib/cloud_crowd/worker.rb', line 20

def initialize(node, unit)
  @start_time = Time.now
  @pid        = $$
  @node       = node
  @unit       = unit
  @status     = @unit['status']
  @retry_wait = RETRY_WAIT
  $0 = "#{unit['action']} (#{unit['id']}) [cloud-crowd-worker]"
end

Instance Attribute Details

#nodeObject (readonly)

Returns the value of attribute node



17
18
19
# File 'lib/cloud_crowd/worker.rb', line 17

def node
  @node
end

#pidObject (readonly)

Returns the value of attribute pid



17
18
19
# File 'lib/cloud_crowd/worker.rb', line 17

def pid
  @pid
end

#statusObject (readonly)

Returns the value of attribute status



17
18
19
# File 'lib/cloud_crowd/worker.rb', line 17

def status
  @status
end

#unitObject (readonly)

Returns the value of attribute unit



17
18
19
# File 'lib/cloud_crowd/worker.rb', line 17

def unit
  @unit
end

Instance Method Details

#base_paramsObject (private)

Common parameters to send back to central upon unit completion, regardless of success or failure.



128
129
130
131
132
# File 'lib/cloud_crowd/worker.rb', line 128

def base_params
  { :pid  => @pid,
    :id   => @unit['id'],
    :time => time_taken }
end

#complete_work_unit(result) ⇒ Object

Return output to the central server, marking the WorkUnit done.



31
32
33
34
35
36
37
# File 'lib/cloud_crowd/worker.rb', line 31

def complete_work_unit(result)
  keep_trying_to "complete work unit" do
    data = base_params.merge({:status => 'succeeded', :output => result})
    @node.central["/work/#{data[:id]}"].put(data)
    log "finished #{display_work_unit} in #{data[:time]} seconds"
  end
end

#display_work_unitObject

Loggable details describing what the Worker is up to.



66
67
68
# File 'lib/cloud_crowd/worker.rb', line 66

def display_work_unit
  "unit ##{@unit['id']} (#{@unit['action']}/#{CloudCrowd.display_status(@status)})"
end

#enhanced_unit_optionsObject

There are some potentially important attributes of the WorkUnit that we'd like to pass into the Action – in case it needs to know them. They will always be made available in the options hash.



110
111
112
113
114
115
116
# File 'lib/cloud_crowd/worker.rb', line 110

def enhanced_unit_options
  @unit['options'].merge({
    'job_id'        => @unit['job_id'],
    'work_unit_id'  => @unit['id'],
    'attempts'      => @unit['attempts']
  })
end

#fail_work_unit(exception) ⇒ Object

Mark the WorkUnit failed, returning the exception to central.



40
41
42
43
44
45
46
# File 'lib/cloud_crowd/worker.rb', line 40

def fail_work_unit(exception)
  keep_trying_to "mark work unit as failed" do
    data = base_params.merge({:status => 'failed', :output => {'output' => exception.message}.to_json})
    @node.central["/work/#{data[:id]}"].put(data)
    log "failed #{display_work_unit} in #{data[:time]} seconds\n#{exception.message}\n#{exception.backtrace}"
  end
end

#keep_trying_to(title) ⇒ Object

We expect and require internal communication between the central server and the workers to succeed. If it fails for any reason, log it, and then keep trying the same request.



51
52
53
54
55
56
57
58
59
60
61
62
63
# File 'lib/cloud_crowd/worker.rb', line 51

def keep_trying_to(title)
  begin
    yield
  rescue RestClient::ResourceNotFound => e
    log "work unit ##{@unit['id']} doesn't exist. discarding..."
  rescue Exception => e
    log "failed to #{title} -- retry in #{@retry_wait} seconds"
    log e.message
    log e.backtrace
    sleep @retry_wait
    retry
  end
end

#log(message) ⇒ Object (private)

Log a message to the daemon log. Includes PID for identification.



135
136
137
# File 'lib/cloud_crowd/worker.rb', line 135

def log(message)
  puts "Worker ##{@pid}: #{message}" unless ENV['RACK_ENV'] == 'test'
end

#runObject

Run this worker inside of a fork. Attempts to exit cleanly. Wraps run_work_unit to benchmark the execution time, if requested.



96
97
98
99
100
101
102
103
104
105
# File 'lib/cloud_crowd/worker.rb', line 96

def run
  trap_signals
  log "starting #{display_work_unit}"
  if @unit['options']['benchmark']
    log("ran #{display_work_unit} in " + Benchmark.measure { run_work_unit }.to_s)
  else
    run_work_unit
  end
  Process.exit!
end

#run_work_unitObject

Executes the WorkUnit by running the Action, catching all exceptions as failures. We capture the thread so that we can kill it from the outside, when exiting.



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
# File 'lib/cloud_crowd/worker.rb', line 73

def run_work_unit
  begin
    result = nil
    action_class = CloudCrowd.actions[@unit['action']]
    action = action_class.new(@status, @unit['input'], enhanced_unit_options, @node.asset_store)
    Dir.chdir(action.work_directory) do
      result = case @status
      when PROCESSING then action.process
      when SPLITTING  then action.split
      when MERGING    then action.merge
      else raise Error::StatusUnspecified, "work units must specify their status"
      end
    end
    action.cleanup_work_directory if action
    complete_work_unit({'output' => result}.to_json)
  rescue Exception => e
    action.cleanup_work_directory if action
    fail_work_unit(e)
  end
end

#time_takenObject

How long has this worker been running for?



119
120
121
# File 'lib/cloud_crowd/worker.rb', line 119

def time_taken
  Time.now - @start_time
end

#trap_signalsObject (private)

When signaled to exit, make sure that the Worker shuts down without firing the Node's at_exit callbacks.



141
142
143
144
145
146
# File 'lib/cloud_crowd/worker.rb', line 141

def trap_signals
  Signal.trap('QUIT') { Process.exit! }
  Signal.trap('INT')  { Process.exit! }
  Signal.trap('KILL') { Process.exit! }
  Signal.trap('TERM') { Process.exit! }
end