diff --git a/example/rhapsody-exchange/dragon/.gitignore b/example/rhapsody-exchange/dragon/.gitignore new file mode 100644 index 0000000..0653a9b --- /dev/null +++ b/example/rhapsody-exchange/dragon/.gitignore @@ -0,0 +1 @@ +ddict_orc_* diff --git a/example/rhapsody-exchange/dragon/driver.py b/example/rhapsody-exchange/dragon/driver.py new file mode 100644 index 0000000..e754c2b --- /dev/null +++ b/example/rhapsody-exchange/dragon/driver.py @@ -0,0 +1,88 @@ +"""RHAPSODY-launched Dragon DDict exchange, executed via RHAPSODY tasks. + +Run with: + dragon -s -- python3 driver.py + +RHAPSODY's DragonDataBackend constructs and owns the `dragon.data.ddict.DDict` +-- this driver never constructs the DDict itself, unlike +../../cpp-exchange/dragon or ../../py-cpp-exchange/dragon. Once the backend +hands back a serialized endpoint, radex takes over exactly as it would +against any other DDict: DragonClient attaches directly from the descriptor. + +The producer/consumer exchange itself runs as RHAPSODY ComputeTasks +dispatched through a DragonExecutionBackend, not inline in this driver +process -- each task function below is fully self-contained (its own +imports, its own radex client, built only from the endpoint passed as an +argument) since a task may execute in a completely separate process with +no knowledge of this module or anything else defined here. +""" + +import asyncio + +from rhapsody.api import ComputeTask, Session +from rhapsody.backends import DragonExecutionBackend +from rhapsody.backends.data import DragonDataBackend + + +def produce(descriptor): + import numpy as np + + from radex.clients.core import DragonClient + from radex.handles.handles import OutgoingHandle + + client = DragonClient(descriptor=descriptor, timeout=5) + + samples = np.arange(10, dtype=np.float64) ** 2 + client.put_tensor(OutgoingHandle("samples"), samples) + client.put_scalar(OutgoingHandle("sample-count"), len(samples)) + return len(samples) + + +def consume(descriptor): + from radex.clients.core import DragonClient + from radex.handles.handles import IncomingHandle + + client = DragonClient(descriptor=descriptor, timeout=5) + + samples = client.wait_for_tensor(IncomingHandle("samples"), 10) + count = client.wait_for_scalar(IncomingHandle("sample-count"), 10) + return { + "count": int(count), + "sum": float(samples.sum()), + "mean": float(samples.mean()), + } + + +async def main() -> int: + print("Driver: Starting Backends", flush=True) + session = Session() + exec_backend = await DragonExecutionBackend() + data_backend = await DragonDataBackend(managers_per_node=1, n_nodes=1) + + session.add_backend(exec_backend) + session.add_backend(data_backend) + + descriptor = data_backend.endpoints[0].serialize() + + tasks = [ + ComputeTask(function=produce, args=(descriptor,)), + ComputeTask(function=consume, args=(descriptor,)), + ] + + print("Driver: Submitting tasks", flush=True) + futures = await session.submit_tasks(tasks) + await asyncio.gather(*futures) + + for task in tasks: + print(f"Driver: Task {task.uid} in {task.state} state.", flush=True) + print(f"Driver: Output: {task.return_value}", flush=True) + + print("Driver: Shutting down", flush=True) + await data_backend.shutdown() + await exec_backend.shutdown() + + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/example/rhapsody-exchange/dragon/requirements.txt b/example/rhapsody-exchange/dragon/requirements.txt new file mode 100644 index 0000000..91e107a --- /dev/null +++ b/example/rhapsody-exchange/dragon/requirements.txt @@ -0,0 +1,3 @@ +numpy +dragonhpc[telemetry]>=0.14.1 +rhapsody-py @ git+https://github.com/radical-cybertools/rhapsody.git@feature/data-backends diff --git a/example/rhapsody-exchange/redis/.gitignore b/example/rhapsody-exchange/redis/.gitignore new file mode 100644 index 0000000..2516d72 --- /dev/null +++ b/example/rhapsody-exchange/redis/.gitignore @@ -0,0 +1,3 @@ +dump.rdb +rhapsody.data.* +rhapsody.session.* diff --git a/example/rhapsody-exchange/redis/driver.py b/example/rhapsody-exchange/redis/driver.py new file mode 100644 index 0000000..66c886d --- /dev/null +++ b/example/rhapsody-exchange/redis/driver.py @@ -0,0 +1,102 @@ +"""RHAPSODY-launched Redis store exchange, executed via RHAPSODY tasks. + +RHAPSODY's RedisDataBackend launches and owns the `redis-server` process -- +this driver never spawns it directly, unlike ../../cpp-exchange/redis +(SmartSim) or a raw `dragon.data.ddict.DDict(...)` construction. Once the +backend hands back a serialized endpoint, radex takes over exactly as it +would against any other Redis deployment: RedisClient reads its connection +info from RADEX_STORE/RADEX_STORE_OPTS. + +The producer/consumer exchange itself runs as RHAPSODY ComputeTasks +dispatched through a ConcurrentExecutionBackend, not inline in this driver +process -- each task function below is fully self-contained (its own +imports, its own radex client, built only from the endpoint passed as an +argument) since a task may execute in a completely separate process with +no knowledge of this module or anything else defined here. +""" + +import asyncio +import os + +from rhapsody.api import ComputeTask, Session +from rhapsody.backends import ConcurrentExecutionBackend +from rhapsody.backends.data import RedisDataBackend + + +def produce(descriptor): + import os + + import numpy as np + + from radex.clients.core import RedisClient + from radex.handles.handles import OutgoingHandle + + os.environ["RADEX_STORE"] = descriptor + os.environ["RADEX_STORE_OPTS"] = "Standalone" + client = RedisClient() + + samples = np.arange(10, dtype=np.float64) ** 2 + client.put_tensor(OutgoingHandle("samples"), samples) + client.put_scalar(OutgoingHandle("sample-count"), len(samples)) + return len(samples) + + +def consume(descriptor): + import os + + from radex.clients.core import RedisClient + from radex.handles.handles import IncomingHandle + + os.environ["RADEX_STORE"] = descriptor + os.environ["RADEX_STORE_OPTS"] = "Standalone" + client = RedisClient() + + samples = client.wait_for_tensor(IncomingHandle("samples"), 10) + count = client.wait_for_scalar(IncomingHandle("sample-count"), 10) + return { + "count": int(count), + "sum": float(samples.sum()), + "mean": float(samples.mean()), + } + + +async def main() -> int: + # Session is constructed first so its work_dir/uid exist before + # RedisDataBackend launches redis-server -- that lets the server's log + # file land inside the session's own directory. + + print("Driver: Starting Backends", flush=True) + session = Session(uid="radex.session.0000") + + data_backend = await RedisDataBackend( + work_dir=os.path.join(session.work_dir, session.uid) + ) + + exec_backend = await ConcurrentExecutionBackend() + + session.add_backend(exec_backend) + session.add_backend(data_backend) + + descriptor = data_backend.endpoints[0].serialize() + + tasks = [ + ComputeTask(function=produce, args=(descriptor,)), + ComputeTask(function=consume, args=(descriptor,)), + ] + + print("Driver: Submitting tasks", flush=True) + futures = await session.submit_tasks(tasks) + await asyncio.gather(*futures) + + for task in tasks: + print(f"Driver: Task {task.uid} in {task.state} state.", flush=True) + print(f"Driver: Output: {task.return_value}", flush=True) + + print("Driver: Shutting down", flush=True) + await session.close() + + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/example/rhapsody-exchange/redis/requirements.txt b/example/rhapsody-exchange/redis/requirements.txt new file mode 100644 index 0000000..5d6a11f --- /dev/null +++ b/example/rhapsody-exchange/redis/requirements.txt @@ -0,0 +1,2 @@ +numpy +rhapsody-py @ git+https://github.com/radical-cybertools/rhapsody.git@feature/data-backends