9.2.11. WFProcessor API

class radical.entk.appman.wfprocessor.WFprocessor(sid, workflow, pending_queue, completed_queue, resubmit_failed, rmq_conn_params)[source]

An WFprocessor (workflow processor) takes the responsibility of dispatching tasks from the various pipelines of the workflow according to their relative order to the TaskManager. All state updates are relflected in the AppManager as we operate on the reference of the same workflow object. The WFprocessor also retrieves completed tasks from the TaskManager and updates states of PST accordingly.

Arguments:
sid:(str) session id used by the profiler and loggers
workflow:(set) REFERENCE of the AppManager’s workflow
pending_queue:(list) queues to hold pending tasks
completed_queue:
 (list) queues to hold completed tasks
resubmit_failed:
 (bool) True if failed tasks should be resubmitted
rmq_conn_params:
 (pika.connection.ConnectionParameters) object of parameters necessary to connect to RabbitMQ
_advance(obj, obj_type, new_state)[source]

transition obj of type obj_type into state new_state

_create_workload()[source]
_dequeue()[source]

Purpose: This is the function that is run in the dequeue thread. This function extracts Tasks from the completed queues and updates the workflow.

_enqueue()[source]

Purpose: This is the function that is run in the enqueue thread. This function extracts Tasks from the workflow that exists in the WFprocessor object and pushes them to the queues in the pending_q list.

_execute_post_exec(pipe, stage)[source]

Purpose: This method executes the post_exec step of a stage for a pipeline.

_execute_workload(workload, scheduled_stages)[source]
_update_dequeued_task(deq_task)[source]
check_processor()[source]
reset_workflow()[source]

When a component is restarted we reset all the tasks that did not finish to the first state. Then they are scheduled again for execution.

start_processor()[source]

Purpose: Method to start the wfp process. The wfp function is not to be accessed directly. The function is started in a separate process using this method.

terminate_processor()[source]

Purpose: Method to terminate the wfp process. This method is blocking as it waits for the wfp process to terminate (aka join).

workflow
workflow_incomplete()[source]

Purpose: Method to check if the workflow execution is incomplete.