torc 0.22.1

Workflow management system
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

"""Example showing per-job scheduling priority.

Jobs with higher priority values are claimed by workers first.
When multiple jobs are ready simultaneously and worker capacity is
limited, priority determines execution order.
"""

import getpass
import os

from torc import create_jobs, make_api, setup_logging
from torc.api import DefaultApi
from torc.openapi_client import JobModel, ResourceRequirementsModel, WorkflowModel

TORC_API_URL = os.getenv("TORC_API_URL", "http://localhost:8080/torc-service/v1")


def build_workflow(api: DefaultApi, workflow: WorkflowModel) -> None:
    workflow_id = workflow.id
    assert workflow_id is not None

    small = api.create_resource_requirements(
        ResourceRequirementsModel(
            workflow_id=workflow_id,
            name="small",
            num_cpus=1,
            memory="2g",
            runtime="PT30M",
        )
    )
    medium = api.create_resource_requirements(
        ResourceRequirementsModel(
            workflow_id=workflow_id,
            name="medium",
            num_cpus=4,
            memory="8g",
            runtime="PT1H",
        )
    )

    # preprocess has no dependencies and the highest priority — it runs first.
    preprocess = api.create_job(
        JobModel(
            workflow_id=workflow_id,
            name="preprocess",
            command="python preprocess.py --input raw.csv --output clean.csv",
            priority=10,
            resource_requirements_id=small.id,
        )
    )

    # These three jobs become ready at the same time once preprocess completes.
    # Workers claim them in descending priority order.
    jobs = [
        JobModel(
            workflow_id=workflow_id,
            name="critical_analysis",
            command="python analyze.py --mode critical --input clean.csv",
            priority=5,  # claimed first among the three
            depends_on_job_ids=[preprocess.id],
            resource_requirements_id=medium.id,
        ),
        JobModel(
            workflow_id=workflow_id,
            name="normal_analysis",
            command="python analyze.py --mode normal --input clean.csv",
            priority=3,  # claimed second
            depends_on_job_ids=[preprocess.id],
            resource_requirements_id=medium.id,
        ),
        JobModel(
            workflow_id=workflow_id,
            name="background_report",
            command="python report.py --input clean.csv",
            priority=1,  # claimed last
            depends_on_job_ids=[preprocess.id],
            resource_requirements_id=small.id,
        ),
    ]
    created = create_jobs(api, workflow_id, jobs)

    # Summary waits for all three; priority defaults to 0.
    api.create_job(
        JobModel(
            workflow_id=workflow_id,
            name="summary",
            command="python summarize.py",
            depends_on_job_ids=[j.id for j in created],
            resource_requirements_id=small.id,
        )
    )


def main() -> None:
    setup_logging()
    api = make_api(TORC_API_URL)
    workflow = api.create_workflow(
        WorkflowModel(
            user=getpass.getuser(),
            name="priority_demo",
            description="Demonstrates per-job scheduling priority",
        )
    )
    try:
        build_workflow(api, workflow)
    except Exception:
        api.delete_workflow(workflow.id)
        raise


if __name__ == "__main__":
    main()