From ee0ba0fe6913c5c92ce86e0175cf6abeb19888df Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Sun, 26 Jul 2026 15:44:47 +0200 Subject: [PATCH] Improved reloading --- src/engine/main.rs | 13 --------- src/engine/store/plugin.rs | 55 ++++++++++++++------------------------ src/engine/terminal.rs | 4 +-- 3 files changed, 22 insertions(+), 50 deletions(-) diff --git a/src/engine/main.rs b/src/engine/main.rs index 184a811..c3c6dc6 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -20,18 +20,5 @@ async fn main() -> tokio::io::Result<()> { terminal_server??; broadcaster??; - let strategy = engine.strategy.strategy_handle.lock().await.take(); - let risk = engine.strategy.risk_handle.lock().await.take(); - - drop(engine); - - if let Some(strategy) = strategy { - strategy.into_future().await??; - } - - if let Some(risk) = risk { - risk.into_future().await??; - } - Ok(()) } diff --git a/src/engine/store/plugin.rs b/src/engine/store/plugin.rs index b891aae..112ee0d 100644 --- a/src/engine/store/plugin.rs +++ b/src/engine/store/plugin.rs @@ -19,23 +19,26 @@ use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, process::{Child, ChildStdout, Command}, sync::Mutex, - task::JoinHandle, }; use crate::{engine::Engine, store::pulse_plugin}; #[derive(Debug)] -pub struct Plugin { +pub struct Plugin Deserialize<'de>> { + pub manifest: Mutex, pub stdout: Mutex, pub process: Mutex, + pub _p: (PhantomData, PhantomData), } -impl Plugin { - pub fn new(mut child: Child) -> Self { +impl Deserialize<'de>> Plugin { + pub fn new(mut child: Child, manifest: M) -> Self { Self { stdout: Mutex::new(child.stdout.take().expect("Failed to obtain child stdout")), process: Mutex::new(child), + manifest: Mutex::new(manifest), + _p: (PhantomData, PhantomData), } } @@ -74,13 +77,13 @@ impl Plugin { Ok(()) } - pub async fn reload(&self, mut child: Child) -> tokio::io::Result<()> { + pub async fn reload(&self, mut child: Child, manifest: M) -> tokio::io::Result<()> { let mut process = self.process.lock().await; process.kill().await?; *self.stdout.lock().await = child.stdout.take().expect("Failed to obtain child stdout"); - + *self.manifest.lock().await = manifest; *process = child; Ok(()) @@ -107,11 +110,8 @@ pub fn get_manifest_plugin_pair<'de, M: Deserialize<'de>>( #[derive(Debug)] pub struct StrategyPair { - pub strategy: Plugin, - pub risk: Plugin, - - pub strategy_manifest: Mutex, - pub risk_manifest: Mutex, + pub strategy: Plugin, + pub risk: Plugin, } impl StrategyPair { @@ -132,10 +132,8 @@ impl StrategyPair { )?; Ok(Arc::new(Self { - strategy: Plugin::new(strategy), - risk: Plugin::new(risk), - strategy_manifest: Mutex::new(strategy_manifest), - risk_manifest: Mutex::new(risk_manifest), + strategy: Plugin::new(strategy, strategy_manifest), + risk: Plugin::new(risk, risk_manifest), })) } } @@ -143,10 +141,6 @@ impl StrategyPair { #[derive(Debug)] pub struct StrategyEngine { pub pair: Mutex>, - - pub strategy_handle: Mutex>>>, - pub risk_handle: Mutex>>>, - pub engine: Weak, } @@ -154,8 +148,6 @@ impl StrategyEngine { pub async fn new(strategy_id: &str, risk_id: &str) -> tokio::io::Result { Ok(Self { pair: Mutex::new(StrategyPair::new(strategy_id, risk_id).await?), - strategy_handle: Mutex::new(None), - risk_handle: Mutex::new(None), engine: Weak::new(), }) } @@ -209,33 +201,26 @@ impl StrategyEngine { pub async fn spawn(self: &Arc) { let engine = self.clone(); - *self.strategy_handle.lock().await = - Some(tokio::spawn(async move { engine.run_strategy().await })); + tokio::spawn(async move { engine.run_strategy().await }); let engine = self.clone(); - *self.risk_handle.lock().await = Some(tokio::spawn(async move { engine.run_risk().await })); + tokio::spawn(async move { engine.run_risk().await }); } pub async fn reload_strategy(self: &Arc, id: &str) -> tokio::io::Result<()> { { - if let Some(handle) = &*self.strategy_handle.lock().await { - handle.abort(); - } - let pair = self.pair.lock().await; - let strategy = pulse_plugin(id)?; + let plugin = pulse_plugin(id)?; let (child, manifest) = get_manifest_plugin_pair( - &strategy, - &strategy.join("strategy.bash"), - &fs::read(strategy.join("strategy.toml")).await?, + &plugin, + &plugin.join("strategy.bash"), + &fs::read(plugin.join("strategy.toml")).await?, )?; - pair.risk.reload(child).await?; - - *pair.strategy_manifest.lock().await = manifest; + pair.strategy.reload(child, manifest).await?; } self.spawn().await; diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index 24849e1..13aec15 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -66,8 +66,8 @@ impl TerminalServer { self.send_to( id, pulse_wire::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: pair.strategy_manifest.lock().await.clone(), - risk: pair.risk_manifest.lock().await.clone(), + strategy: pair.strategy.manifest.lock().await.clone(), + risk: pair.risk.manifest.lock().await.clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: TimeFrame::M15,