Rust中Tokio与daemonize协同工作异常问题及解决方案

背景 #

我编写了一个基于Tokio和fern的TCP服务程序，功能是接收TCP输入并将日志同时输出到标准输出与文件。为了让程序在远程服务器后台持续运行（即使断开终端连接也能记录日志），选用了daemonize crate实现守护进程化。

初始代码 #

use std::fs::File;
use std::str;
use std::time::SystemTime;

use fern::colors::{Color, ColoredLevelConfig};

use log::{debug, info, trace, warn};

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

use daemonize::Daemonize;

fn setup_logging(verbosity: u64) -> Result<(), fern::InitError> {
    let mut base_config = fern::Dispatch::new();

    base_config = match verbosity {
        0 => {
            // Let's say we depend on something which whose "info" level messages are too
            // verbose to include in end-user output. If we don't need them,
            // let's not include them.
            base_config
                .level(log::LevelFilter::Info)
                .level_for("overly-verbose-target", log::LevelFilter::Warn)
        }
        1 => base_config
            .level(log::LevelFilter::Debug)
            .level_for("overly-verbose-target", log::LevelFilter::Info),
        2 => base_config.level(log::LevelFilter::Debug),
        _3_or_more => base_config.level(log::LevelFilter::Trace),
    };

    // Separate file config so we can include year, month and day in file logs
    let file_config = fern::Dispatch::new()
        .format(|out, message, record| {
            let colors = ColoredLevelConfig::new().debug(Color::Magenta);
            out.finish(format_args!(
                "[{} {} {}] {}",
                humantime::format_rfc3339_seconds(SystemTime::now()),
                colors.color(record.level()),
                //record.level(),
                record.target(),
                message
            ))
        })
        .chain(fern::log_file("program.log")?);

    let stdout_config = fern::Dispatch::new()
        .format(|out, message, record| {
            let colors = ColoredLevelConfig::new().debug(Color::Magenta);
            // special format for debug messages coming from our own crate.
            if record.level() > log::LevelFilter::Info && record.target() == "cmd_program" {
                out.finish(format_args!(
                    "DEBUG @ {}: {}",
                    humantime::format_rfc3339_seconds(SystemTime::now()),
                    message
                ))
            } else {
                out.finish(format_args!(
                    "[{} {} {}] {}",
                    humantime::format_rfc3339_seconds(SystemTime::now()),
                    colors.color(record.level()),
                    //record.level(),
                    record.target(),
                    message
                ))
            }
        })
        .chain(std::io::stdout());

    base_config
        .chain(file_config)
        .chain(stdout_config)
        .apply()?;

    Ok(())
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let stdout = File::create("/tmp/daemon.out").unwrap();
    let stderr = File::create("/tmp/daemon.err").unwrap();

    let daemonize = Daemonize::new()
        .pid_file("/tmp/test.pid") // Every method except `new` and `start`
        .chown_pid_file(true) // is optional, see `Daemonize` documentation
        .working_directory("/tmp") // for default behaviour.
        .user("nobody")
        .group("daemon") // Group name
        .group(2) // or group id.
        .umask(0o777) // Set umask, `0o027` by default.
        .stdout(stdout) // Redirect stdout to `/tmp/daemon.out`.
        .stderr(stderr) // Redirect stderr to `/tmp/daemon.err`.
        .privileged_action(|| "Executed before drop privileges");

    match daemonize.start() {
        Ok(_) => println!("Success, daemonized"),
        Err(e) => eprintln!("Error, {}", e),
    }

    setup_logging(3).expect("failed to initialize logging.");
    info!("Server starting...");
    let listener = TcpListener::bind("###IP:port###").await?;
    info!("TcpListener bound to ###IP:port###");
    info!("{:?}", listener.local_addr());
    loop {
        println!("Listening for connection.");
        let (mut socket, _) = listener.accept().await?;
        info!("Connection made from client: {:?}", socket.peer_addr());
        println!("Connection made from client: {:?}", socket.peer_addr());

        tokio::spawn(async move {
            let mut buf = vec![0; 1024];

            loop {
                match socket.read(&mut buf).await {
                    // Return value of `Ok(0)` signifies that the remote has
                    // closed
                    Ok(0) => {
                        info!("Client ({:?}) closed the connection", socket.peer_addr());
                        println!("Client ({:?}) closed the connection", socket.peer_addr());
                        return;
                    }
                    Ok(n) => {
                        let s = match str::from_utf8(buf.as_slice()) {
                            Ok(v) => v,
                            Err(e) => {
                                trace!("Invalid UTF-8 sequence: {}", e);
                                return;
                            }
                        };
                        print!("{}", s);
                        info!("{}", s);
                        // Copy the data back to socket
                        // if socket.write_all(&buf[..n]).await.is_err() {
                        //     // Unexpected socket error. There isn't much we can
                        //     // do here so just stop processing.
                        //     return;
                        // }
                    }
                    Err(_) => {
                        debug!("Something went wrong. Socket Error.");
                        // Unexpected socket error. There isn't much we can do
                        // here so just stop processing.
                        return;
                    }
                }
            }
        });
    }
    debug!("Program outside of loop and exiting. Did something go wrong?");
}

异常现象 #

程序执行到let (mut socket, _) = listener.accept().await?;后不再输出日志，但TCP连接可正常建立：telnet能连接并发送消息，杀死程序时telnet会提示断开。说明Tokio底层处理了TCP连接，但没有触发accept()返回的Future后续逻辑。

已知Tokio与fork存在兼容性问题，但资料显示在Tokio初始化前执行fork即可正常工作。

修复后的代码 #

//////////////////////////////////////////////////////////////
// Changes made after answer that got the code working
//////////////////////////////////////////////////////////////

fn main() -> Result<(), Box<dyn std::error::Error>> {
let stdout = File::create("/tmp/daemon.out").unwrap();
let stderr = File::create("/tmp/daemon.err").unwrap();

let daemonize = Daemonize::new()
    .pid_file("/tmp/test.pid") // Every method except `new` and `start`
    .chown_pid_file(true) // is optional, see `Daemonize` documentation
    .working_directory("/tmp") // for default behaviour.
    .user("nobody")
    .group("daemon") // Group name
    .group(2) // or group id.
    .umask(0o777) // Set umask, `0o027` by default.
    .stdout(stdout) // Redirect stdout to `/tmp/daemon.out`.
    .stderr(stderr) // Redirect stderr to `/tmp/daemon.err`.
    .privileged_action(|| "Executed before drop privileges");

match daemonize.start() {
    Ok(_) => println!("Success, daemonized"),
    Err(e) => eprintln!("Error, {}", e),
}

tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .build()
    .unwrap()
    .block_on(async {
        setup_logging(3).expect("failed to initialize logging.");
        main_functionality().await
    })
//debug!("Program outside of loop and exitting. Did something go wrong?");
}

async fn main_functionality() -> Result<(), Box<dyn 
std::error::Error>> {
info!("Server starting...");
let listener = TcpListener::bind("###IP:port###").await?;
info!("TcpListener bound to ###IP:port###");
info!("{:?}", listener.local_addr());
loop {
    println!("Listening for connection.");
    let (mut socket, _) = listener.accept().await?;
    info!("Connection made from client: {:?}", socket.peer_addr());
    println!("Connection made from client: {:?}", socket.peer_addr());

    tokio::spawn(async move {
        let mut buf = vec![0; 1024];

        loop {
            match socket.read(&mut buf).await {
                // Return value of `Ok(0)` signifies that the remote has
                // closed
                Ok(0) => {
                    info!("Client ({:?}) closed the connection", socket.peer_addr());
                    println!("Client ({:?}) closed the connection", socket.peer_addr());
                    return;
                }
                Ok(n) => {
                    let s = match str::from_utf8(buf.as_slice()) {
                        Ok(v) => v,
                        Err(e) => {
                            trace!("Invalid UTF-8 sequence: {}:[{}]", e, n);
                            return;
                        }
                    };
                    print!("{}", s);
                    info!("{}", s);
                    // Copy the data back to socket
                    // if socket.write_all(&buf[..n]).await.is_err() {
                    //     // Unexpected socket error. There isn't much we can
                    //     // do here so just stop processing.
                    //     return;
                    // }
                }
                Err(_) => {
                    debug!("Something went wrong. Socket Error.");
                    // Unexpected socket error. There isn't much we can do
                    // here so just stop processing.
                    return;
                }
            }
        }
    });
}
}

补充说明 #

修复后程序已正常运行。另外，除了daemonize crate，也可以使用systemd这类系统服务管理工具来实现后台服务的托管与管理。

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