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

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

Overview

{{{ Job

Constant Summary collapse

@@config = 
{}

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.configure(name, value) ⇒ Object 



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

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

.job_info(name) ⇒ Object

Raises:



212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/simplews/jobs.rb', line 212

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 



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

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 



238
239
240
241
242
243
 
# File 'lib/simplews/jobs.rb', line 238

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

.save(name, state) ⇒ Object 



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

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 



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

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

.workdir=(workdir) ⇒ Object 



187
188
189
190
191
192
 
# File 'lib/simplews/jobs.rb', line 187

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

#abort ⇒ Object

Raises:



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

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

#config ⇒ Object 



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

def config
  @@config
end

#error(message = nil) ⇒ Object 



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

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

#info(info) ⇒ Object 



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

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

#job_name ⇒ Object 



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

def job_name
  @name
end

#message(message) ⇒ Object 



273
274
275
276
 
# File 'lib/simplews/jobs.rb', line 273

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

#path(file) ⇒ Object 



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

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

#results(results) ⇒ Object 



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

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

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



307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
 
# File 'lib/simplews/jobs.rb', line 307

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

#save ⇒ Object 



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

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

#step(status, message = nil) ⇒ Object 



277
278
279
280
281
 
# File 'lib/simplews/jobs.rb', line 277

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

#workdir ⇒ Object 



250
251
252
 
# File 'lib/simplews/jobs.rb', line 250

def workdir
  @@workdir
end

#write(file, contents) ⇒ Object 



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

def write(file, contents)
  path = Job.path(file, @name)
  fout = File.open(path,'w')
  fout.write contents
  fout.close
end