From 132a14168e2e93eb6b1247044204e428e4692d61 Mon Sep 17 00:00:00 2001 From: Philipp Oppermann Date: Tue, 11 Aug 2026 13:10:48 +0000 Subject: [PATCH] refactor(runtime): split operator runtime into per-language crates Extract a language-neutral runtime SDK, `dora-runtime-api` (event loop, node harness, and the `OperatorRunner` backend trait + `main(runner)` entry), and move the two runners into their own backend crates: `dora-runtime-shared-lib` (libloading/C-ABI, shipped in the `dora` CLI) and `dora-runtime-python` (the only crate linking PyO3, shipped in the wheel). The `python` cargo feature and all `#[cfg(feature = "python")]` dispatch are gone; adding a language is now one crate implementing `OperatorRunner`. `dora-cli` no longer pulls pyo3 in at all. Table-drive the daemon's runtime spawn logic via a new `spawn/runtime_registry.rs` keyed on `OperatorSource::runtime_name()` (new helper in `dora-message`, with `RUNTIME_*` name constants). The `python` / `shared-library` built-ins reproduce the existing launch commands verbatim (incl. the #1797/#1805 fixes); the registry is the seam where a future third-party-runtime resolver slots in. `dora-runtime-python` also hosts shared-library (and WASM) operators by delegating to `SharedLibRunner`. A daemon that is itself an embedded Python process routes *native* runtime nodes to `python -uc "import dora; dora.start_runtime()"`, and the pre-split `dora-runtime` served them because it compiled the shared-library backend in unconditionally. Without the delegation those operators would fail to init. Both publish workflows drop the now-gone `dora-runtime` for the three new crates, ordered before `dora-cli`. `dora-runtime-python` is excluded from `cargo test --all` alongside `dora-cli-api-python`: it is a plain rlib, so unlike the `extension-module` cdylibs its test binary links libpython, and the CI test job runs without setup-python. Its lib still builds there as a dependency of `dora-cli-api-python`. Behavior-preserving refactor: no descriptor/YAML changes. Co-Authored-By: Claude Opus 4.8 (1M context) Co-Authored-By: Claude Opus 5 (1M context) --- .claude/skills/adora-project/SKILL.md | 2 +- .github/workflows/cargo-release.yml | 6 +- .github/workflows/ci.yml | 8 +- .github/workflows/nightly.yml | 3 + .github/workflows/release.yml | 4 +- .pr-bodies/2777.md | 15 ++ AGENTS.md | 6 +- CLAUDE.md | 14 +- CONTRIBUTING.md | 2 +- Cargo.lock | 55 ++++- Cargo.toml | 8 +- Makefile | 12 +- README.md | 1 + README.zh-CN.md | 1 + apis/python/cli/Cargo.toml | 4 +- apis/python/cli/src/lib.rs | 2 +- apis/python/node/Cargo.toml | 2 +- apis/python/node/src/lib.rs | 2 +- binaries/cli/Cargo.toml | 4 +- binaries/cli/src/command/runtime.rs | 2 +- binaries/cli/src/session.rs | 8 +- binaries/daemon/src/spawn/mod.rs | 1 + binaries/daemon/src/spawn/runtime_registry.rs | 192 ++++++++++++++++ binaries/daemon/src/spawn/spawner.rs | 210 +++--------------- binaries/runtime-api/Cargo.toml | 37 +++ .../operator => runtime-api/src}/channel.rs | 0 binaries/{runtime => runtime-api}/src/lib.rs | 55 +++-- .../mod.rs => runtime-api/src/operator.rs} | 150 ++++--------- binaries/runtime-python/Cargo.toml | 63 ++++++ binaries/runtime-python/build.rs | 13 ++ binaries/runtime-python/src/lib.rs | 149 +++++++++++++ .../src/runner.rs} | 2 +- binaries/runtime-shared-lib/Cargo.toml | 38 ++++ binaries/runtime-shared-lib/src/lib.rs | 144 ++++++++++++ .../src/runner.rs} | 2 +- binaries/runtime/Cargo.toml | 57 ----- binaries/runtime/build.rs | 13 -- docs/architecture.md | 4 +- docs/contributor-qa-cheatsheet.md | 1 + docs/testing-guide.md | 2 + examples/c++-dataflow/README.md | 12 +- examples/c++-dataflow/run.rs | 2 +- examples/c-dataflow/README.md | 12 +- examples/cmake-dataflow/run.rs | 2 +- guide/src/concepts/architecture.md | 4 +- guide/src/development/testing.md | 2 + libraries/core/src/descriptor/mod.rs | 6 +- libraries/message/src/descriptor.rs | 70 ++++++ scripts/qa/all.sh | 2 + scripts/qa/ci-nightly-jobs.sh | 4 + scripts/qa/coverage.sh | 1 + 51 files changed, 973 insertions(+), 438 deletions(-) create mode 100644 .pr-bodies/2777.md create mode 100644 binaries/daemon/src/spawn/runtime_registry.rs create mode 100644 binaries/runtime-api/Cargo.toml rename binaries/{runtime/src/operator => runtime-api/src}/channel.rs (100%) rename binaries/{runtime => runtime-api}/src/lib.rs (94%) rename binaries/{runtime/src/operator/mod.rs => runtime-api/src/operator.rs} (63%) create mode 100644 binaries/runtime-python/Cargo.toml create mode 100644 binaries/runtime-python/build.rs create mode 100644 binaries/runtime-python/src/lib.rs rename binaries/{runtime/src/operator/python.rs => runtime-python/src/runner.rs} (99%) create mode 100644 binaries/runtime-shared-lib/Cargo.toml create mode 100644 binaries/runtime-shared-lib/src/lib.rs rename binaries/{runtime/src/operator/shared_lib.rs => runtime-shared-lib/src/runner.rs} (99%) delete mode 100644 binaries/runtime/Cargo.toml delete mode 100644 binaries/runtime/build.rs diff --git a/.claude/skills/adora-project/SKILL.md b/.claude/skills/adora-project/SKILL.md index 570391a6b4..a7ffb8a385 100644 --- a/.claude/skills/adora-project/SKILL.md +++ b/.claude/skills/adora-project/SKILL.md @@ -181,7 +181,7 @@ Uses `goal_id` and `goal_status` metadata keys. Supports cancellation. cargo build --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python # Test (exclude Python + examples) -cargo test --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python --exclude dora-cli-api-python --exclude dora-examples +cargo test --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python --exclude dora-runtime-python --exclude dora-cli-api-python --exclude dora-examples # Single crate cargo test -p dora-core diff --git a/.github/workflows/cargo-release.yml b/.github/workflows/cargo-release.yml index b2e5a8996f..4d01c1e5f2 100644 --- a/.github/workflows/cargo-release.yml +++ b/.github/workflows/cargo-release.yml @@ -110,7 +110,11 @@ jobs: publish_if_not_exists dora-coordinator publish_if_not_exists dora-daemon publish_if_not_exists dora-operator-api-python - publish_if_not_exists dora-runtime + # The former `dora-runtime`, split per language. `dora-cli` depends on + # `dora-runtime-shared-lib` (→ `-api`), so both must land first. + publish_if_not_exists dora-runtime-api + publish_if_not_exists dora-runtime-shared-lib + publish_if_not_exists dora-runtime-python publish_if_not_exists dora-cli # Publish ROS2 bridge (before the cxx APIs, which optionally diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 69a6d624b3..f78beb2a8e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -177,7 +177,7 @@ jobs: - name: Check examples compile run: cargo check --examples # ...nor does it build test targets, so the `#[cfg(test)]` modules of - # the three PyO3 crates were type-checked nowhere: they're excluded + # the PyO3 crates were type-checked nowhere: they're excluded # from `cargo test` (see the `test` job) and from clippy above, and # `cargo check --all` only covers their lib targets. A syntax error in # one of those test modules could ship. This is rmeta-only and `Check` @@ -189,6 +189,7 @@ jobs: -p dora-node-api-python -p dora-operator-api-python -p dora-ros2-bridge-python + -p dora-runtime-python test: # Linux-only on PR CI (#1716). macOS + Windows coverage runs in nightly. @@ -239,13 +240,16 @@ jobs: # The PyO3 crates stay excluded here — their test binaries link # libpython, and this job sets up no interpreter. Their unit tests run # in `contract-tests`, which does; keep that step in sync if this list - # changes. + # changes. `dora-runtime-python` is one of them: it links pyo3 like the + # rest, and only its own targets are skipped — the lib still builds + # above, as a dependency of dora-cli-api-python. - name: Test run: > cargo test --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python + --exclude dora-runtime-python --exclude dora-cli-api-python --exclude dora-examples diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index c7c1da46d2..d3a786e0d0 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -1207,6 +1207,7 @@ jobs: --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python + --exclude dora-runtime-python --exclude dora-cli-api-python --exclude dora-examples @@ -1847,6 +1848,7 @@ jobs: --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python + --exclude dora-runtime-python --exclude dora-cli-api-python - name: Check (cross) if: matrix.cross @@ -1855,6 +1857,7 @@ jobs: --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python + --exclude dora-runtime-python --exclude dora-cli-api-python # ===== ROS2 bridge basic checks ===== diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 9055c04a78..e89f7dc93f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -86,7 +86,9 @@ jobs: dora-coordinator dora-daemon dora-operator-api-python - dora-runtime + dora-runtime-api + dora-runtime-shared-lib + dora-runtime-python dora-cli dora-ros2-bridge-msg-gen dora-ros2-bridge diff --git a/.pr-bodies/2777.md b/.pr-bodies/2777.md new file mode 100644 index 0000000000..1e7f2aaa8f --- /dev/null +++ b/.pr-bodies/2777.md @@ -0,0 +1,15 @@ +Splits the monolithic `dora-runtime` into a language-neutral SDK plus per-language backend crates, and table-drives the daemon's runtime spawn logic. Behavior-preserving — no descriptor/YAML changes. + +- `dora-runtime-api` — the SDK: operator event loop, node harness, `RuntimeHandle`, and the `OperatorRunner` backend trait + `main(runner)` entry. +- `dora-runtime-shared-lib` — libloading/C-ABI backend, shipped in the `dora` CLI (`dora runtime`). +- `dora-runtime-python` — PyO3 backend, the only crate linking pyo3, shipped in the wheel (`dora.start_runtime()`). + +The `python` cargo feature and all `#[cfg(feature = "python")]` dispatch are gone; adding a language is now one crate implementing `OperatorRunner`. `dora-cli` no longer pulls pyo3 into its tree at all. The daemon's python/shared-library selection moves to `spawn/runtime_registry.rs`, keyed on the new `OperatorSource::runtime_name()` helper, reproducing the existing launch commands verbatim (incl. the #1797/#1805 fixes). The registry is the seam where a follow-up can add third-party runtimes via an explicit `runtimes:` map. + +`dora-runtime-python` also hosts shared-library operators, by delegating to `SharedLibRunner`. That arm is load-bearing: when the daemon is itself an embedded Python process (`current_exe` ends in `python`/`python3`), `native_runtime_command` routes *native* runtime nodes to `python -uc "import dora; dora.start_runtime()"`, and the pre-split `dora-runtime` served them because it compiled the shared-library backend in unconditionally. + +Publishing: both release workflows drop the now-gone `dora-runtime` for `dora-runtime-api` + `dora-runtime-shared-lib` + `dora-runtime-python`, ordered before `dora-cli` (which depends on the shared-lib backend). + +`dora-runtime-python` is excluded from `cargo test --all` alongside `dora-cli-api-python` — it is a plain rlib, so unlike the `extension-module` cdylibs its test binary links libpython, and the CI test job runs without `setup-python`. Its lib still builds there, as a dependency of `dora-cli-api-python`. + +Verified: `cargo test --all` (only the two known container-local `rmw_zenoh_pubsub` failures, which need multicast), `clippy --all -D warnings`, `fmt --check`, `cargo check --examples`, `make qa-fast`, and an end-to-end shared-library operator dataflow (daemon → runtime → dlopen'd operator → sink, all green). diff --git a/AGENTS.md b/AGENTS.md index f1a84b1ce3..f59159f524 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -38,7 +38,9 @@ Important packages: - `binaries/cli`: `dora` CLI - `binaries/daemon`: local process manager and transport bridge - `binaries/coordinator`: distributed orchestration -- `binaries/runtime`: in-process operator runtime +- `binaries/runtime-api`: language-neutral operator runtime SDK (`OperatorRunner` trait + event loop) +- `binaries/runtime-shared-lib`: shared-library (C ABI) operator runtime backend (in the `dora` CLI) +- `binaries/runtime-python`: Python (PyO3) operator runtime backend (in the Python wheel) - `libraries/core`: descriptor parsing and shared build/runtime utilities - `libraries/message`: protocol and message definitions - `apis/rust/node`: Rust node API @@ -74,6 +76,7 @@ cargo test --all \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples @@ -153,6 +156,7 @@ cargo test --all \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples ``` diff --git a/CLAUDE.md b/CLAUDE.md index 00d358b414..d5cdb3bf24 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,7 +20,7 @@ cargo build -p dora-daemon cargo check --all # Test all (excluding Python) -cargo test --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python +cargo test --all --exclude dora-runtime-python --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python # Test single package cargo test -p dora-core @@ -58,7 +58,9 @@ dora run examples/python-dataflow/dataflow.yml --uv --stop-after 10s | `binaries/cli` | dora-cli | CLI binary (`dora` command) - build, run, stop dataflows | | `binaries/daemon` | dora-daemon | Spawns nodes, manages local shared-memory/TCP communication | | `binaries/coordinator` | dora-coordinator | Orchestrates distributed multi-daemon deployments | -| `binaries/runtime` | dora-runtime | In-process operator execution runtime | +| `binaries/runtime-api` | dora-runtime-api | Language-neutral operator runtime SDK (event loop, node harness, `OperatorRunner` trait) | +| `binaries/runtime-shared-lib` | dora-runtime-shared-lib | Shared-library (C ABI) operator runtime backend; shipped in the `dora` CLI (`dora runtime`) | +| `binaries/runtime-python` | dora-runtime-python | Python (PyO3) operator runtime backend; shipped in the Python wheel (`dora.start_runtime()`) | | `libraries/message` | dora-message | All inter-component message types and protocol definitions | | `libraries/core` | dora-core | Dataflow descriptor parsing, build utilities, Zenoh config | | `apis/rust/node` | dora-node-api | Rust API for writing custom nodes | @@ -133,6 +135,7 @@ cargo test --all \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples @@ -142,9 +145,10 @@ cargo test --all \ cargo check --examples # 5. Only if you touched a PyO3 crate (apis/python/node, apis/python/operator, -# libraries/extensions/ros2-bridge/python): their unit tests are excluded from -# the `cargo test --all` above because the test binaries link libpython, so run -# them explicitly. CI runs the same command in ci.yml's `contract-tests` job. +# libraries/extensions/ros2-bridge/python, binaries/runtime-python): their unit +# tests are excluded from the `cargo test --all` above because the test binaries +# link libpython, so run them explicitly. CI runs the same command in ci.yml's +# `contract-tests` job. make qa-test-python # Quick single-crate check while iterating: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1ed4c53a30..120a4272b2 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -17,7 +17,7 @@ Running a command for the whole workspace is possible by passing `--workspace`. cargo build --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python # Test all -cargo test --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python +cargo test --all --exclude dora-runtime-python --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python # Lint cargo clippy --all -- -D warnings diff --git a/Cargo.lock b/Cargo.lock index 57bbeea401..c055854891 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1941,7 +1941,7 @@ dependencies = [ "dora-node-api-c", "dora-operator-api-c", "dora-recording", - "dora-runtime", + "dora-runtime-shared-lib", "dora-tracing", "dunce", "duration-str", @@ -1986,7 +1986,7 @@ dependencies = [ "dora-cli", "dora-download", "dora-node-api", - "dora-runtime", + "dora-runtime-python", "eyre", "pyo3", "pyo3-build-config", @@ -2346,7 +2346,7 @@ dependencies = [ "dora-node-api", "dora-operator-api-python", "dora-ros2-bridge-python", - "dora-runtime", + "dora-runtime-python", "eyre", "flume 0.12.0", "futures", @@ -2563,30 +2563,65 @@ dependencies = [ ] [[package]] -name = "dora-runtime" +name = "dora-runtime-api" version = "1.0.0-rc.4" dependencies = [ - "aligned-vec", "arrow", "dora-core", - "dora-download", "dora-message", "dora-metrics", "dora-node-api", - "dora-operator-api-python", - "dora-operator-api-types", "dora-tracing", "eyre", "flume 0.12.0", "futures", "futures-concurrency", - "libloading 0.9.0", + "serde_yaml", + "tokio", + "tokio-stream", + "tracing", +] + +[[package]] +name = "dora-runtime-python" +version = "1.0.0-rc.4" +dependencies = [ + "arrow", + "dora-core", + "dora-download", + "dora-node-api", + "dora-operator-api-python", + "dora-operator-api-types", + "dora-runtime-api", + "dora-runtime-shared-lib", + "dora-tracing", + "eyre", + "flume 0.12.0", "pyo3", "pyo3-build-config", "pythonize", "serde_yaml", "tokio", - "tokio-stream", + "tracing", + "tracing-opentelemetry", +] + +[[package]] +name = "dora-runtime-shared-lib" +version = "1.0.0-rc.4" +dependencies = [ + "arrow", + "dora-core", + "dora-download", + "dora-node-api", + "dora-operator-api-types", + "dora-runtime-api", + "dora-tracing", + "eyre", + "flume 0.12.0", + "libloading 0.9.0", + "serde_yaml", + "tokio", "tracing", "tracing-opentelemetry", ] diff --git a/Cargo.toml b/Cargo.toml index 48a6a1b004..0565c61c64 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,9 @@ members = [ "binaries/cli", "binaries/coordinator", "binaries/daemon", - "binaries/runtime", + "binaries/runtime-api", + "binaries/runtime-shared-lib", + "binaries/runtime-python", "binaries/mavlink2-bridge-node", "binaries/ros2-bridge-node", "examples/rust-dataflow/node", @@ -123,7 +125,9 @@ dora-download = { version = "1.0.0-rc.4", path = "libraries/extensions/download" dora-log-utils = { version = "1.0.0-rc.4", path = "libraries/log-utils" } dora-coordinator-store = { version = "1.0.0-rc.4", path = "libraries/coordinator-store" } dora-cli = { version = "1.0.0-rc.4", path = "binaries/cli" } -dora-runtime = { version = "1.0.0-rc.4", path = "binaries/runtime" } +dora-runtime-api = { version = "1.0.0-rc.4", path = "binaries/runtime-api" } +dora-runtime-shared-lib = { version = "1.0.0-rc.4", path = "binaries/runtime-shared-lib" } +dora-runtime-python = { version = "1.0.0-rc.4", path = "binaries/runtime-python" } dora-daemon = { version = "1.0.0-rc.4", path = "binaries/daemon" } dora-coordinator = { version = "1.0.0-rc.4", path = "binaries/coordinator" } dora-ros2-bridge = { version = "1.0.0-rc.4", path = "libraries/extensions/ros2-bridge" } diff --git a/Makefile b/Makefile index 2101767981..37135d6404 100644 --- a/Makefile +++ b/Makefile @@ -145,21 +145,29 @@ qa-test: --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples -# The unit tests of the three PyO3 crates `qa-test` excludes. Kept a separate +# The unit tests of the PyO3 crates `qa-test` excludes. Kept a separate # target, not folded into `qa-test`: cargo builds these crates without # `pyo3/extension-module` (unlike the maturin wheel), so the test binaries # link libpython directly and need an interpreter >= 3.11 with a shared # library — a machine set up only for Rust work would start failing the # everyday gate. CI runs this same target in ci.yml's `contract-tests` job, # which sets Python up explicitly. +# +# `dora-runtime-python` is the operator runtime's Python backend. Its tests +# cover the cross-language arms of the runtime split — above all that a +# shared-library operator reaching the Python runtime is *delegated* to the +# shared-lib backend rather than rejected, which is what keeps native +# operators working under an embedded-Python daemon. qa-test-python: @cargo test --lib \ -p dora-node-api-python \ -p dora-operator-api-python \ - -p dora-ros2-bridge-python + -p dora-ros2-bridge-python \ + -p dora-runtime-python qa-coverage: @scripts/qa/coverage.sh diff --git a/README.md b/README.md index e542cdd91b..802fe4db45 100644 --- a/README.md +++ b/README.md @@ -592,6 +592,7 @@ cargo build -p dora-cli ```bash # Run all tests cargo test --all \ + --exclude dora-runtime-python \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python diff --git a/README.zh-CN.md b/README.zh-CN.md index ae1bfc1686..3778a97f07 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -479,6 +479,7 @@ cargo build -p dora-cli ```bash # 运行所有测试 cargo test --all \ + --exclude dora-runtime-python \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python diff --git a/apis/python/cli/Cargo.toml b/apis/python/cli/Cargo.toml index 78bce2f09d..8bcf963c47 100644 --- a/apis/python/cli/Cargo.toml +++ b/apis/python/cli/Cargo.toml @@ -12,11 +12,11 @@ publish = false # PyO3 cdylib, shipped via PyPI as `dora-rs-cli`, not crates.io [features] default = ["telemetry"] -telemetry = ["dora-runtime/telemetry"] +telemetry = ["dora-runtime-python/telemetry"] [dependencies] dora-cli = { workspace = true, features = ["python"] } -dora-runtime = { workspace = true, features = ["tracing", "metrics", "python"] } +dora-runtime-python = { workspace = true, features = ["tracing", "metrics"] } dora-download = { workspace = true } dora-node-api = { workspace = true } eyre = { workspace = true } diff --git a/apis/python/cli/src/lib.rs b/apis/python/cli/src/lib.rs index b9daa69443..4819d7a879 100644 --- a/apis/python/cli/src/lib.rs +++ b/apis/python/cli/src/lib.rs @@ -6,7 +6,7 @@ use pyo3::prelude::*; /// :rtype: None #[pyfunction] pub fn start_runtime() -> eyre::Result<()> { - dora_runtime::main().wrap_err("Dora Runtime raised an error.") + dora_runtime_python::main().wrap_err("Dora Runtime raised an error.") } /// Build a Dataflow, exactly the same way as `dora build` command line tool. diff --git a/apis/python/node/Cargo.toml b/apis/python/node/Cargo.toml index fabfb1b918..2de570ae56 100644 --- a/apis/python/node/Cargo.toml +++ b/apis/python/node/Cargo.toml @@ -22,7 +22,7 @@ async = ["pyo3/experimental-async"] dora-node-api = { workspace = true } dora-message = { workspace = true } dora-operator-api-python = { workspace = true } -dora-runtime = { workspace = true, features = ["python"] } +dora-runtime-python = { workspace = true } dora-cli = { workspace = true, features = ["python"] } chrono = { version = "0.4", features = ["serde"] } pyo3.workspace = true diff --git a/apis/python/node/src/lib.rs b/apis/python/node/src/lib.rs index aabbe2c37c..97b685be23 100644 --- a/apis/python/node/src/lib.rs +++ b/apis/python/node/src/lib.rs @@ -3884,7 +3884,7 @@ impl Node { /// :rtype: None #[pyfunction] pub fn start_runtime() -> eyre::Result<()> { - dora_runtime::main().wrap_err("Dora Runtime raised an error.") + dora_runtime_python::main().wrap_err("Dora Runtime raised an error.") } /// Build a Dataflow, exactly the same way as `dora build` command line tool. diff --git a/binaries/cli/Cargo.toml b/binaries/cli/Cargo.toml index 1a07a1277c..de176119ef 100644 --- a/binaries/cli/Cargo.toml +++ b/binaries/cli/Cargo.toml @@ -18,7 +18,7 @@ path = "src/main.rs" [features] default = ["tracing", "metrics", "redb-backend"] tracing = ["dep:dora-tracing"] -metrics = ["dora-runtime/metrics", "dora-coordinator/metrics"] +metrics = ["dora-runtime-shared-lib/metrics", "dora-coordinator/metrics"] python = ["pyo3"] # Enable extension-module only when building the Python wheel via maturin. # This suppresses Python library linking, which is correct for cdylib targets @@ -57,7 +57,7 @@ tracing-log = "0.2.0" dora-tracing = { workspace = true, optional = true } dora-daemon = { workspace = true } dora-coordinator = { workspace = true } -dora-runtime = { workspace = true } +dora-runtime-shared-lib = { workspace = true } tokio = { workspace = true, features = ["full"] } tokio-stream = { version = "0.1.18", features = ["io-util", "net"] } futures = { workspace = true } diff --git a/binaries/cli/src/command/runtime.rs b/binaries/cli/src/command/runtime.rs index 430e389152..24506bb43a 100644 --- a/binaries/cli/src/command/runtime.rs +++ b/binaries/cli/src/command/runtime.rs @@ -10,6 +10,6 @@ impl Executable for Runtime { fn execute(self) -> eyre::Result<()> { // No tracing: Do not set the runtime in the cli. // ref: 72b4be808122574fcfda69650954318e0355cc7b cli::run - dora_runtime::main().context("Failed to run dora-runtime") + dora_runtime_shared_lib::main().context("Failed to run dora-runtime") } } diff --git a/binaries/cli/src/session.rs b/binaries/cli/src/session.rs index 8d2e38df1c..d464c5e2c1 100644 --- a/binaries/cli/src/session.rs +++ b/binaries/cli/src/session.rs @@ -7,7 +7,7 @@ use dora_core::build::BuildInfo; use dora_message::{ BuildId, SessionId, common::GitSource, - descriptor::{CoreNodeKind, NodeSource, OperatorSource, ResolvedNode}, + descriptor::{CoreNodeKind, NodeSource, ResolvedNode}, id::NodeId, }; use eyre::{Context, ContextCompat}; @@ -272,11 +272,7 @@ impl DataflowSession { // // Kind tag still flips when switching kinds (Python -> // SharedLibrary etc.), which IS a build-system change. - let kind_tag = match &op.config.source { - OperatorSource::SharedLibrary(_) => "shared-library", - OperatorSource::Python(_) => "python", - OperatorSource::Wasm(_) => "wasm", - }; + let kind_tag = op.config.source.runtime_name(); canonical.push_str(" source-kind:"); canonical.push_str(kind_tag); canonical.push('\n'); diff --git a/binaries/daemon/src/spawn/mod.rs b/binaries/daemon/src/spawn/mod.rs index c6893084b7..be09f73edb 100644 --- a/binaries/daemon/src/spawn/mod.rs +++ b/binaries/daemon/src/spawn/mod.rs @@ -3,4 +3,5 @@ pub use spawner::{NodeZenohPeering, Spawner, plan_zenoh_peering}; mod command; mod prepared; +mod runtime_registry; mod spawner; diff --git a/binaries/daemon/src/spawn/runtime_registry.rs b/binaries/daemon/src/spawn/runtime_registry.rs new file mode 100644 index 0000000000..6ddfeb2719 --- /dev/null +++ b/binaries/daemon/src/spawn/runtime_registry.rs @@ -0,0 +1,192 @@ +//! Runtime registry: resolves the base spawn command for a runtime node's +//! operators. +//! +//! Today this maps the two built-in runtimes — `python` and `shared-library` — +//! to their launch commands, reproducing the historical daemon behavior. It is +//! the single seam where a future third-party-runtime resolver (driven by a +//! descriptor `runtimes:` map) will slot in: the caller passes the operators, +//! and this module decides which launcher hosts them. + +use clonable_command::Command; +use dora_core::{ + build::managed_python_interpreter, + config::NodeId, + descriptor::{OperatorDefinition, OperatorSource, PythonSource, RUNTIME_PYTHON}, + get_python_path, +}; +use eyre::{ContextCompat, WrapErr, bail}; +use std::path::Path; + +/// Build the base command that hosts a runtime node's `operators`. +/// +/// All operators in a runtime node must belong to the same runtime family +/// (Python vs. native shared-library/WASM); mixing is rejected. The returned +/// command has only the launcher + args set — the caller injects +/// `DORA_RUNTIME_CONFIG`, zenoh connect vars, per-node env, and stdio. +pub(super) fn runtime_command( + node_id: &NodeId, + operators: &[OperatorDefinition], + uv: bool, + python_env_dir: Option<&Path>, +) -> eyre::Result { + let python_operators: Vec<&OperatorDefinition> = operators + .iter() + .filter(|x| x.config.source.runtime_name() == RUNTIME_PYTHON) + .collect(); + + let other_operators = operators + .iter() + .any(|x| x.config.source.runtime_name() != RUNTIME_PYTHON); + + if !python_operators.is_empty() && !other_operators { + python_runtime_command(node_id, &python_operators, uv, python_env_dir) + } else if python_operators.is_empty() && other_operators { + native_runtime_command(node_id) + } else { + bail!( + "Cannot spawn runtime with both Python and non-Python operators. \ + Please use a single operator or ensure that all operators are Python-based." + ) + } +} + +/// The `python` built-in runtime: launch the runtime inside a Python +/// interpreter via `import dora; dora.start_runtime()`. +fn python_runtime_command( + node_id: &NodeId, + python_operators: &[&OperatorDefinition], + uv: bool, + python_env_dir: Option<&Path>, +) -> eyre::Result { + // Use python to spawn runtime if there is a python operator + + // TODO: Handle multi-operator runtime once sub-interpreter is supported + if python_operators.len() > 1 { + bail!( + "Runtime currently only supports one Python Operator. + This is because PyO3 sub-interpreter is not yet available. + See: https://github.com/PyO3/pyo3/issues/576" + ); + } + + let python_operator = python_operators + .first() + .context("Runtime had no operators definition.")?; + + if let OperatorSource::Python(PythonSource { + source: _, + conda_env: Some(conda_env), + }) = &python_operator.config.source + { + let conda = which::which("conda").context( + "failed to find `conda`, yet a `conda_env` was defined. Make sure that `conda` is available.", + )?; + let mut command = Command::new(conda); + command = command.args([ + "run", + "-n", + conda_env, + "python", + "-uc", + format!("import dora; dora.start_runtime() # {}", node_id).as_str(), + ]); + Ok(command) + } else { + let mut cmd = if uv { + if let Some(python_env_dir) = python_env_dir { + // Reuse the managed interpreter so Python operators run + // against the same environment Dora prepared during build. + let python = managed_python_interpreter(python_env_dir); + if !python.is_file() { + bail!( + "managed Python interpreter `{}` is missing", + python.display() + ); + } + tracing::info!( + "spawning managed Python {} -uc import dora; dora.start_runtime() # {}", + python.display(), + node_id + ); + Command::new(python) + } else { + let mut cmd = Command::new("uv"); + cmd = cmd.arg("run"); + cmd = cmd.arg("python"); + tracing::info!( + "spawning: uv run python -uc import dora; dora.start_runtime() # {}", + node_id + ); + cmd + } + } else { + let python = get_python_path() + .wrap_err("Could not find python path when spawning custom node")?; + tracing::info!( + "spawning: python -uc import dora; dora.start_runtime() # {}", + node_id + ); + + Command::new(python) + }; + // Force python to always flush stdout/stderr buffer + cmd = cmd.args([ + "-uc", + format!("import dora; dora.start_runtime() # {}", node_id).as_str(), + ]); + Ok(cmd) + } +} + +/// The `shared-library` built-in runtime: launch the native `dora runtime` +/// subcommand (which hosts shared-library operators via `libloading`). +fn native_runtime_command(node_id: &NodeId) -> eyre::Result { + let current_exe = std::env::current_exe().wrap_err("failed to get current executable path")?; + let mut file_name = current_exe.clone(); + file_name.set_extension(""); + let file_name = file_name + .file_name() + .and_then(|s| s.to_str()) + .context("failed to get file name from current executable")?; + + // Check if the current executable is a python binary meaning that dora is installed within the python environment + if file_name.ends_with("python") || file_name.ends_with("python3") { + // Use the current executable to spawn runtime. That lands on + // `dora-runtime-python`, which hosts shared-library operators by + // delegating to `dora-runtime-shared-lib` — see that crate's module + // docs. Keep the two in step: dropping the delegation breaks native + // operators under an embedded-Python daemon. + let python = + get_python_path().wrap_err("Could not find python path when spawning custom node")?; + let mut cmd = Command::new(python); + + tracing::info!( + "spawning: python -uc import dora; dora.start_runtime() # {}", + node_id + ); + + cmd = cmd.args([ + "-uc", + format!("import dora; dora.start_runtime() # {}", node_id).as_str(), + ]); + Ok(cmd) + } else if file_name == "dora" { + // current_exe is the dora binary — use it so the + // spawned runtime always matches the daemon version. + // See #1797. + let mut cmd = Command::new(¤t_exe); + cmd = cmd.arg("runtime"); + Ok(cmd) + } else { + // current_exe is something else, e.g. an embedded + // example runner that calls `dora_cli::run()` — + // see examples/c-dataflow/run.rs:21. Spawning + // current_exe with `runtime` would recurse into + // the example runner. Fall back to PATH lookup + // for the dora binary. See #1805. + let mut cmd = + Command::new(which::which("dora").wrap_err("failed to find dora binary on PATH")?); + cmd = cmd.arg("runtime"); + Ok(cmd) + } +} diff --git a/binaries/daemon/src/spawn/spawner.rs b/binaries/daemon/src/spawn/spawner.rs index 646efad896..7e2273ccd2 100644 --- a/binaries/daemon/src/spawn/spawner.rs +++ b/binaries/daemon/src/spawn/spawner.rs @@ -7,12 +7,9 @@ use crate::{ use clonable_command::{Command, Stdio}; use crossbeam::queue::ArrayQueue; use dora_core::{ - build::{managed_python_bin_dir, managed_python_interpreter}, + build::managed_python_bin_dir, config::{Input, InputMapping, NodeId}, - descriptor::{ - CoreNodeKind, Descriptor, OperatorDefinition, OperatorSource, PythonSource, ResolvedNode, - }, - get_python_path, + descriptor::{CoreNodeKind, Descriptor, ResolvedNode}, topics::{ DORA_RUN_PARENT_PID_ENV, DORA_ZENOH_CONNECT_ENV, DORA_ZENOH_LISTEN_ENV, DORA_ZENOH_MULTICAST_ENV, ZENOH_CONFIG_PATH_ENV, @@ -27,7 +24,7 @@ use dora_message::{ descriptor::EnvValue, id::DataId, }; -use eyre::{ContextCompat, WrapErr, bail}; +use eyre::WrapErr; use std::{ collections::{BTreeMap, BTreeSet}, ffi::OsString, @@ -609,189 +606,48 @@ impl Spawner { (command, error_msg) } dora_core::descriptor::CoreNodeKind::Runtime(n) => { - let python_operators: Vec<&OperatorDefinition> = n - .operators - .iter() - .filter(|x| matches!(x.config.source, OperatorSource::Python { .. })) - .collect(); - - let other_operators = n - .operators - .iter() - .any(|x| !matches!(x.config.source, OperatorSource::Python { .. })); - - let command = if !python_operators.is_empty() && !other_operators { - // Use python to spawn runtime if there is a python operator - - // TODO: Handle multi-operator runtime once sub-interpreter is supported - if python_operators.len() > 1 { - eyre::bail!( - "Runtime currently only supports one Python Operator. - This is because PyO3 sub-interpreter is not yet available. - See: https://github.com/PyO3/pyo3/issues/576" - ); - } - - let python_operator = python_operators - .first() - .context("Runtime had no operators definition.")?; - - if let OperatorSource::Python(PythonSource { - source: _, - conda_env: Some(conda_env), - }) = &python_operator.config.source - { - let conda = which::which("conda").context( - "failed to find `conda`, yet a `conda_env` was defined. Make sure that `conda` is available.", - )?; - let mut command = Command::new(conda); - command = command.args([ - "run", - "-n", - conda_env, - "python", - "-uc", - format!("import dora; dora.start_runtime() # {}", node.id).as_str(), - ]); - Some(command) - } else { - let mut cmd = if self.uv { - if let Some(python_env_dir) = python_env_dir.as_deref() { - // Reuse the managed interpreter so Python operators run - // against the same environment Dora prepared during build. - let python = managed_python_interpreter(python_env_dir); - if !python.is_file() { - eyre::bail!( - "managed Python interpreter `{}` is missing", - python.display() - ); - } - tracing::info!( - "spawning managed Python {} -uc import dora; dora.start_runtime() # {}", - python.display(), - node.id - ); - Command::new(python) - } else { - let mut cmd = Command::new("uv"); - cmd = cmd.arg("run"); - cmd = cmd.arg("python"); - tracing::info!( - "spawning: uv run python -uc import dora; dora.start_runtime() # {}", - node.id - ); - cmd - } - } else { - let python = get_python_path() - .wrap_err("Could not find python path when spawning custom node")?; - tracing::info!( - "spawning: python -uc import dora; dora.start_runtime() # {}", - node.id - ); - - Command::new(python) - }; - // Force python to always flush stdout/stderr buffer - cmd = cmd.args([ - "-uc", - format!("import dora; dora.start_runtime() # {}", node.id).as_str(), - ]); - Some(cmd) - } - } else if python_operators.is_empty() && other_operators { - let current_exe = std::env::current_exe() - .wrap_err("failed to get current executable path")?; - let mut file_name = current_exe.clone(); - file_name.set_extension(""); - let file_name = file_name - .file_name() - .and_then(|s| s.to_str()) - .context("failed to get file name from current executable")?; - - // Check if the current executable is a python binary meaning that dora is installed within the python environment - if file_name.ends_with("python") || file_name.ends_with("python3") { - // Use the current executable to spawn runtime - let python = get_python_path() - .wrap_err("Could not find python path when spawning custom node")?; - let mut cmd = Command::new(python); - - tracing::info!( - "spawning: python -uc import dora; dora.start_runtime() # {}", - node.id - ); - - cmd = cmd.args([ - "-uc", - format!("import dora; dora.start_runtime() # {}", node.id).as_str(), - ]); - Some(cmd) - } else if file_name == "dora" { - // current_exe is the dora binary — use it so the - // spawned runtime always matches the daemon version. - // See #1797. - let mut cmd = Command::new(¤t_exe); - cmd = cmd.arg("runtime"); - Some(cmd) - } else { - // current_exe is something else, e.g. an embedded - // example runner that calls `dora_cli::run()` — - // see examples/c-dataflow/run.rs:21. Spawning - // current_exe with `runtime` would recurse into - // the example runner. Fall back to PATH lookup - // for the dora binary. See #1805. - let mut cmd = Command::new( - which::which("dora").wrap_err("failed to find dora binary on PATH")?, - ); - cmd = cmd.arg("runtime"); - Some(cmd) - } - } else { - bail!( - "Cannot spawn runtime with both Python and non-Python operators. \ - Please use a single operator or ensure that all operators are Python-based." - ); - }; + let mut command = super::runtime_registry::runtime_command( + &node.id, + &n.operators, + self.uv, + python_env_dir.as_deref(), + )?; let runtime_config = RuntimeConfig { node: node_config.clone(), operators: n.operators.clone(), }; - let command = if let Some(mut command) = command { - command = command.current_dir(&node_working_dir); - command = self.compose_node_env( - command, - &node.id, - &[node.env.as_ref()], - "DORA_RUNTIME_CONFIG", - serde_yaml::to_string(&runtime_config) - .wrap_err("failed to serialize runtime config")?, - ); + command = command.current_dir(&node_working_dir); + command = self.compose_node_env( + command, + &node.id, + &[node.env.as_ref()], + "DORA_RUNTIME_CONFIG", + serde_yaml::to_string(&runtime_config) + .wrap_err("failed to serialize runtime config")?, + ); - // For managed Python runtime nodes (Python operator + uv on), - // set VIRTUAL_ENV and prepend the env's bin dir to PATH so - // anything the operator spawns sees the managed env. - if self.uv - && let Some(env_dir) = python_env_dir.as_deref() - { - command = - apply_managed_python_runtime_env(command, env_dir, node.env.as_ref())?; - } + // For managed Python runtime nodes (Python operator + uv on), + // set VIRTUAL_ENV and prepend the env's bin dir to PATH so + // anything the operator spawns sees the managed env. + if self.uv + && let Some(env_dir) = python_env_dir.as_deref() + { + command = + apply_managed_python_runtime_env(command, env_dir, node.env.as_ref())?; + } + + command = command + .stdin(Stdio::Null) + .stdout(Stdio::Piped) + .stderr(Stdio::Piped); - command = command - .stdin(Stdio::Null) - .stdout(Stdio::Piped) - .stderr(Stdio::Piped); - Some(command) - } else { - command - }; let error_msg = format!( "failed to run runtime {}/{}", runtime_config.node.dataflow_id, runtime_config.node.node_id ); - (command, error_msg) + (Some(command), error_msg) } }; Ok(PreparedNode { diff --git a/binaries/runtime-api/Cargo.toml b/binaries/runtime-api/Cargo.toml new file mode 100644 index 0000000000..7a4a3b6550 --- /dev/null +++ b/binaries/runtime-api/Cargo.toml @@ -0,0 +1,37 @@ +[package] +name = "dora-runtime-api" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +documentation.workspace = true +readme.workspace = true +description.workspace = true +license.workspace = true +repository.workspace = true + +# Language-neutral runtime SDK: the operator event loop, the daemon/node +# harness, and the `OperatorRunner` backend trait. Per-language backends +# (`dora-runtime-shared-lib`, `dora-runtime-python`) build on top of this. + +[dependencies] +dora-node-api = { workspace = true, default-features = false } +dora-core = { workspace = true } +dora-message = { workspace = true } +dora-tracing = { workspace = true, optional = true } +dora-metrics = { workspace = true, optional = true } +eyre = { workspace = true } +futures = { workspace = true } +futures-concurrency = "7.7.1" +serde_yaml = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tokio-stream = "0.1.18" +tracing = { workspace = true } +flume = { workspace = true } + +[dev-dependencies] +arrow = { workspace = true } + +[features] +default = ["tracing", "metrics"] +tracing = ["dep:dora-tracing"] +metrics = ["dep:dora-metrics"] diff --git a/binaries/runtime/src/operator/channel.rs b/binaries/runtime-api/src/channel.rs similarity index 100% rename from binaries/runtime/src/operator/channel.rs rename to binaries/runtime-api/src/channel.rs diff --git a/binaries/runtime/src/lib.rs b/binaries/runtime-api/src/lib.rs similarity index 94% rename from binaries/runtime/src/lib.rs rename to binaries/runtime-api/src/lib.rs index 377ed42e11..34e69c9ba0 100644 --- a/binaries/runtime/src/lib.rs +++ b/binaries/runtime-api/src/lib.rs @@ -10,8 +10,6 @@ use dora_tracing::TracingBuilder; use eyre::{Context, Result, bail}; use futures::{Stream, StreamExt}; use futures_concurrency::stream::Merge; -use operator::{OperatorEvent, RuntimeHandle, SharedAllocator, StopReason, run_operator}; - use std::{ collections::{BTreeMap, BTreeSet, HashMap, VecDeque}, mem, @@ -23,9 +21,21 @@ use tokio::{ sync::{mpsc, oneshot}, }; use tokio_stream::wrappers::ReceiverStream; + +mod channel; mod operator; -pub fn main() -> eyre::Result<()> { +pub use operator::{ + OperatorEvent, OperatorRunner, RunnerGuard, RuntimeHandle, SharedAllocator, StopReason, +}; + +/// Entry point for a runtime process. +/// +/// Reads the `DORA_RUNTIME_CONFIG` env var, builds the tokio runtime, runs the +/// language-neutral event loop on a spawned thread, and calls +/// [`OperatorRunner::run_operator`] on the main thread. Each per-language +/// backend calls this with its own runner. +pub fn main(runner: impl OperatorRunner) -> eyre::Result<()> { let config: RuntimeConfig = { let raw = std::env::var("DORA_RUNTIME_CONFIG") .wrap_err("env variable DORA_RUNTIME_CONFIG must be set")?; @@ -74,8 +84,7 @@ pub fn main() -> eyre::Result<()> { let mut operator_channels = HashMap::new(); let queue_sizes = queue_sizes(&operator_definition.config); - let (operator_channel, incoming_events) = - operator::channel::channel(tokio_runtime.handle(), queue_sizes); + let (operator_channel, incoming_events) = channel::channel(tokio_runtime.handle(), queue_sizes); operator_channels.insert(operator_definition.id.clone(), operator_channel); tracing::info!("spawning main task"); @@ -102,29 +111,31 @@ pub fn main() -> eyre::Result<()> { }); let operator_id = operator_definition.id.clone(); - // Keep the operator's shared library mapped until *after* the main event - // loop has joined below. Since dora-rs/dora#2742 the loop no longer holds - // Arrow arrays exported by the operator, but values whose vtable lives in - // the `.so` can still be in flight (an `OperatorEvent::Panic` payload). - // Unloading it earlier dangles those (see `run_operator` / - // `shared_lib::run`). - let _operator_library = run_operator( - &node_id, - operator_definition, - incoming_events, - RuntimeHandle::new(operator_events_tx, allocator), - init_done_tx, - &dataflow_descriptor, - ) - .wrap_err_with(|| format!("failed to run operator {operator_id}"))?; + // Hold the backend's guard until *after* the main event loop has joined + // below. The shared-library backend returns the loaded `.so` here. Since + // dora-rs/dora#2742 the loop no longer holds Arrow arrays exported by the + // operator, but values whose vtable lives in the `.so` can still be in + // flight (an `OperatorEvent::Panic` payload). Unloading it earlier dangles + // those. See [`RunnerGuard`]. + let _operator_guard = runner + .run_operator( + &node_id, + operator_definition, + incoming_events, + RuntimeHandle::new(operator_events_tx, allocator), + init_done_tx, + &dataflow_descriptor, + ) + .wrap_err_with(|| format!("failed to run operator {operator_id}"))?; match main_task.join() { Ok(result) => result.wrap_err("main task failed")?, Err(panic) => std::panic::resume_unwind(panic), } - // `_operator_library` drops (unloads the `.so`) at end of scope here, after - // the main loop has joined and released everything the operator handed it. + // `_operator_guard` drops here (unloading the `.so` for the shared-library + // backend), after the main loop has joined and released everything the + // operator handed it. Ok(()) } diff --git a/binaries/runtime/src/operator/mod.rs b/binaries/runtime-api/src/operator.rs similarity index 63% rename from binaries/runtime/src/operator/mod.rs rename to binaries/runtime-api/src/operator.rs index 349fb28c43..4c9b6b6841 100644 --- a/binaries/runtime/src/operator/mod.rs +++ b/binaries/runtime-api/src/operator.rs @@ -1,11 +1,11 @@ use dora_core::{ config::{DataId, NodeId}, - descriptor::{Descriptor, OperatorDefinition, OperatorSource}, + descriptor::{Descriptor, OperatorDefinition}, }; use dora_node_api::{ EncodedSample, Event, MetadataParameters, SampleAllocator, arrow::array::ArrayData, }; -use eyre::{Context, Result}; +use eyre::Result; use std::any::Any; use std::sync::{Arc, OnceLock}; use tokio::sync::{mpsc::Sender, oneshot}; @@ -68,86 +68,51 @@ impl RuntimeHandle { } } -pub mod channel; -#[cfg(feature = "python")] -mod python; -mod shared_lib; +/// A language/ABI-specific operator backend. +/// +/// Each runtime backend (shared-library, Python, WASM, third-party, …) +/// implements this trait and hands it to [`crate::main`], which drives the +/// language-neutral event loop. The implementation is invoked once, on the +/// **main thread** (PyO3 and `libloading` both want a dedicated thread), and is +/// responsible for loading the operator described by `operator` and running it +/// until it stops. +/// +/// The runtime↔operator contract is language-neutral: consume +/// [`dora_node_api::Event`]s off `incoming_events`, emit outputs and lifecycle +/// events through `handle`, and signal readiness (or an init failure) exactly +/// once on `init_done`. +/// +/// A backend that cannot host `operator`'s source kind must return an `Err` +/// **without** signalling `init_done`, so the failure surfaces as a spawn error +/// rather than a runtime hang (dora-rs/dora#2595). +pub trait OperatorRunner { + fn run_operator( + &self, + node_id: &NodeId, + operator: OperatorDefinition, + incoming_events: flume::Receiver, + handle: RuntimeHandle, + init_done: oneshot::Sender>, + dataflow_descriptor: &Descriptor, + ) -> eyre::Result; +} -/// Runs the operator to completion. Returns a shared library that the caller -/// **must keep alive until after the runtime's main event loop has joined**. +/// A backend-owned resource that must stay alive until the runtime's event loop +/// has joined. /// +/// The shared-library backend returns its loaded `libloading::Library` here. /// Since dora-rs/dora#2742 the main loop no longer holds Arrow arrays exported /// by the operator — outputs are encoded into dora-owned samples on the operator /// thread (see [`RuntimeHandle::send_output`]) — but other values can still /// carry `.so`-resident vtables across the channel, most notably an /// [`OperatorEvent::Panic`] payload. Unloading the library while the loop may -/// still hold one dangles those, so the caller keeps it mapped. See -/// `shared_lib::run`. Returns `None` for operator kinds with no such library -/// (Python). -#[allow(unused_variables)] -pub fn run_operator( - node_id: &NodeId, - operator_definition: OperatorDefinition, - incoming_events: flume::Receiver, - handle: RuntimeHandle, - init_done: oneshot::Sender>, - dataflow_descriptor: &Descriptor, -) -> eyre::Result> { - let library = match &operator_definition.config.source { - OperatorSource::SharedLibrary(source) => { - let library = shared_lib::run( - node_id, - &operator_definition.id, - source, - handle, - incoming_events, - init_done, - ) - .wrap_err_with(|| { - format!( - "failed to spawn shared library operator for {}", - operator_definition.id - ) - })?; - Some(library) - } - #[allow(unused_variables)] - OperatorSource::Python(source) => { - #[cfg(feature = "python")] - { - python::run( - node_id, - &operator_definition.id, - source, - handle, - incoming_events, - init_done, - dataflow_descriptor, - ) - .wrap_err_with(|| { - format!( - "failed to spawn Python operator for {}", - operator_definition.id - ) - })?; - None - } - #[cfg(not(feature = "python"))] - eyre::bail!( - "operator `{}` uses a Python source, but this dora-runtime was \ - built without the `python` feature", - operator_definition.id - ); - } - OperatorSource::Wasm(_) => { - eyre::bail!( - "operator `{}` uses a WASM source, which is not supported yet", - operator_definition.id - ); - } - }; - Ok(library) -} +/// still hold one dangles those, so [`crate::main`] binds the guard for the +/// whole run and drops it last. +/// +/// It is deliberately opaque (`Box`) so `dora-runtime-api` stays +/// language-neutral — it never needs to name `libloading` or any other +/// backend-specific type. Backends with nothing to keep alive return `None`. +pub type RunnerGuard = Option>; #[derive(Debug)] #[allow(clippy::large_enum_variant)] @@ -273,39 +238,4 @@ mod tests { "unexpected error: {err}" ); } - - /// An unsupported operator source must surface a descriptive error from - /// `run_operator` rather than returning `Ok(())` while silently dropping - /// the `init_done` sender — which would leave the runtime task blocked in - /// `init_done.await` until it fails with the misleading "the `init_done` - /// channel was closed unexpectedly". - #[test] - fn wasm_source_returns_descriptive_error() { - let operator_definition: OperatorDefinition = - serde_yaml::from_str("id: op\nwasm: model.wasm\n").expect("operator definition parses"); - let dataflow: Descriptor = - serde_yaml::from_str("nodes:\n - id: a\n").expect("descriptor parses"); - let (_events_in_tx, incoming_events) = flume::unbounded::(); - let (events_tx, _events_rx) = tokio::sync::mpsc::channel(1); - let (init_done_tx, mut init_done_rx) = oneshot::channel(); - - let err = run_operator( - &NodeId::from("node".to_string()), - operator_definition, - incoming_events, - RuntimeHandle::new(events_tx, SharedAllocator::default()), - init_done_tx, - &dataflow, - ) - .expect_err("WASM operator source must return an error"); - assert!( - err.to_string().contains("WASM"), - "expected a descriptive WASM error, got: {err}" - ); - // The init_done sender must not have signalled readiness. - assert!( - init_done_rx.try_recv().is_err(), - "init_done must not receive a value for an unsupported source" - ); - } } diff --git a/binaries/runtime-python/Cargo.toml b/binaries/runtime-python/Cargo.toml new file mode 100644 index 0000000000..4ec171238b --- /dev/null +++ b/binaries/runtime-python/Cargo.toml @@ -0,0 +1,63 @@ +[package] +name = "dora-runtime-python" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +documentation.workspace = true +readme.workspace = true +description.workspace = true +license.workspace = true +repository.workspace = true + +# Python (PyO3) operator runtime backend. Loads a Python module's `Operator` +# class and runs it on the `dora-runtime-api` event loop. Shipped inside the +# Python wheel and invoked via `dora.start_runtime()`. This is the only runtime +# crate that links `pyo3`. + +[dependencies] +dora-runtime-api = { workspace = true } +# The wheel's runtime hosts shared-library operators too: an embedded-Python +# daemon routes native runtime nodes here (see this crate's module docs), which +# the pre-split `dora-runtime` handled because it compiled the shared-library +# backend in unconditionally. +dora-runtime-shared-lib = { workspace = true } +dora-node-api = { workspace = true, default-features = false } +dora-core = { workspace = true } +dora-download = { workspace = true } +dora-operator-api-python = { workspace = true } +dora-operator-api-types = { workspace = true } +dora-tracing = { workspace = true, optional = true } +arrow = { workspace = true, features = ["pyarrow"] } +# pyo3-abi3 flag allow simpler linking. See: https://pyo3.rs/v0.13.2/building_and_distribution.html +pyo3 = { workspace = true, features = ["eyre", "abi3-py311"] } +pythonize = { workspace = true } +eyre = { workspace = true } +flume = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +tracing-opentelemetry = { version = "0.33.0", optional = true } + +[build-dependencies] +# pyo3-build-config's own build script (resolve-config feature) probes the +# Python interpreter and fails when it's older than the workspace abi3-py311 +# floor. In the old monolithic `dora-runtime` that had to be gated behind a +# `python` feature, because the crate was linked into `dora-cli` on machines +# with Python <= 3.10. This crate *is* the Python backend and always links +# pyo3, so the dependency is unconditional here. +pyo3-build-config = { workspace = true } + +[dev-dependencies] +serde_yaml = { workspace = true } + +[features] +default = ["tracing", "metrics"] +tracing = ["dora-runtime-api/tracing"] +metrics = ["dora-runtime-api/metrics"] +# Also turned on for the delegated shared-library backend, so native operators +# hosted by an embedded-Python daemon still export OTLP spans. +telemetry = [ + "tracing", + "dora-runtime-shared-lib/telemetry", + "dep:dora-tracing", + "dep:tracing-opentelemetry", +] diff --git a/binaries/runtime-python/build.rs b/binaries/runtime-python/build.rs new file mode 100644 index 0000000000..c8983b1f57 --- /dev/null +++ b/binaries/runtime-python/build.rs @@ -0,0 +1,13 @@ +fn main() { + // Re-emit `Py_3_N` cfgs from pyo3 into this crate so any future code that + // wants to gate on Python ABI levels (`#[cfg(Py_3_11)]` etc.) can do so + // without each contributor rediscovering this setup. See + // apis/python/node/build.rs for context; #1833 hit this issue first. + // + // Unconditional here: this crate *is* the Python runtime backend and always + // links pyo3. In the old monolithic `dora-runtime` this was gated behind the + // `python` feature, because that crate was also linked into `dora-cli` on + // systems below the workspace `abi3-py311` floor (e.g. Python 3.10 on + // ubuntu-22.04), where pyo3-build-config's interpreter probe fails. + pyo3_build_config::use_pyo3_cfgs(); +} diff --git a/binaries/runtime-python/src/lib.rs b/binaries/runtime-python/src/lib.rs new file mode 100644 index 0000000000..e1bf604a1e --- /dev/null +++ b/binaries/runtime-python/src/lib.rs @@ -0,0 +1,149 @@ +//! Python (PyO3) operator runtime backend. +//! +//! Loads a Python module's `Operator` class and runs it on the +//! [`dora_runtime_api`] event loop. Shipped inside the Python wheel and +//! launched by the daemon via `python -uc "import dora; dora.start_runtime()"`. +//! This is the only runtime crate that links `pyo3`. +//! +//! It also hosts shared-library operators, by delegating to +//! [`dora_runtime_shared_lib::SharedLibRunner`]. That is not incidental: when +//! the daemon is itself an embedded Python process (`current_exe` ends in +//! `python`/`python3`, e.g. `python -c "import dora; dora.start_daemon()"`), it +//! routes *native* runtime nodes here too — see `native_runtime_command` in the +//! daemon's `spawn::runtime_registry`. The pre-split `dora-runtime` handled +//! `SharedLibrary` unconditionally, so the wheel's runtime could always dlopen +//! them; keeping that arm wired preserves it. + +use dora_core::{ + config::NodeId, + descriptor::{Descriptor, OperatorDefinition, OperatorSource}, +}; +use dora_node_api::Event; +use dora_runtime_api::{OperatorRunner, RunnerGuard, RuntimeHandle}; +use dora_runtime_shared_lib::SharedLibRunner; +use eyre::{Context, Result}; +use tokio::sync::oneshot; + +mod runner; + +/// Runtime process entry point for Python operators. +pub fn main() -> eyre::Result<()> { + dora_runtime_api::main(PythonRunner) +} + +/// Backend hosting Python operators, and native ones by delegation. +pub struct PythonRunner; + +impl OperatorRunner for PythonRunner { + fn run_operator( + &self, + node_id: &NodeId, + operator: OperatorDefinition, + incoming_events: flume::Receiver, + handle: RuntimeHandle, + init_done: oneshot::Sender>, + dataflow_descriptor: &Descriptor, + ) -> eyre::Result { + match &operator.config.source { + OperatorSource::Python(source) => runner::run( + node_id, + &operator.id, + source, + handle, + incoming_events, + init_done, + dataflow_descriptor, + ) + .wrap_err_with(|| format!("failed to spawn Python operator for {}", operator.id)) + // Nothing to keep alive: the Python interpreter outlives the process. + .map(|()| None), + // `SharedLibrary` is reachable when the daemon runs as an embedded + // Python process (see the module docs): it sends native runtime + // nodes to this runtime, so hosting them here is what keeps that + // deployment working. `Wasm` goes the same way rather than growing a + // second copy of the "not supported yet" arm — the shared-library + // backend already owns that error, including the #2595 contract of + // leaving `init_done` unsignalled. + OperatorSource::SharedLibrary(_) | OperatorSource::Wasm(_) => SharedLibRunner + .run_operator( + node_id, + operator, + incoming_events, + handle, + init_done, + dataflow_descriptor, + ), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use dora_runtime_api::SharedAllocator; + + /// Drives `run_operator` and returns the error plus the `init_done` + /// receiver, for sources that cannot start. + fn run_failing(yaml: &str) -> (eyre::Report, oneshot::Receiver>) { + let operator: OperatorDefinition = + serde_yaml::from_str(yaml).expect("operator definition parses"); + let dataflow: Descriptor = + serde_yaml::from_str("nodes:\n - id: a\n").expect("descriptor parses"); + let (_events_in_tx, incoming_events) = flume::unbounded::(); + let (events_tx, _events_rx) = tokio::sync::mpsc::channel(1); + let (init_done_tx, init_done_rx) = oneshot::channel(); + + let err = PythonRunner + .run_operator( + &NodeId::from("node".to_string()), + operator, + incoming_events, + RuntimeHandle::new(events_tx, SharedAllocator::default()), + init_done_tx, + &dataflow, + ) + .expect_err("operator must fail to start"); + (err, init_done_rx) + } + + /// An unsupported operator source must surface a descriptive error from + /// `run_operator` rather than returning `Ok(())` while silently dropping the + /// `init_done` sender — which would leave the runtime task blocked in + /// `init_done.await` until it fails with the misleading "the `init_done` + /// channel was closed unexpectedly". + #[test] + fn wasm_source_returns_descriptive_error() { + let (err, mut init_done_rx) = run_failing("id: op\nwasm: model.wasm\n"); + assert!( + err.to_string().contains("WASM"), + "expected a descriptive WASM error, got: {err}" + ); + assert!( + init_done_rx.try_recv().is_err(), + "init_done must not receive a value for an unsupported source" + ); + } + + /// A shared-library operator must reach the shared-library backend rather + /// than be rejected as "wrong runtime" — the embedded-Python-daemon path + /// depends on it (see the module docs). A missing `.so` is the closest we + /// can get without building one: the error has to come from *loading* the + /// library, which only the delegated backend attempts. + #[test] + fn shared_library_source_is_delegated_not_rejected() { + let (err, mut init_done_rx) = run_failing("id: op\nshared-library: /nonexistent/dora-op\n"); + let msg = format!("{err:?}"); + assert!( + msg.contains("shared library"), + "expected a load failure from the shared-library backend, got: {msg}" + ); + assert!( + !msg.contains("this is the Python runtime"), + "shared-library operators must not be rejected by the Python runtime: {msg}" + ); + assert!( + init_done_rx.try_recv().is_err(), + "init_done must not receive a value when the library fails to load" + ); + } +} diff --git a/binaries/runtime/src/operator/python.rs b/binaries/runtime-python/src/runner.rs similarity index 99% rename from binaries/runtime/src/operator/python.rs rename to binaries/runtime-python/src/runner.rs index 837a8d58ca..8283e6f8d3 100644 --- a/binaries/runtime/src/operator/python.rs +++ b/binaries/runtime-python/src/runner.rs @@ -1,6 +1,5 @@ #![allow(clippy::borrow_deref_ref)] // clippy warns about code generated by #[pymethods] -use super::{OperatorEvent, StopReason}; use dora_core::{ config::{NodeId, OperatorId}, descriptor::{Descriptor, PythonSource, source_is_url}, @@ -11,6 +10,7 @@ use dora_node_api::Parameter; use dora_node_api::{Event, merged::MergedEvent}; use dora_operator_api_python::PyEvent; use dora_operator_api_types::DoraStatus; +use dora_runtime_api::{OperatorEvent, StopReason}; use eyre::{Context, Result, bail, eyre}; use pyo3::ffi::c_str; use pyo3::{ diff --git a/binaries/runtime-shared-lib/Cargo.toml b/binaries/runtime-shared-lib/Cargo.toml new file mode 100644 index 0000000000..ea8cdcdd3b --- /dev/null +++ b/binaries/runtime-shared-lib/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "dora-runtime-shared-lib" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +documentation.workspace = true +readme.workspace = true +description.workspace = true +license.workspace = true +repository.workspace = true + +# Shared-library (C ABI) operator runtime backend. Loads `.so`/`.dll`/`.dylib` +# operators via `libloading` and runs them on the `dora-runtime-api` event loop. +# Shipped inside the `dora` CLI as the `dora runtime` subcommand. + +[dependencies] +dora-runtime-api = { workspace = true } +dora-node-api = { workspace = true, default-features = false } +dora-core = { workspace = true } +dora-download = { workspace = true } +dora-operator-api-types = { workspace = true } +dora-tracing = { workspace = true, optional = true } +arrow = { workspace = true, features = ["ffi"] } +libloading = "0.9" +eyre = { workspace = true } +flume = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +tracing-opentelemetry = { version = "0.33.0", optional = true } + +[dev-dependencies] +serde_yaml = { workspace = true } + +[features] +default = ["tracing", "metrics"] +tracing = ["dora-runtime-api/tracing"] +metrics = ["dora-runtime-api/metrics"] +telemetry = ["tracing", "dep:dora-tracing", "dep:tracing-opentelemetry"] diff --git a/binaries/runtime-shared-lib/src/lib.rs b/binaries/runtime-shared-lib/src/lib.rs new file mode 100644 index 0000000000..f01fa9151d --- /dev/null +++ b/binaries/runtime-shared-lib/src/lib.rs @@ -0,0 +1,144 @@ +//! Shared-library (C ABI) operator runtime backend. +//! +//! Loads `.so`/`.dll`/`.dylib` operators via `libloading` and runs them on the +//! [`dora_runtime_api`] event loop. Shipped inside the `dora` CLI and launched +//! by the daemon as the `dora runtime` subcommand. +//! +//! [`SharedLibRunner`] is public because the Python runtime embeds it too: a +//! daemon that is itself an embedded Python process routes *native* operators to +//! `python -uc "import dora; dora.start_runtime()"`, so the wheel's runtime has +//! to be able to host them. See `dora-runtime-python`. + +use dora_core::{ + config::NodeId, + descriptor::{Descriptor, OperatorDefinition, OperatorSource}, +}; +use dora_node_api::Event; +use dora_runtime_api::{OperatorRunner, RunnerGuard, RuntimeHandle}; +use eyre::{Context, Result}; +use tokio::sync::oneshot; + +mod runner; + +/// Runtime process entry point for shared-library operators. +pub fn main() -> eyre::Result<()> { + dora_runtime_api::main(SharedLibRunner) +} + +/// Backend hosting `dora_init_operator`/`dora_on_event` C-ABI operators. +pub struct SharedLibRunner; + +impl OperatorRunner for SharedLibRunner { + fn run_operator( + &self, + node_id: &NodeId, + operator: OperatorDefinition, + incoming_events: flume::Receiver, + handle: RuntimeHandle, + init_done: oneshot::Sender>, + _dataflow_descriptor: &Descriptor, + ) -> eyre::Result { + match &operator.config.source { + // The loaded library is handed back as the runner guard so it stays + // mapped until the event loop has joined: values whose vtable lives + // in this `.so` (an `OperatorEvent::Panic` payload) can still be in + // flight, and dropping them after an unload SIGSEGVs. + OperatorSource::SharedLibrary(source) => runner::run( + node_id, + &operator.id, + source, + handle, + incoming_events, + init_done, + ) + .wrap_err_with(|| { + format!( + "failed to spawn shared library operator for {}", + operator.id + ) + }) + .map(|library| Some(Box::new(library) as Box)), + // Unsupported sources must return a descriptive error rather than + // `Ok(())` with a silently dropped `init_done` sender, which would + // leave the runtime task blocked in `init_done.await` until it fails + // with the misleading "the `init_done` channel was closed + // unexpectedly" (#2595). + OperatorSource::Python(_) => eyre::bail!( + "operator `{}` uses a Python source, but this is the shared-library \ + runtime; Python operators are spawned by the Python runtime \ + (`dora-runtime-python`)", + operator.id + ), + OperatorSource::Wasm(_) => eyre::bail!( + "operator `{}` uses a WASM source, which is not supported yet", + operator.id + ), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use dora_runtime_api::SharedAllocator; + + /// Drives `run_operator` for a source this backend cannot host and returns + /// the error plus the still-unsignalled `init_done` receiver. + fn run_unsupported(yaml: &str) -> (eyre::Report, oneshot::Receiver>) { + let operator: OperatorDefinition = + serde_yaml::from_str(yaml).expect("operator definition parses"); + let dataflow: Descriptor = + serde_yaml::from_str("nodes:\n - id: a\n").expect("descriptor parses"); + let (_events_in_tx, incoming_events) = flume::unbounded::(); + let (events_tx, _events_rx) = tokio::sync::mpsc::channel(1); + let (init_done_tx, init_done_rx) = oneshot::channel(); + + let err = SharedLibRunner + .run_operator( + &NodeId::from("node".to_string()), + operator, + incoming_events, + RuntimeHandle::new(events_tx, SharedAllocator::default()), + init_done_tx, + &dataflow, + ) + .expect_err("unsupported operator source must return an error"); + (err, init_done_rx) + } + + /// An unsupported operator source must surface a descriptive error from + /// `run_operator` rather than returning `Ok(())` while silently dropping the + /// `init_done` sender — which would leave the runtime task blocked in + /// `init_done.await` until it fails with the misleading "the `init_done` + /// channel was closed unexpectedly". + #[test] + fn wasm_source_returns_descriptive_error() { + let (err, mut init_done_rx) = run_unsupported("id: op\nwasm: model.wasm\n"); + assert!( + err.to_string().contains("WASM"), + "expected a descriptive WASM error, got: {err}" + ); + assert!( + init_done_rx.try_recv().is_err(), + "init_done must not receive a value for an unsupported source" + ); + } + + /// The cross-language arm this split introduces: a Python operator reaching + /// the shared-library runtime is a routing bug, and it must fail the same + /// way — descriptive error, `init_done` left unsignalled — rather than hang + /// the runtime task. + #[test] + fn python_source_returns_descriptive_error() { + let (err, mut init_done_rx) = run_unsupported("id: op\npython: op.py\n"); + let msg = err.to_string(); + assert!( + msg.contains("Python") && msg.contains("shared-library runtime"), + "expected an error naming the wrong runtime, got: {err}" + ); + assert!( + init_done_rx.try_recv().is_err(), + "init_done must not receive a value for a wrongly routed source" + ); + } +} diff --git a/binaries/runtime/src/operator/shared_lib.rs b/binaries/runtime-shared-lib/src/runner.rs similarity index 99% rename from binaries/runtime/src/operator/shared_lib.rs rename to binaries/runtime-shared-lib/src/runner.rs index f096190c8d..f929289175 100644 --- a/binaries/runtime/src/operator/shared_lib.rs +++ b/binaries/runtime-shared-lib/src/runner.rs @@ -1,4 +1,3 @@ -use super::{OperatorEvent, StopReason}; use dora_core::{ adjust_shared_library_path, config::{DataId, NodeId, OperatorId}, @@ -10,6 +9,7 @@ use dora_operator_api_types::{ DoraDropOperator, DoraInitOperator, DoraInitResult, DoraOnEvent, DoraResult, DoraStatus, Metadata, OnEventResult, Output, SendOutput, safer_ffi::closure::ArcDynFn1, }; +use dora_runtime_api::{OperatorEvent, StopReason}; use eyre::{Context, Result, bail, eyre}; use libloading::Symbol; use std::{ diff --git a/binaries/runtime/Cargo.toml b/binaries/runtime/Cargo.toml deleted file mode 100644 index 4d8f52994c..0000000000 --- a/binaries/runtime/Cargo.toml +++ /dev/null @@ -1,57 +0,0 @@ -[package] -name = "dora-runtime" -version.workspace = true -edition.workspace = true -rust-version.workspace = true -documentation.workspace = true -readme.workspace = true -description.workspace = true -license.workspace = true -repository.workspace = true - -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html - -[dependencies] -dora-node-api = { workspace = true, default-features = false } -dora-operator-api-python = { workspace = true, optional = true } -dora-operator-api-types = { workspace = true } -dora-core = { workspace = true } -dora-tracing = { workspace = true, optional = true } -dora-metrics = { workspace = true, optional = true } -dora-message = { workspace = true } -eyre = { workspace = true } -futures = { workspace = true } -futures-concurrency = "7.7.1" -libloading = "0.9.0" -serde_yaml = { workspace = true } -tokio = { workspace = true, features = ["full"] } -tokio-stream = "0.1.18" -# pyo3-abi3 flag allow simpler linking. See: https://pyo3.rs/v0.13.2/building_and_distribution.html -pyo3 = { workspace = true, features = ["eyre", "abi3-py311"], optional = true } -tracing = { workspace = true } -dora-download = { workspace = true } -flume = { workspace = true } -tracing-opentelemetry = { version = "0.33.0", optional = true } -pythonize = { workspace = true, optional = true } -arrow = { workspace = true, features = ["ffi"] } -aligned-vec = "0.6.4" - -[build-dependencies] -# Optional: pyo3-build-config's own build script (resolve-config feature) -# probes the Python interpreter and fails when it's older than the -# workspace abi3-py311 floor, which broke plain `cargo build -p dora-cli` -# on systems with Python <= 3.10. Only pull it in when Python is wanted. -pyo3-build-config = { workspace = true, optional = true } - -[features] -default = ["tracing", "metrics"] -tracing = ["dora-tracing"] -telemetry = ["tracing", "tracing-opentelemetry"] -metrics = ["dora-metrics"] -python = [ - "pyo3", - "dora-operator-api-python", - "pythonize", - "arrow/pyarrow", - "dep:pyo3-build-config", -] diff --git a/binaries/runtime/build.rs b/binaries/runtime/build.rs deleted file mode 100644 index 8e62d41e45..0000000000 --- a/binaries/runtime/build.rs +++ /dev/null @@ -1,13 +0,0 @@ -fn main() { - // Re-emit `Py_3_N` cfgs from pyo3 into this crate so any future code that - // wants to gate on Python ABI levels (`#[cfg(Py_3_11)]` etc.) can do so - // without each contributor rediscovering this setup. See - // apis/python/node/build.rs for context; #1833 hit this issue first. - // - // pyo3-build-config is an optional build-dependency tied to the `python` - // feature: its own build script probes the Python interpreter and fails - // on systems older than the workspace `abi3-py311` floor (e.g. Python - // 3.10 on ubuntu-22.04), which broke non-Python `cargo build -p dora-cli`. - #[cfg(feature = "python")] - pyo3_build_config::use_pyo3_cfgs(); -} diff --git a/docs/architecture.md b/docs/architecture.md index 55d5eaa3ed..c3e98c563f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -56,7 +56,9 @@ independently. | `binaries/cli` | dora-cli | CLI binary (`dora` command) — build, run, stop dataflows | | `binaries/coordinator` | dora-coordinator | Orchestrates distributed multi-daemon deployments; WebSocket server | | `binaries/daemon` | dora-daemon | Spawns nodes, manages shared-memory/TCP communication per machine | -| `binaries/runtime` | dora-runtime | In-process operator execution (Python/C/C++ via dlopen/PyO3) | +| `binaries/runtime-api` | dora-runtime-api | Language-neutral operator runtime SDK (event loop + `OperatorRunner` backend trait) | +| `binaries/runtime-shared-lib` | dora-runtime-shared-lib | Shared-library operator backend (C/C++/Rust via dlopen); shipped in the `dora` CLI | +| `binaries/runtime-python` | dora-runtime-python | Python operator backend (PyO3); shipped in the Python wheel | | `binaries/ros2-bridge-node` | dora-ros2-bridge-node | ROS2 integration node | | `binaries/record-node` | dora-record-node | Records dataflow messages to `.drec` format | | `binaries/replay-node` | dora-replay-node | Replays recorded messages from `.drec` files | diff --git a/docs/contributor-qa-cheatsheet.md b/docs/contributor-qa-cheatsheet.md index fd39719699..5b031de705 100644 --- a/docs/contributor-qa-cheatsheet.md +++ b/docs/contributor-qa-cheatsheet.md @@ -208,6 +208,7 @@ cargo test --all \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples ``` diff --git a/docs/testing-guide.md b/docs/testing-guide.md index 36d3a78fec..71a7108f57 100644 --- a/docs/testing-guide.md +++ b/docs/testing-guide.md @@ -28,6 +28,7 @@ cargo clippy --all \ # 3. Unit + integration tests (~90s first run) cargo test --all \ + --exclude dora-runtime-python \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python @@ -378,6 +379,7 @@ Add new test files in the `tests/` directory. For tests that need the full CLI s Always exclude Python packages: ```bash cargo test --all \ + --exclude dora-runtime-python \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python diff --git a/examples/c++-dataflow/README.md b/examples/c++-dataflow/README.md index d2e5ba9046..c7eeb663e9 100644 --- a/examples/c++-dataflow/README.md +++ b/examples/c++-dataflow/README.md @@ -43,15 +43,17 @@ For a manual build, follow these steps: ``` Omit the `-fPIC` argument on Windows. Replace the `liboperator_c_api.so` name with the shared library standard library prefix/extensions used on your OS, e.g. `.dll` on Windows. -**Build the dora coordinator and runtime:** +**Build the dora CLI:** -- Build the `dora-coordinator` executable using `cargo build -p dora-coordinator --release` -- Build the `dora-runtime` executable using `cargo build -p dora-runtime --release` +- Build the `dora` executable using `cargo build -p dora-cli --release` + - This is the only dora binary you need: it embeds the coordinator and the + daemon, and hosts shared-library operators like the one above through its + `dora runtime` subcommand, which the daemon spawns for you. **Run the dataflow:** -- Start the `dora-coordinator`, passing the paths to the dataflow file and the `dora-runtime` as arguments: +- Run the dataflow with the CLI built above: ``` - ../../target/release/dora-daemon --run-dataflow dataflow.yml ../../target/release/dora-runtime + ../../target/release/dora run dataflow.yml ``` diff --git a/examples/c++-dataflow/run.rs b/examples/c++-dataflow/run.rs index fd7069c16e..b41599c0b9 100644 --- a/examples/c++-dataflow/run.rs +++ b/examples/c++-dataflow/run.rs @@ -92,7 +92,7 @@ fn main() -> eyre::Result<()> { ], )?; - build_package("dora-runtime")?; + build_package("dora-runtime-shared-lib")?; // Bound the run so a wedged node fails fast via the daemon's stop // escalation instead of hanging until the CI step timeout (#2152). diff --git a/examples/c-dataflow/README.md b/examples/c-dataflow/README.md index 6c6f70414b..7d78ef8b4f 100644 --- a/examples/c-dataflow/README.md +++ b/examples/c-dataflow/README.md @@ -55,15 +55,17 @@ For a manual build, follow these steps: ``` Omit the `-fPIC` argument on Windows. Replace the `liboperator.so` name with the shared library standard library prefix/extensions used on your OS, e.g. `.dll` on Windows. -**Build the dora coordinator and runtime:** +**Build the dora CLI:** -- Build the `dora-coordinator` executable using `cargo build -p dora-coordinator --release` -- Build the `dora-runtime` executable using `cargo build -p dora-runtime --release` +- Build the `dora` executable using `cargo build -p dora-cli --release` + - This is the only dora binary you need: it embeds the coordinator and the + daemon, and hosts shared-library operators like the one above through its + `dora runtime` subcommand, which the daemon spawns for you. **Run the dataflow:** -- Start the `dora-coordinator`, passing the paths to the dataflow file and the `dora-runtime` as arguments: +- Run the dataflow with the CLI built above: ``` - ../../target/release/dora-daemon --run-dataflow dataflow.yml ../../target/release/dora-runtime + ../../target/release/dora run dataflow.yml ``` diff --git a/examples/cmake-dataflow/run.rs b/examples/cmake-dataflow/run.rs index 1a639f1eaa..42ab44f835 100644 --- a/examples/cmake-dataflow/run.rs +++ b/examples/cmake-dataflow/run.rs @@ -35,7 +35,7 @@ fn main() -> eyre::Result<()> { bail!("failed to build a cmake-generated project binary tree"); } - build_package("dora-runtime")?; + build_package("dora-runtime-shared-lib")?; // Bound the run so a wedged node fails fast via the daemon's stop // escalation instead of hanging until the CI step timeout (#2152). diff --git a/guide/src/concepts/architecture.md b/guide/src/concepts/architecture.md index 1bb5cc7906..aad9586307 100644 --- a/guide/src/concepts/architecture.md +++ b/guide/src/concepts/architecture.md @@ -37,7 +37,9 @@ All crates share the workspace version. | `binaries/cli` | dora-cli | CLI binary (`dora` command) — build, run, stop dataflows | | `binaries/coordinator` | dora-coordinator | Orchestrates distributed multi-daemon deployments; WebSocket server | | `binaries/daemon` | dora-daemon | Spawns nodes, manages shared-memory/TCP communication per machine | -| `binaries/runtime` | dora-runtime | In-process operator execution (Python/C/C++ via dlopen/PyO3) | +| `binaries/runtime-api` | dora-runtime-api | Language-neutral operator runtime SDK (event loop + `OperatorRunner` backend trait) | +| `binaries/runtime-shared-lib` | dora-runtime-shared-lib | Shared-library operator backend (C/C++/Rust via dlopen); shipped in the `dora` CLI | +| `binaries/runtime-python` | dora-runtime-python | Python operator backend (PyO3); shipped in the Python wheel | | `binaries/ros2-bridge-node` | dora-ros2-bridge-node | ROS2 integration node | | `binaries/record-node` | dora-record-node | Records dataflow messages to `.drec` format | | `binaries/replay-node` | dora-replay-node | Replays recorded messages from `.drec` files | diff --git a/guide/src/development/testing.md b/guide/src/development/testing.md index 53628c0f6e..2de2bd5c6f 100644 --- a/guide/src/development/testing.md +++ b/guide/src/development/testing.md @@ -19,6 +19,7 @@ cargo clippy --all \ # 3. Unit + integration tests (~90s first run) cargo test --all \ + --exclude dora-runtime-python \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python @@ -366,6 +367,7 @@ Add new test files in the `tests/` directory. For tests that need the full CLI s Always exclude Python packages: ```bash cargo test --all \ + --exclude dora-runtime-python \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python diff --git a/libraries/core/src/descriptor/mod.rs b/libraries/core/src/descriptor/mod.rs index 94e368c744..aa64e1ee17 100644 --- a/libraries/core/src/descriptor/mod.rs +++ b/libraries/core/src/descriptor/mod.rs @@ -14,9 +14,9 @@ use std::{ // reexport for compatibility pub use dora_message::descriptor::{ CoreNodeKind, CustomNode, DYNAMIC_SOURCE, Descriptor, Node, OperatorConfig, OperatorDefinition, - OperatorSource, PythonSource, ResolvedNode, RmwZenohCompatibility, Ros2BridgeConfig, - Ros2Direction, Ros2QosConfig, Ros2TopicConfig, Ros2TransportConfig, RuntimeNode, SHELL_SOURCE, - SingleOperatorDefinition, + OperatorSource, PythonSource, RUNTIME_PYTHON, RUNTIME_SHARED_LIBRARY, RUNTIME_WASM, + ResolvedNode, RmwZenohCompatibility, Ros2BridgeConfig, Ros2Direction, Ros2QosConfig, + Ros2TopicConfig, Ros2TransportConfig, RuntimeNode, SHELL_SOURCE, SingleOperatorDefinition, }; pub use validate::ResolvedNodeExt; pub use visualize::collect_dora_timers; diff --git a/libraries/message/src/descriptor.rs b/libraries/message/src/descriptor.rs index cc2fd96892..40366a7815 100644 --- a/libraries/message/src/descriptor.rs +++ b/libraries/message/src/descriptor.rs @@ -1089,6 +1089,28 @@ impl From for PythonSource { } } +/// Built-in runtime name for shared-library operators. +pub const RUNTIME_SHARED_LIBRARY: &str = "shared-library"; +/// Built-in runtime name for Python operators. +pub const RUNTIME_PYTHON: &str = "python"; +/// Built-in runtime name for WebAssembly operators. +pub const RUNTIME_WASM: &str = "wasm"; + +impl OperatorSource { + /// The name of the runtime that hosts operators declared with this source. + /// + /// This mapping is the single source of truth for "which runtime hosts this + /// operator": the daemon's spawn logic and the CLI's build hashing key on + /// the name rather than matching each variant. + pub fn runtime_name(&self) -> &'static str { + match self { + OperatorSource::SharedLibrary(_) => RUNTIME_SHARED_LIBRARY, + OperatorSource::Python(_) => RUNTIME_PYTHON, + OperatorSource::Wasm(_) => RUNTIME_WASM, + } + } +} + #[allow(missing_docs)] #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] pub struct CustomNode { @@ -1551,4 +1573,52 @@ _unstable_debug: let desc: Descriptor = serde_yaml::from_str(yaml).unwrap(); assert!(desc.debug.enable_debug_inspection); } + + #[test] + fn operator_source_shared_library_names_its_runtime() { + let cfg: OperatorConfig = serde_yaml::from_str("shared-library: build/op").unwrap(); + assert!(matches!(&cfg.source, OperatorSource::SharedLibrary(s) if s == "build/op")); + assert_eq!(cfg.source.runtime_name(), RUNTIME_SHARED_LIBRARY); + assert_eq!(cfg.source.runtime_name(), "shared-library"); + } + + #[test] + fn operator_source_python_source_only_names_its_runtime() { + let cfg: OperatorConfig = serde_yaml::from_str("python: op.py").unwrap(); + assert!(matches!(&cfg.source, OperatorSource::Python(py) if py.source == "op.py")); + assert_eq!(cfg.source.runtime_name(), RUNTIME_PYTHON); + } + + #[test] + fn operator_source_python_with_conda_env_names_its_runtime() { + let cfg: OperatorConfig = + serde_yaml::from_str("python:\n source: op.py\n conda_env: my-env").unwrap(); + match &cfg.source { + OperatorSource::Python(py) => { + assert_eq!(py.source, "op.py"); + assert_eq!(py.conda_env.as_deref(), Some("my-env")); + } + other => panic!("expected python source, got {other:?}"), + } + assert_eq!(cfg.source.runtime_name(), RUNTIME_PYTHON); + } + + #[test] + fn operator_source_wasm_names_its_runtime() { + let cfg: OperatorConfig = serde_yaml::from_str("wasm: op.wasm").unwrap(); + assert!(matches!(&cfg.source, OperatorSource::Wasm(s) if s == "op.wasm")); + assert_eq!(cfg.source.runtime_name(), RUNTIME_WASM); + } + + /// The runtime a node is spawned with must survive a descriptor round-trip: + /// the daemon re-parses the serialized descriptor before spawning. + #[test] + fn operator_source_runtime_survives_a_serde_roundtrip() { + for yaml in ["shared-library: build/op", "python: op.py", "wasm: op.wasm"] { + let cfg: OperatorConfig = serde_yaml::from_str(yaml).unwrap(); + let serialized = serde_yaml::to_string(&cfg).unwrap(); + let reparsed: OperatorConfig = serde_yaml::from_str(&serialized).unwrap(); + assert_eq!(cfg.source.runtime_name(), reparsed.source.runtime_name()); + } + } } diff --git a/scripts/qa/all.sh b/scripts/qa/all.sh index aa573c28f5..a9102df74b 100755 --- a/scripts/qa/all.sh +++ b/scripts/qa/all.sh @@ -273,6 +273,7 @@ case "$MODE" in --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples run "coverage" scripts/qa/coverage.sh @@ -303,6 +304,7 @@ case "$MODE" in --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples \ -- proptest diff --git a/scripts/qa/ci-nightly-jobs.sh b/scripts/qa/ci-nightly-jobs.sh index 7c3f69c508..ec249a42b2 100755 --- a/scripts/qa/ci-nightly-jobs.sh +++ b/scripts/qa/ci-nightly-jobs.sh @@ -1646,6 +1646,7 @@ job_test_cross_platform() { --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python \ --exclude dora-examples } @@ -1966,6 +1967,7 @@ job_cross_check() { --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python ;; Darwin) @@ -1977,12 +1979,14 @@ job_cross_check() { --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python else cargo check --target x86_64-apple-darwin --all \ --exclude dora-node-api-python \ --exclude dora-operator-api-python \ --exclude dora-ros2-bridge-python \ + --exclude dora-runtime-python \ --exclude dora-cli-api-python fi ;; diff --git a/scripts/qa/coverage.sh b/scripts/qa/coverage.sh index 6cfc1974be..9218ba633f 100755 --- a/scripts/qa/coverage.sh +++ b/scripts/qa/coverage.sh @@ -25,6 +25,7 @@ EXCLUDES=( --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python + --exclude dora-runtime-python --exclude dora-cli-api-python --exclude dora-examples # C++ bindings: their build.rs fails under llvm-cov instrumentation