From 8fdd673335c7c6c0459e4271bc1b1446a96e53b0 Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Fri, 31 Jul 2026 14:31:02 -0700 Subject: [PATCH 1/7] feat: implement parallel evaluation and scheduler, add new subcommand in main.rs for processing --- Cargo.lock | 10 ++ addblob_extended.fix | 31 ++++ fix/Cargo.toml | 5 + fix/src/main.rs | 127 +++++++++++++++++ fix/src/parallel_evaluator.rs | 261 ++++++++++++++++++++++++++++++++++ fix/src/scheduler.rs | 69 +++++++++ 6 files changed, 503 insertions(+) create mode 100644 addblob_extended.fix create mode 100644 fix/src/parallel_evaluator.rs create mode 100644 fix/src/scheduler.rs diff --git a/Cargo.lock b/Cargo.lock index 8b77985..164483f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -599,6 +599,15 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" +[[package]] +name = "crossbeam-queue" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -784,6 +793,7 @@ dependencies = [ "chrono", "cmake", "common", + "crossbeam-queue", "derive_more", "fixhandle", "fixshell", diff --git a/addblob_extended.fix b/addblob_extended.fix new file mode 100644 index 0000000..b6e285d --- /dev/null +++ b/addblob_extended.fix @@ -0,0 +1,31 @@ +a = create_blob(Int(1)); +b = create_blob(Int(2)); +c = create_blob(Int(3)); +d = create_blob(Int(4)); +e = create_blob(Int(5)); +f = create_blob(Int(6)); +g = create_blob(Int(7)); +h = create_blob(Int(8)); + +add = create_blob(Path("./target/x86_64-unknown-none/addblob")); + +ab = create_application_thunk(create_tree(add, a, b)); +cd = create_application_thunk(create_tree(add, c, d)); +ef = create_application_thunk(create_tree(add, e, f)); +gh = create_application_thunk(create_tree(add, g, h)); + +sum_ab = create_strict_encode(ab); +sum_cd = create_strict_encode(cd); +sum_ef = create_strict_encode(ef); +sum_gh = create_strict_encode(gh); + +left = create_application_thunk(create_tree(add, sum_ab, sum_cd)); +right = create_application_thunk(create_tree(add, sum_ef, sum_gh)); + +sum_left = create_strict_encode(left); +sum_right = create_strict_encode(right); + +final = create_application_thunk(create_tree(add, sum_left, sum_right)); + +eval(create_strict_encode(final)); + diff --git a/fix/Cargo.toml b/fix/Cargo.toml index 7de7790..6572031 100644 --- a/fix/Cargo.toml +++ b/fix/Cargo.toml @@ -37,6 +37,11 @@ bitfield-struct = "0.11.0" blake3 = { version = "1.8.5", default-features = false } hex = { version = "0.4.3", default-features = false, features = ["alloc"] } bitint = "0.1.1" +crossbeam-queue = { + version = "0.3.13", + default-features = false, + features = ["alloc"] +} [build-dependencies] fixshell = { path = "shell", artifact="staticlib", target = "x86_64-unknown-none" } diff --git a/fix/src/main.rs b/fix/src/main.rs index 0321127..0ce4c2a 100644 --- a/fix/src/main.rs +++ b/fix/src/main.rs @@ -1,5 +1,9 @@ #![no_main] #![no_std] + +mod scheduler; +mod parallel_evaluator; + use kernel::host::fs::{self, File, Whence}; use kernel::host::os; use kernel::prelude::*; @@ -21,6 +25,11 @@ fn main() { Some("eval") => { let filename = argv.get(2).expect("fix eval: expected a command file"); eval_file(filename); + }, + //test to run the parallel evaluator + Some("parallel_eval") => { + let filename = argv.get(2).expect("fix eval: expected a command file"); + eval_file_parallel(filename); } Some(other) => panic!("fix: unknown command '{other}' (expected: init | eval )"), None => panic!("fix: expected a command (init | eval )"), @@ -123,3 +132,121 @@ fn eval(evaluator: &Evaluator, e: &Expr, ctx: &mut BTreeMap eval(evaluator, x, ctx), } } + +// Jennifer: tons of redundancy but I just didn't want to change original code, +// in case errors showed up +// the main change is just calling the parallel evaluator +fn eval_file_parallel(filename: &str) { + let mut file = File::open(filename, true, false, false, false, false).unwrap(); + let len = file.seek(Whence::End(0)) as usize; + file.seek(Whence::Start(0)); + let mut buf = vec![0; len]; + file.read_exact(&mut buf); + + let file = core::str::from_utf8(&buf).unwrap(); + + let lexer = Lexer::new(&file); + let tokens = lexer.tokenize().unwrap(); + let mut parser = Parser::new(&tokens); + let program = parser.parse_program().unwrap(); + + let runtime = FixOnArca::default(); + let evaluator = parallel_evaluator::Evaluator::new(runtime); + + let mut context = BTreeMap::new(); + for statement in program { + match statement { + Statement::Assign { name, expr } => { + let result = eval_parallel(evaluator.as_ref(), &expr, &mut context); + context.insert(name, result); + } + Statement::Print(expr) | Statement::Expr(expr) => { + let x = eval_parallel(evaluator.as_ref(), &expr, &mut context); + match x { + Value::Handle(x) => { + println!("handle: {x}"); + if let Some(blob) = x + .try_unwrap_object() + .ok() + .and_then(|x| x.try_unwrap_blob().ok()) + { + let contents = evaluator.storage().get_blob(blob).unwrap(); + println!("result is a Blob: {contents:?}"); + if contents.len() == 8 { + let bytes: [u8; 8] = (*contents).try_into().unwrap(); + let value = u64::from_le_bytes(bytes); + println!("\tas a u64: {value}"); + } + } + } + Value::Int(x) => { + println!("int: {x}"); + } + Value::String(x) => { + println!("string: {x}"); + } + Value::Path(x) => { + println!("path: {x}"); + } + } + } + } + } +} + +fn eval_parallel(evaluator: ¶llel_evaluator::Evaluator, e: &Expr, ctx: &mut BTreeMap) -> Value { + match e { + Expr::Number(x) => Value::Int(*x), + Expr::Identifier(x) => ctx.get(x).expect("undefined identifier").clone(), + Expr::String(x) => Value::String(x.clone()), + Expr::Call { name, args } => { + let args: Vec = args.into_iter().map(|x| eval_parallel(evaluator, x, ctx)).collect(); + match name.as_str() { + "Int" => args[0].clone(), + "create_blob" => match args[0] { + Value::Handle(_) => panic!("create blob with handle?"), + Value::Int(x) => { + let bytes = i64::to_le_bytes(x); + Value::Handle(evaluator.storage().add_blob(&bytes).into()) + } + Value::String(ref x) => { + Value::Handle(evaluator.storage().add_blob(x.as_bytes()).into()) + } + Value::Path(ref x) => { + let mut file = File::open(x, true, false, false, false, false).unwrap(); + let len = file.seek(Whence::End(0)); + file.seek(Whence::Start(0)); + let mut buf = vec![0; len as usize]; + file.read_exact(&mut buf); + core::mem::forget(file); + Value::Handle(evaluator.storage().add_blob(&buf).into()) + } + }, + "create_tree" => { + let handles: Vec = args.into_iter().map(Value::unwrap_handle).collect(); + Value::Handle(evaluator.storage().add_tree(&handles).into()) + } + "create_application_thunk" => Value::Handle( + Thunk::Application( + args[0] + .clone() + .unwrap_handle() + .unwrap_object() + .unwrap_tree(), + ) + .into(), + ), + "create_strict_encode" => Value::Handle( + Encode::Strict(args[0].clone().unwrap_handle().unwrap_thunk()).into(), + ), + "eval" => Value::Handle(evaluator.eval(args[0].clone().unwrap_handle())), + "Path" => match args[0] { + Value::String(ref x) => Value::Path(x.clone()), + _ => panic!("bad path"), + }, + name => todo!("call {name} {args:?}"), + } + } + Expr::Group(x) => eval_parallel(evaluator, x, ctx), + } +} diff --git a/fix/src/parallel_evaluator.rs b/fix/src/parallel_evaluator.rs new file mode 100644 index 0000000..0ffc9ff --- /dev/null +++ b/fix/src/parallel_evaluator.rs @@ -0,0 +1,261 @@ +extern crate alloc; +use alloc::sync::Arc; +//use core::sync::atomic::{AtomicBool, Ordering}; +use kernel::{coreid, kthread}; +//use kernel::kthread::{KMutex, yield_now}; +//use kernel::tsc; +use crate::scheduler::{Scheduler, Task}; + + +use crate::handle::*; +use crate::runtime::Runtime; +use crate::storage::Storage; +use kernel::prelude::*; + +// use fixhandle::rawhandle::{Encode, Handle, Object, Ref, Thunk, TreeName}; + +// use fixruntime::{ +// common::CouponTrades, +// fixruntime::{FixRuntime, FixTreeData}, +// runtime::{DeterministicEquivRuntime, Executor}, +// storage::FixData, +// }; + +// use common::bitpack::BitPack; +// use kernel::prelude::*; + +const NUM_WORKERS: usize = 2; +#[derive(Clone, Copy)] +enum EvalType { + Parallel, Serial, +} +pub struct Evaluator { + runtime: R, + scheduler: Scheduler, +} + +impl Evaluator { + + pub fn new(runtime: R) -> Arc{ + let evaluator = Arc::new(Self { + runtime, + scheduler: Scheduler::new(), + }); + + evaluator.start_workers(NUM_WORKERS); + evaluator + } + + pub fn start_workers(self: &Arc, num_workers: usize) { + for i in 0..num_workers { + let evaluator = Arc::clone(self); + kthread::spawn(move || { + println!("worker {} started on core {}", i, coreid()); + evaluator.worker_loop(i); + } + ) + } + } + + pub fn worker_loop(self: &Arc, worker_id: usize) { + loop { + // eventually make this a queue to handle local queue of work + if let Some(work) = self.scheduler.get_work() { + println!( + "worker {} evaluating task on core {}", + worker_id, + coreid() + ); + + let result = self.eval_test(work.get_handle(), EvalType::Parallel); + work.task_complete(result) + } + else { + //change this to condition variable, so it does no busy waiting? + kthread::yield_now() + } + } + + } + + fn wait_while_helping(&self, target: &Arc) -> Handle { + loop { + // The task we are waiting for has completed. + if target.is_complete() { + return target.take_result(); + } + + if let Some(work) = self.scheduler.get_work() { + println!("calling thread helping on core {}", coreid()); + + let result = + self.eval_test(work.get_handle(), EvalType::Parallel); + + work.task_complete(result); + } else { + // Workers already claimed all available tasks. + kthread::yield_now(); + } + } +} + + pub fn runtime(&self) -> &R { + &self.runtime + } + + pub fn storage(&self) -> &dyn Storage { + self.runtime.storage() + } + + fn apply(&self, combination: Tree) -> Handle { + self.runtime.execute(combination) + } + + fn lift(&self, handle: Handle) -> Handle { + match handle { + Handle::Ref(r) => match r { + Ref::Tree(t) => Object::Tree(t).into(), + Ref::Blob(b) => Object::Blob(b).into(), + }, + _ => handle, + } + } + + fn lower(&self, handle: Handle) -> Handle { + match handle { + Handle::Object(r) => match r { + Object::Tree(t) => Ref::Tree(t).into(), + Object::Blob(b) => Ref::Blob(b).into(), + }, + _ => handle, + } + } + + fn think(&self, thunk: Thunk, eval_mode: EvalType) -> Handle { + match thunk { + Thunk::Identification(_) => todo!(), + Thunk::Selection(_) => todo!(), + Thunk::Application(tree) => { + let evaled = self.eval_tree(tree,eval_mode); + self.apply(evaled) + } + } + } + + fn force(&self, thunk: Thunk, eval_mode: EvalType) -> Handle { + let thought = self.think(thunk, eval_mode); + match thought { + Handle::Object(_) => thought, + Handle::Ref(_) => self.lift(thought), + Handle::Thunk(_) | Handle::Encode(_) => todo!(), + } + } + + fn encode(&self, encode: Encode, eval_mode: EvalType) -> Handle { + match encode { + Encode::Strict(thunk) => self.lift(self.force(thunk, eval_mode)), + Encode::Shallow(thunk) => self.lower(self.force(thunk, eval_mode)), + } + } + + + fn eval_tree(&self, handle: Tree, eval_mode: EvalType) -> Tree { + match eval_mode { + EvalType::Serial => self.eval_tree_seq(handle), + EvalType::Parallel => self.eval_tree_parallel(handle) + } + } + + fn eval_tree_seq(&self, handle: Tree) -> Tree { + let tree = self.runtime.storage().get_tree(handle).unwrap(); + let evaled: Vec = tree + .as_ref() + .iter() + .copied() + .map(|x| self.eval_test(x, EvalType::Serial)) + .collect(); + self.runtime.storage().add_tree(&evaled) + } + + fn eval_tree_parallel(&self, handle: Tree) -> Tree { + + let tree = self.runtime.storage().get_tree(handle).unwrap(); + if tree.len() <= 2 { + return self.eval_tree_seq(handle) + } + let mut evaled = Vec::with_capacity(tree.len()); + // evaluate the first + evaled.push(self.eval_test(tree[0], EvalType::Serial)); + let mut tasks = Vec::with_capacity(tree.len() - 1); + for child in tree[1..].iter().copied() { + tasks.push(self.scheduler.push_work(child)); + } + for task in tasks { + evaled.push(self.wait_while_helping(&task)); + } + self.runtime.storage().add_tree(&evaled) +} + // elimnate redundancy i think + pub fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle { + println!("evaluating {handle}"); + match handle { + Handle::Thunk(_) | Handle::Ref(_) => todo!(), + Handle::Object(obj) => match obj { + Object::Blob(x) => x.into(), + Object::Tree(tree) => self.eval_tree(tree, eval_mode).into(), + }, + Handle::Encode(e) => self.eval_test(self.encode(e, eval_mode), eval_mode), + } + } + + pub fn eval(&self, handle: Handle) -> Handle { + //let start = tsc::read_cycles(); + self.eval_test(handle, EvalType::Parallel) + //println!("total_cycles = {}", cycles_since(start)); + //result + } + + /* + fn eval_tree_seq(&self, handle: Tree) -> Tree { + let start = tsc::read_cycles(); + + println!( + "[timing] eval_tree_seq start on core {}: {:?}", + kernel::coreid(), + handle + ); + + let tree = self.runtime.storage().get_tree(handle).unwrap(); + + let evaled: Vec = tree + .as_ref() + .iter() + .copied() + .map(|x| { + let child_start = tsc::read_cycles(); + let result = self.eval(x); + + println!( + "[timing] seq child {:?} -> {:?}, cycles={}", + x, + result, + cycles_since(child_start) + ); + + result + }) + .collect(); + + let result_tree = self.runtime.storage().add_tree(&evaled); + + println!( + "[timing] eval_tree_seq done: {:?}, total_cycles={}", + handle, + cycles_since(start) + ); + + result_tree + } + */ + +} diff --git a/fix/src/scheduler.rs b/fix/src/scheduler.rs new file mode 100644 index 0000000..1437d66 --- /dev/null +++ b/fix/src/scheduler.rs @@ -0,0 +1,69 @@ +//use alloc::collections::VecDeque; +use alloc::sync::Arc; +use fix::Handle; +extern crate alloc; +use core::sync::atomic::{AtomicBool, Ordering}; +//use kernel::kthread; +use kernel::kthread::KMutex; + +use crossbeam_queue::SegQueue; + +//task that must be added +pub struct Task { + handle: Handle, + result: KMutex>, + done: AtomicBool, + +} +impl Task { + // create new task + fn new (value: Handle) -> Self{ + Self { + handle: value, + result: KMutex::new(None), + done: AtomicBool::new(false), + } + } + + pub fn task_complete (&self, value: Handle) { + *self.result.lock() = Some(value); + self.done.store(true, Ordering::Release); + } + + pub fn get_handle(&self) -> Handle { + self.handle + } + + pub fn is_complete(&self) -> bool { + self.done.load(Ordering::Acquire) + } + + pub fn take_result(&self) -> Handle { + self.result.lock().take().expect("task marked complete without a result") + } +} +pub struct Scheduler { + global: SegQueue>, + //locals: Vec>>>, + //results: , + //num_work_left:, + //live_worker_count:, scheduler calls into executer??? scheduler thread, thread is like arca user pipe opens user programarca blob + +} + +impl Scheduler { + pub const fn new () -> Self { + Self { global: SegQueue::new() } + } + + pub fn get_work (self: &Self) -> Option> { + self.global.pop() + } + + pub fn push_work(&self, handle: Handle)-> Arc { + let task = Arc::new(Task::new(handle)); + self.global.push(Arc::clone(&task)); + task + } + +} \ No newline at end of file From 2b10ea785df2e2509eae4a511a8e9b56293eab78 Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Fri, 31 Jul 2026 15:35:45 -0700 Subject: [PATCH 2/7] refactor: change blob creation for edited scripting language and improve eval_parallel handling in main.rs --- addblob_extended.fix | 20 ++++---- fix/src/main.rs | 106 ++++++++++++++----------------------------- 2 files changed, 44 insertions(+), 82 deletions(-) diff --git a/addblob_extended.fix b/addblob_extended.fix index b6e285d..6534593 100644 --- a/addblob_extended.fix +++ b/addblob_extended.fix @@ -1,13 +1,13 @@ -a = create_blob(Int(1)); -b = create_blob(Int(2)); -c = create_blob(Int(3)); -d = create_blob(Int(4)); -e = create_blob(Int(5)); -f = create_blob(Int(6)); -g = create_blob(Int(7)); -h = create_blob(Int(8)); - -add = create_blob(Path("./target/x86_64-unknown-none/addblob")); +a = 1; +b = 2; +c = 3; +d = 4; +e = 5; +f = 6; +g = 7; +h = 8; + +add = create_blob("./target/x86_64-unknown-none/addblob"); ab = create_application_thunk(create_tree(add, a, b)); cd = create_application_thunk(create_tree(add, c, d)); diff --git a/fix/src/main.rs b/fix/src/main.rs index 0ce4c2a..ea59006 100644 --- a/fix/src/main.rs +++ b/fix/src/main.rs @@ -135,7 +135,7 @@ fn eval(evaluator: &Evaluator, e: &Expr, ctx: &mut BTreeMap { let x = eval_parallel(evaluator.as_ref(), &expr, &mut context); - match x { - Value::Handle(x) => { - println!("handle: {x}"); - if let Some(blob) = x - .try_unwrap_object() - .ok() - .and_then(|x| x.try_unwrap_blob().ok()) - { - let contents = evaluator.storage().get_blob(blob).unwrap(); - println!("result is a Blob: {contents:?}"); - if contents.len() == 8 { - let bytes: [u8; 8] = (*contents).try_into().unwrap(); - let value = u64::from_le_bytes(bytes); - println!("\tas a u64: {value}"); - } - } - } - Value::Int(x) => { - println!("int: {x}"); - } - Value::String(x) => { - println!("string: {x}"); - } - Value::Path(x) => { - println!("path: {x}"); + println!("handle: {x}"); + if let Handle::Object(Object::Blob(blob)) = x { + let contents = evaluator.storage().get_blob(blob).unwrap(); + println!("result is a Blob: {contents:?}"); + if contents.len() == 8 { + let bytes: [u8; 8] = (*contents).try_into().unwrap(); + let value = u64::from_le_bytes(bytes); + println!("\tas a u64: {value}"); } } } @@ -194,56 +177,35 @@ fn eval_file_parallel(filename: &str) { } } -fn eval_parallel(evaluator: ¶llel_evaluator::Evaluator, e: &Expr, ctx: &mut BTreeMap) -> Value { +fn eval_parallel(evaluator: ¶llel_evaluator::Evaluator, e: &Expr, ctx: &mut BTreeMap) -> Handle { match e { - Expr::Number(x) => Value::Int(*x), - Expr::Identifier(x) => ctx.get(x).expect("undefined identifier").clone(), - Expr::String(x) => Value::String(x.clone()), + Expr::Identifier(x) => *ctx.get(x).expect("undefined identifier"), + Expr::Number(x) => { + let bytes = i64::to_le_bytes(*x); + evaluator.storage().add_blob(&bytes).into() + } + Expr::String(x) => { + let bytes = x.as_bytes(); + evaluator.storage().add_blob(bytes).into() + } Expr::Call { name, args } => { - let args: Vec = args.into_iter().map(|x| eval_parallel(evaluator, x, ctx)).collect(); + let arg_handles: Vec = args.iter().map(|x| eval_parallel(evaluator, x, ctx)).collect(); match name.as_str() { - "Int" => args[0].clone(), - "create_blob" => match args[0] { - Value::Handle(_) => panic!("create blob with handle?"), - Value::Int(x) => { - let bytes = i64::to_le_bytes(x); - Value::Handle(evaluator.storage().add_blob(&bytes).into()) - } - Value::String(ref x) => { - Value::Handle(evaluator.storage().add_blob(x.as_bytes()).into()) - } - Value::Path(ref x) => { - let mut file = File::open(x, true, false, false, false, false).unwrap(); - let len = file.seek(Whence::End(0)); - file.seek(Whence::Start(0)); - let mut buf = vec![0; len as usize]; - file.read_exact(&mut buf); - core::mem::forget(file); - Value::Handle(evaluator.storage().add_blob(&buf).into()) - } - }, - "create_tree" => { - let handles: Vec = args.into_iter().map(Value::unwrap_handle).collect(); - Value::Handle(evaluator.storage().add_tree(&handles).into()) + "create_blob" if let Expr::String(path) = &args.get(0).expect("no path") => { + let mut file = File::open(path, true, false, false, false, false).unwrap(); + let len = file.seek(Whence::End(0)); + file.seek(Whence::Start(0)); + let mut buf = vec![0; len as usize]; + file.read_exact(&mut buf); + core::mem::forget(file); + evaluator.storage().add_blob(&buf).into() } - "create_application_thunk" => Value::Handle( - Thunk::Application( - args[0] - .clone() - .unwrap_handle() - .unwrap_object() - .unwrap_tree(), - ) - .into(), - ), - "create_strict_encode" => Value::Handle( - Encode::Strict(args[0].clone().unwrap_handle().unwrap_thunk()).into(), - ), - "eval" => Value::Handle(evaluator.eval(args[0].clone().unwrap_handle())), - "Path" => match args[0] { - Value::String(ref x) => Value::Path(x.clone()), - _ => panic!("bad path"), - }, + "create_tree" => evaluator.storage().add_tree(&arg_handles).into(), + "create_application_thunk" => { + Thunk::Application(arg_handles[0].unwrap_object().unwrap_tree()).into() + } + "create_strict_encode" => Encode::Strict(arg_handles[0].unwrap_thunk()).into(), + "eval" => evaluator.eval(arg_handles[0]), name => todo!("call {name} {args:?}"), } } From c5757059a9b3dbdcf3ae872fdc984895e0f227e0 Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Sun, 2 Aug 2026 21:59:20 -0700 Subject: [PATCH 3/7] refactor: extend addblob_extended script with a bigger workload --- addblob_extended.fix | 32 +++++++++++++++++++++++++++++--- fix/src/main.rs | 2 +- fix/src/parallel_evaluator.rs | 9 +++++---- fix/src/scheduler.rs | 2 +- 4 files changed, 36 insertions(+), 9 deletions(-) diff --git a/addblob_extended.fix b/addblob_extended.fix index 6534593..7c709a0 100644 --- a/addblob_extended.fix +++ b/addblob_extended.fix @@ -6,6 +6,14 @@ e = 5; f = 6; g = 7; h = 8; +i = 9; +j = 10; +k = 11; +l = 12; +m = 13; +n = 14; +o = 15; +p = 16; add = create_blob("./target/x86_64-unknown-none/addblob"); @@ -13,14 +21,32 @@ ab = create_application_thunk(create_tree(add, a, b)); cd = create_application_thunk(create_tree(add, c, d)); ef = create_application_thunk(create_tree(add, e, f)); gh = create_application_thunk(create_tree(add, g, h)); +ij = create_application_thunk(create_tree(add, i, j)); +kl = create_application_thunk(create_tree(add, k, l)); +mn = create_application_thunk(create_tree(add, m, n)); +op = create_application_thunk(create_tree(add, o, p)); sum_ab = create_strict_encode(ab); sum_cd = create_strict_encode(cd); sum_ef = create_strict_encode(ef); sum_gh = create_strict_encode(gh); - -left = create_application_thunk(create_tree(add, sum_ab, sum_cd)); -right = create_application_thunk(create_tree(add, sum_ef, sum_gh)); +sum_ij = create_strict_encode(ij); +sum_kl = create_strict_encode(kl); +sum_mn = create_strict_encode(mn); +sum_op = create_strict_encode(op); + +lefti = create_application_thunk(create_tree(add, sum_ab, sum_cd)); +righti = create_application_thunk(create_tree(add, sum_ef, sum_gh)); +leftr= create_application_thunk(create_tree(add, sum_ij, sum_kl)); +rightr = create_application_thunk(create_tree(add, sum_mn, sum_op)); + +sum_lefti = create_strict_encode(lefti); +sum_righti = create_strict_encode(righti); +sum_leftr = create_strict_encode(leftr); +sum_rightr = create_strict_encode(rightr); + +left = create_application_thunk(create_tree(add, sum_lefti, sum_leftr)); +right = create_application_thunk(create_tree(add, sum_righti, sum_rightr)); sum_left = create_strict_encode(left); sum_right = create_strict_encode(right); diff --git a/fix/src/main.rs b/fix/src/main.rs index ea59006..0ef93ae 100644 --- a/fix/src/main.rs +++ b/fix/src/main.rs @@ -26,7 +26,7 @@ fn main() { let filename = argv.get(2).expect("fix eval: expected a command file"); eval_file(filename); }, - //test to run the parallel evaluator + // test to run the parallel evaluator Some("parallel_eval") => { let filename = argv.get(2).expect("fix eval: expected a command file"); eval_file_parallel(filename); diff --git a/fix/src/parallel_evaluator.rs b/fix/src/parallel_evaluator.rs index 0ffc9ff..9898e1b 100644 --- a/fix/src/parallel_evaluator.rs +++ b/fix/src/parallel_evaluator.rs @@ -24,7 +24,7 @@ use kernel::prelude::*; // use common::bitpack::BitPack; // use kernel::prelude::*; -const NUM_WORKERS: usize = 2; +const NUM_WORKERS: usize = 8; #[derive(Clone, Copy)] enum EvalType { Parallel, Serial, @@ -98,11 +98,12 @@ impl Evaluator { } } } - + /* pub fn runtime(&self) -> &R { &self.runtime } - + */ + pub fn storage(&self) -> &dyn Storage { self.runtime.storage() } @@ -196,7 +197,7 @@ impl Evaluator { self.runtime.storage().add_tree(&evaled) } // elimnate redundancy i think - pub fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle { + fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle { println!("evaluating {handle}"); match handle { Handle::Thunk(_) | Handle::Ref(_) => todo!(), diff --git a/fix/src/scheduler.rs b/fix/src/scheduler.rs index 1437d66..d38d894 100644 --- a/fix/src/scheduler.rs +++ b/fix/src/scheduler.rs @@ -16,7 +16,7 @@ pub struct Task { } impl Task { - // create new task + fn new (value: Handle) -> Self{ Self { handle: value, From b0f5c5514ee1add8d82c0321338efe8f506deea0 Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Mon, 3 Aug 2026 09:29:29 -0700 Subject: [PATCH 4/7] style: run cargo fmt --- fix/src/main.rs | 19 ++++++++----- fix/src/parallel_evaluator.rs | 50 +++++++++++++---------------------- fix/src/scheduler.rs | 31 +++++++++++----------- 3 files changed, 48 insertions(+), 52 deletions(-) diff --git a/fix/src/main.rs b/fix/src/main.rs index 0ef93ae..1da9735 100644 --- a/fix/src/main.rs +++ b/fix/src/main.rs @@ -1,8 +1,8 @@ #![no_main] #![no_std] -mod scheduler; mod parallel_evaluator; +mod scheduler; use kernel::host::fs::{self, File, Whence}; use kernel::host::os; @@ -25,7 +25,7 @@ fn main() { Some("eval") => { let filename = argv.get(2).expect("fix eval: expected a command file"); eval_file(filename); - }, + } // test to run the parallel evaluator Some("parallel_eval") => { let filename = argv.get(2).expect("fix eval: expected a command file"); @@ -133,8 +133,8 @@ fn eval(evaluator: &Evaluator, e: &Expr, ctx: &mut BTreeMap, e: &Expr, ctx: &mut BTreeMap) -> Handle { +fn eval_parallel( + evaluator: ¶llel_evaluator::Evaluator, + e: &Expr, + ctx: &mut BTreeMap, +) -> Handle { match e { Expr::Identifier(x) => *ctx.get(x).expect("undefined identifier"), Expr::Number(x) => { @@ -189,7 +193,10 @@ fn eval_parallel(evaluator: ¶llel_evaluator::Evaluator, e: &Expr, evaluator.storage().add_blob(bytes).into() } Expr::Call { name, args } => { - let arg_handles: Vec = args.iter().map(|x| eval_parallel(evaluator, x, ctx)).collect(); + let arg_handles: Vec = args + .iter() + .map(|x| eval_parallel(evaluator, x, ctx)) + .collect(); match name.as_str() { "create_blob" if let Expr::String(path) = &args.get(0).expect("no path") => { let mut file = File::open(path, true, false, false, false, false).unwrap(); diff --git a/fix/src/parallel_evaluator.rs b/fix/src/parallel_evaluator.rs index 9898e1b..afd8d73 100644 --- a/fix/src/parallel_evaluator.rs +++ b/fix/src/parallel_evaluator.rs @@ -6,9 +6,8 @@ use kernel::{coreid, kthread}; //use kernel::tsc; use crate::scheduler::{Scheduler, Task}; - use crate::handle::*; -use crate::runtime::Runtime; +use crate::runtime::Runtime; use crate::storage::Storage; use kernel::prelude::*; @@ -27,7 +26,8 @@ use kernel::prelude::*; const NUM_WORKERS: usize = 8; #[derive(Clone, Copy)] enum EvalType { - Parallel, Serial, + Parallel, + Serial, } pub struct Evaluator { runtime: R, @@ -35,8 +35,7 @@ pub struct Evaluator { } impl Evaluator { - - pub fn new(runtime: R) -> Arc{ + pub fn new(runtime: R) -> Arc { let evaluator = Arc::new(Self { runtime, scheduler: Scheduler::new(), @@ -49,11 +48,10 @@ impl Evaluator { pub fn start_workers(self: &Arc, num_workers: usize) { for i in 0..num_workers { let evaluator = Arc::clone(self); - kthread::spawn(move || { + kthread::spawn(move || { println!("worker {} started on core {}", i, coreid()); evaluator.worker_loop(i); - } - ) + }) } } @@ -61,21 +59,15 @@ impl Evaluator { loop { // eventually make this a queue to handle local queue of work if let Some(work) = self.scheduler.get_work() { - println!( - "worker {} evaluating task on core {}", - worker_id, - coreid() - ); + println!("worker {} evaluating task on core {}", worker_id, coreid()); let result = self.eval_test(work.get_handle(), EvalType::Parallel); work.task_complete(result) - } - else { + } else { //change this to condition variable, so it does no busy waiting? kthread::yield_now() } } - } fn wait_while_helping(&self, target: &Arc) -> Handle { @@ -84,12 +76,11 @@ impl Evaluator { if target.is_complete() { return target.take_result(); } - + if let Some(work) = self.scheduler.get_work() { println!("calling thread helping on core {}", coreid()); - let result = - self.eval_test(work.get_handle(), EvalType::Parallel); + let result = self.eval_test(work.get_handle(), EvalType::Parallel); work.task_complete(result); } else { @@ -97,13 +88,13 @@ impl Evaluator { kthread::yield_now(); } } -} + } /* pub fn runtime(&self) -> &R { &self.runtime } */ - + pub fn storage(&self) -> &dyn Storage { self.runtime.storage() } @@ -137,7 +128,7 @@ impl Evaluator { Thunk::Identification(_) => todo!(), Thunk::Selection(_) => todo!(), Thunk::Application(tree) => { - let evaled = self.eval_tree(tree,eval_mode); + let evaled = self.eval_tree(tree, eval_mode); self.apply(evaled) } } @@ -159,14 +150,13 @@ impl Evaluator { } } - fn eval_tree(&self, handle: Tree, eval_mode: EvalType) -> Tree { match eval_mode { EvalType::Serial => self.eval_tree_seq(handle), - EvalType::Parallel => self.eval_tree_parallel(handle) - } + EvalType::Parallel => self.eval_tree_parallel(handle), + } } - + fn eval_tree_seq(&self, handle: Tree) -> Tree { let tree = self.runtime.storage().get_tree(handle).unwrap(); let evaled: Vec = tree @@ -179,13 +169,12 @@ impl Evaluator { } fn eval_tree_parallel(&self, handle: Tree) -> Tree { - let tree = self.runtime.storage().get_tree(handle).unwrap(); if tree.len() <= 2 { - return self.eval_tree_seq(handle) + return self.eval_tree_seq(handle); } let mut evaled = Vec::with_capacity(tree.len()); - // evaluate the first + // evaluate the first evaled.push(self.eval_test(tree[0], EvalType::Serial)); let mut tasks = Vec::with_capacity(tree.len() - 1); for child in tree[1..].iter().copied() { @@ -195,7 +184,7 @@ impl Evaluator { evaled.push(self.wait_while_helping(&task)); } self.runtime.storage().add_tree(&evaled) -} + } // elimnate redundancy i think fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle { println!("evaluating {handle}"); @@ -258,5 +247,4 @@ impl Evaluator { result_tree } */ - } diff --git a/fix/src/scheduler.rs b/fix/src/scheduler.rs index d38d894..e52fab0 100644 --- a/fix/src/scheduler.rs +++ b/fix/src/scheduler.rs @@ -13,23 +13,21 @@ pub struct Task { handle: Handle, result: KMutex>, done: AtomicBool, - } impl Task { - - fn new (value: Handle) -> Self{ + fn new(value: Handle) -> Self { Self { handle: value, result: KMutex::new(None), done: AtomicBool::new(false), } } - - pub fn task_complete (&self, value: Handle) { + + pub fn task_complete(&self, value: Handle) { *self.result.lock() = Some(value); self.done.store(true, Ordering::Release); } - + pub fn get_handle(&self) -> Handle { self.handle } @@ -39,7 +37,10 @@ impl Task { } pub fn take_result(&self) -> Handle { - self.result.lock().take().expect("task marked complete without a result") + self.result + .lock() + .take() + .expect("task marked complete without a result") } } pub struct Scheduler { @@ -48,22 +49,22 @@ pub struct Scheduler { //results: , //num_work_left:, //live_worker_count:, scheduler calls into executer??? scheduler thread, thread is like arca user pipe opens user programarca blob - } impl Scheduler { - pub const fn new () -> Self { - Self { global: SegQueue::new() } + pub const fn new() -> Self { + Self { + global: SegQueue::new(), + } } - pub fn get_work (self: &Self) -> Option> { + pub fn get_work(self: &Self) -> Option> { self.global.pop() } - pub fn push_work(&self, handle: Handle)-> Arc { + pub fn push_work(&self, handle: Handle) -> Arc { let task = Arc::new(Task::new(handle)); self.global.push(Arc::clone(&task)); - task + task } - -} \ No newline at end of file +} From d34a593072ed8345d4e7d0a173feb8dded5d4dc5 Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Thu, 6 Aug 2026 09:39:40 -0700 Subject: [PATCH 5/7] fix: remove unnecessary modules and add slowaddblob module --- addblob_extended.fix | 2 +- fix/src/parallel_evaluator.rs | 80 +++---------------------- fix/src/scheduler.rs | 2 - fix/wasm/slowaddblob.wat | 109 ++++++++++++++++++++++++++++++++++ 4 files changed, 117 insertions(+), 76 deletions(-) create mode 100644 fix/wasm/slowaddblob.wat diff --git a/addblob_extended.fix b/addblob_extended.fix index 7c709a0..e051e77 100644 --- a/addblob_extended.fix +++ b/addblob_extended.fix @@ -15,7 +15,7 @@ n = 14; o = 15; p = 16; -add = create_blob("./target/x86_64-unknown-none/addblob"); +add = create_blob("./target/x86_64-unknown-none/slowaddblob"); ab = create_application_thunk(create_tree(add, a, b)); cd = create_application_thunk(create_tree(add, c, d)); diff --git a/fix/src/parallel_evaluator.rs b/fix/src/parallel_evaluator.rs index afd8d73..d626750 100644 --- a/fix/src/parallel_evaluator.rs +++ b/fix/src/parallel_evaluator.rs @@ -1,29 +1,14 @@ extern crate alloc; +use crate::scheduler::{Scheduler, Task}; use alloc::sync::Arc; -//use core::sync::atomic::{AtomicBool, Ordering}; use kernel::{coreid, kthread}; -//use kernel::kthread::{KMutex, yield_now}; -//use kernel::tsc; -use crate::scheduler::{Scheduler, Task}; use crate::handle::*; use crate::runtime::Runtime; use crate::storage::Storage; use kernel::prelude::*; -// use fixhandle::rawhandle::{Encode, Handle, Object, Ref, Thunk, TreeName}; - -// use fixruntime::{ -// common::CouponTrades, -// fixruntime::{FixRuntime, FixTreeData}, -// runtime::{DeterministicEquivRuntime, Executor}, -// storage::FixData, -// }; - -// use common::bitpack::BitPack; -// use kernel::prelude::*; - -const NUM_WORKERS: usize = 8; +const NUM_WORKERS: usize = 19; #[derive(Clone, Copy)] enum EvalType { Parallel, @@ -49,7 +34,7 @@ impl Evaluator { for i in 0..num_workers { let evaluator = Arc::clone(self); kthread::spawn(move || { - println!("worker {} started on core {}", i, coreid()); + //println!("worker {} started on core {}", i, coreid()); evaluator.worker_loop(i); }) } @@ -59,7 +44,7 @@ impl Evaluator { loop { // eventually make this a queue to handle local queue of work if let Some(work) = self.scheduler.get_work() { - println!("worker {} evaluating task on core {}", worker_id, coreid()); + //println!("worker {} evaluating task on core {}", worker_id, coreid()); let result = self.eval_test(work.get_handle(), EvalType::Parallel); work.task_complete(result) @@ -78,7 +63,7 @@ impl Evaluator { } if let Some(work) = self.scheduler.get_work() { - println!("calling thread helping on core {}", coreid()); + //println!("calling thread helping on core {}", coreid()); let result = self.eval_test(work.get_handle(), EvalType::Parallel); @@ -89,11 +74,6 @@ impl Evaluator { } } } - /* - pub fn runtime(&self) -> &R { - &self.runtime - } - */ pub fn storage(&self) -> &dyn Storage { self.runtime.storage() @@ -170,7 +150,7 @@ impl Evaluator { fn eval_tree_parallel(&self, handle: Tree) -> Tree { let tree = self.runtime.storage().get_tree(handle).unwrap(); - if tree.len() <= 2 { + if tree.len() <= 3 { return self.eval_tree_seq(handle); } let mut evaled = Vec::with_capacity(tree.len()); @@ -187,7 +167,7 @@ impl Evaluator { } // elimnate redundancy i think fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle { - println!("evaluating {handle}"); + //println!("evaluating {handle}"); match handle { Handle::Thunk(_) | Handle::Ref(_) => todo!(), Handle::Object(obj) => match obj { @@ -199,52 +179,6 @@ impl Evaluator { } pub fn eval(&self, handle: Handle) -> Handle { - //let start = tsc::read_cycles(); self.eval_test(handle, EvalType::Parallel) - //println!("total_cycles = {}", cycles_since(start)); - //result - } - - /* - fn eval_tree_seq(&self, handle: Tree) -> Tree { - let start = tsc::read_cycles(); - - println!( - "[timing] eval_tree_seq start on core {}: {:?}", - kernel::coreid(), - handle - ); - - let tree = self.runtime.storage().get_tree(handle).unwrap(); - - let evaled: Vec = tree - .as_ref() - .iter() - .copied() - .map(|x| { - let child_start = tsc::read_cycles(); - let result = self.eval(x); - - println!( - "[timing] seq child {:?} -> {:?}, cycles={}", - x, - result, - cycles_since(child_start) - ); - - result - }) - .collect(); - - let result_tree = self.runtime.storage().add_tree(&evaled); - - println!( - "[timing] eval_tree_seq done: {:?}, total_cycles={}", - handle, - cycles_since(start) - ); - - result_tree } - */ } diff --git a/fix/src/scheduler.rs b/fix/src/scheduler.rs index e52fab0..320dddc 100644 --- a/fix/src/scheduler.rs +++ b/fix/src/scheduler.rs @@ -1,9 +1,7 @@ -//use alloc::collections::VecDeque; use alloc::sync::Arc; use fix::Handle; extern crate alloc; use core::sync::atomic::{AtomicBool, Ordering}; -//use kernel::kthread; use kernel::kthread::KMutex; use crossbeam_queue::SegQueue; diff --git a/fix/wasm/slowaddblob.wat b/fix/wasm/slowaddblob.wat new file mode 100644 index 0000000..d992ad6 --- /dev/null +++ b/fix/wasm/slowaddblob.wat @@ -0,0 +1,109 @@ +(module + (import "fixpoint" "create_blob_i64" + (func $create_blob_i64 (param i64) (result externref))) + (import "fixpoint" "attach_blob" + (func $attach_blob (param i32) (param externref))) + (import "fixpoint" "attach_tree" + (func $attach_tree (param i32) (param externref))) + + (memory $mem_0 1) + (memory $mem_1 0) + (memory $mem_2 0) + (table $tab_0 0 externref) + + (func (export "_fixpoint_apply") + (param $encode externref) + (result externref) + + (local $counter i64) + (local $spin i64) + (local $left i64) + (local $right i64) + + ;; Attach the combination tree. + (call $attach_tree + (i32.const 0) + (local.get $encode)) + + ;; Grow rw-memory by zero pages, preserving the original behavior. + (memory.grow + (memory $mem_0) + (i32.const 0)) + drop + + ;; Attach the two input blobs. + (call $attach_blob + (i32.const 1) + (table.get $tab_0 (i32.const 1))) + (call $attach_blob + (i32.const 2) + (table.get $tab_0 (i32.const 2))) + + ;; Load both operands once. + (local.set $left + (i64.load + (memory $mem_1) + (i32.const 0))) + + (local.set $right + (i64.load + (memory $mem_2) + (i32.const 0))) + + ;; Artificial CPU work before the addition. + ;; Change this constant to tune how slow each addition is. + (local.set $counter (i64.const 100000000)) + (local.set $spin (local.get $left)) + + (block $spin_done + (loop $spin_loop + (br_if $spin_done + (i64.eqz (local.get $counter))) + + (local.set $spin + (i64.add + (i64.xor + (local.get $spin) + (local.get $counter)) + (i64.const 6364136223846793005))) + + (local.set $counter + (i64.sub + (local.get $counter) + (i64.const 1))) + + (br $spin_loop))) + + ;; Store the spin result so the work is observable. + (i64.store + (memory $mem_0) + (i32.const 8) + (local.get $spin)) + + ;; Both branches return left + right. + ;; The condition depends on the spin result. + (i64.store + (memory $mem_0) + (i32.const 0) + (if (result i64) + (i64.eqz + (i64.and + (local.get $spin) + (i64.const 1))) + (then + (i64.add + (local.get $left) + (local.get $right))) + (else + (i64.sub + (local.get $left) + (i64.sub + (i64.const 0) + (local.get $right)))))) + + ;; Return the sum as a Fix blob. + (call $create_blob_i64 + (i64.load + (memory $mem_0) + (i32.const 0)))) +) \ No newline at end of file From cdf4249cce58f538477c72cc91f930bad32705a1 Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Thu, 6 Aug 2026 11:07:02 -0700 Subject: [PATCH 6/7] fix: update parallel evaluator after rebase --- fix/src/main.rs | 6 +++++- fix/src/parallel_evaluator.rs | 15 ++++++++------- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/fix/src/main.rs b/fix/src/main.rs index 1da9735..88222c1 100644 --- a/fix/src/main.rs +++ b/fix/src/main.rs @@ -216,6 +216,10 @@ fn eval_parallel( name => todo!("call {name} {args:?}"), } } - Expr::Group(x) => eval_parallel(evaluator, x, ctx), + Expr::IdentificationThunk(x) => { + Thunk::Identification(eval_parallel(&evaluator, x, ctx).unwrap_ref()).into() + } + Expr::Ref(reference) => evaluator.lower(eval_parallel(&evaluator, reference, ctx)), + Expr::Group(x) => eval_parallel(&evaluator, x, ctx), } } diff --git a/fix/src/parallel_evaluator.rs b/fix/src/parallel_evaluator.rs index d626750..e1886b6 100644 --- a/fix/src/parallel_evaluator.rs +++ b/fix/src/parallel_evaluator.rs @@ -34,13 +34,13 @@ impl Evaluator { for i in 0..num_workers { let evaluator = Arc::clone(self); kthread::spawn(move || { - //println!("worker {} started on core {}", i, coreid()); + println!("worker {} started on core {}", i, coreid()); evaluator.worker_loop(i); }) } } - pub fn worker_loop(self: &Arc, worker_id: usize) { + pub fn worker_loop(self: &Arc, _worker_id: usize) { loop { // eventually make this a queue to handle local queue of work if let Some(work) = self.scheduler.get_work() { @@ -83,7 +83,7 @@ impl Evaluator { self.runtime.execute(combination) } - fn lift(&self, handle: Handle) -> Handle { + pub fn lift(&self, handle: Handle) -> Handle { match handle { Handle::Ref(r) => match r { Ref::Tree(t) => Object::Tree(t).into(), @@ -93,7 +93,7 @@ impl Evaluator { } } - fn lower(&self, handle: Handle) -> Handle { + pub fn lower(&self, handle: Handle) -> Handle { match handle { Handle::Object(r) => match r { Object::Tree(t) => Ref::Tree(t).into(), @@ -105,7 +105,7 @@ impl Evaluator { fn think(&self, thunk: Thunk, eval_mode: EvalType) -> Handle { match thunk { - Thunk::Identification(_) => todo!(), + Thunk::Identification(reference) => self.lift(Handle::Ref(reference)), Thunk::Selection(_) => todo!(), Thunk::Application(tree) => { let evaled = self.eval_tree(tree, eval_mode); @@ -169,9 +169,10 @@ impl Evaluator { fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle { //println!("evaluating {handle}"); match handle { - Handle::Thunk(_) | Handle::Ref(_) => todo!(), + Handle::Ref(reference) => self.eval(self.lift(Handle::Ref(reference))), + Handle::Thunk(_) => todo!(), Handle::Object(obj) => match obj { - Object::Blob(x) => x.into(), + Object::Blob(blob) => blob.into(), Object::Tree(tree) => self.eval_tree(tree, eval_mode).into(), }, Handle::Encode(e) => self.eval_test(self.encode(e, eval_mode), eval_mode), From ea399b51571cfe68137b544ab83279367225528d Mon Sep 17 00:00:00 2001 From: Jennifer Mori Date: Fri, 7 Aug 2026 11:12:26 -0700 Subject: [PATCH 7/7] fix: switch addblob_extended back to add blob --- addblob_extended.fix | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/addblob_extended.fix b/addblob_extended.fix index e051e77..7c709a0 100644 --- a/addblob_extended.fix +++ b/addblob_extended.fix @@ -15,7 +15,7 @@ n = 14; o = 15; p = 16; -add = create_blob("./target/x86_64-unknown-none/slowaddblob"); +add = create_blob("./target/x86_64-unknown-none/addblob"); ab = create_application_thunk(create_tree(add, a, b)); cd = create_application_thunk(create_tree(add, c, d));