Blue/various changes (#674)

* simplify the ex `erigon` example
Update exescript commands in all examples to use the new interface
Add a common "environment info" message for all examples
Add a common boilerplate to all examples
Add ability to pass parameters to `Script()` through `WorkContext.new_script()`
Update default subnet to `devnet-beta`

README.md

@@ -198,7 +198,7 @@ It's possible to set various elements of `yagna` configuration through environme
 - `YAGNA_PAYMENT_NETWORK`, Ethereum network name for `yagna` to use, e.g. `rinkeby`
 - `YAGNA_PAYMENT_DRIVER`, payment driver name for `yagna` to use, e.g. `zksync`
 - `YAGNA_PAYMENT_URL`, URL to `yagna` payment API, e.g. `http://localhost:7500/payment-api/v1`
-- `YAGNA_SUBNET`, name of the `yagna` sub network to be used, e.g. `devnet-beta.2`
+- `YAGNA_SUBNET`, name of the `yagna` sub network to be used, e.g. `devnet-beta`
 - `YAPAPI_USE_GFTP_CLOSE`, if set to a _truthy_ value (e.g. "1", "Y", "True", "on") then `yapapi`
   will ask `gftp` to close files when there's no need to publish them any longer. This may greatly
   reduce the number of files kept open while `yapapi` is running but requires `yagna`

examples/blender/blender.py

@@ -1,18 +1,13 @@
 #!/usr/bin/env python3
-import asyncio
 from datetime import datetime, timedelta
 import pathlib
 import sys
 
 from yapapi import (
     Golem,
-    NoPaymentAccountError,
     Task,
-    __version__ as yapapi_version,
     WorkContext,
-    windows_event_loop_fix,
 )
-from yapapi.log import enable_default_logger
 from yapapi.payload import vm
 from yapapi.rest.activity import BatchTimeoutError
 
@@ -27,6 +22,8 @@
     TEXT_COLOR_YELLOW,
     TEXT_COLOR_MAGENTA,
     format_usage,
+    run_golem_example,
+    print_env_info,
 )
 
 
@@ -40,12 +37,19 @@ async def main(subnet_tag, payment_driver=None, payment_network=None, show_usage
     async def worker(ctx: WorkContext, tasks):
         script_dir = pathlib.Path(__file__).resolve().parent
         scene_path = str(script_dir / "cubes.blend")
-        ctx.send_file(scene_path, "/golem/resource/scene.blend")
+
+        # Set timeout for the first script executed on the provider. Usually, 30 seconds
+        # should be more than enough for computing a single frame of the provided scene,
+        # however a provider may require more time for the first task if it needs to download
+        # the VM image first. Once downloaded, the VM image will be cached and other tasks that use
+        # that image will be computed faster.
+        script = ctx.new_script(timeout=timedelta(minutes=10))
+        script.upload_file(scene_path, "/golem/resource/scene.blend")
+
         async for task in tasks:
             frame = task.data
             crops = [{"outfilebasename": "out", "borders_x": [0.0, 1.0], "borders_y": [0.0, 1.0]}]
-            ctx.send_json(
-                "/golem/work/params.json",
+            script.upload_json(
                 {
                     "scene_file": "/golem/resource/scene.blend",
                     "resolution": (400, 300),
@@ -58,17 +62,14 @@ async def worker(ctx: WorkContext, tasks):
                     "WORK_DIR": "/golem/work",
                     "OUTPUT_DIR": "/golem/output",
                 },
+                "/golem/work/params.json",
             )
-            ctx.run("/golem/entrypoints/run-blender.sh")
+
+            script.run("/golem/entrypoints/run-blender.sh")
             output_file = f"output_{frame}.png"
-            ctx.download_file(f"/golem/output/out{frame:04d}.png", output_file)
+            script.download_file(f"/golem/output/out{frame:04d}.png", output_file)
             try:
-                # Set timeout for executing the script on the provider. Usually, 30 seconds
-                # should be more than enough for computing a single frame, however a provider
-                # may require more time for the first task if it needs to download a VM image
-                # first. Once downloaded, the VM image will be cached and other tasks that use
-                # that image will be computed faster.
-                yield ctx.commit(timeout=timedelta(minutes=10))
+                yield script
                 # TODO: Check if job results are valid
                 # and reject by: task.reject_task(reason = 'invalid file')
                 task.accept_result(result=output_file)
@@ -80,6 +81,9 @@ async def worker(ctx: WorkContext, tasks):
                 )
                 raise
 
+            # reinitialize the script which we send to the engine to compute subsequent frames
+            script = ctx.new_script(timeout=timedelta(minutes=1))
+
             if show_usage:
                 raw_state = await ctx.get_raw_state()
                 usage = format_usage(await ctx.get_usage())
@@ -118,13 +122,7 @@ async def worker(ctx: WorkContext, tasks):
         payment_driver=payment_driver,
         payment_network=payment_network,
     ) as golem:
-
-        print(
-            f"yapapi version: {TEXT_COLOR_YELLOW}{yapapi_version}{TEXT_COLOR_DEFAULT}\n"
-            f"Using subnet: {TEXT_COLOR_YELLOW}{subnet_tag}{TEXT_COLOR_DEFAULT}, "
-            f"payment driver: {TEXT_COLOR_YELLOW}{golem.payment_driver}{TEXT_COLOR_DEFAULT}, "
-            f"and network: {TEXT_COLOR_YELLOW}{golem.payment_network}{TEXT_COLOR_DEFAULT}\n"
-        )
+        print_env_info(golem)
 
         num_tasks = 0
         start_time = datetime.now()
@@ -158,52 +156,12 @@ async def worker(ctx: WorkContext, tasks):
     parser.set_defaults(log_file=f"blender-yapapi-{now}.log")
     args = parser.parse_args()
 
-    # This is only required when running on Windows with Python prior to 3.8:
-    windows_event_loop_fix()
-
-    enable_default_logger(
-        log_file=args.log_file,
-        debug_activity_api=True,
-        debug_market_api=True,
-        debug_payment_api=True,
-    )
-
-    loop = asyncio.get_event_loop()
-    task = loop.create_task(
+    run_golem_example(
         main(
             subnet_tag=args.subnet_tag,
             payment_driver=args.payment_driver,
             payment_network=args.payment_network,
             show_usage=args.show_usage,
-        )
+        ),
+        log_file=args.log_file,
     )
-
-    try:
-        loop.run_until_complete(task)
-    except NoPaymentAccountError as e:
-        handbook_url = (
-            "https://handbook.golem.network/requestor-tutorials/"
-            "flash-tutorial-of-requestor-development"
-        )
-        print(
-            f"{TEXT_COLOR_RED}"
-            f"No payment account initialized for driver `{e.required_driver}` "
-            f"and network `{e.required_network}`.\n\n"
-            f"See {handbook_url} on how to initialize payment accounts for a requestor node."
-            f"{TEXT_COLOR_DEFAULT}"
-        )
-    except KeyboardInterrupt:
-        print(
-            f"{TEXT_COLOR_YELLOW}"
-            "Shutting down gracefully, please wait a short while "
-            "or press Ctrl+C to exit immediately..."
-            f"{TEXT_COLOR_DEFAULT}"
-        )
-        task.cancel()
-        try:
-            loop.run_until_complete(task)
-            print(
-                f"{TEXT_COLOR_YELLOW}Shutdown completed, thank you for waiting!{TEXT_COLOR_DEFAULT}"
-            )
-        except (asyncio.CancelledError, KeyboardInterrupt):
-            pass

examples/custom-usage-counter/custom_usage_counter.py

@@ -44,8 +44,9 @@ async def run(self):
         start_time = datetime.now()
         print(f"service {self.id} running on '{self.provider_name}'...")
         while datetime.now() < start_time + timedelta(seconds=self._running_time_sec):
-            self._ctx.run("sleep", "1000")
-            yield self._ctx.commit()
+            script = self._ctx.new_script()
+            script.run("sleep", "1000")
+            yield script
             usage: ActivityUsage = await self._ctx.get_usage()
             cost = await self._ctx.get_cost()
             print(f"total cost so far: {cost}; activity usage: {usage.current_usage}")

examples/custom_runtime/custom_runtime.py

@@ -0,0 +1,72 @@
+import asyncio
+
+from dataclasses import dataclass
+
+from yapapi.props.base import prop, constraint
+from yapapi.props import inf
+
+from yapapi.payload import Payload
+from yapapi import Golem
+from yapapi.services import Service
+
+
+RUNTIME_NAME = "my-runtime"
+SOME_CUSTOM_PROPERTY = "golem.srv.app.eth.network"
+
+# The `CustomPayload` in this example is an arbitrary definition of some demand
+# that the requestor might wish to be fulfilled by the providers on the Golem network
+#
+# This payload must correspond with a runtime running on providers, either a runtime
+# written by some other party, or developed alongside its requestor counterpart using
+# the Runtime SDK (https://github.com/golemfactory/ya-runtime-sdk/)
+#
+# It is up to the author of said runtime to define any additional properties that would
+# describe the requestor's demand and `custom_property` is just an example.
+
+
+@dataclass
+class CustomPayload(Payload):
+    custom_property: str = prop(SOME_CUSTOM_PROPERTY)
+
+    runtime: str = constraint(inf.INF_RUNTIME_NAME, default=RUNTIME_NAME)
+    min_mem_gib: float = constraint(inf.INF_MEM, operator=">=", default=16)
+    min_storage_gib: float = constraint(inf.INF_STORAGE, operator=">=", default=1024)
+
+
+class CustomRuntimeService(Service):
+    @staticmethod
+    async def get_payload():
+        return CustomPayload(custom_property="whatever")
+
+
+async def main(subnet_tag, driver=None, network=None):
+
+    async with Golem(
+        budget=10.0,
+        subnet_tag=subnet_tag,
+        payment_driver=driver,
+        payment_network=network,
+    ) as golem:
+        cluster = await golem.run_service(
+            CustomRuntimeService,
+        )
+
+        def instances():
+            return [f"{s.provider_name}: {s.state.value}" for s in cluster.instances]
+
+        cnt = 0
+        while cnt < 10:
+            print(f"instances: {instances()}")
+            await asyncio.sleep(3)
+
+        cluster.stop()
+
+        cnt = 0
+        while cnt < 10 and any(s.is_available for s in cluster.instances):
+            print(f"instances: {instances()}")
+            await asyncio.sleep(1)
+
+    print(f"instances: {instances()}")
+
+
+asyncio.run(main(None))

examples/hello-world/hello.py

@@ -9,9 +9,10 @@
 
 async def worker(context: WorkContext, tasks: AsyncIterable[Task]):
     async for task in tasks:
-        context.run("/bin/sh", "-c", "date")
+        script = context.new_script()
+        script.run("/bin/sh", "-c", "date")
 
-        future_results = yield context.commit()
+        future_results = yield script
         results = await future_results
         task.accept_result(result=results[-1])
 
@@ -23,7 +24,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)