Implementation of task collections.
Tasks can be grouped into collections, which are tasks themselves, therefore can be controlled (started/stopped/cancelled) like a single whole. Collection classes provided in this module implement the basic patterns of job group execution; they can be combined to form more complex workflows. Hook methods are provided so that derived classes can implement problem-specific job control policies.
A ParallelTaskCollection runs all of its tasks concurrently.
The collection state is set to TERMINATED once all tasks have reached the same terminal status.
Add a task to the collection.
Use the given Controller interface for operations on the job associated with this task.
Terminate all tasks in the collection, and set collection state to TERMINATED.
Try to advance all jobs in the collection to the next state in a normal lifecycle. Return list of task execution states.
Start all tasks in the collection.
Update state of all tasks in the collection.
Wrap a Task instance and re-submit it until a specified termination condition is met.
By default, the re-submission upon failure happens iff execution terminated with nonzero return code; the failed task is retried up to self.max_retries times (indefinitely if self.max_retries is 0).
Override the retry method to implement a different retryal policy.
Note: The resubmission code is implemented in the terminated(), so be sure to call it if you override in derived classes.
Evaluates to True if this task or any of its subtasks has been modified and should be saved to persistent storage.
Return True or False, depending on whether the failed task should be re-submitted or not.
The default behavior is to retry a task iff its execution terminated with nonzero returncode and the maximum retry limit has not been reached. If self.max_retries is 0, then the dependent task is retried indefinitely.
Override this method in subclasses to implement a different policy.
Update the state of the dependent task, then resubmit it if it’s TERMINATED and self.retry() is True.
A SequentialTaskCollection runs its tasks one at a time.
After a task has completed, the next method is called with the index of the finished task in the self.tasks list; the return value of the next method is then made the collection execution.state. If the returned state is RUNNING, then the subsequent task is started, otherwise no action is performed.
The default next implementation just runs the tasks in the order they were given to the constructor, and sets the state to TERMINATED when all tasks have been run.
Use the given Controller interface for operations on the job associated with this task.
Stop execution of this sequence. Kill currently-running task (if any), then set collection state to TERMINATED.
Return the state or task to run when step number done is completed.
This method is called when a task is finished; the done argument contains the index number of the just-finished task into the self.tasks list. In other words, the task that just completed is available as self.tasks[done].
The return value from next can be either a task state (i.e., an instance of Run.State), or a valid index number for self.tasks. In the first case:
If instead the return value is a (nonnegative) number, then tasks in the sequence will be re-run starting from that index.
The default implementation runs tasks in the order they were given to the constructor, and sets the state to TERMINATED when all tasks have been run. This method can (and should) be overridden in derived classes to implement policies for serial job execution.
Start the current task in the collection.
Update state of the collection, based on the jobs’ statuses.
Simplified interface for creating a sequence of Tasks. This can be used when the number of Tasks to run is fixed and known at program writing time.
A StagedTaskCollection subclass should define methods stage0, stage1, ... up to stageN (for some arbitrary value of N positive integer). Each of these stageN must return a Task instance; the task returned by the stage0 method will be executed first, followed by the task returned by stage1, and so on. The sequence stops at the first N such that stageN is not defined.
The exit status of the whole sequence is the exit status of the last Task instance run. However, if any of the stageX methods returns an integer value instead of a Task instance, then the sequence stops and that number is used as the sequence exit code.
Base class for all task collections. A “task collection” is a group of tasks, that can be managed collectively as a single one.
A task collection implements the same interface as the Task class, so you can use a TaskCollection everywhere a Task is required. A task collection has a state attribute, which is an instance of gc3libs.Run.State; each concrete collection class decides how to deduce a collective state based on the individual task states.
Add a task to the collection.
Use the given Controller interface for operations on the job associated with this task.
Evaluates to True if this task or any of its subtasks has been modified and should be saved to persistent storage.
This method just asks the Engine to free the contained tasks.
Raise a gc3libs.exceptions.InvalidOperation error, as there is no meaningful semantics that can be defined for peek into a generic collection of tasks.
Remove a task from the collection.
Return a dictionary mapping each state name into the count of tasks in that state. In addition, the following keys are defined:
If the optional argument only is not None, tasks whose class is not contained in only are ignored.
| Parameters: | only (tuple) – Restrict counting to tasks of these classes. |
|---|
Called when the job state transitions to TERMINATED, i.e., the job has finished execution (with whatever exit status, see returncode) and the final output has been retrieved.
Default implementation for TaskCollection is to set the exitcode to the maximum of the exit codes of its tasks.
Update the running state of all managed tasks.