Class Dsadmin::Admind::Cron
In: lib/dsadmin/admind/Cron.rb
Parent: Object

Internal task scheduler.

A replacement for cron(8) adapted to our special needs:

  • Running internally => no call overhead and very simple to use and configure from the admind code.
  • Schedules Dsadmin::Request objects (including all contained data) instead of general Unix commands.
  • Supports flexible retry schemas on failed and delayed job executions (see class RetryPolicy).
  • Supports delaying of job executions (typically depending on the server load), allowing automatic distribution of jobs in time and avoiding resource bottlenecks (see Dsadmin::Response::TEMP_UNAVAILABLE).

This is intended to run as a "background service" (singleton) on the master admind instance. Scheduling has a granularity/precision of 60s.

Note: "Normal" clients should not access the Cron instance directly, but instead use a CronProxy, which will then automatically forward everything to the "real" Cron, no matter if it‘s running locally or on a different machine.

See also:class CronjobSpec
See also:class CronProxy
See also:class Dsadmin::Common::TimeSpec
See also:class RetryPolicy

Methods

add   forcerun   new   pause   ps   remove   resume   run   runMain   running?   shutdown   visitJobs  

Included Modules

Dsadmin::CoreClient Dsadmin::Contractor Dsadmin::ValueChecker Singleton

Constants

RETRY_REG15 = RetryPolicy.new(:regular, [15])   Common retry policy: Every 15 mins
RETRY_EXP15 = RetryPolicy.new(:exponential, [15])   Common retry policy: After 15 mins, then doubling that interval each time
RETRY_ONCE15 = RetryPolicy.new(:fixed, [15])   Common retry policy: Once, after 15 mins
RETRY_REG015 = RetryPolicy.new(:regular, [0, 15])   Common retry policy: Retry immediately, then every 15 mins
RETRY_EXP015 = RetryPolicy.new(:exponential, [0, 15])   Common retry policy: Retry immediately, then after 15 mins, then doubling that interval each time

Attributes

thread  [R] 

Public Class methods

new()

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 398
398:       def initialize
399:         @jobs       = Hash.new # jobid => Cronjob
400:         @mutex      = Mutex.new
401:         @running_id = 0
402:         @shutdown   = false
403:       end

Public Instance methods

add(aCronjobSpec)

Add and schedule a job (given as a CronjobSpec).

Returns:The job‘s ID

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 302
302:       def add(aCronjobSpec)
303:         requireKindOf(CronjobSpec, aCronjobSpec)
304:         id = nil
305:         
306:         @mutex.synchronize {
307:           while(@jobs.has_key?(@running_id))
308:             @running_id += 1
309:           end
310:           id = @running_id
311:           
312:           @jobs[id] = Cronjob.new(aCronjobSpec)
313:           s = aCronjobSpec
314:           log.notice("Scheduling '#{s.request.uri}' for execution at/every #{s.timespec.to_s}")
315:         }
316:         
317:         return id
318:       end
forcerun(aJobID)

Force the specified job to be executed now (in a seperate thread).

If the job is already running, nothing happens.

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 356
356:       def forcerun(aJobID)
357:         return if @shutdown
358:         
359:         @mutex.synchronize {
360:           checkNotNil(@jobs[aJobID])
361:           @jobs[aJobID].force_run()
362:         }
363:       end
pause(aJobID)

Pause scheduling of the specified cronjob.

This means no further execution of it is triggered until resume is called for it. This does not pause a currently running job — only its further scheduling.

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 336
336:       def pause(aJobID)
337:         @mutex.synchronize {
338:           checkNotNil(@jobs[aJobID])
339:           @jobs[aJobID].pause()
340:         }
341:       end
ps()

Get a list of scheduled tasks and their status (Array of Hashes)

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 367
367:       def ps
368:         res = Array.new
369:         
370:         @mutex.synchronize do
371:           @jobs.each do |key, job|
372:             entry = {
373:               'jobid'         => key,
374:               'uri'           => job.uri,
375:               'user'          => job.username,
376:               'last_run_time' => job.last_run_time,
377:               'last_end_time' => job.last_end_time,
378:               'next_run_time' => job.next_run_time,
379:               'success_count' => job.success_count,
380:               'failure_count' => job.failure_count,
381:               'delay_count'   => job.delay_count,
382:               'state'         => job.state_s,
383:               'last_result'   => job.last_result_s
384:             }
385:             
386:             res << entry
387:           end
388:         end
389:         
390:         return res
391:       end
remove(aJobID)

Remove the specified cronjob. If the job is running, this does not abort it.

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 322
322:       def remove(aJobID)
323:         #        abort(aJobID)
324:         @mutex.synchronize {
325:           checkNotNil(@jobs[aJobID])
326:           @jobs.delete(aJobID)
327:         }
328:       end
resume(aJobID)

The counterpiece to pause

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 345
345:       def resume(aJobID)
346:         @mutex.synchronize {
347:           checkNotNil(@jobs[aJobID])
348:           @jobs[aJobID].run()
349:         }
350:       end
run()

Start the Cron service (in a separate thread)

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 274
274:       def run
275:         requireTrue(running? != true)
276:         
277:         @shutdown = false
278:         
279:         @thread = spawnThread {
280:           runMain
281:         }
282:         
283:         @thread.wakeup
284:       end
running?()

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 293
293:       def running?
294:         return thread && (thread.alive?)
295:       end
shutdown()

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 287
287:       def shutdown
288:         @shutdown = true
289:         thread.wakeup if running?
290:       end

Private Instance methods

runMain()

Main loop

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 407
407:       def runMain
408:         while(! @shutdown)
409:           t_start = sys.now.to_i
410:           visitJobs
411:           t_end   = sys.now.to_i
412:           t_sleep = 60 - (t_end - t_start)
413:           #puts "Next pass in #{t_sleep} seconds"
414:           sleep(t_sleep) if(t_sleep > 0)
415:         end
416:       end
visitJobs()

[Source]

     # File lib/dsadmin/admind/Cron.rb, line 419
419:       def visitJobs
420:         #puts "visiting cronjobs"
421:         
422:         @mutex.synchronize {
423:           @jobs.delete_if { |id, job| job.done? }
424:         }
425:         
426:         tmp = @jobs.clone
427:         
428:         tmp.each { |id, job|
429:           job.tick
430:         }
431:       end

[Validate]