向 mio 注册频道

Posted 2023-02-16

【中文标题】向 mio 注册频道

【英文标题】：Register a channel with mio

【发布时间】：2019-04-26 06:59:50

【问题描述】：

在旧版本的 mio 文档中，我找到了 mio::channel，它似乎用于创建一个实现 EventedFd 的通道，该通道可以使用 Poll 注册。 我还在 reddit 上看到这已被更改为有利于其他内容，但我无法弄清楚注册频道的新方法是什么。

目前使用 mio 等待频道（std::sync::mpsc 或其他）的当前方式是什么？我在谷歌上只能找到旧版本的 mio 文档的链接。 谢谢！

当我遇到同样的问题时，我希望我提供一些代码以使问题更容易理解：

use std::io;
use std::net::*; //TcpListener,TcpStream,IpAddr,Ipv4Addr,SocketAddr;
use std::thread::*;
use std::sync::mpsc::*; //Receiver,SyncSender,Sender,channel,sync_channel;

fn poll_quit( rx : &Receiver::<u8> ) -> bool 
    match rx.try_recv() 
        Ok(_) => true,
        Err(TryRecvError::Empty) => false,
        Err(TryRecvError::Disconnected) => true
    


fn guard_external_tcp_port( rx : Receiver::<u8>) -> () 
    let listener = TcpListener::bind("127.0.0.1:8384").expect("tcp guard - bind failed!");
    listener.set_nonblocking(true).expect("cannot set tcp listener to nonblocking!");
    while false == poll_quit(&rx) 
        match listener.accept() 
            Ok((_s,pa)) => 
                println!("tcp client connected:  - closing it down."
                    , pa);
            
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => 
                //wait_for_fd();
                continue;
            
            Err(e) => panic!("encountered IO error: ", e),

        
    


fn start_tcpguard() -> (JoinHandle<()>,SyncSender::<u8>) 
    let (tx,rx) = sync_channel::<u8>(0);
    let thrd = spawn(move || 
        guard_external_tcp_port(rx);
    );
    (thrd,tx)

到目前为止，我尝试不使用 mio，但问题很明显：您希望阻塞直到关闭通道消息到达或侦听器套接字发出信号。有足够的动力转向mio。除非 mio 无法等待Receiver。

【问题讨论】：

让服务器在线程中运行并使用通道来通知线程关闭是一个主要用例。是的，您可以轮询Receiver，但除非您将TcpListener 设置为非阻塞并轮询它，否则您将无法彻底关闭。所以，一个人必须mio - 等待一个频道和一个听众。

【参考方案1】：

现在还有另一个 mio crate，它具有与频道相关的功能：

mio-extras

所以，您所要做的就是将该板条箱添加到您的 cargo.toml [dependencies]。 然后，在您的main.rs 中添加extern crate mio_extras，您应该已经准备好使用它了。

所以，如果我的代码 sn-p 我添加到问题中，它最终看起来像这样：

extern crate mio;
extern crate mio_extras;

use std::io;
use mio::*;
use mio::net::*;
use mio_extras::channel::*;
use std::thread::*;

fn guard_external_tcp_port( rx : mio_extras::channel::Receiver::<u8>) -> () 
    let poll = Poll::new().expect("could not create a new Poll instance.");
    const QUIT : Token = Token(0);
    const CONNECT : Token = Token(1);


    let mut events = Events::with_capacity(5);
    poll.register(&rx,QUIT,Ready::readable(),PollOpt::level() ).expect("could not register channel Receiver.");


    let addr : std::net::SocketAddr = "127.0.0.1:8384".parse().unwrap();
    let listener = TcpListener::bind(&addr).expect("tcp guard - bind failed!");

    poll.register(&listener,CONNECT,Ready::all(),PollOpt::edge()).expect("could not register our listening socket.");

    let mut running : bool = true;
    //listener.set_nonblocking(true).expect("cannot set tcp listener to nonblocking!");
    while running 
        let _nevents = poll.poll(&mut events,None).unwrap();
        println!("poll returned!");
        for event in &events 
            match event.token() 
                QUIT => running = false,
                CONNECT => 
                    match listener.accept() 
                        Ok((_s,pa)) => 
                            println!("tcp client connected:  - closing it down."
                                , pa);
                        
                        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => 
                            //wait_for_fd();
                            continue;
                        
                        Err(e) => panic!("encountered IO error: ", e),
                    
                ,
                Token(_) => continue
            
        
    


fn start_tcpguard() -> (JoinHandle<()>,Sender::<u8>) 
    let (tx,rx) = mio_extras::channel::channel::<u8>();
    let thrd = spawn(move || 
        guard_external_tcp_port(rx);
    );
    (thrd,tx)

我也尝试了sync_channel(0) 版本，但它无法按预期工作。上面代码中的版本，虽然就轮询接收关闭事件而言是有效的。

【讨论】：

很遗憾，mio-extras 与最新的mio 0.7.0 不兼容。