Improved architecture with signle mutex

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<()> {

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,

src/engine/engine/mod.rs

@@ -20,7 +20,7 @@ pub struct WatchList {
 #[derive(Clone)]
 pub struct Engine {
     pub terminal_server: Arc<TerminalServer>,
-    pub strategy: Arc<StrategyEngine>,
+    pub strategy_engine: Arc<StrategyEngine>,
     pub config: Arc<Mutex<Config>>,
     pub accounts: Arc<Mutex<AccountList>>,
     pub watch_list: Arc<Mutex<WatchList>>,
@@ -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 {

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::{
+use tokio::{fs, sync::Mutex};
-    fs,
-    process::{Child, Command},
-    sync::Mutex,
-};
 
 use crate::{
     engine::Engine,
-    store::{pulse_strategy, strategy::StrategyChild},
+    store::{
+        pulse_strategy,
+        strategy::{StrategyChild, get_manifest},
+    },
 };
 
 pub struct StrategyEngine {
     pub engine: Weak<Engine>,
-    pub child: Arc<StrategyChild>,
+    pub strategy: Mutex<StrategyChild>,
     pub ws: WebSocket,
     pub subscriptions: Mutex<HashSet<Subscription>>,
 }
@@ -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,21 +53,63 @@ impl StrategyEngine {
         Arc::new(self)
     }
 
-    pub async fn run(&self) -> anyhow::Result<()> {
+    pub async fn send(self: &Arc<Self>, msg: &StrategyEngineMessage) -> tokio::io::Result<()> {
+        self.strategy.lock().await.send(msg).await
+    }
+
+    pub async fn spawn(self: &Arc<Self>) {
+        let engine = self.clone();
+
+        tokio::spawn(async move { engine.run().await });
+    }
+
+    pub async fn reload(self: &Arc<Self>, 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<Self>) -> 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
+                    self.send(&StrategyEngineMessage::WatchList(
-                        .send(&StrategyEngineMessage::WatchList(
                         engine.watch_list.lock().await.clone().items,
                     ))
                     .await?;
@@ -150,8 +190,7 @@ impl StrategyEngine {
 
                     let start_time = now.saturating_sub(interval_ms * count as u64);
 
-                    self.child
+                    self.send(&StrategyEngineMessage::Candlestick {
-                        .send(&StrategyEngineMessage::Candlestick {
                         candles: client
                             .candle_snapshot(&symbol, interval, start_time, now)
                             .await?,
@@ -163,45 +202,4 @@ impl StrategyEngine {
             }
         }
     }
-
-    pub async fn spawn(self: &Arc<Self>) {
-        let engine = self.clone();
-
-        tokio::spawn(async move { engine.run().await });
-    }
-
-    pub async fn reload(self: &Arc<Self>, 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()))?,
-    ))
 }

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?;
 

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},
+    process::{Child, ChildStdout, Command},
-    sync::Mutex,
 };
 #[derive(Debug)]
 pub struct StrategyChild {
-    pub manifest: Mutex<StrategyManifest>,
+    pub manifest: StrategyManifest,
-    pub stdout: Mutex<ChildStdout>,
+    pub child: Child,
-    pub process: Mutex<Child>,
 }
 
 impl StrategyChild {
-    pub fn new(mut child: Child, manifest: StrategyManifest) -> Self {
+    pub fn new(child: Child, manifest: StrategyManifest) -> Self {
-        Self {
+        Self { child, manifest }
-            stdout: Mutex::new(child.stdout.take().expect("Failed to obtain child stdout")),
-            process: Mutex::new(child),
-            manifest: Mutex::new(manifest),
-        }
     }
 
-    pub async fn recv(&self) -> tokio::io::Result<Option<StrategyMessage>> {
+    pub async fn read(stdout: &mut ChildStdout) -> tokio::io::Result<Option<StrategyMessage>> {
-        let mut stdout = self.stdout.lock().await;
-
         let mut len_buf = [0u8; size_of::<usize>()];
         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<()> {
+    pub async fn send_raw(&mut self, msg: &[u8]) -> tokio::io::Result<()> {
-        let mut process = self.process.lock().await;
+        let stdin = self.child.stdin.as_mut().unwrap();
-        let stdin = process.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()))?,
+    ))
 }

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,