import datetime
import os
import time
from ewokscore import Task
[docs]
class ConcurencyTestTask(
Task,
input_names=["output_filename", "sleep_time", "task_id"],
output_names=["slept_time"],
):
"""
This task takes a fulltomo sequence and creates the associated NXtomo file.
"""
[docs]
def run(self):
with open(self.inputs.output_filename, "a") as fd:
fd.write(
f"Task {self.inputs.task_id} started at {datetime.datetime.now().isoformat()} {os.getpid()}\n"
)
t0 = time.perf_counter()
time.sleep(self.inputs.sleep_time)
t1 = time.perf_counter()
fd.write(
f"Task {self.inputs.task_id} stopped at {datetime.datetime.now().isoformat()} slept {t1 - t0}s\n"
)
self.outputs.slept_time = t1 - t0
# wrk = {"graph": {'id': 'sinogram2nx'}, 'nodes': [{'id': 'sino2nx', 'task_type': 'class', 'task_identifier': 'ewoksid16a.tasks.fluotomo.nxtomo_sinogram.FluoSinogram2Nx'}]}
# inputs = [{'id': 'sino2nx', 'name':'output_root_uri', 'value': "/data/visitor/blc16919/id16a/20260303/PROCESSED_DATA/AlGe40/AlGe40_sinogram_050nm_0001/fluofit/scan_0001/AlGe40_sinogram_050nm_0001_0001.h5::/1.1"},
# ...: {'id': 'sino2nx', 'name':'output_root_group', 'value': "weighted"}, ]
# ...: kwargs = { "engine": "dask", 'inputs': inputs, "outputs": [{"all": True}], "scheduler": "multiprocessing",}
# ...: from ewoksjob.client import submit
# ...: submit(args=(wrk,), kwargs=kwargs, queue='fluo_dev').get()
# if False:
#
# wrk = {"graph": {'id': 'concurencyTest'},
# "nodes": [
# {'id': "start", 'task_type': 'class', 'task_identifier': 'ewoksid16a.tasks.common.tests.ConcurencyTestTask'},
# {'id': "branchA", 'task_type': 'class', 'task_identifier': 'ewoksid16a.tasks.common.tests.ConcurencyTestTask'},
# {'id': "branchB", 'task_type': 'class', 'task_identifier': 'ewoksid16a.tasks.common.tests.ConcurencyTestTask'},
# {'id': "branchC", 'task_type': 'class', 'task_identifier': 'ewoksid16a.tasks.common.tests.ConcurencyTestTask'},
# {'id': "end", 'task_type': 'class', 'task_identifier': 'ewoksid16a.tasks.common.tests.ConcurencyTestTask'},
# ],
# "links": [
# {'source': "start", "target": "branchA"},
# {'source': "start", "target": "branchB"},
# {'source': "start", "target": "branchC"},
# {'source': "branchA", "target": "end"},
# {'source': "branchB", "target": "end"},
# {'source': "branchC", "target": "end"},
# ]
# }
#
# inputs = [
# {'id': "start", 'name': "output_filename", "value": "/tmp_14_days/chevremo/output_start_end.log"},
# {'id': "start", 'name': "sleep_time", "value": 10},
# {'id': "start", 'name': "task_id", "value": "start"},
#
# {'id': "branchA", 'name': "output_filename", "value": "/tmp_14_days/chevremo/output_branch_A.log"},
# {'id': "branchA", 'name': "sleep_time", "value": 10},
# {'id': "branchA", 'name': "task_id", "value": "branchA"},
#
# {'id': "branchB", 'name': "output_filename", "value": "/tmp_14_days/chevremo/output_branch_B.log"},
# {'id': "branchB", 'name': "sleep_time", "value": 10},
# {'id': "branchB", 'name': "task_id", "value": "branchB"},
#
# {'id': "branchC", 'name': "output_filename", "value": "/tmp_14_days/chevremo/output_branch_C.log"},
# {'id': "branchC", 'name': "sleep_time", "value": 10},
# {'id': "branchC", 'name': "task_id", "value": "branchC"},
#
# {'id': "end", 'name': "output_filename", "value": "/tmp_14_days/chevremo/output_start_end.log"},
# {'id': "end", 'name': "sleep_time", "value": 10},
# {'id': "end", 'name': "task_id", "value": "end"},
# ]
#
# kwargs = { "engine": "dask",
# 'inputs': inputs,
# "outputs": [{"all": True}],
# # "scheduler": "multithreading",
# "scheduler": "cluster", "scheduler_options": {"n_workers": 4, "threads_per_worker": 1},
# # "scheduler": "multiprocessing", "scheduler_options": {"num_workers": 4, "context": "spawn"},
# "convert_destination": "/tmp_14_days/chevremo/workflow.json",
# }
#
# from ewoksjob.client import submit
# submit(args=(wrk,), kwargs=kwargs, queue="fluo_dev")