Multi-endpoint baseline

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

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()

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))