diff --git a/test/09-service/agent.py b/test/09-service/agent.py index e1bc94b..d6b5ae4 100644 --- a/test/09-service/agent.py +++ b/test/09-service/agent.py @@ -23,10 +23,11 @@ def __init__(self, flow: WorkflowEngine, *args, **kwargs): # no learning. Simple investigator self.investigator = MyModel(flow) - @self.flow.function_task + @self.flow.function_task(backend="inference") async def model_select( in_data: TypedData, i_id=self.investigator.get_id(), model_kwargs={} ): + print("\n RUNNING SELECTOR............ \n") return i_id # default to latest model self.model_selector = model_select diff --git a/test/09-service/model.py b/test/09-service/model.py index 676d8bd..e883586 100644 --- a/test/09-service/model.py +++ b/test/09-service/model.py @@ -1,5 +1,6 @@ import asyncio import logging +import time from radical.asyncflow import WorkflowEngine from digitaltwin.components import ModelInvestigator, TypedData @@ -25,21 +26,28 @@ def __init__(self, flow: WorkflowEngine, *args, **kwargs): super().__init__(flow) self.flow = flow - @self.flow.function_task - async def compute(in_data: TypedData, offset=1): - return offset - in_data.data + @self.flow.function_task(backend="learning") + async def compute(): + print("\n RUNNING SIM ............ \n") + time.sleep(5) + return self.compute = compute - async def main_loop(self, runtime: RuntimeAPI): + @self.flow.function_task(backend="inference") async def do_inference(in_data: TypedData, offset=1): - return TypedData(INFERENCE_DTYPE, await self.compute(in_data, - offset=offset)) + print("\n RUNNING INFERENCE............ \n") + return TypedData(INFERENCE_DTYPE, offset - in_data.data) + + self.do_inference = do_inference + + async def main_loop(self, runtime: RuntimeAPI): - runtime.set_inference_task(do_inference) + runtime.set_inference_task(self.do_inference) offset = 2 while True: runtime.publish_new_model({"offset": offset}) offset += 1 - await asyncio.sleep(5) + # simulate a long sim + await self.compute() diff --git a/test/09-service/run_me.py b/test/09-service/run_me.py index 4f2f028..80acc0f 100644 --- a/test/09-service/run_me.py +++ b/test/09-service/run_me.py @@ -41,10 +41,20 @@ # engine wiring, stated explicitly: one 'inference' backend per session, on a # co-located endpoint, with the concurrent backend +# ENGINES = { +# "engines": { +# "inference": {"endpoint_name": TASK_ENDPOINT, "backends": ["concurrent"]} +# } +# } + +# "pools": [{"name": "exsitu", "endpoint_name": "hpc"}, +# {"name": "insitu", "endpoint_name": "pi"}], + ENGINES = { - "engines": { - "inference": {"endpoint_name": TASK_ENDPOINT, "backends": ["concurrent"]} - } + "engines": { + "inference" : {"endpoint_name": "pi", "backends": ["concurrent"]}, + "learning" : {"endpoint_name": "hpc", "backends": ["dragon"]}, + } } @@ -72,7 +82,7 @@ def main(): print(json.dumps(dt.describe(twin), indent=2)) dt.start(twin) - time.sleep(15) + time.sleep(30) print(json.dumps(dt.twin(twin), indent=2))