pycolonies/examples/workflow_example2.py at main · ossko/pycolonies · GitHub

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
from pycolonies import colonies_client
from pycolonies import func_spec
from pycolonies import Workflow

colonies, colonyname, colony_prvkey, executor_name, prvkey = colonies_client()

def gen_nums(ctx={}):
    return 1, 2

def reduce(*nums, ctx={}):
    total = 0
    for n in nums:
        total += n
    return total

wf = Workflow(colonyname=colonyname)
f = func_spec(func=gen_nums,
              args=[],
              colonyname=colonyname,
              executortype="python-executor",
              priority=200,
              maxexectime=100,
              maxretries=3,
              maxwaittime=100)

f.nodename = "gen_nums1"
wf.functionspecs.append(f)

f = func_spec(func=gen_nums,
              args=[],
              colonyname=colonyname,
              executortype="python-executor",
              priority=200,
              maxexectime=100,
              maxretries=3,
              maxwaittime=100)

f.nodename = "gen_nums2"
wf.functionspecs.append(f)

func_spec = func_spec(func=reduce,
                             args=[],
                             colonyname=colonyname,
                             executortype="python-executor",
                             priority=200,
                             maxexectime=100,
                             maxretries=3,
                             maxwaittime=100)

func_spec.conditions.dependencies = ["gen_nums1", "gen_nums2"]
wf.functionspecs.append(func_spec)

processgraph = colonies.submit_workflow(wf, prvkey)
print("Workflow", processgraph.processgraphid, "submitted")

# wait for the sum_list process
process = colonies.find_process("reduce", processgraph.processids, prvkey)
process = colonies.wait(process, 100, prvkey)
print(process.output[0])