dora-rs

"dora-rs 1.1 を Rust で触る ― ノード・名前の対応・queue_policy を実験で確かめる" はじめに dora-rs は、ロボットや AI のアプリを「ノードをつないだデータフロー」として組むためのフレームワークです。ノードは Rust や Python で書け、つなぎ方は YAML で宣言します。 Rust でロボット制御の閉ループ（状態を読む → 指令を出す → 状態が変わる → …）を組みたくて、まずは dora の基本を手で確かめることにしました。この記事は、その過程で分かったこと・ハマったことのまとめです。 ノードはどういう形で書くのか YAML とコードで、名前はどう対応するのか 受け取る側の処理が遅いとき、queue_policy で何が変わるのか（実験あり） 特に最後の queue_policy は、予想を立てて外し、ログで確かめ直しました。 環境とログの取り方 dora-cli 1.1.0 / dora-node-api 1.1.0 rustc 1.97.0 WSL2（Ubuntu） cargo install dora-cli dora new --lang rust # ひな形の作成 dora run dataflow.yml # coordinator / daemon を別に起動しなくてよい dora run は YAML の build: を自動で実行してくれるので、事前に cargo build しなくても動きます。 出力をファイルにも残したいときは、標準エラーも合流させて tee します。 dora run dataflow.yml 2>&1 | tee datas/log.txt ! WSL2 では WARN ... failed to update HLC ... exceeding delta 500ms という警告が大量に出ることがあります。ノード間の時計のずれに関する警告で、今回の実験の動作には影響しませんでした。 ノードの形：待つ → 処理 → 送る dora new --lang rust で作ったひな形を少し変えた、送る側のノードです。 talker_1/src/main.rs use dora_node_api::{dora_core::config::DataId, DoraNode, Event, IntoArrow}; use std::error::Error; fn main() -> Result<(), Box<dyn Error>> { let (mut node, mut events) = DoraNode::init_from_env()?; let mut counter = 0u32; while let Some(event) = events.recv() { // ① イベントが来るまで待つ match event { Event::Input { id, metadata, data: _ } => match id.as_str() { "tick" => { // ② どの入力かで処理を分ける counter += 1; node.send_output( // ③ 出力を送る DataId::from("speech".to_owned()), metadata.parameters, counter.into_arrow(), )?; } _ => {} }, _ => {} } } Ok(()) } どのノードも、init → recv で待つ → match で振り分ける → send_output の繰り返しです。 ノードは events.recv() で何かが届くまで止まっています。では、誰からも入力を受け取らない「送るだけのノード」はどうやって動くのでしょうか。そこで使うのが、dora が持っているタイマーです。 inputs: tick: dora/timer/millis/10 # 10ms ごと（100Hz）にイベントが届く タイマーは、ノードを一定間隔で起こす目覚まし時計です。制御ノードを作るときには、このタイマーの間隔がそのまま制御周期になります。 書き方 間隔 頻度 dora/timer/millis/10 0.01 秒 100 Hz dora/timer/millis/100 0.1 秒 10 Hz dora/timer/secs/1 1 秒 1 Hz なお tick という名前に特別な意味はありません。次の章で書くとおり、YAML の左側のキーなので、clock: にしてもコード側を合わせれば動きます。 名前の対応：YAML とコード 送る側 2 つ、受け取る側 1 つの構成にしてみました。 dataflow.yml nodes: - id: talker_1 build: cargo build -p talker_1 path: target/debug/talker_1 inputs: tick: dora/timer/millis/10 outputs: - speech - id: talker_2 build: cargo build -p talker_2 path: target/debug/talker_2 inputs: tick: dora/timer/secs/2 outputs: - speech - id: listener_1 build: cargo build -p listener_1 path: target/debug/listener_1 inputs: speech-1: talker_1/speech speech-2: talker_2/speech 名前は 3 か所で対応しています。 talker_1 のコード YAML listener_1 のコード DataId "speech" ──→ speech-1: talker_1/speech ──→ match id.as_str() { "speech-1" => ... } ^^^^^^^^ ^^^^^^^^ ^^^^^^ (2) (3) (1) outputs: の名前（speech）は、送る側のコードの DataId と一致させる inputs: の左のキー（speech-1）は、受け取る側のコードの match と一致させる talker_1/ は YAML の id:（ノード名）。crate 名ではない ハマり：受け取っているのに表示されない 最初、listener のコードはひな形のままで、"speech" しか拾っていませんでした。 "speech" => { /* 表示する */ } other => println!("Received input {other}"), YAML では speech-1: という名前で受け取っているので、届いたデータは other => に落ちて、Received input speech-1`` としか表示されません。受け取る側の id は送る側の出力名ではなく、YAML の左のキーです。直すのは受け取る側の match です。 名前の食い違いはコンパイルエラーにも実行時エラーにもならず、黙って無視されます。other => に何か表示させておくと、気づきやすくなります。 型：受け取る側も型を指定する 送る側を文字列から数値に変えたところ、受け取る側（文字列のまま）がすぐに落ちました。 // 送る側 node.send_output(..., 1.0f32.into_arrow())?; // 受け取る側（変更し忘れ） let message: &str = (&data).try_into()?; listener_1: Error: not a string array (expected Utf8, LargeUtf8 or Utf8View, got Float32) ノード間を流れるのは Arrow の配列です。Rust のコードからはどれも同じ ArrowData 型に見えるので、受け取る側で let message: f32 = (&data).try_into()?; のように変換先の型を指定します。 注意したいのは、この食い違いがコンパイルでは見つからないことです。 YAML が決めているのは「どこからどこへ」だけで、型は書いていない 送る側と受け取る側は別の crate なので、コンパイラは両者の型を突き合わせられない 実行前に見つけたいなら、送るメッセージの型を共通のライブラリ crate に定義して、両側でそれを使うのがよさそうです（これは次回試します）。 queue_policy の実験 受け取る側の処理が遅いと、入力はキューに溜まっていきます。溢れたときの振る舞いを決めるのが queue_policy です。 既定値 ドキュメントに見当たらなかったので、ソース（dora-message-1.1.0/src/config.rs）を読みました。 pub const DEFAULT_QUEUE_SIZE: usize = 10; pub enum QueuePolicy { /// Drop the oldest queued message when the queue is full (default). #[default] DropOldest, /// Buffer up to 10x queue_size without dropping. Drops with ERROR log at hard cap. Backpressure, } pub fn effective_cap(&self, queue_size: usize) -> usize { match self { Self::DropOldest => queue_size.max(1), Self::Backpressure => queue_size.saturating_mul(10).max(100), } } 何も書かなければ drop_oldest・queue_size: 10 1.1.0 の backpressure は、送る側を待たせるのではなく、queue_size の 10 倍（最低 100）まで溜め、超えたら捨てる YAML では、入力を長い書き方にして指定します。 - id: listener_1 inputs: speech-1: source: talker_1/speech queue_size: 1 queue_policy: drop_oldest # または backpressure speech-2: talker_2/speech 実験のしかた talker_1 は 100 Hz で連番（1, 2, 3, …）を送る。番号の抜けを見れば、何が捨てられたかが分かる listener_1 は、talker_2 の入力（2 秒ごと）を受け取るたびに 10 秒 sleep する。その間、talker_1 の値はキューに溜まる listener_1/src/main.rs（抜粋） "speech-1" => { let message: u32 = (&data).try_into()?; println!("I heard: {} from {}", message, id); } "speech-2" => { let message: &str = (&data).try_into()?; println!("I heard: {message} from {id}さん"); thread::sleep(Duration::from_secs(10)); } queue_size: 1 に固定して、queue_policy だけを変えて比べました。 結果 ログから、listener_1 に届いた番号を並べるとこうなりました。 起きた直後 そのあと drop_oldest: 64 → 1005, 1006, 1007, … 最新の値にすぐ追いつく backpressure: 2 … 66 → 902 … 1001 → … 古い値を約 100 個処理してから追いつく 結果 drop_oldest 寝ている間の値はほぼ捨てられ、起きるとすぐ最新値（1005）から処理する backpressure 902〜1001 のちょうど約 100 個を溜めて古い順に処理し、それより前（67〜901）は捨てる backpressure で溜まった数（約 100）は、ソースの effective_cap（1 * 10 と 100 の大きいほう）と一致しました。一方、コメントにある「上限で ERROR ログ」は、tee したログにも out/ 以下のログにも見当たりませんでした。 どちらの policy でも、起きた直後に少し古い値（64 や 2〜66）が出ています。dora のキューとは別に、ノード側にも受け取り用のバッファがあるのではないかと推測していますが、確かめてはいません。 予想との答え合わせ 予想 結果 drop_oldest 最新だけ ほぼそのとおり backpressure 全部順番に 上限（100 個）までは順番。超えたら捨てる listener を「関節角を受け取って指令を出す controller」だと考えます。backpressure だと、起きたあとに 10 秒近く前の角度を 100 個以上処理してから、やっと今に追いつきます。古い角度に対して指令を出しても、今のアームの位置とはずれています。そのずれを直そうとすれば、指令はかえっておかしくなります。制御に必要なのは「今の状態に合わせた指令」なので、drop_oldest（queue_size: 1）が向いています。 逆に、1 つも抜けてほしくない記録（ロガーなど）には backpressure が候補になります。queue_policy は入力ごとに指定できるので、同じデータフローの中で使い分けられます。 plant ──(状態)──> controller drop_oldest, queue_size: 1 ← 今の値だけ欲しい └─(状態)──> logger backpressure ← 全部欲しい まとめ dora のノードは「待つ → 処理 → 送る」の繰り返し。入力のないノードはタイマーで起こす 名前は、outputs: ↔ 送る側の DataId、inputs: の左キー ↔ 受け取る側の match で対応する。食い違いは黙って無視される 型の食い違いは実行時まで分からない queue_policy の既定は drop_oldest（queue_size: 10）。1.1.0 の backpressure は「上限まで溜めて、超えたら捨てる」 制御の入力には drop_oldest、全件必要な記録には backpressure