如何在 Rust 中干净利落地打破 tokio-core 事件循环和 futures::Stream

Question

我正在涉足 tokio-core，并且可以弄清楚如何生成事件循环。但是有两件事我不确定 - 如何优雅地退出事件循环以及如何退出在事件循环内运行的流。例如，考虑这段简单的代码，它在事件循环中生成两个侦听器并等待另一个线程指示退出条件：

extern crate tokio_core;
extern crate futures;

use tokio_core::reactor::Core;
use futures::sync::mpsc::unbounded;
use tokio_core::net::TcpListener;
use std::net::SocketAddr;
use std::str::FromStr;
use futures::{Stream, Future};
use std::thread;
use std::time::Duration;
use std::sync::mpsc::channel;

fn main() {
    let (get_tx, get_rx) = channel();

    let j = thread::spawn(move || {
        let mut core = Core::new().unwrap();
        let (tx, rx) = unbounded();
        get_tx.send(tx).unwrap(); // <<<<<<<<<<<<<<< (1)

        // Listener-0
        {
            let l = TcpListener::bind(&SocketAddr::from_str("127.0.0.1:44444").unwrap(),
                                      &core.handle())
                .unwrap();

            let fe = l.incoming()
                .for_each(|(_sock, peer)| {
                    println!("Accepted from {}", peer);
                    Ok(())
                })
                .map_err(|e| println!("----- {:?}", e));

            core.handle().spawn(fe);
        }

        // Listener1
        {
            let l = TcpListener::bind(&SocketAddr::from_str("127.0.0.1:55555").unwrap(),
                                      &core.handle())
                .unwrap();

            let fe = l.incoming()
                .for_each(|(_sock, peer)| {
                    println!("Accepted from {}", peer);
                    Ok(())
                })
                .map_err(|e| println!("----- {:?}", e));

            core.handle().spawn(fe);
        }

        let work = rx.for_each(|v| {
            if v {
                // (3) I want to shut down listener-0 above the release the resources
                Ok(())
            } else {
                Err(()) // <<<<<<<<<<<<<<< (2)

            }
        });

        let _ = core.run(work);
        println!("Exiting event loop thread");
    });

    let tx = get_rx.recv().unwrap();

    thread::sleep(Duration::from_secs(2));
    println!("Want to terminate listener-0"); // <<<<<< (3)
    tx.send(true).unwrap();

    thread::sleep(Duration::from_secs(2));
    println!("Want to exit event loop");
    tx.send(false).unwrap();

    j.join().unwrap();
}

所以说在主线程睡眠后我想要事件循环线程的干净退出。目前我向事件循环发送一些东西以使其退出并释放线程。

但是，(1) 和 (2) 都感觉很老套——我是在强制一个错误作为退出条件。我的问题是：

1) 我做得对吗？如果不是，那么优雅退出事件循环线程的正确方法是什么。

2) 我不知道该怎么做(3) - 即在外部指示条件以关闭 listener-0 并释放它的所有资源。我如何做到这一点？

Answer 1

事件循环 (core) 不再被转动（例如被run()）或被遗忘（drop()ed）。没有同步退出。 core.run() 在传递给它的 Future 完成时返回并停止转动循环。

Stream 通过产生 None 完成（在下面的代码中标记为 (3)）。 当例如TCP 连接关闭，Stream 表示它完成，反之亦然。

extern crate tokio_core;
extern crate futures;

use tokio_core::reactor::Core;
use futures::sync::mpsc::unbounded;
use tokio_core::net::TcpListener;
use std::net::SocketAddr;
use std::str::FromStr;
use futures::{Async, Stream, Future, Poll};
use std::thread;
use std::time::Duration;

struct CompletionPact<S, C>
    where S: Stream,
          C: Stream, 
{
    stream: S,
    completer: C,
}

fn stream_completion_pact<S, C>(s: S, c: C) -> CompletionPact<S, C>
    where S: Stream,
          C: Stream,
{
    CompletionPact {
        stream: s,
        completer: c,
    }
}

impl<S, C> Stream for CompletionPact<S, C>
    where S: Stream,
          C: Stream,
{
    type Item = S::Item;
    type Error = S::Error;

    fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
        match self.completer.poll() {
            Ok(Async::Ready(None)) |
            Err(_) |
            Ok(Async::Ready(Some(_))) => {
                // We are done, forget us
                Ok(Async::Ready(None)) // <<<<<< (3)
            },
            Ok(Async::NotReady) => {
                self.stream.poll()
            },
        }
    }
}

fn main() {
    // unbounded() is the equivalent of a Stream made from a channel()
    // directly create it in this thread instead of receiving a Sender
    let (tx, rx) = unbounded::<()>();
    // A second one to cause forgetting the listener
    let (l0tx, l0rx) = unbounded::<()>();

    let j = thread::spawn(move || {
        let mut core = Core::new().unwrap();

        // Listener-0
        {
            let l = TcpListener::bind(
                    &SocketAddr::from_str("127.0.0.1:44444").unwrap(),
                    &core.handle())
                .unwrap();

            // wrap the Stream of incoming connections (which usually doesn't
            // complete) into a Stream that completes when the
            // other side is drop()ed or sent on
            let fe = stream_completion_pact(l.incoming(), l0rx)
                .for_each(|(_sock, peer)| {
                    println!("Accepted from {}", peer);
                    Ok(())
                })
                .map_err(|e| println!("----- {:?}", e));

            core.handle().spawn(fe);
        }

        // Listener1
        {
            let l = TcpListener::bind(
                    &SocketAddr::from_str("127.0.0.1:55555").unwrap(),
                    &core.handle())
                .unwrap();

            let fe = l.incoming()
                .for_each(|(_sock, peer)| {
                    println!("Accepted from {}", peer);
                    Ok(())
                })
                .map_err(|e| println!("----- {:?}", e));

            core.handle().spawn(fe);
        }

        let _ = core.run(rx.into_future());
        println!("Exiting event loop thread");
    });

    thread::sleep(Duration::from_secs(2));
    println!("Want to terminate listener-0");
    // A drop() will result in the rx side Stream being completed,
    // which is indicated by Ok(Async::Ready(None)).
    // Our wrapper behaves the same when something is received.
    // When the event loop encounters a
    // Stream that is complete it forgets about it. Which propagates to a
    // drop() that close()es the file descriptor, which closes the port if
    // nothing else uses it.
    l0tx.send(()).unwrap(); // alternatively: drop(l0tx);
    // Note that this is async and is only the signal
    // that starts the forgetting.

    thread::sleep(Duration::from_secs(2));
    println!("Want to exit event loop");
    // Same concept. The reception or drop() will cause Stream completion.
    // A completed Future will cause run() to return.
    tx.send(()).unwrap();

    j.join().unwrap();
}

Answer 2

我通过oneshot 频道实现了正常关机。

诀窍是同时使用oneshot 通道来取消tcp 侦听器，并在两个期货上使用select!。请注意，我在下面的示例中使用的是 tokio 0.2 和 futures 0.3。

use futures::channel::oneshot;
use futures::{FutureExt, StreamExt};
use std::thread;
use tokio::net::TcpListener;

pub struct ServerHandle {
    // This is the thread in which the server will block
    thread: thread::JoinHandle<()>,

    // This switch can be used to trigger shutdown of the server.
    kill_switch: oneshot::Sender<()>,
}

impl ServerHandle {
    pub fn stop(self) {
        self.kill_switch.send(()).unwrap();
        self.thread.join().unwrap();
    }
}

pub fn run_server() -> ServerHandle {
    let (kill_switch, kill_switch_receiver) = oneshot::channel::<()>();

    let thread = thread::spawn(move || {
        info!("Server thread begun!!!");
        let mut runtime = tokio::runtime::Builder::new()
            .basic_scheduler()
            .enable_all()
            .thread_name("Tokio-server-thread")
            .build()
            .unwrap();

        runtime.block_on(async {
            server_prog(kill_switch_receiver).await.unwrap();
        });

        info!("Server finished!!!");
    });

    ServerHandle {
        thread,
        kill_switch,
    }
}

async fn server_prog(kill_switch_receiver: oneshot::Receiver<()>) -> std::io::Result<()> {
    let addr = "127.0.0.1:12345";
    let addr: std::net::SocketAddr = addr.parse().unwrap();
    let mut listener = TcpListener::bind(&addr).await?;
    let mut kill_switch_receiver = kill_switch_receiver.fuse();
    let mut incoming = listener.incoming().fuse();

    loop {
        futures::select! {
            x = kill_switch_receiver => {
                break;
            },
            optional_new_client = incoming.next() => {
                if let Some(new_client) = optional_new_client {
                    let peer_socket = new_client?;
                    info!("Client connected!");
                    let peer = process_client(peer_socket, db.clone());
                    peers.lock().unwrap().push(peer);
                } else {
                    info!("No more incoming connections.");
                    break;
                }
            },
        };
    }
    Ok(())
}

希望这对其他人（或未来的我）有所帮助。

我的代码在这里：

https://github.com/windelbouwman/lognplot/blob/master/lognplot/src/server/server.rs