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 collapse

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

#node ⇒ Object (readonly)

Returns the value of attribute node.



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

def node
  @node
end

#pid ⇒ Object (readonly)

Returns the value of attribute pid.



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

def pid
  @pid
end

#status ⇒ Object (readonly)

Returns the value of attribute status.



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

def status
  @status
end

#unit ⇒ Object (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

#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_unit ⇒ Object

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_options ⇒ Object

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.



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

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

#run ⇒ Object

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



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

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_unit ⇒ Object

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
93
 
# 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
  @node.resolve_work(@unit['id'])
end

#time_taken ⇒ Object

How long has this worker been running for?



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

def time_taken
  Time.now - @start_time
end