gpt4 book ai didi

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

转载 作者:行者123 更新时间:2023-12-03 11:36:32 32 4
gpt4 key购买 nike

我最近开始用 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?)

  • 要将接收器不存储在互斥锁中,可以执行以下操作:
    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(...);

    关于asynchronous - rust lazy_static和tokio::select中的tokio::sync::mpsc::channel,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/65452692/

    32 4 0
    Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
    广告合作:1813099741@qq.com 6ren.com