asynchronous - rust lazy_static和tokio::select中的tokio::sync::mpsc::channel

短命女 2022-10-18 01:29 158阅读 0赞

我最近开始用 rust 编码，我很喜欢它。我在一个要“包装” C-API的项目上编码。在一种情况下，我必须在Rust中定义C可以调用的回调。我让bindgen创建了回调。由于代码需要异步运行，因此我使用了tokio。  
我想要达成的目标  
我将主要功能创建为tokio::main。在主要功能中，我创建了2个异步任务，一个监听 channel ，另一个监听C-API中的消息队列。如果消息可用，我想通过回调函数上的 channel 发送消息，这样我就可以在任务中接收消息，而我正在监听事件。稍后，我想通过SSE或GraphQL订阅将这些消息发送给多个客户端。  
我无法更改C回调，因为它们需要可传递到C-API，并且必须使用回调，否则我不会收到消息。  
我的最新方法看起来像这样简化:

use lazy_static::lazy_static; use tokio::sync::{ mpsc::{channel, Receiver, Sender}, Mutex, }; use bindgen::{notify_connect, notify_connectionstate}; lazy_static! { static ref BROADCAST_CONNECT: Mutex<(Sender<bool>, Receiver<bool>)> = Mutex::new(channel(128)); static ref BROADCAST_CONNECTIONSTATE: Mutex<(Sender<u32>, Receiver<u32>)> = Mutex::new(channel(128)); } #[tokio::main] async fn main() { unsafe { notify_connect(Some(_notify_connect)) } // pass the callback function to the C-API unsafe { notify_connectionstate(Some(_notify_connectionstate)) } // pass the callback function to the C-API tokio::spawn(async move { // wait for a channel to have a message loop { tokio::select! { // wating for a channel to receive a message Some(msg) = BROADCAST_CONNECT.lock().await.1.recv() => println!("{}", msg), Some(msg) = BROADCAST_CONNECTIONSTATE.lock().await.1.recv() => println!("{}", msg), } } }); let handle2 = tokio::spawn(async move { loop { unsafe { message_queue_in_c( some_handle, true, Duration::milliseconds(100).num_microseconds().unwrap(), ) } } }); handle.await.unwrap(); habdle2.await.unwrap(); } // the callback function that gets called from the C-API unsafe extern "C" fn _notify_connect(is_connected: bool) { // C-API is not async, so use synchronous lock match BROADCAST_CONNECT.try_lock() { Ok(value) => match value.0.blocking_send(is_connected) { Ok(_) => {} Err(e) => { eprintln!("{}", e) } }, Err(e) => { eprintln!("{}", e) } } } unsafe extern "C" fn _notify_connectionstate(connectionstate: u32) { match BROADCAST_CONNECTIONSTATE.try_lock() { Ok(value) => match value.0.blocking_send(connectionstate) { Ok(_) => {} Err(e) => { eprintln!("{}", e) } }, Err(e) => { eprintln!("{}", e) } } }

问题:

error[E0716]: temporary value dropped while borrowed --> src/main.rs:37:29 | 35 | / tokio::select! { 36 | | Some(msg) = BROADCAST_CONNECT.lock().await.1.recv() => println!("{}", msg), 37 | | Some(msg) = BROADCAST_CONNECTIONSTATE.lock().await.1.recv() => println!("{}", msg), | | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ creates a temporary which is freed while still in use 38 | | } | | - | | | | |_____________temporary value is freed at the end of this statement | borrow later captured here by closure | = note: consider using a `let` binding to create a longer lived value

我了解该消息以及发生这种情况的原因，但是我想不出解决方案。  
我有一个使用横梁 channel 的工作示例，但是我宁愿使用来自tokio的异步 channel ，所以我没有那么多的依赖关系，并且一切都是异步的。  
工作示例:

use lazy_static::lazy_static; use crossbeam::{ channel::{bounded, Receiver, Sender}, select, }; use bindgen::{notify_connect, notify_connectionstate}; lazy_static! { static ref BROADCAST_CONNECT: (Sender<bool>, Receiver<bool>) = bounded(128); static ref BROADCAST_CONNECTIONSTATE: (Sender<u32>, Receiver<u32>) = bounded(128); } #[tokio::main] async fn main() { unsafe { notify_connect(Some(_notify_connect)) } // pass the callback function to the C-API unsafe { notify_connectionstate(Some(_notify_connectionstate)) } // pass the callback function to the C-API let handle1 = tokio::spawn(async move { loop { select! { recv(&BROADCAST_CONNECT.1) -> msg => println!("is_connected: {:?}", msg.unwrap()), recv(&BROADCAST_CONNECTIONSTATE.1) -> msg => println!("connectionstate: {:?}", msg.unwrap()), } } }); let handle2 = tokio::spawn(async move { loop { unsafe { message_queue_in_c( some_handle, true, Duration::milliseconds(100).num_microseconds().unwrap(), ) } } }); handle.await.unwrap(); handle2.await.unwrap(); } // the callback function thats gets called from the C-API unsafe extern "C" fn _notify_connect(is_connected: bool) { match &BROADCAST_CONNECT.0.send(is_connected) { Ok(_) => {} Err(e) => eprintln!("{}", e), }; } unsafe extern "C" fn _notify_connectionstate(connectionstate: u32) { match BROADCAST_CONNECTIONSTATE.0.send(connectionstate) { Ok(_) => {} Err(e) => eprintln!("{}", e), } }

选择  
我也无法使用的一种替代方法是使用某种局部函数或使用闭包。但是我不确定这是否甚至可以工作。也许有人有一个主意。如果这样的事情行得通，那就太好了，所以我不必使用lazy\_static(我宁愿在我的代码中没有global/static变量)

use tokio::sync::{ mpsc::{channel, Receiver, Sender}, Mutex, }; use bindgen::{notify_connect, notify_connectionstate}; #[tokio::main] async fn main() { let app = app::App::new(); let mut broadcast_connect = channel::<bool>(128); let mut broadcast_connectionstate = channel::<bool>(128); let notify_connect = { unsafe extern "C" fn _notify_connect(is_connected: bool) { match broadcast_connect.0.blocking_send(is_connected) { Ok(_) => {} Err(e) => { eprintln!("{}", e) } } } }; let notify_connectionstate = { unsafe extern "C" fn _notify_connectionstate(connectionstate: u32) { match broadcast_connectionstate.0.blocking_send(connectionstate) { Ok(_) => {} Err(e) => { eprintln!("{}", e) } } } }; unsafe { notify_connect(Some(notify_connect)) } // pass the callback function to the C-API unsafe { notify_connectionstate(Some(notify_connectionstate)) } // pass the callback function to the C-API let handle = tokio::spawn(async move { loop { tokio::select! { Some(msg) = broadcast_connect.1.recv() => println!("{}", msg), Some(msg) = broadcast_connectionstate.1.recv() => println!("{}", msg), } } }); let handle2 = tokio::spawn(async move { loop { unsafe { message_queue_in_c( some_handle, true, Duration::milliseconds(100).num_microseconds().unwrap(), ) } } }); handle.await.unwrap(); handle2.await.unwrap(); }

这种方法的问题

can't capture dynamic environment in a fn item --> src/main.rs:47:19 | 47 | match broadcast_connectionstate.0.blocking_send(connectionstate) { | ^^^^^^^^^^^^^^^^^^^^^^^^^ | = help: use the `|| { ... }` closure form instead

如果有人可以解决我的任何一个问题，那就太好了。如果这是一种全新的方法，那也可以。如果没有 channel ，tokio或其他方法，那也没关系。我主要是使用tokio的，因为我也使用了tokio的 crate ，因此我不必再有更多依赖项。  
非常感谢您的阅读，直到这里为止。

**最佳答案**

如果您对第一个示例进行了以下更改，则它应该可以工作:

*  用`tokio::sync::Mutex`替换`std::sync::Mutex`，因此您不必在回调中使用`try_lock`。
 *  不要将接收方存储在互斥锁中，而仅将发送方存储。
 *  在回调中，使用无限制 channel ，或确保在发送之前释放锁定。
 *  使用`std::thread::spawn`而不是`tokio::spawn`在专用线程上运行阻塞的C代码。 ([why?][why])
 *  要将接收器不存储在互斥锁中，可以执行以下操作:
    
        static ref BROADCAST_CONNECT: Mutex<Option<Sender<bool>>> = Mutex::new(None); // in main let (send, recv) = channel(128); *BROADCAST_CONNECT.lock().unwrap() = Some(send);
    
    如果要使用有界 channel ，可以先克隆 channel ，然后在该锁上调用`drop`，然后使用`blocking_send`来释放该锁。对于无限制的 channel ，这无关紧要，因为发送是即时的。
    
        // in C callback let lock = BROADCAST_CONNECT.lock().unwrap(); let send = lock.as_ref().clone(); drop(lock); send.blocking_send(...);

原文链接：[https://stackoverflow.com/questions/65452692/rust-lazy-static-and-tokiosyncmpscchannel-in-tokioselect][https_stackoverflow.com_questions_65452692_rust-lazy-static-and-tokiosyncmpscchannel-in-tokioselect]

[why]: https://ryhl.io/blog/async-what-is-blocking/
[https_stackoverflow.com_questions_65452692_rust-lazy-static-and-tokiosyncmpscchannel-in-tokioselect]: https://stackoverflow.com/questions/65452692/rust-lazy-static-and-tokiosyncmpscchannel-in-tokioselect