こんにちは。κeenです。
Rustでスレッドに終了メッセージを送るにはいくつか方法があります。
1つ目は AtomicBool
を使って終了フラグを共有する方法。コードだと以下のようになります。
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::{thread, time};
pub struct Worker {
to_stop: Arc<AtomicBool>,
}
trait WorkerTrait {
fn run(&self);
fn stop(&self);
}
impl Worker {
fn new() -> Self {
Worker {
to_stop: Arc::new(AtomicBool::new(false)),
}
}
}
impl WorkerTrait for Worker {
fn run(&self) {
let to_stop = Arc::clone(&self.to_stop);
thread::spawn(move || loop {
if to_stop.load(Ordering::Relaxed) {
break;
}
println!("worker1 working ");
thread::sleep(time::Duration::from_millis(1000)); // 擬似処理.
// thread break?
});
}
fn stop(&self) {
self.to_stop.store(true, Ordering::Relaxed)
}
}
fn main() {
let w = Worker::new();
w.run();
thread::sleep(time::Duration::from_secs(10));
// stop thread.
w.stop();
}
もう1つはチャネルを使ってメッセージを送る方法。
コード例は以下です。少し真面目に書いてみましたので長くなっています。
use std::any::Any;
use std::sync::mpsc::{channel, SendError, Sender, TryRecvError};
use std::thread::JoinHandle;
use std::{thread, time};
pub struct Worker {
//スレッドに終了メッセージを送信するチャネル
control: Option<Sender<Message>>,
// threadのハンドル
handle: Option<JoinHandle<()>>,
}
#[derive(Debug)]
enum WorkerError {
Channel(SendError<Message>),
Thread(Box<dyn Any + Send + 'static>),
ThreadNotStarted,
}
trait WorkerTrait {
type Error;
fn run(&mut self);
fn stop(&mut self) -> Result<(), Self::Error>;
}
// スレッドに対するメッセージ。他に必要なら増やす。
enum Message {
Stop,
}
impl Worker {
fn new() -> Self {
Self {
control: None,
handle: None,
}
}
}
impl WorkerTrait for Worker {
type Error = WorkerError;
fn run(&mut self) {
let (tx, rx) = channel();
let handle = thread::spawn(move || loop {
match rx.try_recv() {
// メッセージが来るか、チャネルが閉じられたら(=Workerが破棄されたら)スレッドを止める
Ok(Message::Stop) | Err(TryRecvError::Disconnected) => break,
Err(TryRecvError::Empty) => {
println!("worker1 working ");
thread::sleep(time::Duration::from_millis(1000)); // 擬似処理.
// thread break?
}
}
});
self.control = Some(tx);
self.handle = Some(handle);
}
fn stop(&mut self) -> Result<(), Self::Error> {
if let (Some(control), Some(handle)) = (self.control.take(), self.handle.take()) {
// スレッドに終了メッセージを送り、
control.send(Message::Stop).map_err(WorkerError::Channel)?;
// 終了を待ち合わせる
handle.join().map_err(WorkerError::Thread)?;
Ok(())
} else {
Err(WorkerError::ThreadNotStarted)
}
}
}
fn main() {
let mut w = Worker::new();
w.run();
thread::sleep(time::Duration::from_secs(10));
// stop thread.
w.stop().expect("stopping thread failed");
}
AtomicBoolを送る方法は簡単なのが特徴です。しかし1種類のメッセージ1度だけしか送れません。
チャネルを使う方は様々なメッセージを送れるので終了以外にもコミュニケーションを行いたい場合に有用です。しかしご覧の通りセットアップが多少手間です。
用途に合わせてお使い下さい。
因みにチャネルの方を本気で使いたいならcrossbeam_channelのようなクレイトもあるので併せて検討下さい。
https://docs.rs/crossbeam-channel/0.3.8/crossbeam_channel/
因みに出来る限り元のコードに合わせてサンプルコードを書きましたが、WorkerTrait
は以下のようなシグネチャの方が使いやすいと思います。
trait WorkerTrait {
fn run() -> Self;
fn stop(self);
}
スレッドが走っていない状態のワーカがあると管理が面倒なので run
でコンストラクタを兼ね、やはり終了したあとのワーカが残ると管理が面倒なので stop
で self
を消費してしまった方が良いと思います。もちろん、ワーカを使いまわしたいなどの要件があるならその限りではありませんが。