Rust でカメラ映像やセンサーデータなど、常に最新の情報だけを処理したい場合、どのようなチャネル設計が適切でしょう？
例えば、30ms ごとに新しいフレームが生成されるが、処理には 300ms かかるようなケースを考えてみます。処理スレッドが追いつかない場合、「古いデータは捨てて、常に最新のデータだけを処理したい」というのは自然な要求です。
Rust における「最新データ優先」の並行処理パターンについて、複数の実装アプローチを比較します。「古いデータを上書きするチャネル」の実装方法に焦点を当てます。
想定する条件
- カメラキャプチャスレッドが約30msごとに画像を取得している
- 処理スレッドは1フレームの処理に約300msかかる
- 複数の処理スレッドを使うが、全フレームを処理できるほど多くのスレッドは用意できない
- フレームをスキップすることは許容できるが、処理スレッドは常に「最新の」フレームを処理したい
チャネルが満杯になったとき、送信側がブロックされては要求を満たせません。古いデータを上書きするようなチャネルが必要となります。
実装アプローチ
1. tokio::sync::watch - 単一値の監視
Tokio の watch チャネルは、まさにこの「最新値のみを保持する」パターンのために設計されています。
use tokio::sync::watch;
fn main() {
// 送信者と受信者を作成
let (tx, rx) = watch::channel(Frame::default());
// 送信者側（カメラキャプチャスレッド）
std::thread::spawn(move || {
loop {
let new_frame = capture_frame();
// 常に最新の値で上書き
let _ = tx.send(new_frame);
std::thread::sleep(std::time::Duration::from_millis(30));
}
});
// 受信者側（処理スレッド）
std::thread::spawn(move || {
let mut receiver = rx.clone();
loop {
// 変更を待ち、最新の値を取得
if receiver.changed().await.is_ok() {
let frame = receiver.borrow_and_update().clone();
process_frame(frame); // 約300ms
}
}
});
}
特徴:
- シンプルな実装
- 単一の値のみ保持する（最新のフレームだけを保持）
- 複数の受信者が同じ最新値を受け取れる（MPMC）
- 送信操作は非ブロッキング
- 古い値は新しい値によって自動的に上書きされる
ただし、watch チャネルは単一の値しか保持できないため、複数の最新フレームを保持したい場合は別のアプローチが必要です。
2. tokio::sync::broadcast - 設定したバッファサイズまでの値を保持
より柔軟なアプローチとしては、Tokio の broadcast チャネルがあります。
use tokio::sync::broadcast;
fn main() {
// キャパシティを持つチャネルを作成
let (tx, _) = broadcast::channel(10); // 10フレームまでバッファリング
// 送信者側
let tx_clone = tx.clone();
std::thread::spawn(move || {
loop {
let new_frame = capture_frame();
// バッファがいっぱいの場合、送信は失敗せず古いメッセージが削除される
let _ = tx_clone.send(new_frame);
std::thread::sleep(std::time::Duration::from_millis(30));
}
});
// 受信者側
let mut rx = tx.subscribe();
std::thread::spawn(move || {
loop {
match rx.recv().await {
Ok(frame) => {
process_frame(frame); // 処理（約300ms）
}
Err(broadcast::error::RecvError::Lagged(n)) => {
println!("{}フレームスキップしました", n);
}
Err(_) => break,
}
}
});
}
特徴:
- 指定したキャパシティまでメッセージをバッファリング
- バッファがいっぱいになると、古いメッセージが自動的に削除される
- 複数の受信者がそれぞれ独立してメッセージを消費できる
- 受信者が処理に追いつけないと
Laggedエラーを受け取る - どのメッセージが処理されるかをより細かく制御できる
3. crossbeam + semaphore
crossbeam チャネルと semaphore を組み合わせた実装も可能です。
use crossbeam_channel::{bounded, select};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
struct Semaphore {
count: AtomicUsize,
}
impl Semaphore {
fn new(count: usize) -> Self {
Self { count: AtomicUsize::new(count) }
}
fn try_acquire(&self) -> bool {
let mut current = self.count.load(Ordering::SeqCst);
while current > 0 {
match self.count.compare_exchange(
current, current - 1, Ordering::SeqCst, Ordering::SeqCst
) {
Ok(_) => return true,
Err(val) => current = val,
}
}
false
}
fn release(&self) {
self.count.fetch_add(1, Ordering::SeqCst);
}
}
fn main() {
// 処理スレッド数に基づいたセマフォを作成
let semaphore = Arc::new(Semaphore::new(3)); // 同時に3つのフレームまで処理可能
// チャネルを作成
let (tx, rx) = bounded(10); // バッファサイズは適宜調整
let rx = Arc::new(Mutex::new(rx));
// 送信者側
std::thread::spawn(move || {
loop {
let new_frame = capture_frame();
// セマフォで処理スロットが空いているときだけ送信を試みる
if semaphore.try_acquire() {
if tx.send(new_frame).is_err() {
// 送信エラー時はセマフォを解放
semaphore.release();
break;
}
} else {
// 処理スロットがない場合は古いフレームをスキップ
println!("処理スレッドがビジー状態: フレームをスキップします");
}
std::thread::sleep(std::time::Duration::from_millis(30));
}
});
// 複数の処理スレッド
for _ in 0..3 {
let rx_clone = Arc::clone(&rx);
let semaphore_clone = Arc::clone(&semaphore);
std::thread::spawn(move || {
loop {
let frame = {
let rx_guard = rx_clone.lock().unwrap();
match rx_guard.recv() {
Ok(frame) => frame,
Err(_) => break,
}
};
// フレーム処理
process_frame(frame);
// 処理が終わったらセマフォを解放
semaphore_clone.release();
}
});
}
}
特徴:
- 明示的に古い値を捨てる処理が必要
- 他のアプローチに比べて制御が複雑になる可能性がある
4. ringbuf - リングバッファを使用
ringbuf クレートを使用したリングバッファベースの実装も考えられます。
use ringbuf::{HeapRb, Rb, Consumer, Producer};
fn main() {
// リングバッファを作成
let rb = HeapRb::<Frame>::new(1); // バッファサイズ1
let (mut prod, mut cons) = rb.split();
// 送信者側
std::thread::spawn(move || {
loop {
let new_frame = capture_frame();
// 古い要素を上書き
let _ = prod.push_overwrite(new_frame);
std::thread::sleep(std::time::Duration::from_millis(30));
}
});
// 受信者側
std::thread::spawn(move || {
loop {
if let Some(frame) = cons.pop() {
process_frame(frame); // 処理（約300ms）
} else {
std::thread::sleep(std::time::Duration::from_millis(10));
}
}
});
}
特徴:
-
push_overwriteで古い要素を簡単に上書き可能 - リングバッファの操作にはロックが必要
- 明示的なバッファサイズ管理が可能
- シンプルな API
ただし、push_overwrite にはロックが必要なため、高速な処理が求められる場合には注意が必要です。
5. リングバッファ + Mutex + CondVar
リングバッファと同期プリミティブを組み合わせる方法です。
use std::sync::{Arc, Mutex, Condvar};
use std::collections::VecDeque;
struct OverwritingChannel<T> {
buffer: Mutex<VecDeque<T>>,
capacity: usize,
not_empty: Condvar,
}
impl<T> OverwritingChannel<T> {
fn new(capacity: usize) -> Self {
Self {
buffer: Mutex::new(VecDeque::with_capacity(capacity)),
capacity,
not_empty: Condvar::new(),
}
}
fn send(&self, item: T) {
let mut buffer = self.buffer.lock().unwrap();
if buffer.len() == self.capacity {
buffer.pop_front(); // 最も古い要素を削除
}
buffer.push_back(item);
self.not_empty.notify_one();
}
fn recv(&self) -> T {
let mut buffer = self.buffer.lock().unwrap();
while buffer.is_empty() {
buffer = self.not_empty.wait(buffer).unwrap();
}
buffer.pop_front().unwrap()
}
}
特徴:
- 要件に完全に合わせたカスタム実装が可能
- 細かい挙動を制御できる
- 実装が複雑
- バグを作り込みそう
- パフォーマンスの最適化が難しそう
まとめ
Rust で「古いデータを上書きするチャネル」を実装する以下のアプローチを紹介しました：
- tokio::sync::watch - 最も簡単で単一値のみを扱う場合
- tokio::sync::broadcast - 複数のメッセージをバッファリングする場合
- crossbeam + semaphore - 非同期コードを避けたい場合
- ringbuf - リングバッファベースのシンプルな実装
- 自作実装 - 特殊な要件に対応する必要がある場合
要件を見直して、実装はシンプルにするのがいいかなとおもいました