From 2457eec1bfac4b1792e2eed77ff7e879c3ed496b Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 2 Mar 2026 00:32:07 +0100 Subject: [PATCH] feat: Implement Pipeline support Adds support for pipelines, allowing streams to be chained together and transformed. This includes new types for `PipeStream` and `SourceAdapter`, along with modifications to the `Environment` to manage pipeline execution. A new example `pipeline.myc` demonstrates its usage. --- examples/pipeline.myc | 23 ++++++++++++++ examples/soa_series.myc | 4 +-- src/ast/compiler/binder.rs | 2 +- src/ast/environment.rs | 11 +++++-- src/ast/rtl/streams.rs | 62 ++++++++++++++++++++++++++------------ src/ast/types.rs | 10 +++++- src/ast/vm.rs | 20 ++++++++++-- 7 files changed, 103 insertions(+), 29 deletions(-) create mode 100644 examples/pipeline.myc diff --git a/examples/pipeline.myc b/examples/pipeline.myc new file mode 100644 index 0000000..24d19b9 --- /dev/null +++ b/examples/pipeline.myc @@ -0,0 +1,23 @@ +;; Output: (does not matter) + +(do + (def src1 (create-random-ohlc 42 3)) + + (def combined + (pipe [src1] + (fn [t1] + (.close t1) + ) + ) + ) + + (def my_indicator + (pipe [combined] + (fn [close_price] + (+ close_price 10.0) + ) + ) + ) + + my_indicator +) \ No newline at end of file diff --git a/examples/soa_series.myc b/examples/soa_series.myc index 64d0d42..119c863 100644 --- a/examples/soa_series.myc +++ b/examples/soa_series.myc @@ -1,6 +1,6 @@ ;; Output: 26.7 -;; Benchmark: 998ns -;; Benchmark-Repeat: 2047 +;; Benchmark: 1.1us +;; Benchmark-Repeat: 1753 (do (def template {:price 10.0 :volume 100 :msg "hi"}) (def my_ticks (create-series template)) diff --git a/src/ast/compiler/binder.rs b/src/ast/compiler/binder.rs index 6f73328..da82f4e 100644 --- a/src/ast/compiler/binder.rs +++ b/src/ast/compiler/binder.rs @@ -279,7 +279,7 @@ impl Binder { UntypedKind::Pipe { inputs, lambda } => { let mut bound_inputs = Vec::with_capacity(inputs.len()); for input in inputs { - bound_inputs.push(self.bind(&input)?); + bound_inputs.push(self.bind(input)?); } let bound_lambda = Box::new(self.bind(lambda.as_ref())?); Ok(self.make_node( diff --git a/src/ast/environment.rs b/src/ast/environment.rs index 46c361a..980eb08 100644 --- a/src/ast/environment.rs +++ b/src/ast/environment.rs @@ -21,6 +21,8 @@ use crate::ast::types::{ Identity, NodeIdentity, Object, Purity, SourceLocation, StaticType, Value, }; +pub type PipelineGenerator = Box bool>; + pub struct Environment { pub global_names: Rc>>, pub global_types: Rc>>, @@ -33,7 +35,7 @@ pub struct Environment { pub debug_mode: bool, pub optimization: bool, pub macro_registry: Rc>, - pub pipeline_generators: Rc bool>>>>, + pub pipeline_generators: Rc>>, } struct EnvFunctionRegistry { @@ -394,7 +396,9 @@ impl Environment { let compiled = self.compile(source)?; let linked = self.link(compiled); let func = self.instantiate(linked); - Ok((func.func)(vec![])) + let res = (func.func)(vec![]); + self.run_pipeline(); + Ok(res) } } @@ -423,6 +427,9 @@ impl Environment { break; } } + + self.run_pipeline(); + Ok((result, observer.logs)) } } diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index a6bc2ad..f12df26 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -24,6 +24,19 @@ pub trait Observer { fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value); } +/// A lightweight adapter to map an unknown source index to a specific target index. +/// This prevents index collisions when a Pipe listens to multiple independent RootStreams. +pub struct SourceAdapter { + pub target: Rc>, + pub target_index: usize, +} + +impl Observer for SourceAdapter { + fn notify(&mut self, _ignored_source: usize, cycle_id: u64, value: Value) { + self.target.borrow_mut().notify(self.target_index, cycle_id, value); + } +} + /// Polymorphic Interface for any stream that can accept observers (like Delphi's IStream). pub trait ObservableStream { fn add_observer(&self, observer: Rc>); @@ -95,6 +108,12 @@ impl ObservableStream for RootStream { } } +impl Default for RootStream { + fn default() -> Self { + Self::new() + } +} + impl RootStream { pub fn new() -> Self { Self { @@ -136,32 +155,33 @@ pub struct PipeStream { /// For the MVP, we assume the Pipe is notified by the Root or its parents. pub input_count: usize, /// Tracks the last cycle_id received from each input. - last_cycle_per_input: RefCell>, + last_cycle_per_input: Vec, + /// Stores the current value for each input to construct the argument tuple. + current_values: Vec, /// The current output signal of this pipe. current_signal: RefCell>, - /// The VM closure (lambda) to execute. - pub lambda: Option>, + /// The executable closure representing the Lambda. Expects a Vec of arguments. + pub executor: Option) -> Value>>, /// Observers of THIS pipe. observers: RefCell>>>, } impl PipeStream { - pub fn new(name: String, input_count: usize, lambda: Option>) -> Self { + pub fn new( + name: String, + input_count: usize, + executor: Option) -> Value>> + ) -> Self { Self { name, input_count, - last_cycle_per_input: RefCell::new(vec![0; input_count]), + last_cycle_per_input: vec![0; input_count], + current_values: vec![Value::Void; input_count], current_signal: RefCell::new(None), - lambda, + executor, observers: RefCell::new(Vec::new()), } } - - /// Internal: Checks if all inputs have reached the target cycle. - fn is_barrier_reached(&self, cycle_id: u64) -> bool { - let cycles = self.last_cycle_per_input.borrow(); - cycles.iter().all(|&c| c == cycle_id) - } } impl ObservableStream for PipeStream { @@ -179,22 +199,24 @@ impl Stream for PipeStream { impl Observer for PipeStream { fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value) { let barrier_reached = { - let mut cycles = self.last_cycle_per_input.borrow_mut(); if source_index < self.input_count { - cycles[source_index] = cycle_id; + self.last_cycle_per_input[source_index] = cycle_id; + self.current_values[source_index] = value; } // Check if all inputs reached the same cycle. - cycles.iter().all(|&c| c == cycle_id) + self.last_cycle_per_input.iter().all(|&c| c == cycle_id) }; if barrier_reached { // 1. Prepare Arguments for Lambda (Current values of all inputs) - let args = vec![value]; // Simplified for now + let args = self.current_values.clone(); - // 2. Execute Lambda (This requires a VM instance!) - // TODO: Actually execute the lambda using a VM context. - // For now, we mock the result to be the input value itself. - let result = args[0].clone(); + // 2. Execute Lambda using the encapsulated VM executor + let result = if let Some(exec) = &mut self.executor { + exec(args) + } else { + self.current_values[0].clone() // Identity bypass (defaults to first input) + }; // 3. Handle Void case! (Filter pattern) if matches!(result, Value::Void) { diff --git a/src/ast/types.rs b/src/ast/types.rs index ade336c..8a5922d 100644 --- a/src/ast/types.rs +++ b/src/ast/types.rs @@ -592,7 +592,15 @@ impl fmt::Display for Value { } Value::FieldAccessor(k) => write!(f, ".{}", k.name()), Value::Function(f_meta) => write!(f, "", f_meta.purity), - Value::Object(o) => write!(f, "<{}>", o.type_name()), + Value::Object(o) => { + if let Some(pipe) = o.as_any().downcast_ref::() { + write!(f, "PipelineNode[last: {:?}]", pipe.series.get_item(0)) + } else if let Some(val_series) = o.as_any().downcast_ref::() { + write!(f, "SharedValueSeries[len: {}]", val_series.buffer.borrow().len()) + } else { + write!(f, "<{}>", o.type_name()) + } + } Value::Cell(c) => write!(f, "{}", c.borrow()), Value::TailCallRequest(_) => write!(f, ""), } diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 244fcca..4e366a3 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -395,14 +395,28 @@ impl VM { return Err("Pipe lambda must be a function/closure".to_string()); }; + // Create the persistent execution closure for the PipeStream + let mut pipe_vm = crate::ast::vm::VM::new(self.globals.clone()); + let my_closure = lambda_obj.clone(); + let executor: Box) -> Value> = Box::new(move |args: Vec| -> Value { + match pipe_vm.run_with_args(my_closure.clone(), args) { + Ok(res) => res, + Err(e) => panic!("Pipeline lambda execution failed: {}", e), + } + }); + let pipe = Rc::new(RefCell::new(PipeStream::new( "pipe".to_string(), inputs.len(), - Some(lambda_obj) + Some(executor) ))); - for stream in obs_streams { - stream.add_observer(pipe.clone() as Rc>); + for (i, stream) in obs_streams.into_iter().enumerate() { + let adapter = Rc::new(RefCell::new(crate::ast::rtl::streams::SourceAdapter { + target: pipe.clone() as Rc>, + target_index: i, + })); + stream.add_observer(adapter); } // Create the data buffer and pusher for the pipe's output