From 7f1e3f8d7d03e9dfa6b1d5fb4ed9630ceaabd36f Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 05:56:59 +0200 Subject: [PATCH] Improved architecture with signle mutex --- pulse-sdk/src/lib.rs | 5 -- src/engine/engine/command.rs | 4 +- src/engine/engine/mod.rs | 4 +- src/engine/engine/strategy.rs | 134 +++++++++++++++++----------------- src/engine/main.rs | 2 +- src/engine/store/strategy.rs | 61 +++++++--------- src/engine/terminal.rs | 8 +- 7 files changed, 106 insertions(+), 112 deletions(-) diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index 6d644bd..eddadf3 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -35,11 +35,6 @@ pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> { Ok(()) } -use std::sync::{ - Arc, - atomic::{AtomicBool, Ordering}, -}; - #[allow(async_fn_in_trait)] pub trait Strategy { async fn start(&self) -> tokio::io::Result<()> { diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 3644ad3..91c2c01 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -60,7 +60,7 @@ impl Engine { .await; } - self.strategy.reload(id.as_str()).await?; + self.strategy_engine.reload(id.as_str()).await?; self.config.lock().await.strategy = id; }); @@ -75,7 +75,7 @@ impl Engine { self.terminal_server.broadcast( pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: self.strategy.child.manifest.lock().await.clone(), + strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), mode: Mode::Auto, state: ItemState::Running, diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index f8011b9..b9cc3f6 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -20,7 +20,7 @@ pub struct WatchList { #[derive(Clone)] pub struct Engine { pub terminal_server: Arc, - pub strategy: Arc, + pub strategy_engine: Arc, pub config: Arc>, pub accounts: Arc>, pub watch_list: Arc>, @@ -38,7 +38,7 @@ impl Engine { Ok(Arc::new_cyclic(|engine| Self { terminal_server: TerminalServer::new(engine.clone()), - strategy: strategy.initialize(engine.clone()), + strategy_engine: strategy.initialize(engine.clone()), config, accounts, watch_list: Arc::new(Mutex::new(WatchList { diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index d45db97..6c1d372 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -1,26 +1,24 @@ +use anyhow::Context; use hypersdk::hypercore::{self, CandleInterval, Subscription, WebSocket}; use pulse_sdk::prelude::*; use std::{ collections::HashSet, - path::PathBuf, - process::Stdio, sync::{Arc, Weak}, time::{SystemTime, UNIX_EPOCH}, }; -use tokio::{ - fs, - process::{Child, Command}, - sync::Mutex, -}; +use tokio::{fs, sync::Mutex}; use crate::{ engine::Engine, - store::{pulse_strategy, strategy::StrategyChild}, + store::{ + pulse_strategy, + strategy::{StrategyChild, get_manifest}, + }, }; pub struct StrategyEngine { pub engine: Weak, - pub child: Arc, + pub strategy: Mutex, pub ws: WebSocket, pub subscriptions: Mutex>, } @@ -42,7 +40,7 @@ impl StrategyEngine { )?; Ok(Self { - child: Arc::new(StrategyChild::new(strategy, strategy_manifest)), + strategy: Mutex::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), ws: hypercore::mainnet_ws(), subscriptions: Mutex::new(HashSet::new()), @@ -55,24 +53,66 @@ impl StrategyEngine { Arc::new(self) } - pub async fn run(&self) -> anyhow::Result<()> { + pub async fn send(self: &Arc, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { + self.strategy.lock().await.send(msg).await + } + + pub async fn spawn(self: &Arc) { + let engine = self.clone(); + + tokio::spawn(async move { engine.run().await }); + } + + pub async fn reload(self: &Arc, id: &str) -> tokio::io::Result<()> { + { + let strategy = pulse_strategy(id)?; + + let (child, manifest) = get_manifest( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, + )?; + + let mut strategy = self.strategy.lock().await; + + strategy.child.kill().await?; + strategy.child = child; + strategy.manifest = manifest; + } + + let engine = self.clone(); + tokio::spawn(async move { engine.run().await }); + + Ok(()) + } + + pub async fn run(self: &Arc) -> anyhow::Result<()> { let engine = self .engine .upgrade() .expect("Failed to upgrade engine (StrategyEngine)"); - self.child.send(&StrategyEngineMessage::Initialize).await?; + let mut stdout = { + let mut child = self.strategy.lock().await; + + child + .child + .stdout + .take() + .context("Failed to obtain strategy stdout")? + }; + + self.send(&StrategyEngineMessage::Initialize).await?; loop { - match self.child.recv().await? { + match StrategyChild::read(&mut stdout).await? { None => {} Some(StrategyMessage::GetWatchList) => { - self.child - .send(&StrategyEngineMessage::WatchList( - engine.watch_list.lock().await.clone().items, - )) - .await?; + self.send(&StrategyEngineMessage::WatchList( + engine.watch_list.lock().await.clone().items, + )) + .await?; } Some(StrategyMessage::Log(mut log)) => { @@ -150,58 +190,16 @@ impl StrategyEngine { let start_time = now.saturating_sub(interval_ms * count as u64); - self.child - .send(&StrategyEngineMessage::Candlestick { - candles: client - .candle_snapshot(&symbol, interval, start_time, now) - .await?, - symbol, - interval, - }) - .await?; + self.send(&StrategyEngineMessage::Candlestick { + candles: client + .candle_snapshot(&symbol, interval, start_time, now) + .await?, + symbol, + interval, + }) + .await?; } } } } - - pub async fn spawn(self: &Arc) { - let engine = self.clone(); - - tokio::spawn(async move { engine.run().await }); - } - - pub async fn reload(self: &Arc, id: &str) -> tokio::io::Result<()> { - let strategy = pulse_strategy(id)?; - - let (child, manifest) = get_manifest( - &strategy, - &strategy.join("strategy.bash"), - &fs::read(strategy.join("strategy.toml")).await?, - )?; - - self.child.reload(child, manifest).await?; - - let engine = self.clone(); - tokio::spawn(async move { engine.run().await }); - - Ok(()) - } -} - -fn get_manifest<'de, M: serde::Deserialize<'de>>( - strategy_dir: &PathBuf, - strategy_path: &PathBuf, - manifest: &'de [u8], -) -> tokio::io::Result<(Child, M)> { - Ok(( - Command::new("bash") - .arg(strategy_path) - .current_dir(strategy_dir) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()) - .spawn()?, - toml::from_slice(manifest) - .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?, - )) } diff --git a/src/engine/main.rs b/src/engine/main.rs index 1d45b2d..1393e79 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -9,7 +9,7 @@ async fn main() -> anyhow::Result<()> { let broadcaster = engine.spawn_broadcaster().await; - engine.strategy.spawn().await; + engine.strategy_engine.spawn().await; engine.terminal_server.run().await?; diff --git a/src/engine/store/strategy.rs b/src/engine/store/strategy.rs index fa7a99f..858fd9a 100644 --- a/src/engine/store/strategy.rs +++ b/src/engine/store/strategy.rs @@ -1,31 +1,25 @@ +use std::{path::PathBuf, process::Stdio}; + use pulse_sdk::{ map_postcard_err, strategy::{StrategyEngineMessage, StrategyManifest, StrategyMessage}, }; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, - process::{Child, ChildStdout}, - sync::Mutex, + process::{Child, ChildStdout, Command}, }; #[derive(Debug)] pub struct StrategyChild { - pub manifest: Mutex, - pub stdout: Mutex, - pub process: Mutex, + pub manifest: StrategyManifest, + pub child: Child, } impl StrategyChild { - pub fn new(mut child: Child, manifest: StrategyManifest) -> Self { - Self { - stdout: Mutex::new(child.stdout.take().expect("Failed to obtain child stdout")), - process: Mutex::new(child), - manifest: Mutex::new(manifest), - } + pub fn new(child: Child, manifest: StrategyManifest) -> Self { + Self { child, manifest } } - pub async fn recv(&self) -> tokio::io::Result> { - let mut stdout = self.stdout.lock().await; - + pub async fn read(stdout: &mut ChildStdout) -> tokio::io::Result> { let mut len_buf = [0u8; size_of::()]; let size = stdout.read_exact(&mut len_buf).await?; @@ -42,14 +36,13 @@ impl StrategyChild { Ok(Some(map_postcard_err(postcard::from_bytes(&buffer))?)) } - pub async fn send(&self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { + pub async fn send(&mut self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { self.send_raw(&map_postcard_err(postcard::to_allocvec(msg))?) .await } - pub async fn send_raw(&self, msg: &[u8]) -> tokio::io::Result<()> { - let mut process = self.process.lock().await; - let stdin = process.stdin.as_mut().unwrap(); + pub async fn send_raw(&mut self, msg: &[u8]) -> tokio::io::Result<()> { + let stdin = self.child.stdin.as_mut().unwrap(); stdin.write_all(&msg.len().to_le_bytes()).await?; stdin.write_all(msg).await?; @@ -57,20 +50,22 @@ impl StrategyChild { Ok(()) } +} - pub async fn reload( - &self, - mut child: Child, - manifest: StrategyManifest, - ) -> 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(()) - } +pub fn get_manifest<'de, M: serde::Deserialize<'de>>( + strategy_dir: &PathBuf, + strategy_path: &PathBuf, + manifest: &'de [u8], +) -> tokio::io::Result<(Child, M)> { + Ok(( + Command::new("bash") + .arg(strategy_path) + .current_dir(strategy_dir) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn()?, + toml::from_slice(manifest) + .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?, + )) } diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index a08f521..5dcfe20 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -65,7 +65,13 @@ impl TerminalServer { self.send_to( id, pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: engine.strategy.child.manifest.lock().await.clone(), + strategy: engine + .strategy_engine + .strategy + .lock() + .await + .manifest + .clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: engine.config.lock().await.cooldown,