diff --git a/src/ast/data_server/cache.rs b/src/ast/data_server/cache.rs index 1c44407..e35aaf9 100644 --- a/src/ast/data_server/cache.rs +++ b/src/ast/data_server/cache.rs @@ -2,11 +2,12 @@ //! //! Designed for a many-readers / rare-writer pattern: hundreds of VM threads //! read concurrently via `RwLock`, while the loader writes each file exactly -//! once. Duplicate loads are prevented by a `loading` guard set. +//! once. Duplicate loads are prevented by a `loading` guard set, and waiting +//! threads are woken via `Condvar` (not spin-waiting). use super::records::{DataFormat, M1Parsed, TickParsed}; use std::collections::{HashMap, HashSet}; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::{Arc, Condvar, Mutex, RwLock}; /// Uniquely identifies one data file on disk. #[derive(Clone, Debug, Hash, Eq, PartialEq)] @@ -28,13 +29,17 @@ type CacheMap = RwLock>>; /// The outer `Arc` allows the entire file to be shared without copying, /// and individual chunk `Arc`s let iterators hold references to just the /// chunk they are consuming. +/// +/// Waiting threads are notified via a [`Condvar`] when a file finishes +/// loading — no spin-waiting required. pub struct FileCache { m1: CacheMap, tick: CacheMap, /// Tracks files currently being loaded to prevent duplicate work. - /// A key is inserted before loading starts and removed after the - /// result is written to the cache. loading: Mutex>, + /// Signaled whenever a file finishes loading (insert completes). + /// Waiters re-check `is_cached` after each wake-up. + loaded: Condvar, } impl Default for FileCache { @@ -49,6 +54,7 @@ impl FileCache { m1: RwLock::new(HashMap::new()), tick: RwLock::new(HashMap::new()), loading: Mutex::new(HashSet::new()), + loaded: Condvar::new(), } } @@ -59,13 +65,15 @@ impl FileCache { self.m1.read().unwrap().get(key).cloned() } - /// Inserts M1 chunks into the cache and clears the loading guard. + /// Inserts M1 chunks into the cache, clears the loading guard, and + /// notifies all waiting threads. pub fn insert_m1(&self, key: FileKey, chunks: Vec>) { self.m1 .write() .unwrap() .insert(key.clone(), Arc::new(chunks)); self.loading.lock().unwrap().remove(&key); + self.loaded.notify_all(); } // -- Tick ---------------------------------------------------------------- @@ -75,13 +83,15 @@ impl FileCache { self.tick.read().unwrap().get(key).cloned() } - /// Inserts tick chunks into the cache and clears the loading guard. + /// Inserts tick chunks into the cache, clears the loading guard, and + /// notifies all waiting threads. pub fn insert_tick(&self, key: FileKey, chunks: Vec>) { self.tick .write() .unwrap() .insert(key.clone(), Arc::new(chunks)); self.loading.lock().unwrap().remove(&key); + self.loaded.notify_all(); } // -- Loading guard ------------------------------------------------------- @@ -91,11 +101,7 @@ impl FileCache { /// thread is already loading it or it is already cached. pub fn try_claim_loading(&self, key: &FileKey) -> bool { // Fast path: already cached? - if key.format == DataFormat::M1 { - if self.m1.read().unwrap().contains_key(key) { - return false; - } - } else if self.tick.read().unwrap().contains_key(key) { + if self.is_cached(key) { return false; } // Slow path: try to claim the loading slot @@ -115,20 +121,16 @@ impl FileCache { } } - /// Spins briefly until the key is no longer in the loading set, i.e. - /// another thread has finished loading it. Returns `true` if the key - /// is now cached, `false` on timeout. - pub fn wait_for_load(&self, key: &FileKey) -> bool { - // Simple spin-wait with yield. In practice, file loads take - // milliseconds and this is only hit during prefetch races. - for _ in 0..10_000 { - if self.is_cached(key) { - return true; - } - std::thread::yield_now(); + /// Blocks until the given key is present in the cache. + /// + /// Uses a [`Condvar`] to sleep efficiently — the thread is woken when + /// any file finishes loading. This replaces the earlier spin-wait and + /// is equivalent to Delphi's `TEvent.WaitFor()`. + pub fn wait_for_load(&self, key: &FileKey) { + let mut guard = self.loading.lock().unwrap(); + while !self.is_cached(key) && guard.contains(key) { + guard = self.loaded.wait(guard).unwrap(); } - // Fallback: still not cached, caller should load it themselves. - false } } @@ -205,4 +207,30 @@ mod tests { cache.insert_tick(tick_key.clone(), vec![]); assert!(cache.is_cached(&tick_key)); } + + #[test] + fn test_condvar_wakeup() { + let cache = Arc::new(FileCache::new()); + let key = test_key("TEST", 2020, 1, DataFormat::M1); + + // Claim loading on main thread + assert!(cache.try_claim_loading(&key)); + + // Spawn a waiter thread + let cache2 = Arc::clone(&cache); + let key2 = key.clone(); + let handle = std::thread::spawn(move || { + cache2.wait_for_load(&key2); + cache2.is_cached(&key2) + }); + + // Give the waiter time to block + std::thread::sleep(std::time::Duration::from_millis(10)); + + // Insert data — this wakes the waiter + cache.insert_m1(key, vec![]); + + // Waiter should now see the cached data + assert!(handle.join().unwrap()); + } } diff --git a/src/ast/data_server/mod.rs b/src/ast/data_server/mod.rs index 9d8f544..6cd4e46 100644 --- a/src/ast/data_server/mod.rs +++ b/src/ast/data_server/mod.rs @@ -30,6 +30,9 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Arc, OnceLock}; +/// Default path for Pepperstone tick data files. +pub const DEFAULT_DATA_PATH: &str = "/mnt/tickdata/Pepperstone"; + /// Global data server instance, initialized once before any VM threads start. static DATA_SERVER: OnceLock> = OnceLock::new(); @@ -50,6 +53,20 @@ pub fn init_data_server(base_path: impl AsRef) -> Arc { server } +/// Initializes the global data server with [`DEFAULT_DATA_PATH`] if the +/// directory exists. Does nothing if the path is missing or the server +/// is already initialized. Safe to call multiple times. +pub fn init_default_data_server() { + if DATA_SERVER.get().is_some() { + return; + } + if Path::new(DEFAULT_DATA_PATH).is_dir() { + // Ignore the error if another thread raced us + let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let _ = DATA_SERVER.set(server); + } +} + /// Returns the global data server, or `None` if not yet initialized. pub fn get_data_server() -> Option> { DATA_SERVER.get().cloned() @@ -233,26 +250,19 @@ impl DataServer { // -- M1 loading ---------------------------------------------------------- /// Ensures the given M1 file is loaded into the cache (blocking). + /// + /// If another thread is already loading this file, blocks on a `Condvar` + /// until the load completes. If no one is loading, loads synchronously. fn ensure_m1_loaded(&self, key: &FileKey) { if self.cache.is_cached(key) { return; } - if self.cache.is_loading(key) { - self.cache.wait_for_load(key); - if self.cache.is_cached(key) { - return; - } - } - self.load_m1_sync(key); - } - - /// Synchronously loads an M1 file into the cache. - fn load_m1_sync(&self, key: &FileKey) { if !self.cache.try_claim_loading(key) { - // Another thread is loading or already cached — wait for it + // Another thread owns this load — block until it finishes. self.cache.wait_for_load(key); return; } + // We own the load — do it now. let path = self.index.file_path(key); match loader::load_m1_file(&path) { Ok(chunks) => self.cache.insert_m1(key.clone(), chunks), @@ -292,17 +302,6 @@ impl DataServer { if self.cache.is_cached(key) { return; } - if self.cache.is_loading(key) { - self.cache.wait_for_load(key); - if self.cache.is_cached(key) { - return; - } - } - self.load_tick_sync(key); - } - - /// Synchronously loads a tick file into the cache. - fn load_tick_sync(&self, key: &FileKey) { if !self.cache.try_claim_loading(key) { self.cache.wait_for_load(key); return; diff --git a/src/ast/rtl/data_streams.rs b/src/ast/rtl/data_streams.rs new file mode 100644 index 0000000..37464f9 --- /dev/null +++ b/src/ast/rtl/data_streams.rs @@ -0,0 +1,188 @@ +//! RTL registration for file-based market data streams. +//! +//! Bridges the `Arc`-based [`DataServer`] cache with the `Rc`-based VM world +//! by registering generator closures that read pre-parsed chunks and convert +//! them into `Value::Record` items for `RootStream::tick()`. + +use crate::ast::data_server::get_data_server; +use crate::ast::environment::Environment; +use crate::ast::rtl::streams::{RootStream, StreamNode}; +use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType, Value}; +use std::rc::Rc; + +/// Registers `create-m1-stream` and `create-tick-stream` as RTL functions. +pub fn register(env: &Environment) { + register_m1_stream(env); + register_tick_stream(env); +} + +// -- M1 Stream --------------------------------------------------------------- + +fn register_m1_stream(env: &Environment) { + let generators = env.pipeline_generators.clone(); + + let m1_layout = RecordLayout::get_or_create(vec![ + (Keyword::intern("time"), StaticType::DateTime), + (Keyword::intern("open"), StaticType::Float), + (Keyword::intern("high"), StaticType::Float), + (Keyword::intern("low"), StaticType::Float), + (Keyword::intern("close"), StaticType::Float), + (Keyword::intern("spread"), StaticType::Float), + (Keyword::intern("volume"), StaticType::Int), + ]); + + let ret_type = StaticType::Stream(Box::new(StaticType::Record(m1_layout.clone()))); + + env.register_native_fn( + "create-m1-stream", + StaticType::Function(Box::new(Signature { + params: StaticType::Tuple(vec![StaticType::Text]), + ret: ret_type, + })), + Purity::Impure, + move |args: &[Value]| { + let symbol = match &args[0] { + Value::Text(s) => s.as_ref(), + _ => panic!("create-m1-stream: first argument must be a string"), + }; + + let server = get_data_server() + .expect("create-m1-stream: DataServer not initialized. Call init_data_server() first."); + + let mut iter = server + .stream_m1(symbol) + .unwrap_or_else(|| panic!("create-m1-stream: symbol '{}' not found", symbol)); + + // Create the pipeline entry point + let root_stream = Rc::new(RootStream::new()); + let layout = m1_layout.clone(); + let stream_node = StreamNode { + inner: root_stream.clone(), + element_type: StaticType::Record(layout.clone()), + }; + + // Pre-fetch the first chunk + let mut current_chunk = iter.next_chunk(); + let mut pos = 0; + + let generator = move || -> bool { + loop { + let chunk = match ¤t_chunk { + Some(c) => c, + None => return false, + }; + + if pos < chunk.len() { + let rec = &chunk[pos]; + pos += 1; + + let value = Value::Record( + layout.clone(), + Rc::new(vec![ + Value::DateTime(rec.time_ms), + Value::Float(rec.open), + Value::Float(rec.high), + Value::Float(rec.low), + Value::Float(rec.close), + Value::Float(rec.spread), + Value::Int(rec.volume), + ]), + ); + root_stream.tick(value); + return true; + } + + // Advance to next chunk (triggers prefetch internally) + current_chunk = iter.next_chunk(); + pos = 0; + } + }; + + generators.borrow_mut().push(Box::new(generator)); + Value::Stream(Rc::new(stream_node)) + }, + ) + .doc("Creates a stream of M1 (minute) OHLCV bars from tick data files.") + .description("Reads pre-cached binary data from the DataServer. The symbol must match a directory entry (e.g. \"EURUSD\", \"AAPL.US\").") + .examples(&["(def ohlc (create-m1-stream \"EURUSD\"))"]); +} + +// -- Tick Stream ------------------------------------------------------------- + +fn register_tick_stream(env: &Environment) { + let generators = env.pipeline_generators.clone(); + + let tick_layout = RecordLayout::get_or_create(vec![ + (Keyword::intern("time"), StaticType::DateTime), + (Keyword::intern("ask"), StaticType::Float), + (Keyword::intern("bid"), StaticType::Float), + ]); + + let ret_type = StaticType::Stream(Box::new(StaticType::Record(tick_layout.clone()))); + + env.register_native_fn( + "create-tick-stream", + StaticType::Function(Box::new(Signature { + params: StaticType::Tuple(vec![StaticType::Text]), + ret: ret_type, + })), + Purity::Impure, + move |args: &[Value]| { + let symbol = match &args[0] { + Value::Text(s) => s.as_ref(), + _ => panic!("create-tick-stream: first argument must be a string"), + }; + + let server = get_data_server() + .expect("create-tick-stream: DataServer not initialized."); + + let mut iter = server + .stream_tick(symbol) + .unwrap_or_else(|| panic!("create-tick-stream: symbol '{}' not found", symbol)); + + let root_stream = Rc::new(RootStream::new()); + let layout = tick_layout.clone(); + let stream_node = StreamNode { + inner: root_stream.clone(), + element_type: StaticType::Record(layout.clone()), + }; + + let mut current_chunk = iter.next_chunk(); + let mut pos = 0; + + let generator = move || -> bool { + loop { + let chunk = match ¤t_chunk { + Some(c) => c, + None => return false, + }; + + if pos < chunk.len() { + let rec = &chunk[pos]; + pos += 1; + + let value = Value::Record( + layout.clone(), + Rc::new(vec![ + Value::DateTime(rec.time_ms), + Value::Float(rec.ask), + Value::Float(rec.bid), + ]), + ); + root_stream.tick(value); + return true; + } + + current_chunk = iter.next_chunk(); + pos = 0; + } + }; + + generators.borrow_mut().push(Box::new(generator)); + Value::Stream(Rc::new(stream_node)) + }, + ) + .doc("Creates a stream of tick data (ask/bid) from tick data files.") + .description("Reads pre-cached binary data from the DataServer. The symbol must match a directory entry.") + .examples(&["(def ticks (create-tick-stream \"EURUSD\"))"]); +} diff --git a/src/ast/rtl/mod.rs b/src/ast/rtl/mod.rs index 9802a42..65c4de6 100644 --- a/src/ast/rtl/mod.rs +++ b/src/ast/rtl/mod.rs @@ -1,4 +1,5 @@ pub mod core; +pub mod data_streams; pub mod datetime; pub mod docs; pub mod intrinsics; @@ -13,6 +14,7 @@ use crate::ast::environment::Environment; pub fn register(env: &Environment) { core::register(env); + data_streams::register(env); datetime::register(env); math::register(env); series::register(env); diff --git a/src/bin/ast.rs b/src/bin/ast.rs index 8bc3c94..2b57848 100644 --- a/src/bin/ast.rs +++ b/src/bin/ast.rs @@ -53,11 +53,22 @@ struct Cli { /// Emit the parsed source back as Myc code (round-trip test) #[arg(long)] emit: bool, + + /// Path to tick data directory (overrides default /mnt/tickdata/Pepperstone) + #[arg(long)] + data_path: Option, } fn main() { let cli = Cli::parse(); + // Initialize data server (before Environment::new so RTL has access) + if let Some(path) = &cli.data_path { + myc::ast::data_server::init_data_server(path); + } else { + myc::ast::data_server::init_default_data_server(); + } + if cli.mcp { if let Err(e) = myc::ast::mcp_server::run() { eprintln!("MCP server error: {}", e); diff --git a/src/main.rs b/src/main.rs index 3129070..13b7101 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,6 +5,7 @@ use myc::ast::parser::Parser; use std::path::PathBuf; fn main() -> eframe::Result { + myc::ast::data_server::init_default_data_server(); let options = eframe::NativeOptions { viewport: egui::ViewportBuilder::default().with_inner_size([1024.0, 768.0]), ..Default::default() diff --git a/tests/data_server.rs b/tests/data_server.rs new file mode 100644 index 0000000..a739371 --- /dev/null +++ b/tests/data_server.rs @@ -0,0 +1,159 @@ +//! Integration tests for the DataServer, verifying correct parsing of real +//! tick data files from /mnt/tickdata/Pepperstone/. + +use std::path::Path; +use std::sync::Arc; + +const TICKDATA_PATH: &str = "/mnt/tickdata/Pepperstone"; + +/// Skip tests when the tick data directory is not available. +fn skip_if_no_data() -> bool { + !Path::new(TICKDATA_PATH).exists() +} + +#[test] +fn test_real_symbol_index() { + if skip_if_no_data() { + return; + } + let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH)); + let symbols = server.symbols(); + + assert!(!symbols.is_empty(), "Should find at least one symbol"); + assert!( + symbols.iter().any(|s| s.as_ref() == "EURUSD"), + "EURUSD should be in the symbol list" + ); + + // Verify file count for a known symbol + let m1_count = server + .file_count("EURUSD", myc::ast::data_server::records::DataFormat::M1) + .unwrap(); + assert!(m1_count > 0, "EURUSD should have M1 files, found {m1_count}"); +} + +#[test] +fn test_real_m1_stream_eurusd() { + if skip_if_no_data() { + return; + } + let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH)); + let mut iter = server.stream_m1("EURUSD").expect("EURUSD M1 stream"); + + // Read the first chunk + let first_chunk = iter.next_chunk().expect("At least one chunk"); + assert!(!first_chunk.is_empty()); + + let rec = &first_chunk[0]; + // Verify the record has sane values + assert!(rec.time_ms > 0, "Timestamp should be positive"); + assert!(rec.open > 0.0, "EURUSD open should be > 0"); + assert!(rec.high >= rec.low, "High should be >= Low"); + assert!(rec.close > 0.0, "EURUSD close should be > 0"); + + // EURUSD is always between ~0.8 and ~1.6 + assert!( + (0.5..2.0).contains(&rec.open), + "EURUSD open should be between 0.5 and 2.0, got {}", + rec.open + ); + + println!( + "First EURUSD M1 record: time_ms={}, O={:.5} H={:.5} L={:.5} C={:.5} S={:.1} V={}", + rec.time_ms, rec.open, rec.high, rec.low, rec.close, rec.spread, rec.volume + ); +} + +#[test] +fn test_real_tick_stream_eurusd() { + if skip_if_no_data() { + return; + } + let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH)); + + // EURUSD tick data may not exist — check gracefully + let Some(mut iter) = server.stream_tick("EURUSD") else { + println!("No EURUSD tick data found, skipping"); + return; + }; + + let first_chunk = iter.next_chunk().expect("At least one tick chunk"); + let rec = &first_chunk[0]; + + assert!(rec.time_ms > 0); + assert!(rec.ask > 0.0); + assert!(rec.bid > 0.0); + // Spread should be small for EURUSD + let spread = (rec.ask - rec.bid).abs(); + assert!( + spread < 0.01, + "EURUSD spread should be < 100 pips, got {spread}" + ); + + println!( + "First EURUSD tick: time_ms={}, Ask={:.5} Bid={:.5} Spread={:.5}", + rec.time_ms, rec.ask, rec.bid, spread + ); +} + +#[test] +fn test_real_m1_full_iteration() { + if skip_if_no_data() { + return; + } + let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH)); + + // Use a symbol with few files for speed (AAPL.US starts 2006) + let Some(mut iter) = server.stream_m1("AAPL.US") else { + println!("No AAPL.US data found, skipping"); + return; + }; + + let mut total_records = 0u64; + let mut chunks = 0u64; + while let Some(chunk) = iter.next_chunk() { + chunks += 1; + total_records += chunk.len() as u64; + } + + println!("AAPL.US M1: {total_records} records in {chunks} chunks"); + assert!(total_records > 0, "Should have read some records"); +} + +#[test] +fn test_real_concurrent_access() { + if skip_if_no_data() { + return; + } + let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH)); + + // Spawn 4 threads all reading the same symbol simultaneously + let handles: Vec<_> = (0..4) + .map(|_| { + let srv = Arc::clone(&server); + std::thread::spawn(move || { + let mut iter = srv.stream_m1("EURUSD").expect("EURUSD"); + let mut count = 0u64; + // Read first 10 chunks only (for speed) + for _ in 0..10 { + if let Some(chunk) = iter.next_chunk() { + count += chunk.len() as u64; + } else { + break; + } + } + count + }) + }) + .collect(); + + let counts: Vec = handles.into_iter().map(|h| h.join().unwrap()).collect(); + + // All threads should read the same number of records + assert!(counts[0] > 0); + assert!( + counts.iter().all(|&c| c == counts[0]), + "All threads should read the same amount: {counts:?}" + ); + println!("Concurrent access: 4 threads each read {} records", counts[0]); +}