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.