Class: SimpleWS::Jobs::Scheduler::Job

Inherits:
Object
  • Object
show all
Defined in:
lib/simplews/jobs.rb,
lib/simplews/rake.rb

Overview

{{{ Job

Constant Summary collapse

@@config =
{}

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.configure(name, value) ⇒ Object


253
254
255
# File 'lib/simplews/jobs.rb', line 253

def self.configure(name, value)
  @@config[name] = value
end

.job_info(name) ⇒ Object

Raises:


219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
# File 'lib/simplews/jobs.rb', line 219

def self.job_info(name)
  info = nil

  retries = 2
  begin
    info = Marshal::load(File.open(File.join(@@savedir,name + '.marshal')))
    raise Exception unless info.is_a?(Hash) && info[:info]
  rescue Exception
    if retries > 0
      retries -= 1
      sleep SimpleWS::Jobs::SLEEP_TIMES[:job_info]
      retry
    end
    info = nil
  end

  raise JobNotFound, "Job with name '#{ name }' was not found" if info.nil?

  if info[:queued] && !@@queue.collect{|info| info[:name]}.include?(name)
    FileUtils.rm(File.join(@@savedir, name + '.marshal'))
    raise Aborted, "Job #{ name } has been removed from the queue"
  end

  info
end

.path(file, name) ⇒ Object


205
206
207
208
209
210
211
# File 'lib/simplews/jobs.rb', line 205

def self.path(file, name)
  if file =~ /^\/|#{@@workdir}/
    file.gsub(/\{JOB\}/, name)
  else
    File.join(@@workdir, file.gsub(/\{JOB\}/,name))
  end
end

.results(name) ⇒ Object


245
246
247
248
249
250
# File 'lib/simplews/jobs.rb', line 245

def self.results(name)
  job_info(name)[:results].collect{|file|
    code = Scheduler.random_name("res-")
    [code, file]
  }
end

.save(name, state) ⇒ Object


213
214
215
216
217
# File 'lib/simplews/jobs.rb', line 213

def self.save(name, state)
  fout = File.open(File.join(@@savedir,name + '.marshal'),'w')
  fout.write Marshal::dump(state)
  fout.close
end

.taken(name = "") ⇒ Object


201
202
203
204
# File 'lib/simplews/jobs.rb', line 201

def self.taken(name = "")
  Dir.glob(@@savedir + "/#{ name }*.marshal").
    collect{|n| n.match(/\/(#{ Regexp.quote name }(?:-\d+)?).marshal/); $1}.compact
end

.workdir=(workdir) ⇒ Object


194
195
196
197
198
199
# File 'lib/simplews/jobs.rb', line 194

def self.workdir=(workdir)
  @@workdir = workdir
  @@savedir = File.join(@@workdir, '.save')
  FileUtils.mkdir_p @@workdir unless File.exist? @@workdir
  FileUtils.mkdir_p @@savedir unless File.exist? @@savedir
end

Instance Method Details

#abortObject


312
313
314
315
# File 'lib/simplews/jobs.rb', line 312

def abort
  raise SimpleWS::Jobs::Aborted
  save
end

#add_description(reg_exp, step, message) ⇒ Object

Add step information to rule tasks, as the 'desc' method cannot be used to describe them for the time being.


11
12
13
14
# File 'lib/simplews/rake.rb', line 11

def add_description(reg_exp, step, message)
  @step_descriptions ||= {}
  @step_descriptions[Regexp.new(reg_exp)] = "#{ step }: #{ message }"
end

#configObject


261
262
263
# File 'lib/simplews/jobs.rb', line 261

def config
  @@config
end

#error(message = nil) ⇒ Object


292
293
294
295
# File 'lib/simplews/jobs.rb', line 292

def error(message = nil)
  step(:error, message)
  save
end

#info(info = {}) ⇒ Object


297
298
299
300
301
# File 'lib/simplews/jobs.rb', line 297

def info(info = {})
  @state[:info].merge!(info)
  save
  @state[:info]
end

#job_nameObject


317
318
319
# File 'lib/simplews/jobs.rb', line 317

def job_name
  @name
end

#message(message) ⇒ Object


282
283
284
285
# File 'lib/simplews/jobs.rb', line 282

def message(message)
  @state[:messages] << message 
  save
end

#path(file) ⇒ Object


265
266
267
# File 'lib/simplews/jobs.rb', line 265

def path(file)
  Job.path(file,  @name)
end

#rake(rakefile = "Rakefile", target = nil) ⇒ Object

Instruct rake to load the rakefile, named Rakefile by default, and use it to produce the file specified first as product of the web service task. The 'execute' method of the Rake::Tasks class method execute is monkey-patched to log the steps. Since this is executed on a new process, there should be no side-effects from the patching.


21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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
65
66
67
68
69
# File 'lib/simplews/rake.rb', line 21

def rake(rakefile = "Rakefile", target = nil)
  Rake::Task.class_eval "alias_method :old_execute, :execute\ndef execute(*args)\naction = name\nmessage = $_step_descriptions.collect{|rexp, msg|\nif name.match(rexp)\nmsg\nelse\nnil\nend\n}.compact.first\n\nmessage ||= comment\n\nmessage ||= \"Invoking \#{name}\"\n\nif message.match(/^(\\w+): (.*)/)\n$_current_job.step($1, $2)\nelse\n$_current_job.step(action, message)\nend\n\nold_execute(*args)\nend\n"

  load rakefile
  @@steps.each{|step|
    step_dirname = File.join(workdir, step.to_s)
    FileUtils.mkdir_p step_dirname unless File.exists? step_dirname
  }

  if defined? Rake::Pipeline
    Rake::Pipeline::step_descriptions.each{|re, description|
      if description.match(/(.*): (.*)/)
        add_description(re, $1, $2)
      end
    }
  end

  files = result_filenames
  target ||= files.first

  $_current_job = self
  $_step_descriptions = @step_descriptions || {}

  Rake::Task[target].invoke
end

#result_filenamesObject


308
309
310
# File 'lib/simplews/jobs.rb', line 308

def result_filenames
  @state[:results]
end

#results(results) ⇒ Object


303
304
305
306
# File 'lib/simplews/jobs.rb', line 303

def results(results)
  @state[:results] = results.collect{|file| path(file)}
  save
end

#run(task, name, results, *args) ⇒ Object


321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
# File 'lib/simplews/jobs.rb', line 321

def run(task, name, results, *args)
  @name = name
  @state = {
    :name => @name, 
    :status => :prepared, 
    :messages => [], 
    :info => {}, 
    :results => results.collect{|file| path(file)},
  }
  save
  @pid = Process.fork do
    begin
      puts "Job #{@name} starting with pid #{Process.pid}"

      trap(:INT) { raise SimpleWS::Jobs::Aborted }
      self.send task, *args
      step :done
      exit(0)
    rescue  SimpleWS::Jobs::Aborted
      step(:aborted, "Job Aborted")
      exit(-1)
    rescue Exception
      if !$!.kind_of? SystemExit
        error($!.message)
        puts "Error in job #{ @name }"
        puts $!.message
        puts $!.backtrace
        exit(-1)
      else
        exit($!.status)
      end
    end
  end

  @pid
end

#saveObject


269
270
271
# File 'lib/simplews/jobs.rb', line 269

def save
  Job.save(@name, @state)
end

#step(status, message = nil) ⇒ Object


286
287
288
289
290
# File 'lib/simplews/jobs.rb', line 286

def step(status, message = nil)
  @state[:status] = status
  @state[:messages] << message if message && message != ""
  save
end

#workdirObject


257
258
259
# File 'lib/simplews/jobs.rb', line 257

def workdir
  @@workdir
end

#write(file, contents) ⇒ Object


273
274
275
276
277
278
279
280
# File 'lib/simplews/jobs.rb', line 273

def write(file, contents)
  path = Job.path(file, @name)
  directory = File.dirname(File.expand_path(path))
  FileUtils.mkdir_p directory unless File.exists? directory
  File.open(path,'w') do |fout|
    fout.write contents
  end
end