From 93a4513e08bab731581ce4a0217f9b76cbf027e9 Mon Sep 17 00:00:00 2001 From: AymenFJA Date: Mon, 24 Aug 2026 19:32:41 +0000 Subject: [PATCH 1/2] Add RHAPSODY DataBackend examples using RADEX clients Demonstrate RedisDataBackend/DragonDataBackend launching the store and handing off its endpoint to a RADEX client (RedisClient/DragonClient), with the exchange executed as RHAPSODY tasks via Concurrent/Dragon execution backends. --- example/rhapsody-exchange/dragon/.gitignore | 1 + example/rhapsody-exchange/dragon/driver.py | 88 +++++++++++++++ .../rhapsody-exchange/dragon/requirements.txt | 3 + example/rhapsody-exchange/dump.rdb | Bin 0 -> 253 bytes example/rhapsody-exchange/redis/.gitignore | 3 + example/rhapsody-exchange/redis/driver.py | 102 ++++++++++++++++++ .../rhapsody-exchange/redis/requirements.txt | 2 + 7 files changed, 199 insertions(+) create mode 100644 example/rhapsody-exchange/dragon/.gitignore create mode 100644 example/rhapsody-exchange/dragon/driver.py create mode 100644 example/rhapsody-exchange/dragon/requirements.txt create mode 100644 example/rhapsody-exchange/dump.rdb create mode 100644 example/rhapsody-exchange/redis/.gitignore create mode 100644 example/rhapsody-exchange/redis/driver.py create mode 100644 example/rhapsody-exchange/redis/requirements.txt 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/dump.rdb b/example/rhapsody-exchange/dump.rdb new file mode 100644 index 0000000000000000000000000000000000000000..44da6c3fa4b8b90664dcfe322ec8bcc4e2ce9de5 GIT binary patch literal 253 zcmWG?b@2=~FfcUz#aWb^l3A=5xUrsf{1*u%r{iz6{VO*bjAIQ0O-KZf5d3=D#~sU?Xii6x0vR>g_A1v#n3 zhlL~ 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 From e453ae0e00d09b7dc5909c86cce600356674ea9a Mon Sep 17 00:00:00 2001 From: AymenFJA Date: Wed, 26 Aug 2026 03:59:55 +0000 Subject: [PATCH 2/2] remove rdb file --- example/rhapsody-exchange/dump.rdb | Bin 253 -> 0 bytes 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 example/rhapsody-exchange/dump.rdb diff --git a/example/rhapsody-exchange/dump.rdb b/example/rhapsody-exchange/dump.rdb deleted file mode 100644 index 44da6c3fa4b8b90664dcfe322ec8bcc4e2ce9de5..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 253 zcmWG?b@2=~FfcUz#aWb^l3A=5xUrsf{1*u%r{iz6{VO*bjAIQ0O-KZf5d3=D#~sU?Xii6x0vR>g_A1v#n3 zhlL~