Another issue with graph analysis and conditional branches: error handler

[woutdenolf]

import logging
from ewoks import execute_graph
from ewokscore.task import Task


class MyError(Exception):
    pass


class Dummy(Task, optional_input_names=["raise_error"]):
    def run(self):
        if self.inputs.raise_error:
            raise MyError(f"Error raise in {self.label!r}")
        else:
            print(self.label, "Done")


class Error(Task, optional_input_names=["WorkflowExceptionInstance"]):
    def run(self):
        print(
            f"Error captured in {self.label!r}: {self.inputs.WorkflowExceptionInstance}"
        )


node_locate_data = {
    "id": "locate_data",
    "task_type": "class",
    "force_start_node": True,
    "task_identifier": "__main__.Dummy",
}

node_termination = {
    "id": "terminate",
    "task_type": "class",
    "task_identifier": "__main__.Error",
}


node_setup_workdir = {
    "id": "setup_workdir",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

node_load_master = {
    "id": "load_master",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

node_load_metadata = {
    "id": "load_metadata",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

node_draw_plots = {
    "id": "draw_plots",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

link_termination = {
    "source": "locate_data",
    "target": "terminate",
    "on_error": True,
}

link_locate_to_setup = {
    "source": "locate_data",
    "target": "setup_workdir",
}

link_locate_to_master = {
    "source": "locate_data",
    "target": "load_master",
}

link_locate_to_metadata = {
    "source": "locate_data",
    "target": "load_metadata",
}

link_setup_to_master = {
    "source": "setup_workdir",
    "target": "load_master",
}

link_setup_to_metadata = {
    "source": "setup_workdir",
    "target": "load_metadata",
}

link_master_to_draw = {
    "source": "load_master",
    "target": "draw_plots",
}

link_metadata_to_draw = {
    "source": "load_metadata",
    "target": "draw_plots",
}


workflow = {
    "graph": {"id": "workflow_ID29ESRF_draw_foil_2Dmap"},
    "nodes": [
        node_locate_data,
        node_termination,
        node_setup_workdir,
        node_load_master,
        node_load_metadata,
        node_draw_plots,
    ],
    "links": [
        link_termination,
        link_locate_to_setup,
        link_locate_to_master,
        link_locate_to_metadata,
        link_setup_to_master,
        link_setup_to_metadata,
        link_master_to_draw,
        link_metadata_to_draw,
    ],
}

print("\nSTART\n")
logging.basicConfig(level=logging.CRITICAL)

execute_graph(
    workflow,
    engine="ppf",
    pool_type="thread",
    inputs=[{"name": "raise_error", "value": False, "id": "locate_data"}],
)
print("\nEXIT\n")

START

locate_data Done
setup_workdir Done
load_master Done
load_master Done
load_metadata Done
draw_plots Done
draw_plots Done
draw_plots Done
load_metadata Done
draw_plots Done

EXIT

ewoksppf sees this:

locate_data -> terminate required=False
locate_data -> setup_workdir required=True
locate_data -> load_master required=True
locate_data -> load_metadata required=True
setup_workdir -> load_master required=False          ????
setup_workdir -> load_metadata required=False     ????
load_master -> draw_plots required=False               ????
load_metadata -> draw_plots required=False

[woutdenolf]

@TamasFoldes FYI

[woutdenolf]

The current solution to this case is to use the link property required

import logging
from ewoks import execute_graph
from ewokscore.task import Task


class MyError(Exception):
    pass


class Dummy(Task, optional_input_names=["raise_error"]):
    def run(self):
        if self.inputs.raise_error:
            raise MyError(f"Error raise in {self.label!r}")
        else:
            print(self.label, "Done")


class Error(Task, optional_input_names=["WorkflowExceptionInstance"]):
    def run(self):
        print(
            f"Error captured in {self.label!r}: {self.inputs.WorkflowExceptionInstance}"
        )


node_locate_data = {
    "id": "locate_data",
    "task_type": "class",
    "force_start_node": True,
    "task_identifier": "__main__.Dummy",
}

node_termination = {
    "id": "terminate",
    "task_type": "class",
    "task_identifier": "__main__.Error",
}


node_setup_workdir = {
    "id": "setup_workdir",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

node_load_master = {
    "id": "load_master",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

node_load_metadata = {
    "id": "load_metadata",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

node_draw_plots = {
    "id": "draw_plots",
    "task_type": "class",
    "task_identifier": "__main__.Dummy",
}

link_termination = {
    "source": "locate_data",
    "target": "terminate",
    "on_error": True,
}

link_locate_to_setup = {
    "source": "locate_data",
    "target": "setup_workdir",
}

link_locate_to_master = {
    "source": "locate_data",
    "target": "load_master",
}

link_locate_to_metadata = {
    "source": "locate_data",
    "target": "load_metadata",
}

link_setup_to_master = {
    "source": "setup_workdir",
    "target": "load_master",
    "required": True,
}

link_setup_to_metadata = {
    "source": "setup_workdir",
    "target": "load_metadata",
    "required": True,
}

link_master_to_draw = {
    "source": "load_master",
    "target": "draw_plots",
    "required": True,
}

link_metadata_to_draw = {
    "source": "load_metadata",
    "target": "draw_plots",
    "required": True,
}


workflow = {
    "graph": {"id": "workflow_ID29ESRF_draw_foil_2Dmap"},
    "nodes": [
        node_locate_data,
        node_termination,
        node_setup_workdir,
        node_load_master,
        node_load_metadata,
        node_draw_plots,
    ],
    "links": [
        link_termination,
        link_locate_to_setup,
        link_locate_to_master,
        link_locate_to_metadata,
        link_setup_to_master,
        link_setup_to_metadata,
        link_master_to_draw,
        link_metadata_to_draw,
    ],
}

print("\nSTART\n")
logging.basicConfig(level=logging.CRITICAL)

execute_graph(
    workflow,
    engine="ppf",
    pool_type="thread",
    inputs=[{"name": "raise_error", "value": False, "id": "locate_data"}],
)
print("\nEXIT\n")

[woutdenolf]

When locate_data does not raise an error, you have this without using "required"

locate_data Done
setup_workdir Done
load_master Done
load_master Done
load_metadata Done
load_metadata Done
draw_plots Done
draw_plots Done
draw_plots Done
draw_plots Done

and this when using "required" is True on the appropriate links

locate_data Done
setup_workdir Done
load_master Done
load_metadata Done
draw_plots Done