Refactor FileCache to use Condvar for efficient waits

The `FileCache` now uses `std::sync::Condvar` to signal when files have
finished loading. This replaces the previous spin-wait mechanism,
significantly improving efficiency by allowing waiting threads to sleep
until explicitly woken up.

The `wait_for_load` method now blocks on the `Condvar`, and the
`insert_m1` and `insert_tick` methods call `notify_all` after updating
the cache and removing the loading guard. This ensures that all waiting
threads are woken up and can re-check the cache.

Additionally, `ensure_m1_loaded` and `ensure_tick_loaded` in
`DataServer` have been simplified to directly call `wait_for_load` when
another thread is already loading the file, removing the redundant
synchronous loading logic.
Refactor FileCache to use Condvar

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<T> = RwLock<HashMap<FileKey, ChunkVec<T>>>;
 /// 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<M1Parsed>,
     tick: CacheMap<TickParsed>,
     /// 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<HashSet<FileKey>>,
+    /// 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<Arc<[M1Parsed]>>) {
         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<Arc<[TickParsed]>>) {
         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());
+    }
 }

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<Arc<DataServer>> = OnceLock::new();
 
@@ -50,6 +53,20 @@ pub fn init_data_server(base_path: impl AsRef<Path>) -> Arc<DataServer> {
     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<Arc<DataServer>> {
     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;

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 &current_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 &current_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\"))"]);
+}

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

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<PathBuf>,
 }
 
 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);

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

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<u64> = 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]);
+}