如何在rust中触发异步回调

mwg9r5ms  于 2022-11-30  发布在  其他
关注(0)|答案(1)|浏览(336)

我试图在Rust中实现一个状态机，但是在一个spawn线程中触发状态机的回调时遇到了一些问题。
这是我的StateMachine结构。状态是一个通用的T，因为我想在许多不同的场景中使用它，我使用Vec来存储所有注册到这个StateMachine中的回调。
一开始，我没有使用lifetime 'a，但这样会遇到一些lifetime问题，所以我通过以下建议添加了lifetime 'a：Idiomatic callbacks in Rust

pub struct StateMachine<'a, T> where T:Clone+Eq+'a {
    state: RwLock<T>,
    listeners2: Vec<Arc<Mutex<ListenerCallback<'a, T>>>>,
}

pub type ListenerCallback<'a, T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'a ;

当状态更改时，StateMachine将激发所有回调，如下所示。

pub async fn try_set(&mut self, new_state:T) -> Result<()> {
        if (block_on(self.state.read()).deref().eq(&new_state)) {
            return Ok(())
        }
        // todo change the state

        // fire every listener in spawn
        let mut fire_results = vec![];
        for listener in &mut self.listeners2 {
            let state = new_state.clone();
            let fire_listener = listener.clone();
            fire_results.push(tokio::spawn(async move {
                let mut guard  = fire_listener.lock().unwrap();
                guard.deref_mut()(state);
            }));
        }
        // if fire result return Err, return it
        for fire_result in fire_results {
            fire_result.await?;
        }
        Ok(())
    }

但会导致编译错误。

error[E0521]: borrowed data escapes outside of associated function
  --> src/taf/taf-core/src/execution/state_machine.rs:54:33
   |
15 | impl<'a,T> StateMachine<'a,T> where T:Clone+Eq+Send {
   |      -- lifetime `'a` defined here
...
34 |     pub async fn try_set(&mut self, new_state:T) -> Result<()> {
   |                          --------- `self` is a reference that is only valid in the associated function body
...
54 |             let fire_listener = listener.clone();
   |                                 ^^^^^^^^^^^^^^^^
   |                                 |
   |                                 `self` escapes the associated function body here
   |                                 argument requires that `'a` must outlive `'static`

##########################################################

完整的代码与大量的业务逻辑结合在一起，所以我重写了如下两个演示，问题是相同的。第一个演示同步触发回调，它工作，第二个演示尝试异步触发回调，它遇到同样的问题：self在此处转义关联的函数体。
第一个演示（有效）：

use std::alloc::alloc;
use std::ops::DerefMut;
use std::sync::{Arc, Mutex, RwLock};
use anyhow::Result;
use dashmap::DashMap;

struct StateMachine<'a,T> where T:Clone+Eq+'a {
    state: T,
    listeners: Vec<Box<Callback<'a, T>>>,
}

type Callback<'a, T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'a;

impl<'a, T> StateMachine<'a,T> where T:Clone+Eq+'a {

    pub fn new(init_state: T) -> Self {
        StateMachine {
            state: init_state,
            listeners: vec![]
        }
    }

    pub fn add_listener(&mut self, listener: Box<Callback<'a, T>>) -> Result<()> {
        self.listeners.push(listener);
        Ok(())
    }

    pub fn set(&mut self, new_state: T) -> Result<()> {

        self.state = new_state.clone();

        for listener in &mut self.listeners {
            listener(new_state.clone());
        }
        Ok(())
    }
}

#[derive(Clone, Eq, PartialEq, Hash)]
enum ExeState {
    Waiting,
    Running,
    Finished,
    Failed,
}

struct Execution<'a> {
    exec_id: String,
    pub state_machine: StateMachine<'a, ExeState>,
}

struct ExecManager<'a> {
    all_jobs: Arc<RwLock<DashMap<String, Execution<'a>>>>,
    finished_jobs: Arc<RwLock<Vec<String>>>,
}

impl<'a> ExecManager<'a> {

    pub fn new() -> Self {
        ExecManager {
            all_jobs: Arc::new(RwLock::new(DashMap::new())),
            finished_jobs: Arc::new(RwLock::new(vec![]))
        }
    }

    fn add_job(&mut self, job_id: String) {
        let mut execution = Execution {
            exec_id: job_id.clone(),
            state_machine: StateMachine::new(ExeState::Waiting)
        };

        // add listener
        let callback_finished_jobs = self.finished_jobs.clone();
        let callback_job_id = job_id.clone();
        execution.state_machine.add_listener( Box::new(move |new_state| {
            println!("listener fired!, job_id {}", callback_job_id.clone());
            if new_state == ExeState::Finished || new_state == ExeState::Failed {
                let mut guard = callback_finished_jobs.write().unwrap();
                guard.deref_mut().push(callback_job_id.clone());

            }
            Ok(())
        }));

        let mut guard = self.all_jobs.write().unwrap();
        guard.deref_mut().insert(job_id, execution);
    }

    fn mock_exec(&mut self, job_id: String) {
        let mut guard = self.all_jobs.write().unwrap();
        let mut exec = guard.deref_mut().get_mut(&job_id).unwrap();

        exec.state_machine.set(ExeState::Finished);
    }

}

#[test]
fn test() {
    let mut manager = ExecManager::new();

    manager.add_job(String::from("job_id1"));
    manager.add_job(String::from("job_id2"));

    manager.mock_exec(String::from("job_id1"));
    manager.mock_exec(String::from("job_id2"));

}

第二个演示：

use std::alloc::alloc;
use std::ops::DerefMut;
use std::sync::{Arc, Mutex, RwLock};
use anyhow::Result;
use dashmap::DashMap;
use petgraph::algo::astar;

struct StateMachine<'a,T> where T:Clone+Eq+Send+'a {
    state: T,
    listeners: Vec<Arc<Mutex<Box<Callback<'a, T>>>>>,
}

type Callback<'a, T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'a;

impl<'a, T> StateMachine<'a,T> where T:Clone+Eq+Send+'a {

    pub fn new(init_state: T) -> Self {
        StateMachine {
            state: init_state,
            listeners: vec![]
        }
    }

    pub fn add_listener(&mut self, listener: Box<Callback<'a, T>>) -> Result<()> {
        self.listeners.push(Arc::new(Mutex::new(listener)));
        Ok(())
    }

    pub fn set(&mut self, new_state: T) -> Result<()> {

        self.state = new_state.clone();

        for listener in &mut self.listeners {
            let spawn_listener = listener.clone();
            tokio::spawn(async move {
                let mut guard = spawn_listener.lock().unwrap();
                guard.deref_mut()(new_state.clone());
            });
        }
        Ok(())
    }
}

#[derive(Clone, Eq, PartialEq, Hash)]
enum ExeState {
    Waiting,
    Running,
    Finished,
    Failed,
}

struct Execution<'a> {
    exec_id: String,
    pub state_machine: StateMachine<'a, ExeState>,
}

struct ExecManager<'a> {
    all_jobs: Arc<RwLock<DashMap<String, Execution<'a>>>>,
    finished_jobs: Arc<RwLock<Vec<String>>>,
}

impl<'a> ExecManager<'a> {

    pub fn new() -> Self {
        ExecManager {
            all_jobs: Arc::new(RwLock::new(DashMap::new())),
            finished_jobs: Arc::new(RwLock::new(vec![]))
        }
    }

    fn add_job(&mut self, job_id: String) {
        let mut execution = Execution {
            exec_id: job_id.clone(),
            state_machine: StateMachine::new(ExeState::Waiting)
        };

        // add listener
        let callback_finished_jobs = self.finished_jobs.clone();
        let callback_job_id = job_id.clone();
        execution.state_machine.add_listener( Box::new(move |new_state| {
            println!("listener fired!, job_id {}", callback_job_id.clone());
            if new_state == ExeState::Finished || new_state == ExeState::Failed {
                let mut guard = callback_finished_jobs.write().unwrap();
                guard.deref_mut().push(callback_job_id.clone());

            }
            Ok(())
        }));

        let mut guard = self.all_jobs.write().unwrap();
        guard.deref_mut().insert(job_id, execution);
    }

    fn mock_exec(&mut self, job_id: String) {
        let mut guard = self.all_jobs.write().unwrap();
        let mut exec = guard.deref_mut().get_mut(&job_id).unwrap();

        exec.state_machine.set(ExeState::Finished);
    }

}

#[test]
fn test() {
    let mut manager = ExecManager::new();

    manager.add_job(String::from("job_id1"));
    manager.add_job(String::from("job_id2"));

    manager.mock_exec(String::from("job_id1"));
    manager.mock_exec(String::from("job_id2"));

}

第二个演示的编译错误：

error[E0521]: borrowed data escapes outside of associated function
  --> generic/src/callback2.rs:34:34
   |
15 | impl<'a, T> StateMachine<'a,T> where T:Clone+Eq+Send+'a {
   |      -- lifetime `'a` defined here
...
29 |     pub fn set(&mut self, new_state: T) -> Result<()> {
   |                --------- `self` is a reference that is only valid in the associated function body
...
34 |             let spawn_listener = listener.clone();
   |                                  ^^^^^^^^^^^^^^^^
   |                                  |
   |                                  `self` escapes the associated function body here
   |                                  argument requires that `'a` must outlive `'static`
   |
   = note: requirement occurs because of the type `std::sync::Mutex<Box<dyn FnMut(T) -> Result<(), anyhow::Error> + Send + Sync>>`, which makes the generic argument `Box<dyn FnMut(T) -> Result<(), anyhow::Error> + Send + Sync>` invariant
   = note: the struct `std::sync::Mutex<T>` is invariant over the parameter `T`
   = help: see <https://doc.rust-lang.org/nomicon/subtyping.html> for more information about variance
rust

1条答案

按热度按时间
disbfnqx

disbfnqx1#

使用tokio::spawn() * 衍生的任务不能使用借用的数据 *（在这里，是指生存期为'a的数据，无论它是什么）。这是因为当前没有（并且可能永远不会有）任何方法来保证借用的数据可靠地生存于衍生的任务之后。
您有两个选择：
1.在不产生通知的情况下触发通知。您可以将通知future放入FuturesUnordered以同时运行它们，但它们仍然必须在try_set()之前完成。
1.删除生存期参数;停止允许借用数据的回调。在必要的地方将'static放在您的dyn类型上。更改StateMachine的用户，以便他们不尝试使用借用的数据，而是在必要时使用Arc。

pub struct StateMachine<T> where T: Clone + Eq + 'static {
   state: RwLock<T>,
   listeners2: Vec<Arc<Mutex<ListenerCallback<T>>>>,
}

pub type ListenerCallback<T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'static;
赞(0）分享  回复(0）举报  2022-11-30
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com