1

我最近开始使用 rust 进行编码,我很喜欢它。我正在编写一个想要“包装”C-API 的项目。在一种情况下,我必须在 Rust 中定义 C 可以调用的回调。我让 bindgen 创建了回调。由于代码需要在某种程度上异步运行,因此我使用 tokio。

我想要达到的目标

我将 main 函数创建为 tokio::main。在主函数中,我创建了 2 个异步任务,一个监听通道,另一个触发 C-API 中的消息队列。如果消息可用,我想通过回调函数上的通道发送它们,这样我就可以在我正在侦听事件的任务上接收消息。稍后我想通过 SSE 或 GraphQL 订阅将这些消息发送给多个客户端。

我无法更改 C-Callbacks,因为它们需要传递给 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

我了解该消息以及为什么会发生这种情况,但我想不出解决方案如何工作。

我有一个使用横梁通道的工作示例,但我宁愿使用来自 tokio 的异步通道,所以我没有那么多依赖项,一切都是异步的。

工作示例:

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(我宁愿在我的代码中没有全局/静态变量)

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

如果有人能解决我的任何一个问题,那就太好了。如果这是一种全新的方法,那也很好。如果频道或 tokio 或其他任何方式不可行,那也没关系。主要是我用了tokio,因为一个crate我也在用tokio,所以我不必有更多的依赖。

已经谢谢你阅读到这里。

4

1 回答 1

2

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

  1. 替换tokio::sync::Mutexstd::sync::Mutex这样您就不必try_lock在回调中使用。
  2. 不要将接收者存储在互斥体中,只有发送者。
  3. 在回调中,要么使用无界通道,要么确保在发送前释放锁。
  4. 在专用线程上运行阻塞 C 代码,std::thread::spawn而不是在tokio::spawn. (为什么?

要不将接收器存储在互斥锁中,您可以执行以下操作:

static ref BROADCAST_CONNECT: Mutex<Option<Sender<bool>>> = Mutex::new(None);

// in main
let (send, recv) = channel(128);
*BROADCAST_CONNECT.lock().unwrap() = Some(send);

如果你想要一个有界通道,你可以先克隆通道,然后调用drop锁,然后使用blocking_send发送来释放锁。对于无限制的通道,这并不重要,因为发送是即时的。

// in C callback
let lock = BROADCAST_CONNECT.lock().unwrap();
let send = lock.as_ref().clone();
drop(lock);
send.blocking_send(...);
于 2020-12-26T12:38:47.183 回答