Brummel/RustAst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Specialize pipeline types for performance

Browse Source 
Introduces type specialization for reactive pipeline nodes and their
data buffers. This eliminates the overhead of generic `Value` enums and
`SharedValueSeries` when dealing with known scalar or record types.

The architecture shifts buffer instantiation from the VM to the runtime
(RTL), leveraging type information from the AST. A new `out_type` field
is added to `BoundKind::Pipe` to store the static type of the pipeline's
output, determined by the type checker.

This enables the creation of specialized `RingBuffer<T>` and
`SharedRecordSeries` (for Struct-of-Arrays layout) when the output type
is known, significantly improving memory usage and processing speed for
time-series data, especially in financial analysis.
This commit is contained in:
Michael Schimmel
2026-03-02 22:03:24 +01:00
parent a4af142719
commit f7cb6655af
14 changed files with 295 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ast/compiler/analyzer.rs
+2 -1
View File
@@ -237,7 +237,7 @@ impl<'a> Analyzer<'a> {
Purity::Impure,
)
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, out_type } => {
let mut analyzed_inputs = Vec::with_capacity(inputs.len());
for input in inputs {
analyzed_inputs.push(self.visit(Rc::new(input.clone())));
@@ -247,6 +247,7 @@ impl<'a> Analyzer<'a> {
BoundKind::Pipe {
inputs: analyzed_inputs,
lambda: a_lambda,
out_type: out_type.clone(),
},
Purity::Impure,
)
src/ast/compiler/binder.rs
+1
View File
@@ -287,6 +287,7 @@ impl Binder {
BoundKind::Pipe {
inputs: bound_inputs,
lambda: bound_lambda,
out_type: crate::ast::types::StaticType::Any,
},
))
}
src/ast/compiler/bound_nodes.rs
+1
View File
@@ -138,6 +138,7 @@ pub enum BoundKind<T = ()> {
Pipe {
inputs: Vec<BoundNode<T>>,
lambda: Box<BoundNode<T>>,
out_type: crate::ast::types::StaticType,
},
Block {
exprs: Vec<BoundNode<T>>,
src/ast/compiler/captures.rs
+2 -1
View File
@@ -89,7 +89,7 @@ impl CapturePass {
args: Box::new(Self::transform(*args, capture_map)),
};
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, out_type } => {
let mut t_inputs = Vec::with_capacity(inputs.len());
for input in inputs {
t_inputs.push(Self::transform(input, capture_map));
@@ -97,6 +97,7 @@ impl CapturePass {
node.kind = BoundKind::Pipe {
inputs: t_inputs,
lambda: Box::new(Self::transform(*lambda, capture_map)),
out_type: out_type.clone(),
};
}
BoundKind::Block { exprs } => {
src/ast/compiler/dumper.rs
+1 -1
View File
@@ -203,7 +203,7 @@ impl Dumper {
self.visit(args);
self.indent -= 1;
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, .. } => {
self.log("Pipe", node);
self.indent += 1;
for input in inputs {
src/ast/compiler/optimizer/engine.rs
+2 -1
View File
@@ -386,7 +386,7 @@ impl Optimizer {
)
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, out_type } => {
let mut o_inputs = Vec::with_capacity(inputs.len());
for input in inputs {
o_inputs.push(self.visit_node(input, sub, path));
@@ -396,6 +396,7 @@ impl Optimizer {
BoundKind::Pipe {
inputs: o_inputs,
lambda: o_lambda,
out_type: out_type.clone(),
},
node.ty.clone(),
)
src/ast/compiler/optimizer/utils.rs
+1 -1
View File
@@ -152,7 +152,7 @@ impl UsageInfo {
BoundKind::Again { args } => {
self.collect(args);
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, .. } => {
for input in inputs {
self.collect(input);
}
src/ast/compiler/tco.rs
+2 -1
View File
@@ -48,7 +48,7 @@ impl TCO {
args: Box::new(Self::transform(Rc::new((**args).clone()), false)),
}
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, out_type } => {
let mut t_inputs = Vec::with_capacity(inputs.len());
for input in inputs {
t_inputs.push(Self::transform(Rc::new(input.clone()), false));
@@ -56,6 +56,7 @@ impl TCO {
BoundKind::Pipe {
inputs: t_inputs,
lambda: Box::new(Self::transform(Rc::new((**lambda).clone()), false)),
out_type: out_type.clone(),
}
}
src/ast/compiler/type_checker.rs
+16 -3
View File
@@ -387,12 +387,24 @@ impl TypeChecker {
)
}
BoundKind::Pipe { inputs, lambda } => {
BoundKind::Pipe { inputs, lambda, .. } => {
let mut typed_inputs = Vec::with_capacity(inputs.len());
let mut arg_types = Vec::with_capacity(inputs.len());
for input in inputs {
typed_inputs.push(self.check_node(input, ctx)?);
let typed_input = self.check_node(input, ctx)?;
// Unwrap Series(T) to T for the lambda argument
let arg_ty = if let StaticType::Series(inner) = &typed_input.ty {
*inner.clone()
} else {
StaticType::Any
};
arg_types.push(arg_ty);
typed_inputs.push(typed_input);
}
let typed_lambda = self.check_node(*lambda, ctx)?;
// Specialized check for the lambda using the input types!
let typed_lambda = self.check(*lambda, &arg_types)?;
let ret_ty = if let StaticType::Function(sig) = &typed_lambda.ty {
// If the lambda returns an Optional(T), the pipeline filters Void and stores T!
if let StaticType::Optional(inner) = &sig.ret {
@@ -407,6 +419,7 @@ impl TypeChecker {
BoundKind::Pipe {
inputs: typed_inputs,
lambda: Box::new(typed_lambda),
out_type: ret_ty.clone(),
},
StaticType::Series(Box::new(ret_ty)),
)
src/ast/rtl/streams.rs
+145 -3
View File
@@ -240,11 +240,14 @@ impl Observer for PipeStream {
pub struct SeriesPusher<T: crate::ast::rtl::series::ScalarValue> {
pub buffer: Rc<RefCell<crate::ast::rtl::series::RingBuffer<T>>>,
pub lookback: Option<usize>,
pub extractor: fn(Value) -> Option<T>,
}
impl<T: crate::ast::rtl::series::ScalarValue> Observer for SeriesPusher<T> {
fn notify(&mut self, _source_index: usize, _cycle_id: u64, _value: Value) {
// ... (Downcast and push logic)
fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) {
if let Some(v) = (self.extractor)(value) {
self.buffer.borrow_mut().push(v, self.lookback);
}
}
}
@@ -260,6 +263,137 @@ impl Observer for ValuePusher {
}
}
/// A specialized observer that splits a Record into its fields and pushes them into SoA buffers.
pub struct RecordPusher {
pub field_buffers: Vec<Rc<RefCell<dyn crate::ast::rtl::series::SeriesMember>>>,
pub lookback: Option<usize>,
}
impl Observer for RecordPusher {
fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) {
if let Value::Record(_, values) = value {
for (i, v) in values.iter().enumerate() {
if let Some(buf) = self.field_buffers.get(i) {
buf.borrow_mut().push_value(v.clone(), self.lookback);
}
}
}
}
}
/// Factory function to build a specialized pipeline node based on the output type.
/// This keeps the VM "dumb" and moves the buffer selection logic to the RTL.
pub fn build_pipeline_node(
inputs: Vec<Rc<dyn ObservableStream>>,
executor: Box<dyn FnMut(Vec<Value>) -> Value>,
out_type: &StaticType,
) -> Rc<PipelineNode> {
use crate::ast::rtl::series::*;
let pipe = Rc::new(RefCell::new(PipeStream::new(
"pipe".to_string(),
inputs.len(),
Some(executor),
)));
// Connect inputs to the pipe
for (i, input) in inputs.into_iter().enumerate() {
let adapter = Rc::new(RefCell::new(SourceAdapter {
target: pipe.clone(),
target_index: i,
}));
input.add_observer(adapter);
}
let series: Rc<dyn crate::ast::types::Series> = match out_type {
StaticType::Float => {
let buffer = Rc::new(RefCell::new(RingBuffer::<f64>::new()));
let pusher = Rc::new(RefCell::new(SeriesPusher {
buffer: buffer.clone(),
lookback: Some(100), // TODO: Configuration
extractor: |v| if let Value::Float(f) = v { Some(f) } else { None },
}));
pipe.borrow().add_observer(pusher);
Rc::new(SharedSeries {
buffer,
type_name: "SharedFloatSeries",
converter: |v| Value::Float(v),
})
}
StaticType::Int | StaticType::DateTime => {
let buffer = Rc::new(RefCell::new(RingBuffer::<i64>::new()));
let pusher = Rc::new(RefCell::new(SeriesPusher {
buffer: buffer.clone(),
lookback: Some(100),
extractor: |v| match v {
Value::Int(i) => Some(i),
Value::DateTime(i) => Some(i),
_ => None,
},
}));
pipe.borrow().add_observer(pusher);
Rc::new(SharedSeries {
buffer,
type_name: "SharedIntSeries",
converter: |v| Value::Int(v),
})
}
StaticType::Bool => {
let buffer = Rc::new(RefCell::new(RingBuffer::<bool>::new()));
let pusher = Rc::new(RefCell::new(SeriesPusher {
buffer: buffer.clone(),
lookback: Some(100),
extractor: |v| if let Value::Bool(b) = v { Some(b) } else { None },
}));
pipe.borrow().add_observer(pusher);
Rc::new(SharedSeries {
buffer,
type_name: "SharedBoolSeries",
converter: |v| Value::Bool(v),
})
}
StaticType::Record(layout) => {
let mut field_buffers = Vec::with_capacity(layout.fields.len());
for (_, ty) in &layout.fields {
let member: Rc<RefCell<dyn SeriesMember>> = match ty {
StaticType::Float => Rc::new(RefCell::new(ScalarSeries::<f64>::new("FloatSeries"))),
StaticType::Int | StaticType::DateTime => {
Rc::new(RefCell::new(ScalarSeries::<i64>::new("IntSeries")))
}
StaticType::Bool => Rc::new(RefCell::new(ScalarSeries::<bool>::new("BoolSeries"))),
_ => Rc::new(RefCell::new(ValueSeries::new())),
};
field_buffers.push(member);
}
let pusher = Rc::new(RefCell::new(RecordPusher {
field_buffers: field_buffers.clone(),
lookback: Some(100),
}));
pipe.borrow().add_observer(pusher);
Rc::new(SharedRecordSeries {
layout: layout.clone(),
field_buffers,
})
}
_ => {
let buffer = Rc::new(RefCell::new(RingBuffer::<Value>::new()));
let pusher = Rc::new(RefCell::new(ValuePusher {
buffer: buffer.clone(),
lookback: Some(100),
}));
pipe.borrow().add_observer(pusher);
Rc::new(SharedValueSeries { buffer })
}
};
Rc::new(PipelineNode {
stream: pipe,
series,
})
}
// ============================================================================
// Script Integration (RTL Registration)
// ============================================================================
@@ -271,11 +405,19 @@ pub fn register(env: &Environment) {
// (create-random-ohlc seed limit) -> StreamNode
let generators = env.pipeline_generators.clone();
// Define the OHLC layout for typing
let ohlc_layout = RecordLayout::get_or_create(vec![
(Keyword::intern("open"), StaticType::Float),
(Keyword::intern("high"), StaticType::Float),
(Keyword::intern("low"), StaticType::Float),
(Keyword::intern("close"), StaticType::Float),
]);
env.register_native_fn(
"create-random-ohlc",
StaticType::Function(Box::new(Signature {
params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]),
ret: StaticType::Any, // Returns a StreamNode
ret: StaticType::Series(Box::new(StaticType::Record(ohlc_layout.clone()))),
})),
Purity::Impure, // Modifies global generator registry
move |args: std::vec::Vec<Value>| {
src/ast/vm.rs
+24 -44
View File
@@ -368,10 +368,13 @@ impl VM {
Ok(Value::Void)
}
}
BoundKind::Pipe { inputs, lambda } => {
use crate::ast::rtl::streams::{PipelineNode, PipeStream, StreamNode, ValuePusher, ObservableStream};
use crate::ast::rtl::series::{RingBuffer, SharedValueSeries};
BoundKind::Pipe {
inputs,
lambda,
out_type,
} => {
use crate::ast::rtl::streams::{PipelineNode, StreamNode};
let mut obs_streams = Vec::new();
for input in inputs {
let val = self.eval_internal(obs, input)?;
@@ -381,7 +384,10 @@ impl VM {
} else if let Some(p) = obj.as_any().downcast_ref::<PipelineNode>() {
obs_streams.push(p.stream.clone());
} else {
return Err(format!("Pipe input must be a stream, found {}", obj.type_name()));
return Err(format!(
"Pipe input must be a stream, found {}",
obj.type_name()
));
}
} else {
return Err("Pipe input must be an object (stream)".to_string());
@@ -398,46 +404,20 @@ impl VM {
// 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<dyn FnMut(Vec<Value>) -> Value> = Box::new(move |args: Vec<Value>| -> Value {
match pipe_vm.run_with_args(my_closure.clone(), args) {
Ok(res) => res,
Err(e) => panic!("Pipeline lambda execution failed: {}", e),
}
});
let executor: Box<dyn FnMut(Vec<Value>) -> Value> =
Box::new(move |args: Vec<Value>| -> 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(executor)
)));
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<RefCell<dyn crate::ast::rtl::streams::Observer>>,
target_index: i,
}));
stream.add_observer(adapter);
}
// Create the data buffer and pusher for the pipe's output
let buffer = Rc::new(RefCell::new(RingBuffer::<Value>::new()));
let pusher = Rc::new(RefCell::new(ValuePusher {
buffer: buffer.clone(),
lookback: Some(100), // Default lookback for now
}));
// The pipe pushes to the buffer
pipe.borrow().add_observer(pusher);
// Create the Series view for script access
let series = Rc::new(SharedValueSeries {
buffer,
});
let node = Rc::new(PipelineNode {
stream: pipe,
series,
});
// Delegate to the RTL Factory for specialized buffer instantiation
let node = crate::ast::rtl::streams::build_pipeline_node(
obs_streams,
executor,
out_type,
);
Ok(Value::Object(node))
}