53097


Rustの無限ループで安全にloop処理から抜ける方法は  
どの様に書けば良いですか。

[code]

    use std::{thread, time};
    
    pub struct Worker { }
    
    trait WorkerTrait {
        fn run(&self);
        fn stop(&self);
    }
    
    impl Worker {
        fn run(&self) {
            thread::spawn(move || loop {
                println!("worker1 working ");
                thread::sleep(time::Duration::from_millis(1000)); // 擬似処理.
                // thread break?
            });
        }
        fn stop(&self) {
            // to break thread.
        }
    }
    fn main() {
        let w = Worker {};
    
        w.run();
    
        thread::sleep(time::Duration::from_secs(10));
    
        // stop thread.
        w.stop();
    }
    
    
[/code]


CC BY-SA 4.0
2019-02-28T07:59:47.457
53094
2019-02-28T09:30:55.247
32285
question
rust
Rustで安全にthread loopを止める方法


こんにちは。κeenです。  
Rustでスレッドに終了メッセージを送るにはいくつか方法があります。

1つ目は `AtomicBool` を使って終了フラグを共有する方法。コードだと以下のようになります。

[code]

    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();
    }
    
[/code]

もう1つはチャネルを使ってメッセージを送る方法。  
コード例は以下です。少し真面目に書いてみましたので長くなっています。

[code]

    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");
    }
    
    
[/code]

AtomicBoolを送る方法は簡単なのが特徴です。しかし1種類のメッセージ1度だけしか送れません。  
チャネルを使う方は様々なメッセージを送れるので終了以外にもコミュニケーションを行いたい場合に有用です。しかしご覧の通りセットアップが多少手間です。  
用途に合わせてお使い下さい。  
因みにチャネルの方を本気で使いたいならcrossbeam_channelのようなクレイトもあるので併せて検討下さい。  
<https://docs.rs/crossbeam-channel/0.3.8/crossbeam_channel/>

因みに出来る限り元のコードに合わせてサンプルコードを書きましたが、`WorkerTrait` は以下のようなシグネチャの方が使いやすいと思います。

[code]

    trait WorkerTrait {
        fn run() -> Self;
        fn stop(self);
    }
    
[/code]

スレッドが走っていない状態のワーカがあると管理が面倒なので `run` でコンストラクタを兼ね、やはり終了したあとのワーカが残ると管理が面倒なので
`stop` で `self` を消費してしまった方が良いと思います。もちろん、ワーカを使いまわしたいなどの要件があるならその限りではありませんが。


CC BY-SA 4.0
2019-02-28T09:30:55.247
53097
2019-02-28T09:30:55.247
22816
53094
answer