如何在rust中触发异步回调

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

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

  1. pub struct StateMachine<'a, T> where T:Clone+Eq+'a {
  2.     state: RwLock<T>,
  3.     listeners2: Vec<Arc<Mutex<ListenerCallback<'a, T>>>>,
  4. }
  6. pub type ListenerCallback<'a, T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'a ;

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

  1. pub async fn try_set(&mut self, new_state:T) -> Result<()> {
  2.         if (block_on(self.state.read()).deref().eq(&new_state)) {
  3.             return Ok(())
  4.         }
  5.         // todo change the state
  7.         // fire every listener in spawn
  8.         let mut fire_results = vec![];
  9.         for listener in &mut self.listeners2 {
  10.             let state = new_state.clone();
  11.             let fire_listener = listener.clone();
  12.             fire_results.push(tokio::spawn(async move {
  13.                 let mut guard  = fire_listener.lock().unwrap();
  14.                 guard.deref_mut()(state);
  15.             }));
  16.         }
  17.         // if fire result return Err, return it
  18.         for fire_result in fire_results {
  19.             fire_result.await?;
  20.         }
  21.         Ok(())
  22.     }

但会导致编译错误。

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

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

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

  1. use std::alloc::alloc;
  2. use std::ops::DerefMut;
  3. use std::sync::{Arc, Mutex, RwLock};
  4. use anyhow::Result;
  5. use dashmap::DashMap;
  7. struct StateMachine<'a,T> where T:Clone+Eq+'a {
  8.     state: T,
  9.     listeners: Vec<Box<Callback<'a, T>>>,
  10. }
  12. type Callback<'a, T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'a;
  14. impl<'a, T> StateMachine<'a,T> where T:Clone+Eq+'a {
  16.     pub fn new(init_state: T) -> Self {
  17.         StateMachine {
  18.             state: init_state,
  19.             listeners: vec![]
  20.         }
  21.     }
  23.     pub fn add_listener(&mut self, listener: Box<Callback<'a, T>>) -> Result<()> {
  24.         self.listeners.push(listener);
  25.         Ok(())
  26.     }
  28.     pub fn set(&mut self, new_state: T) -> Result<()> {
  30.         self.state = new_state.clone();
  32.         for listener in &mut self.listeners {
  33.             listener(new_state.clone());
  34.         }
  35.         Ok(())
  36.     }
  37. }
  39. #[derive(Clone, Eq, PartialEq, Hash)]
  40. enum ExeState {
  41.     Waiting,
  42.     Running,
  43.     Finished,
  44.     Failed,
  45. }
  47. struct Execution<'a> {
  48.     exec_id: String,
  49.     pub state_machine: StateMachine<'a, ExeState>,
  50. }
  52. struct ExecManager<'a> {
  53.     all_jobs: Arc<RwLock<DashMap<String, Execution<'a>>>>,
  54.     finished_jobs: Arc<RwLock<Vec<String>>>,
  55. }
  57. impl<'a> ExecManager<'a> {
  59.     pub fn new() -> Self {
  60.         ExecManager {
  61.             all_jobs: Arc::new(RwLock::new(DashMap::new())),
  62.             finished_jobs: Arc::new(RwLock::new(vec![]))
  63.         }
  64.     }
  66.     fn add_job(&mut self, job_id: String) {
  67.         let mut execution = Execution {
  68.             exec_id: job_id.clone(),
  69.             state_machine: StateMachine::new(ExeState::Waiting)
  70.         };
  72.         // add listener
  73.         let callback_finished_jobs = self.finished_jobs.clone();
  74.         let callback_job_id = job_id.clone();
  75.         execution.state_machine.add_listener( Box::new(move |new_state| {
  76.             println!("listener fired!, job_id {}", callback_job_id.clone());
  77.             if new_state == ExeState::Finished || new_state == ExeState::Failed {
  78.                 let mut guard = callback_finished_jobs.write().unwrap();
  79.                 guard.deref_mut().push(callback_job_id.clone());
  81.             }
  82.             Ok(())
  83.         }));
  85.         let mut guard = self.all_jobs.write().unwrap();
  86.         guard.deref_mut().insert(job_id, execution);
  87.     }
  89.     fn mock_exec(&mut self, job_id: String) {
  90.         let mut guard = self.all_jobs.write().unwrap();
  91.         let mut exec = guard.deref_mut().get_mut(&job_id).unwrap();
  93.         exec.state_machine.set(ExeState::Finished);
  94.     }
  96. }
  98. #[test]
  99. fn test() {
  100.     let mut manager = ExecManager::new();
  102.     manager.add_job(String::from("job_id1"));
  103.     manager.add_job(String::from("job_id2"));
  105.     manager.mock_exec(String::from("job_id1"));
  106.     manager.mock_exec(String::from("job_id2"));
  108. }

第二个演示：

  1. use std::alloc::alloc;
  2. use std::ops::DerefMut;
  3. use std::sync::{Arc, Mutex, RwLock};
  4. use anyhow::Result;
  5. use dashmap::DashMap;
  6. use petgraph::algo::astar;
  8. struct StateMachine<'a,T> where T:Clone+Eq+Send+'a {
  9.     state: T,
  10.     listeners: Vec<Arc<Mutex<Box<Callback<'a, T>>>>>,
  11. }
  13. type Callback<'a, T> = dyn FnMut(T) -> Result<()> + Send + Sync + 'a;
  15. impl<'a, T> StateMachine<'a,T> where T:Clone+Eq+Send+'a {
  17.     pub fn new(init_state: T) -> Self {
  18.         StateMachine {
  19.             state: init_state,
  20.             listeners: vec![]
  21.         }
  22.     }
  24.     pub fn add_listener(&mut self, listener: Box<Callback<'a, T>>) -> Result<()> {
  25.         self.listeners.push(Arc::new(Mutex::new(listener)));
  26.         Ok(())
  27.     }
  29.     pub fn set(&mut self, new_state: T) -> Result<()> {
  31.         self.state = new_state.clone();
  33.         for listener in &mut self.listeners {
  34.             let spawn_listener = listener.clone();
  35.             tokio::spawn(async move {
  36.                 let mut guard = spawn_listener.lock().unwrap();
  37.                 guard.deref_mut()(new_state.clone());
  38.             });
  39.         }
  40.         Ok(())
  41.     }
  42. }
  44. #[derive(Clone, Eq, PartialEq, Hash)]
  45. enum ExeState {
  46.     Waiting,
  47.     Running,
  48.     Finished,
  49.     Failed,
  50. }
  52. struct Execution<'a> {
  53.     exec_id: String,
  54.     pub state_machine: StateMachine<'a, ExeState>,
  55. }
  57. struct ExecManager<'a> {
  58.     all_jobs: Arc<RwLock<DashMap<String, Execution<'a>>>>,
  59.     finished_jobs: Arc<RwLock<Vec<String>>>,
  60. }
  62. impl<'a> ExecManager<'a> {
  64.     pub fn new() -> Self {
  65.         ExecManager {
  66.             all_jobs: Arc::new(RwLock::new(DashMap::new())),
  67.             finished_jobs: Arc::new(RwLock::new(vec![]))
  68.         }
  69.     }
  71.     fn add_job(&mut self, job_id: String) {
  72.         let mut execution = Execution {
  73.             exec_id: job_id.clone(),
  74.             state_machine: StateMachine::new(ExeState::Waiting)
  75.         };
  77.         // add listener
  78.         let callback_finished_jobs = self.finished_jobs.clone();
  79.         let callback_job_id = job_id.clone();
  80.         execution.state_machine.add_listener( Box::new(move |new_state| {
  81.             println!("listener fired!, job_id {}", callback_job_id.clone());
  82.             if new_state == ExeState::Finished || new_state == ExeState::Failed {
  83.                 let mut guard = callback_finished_jobs.write().unwrap();
  84.                 guard.deref_mut().push(callback_job_id.clone());
  86.             }
  87.             Ok(())
  88.         }));
  90.         let mut guard = self.all_jobs.write().unwrap();
  91.         guard.deref_mut().insert(job_id, execution);
  92.     }
  94.     fn mock_exec(&mut self, job_id: String) {
  95.         let mut guard = self.all_jobs.write().unwrap();
  96.         let mut exec = guard.deref_mut().get_mut(&job_id).unwrap();
  98.         exec.state_machine.set(ExeState::Finished);
  99.     }
  101. }
  103. #[test]
  104. fn test() {
  105.     let mut manager = ExecManager::new();
  107.     manager.add_job(String::from("job_id1"));
  108.     manager.add_job(String::from("job_id2"));
  110.     manager.mock_exec(String::from("job_id1"));
  111.     manager.mock_exec(String::from("job_id2"));
  113. }

第二个演示的编译错误：

  1. error[E0521]: borrowed data escapes outside of associated function
  2.   --> generic/src/callback2.rs:34:34
  3.    |
  4. 15 | impl<'a, T> StateMachine<'a,T> where T:Clone+Eq+Send+'a {
  5.    |      -- lifetime `'a` defined here
  6. ...
  7. 29 |     pub fn set(&mut self, new_state: T) -> Result<()> {
  8.    |                --------- `self` is a reference that is only valid in the associated function body
  9. ...
  10. 34 |             let spawn_listener = listener.clone();
  11.    |                                  ^^^^^^^^^^^^^^^^
  12.    |                                  |
  13.    |                                  `self` escapes the associated function body here
  14.    |                                  argument requires that `'a` must outlive `'static`
  15.    |
  16.    = 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
  17.    = note: the struct `std::sync::Mutex<T>` is invariant over the parameter `T`
  18.    = 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。

  1. pub struct StateMachine<T> where T: Clone + Eq + 'static {
  2.    state: RwLock<T>,
  3.    listeners2: Vec<Arc<Mutex<ListenerCallback<T>>>>,
  4. }
  6. 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