bcftbx.JobRunner

Classes for starting, stopping and managing jobs.

Class BaseJobRunner is a template with methods that need to be implemented by subclasses. The subclasses implemented here are:

  • SimpleJobRunner: run jobs (e.g. scripts) on a local file system.

  • GEJobRunner : run jobs using Grid Engine (GE) i.e. qsub, qdel etc

A single JobRunner instance can be used to start and manage multiple processes.

Each job is started by invoking the ‘run’ method of the runner. This returns an id string which is then used in calls to the ‘isRunning’, ‘terminate’ etc methods to check on and control the job.

The runner’s ‘list’ method returns a list of running job ids.

Simple usage example:

>>> # Create a JobRunner instance
>>> runner = SimpleJobRunner()
>>> # Start a job using the runner and collect its id
>>> job_id = runner.run('Example',None,'myscript.sh')
>>> # Wait for job to complete
>>> import time
>>> while runner.isRunning(job_id):
>>>     time.sleep(10)
>>> # Get the names of the output files
>>> log,err = (runner.logFile(job_id),runner.errFile(job_id))

Processes run using a job runner inherit the environment where the runner is created and executed.

Additionally runners set an ‘BCFTBX_RUNNER_NSLOTS’ environment variable, which is set to the number of slots (aka CPUs/cores/threads) available to processes executed by the runner. For both ‘SimpleJobRunner’ and ‘GEJobRunner’, this defaults to one (i.e. serial jobs); the ‘nslots’ option can be used when instantiating ‘SimpleJobRunner’ objects to specify more cores, for example:

>>> multicore_runner = SimpleJobRunner(nslots=4)

For ‘GEJobRunner’ instances the number of cores is set by specifying ‘-pe smp.pe’ as part of the ‘ge_extra_args’ option, for example:

>>> multicore_runner = GEJobRunner(extra_ge_args=('-pe','smp.pe','4'))
class bcftbx.JobRunner.BaseJobRunner

Base class for implementing job runners

This class can be used as a template for implementing custom job runners. The idea is that the runners wrap the specifics of interacting with an underlying job control system and thus provide a generic interface to be used by higher level classes.

A job runner needs to implement the following methods:

  • run : starts a job running

  • terminate : kills a running job

  • list : lists the running job ids

  • logFile : returns the name of the log file for a job

  • errFile : returns the name of the error file for a job

  • exit_status: returns the exit status for the command (or None if the job is still running)

Optionally it can also implement the methods:

  • errorState: indicates if running job is in an “error state”

  • isRunning : checks if a specific job is running

if the default implementations are not sufficient.

errFile(job_id)

Return name of error file relative to working directory

errorState(job_id)

Check if the job is in an error state

Return True if the job is deemed to be in an ‘error state’, False otherwise.

exit_status(job_id)

Return the exit status code for the command

Return the exit status code from the command that was run by the specified job, or None if the job hasn’t exited yet.

isRunning(job_id)

Check if a job is running

Returns True if job is still running, False if not

list()

Return a list of running job_ids

logFile(job_id)

Return name of log file relative to working directory

property log_dir

Return the current log directory setting

run(name, working_dir, script, args)

Start a job running

Parameters:
  • name – Name to give the job

  • working_dir – Directory to run the job in

  • script – Script file to run

  • args – List of arguments to supply to the script

Returns:

Returns a job id, or None if the job failed to start

set_log_dir(log_dir)

(Re)set the directory to write log files to

terminate(job_id)

Terminate a job

Returns True if termination was successful, False otherwise

class bcftbx.JobRunner.GEJobRunner(queue=None, log_dir=None, ge_extra_args=None, poll_interval=5.0, timeout=30.0)

Class implementing job runner for Grid Engine

GEJobRunner submits jobs to a Grid Engine cluster using the ‘qsub’ command, determines the status of jobs using ‘qstat’ and terminates then using ‘qdel’.

Additionally the runner can be configured for a specific GE queue on initialisation.

Each GEJobRunner instance creates a temporary directory which it uses for internal admin; this will be removed at program exit via ‘atexit’.

errFile(job_id)

Return the error file name for a job

The name should be ‘<name>.e<job_id>’

errorState(job_id)

Check if the job is in an error state

Return True if the job is deemed to be in an ‘error state’ (i.e. qstat returns the state as ‘E..’), False otherwise.

exit_status(job_id)

Return exit status from command run by a job

If the job is still running then returns ‘None’.

property ge_extra_args

Return the extra GE arguments

list()

Get list of job ids which are queued or running

logFile(job_id)

Return the log file name for a job

The name should be ‘<name>.o<job_id>’

name(job_id)

Return the name for a job

property nslots

Return the number of associated slots

This is extracted from the ‘ge_extra_args’ property, by looking for qsub options of the form ‘-pe smp.pe N’ (in which case ‘nslots’ will be N).

queue(job_id)

Fetch the job queue name

Returns the queue as reported by qstat, or None if not found.

run(name, working_dir, script, args)

Submit a script or command to the cluster via ‘qsub’

Parameters:
  • name – Name to give the job

  • working_dir – Directory to run the job in

  • script – Script file to run

  • args – List of arguments to supply to the script

Returns:

Job id for submitted job, or ‘None’ if job failed to start.

terminate(job_id)

Remove a job from the GE queue using ‘qdel’

class bcftbx.JobRunner.ResourceLock

Class for managing in-process locks on ‘resources’

A ‘resource’ is identified by an arbitrary string.

Example usage: create a new ResourceLock instance and check if a resource is locked:

>>> r = ResourceLock()
>>> r.is_locked("resource1")
False

Try to acquire the lock on the resource:

>>> lock = r.acquire("resource1")
>>> r.is_locked("resource1")
True

Release the lock on the resource:

>>> r.release(lock)
>>> r.is_locked("resource1")
False
acquire(resource_name, timeout=None)

Attempt to acquire the lock on a resource

Parameters:
  • resource_name (str) – name of the resource to acquire the lock name for

  • timeout (float) – optional, specifies a timeout period after which failure to acquire the lock raises an exception.

Returns:

lock name.

Return type:

String

is_locked(resource_name)

Check if a resource is locked

Parameters:

resource_name (str) – name of the resource to check the lock for

Returns:

True if resource is locked, False

if not.

Return type:

Boolean

release(lock)

Release a lock on a resource

Parameters:

lock (str) – lock to release.

class bcftbx.JobRunner.SimpleJobRunner(log_dir=None, join_logs=False, nslots=1)

Class implementing job runner for local system

SimpleJobRunner starts jobs as processes on a local system; the status of jobs is determined using the Linux ‘ps eu’ command, and jobs are terminated using ‘kill -9’.

errFile(job_id)

Return the error file name for a job

exit_status(job_id)

Return exit status from command run by a job

list()

Return a list of running job_ids

logFile(job_id)

Return the log file name for a job

name(job_id)

Return the name for a job

property nslots

Return the number of associated slots

run(name, working_dir, script, args)

Run a command and return the PID (=job id)

Parameters:
  • name – Name to give the job

  • working_dir – Directory to run the job in

  • script – Script file to run

  • args – List of arguments to supply to the script

Returns:

Job id for submitted job, or ‘None’ if job failed to start.

terminate(job_id)

Kill a running job using ‘kill -9’

bcftbx.JobRunner.fetch_runner(definition)

Return job runner instance based on a definition string

Given a definition string, returns an appropriate runner instance.

Definitions are of the form:

RunnerName[(args)]

RunnerName can be ‘SimpleJobRunner’ or ‘GEJobRunner’. If ‘(args)’ are also supplied then:

  • for SimpleJobRunners, this can be a list of optional arguments separated by spaces:

    • ‘nslots=N’ (where N is an integer; sets a non-default number of slots

    • ‘join_logs=BOOLEAN’ (where BOOLEAN can be ‘True’, ‘true’,’y’,’False’,’false’,’n’; sets whether stdout and stderr should be written to the same file)

  • for GEJobRunners, this is a set of arbitrary ‘qsub’ options that will be used on job submission.