Dora 项目架构与开发模式完全指南从 Dataflow 编排、节点生命周期到零拷贝数据通路【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora本指南以.claude/skills/adora-project/SKILL.md为骨架系统梳理 DoraDataflow-Oriented Robotic Architecture的运行时架构、消息协议、节点/算子 API、Dataflow YAML 模式与工程约定并深入到本仓库源码apis/、libraries/、binaries/验证其实现细节。读完本文你将掌握 Dora 四大组件CLI / Coordinator / Daemon / Node-Operator的职责边界与协作流程能够编写 Dataflow YAML 描述管道、用 Rust 实现 Node 与 Operator、理解共享内存与零拷贝的 4KB 阈值机制并遵循仓库的构建、测试与提交约定参与开发。架构总览四个组件一条流水线Dora 的运行时由四个层次组成它们在启动一条数据流时按以下方式连接CLI (WS:6013) -- Coordinator -- Daemon(s) -- Nodes / Operators (orchestrate) (per machine) (user code)CLI通过 WebSocket端口6013向 Coordinator 下发指令对应仓库binaries/cli/与libraries/message/src/cli_to_coordinator.rs、coordinator_to_cli.rs中的控制协议。Coordinator单一编排者维护持久化状态内存或 redb 后端见下文 CoordinatorStore负责基于标签label调度节点到目标机器。代码位于binaries/coordinator/。Daemon每台机器一个的进程管理器拥有共享内存区域通过 TCP/Unix Socket 与节点通信负责拉起、监控和终止节点进程。代码位于binaries/daemon/。Node独立的操作系统进程承载用户代码大于 4KB 的消息走共享内存小消息走 TCP。Operator进程内in-process运行在运行时runtime中没有 IPC 开销是官方推荐的使用方式见apis/rust/operator/src/lib.rs的文档注释It is the recommended way of usingdora。消息协议libraries/message/中的类型体系所有组件间消息都定义在 libraries/message/src/这是理解 Dora 内部通信的钥匙。核心消息类型如下表消息方向文件用途DaemonRequest/DaemonReplyNode ↔ Daemonnode_to_daemon.rs/daemon_to_node.rs数据收发、订阅、注册DaemonCoordinatorEventCoordinator → Daemoncoordinator_to_daemon.rs启动/停止节点、reloadCoordinatorRequestDaemon → Coordinatordaemon_to_coordinator.rs状态、日志、节点结果上报InterDaemonEventDaemon ↔ Daemondaemon_to_daemon.rs跨机器消息路由从源码看DaemonRequest的实际变体DaemonRequestlibraries/message/src/node_to_daemon.rs是一个#[non_exhaustive]枚举实际包含Register(NodeRegisterRequest)节点启动后向 Daemon 注册自己Subscribe订阅数据通道SendMessage { output_id, metadata, data }发送一条输出数据OutputSent/CloseOutputs/OutputsDone输出确认与收尾信号NextEvent/EventStreamDropped事件流轮询与释放NodeConfig { node_id }查询节点配置ExtensionStore/ExtensionLoad/ExtensionDrop/ExtensionRequest扩展表extension table机制——Dora 本身不解释namespace/key/value它为数据流作用域内的传输扩展提供带外通道tensor-pool 扩展就依赖它做跨机器注册与池写入。值得注意的是该枚举还有两个与性能相关的方法encode_size_hint()为encode_presized预分配缓冲和expects_tcp_binary_reply()/expects_tcp_json_reply()决定应答走二进制还是 JSON。源码注释强调匹配必须是穷尽的——忘记给新变体上报体积会让预分配退化为从空缓冲增长且没有任何测试能发现。这是本项目性能关键路径显式化编码风格的典型例子。版本兼容消息是带版本号的连接初始化时进行版本校验versions_compatible见libraries/message/src/lib.rs。daemon_to_node.rs中DaemonReply::ExtensionReply被特意追加在枚举末尾注释说明原因Python 节点 API 与 daemon 分开发布PyPI混合版本配对时不能误解码旧应答——枚举变体追加要放最后是这套协议的一条隐式规则。节点生命周期从 Spawn 到 SIGKILL节点从启动到退出的完整流程SKILL.md中的 6 步Coordinator 向目标 Daemon 发送SpawnNode依据标签选择目标Daemon 创建RunningNode并把RuntimeConfig作为JSON 环境变量传给节点进程节点进程启动通过 TCP/Unix Socket 回连 Daemon节点注册输入/输出订阅数据通道数据通过有界的flumechannel 在进程内部路由大负载走共享内存停止时软终止SIGTERM→ 宽限期 → 硬终止SIGKILL。宽限期常量源码中的硬数字binaries/daemon/src/running_dataflow.rs定义了DEFAULT_STOP_GRACE Duration::from_millis(10_000)10 秒。更重要的约束写在apis/rust/node/src/node/mod.rs中ZENOH_TEARDOWN_TIMEOUT的注释里该常量3 秒必须稳定低于daemon 的强制 kill 宽限10s 10s/2 15s否则一个网络运行时被卡住的节点会在 teardown 期间被TerminateProcess在 Windows 上表现为ExitCode(1)并污染 nightly 测试对应 issue dora-rs/dora#2742。仓库中还有专门的测试zenoh_teardown_fits_within_daemon_force_kill_grace守护这一不变量。节点初始化环境变量DoraNode::init_from_env()apis/rust/node/src/node/mod.rs按以下优先级决定运行模式DORA_NODE_CONFIG已设置 → 正常 daemon 模式由 daemon 注入的 YAML 反序列化为NodeConfigDORA_TEST_WITH_INPUTS已设置 → 集成测试模式从 JSONL 读取输入默认写到同目录outputs.jsonl可用DORA_TEST_WRITE_OUTPUTS_TO覆盖否则若 stdin 是终端isatty→ 回退到init_interactive()交互模式。另有DORA_RUNTIME_TYPE_CHECK环境变量控制运行时类型检查error为严格报错、warn/1/true/on为警告、0/false/off或未设置为关闭默认。关键抽象Node API 与 Operator APIDoraNodeRust 节点 APIDoraNode定义在 apis/rust/node/src/node/mod.rs核心接口pub struct DoraNode { /* ... */ } impl DoraNode { pub fn init(node_config: NodeConfig) - NodeResult(Self, EventStream); pub fn init_from_env() - NodeResult(Self, EventStream); pub fn init_from_node_id(node_id: NodeId) - NodeResult(Self, EventStream); pub fn init_flexible(node_id: NodeId) - NodeResult(Self, EventStream); pub fn init_interactive() - NodeResult(Self, EventStream); pub fn send_output(mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array) - NodeResult(); pub fn send_service_request(mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array) - NodeResultString; // returns auto-generated request_id pub fn send_service_response(mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array) - NodeResult(); }各初始化函数的分工init_from_env()推荐方式适配 daemon 注入的环境变量且能优雅回退到交互/测试模式init_from_env_force()环境变量缺失时直接报错不回退init_from_node_id(node_id)供动态节点dynamic nodes使用通过DoraNodeBuilder建立连接可自定义 daemon 端口daemon_portinit_flexible(node_id)先试传统环境变量路径失败再回退到动态节点路径适合有时静态、有时动态的节点init_interactive()独立运行模式从终端提示输入不连接 daemon、无法参与数据流官方注释建议不要直接使用优先init_from_env。DoraNode结构体还携带了完整的内部状态zenoh_session直接节点间 pub/sub 数据平面交互/测试模式下为None、sample_allocator共享内存 provider、zenoh_publishers每个输出的 zenoh 发布器及启动握手状态、restart_count重启次数以及可选的运行时类型检查状态。DoraOperatorRust 算子 API// apis/rust/operator/src/lib.rs pub trait DoraOperator: Default { fn on_event(mut self, event: Event, output_sender: mut DoraOutputSender) - ResultDoraStatus, String; }Operator 由数据流运行时驱动通过Default构造然后通过on_event接收Event流。Event是#[non_exhaustive]枚举包含Input { id, metadata, data }某个输入到达InputParseError { id, error }某输入负载无法解码为 Arrow 数组InputClosed { id }某输入源结束后续不再有InputStop运行时请求优雅退出应返回DoraStatus::StopError { error }事件流本身的错误区别于单个输入的解码失败。因为Event是non_exhaustive实现必须带 catch-all 分支以兼容未来新增变体。创建算子的脚手架命令是dora new op --kind operator。数据格式Apache Arrow 与 4KB 零拷贝阈值Dora 全程使用Apache Arrow 列式格式零序列化Python 绑定通过arrow::pyarrow做跨 FFI 零拷贝。消息 4KB → 使用共享内存每个节点 4 个命名区域消息 4KB → 直接走 TCP。阈值常量定义在 apis/rust/node/src/node/mod.rspub const ZERO_COPY_THRESHOLD: usize 4096;。源码注释解释了为什么是 4KB共享内存以内存页为单位共享而典型内存页就是 4KiB小于页大小的消息共享整页反而有内存与建立共享段的开销对小消息而言拷贝进堆缓冲发布更便宜。该阈值在运行时可通过DORA_ZERO_COPY_THRESHOLD环境变量覆盖见DoraNode::zero_copy_threshold。在 zenoh 数据平面上该阈值决定输出的发布方式达到或超过阈值的负载走 zenoh 共享内存本地订阅者零拷贝更小的负载走 zenoh 堆缓冲put。若一个大负载没拿到共享内存缓冲则走可靠的 daemon 路径而不是 zenoh——因为分片的 express 发布会被静默丢弃issue dora-rs/dora#2366。启动握手StartupHandshake可靠的快速路径切换源码中StartupHandshakeapis/rust/node/src/node/mod.rs实现了一个巧妙的机制zenoh 数据平面是节点间直接 pub/sub对尚未传播到发布器的订阅会丢弃样本快源因此可能丢失首批消息。Dora 的解法是端到端证明路由marker 走输出真实 topic消费方 ack 走acktopic 返回ack 到达即证明该路由对能承载数据。在此之前的每次发送都走 daemon 路径因此什么都不会丢宽限期ZENOH_STARTUP_GRACE500ms可调结束后仍未 ack 的输出被冻结在 daemon 路径上——只损失快速路径不损失正确性。该握手在init返回前必定结束settle()要么看到 ack 完成、要么冻结。Dataflow YAML Schema描述管道的标准语法Dataflow YAML 是 Dora 管道的图纸。SKILL.md给出的完整模式如下nodes: my-node: path: path/to/executable # or shell: or Python script inputs: input_name: other-node/output # Simple form input_with_opts: # Extended form source: other-node/output queue_size: 10 queue_policy: drop_oldest # drop_oldest (default) | backpressure outputs: - output_name env: KEY: value args: -v --some-flag foo # String, not a list restart_policy: on-failure # never | on-failure | always health_check_timeout: 2.0 # seconds (per node) deploy: # unstable, may change machine: gpu-server顶层描述符字段health_check_interval: 5.0 # seconds (global)各字段的要点path节点可执行文件路径或shell:前缀命令或 Python 脚本inputs简单形式input_name: other-node/output直接引用上游扩展形式可配置queue_size队列容量与queue_policydrop_oldest丢弃最旧默认backpressure背压。YAML 校验实现在 libraries/core/src/descriptor/validate.rsoutputs声明输出名env注入到节点进程的环境变量args注意是字符串而非列表restart_policynever不重启/on-failure失败重启默认示例值/always总是重启health_check_timeout单节点健康检查超时秒deploy.machine部署目标机器标签注意标记为 unstable未来可能变更——实际节点放置由标签调度在 spawn 时决定见daemon_to_node.rs中OutputRouting的注释descriptor 的deploy是意图而非落点。虚拟输入daemon 生成每个节点都可以消费 daemon 自动生成的虚拟输入dora/timer/secs/N或dora/timer/millis/N周期性定时器dora/logs、dora/logs/level、dora/logs/level/node日志流。通信模式Topic、Service、ActionTopic默认标准发布/订阅。节点发送输出所有订阅者收到。Service请求/应答客户端调用send_service_request会获得自动生成的request_id// Client: request_id is auto-generated and returned let request_id node.send_service_request( service_output.into(), MetadataParameters::default(), data, )?; // Server: pass through request_id from incoming metadata if let Some(req_id) metadata.get(dora_message::metadata::REQUEST_ID) { let mut params MetadataParameters::default(); params.insert(REQUEST_ID.to_string(), req_id.clone()); node.send_service_response(response_output.into(), params, result)?; }关键点服务端必须把传入 metadata 中的REQUEST_ID原样回传应答才能被路由到正确的请求方。ActionGoal/Feedback/Result使用goal_id和goal_statusmetadata 键支持取消。可参考examples/action-example/client/server 双节点与examples/service-example/、examples/c-service-action/中的完整可运行示例。状态与参数Coordinator Store通过CoordinatorStoretrait 抽象持久化状态后端有内存实现InMemoryStore见binaries/coordinator/src/lib.rs的导出与 redb 实现libraries/coordinator-store/参数系统按 dataflow 作用域的键值对通过SetParam/GetParam/DeleteParam传播分布式状态使用UHLChybrid logical clock混合逻辑时钟保证跨机器因果序节点状态restart_count、last_activity原子变量、pid容错统计FaultToleranceStats原子计数器binaries/daemon/src/fault_tolerance.rs。ID 约定DataflowIdUUID v7——可排序、基于时间戳天然支持按时间顺序归档与检索NodeId来自 YAML 的字符串标识时间戳UHLC保证跨机器排序Metadata 键定义在 libraries/message/src/metadata.rs包括REQUEST_ID、GOAL_ID、GOAL_STATUS、SESSION_ID等。错误处理模式应用层错误使用eyre::Result配合.context()链补充上下文NodeErrorCause枚举区分错误归因GraceDuration宽限期相关、Cascading级联错误、FailedToSpawn拉取失败、Other重启策略带指数退避可配置窗口级联错误追踪记录哪个节点导致了哪个失败这是Cascading归因的基础。测试约定分层测试矩阵变更类型测试层级位置库函数单元#[cfg(test)]同文件Coordinator/Daemon 行为集成测试binaries/coordinator/tests/CLI 命令Smoke联网tests/example-smoke.rsDataflow 功能Smoke两种模式tests/example-smoke.rsBug 修复回归测试能复现问题的层级Smoke 测试辅助函数run_smoke_test(name, yaml, timeout)—— 联网模式up/start/poll/stop/downrun_smoke_test_local(name, yaml, stop_after_secs)—— 本地模式run --stop-after。集成测试setup_integration_testing()通过DORA_TEST_WITH_INPUTS环境变量注入 JSON 输入见上文节点初始化的测试模式。构建命令速查# Build (exclude Python) 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-runtime-python --exclude dora-cli-api-python --exclude dora-examples # Single crate cargo test -p dora-core # Pre-commit (mandatory) cargo fmt --all -- --check cargo clippy --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python -- -D warnings注意Python 相关 crate 在多数构建/测试命令中被显式排除它们体积大且依赖特定环境cargo clippy必须以-D warnings通过这是提交前的强制门槛。PyO3 绑定Python 生态接入Python Node APIapis/python/node/——#[pyclass] Node实现迭代器协议Python Operator APIapis/python/operator/—— 类型转换工具Python CLI APIapis/python/cli/——build()、run()、start_runtime()使用 PyO3 0.29启用eyre、abi3-py311、multiple-pymethodsfeaturesArrow 数组通过arrow::pyarrowFFI 零拷贝传递阻塞 Rust 操作期间释放 GILpy.detach()避免阻塞 Python 事件循环。关键工程约定速查Rust edition 2024MSRV 1.85.0Workspace 版本统一为 0.2.0所有 crate 共享进程内部事件路由使用flume有界 MPSC异步运行时为tokiofull features跨机器 pub/sub 使用Zenoh零拷贝阈值4KB日志/指标走fire-and-forget策略——绝不在数据路径上阻塞相关实现见libraries/log-utils/与binaries/daemon/src/log.rs。进一步阅读数据流示例从 rust-dataflow/Rust 节点、python-dataflow/Python 节点、c-dataflow/C 节点出发覆盖三种主流语言通信模式示例service-example/、action-example/、streaming-example/容错与生命周期测试tests/dataflows/ 下的 YAML 与 tests/fault-tolerance-e2e.rs动态拓扑dynamic-add-remove/Python与 rust-dynamic-add-remove/Rust深入文档architecture.md、yaml-spec.md、api-rust.md、api-python.md、api-c.md、api-cxx.md。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Dora 项目架构与开发模式完全指南：从 Dataflow 编排、节点生命周期到零拷贝数据通路
Dora 项目架构与开发模式完全指南从 Dataflow 编排、节点生命周期到零拷贝数据通路【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora本指南以.claude/skills/adora-project/SKILL.md为骨架系统梳理 DoraDataflow-Oriented Robotic Architecture的运行时架构、消息协议、节点/算子 API、Dataflow YAML 模式与工程约定并深入到本仓库源码apis/、libraries/、binaries/验证其实现细节。读完本文你将掌握 Dora 四大组件CLI / Coordinator / Daemon / Node-Operator的职责边界与协作流程能够编写 Dataflow YAML 描述管道、用 Rust 实现 Node 与 Operator、理解共享内存与零拷贝的 4KB 阈值机制并遵循仓库的构建、测试与提交约定参与开发。架构总览四个组件一条流水线Dora 的运行时由四个层次组成它们在启动一条数据流时按以下方式连接CLI (WS:6013) -- Coordinator -- Daemon(s) -- Nodes / Operators (orchestrate) (per machine) (user code)CLI通过 WebSocket端口6013向 Coordinator 下发指令对应仓库binaries/cli/与libraries/message/src/cli_to_coordinator.rs、coordinator_to_cli.rs中的控制协议。Coordinator单一编排者维护持久化状态内存或 redb 后端见下文 CoordinatorStore负责基于标签label调度节点到目标机器。代码位于binaries/coordinator/。Daemon每台机器一个的进程管理器拥有共享内存区域通过 TCP/Unix Socket 与节点通信负责拉起、监控和终止节点进程。代码位于binaries/daemon/。Node独立的操作系统进程承载用户代码大于 4KB 的消息走共享内存小消息走 TCP。Operator进程内in-process运行在运行时runtime中没有 IPC 开销是官方推荐的使用方式见apis/rust/operator/src/lib.rs的文档注释It is the recommended way of usingdora。消息协议libraries/message/中的类型体系所有组件间消息都定义在 libraries/message/src/这是理解 Dora 内部通信的钥匙。核心消息类型如下表消息方向文件用途DaemonRequest/DaemonReplyNode ↔ Daemonnode_to_daemon.rs/daemon_to_node.rs数据收发、订阅、注册DaemonCoordinatorEventCoordinator → Daemoncoordinator_to_daemon.rs启动/停止节点、reloadCoordinatorRequestDaemon → Coordinatordaemon_to_coordinator.rs状态、日志、节点结果上报InterDaemonEventDaemon ↔ Daemondaemon_to_daemon.rs跨机器消息路由从源码看DaemonRequest的实际变体DaemonRequestlibraries/message/src/node_to_daemon.rs是一个#[non_exhaustive]枚举实际包含Register(NodeRegisterRequest)节点启动后向 Daemon 注册自己Subscribe订阅数据通道SendMessage { output_id, metadata, data }发送一条输出数据OutputSent/CloseOutputs/OutputsDone输出确认与收尾信号NextEvent/EventStreamDropped事件流轮询与释放NodeConfig { node_id }查询节点配置ExtensionStore/ExtensionLoad/ExtensionDrop/ExtensionRequest扩展表extension table机制——Dora 本身不解释namespace/key/value它为数据流作用域内的传输扩展提供带外通道tensor-pool 扩展就依赖它做跨机器注册与池写入。值得注意的是该枚举还有两个与性能相关的方法encode_size_hint()为encode_presized预分配缓冲和expects_tcp_binary_reply()/expects_tcp_json_reply()决定应答走二进制还是 JSON。源码注释强调匹配必须是穷尽的——忘记给新变体上报体积会让预分配退化为从空缓冲增长且没有任何测试能发现。这是本项目性能关键路径显式化编码风格的典型例子。版本兼容消息是带版本号的连接初始化时进行版本校验versions_compatible见libraries/message/src/lib.rs。daemon_to_node.rs中DaemonReply::ExtensionReply被特意追加在枚举末尾注释说明原因Python 节点 API 与 daemon 分开发布PyPI混合版本配对时不能误解码旧应答——枚举变体追加要放最后是这套协议的一条隐式规则。节点生命周期从 Spawn 到 SIGKILL节点从启动到退出的完整流程SKILL.md中的 6 步Coordinator 向目标 Daemon 发送SpawnNode依据标签选择目标Daemon 创建RunningNode并把RuntimeConfig作为JSON 环境变量传给节点进程节点进程启动通过 TCP/Unix Socket 回连 Daemon节点注册输入/输出订阅数据通道数据通过有界的flumechannel 在进程内部路由大负载走共享内存停止时软终止SIGTERM→ 宽限期 → 硬终止SIGKILL。宽限期常量源码中的硬数字binaries/daemon/src/running_dataflow.rs定义了DEFAULT_STOP_GRACE Duration::from_millis(10_000)10 秒。更重要的约束写在apis/rust/node/src/node/mod.rs中ZENOH_TEARDOWN_TIMEOUT的注释里该常量3 秒必须稳定低于daemon 的强制 kill 宽限10s 10s/2 15s否则一个网络运行时被卡住的节点会在 teardown 期间被TerminateProcess在 Windows 上表现为ExitCode(1)并污染 nightly 测试对应 issue dora-rs/dora#2742。仓库中还有专门的测试zenoh_teardown_fits_within_daemon_force_kill_grace守护这一不变量。节点初始化环境变量DoraNode::init_from_env()apis/rust/node/src/node/mod.rs按以下优先级决定运行模式DORA_NODE_CONFIG已设置 → 正常 daemon 模式由 daemon 注入的 YAML 反序列化为NodeConfigDORA_TEST_WITH_INPUTS已设置 → 集成测试模式从 JSONL 读取输入默认写到同目录outputs.jsonl可用DORA_TEST_WRITE_OUTPUTS_TO覆盖否则若 stdin 是终端isatty→ 回退到init_interactive()交互模式。另有DORA_RUNTIME_TYPE_CHECK环境变量控制运行时类型检查error为严格报错、warn/1/true/on为警告、0/false/off或未设置为关闭默认。关键抽象Node API 与 Operator APIDoraNodeRust 节点 APIDoraNode定义在 apis/rust/node/src/node/mod.rs核心接口pub struct DoraNode { /* ... */ } impl DoraNode { pub fn init(node_config: NodeConfig) - NodeResult(Self, EventStream); pub fn init_from_env() - NodeResult(Self, EventStream); pub fn init_from_node_id(node_id: NodeId) - NodeResult(Self, EventStream); pub fn init_flexible(node_id: NodeId) - NodeResult(Self, EventStream); pub fn init_interactive() - NodeResult(Self, EventStream); pub fn send_output(mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array) - NodeResult(); pub fn send_service_request(mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array) - NodeResultString; // returns auto-generated request_id pub fn send_service_response(mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array) - NodeResult(); }各初始化函数的分工init_from_env()推荐方式适配 daemon 注入的环境变量且能优雅回退到交互/测试模式init_from_env_force()环境变量缺失时直接报错不回退init_from_node_id(node_id)供动态节点dynamic nodes使用通过DoraNodeBuilder建立连接可自定义 daemon 端口daemon_portinit_flexible(node_id)先试传统环境变量路径失败再回退到动态节点路径适合有时静态、有时动态的节点init_interactive()独立运行模式从终端提示输入不连接 daemon、无法参与数据流官方注释建议不要直接使用优先init_from_env。DoraNode结构体还携带了完整的内部状态zenoh_session直接节点间 pub/sub 数据平面交互/测试模式下为None、sample_allocator共享内存 provider、zenoh_publishers每个输出的 zenoh 发布器及启动握手状态、restart_count重启次数以及可选的运行时类型检查状态。DoraOperatorRust 算子 API// apis/rust/operator/src/lib.rs pub trait DoraOperator: Default { fn on_event(mut self, event: Event, output_sender: mut DoraOutputSender) - ResultDoraStatus, String; }Operator 由数据流运行时驱动通过Default构造然后通过on_event接收Event流。Event是#[non_exhaustive]枚举包含Input { id, metadata, data }某个输入到达InputParseError { id, error }某输入负载无法解码为 Arrow 数组InputClosed { id }某输入源结束后续不再有InputStop运行时请求优雅退出应返回DoraStatus::StopError { error }事件流本身的错误区别于单个输入的解码失败。因为Event是non_exhaustive实现必须带 catch-all 分支以兼容未来新增变体。创建算子的脚手架命令是dora new op --kind operator。数据格式Apache Arrow 与 4KB 零拷贝阈值Dora 全程使用Apache Arrow 列式格式零序列化Python 绑定通过arrow::pyarrow做跨 FFI 零拷贝。消息 4KB → 使用共享内存每个节点 4 个命名区域消息 4KB → 直接走 TCP。阈值常量定义在 apis/rust/node/src/node/mod.rspub const ZERO_COPY_THRESHOLD: usize 4096;。源码注释解释了为什么是 4KB共享内存以内存页为单位共享而典型内存页就是 4KiB小于页大小的消息共享整页反而有内存与建立共享段的开销对小消息而言拷贝进堆缓冲发布更便宜。该阈值在运行时可通过DORA_ZERO_COPY_THRESHOLD环境变量覆盖见DoraNode::zero_copy_threshold。在 zenoh 数据平面上该阈值决定输出的发布方式达到或超过阈值的负载走 zenoh 共享内存本地订阅者零拷贝更小的负载走 zenoh 堆缓冲put。若一个大负载没拿到共享内存缓冲则走可靠的 daemon 路径而不是 zenoh——因为分片的 express 发布会被静默丢弃issue dora-rs/dora#2366。启动握手StartupHandshake可靠的快速路径切换源码中StartupHandshakeapis/rust/node/src/node/mod.rs实现了一个巧妙的机制zenoh 数据平面是节点间直接 pub/sub对尚未传播到发布器的订阅会丢弃样本快源因此可能丢失首批消息。Dora 的解法是端到端证明路由marker 走输出真实 topic消费方 ack 走acktopic 返回ack 到达即证明该路由对能承载数据。在此之前的每次发送都走 daemon 路径因此什么都不会丢宽限期ZENOH_STARTUP_GRACE500ms可调结束后仍未 ack 的输出被冻结在 daemon 路径上——只损失快速路径不损失正确性。该握手在init返回前必定结束settle()要么看到 ack 完成、要么冻结。Dataflow YAML Schema描述管道的标准语法Dataflow YAML 是 Dora 管道的图纸。SKILL.md给出的完整模式如下nodes: my-node: path: path/to/executable # or shell: or Python script inputs: input_name: other-node/output # Simple form input_with_opts: # Extended form source: other-node/output queue_size: 10 queue_policy: drop_oldest # drop_oldest (default) | backpressure outputs: - output_name env: KEY: value args: -v --some-flag foo # String, not a list restart_policy: on-failure # never | on-failure | always health_check_timeout: 2.0 # seconds (per node) deploy: # unstable, may change machine: gpu-server顶层描述符字段health_check_interval: 5.0 # seconds (global)各字段的要点path节点可执行文件路径或shell:前缀命令或 Python 脚本inputs简单形式input_name: other-node/output直接引用上游扩展形式可配置queue_size队列容量与queue_policydrop_oldest丢弃最旧默认backpressure背压。YAML 校验实现在 libraries/core/src/descriptor/validate.rsoutputs声明输出名env注入到节点进程的环境变量args注意是字符串而非列表restart_policynever不重启/on-failure失败重启默认示例值/always总是重启health_check_timeout单节点健康检查超时秒deploy.machine部署目标机器标签注意标记为 unstable未来可能变更——实际节点放置由标签调度在 spawn 时决定见daemon_to_node.rs中OutputRouting的注释descriptor 的deploy是意图而非落点。虚拟输入daemon 生成每个节点都可以消费 daemon 自动生成的虚拟输入dora/timer/secs/N或dora/timer/millis/N周期性定时器dora/logs、dora/logs/level、dora/logs/level/node日志流。通信模式Topic、Service、ActionTopic默认标准发布/订阅。节点发送输出所有订阅者收到。Service请求/应答客户端调用send_service_request会获得自动生成的request_id// Client: request_id is auto-generated and returned let request_id node.send_service_request( service_output.into(), MetadataParameters::default(), data, )?; // Server: pass through request_id from incoming metadata if let Some(req_id) metadata.get(dora_message::metadata::REQUEST_ID) { let mut params MetadataParameters::default(); params.insert(REQUEST_ID.to_string(), req_id.clone()); node.send_service_response(response_output.into(), params, result)?; }关键点服务端必须把传入 metadata 中的REQUEST_ID原样回传应答才能被路由到正确的请求方。ActionGoal/Feedback/Result使用goal_id和goal_statusmetadata 键支持取消。可参考examples/action-example/client/server 双节点与examples/service-example/、examples/c-service-action/中的完整可运行示例。状态与参数Coordinator Store通过CoordinatorStoretrait 抽象持久化状态后端有内存实现InMemoryStore见binaries/coordinator/src/lib.rs的导出与 redb 实现libraries/coordinator-store/参数系统按 dataflow 作用域的键值对通过SetParam/GetParam/DeleteParam传播分布式状态使用UHLChybrid logical clock混合逻辑时钟保证跨机器因果序节点状态restart_count、last_activity原子变量、pid容错统计FaultToleranceStats原子计数器binaries/daemon/src/fault_tolerance.rs。ID 约定DataflowIdUUID v7——可排序、基于时间戳天然支持按时间顺序归档与检索NodeId来自 YAML 的字符串标识时间戳UHLC保证跨机器排序Metadata 键定义在 libraries/message/src/metadata.rs包括REQUEST_ID、GOAL_ID、GOAL_STATUS、SESSION_ID等。错误处理模式应用层错误使用eyre::Result配合.context()链补充上下文NodeErrorCause枚举区分错误归因GraceDuration宽限期相关、Cascading级联错误、FailedToSpawn拉取失败、Other重启策略带指数退避可配置窗口级联错误追踪记录哪个节点导致了哪个失败这是Cascading归因的基础。测试约定分层测试矩阵变更类型测试层级位置库函数单元#[cfg(test)]同文件Coordinator/Daemon 行为集成测试binaries/coordinator/tests/CLI 命令Smoke联网tests/example-smoke.rsDataflow 功能Smoke两种模式tests/example-smoke.rsBug 修复回归测试能复现问题的层级Smoke 测试辅助函数run_smoke_test(name, yaml, timeout)—— 联网模式up/start/poll/stop/downrun_smoke_test_local(name, yaml, stop_after_secs)—— 本地模式run --stop-after。集成测试setup_integration_testing()通过DORA_TEST_WITH_INPUTS环境变量注入 JSON 输入见上文节点初始化的测试模式。构建命令速查# Build (exclude Python) 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-runtime-python --exclude dora-cli-api-python --exclude dora-examples # Single crate cargo test -p dora-core # Pre-commit (mandatory) cargo fmt --all -- --check cargo clippy --all --exclude dora-node-api-python --exclude dora-operator-api-python --exclude dora-ros2-bridge-python -- -D warnings注意Python 相关 crate 在多数构建/测试命令中被显式排除它们体积大且依赖特定环境cargo clippy必须以-D warnings通过这是提交前的强制门槛。PyO3 绑定Python 生态接入Python Node APIapis/python/node/——#[pyclass] Node实现迭代器协议Python Operator APIapis/python/operator/—— 类型转换工具Python CLI APIapis/python/cli/——build()、run()、start_runtime()使用 PyO3 0.29启用eyre、abi3-py311、multiple-pymethodsfeaturesArrow 数组通过arrow::pyarrowFFI 零拷贝传递阻塞 Rust 操作期间释放 GILpy.detach()避免阻塞 Python 事件循环。关键工程约定速查Rust edition 2024MSRV 1.85.0Workspace 版本统一为 0.2.0所有 crate 共享进程内部事件路由使用flume有界 MPSC异步运行时为tokiofull features跨机器 pub/sub 使用Zenoh零拷贝阈值4KB日志/指标走fire-and-forget策略——绝不在数据路径上阻塞相关实现见libraries/log-utils/与binaries/daemon/src/log.rs。进一步阅读数据流示例从 rust-dataflow/Rust 节点、python-dataflow/Python 节点、c-dataflow/C 节点出发覆盖三种主流语言通信模式示例service-example/、action-example/、streaming-example/容错与生命周期测试tests/dataflows/ 下的 YAML 与 tests/fault-tolerance-e2e.rs动态拓扑dynamic-add-remove/Python与 rust-dynamic-add-remove/Rust深入文档architecture.md、yaml-spec.md、api-rust.md、api-python.md、api-c.md、api-cxx.md。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考