toil.server.wes.tasks

Module Contents

Classes

ToilWorkflowRunner

A class to represent a workflow runner to run the requested workflow.

TaskRunner

Abstraction over the Celery API. Runs our run_wes task and allows canceling it.

MultiprocessingTaskRunner

Version of TaskRunner that just runs tasks with Multiprocessing.

Functions

run_wes_task(base_scratch_dir, state_store_url, ...)

Run a requested workflow.

cancel_run(task_id)

Send a SIGTERM signal to the process that is running task_id.

Attributes

logger

WAIT_FOR_DEATH_TIMEOUT

run_wes

toil.server.wes.tasks.logger
toil.server.wes.tasks.WAIT_FOR_DEATH_TIMEOUT = 20
class toil.server.wes.tasks.ToilWorkflowRunner(base_scratch_dir, state_store_url, workflow_id, request, engine_options)

A class to represent a workflow runner to run the requested workflow.

Responsible for parsing the user request into a shell command, executing that command, and collecting the outputs of the resulting workflow run.

Parameters

  • base_scratch_dir (str) –

  • state_store_url (str) –

  • workflow_id (str) –

  • request (Dict[str, Any]) –

  • engine_options (List[str]) –

write_scratch_file(filename, contents)

Write a file to the scratch directory.

Parameters

  • filename (str) –

  • contents (str) –

Return type

None

get_state()
Return type

str

write_workflow(src_url)

Fetch the workflow file from its source and write it to a destination file.

Parameters

src_url (str) –

Return type

str

sort_options(workflow_engine_parameters=None)

Sort the command line arguments in the order that can be recognized by the workflow execution engine.

Parameters

workflow_engine_parameters (Optional[Dict[str, Optional[str]]]) – User-specified parameters for this

Return type

List[str]

particular workflow. Keys are command-line options, and values are option arguments, or None for options that are flags.

initialize_run()

Write workflow and input files and construct a list of shell commands to be executed. Return that list of shell commands that should be executed in order to complete this workflow run.

Return type

List[str]

call_cmd(cmd, cwd)

Calls a command with Popen. Writes stdout, stderr, and the command to separate files.

Parameters

  • cmd (Union[List[str], str]) –

  • cwd (str) –

Return type

subprocess.Popen[bytes]

run()

Construct a command to run a the requested workflow with the options, run it, and deposit the outputs in the output directory.

Return type

None

write_output_files()

Fetch all the files that this workflow generated and output information about them to outputs.json.

Return type

None

toil.server.wes.tasks.run_wes_task(base_scratch_dir, state_store_url, workflow_id, request, engine_options)

Run a requested workflow.

Parameters

  • base_scratch_dir (str) – Directory where the workflow’s scratch dir will live, under the workflow’s ID.

  • state_store_url (str) – URL/path at which the server and Celery task communicate about workflow state.

  • workflow_id (str) – ID of the workflow run.

  • request (Dict[str, Any]) –

  • engine_options (List[str]) –

Returns

the state of the workflow run.

Return type

str

toil.server.wes.tasks.run_wes
toil.server.wes.tasks.cancel_run(task_id)

Send a SIGTERM signal to the process that is running task_id.

Parameters

task_id (str) –

Return type

None

class toil.server.wes.tasks.TaskRunner

Abstraction over the Celery API. Runs our run_wes task and allows canceling it.

We can swap this out in the server to allow testing without Celery.

static run(args, task_id)

Run the given task args with the given ID on Celery.

Parameters
Return type

None

static cancel(task_id)

Cancel the task with the given ID on Celery.

Parameters

task_id (str) –

Return type

None

static is_ok(task_id)

Make sure that the task running system is working for the given task. If the task system has detected an internal failure, return False.

Parameters

task_id (str) –

Return type

bool

class toil.server.wes.tasks.MultiprocessingTaskRunner

Bases: TaskRunner

digraph inheritance47fd0023c3 { bgcolor=transparent; rankdir=LR; size="8.0, 12.0"; "MultiprocessingTaskRunner" [URL="#toil.server.wes.tasks.MultiprocessingTaskRunner",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Version of TaskRunner that just runs tasks with Multiprocessing."]; "TaskRunner" -> "MultiprocessingTaskRunner" [arrowsize=0.5,style="setlinewidth(0.5)"]; "TaskRunner" [URL="#toil.server.wes.tasks.TaskRunner",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Abstraction over the Celery API. Runs our run_wes task and allows canceling it."]; }

Version of TaskRunner that just runs tasks with Multiprocessing.

Can’t use threading because there’s no way to send a cancel signal or exception to a Python thread, if loops in the task (i.e. ToilWorkflowRunner) don’t poll for it.

static set_up_and_run_task(output_path, args)

Set up logging for the process into the given file and then call run_wes_task with the given arguments.

If the process finishes successfully, it will clean up the log, but if the process crashes, the caller must clean up the log.

Parameters
Return type

None

classmethod run(args, task_id)

Run the given task args with the given ID.

Parameters
Return type

None

classmethod cancel(task_id)

Cancel the task with the given ID.

Parameters

task_id (str) –

Return type

None

classmethod is_ok(task_id)

Make sure that the task running system is working for the given task. If the task system has detected an internal failure, return False.

Parameters

task_id (str) –

Return type

bool