get individual results instead of the accumulated ones (#682)

* get individual results instead of the accumulated ones * update readme and docstrings...
golemfactory · Sep 27, 2021 · b649a89 · b649a89
1 parent 106b02a
commit b649a89
Show file tree

Hide file tree

Showing 6 changed files with 42 additions and 33 deletions.
diff --git a/README.md b/README.md
@@ -160,11 +160,11 @@ from yapapi.payload import vm
 
 async def worker(context: WorkContext, tasks: AsyncIterable[Task]):
     async for task in tasks:
-        context.run("/bin/sh", "-c", "date")
+        script = context.new_script()
+        future_result = script.run("/bin/sh", "-c", "date")
 
-        future_results = yield context.commit()
-        results = await future_results
-        task.accept_result(result=results[-1])
+        yield script
+        task.accept_result(result=await future_result)
 
 
 async def main():
@@ -174,7 +174,7 @@ async def main():
 
     tasks = [Task(data=None)]
 
-    async with Golem(budget=1.0, subnet_tag="devnet-beta.2") as golem:
+    async with Golem(budget=1.0, subnet_tag="devnet-beta") as golem:
         async for completed in golem.execute_tasks(worker, tasks, payload=package):
             print(completed.result.stdout)
 

diff --git a/examples/hello-world/hello.py b/examples/hello-world/hello.py
@@ -10,11 +10,11 @@
 async def worker(context: WorkContext, tasks: AsyncIterable[Task]):
     async for task in tasks:
         script = context.new_script()
-        script.run("/bin/sh", "-c", "date")
+        future_result = script.run("/bin/sh", "-c", "date")
 
-        future_results = yield script
-        results = await future_results
-        task.accept_result(result=results[-1])
+        yield script
+
+        task.accept_result(result=await future_result)
 
 
 async def main():

diff --git a/examples/hello-world/hello_service.py b/examples/hello-world/hello_service.py
@@ -35,15 +35,16 @@ async def run(self):
         while True:
             await asyncio.sleep(REFRESH_INTERVAL_SEC)
             script = self._ctx.new_script()
-            script.run(
+            future_result = script.run(
                 "/bin/sh",
                 "-c",
                 f"cat {DATE_OUTPUT_PATH}",
             )
 
-            future_results = yield script
-            results = await future_results
-            print(results[0].stdout.strip() if results[0].stdout else "")
+            yield script
+
+            result = (await future_result).stdout
+            print(result.strip() if result else "")
 
 
 async def main():

diff --git a/examples/simple-service-poc/simple_service.py b/examples/simple-service-poc/simple_service.py
@@ -75,13 +75,13 @@ async def run(self):
         while True:
             await asyncio.sleep(10)
             script = self._ctx.new_script()
-            script.run(self.SIMPLE_SERVICE, "--stats")  # idx 0
-            script.run(self.SIMPLE_SERVICE, "--plot", "dist")  # idx 1
+            stats_results = script.run(self.SIMPLE_SERVICE, "--stats")
+            plot_results = script.run(self.SIMPLE_SERVICE, "--plot", "dist")
 
-            future_results = yield script
-            results = await future_results
-            stats = results[0].stdout.strip()
-            plot = results[1].stdout.strip().strip('"')
+            yield script
+
+            stats = (await stats_results).stdout.strip()
+            plot = (await plot_results).stdout.strip().strip('"')
 
             print(f"{TEXT_COLOR_CYAN}stats: {stats}{TEXT_COLOR_DEFAULT}")
 

diff --git a/yapapi/golem.py b/yapapi/golem.py
@@ -335,26 +335,32 @@ async def get_payload():
                     )
 
                 async def start(self):
+                    async for script in super().start():
+                        yield script
+
                     # every `DATE_POLL_INTERVAL` write output of `date` to `DATE_OUTPUT_PATH`
-                    self._ctx.run(
+                    script = self._ctx.new_script()
+                    script.run(
                         "/bin/sh",
                         "-c",
                         f"while true; do date > {DATE_OUTPUT_PATH}; sleep {REFRESH_INTERVAL_SEC}; done &",
                     )
-                    yield self._ctx.commit()
+                    yield script
 
                 async def run(self):
                     while True:
                         await asyncio.sleep(REFRESH_INTERVAL_SEC)
-                        self._ctx.run(
+                        script = self._ctx.new_script()
+                        future_result = script.run(
                             "/bin/sh",
                             "-c",
                             f"cat {DATE_OUTPUT_PATH}",
                         )
 
-                        future_results = yield self._ctx.commit()
-                        results = await future_results
-                        print(results[0].stdout.strip())
+                        yield script
+
+                        result = (await future_result).stdout
+                        print(result.strip() if result else "")
 
 
             async def main():

diff --git a/yapapi/services.py b/yapapi/services.py
@@ -268,15 +268,17 @@ async def start(self) -> AsyncGenerator[Script, Awaitable[List[events.CommandEve
 
         **Example**::
 
+
             async def start(self):
+                s = self._ctx.new_script()
                 # deploy the exe-unit
-                self._ctx.deploy()
+                s.deploy(**self.get_deploy_args())
                 # start the exe-unit's container
-                self._ctx.start()
+                s.start()
                 # start some service process within the container
-                self._ctx.run("/golem/run/service_ctl", "--start")
+                s.run("/golem/run/service_ctl", "--start")
                 # send the batch to the provider
-                yield self._ctx.commit()
+                yield s
 
         ### Default implementation
 
@@ -332,10 +334,10 @@ async def run(self) -> AsyncGenerator[Script, Awaitable[List[events.CommandEvent
 
             async def run(self):
                 while True:
-                    self._ctx.run("/golem/run/report", "--stats")  # index 0
-                    future_results = yield self._ctx.commit()
-                    results = await future_results
-                    stats = results[0].stdout.strip()  # retrieve from index 0
+                    script = self._ctx.new_script()
+                    stats_results = script.run(self.SIMPLE_SERVICE, "--stats")
+                    yield script
+                    stats = (await stats_results).stdout.strip()
                     print(f"stats: {stats}")
 
         **Default implementation**