Rust中闭包使用std::sync::mpsc发送端时通道意外关闭问题

问题现象 #

运行监听文件变更的代码后，接收端输出错误：

Error while watching config file: receiving on a closed channel

预期发送端tx被移入回调闭包后通道不会关闭，但实际通道被提前关闭。

原代码 #

Rust 代码 #

use std::{path::PathBuf, thread};

use log::{error, info};
use notify::Watcher;

fn main() {
    env_logger::init();

    let (tx, rx) = std::sync::mpsc::channel();

    thread::spawn({
        let config_path = PathBuf::from("config.toml");

        move || {
            let mut watcher =
                notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| {
                    match res {
                        Ok(event) => match event.kind {
                            notify::event::EventKind::Modify(notify::event::ModifyKind::Data(
                                notify::event::DataChange::Any,
                            )) => {
                                info!("Reloading config file");
                                tx.send(true).ok();
                            }
                            _ => (),
                        },
                        Err(e) => {
                            error!("Notify Error: {}", e);
                        }
                    }
                })
                .unwrap();
            watcher.watch(&config_path, notify::RecursiveMode::NonRecursive)
        }
    });

    loop {
        match rx.recv() {
            Ok(_) => {
                // reload config
            }
            Err(e) => {
                error!("Error while watching config file: {}", e);
                break;
            }
        }
    }
}

Cargo.toml #

# Cargo.toml

[dependencies]
env_logger = "0.10.0"
log = "0.4.19"
notify = "6.0.1"

问题原因 #

线程闭包执行到watcher.watch(...)后就结束了，局部变量watcher随即被销毁：

1. notify的Watcher在被销毁（drop）时，会停止文件监听并释放相关资源；
2. 持有发送端tx的回调闭包是watcher的一部分，watcher被销毁后，tx也会被销毁；
3. mpsc通道的规则是：当所有发送端都被销毁时，通道会自动关闭，此时接收端rx.recv()就会返回RecvError，也就是你看到的报错。

修复方案 #

需要让watcher在线程内持续存在，避免被提前销毁。可以在调用watch后添加代码让线程保持运行状态，比如用thread::park()挂起线程：

修改后的线程闭包代码：

move || {
    let mut watcher =
        notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| {
            match res {
                Ok(event) => match event.kind {
                    notify::event::EventKind::Modify(notify::event::ModifyKind::Data(
                        notify::event::DataChange::Any,
                    )) => {
                        info!("Reloading config file");
                        tx.send(true).ok();
                    }
                    _ => (),
                },
                Err(e) => {
                    error!("Notify Error: {}", e);
                }
            }
        })
        .unwrap();

    // 处理watch的错误，避免无声失败
    if let Err(e) = watcher.watch(&config_path, notify::RecursiveMode::NonRecursive) {
        error!("Failed to watch config file: {}", e);
        return;
    }

    // 挂起线程，让watcher持续存在
    std::thread::park();
}

说明 #

• thread::park()会让线程进入休眠状态，直到被其他线程唤醒，这样watcher作为线程内的局部变量会一直存活，不会被销毁；
• 额外处理了watcher.watch()的错误，避免因监听失败导致线程意外退出；
• 此时tx会被回调闭包持有，而回调闭包被watcher持有，只要watcher存活，tx就不会被销毁，通道也就不会关闭。

内容的提问来源于stack exchange，提问作者tkr3