JobWrapper

The Job Wrapper Class is instantiated with arguments tailored for running a particular job. The JobWrapper starts a thread for execution of the job and a Watchdog Agent that can monitor its progress.

JobWrapper options

JobWrapper
{
  BufferLimit = 10485760
  CleanUpFlag = True
  DefaultCatalog = []
  DefaultCPUTime = 600
  DefaultErrorFile = 'std.err'
  DefaultOutputFile = 'std.out'
  DiskSE = ['-disk', '-DST', '-USER']
  MasterCatalogOnlyFlag = True
  MaxJobPeekLines = 20
  OutputSandboxLimit = 1024 * 1024 * 10
  # Retry the upload of the output file if only one output SE is defined
  RetryUpload = False
  TapeSE = ['-tape', '-RDST', '-RAW']
  MinOutputDataBufferGB = 5
}

class DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapper.ExecutionThread(spObject, cmd, maxPeekLines, stdoutFile, stderrFile, exeEnv, executionResults)

Bases: Thread

__init__(spObject, cmd, maxPeekLines, stdoutFile, stderrFile, exeEnv, executionResults)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

property daemon

A boolean value indicating whether this thread is a daemon thread.

This must be set before start() is called, otherwise RuntimeError is raised. Its initial value is inherited from the creating thread; the main thread is not a daemon thread and therefore all threads created in the main thread default to daemon = False.

The entire Python program exits when only daemon threads are left.

getCurrentPID()

getName()

Return a string used for identification purposes only.

This method is deprecated, use the name attribute instead.

getOutput(lines=0)

property ident

Thread identifier of this thread or None if it has not been started.

This is a nonzero integer. See the get_ident() function. Thread identifiers may be recycled when a thread exits and another thread is created. The identifier is available even after the thread has exited.

isDaemon()

Return whether this thread is a daemon.

This method is deprecated, use the daemon attribute instead.

is_alive()

Return whether the thread is alive.

This method returns True just before the run() method starts until just after the run() method terminates. See also the module function enumerate().

join(timeout=None)

Wait until the thread terminates.

This blocks the calling thread until the thread whose join() method is called terminates – either normally or through an unhandled exception or until the optional timeout occurs.

When the timeout argument is present and not None, it should be a floating point number specifying a timeout for the operation in seconds (or fractions thereof). As join() always returns None, you must call is_alive() after join() to decide whether a timeout happened – if the thread is still alive, the join() call timed out.

When the timeout argument is not present or None, the operation will block until the thread terminates.

A thread can be join()ed many times.

join() raises a RuntimeError if an attempt is made to join the current thread as that would cause a deadlock. It is also an error to join() a thread before it has been started and attempts to do so raises the same exception.

property name

A string used for identification purposes only.

It has no semantics. Multiple threads may be given the same name. The initial name is set by the constructor.

property native_id

Native integral thread ID of this thread, or None if it has not been started.

This is a non-negative integer. See the get_native_id() function. This represents the Thread ID as reported by the kernel.

run(): Method representing the thread activity. This one overrides the ~threading.Thread run method

sendOutput(stdid, line)

setDaemon(daemonic)

Set whether this thread is a daemon.

This method is deprecated, use the .daemon property instead.

setName(name)

Set the name string for this thread.

This method is deprecated, use the name attribute instead.

start()

Start the thread’s activity.

It must be called at most once per thread object. It arranges for the object’s run() method to be invoked in a separate thread of control.

This method will raise a RuntimeError if called more than once on the same thread object.

class DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapper.JobWrapper(jobID: int | None = None, jobReport: JobReport | None = None)

Bases: object

The only user of the JobWrapper is the JobWrapperTemplate

__init__(jobID: int | None = None, jobReport: JobReport | None = None): Standard constructor

execute(): Main execution method of the Job Wrapper

finalize(): Perform any final actions to clean up after job execution.

initialize(arguments): Initializes parameters and environment for job.

postProcess(payloadStatus: int | None, payloadOutput: str, payloadExecutorError: str, cpuTimeConsumed: list, watchdogError: str, watchdogStats: dict): This method is called after the payload has finished running.

preProcess(): This method is called before the payload starts.

process(command: str, env: dict): This method calls the payload.

processJobOutputs(): Outputs for a job may be treated here.

resolveInputData(): Input data is resolved here using a VO specific plugin module.

sendFailoverRequest(): Create and send a combined job failover request if any

sendJobAccounting(status='', minorStatus=''): Send WMS accounting data.

transferInputSandbox(inputSandbox): Downloads the input sandbox for the job