Continue LiveNode and AsyncRunner in Rust

Author: cjdsellers

Cargo.lock

@@ -32,10 +32,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // #[cfg(feature = "python")]
     pyo3::prepare_freethreaded_python();
 
-    tracing_subscriber::fmt()
-        .with_max_level(tracing::Level::INFO)
-        .init();
-
     dotenvy::dotenv().ok();
 
     let environment = Environment::Live;
@@ -70,7 +66,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Create and register a Databento subscriber actor
     let client_id = ClientId::new("DATABENTO");
     let instrument_ids = vec![
-        InstrumentId::from("ES.c.0.GLBX"),
+        InstrumentId::from("ESM5.XCME"),
         // Add more instruments as needed
     ];
 

crates/adapters/databento/bin/node_test.rs

@@ -26,10 +26,16 @@ use std::{
 use ahash::AHashMap;
 use databento::live::Subscription;
 use indexmap::IndexMap;
-use nautilus_common::messages::data::{
-    RequestBars, RequestInstruments, RequestQuotes, RequestTrades, SubscribeBookDeltas,
-    SubscribeInstrumentStatus, SubscribeQuotes, SubscribeTrades, UnsubscribeBookDeltas,
-    UnsubscribeInstrumentStatus, UnsubscribeQuotes, UnsubscribeTrades,
+use nautilus_common::{
+    messages::{
+        DataEvent,
+        data::{
+            RequestBars, RequestInstruments, RequestQuotes, RequestTrades, SubscribeBookDeltas,
+            SubscribeInstrumentStatus, SubscribeQuotes, SubscribeTrades, UnsubscribeBookDeltas,
+            UnsubscribeInstrumentStatus, UnsubscribeQuotes, UnsubscribeTrades,
+        },
+    },
+    runner::get_data_event_sender,
 };
 use nautilus_core::time::AtomicTime;
 use nautilus_data::client::DataClient;
@@ -38,7 +44,10 @@ use nautilus_model::{
     identifiers::{ClientId, Symbol, Venue},
     instruments::Instrument,
 };
-use tokio::{sync::mpsc, task::JoinHandle};
+use tokio::{
+    sync::mpsc::{self, UnboundedSender},
+    task::JoinHandle,
+};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
@@ -108,6 +117,8 @@ pub struct DatabentoDataClient {
     publisher_venue_map: Arc<IndexMap<PublisherId, Venue>>,
     /// Symbol to venue mapping (for caching).
     symbol_venue_map: Arc<RwLock<AHashMap<Symbol, Venue>>>,
+    /// Data event sender for forwarding data to AsyncRunner.
+    data_sender: UnboundedSender<DataEvent>,
 }
 
 impl DatabentoDataClient {
@@ -141,6 +152,9 @@ impl DatabentoDataClient {
             .map(|p| (p.publisher_id, Venue::from(p.venue.as_str())))
             .collect::<IndexMap<u16, Venue>>();
 
+        // Get the data event sender for forwarding data to AsyncRunner
+        let data_sender = get_data_event_sender();
+
         Ok(Self {
             client_id,
             config,
@@ -152,6 +166,7 @@ impl DatabentoDataClient {
             cancellation_token: CancellationToken::new(),
             publisher_venue_map: Arc::new(publisher_venue_map),
             symbol_venue_map: Arc::new(RwLock::new(AHashMap::new())),
+            data_sender,
         })
     }
 
@@ -238,6 +253,7 @@ impl DatabentoDataClient {
         });
 
         let cancellation_token = self.cancellation_token.clone();
+        let data_sender = self.data_sender.clone();
 
         // Spawn message processing task with cancellation support
         let msg_handle = tokio::spawn(async move {
@@ -248,7 +264,9 @@ impl DatabentoDataClient {
                         match msg {
                             Some(LiveMessage::Data(data)) => {
                                 tracing::debug!("Received data: {data:?}");
-                                // TODO: Forward to message bus or data engine
+                                if let Err(e) = data_sender.send(DataEvent::Data(data)) {
+                                    tracing::error!("Failed to send data event: {e}");
+                                }
                             }
                             Some(LiveMessage::Instrument(instrument)) => {
                                 tracing::debug!("Received instrument: {}", instrument.id());

crates/adapters/databento/src/data.rs

@@ -219,6 +219,11 @@ impl MessagingSwitchboard {
         "DataEngine.process".into()
     }
 
+    #[must_use]
+    pub fn data_engine_response() -> MStr<Endpoint> {
+        "DataEngine.response".into()
+    }
+
     #[must_use]
     pub fn exec_engine_execute() -> MStr<Endpoint> {
         "ExecEngine.execute".into()

crates/common/src/msgbus/switchboard.rs

@@ -19,6 +19,8 @@ use std::{
     rc::Rc,
 };
 
+use tokio::sync::mpsc::UnboundedSender;
+
 use crate::{
     clock::Clock,
     messages::{DataEvent, data::DataCommand},
@@ -56,6 +58,7 @@ pub fn set_global_clock(c: Rc<RefCell<dyn Clock>>) {
 pub type DataCommandQueue = Rc<RefCell<VecDeque<DataCommand>>>;
 
 /// Get globally shared message bus command queue
+///
 /// # Panics
 ///
 /// Panics if thread-local storage cannot be accessed.
@@ -106,16 +109,64 @@ pub fn set_data_evt_queue(dq: Rc<RefCell<dyn DataQueue>>) {
         .expect("Should be able to access thread local storage");
 }
 
+/// Sends a data event to the global data event queue.
+///
+/// This function provides a convenient way for data clients and feed handlers
+/// to send data events to the AsyncRunner for processing.
+///
+/// # Panics
+///
+/// Panics if thread-local storage cannot be accessed or the data event queue is uninitialized.
+pub fn send_data_event(event: DataEvent) {
+    get_data_evt_queue().borrow_mut().push(event);
+}
+
+/// Sets the global data event sender.
+///
+/// This should be called by the AsyncRunner when it creates the channel.
+///
+/// # Panics
+///
+/// Panics if thread-local storage cannot be accessed or a sender is already set.
+pub fn set_data_event_sender(sender: UnboundedSender<DataEvent>) {
+    DATA_EVT_SENDER
+        .try_with(|s| {
+            assert!(s.set(sender).is_ok(), "Data event sender already set");
+        })
+        .expect("Should be able to access thread local storage");
+}
+
+/// Gets a cloned data event sender.
+///
+/// This allows data clients to send events directly to the AsyncRunner
+/// without going through shared mutable state.
+///
+/// # Panics
+///
+/// Panics if thread-local storage cannot be accessed or the sender is uninitialized.
+#[must_use]
+pub fn get_data_event_sender() -> UnboundedSender<DataEvent> {
+    DATA_EVT_SENDER
+        .try_with(|s| {
+            s.get()
+                .expect("Data event sender should be initialized by AsyncRunner")
+                .clone()
+        })
+        .expect("Should be able to access thread local storage")
+}
+
 thread_local! {
     static CLOCK: OnceCell<GlobalClock> = OnceCell::new();
     static DATA_EVT_QUEUE: OnceCell<GlobalDataQueue> = OnceCell::new();
     static DATA_CMD_QUEUE: DataCommandQueue = Rc::new(RefCell::new(VecDeque::new()));
+    // TODO: Potentially redundant but added to simplify the abstraction layers for now
+    static DATA_EVT_SENDER: OnceCell<UnboundedSender<DataEvent>> = const { OnceCell::new() };
 }
 
 // Represents different event types for the runner.
 #[allow(clippy::large_enum_variant)]
 #[derive(Debug)]
 pub enum RunnerEvent {
+    Time(TimeEvent),
     Data(DataEvent),
-    Timer(TimeEvent),
 }

crates/common/src/runner.rs

@@ -571,7 +571,9 @@ impl DataEngine {
     /// Currently supports `InstrumentAny`; unrecognized types are logged as errors.
     pub fn process(&mut self, data: &dyn Any) {
         // TODO: Eventually these could be added to the `Data` enum? process here for now
-        if let Some(instrument) = data.downcast_ref::<InstrumentAny>() {
+        if let Some(data) = data.downcast_ref::<Data>() {
+            self.process_data(data.clone()); // TODO: Optimize (not necessary if we change handler)
+        } else if let Some(instrument) = data.downcast_ref::<InstrumentAny>() {
             self.handle_instrument(instrument.clone());
         } else {
             log::error!("Cannot process data {data:?}, type is unrecognized");

crates/data/src/engine/mod.rs

@@ -71,6 +71,7 @@ pyo3 = { workspace = true, optional = true }
 serde = { workspace = true }
 strum = { workspace = true }
 tokio = { workspace = true }
+tracing = { workspace = true }
 ustr = { workspace = true }
 
 [dev-dependencies]

crates/live/Cargo.toml

@@ -19,8 +19,12 @@
 
 use std::{cell::RefCell, collections::HashMap, rc::Rc};
 
+use anyhow::Context;
 use nautilus_common::{
-    actor::DataActor, clock::LiveClock, component::Component, enums::Environment,
+    actor::{Actor, DataActor},
+    clock::LiveClock,
+    component::Component,
+    enums::Environment,
 };
 use nautilus_core::UUID4;
 use nautilus_data::client::DataClientAdapter;
@@ -30,6 +34,7 @@ use nautilus_system::{
     factories::{ClientConfig, DataClientFactory, ExecutionClientFactory},
     kernel::NautilusKernel,
 };
+use tokio::sync::mpsc::UnboundedSender;
 
 use crate::{config::LiveNodeConfig, runner::AsyncRunner};
 
@@ -42,6 +47,7 @@ pub struct LiveNode {
     clock: Rc<RefCell<LiveClock>>,
     kernel: NautilusKernel,
     runner: AsyncRunner,
+    signal_tx: Option<UnboundedSender<()>>,
     config: LiveNodeConfig,
     is_running: bool,
 }
@@ -83,14 +89,15 @@ impl LiveNode {
 
         let clock = Rc::new(RefCell::new(LiveClock::new()));
         let kernel = NautilusKernel::new(name, config.clone())?;
-        let runner = AsyncRunner::new(clock.clone());
+        let (runner, signal_tx) = AsyncRunner::new(clock.clone());
 
         log::info!("LiveNode built successfully with kernel config");
 
         Ok(Self {
             clock,
             kernel,
             runner,
+            signal_tx: Some(signal_tx),
             config,
             is_running: false,
         })
@@ -106,9 +113,9 @@ impl LiveNode {
             anyhow::bail!("LiveNode is already running");
         }
 
-        log::info!("Starting live node");
+        log::info!("Starting LiveNode");
+
         self.kernel.start_async().await;
-        log::info!("Kernel start completed");
         self.is_running = true;
 
         log::info!("LiveNode started successfully");
@@ -125,7 +132,8 @@ impl LiveNode {
             anyhow::bail!("LiveNode is not running");
         }
 
-        log::info!("Stopping live node");
+        log::info!("Stopping LiveNode");
+
         self.kernel.stop_async().await;
         self.is_running = false;
 
@@ -142,40 +150,35 @@ impl LiveNode {
     ///
     /// Returns an error if the node fails to start or encounters a runtime error.
     pub async fn run(&mut self) -> anyhow::Result<()> {
-        log::info!("Starting LiveNode...");
-        self.start().await?;
+        let signal_tx = self.signal_tx.take().context("LiveNode already running")?;
 
-        log::info!("LiveNode started, setting up signal handler...");
-
-        let loop_duration = tokio::time::Duration::from_millis(100);
+        self.start().await?;
 
-        // TODO: Temporary count logging for development
-        let mut count = 0;
-        loop {
-            count += 1;
-            if count % 10 == 0 {
-                log::info!("Signal handler loop iteration {count}");
+        tokio::select! {
+            // Run on main thread
+            _ = self.runner.run() => {
+                log::info!("AsyncRunner finished");
             }
-
-            tokio::select! {
-                result = tokio::signal::ctrl_c() => {
-                    match result {
-                        Ok(()) => {
-                            log::info!("Received SIGINT, shutting down...");
-                            break;
-                        }
-                        Err(e) => {
-                            log::error!("Failed to listen for SIGINT: {e}");
-                            anyhow::bail!("Signal handling failed: {e}");
+            // Handle SIGINT signal
+            result = tokio::signal::ctrl_c() => {
+                match result {
+                    Ok(()) => {
+                        log::info!("Received SIGINT, shutting down");
+                        if let Err(e) = signal_tx.send(()) {
+                            log::error!("Failed to send shutdown signal: {e}");
                         }
+                        // Give the AsyncRunner a moment to process the shutdown signal
+                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+                    }
+                    Err(e) => {
+                        log::error!("Failed to listen for SIGINT: {e}");
                     }
-                }
-                _ = tokio::time::sleep(loop_duration) => {
                 }
             }
         }
 
-        log::info!("Shutting down after signal...");
+        log::debug!("AsyncRunner and signal handling finished"); // TODO: Temp logging
+
         self.stop().await?;
         Ok(())
     }
@@ -224,7 +227,7 @@ impl LiveNode {
     /// - The node is currently running.
     pub fn add_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
     where
-        T: DataActor + Component + 'static,
+        T: DataActor + Component + Actor + 'static,
     {
         if self.is_running {
             anyhow::bail!(
@@ -417,7 +420,7 @@ impl LiveNodeBuilder {
 
         let clock = Rc::new(RefCell::new(LiveClock::new()));
         let kernel = NautilusKernel::new("LiveNode".to_string(), self.config.clone())?;
-        let runner = AsyncRunner::new(clock.clone());
+        let (runner, signal_tx) = AsyncRunner::new(clock.clone());
 
         // Create and register data clients
         for (name, factory) in self.data_client_factories.into_iter() {
@@ -471,6 +474,7 @@ impl LiveNodeBuilder {
             clock,
             kernel,
             runner,
+            signal_tx: Some(signal_tx),
             config: self.config,
             is_running: false,
         })