Parallel evaluator

addblob_extended.fix

@@ -0,0 +1,57 @@
+a = 1;
+b = 2;
+c = 3;
+d = 4;
+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");
+
+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);
+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);
+
+final = create_application_thunk(create_tree(add, sum_left, sum_right));
+
+eval(create_strict_encode(final));
+

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" }

fix/src/main.rs

@@ -1,5 +1,9 @@
 #![no_main]
 #![no_std]
+
+mod parallel_evaluator;
+mod scheduler;
+
 use kernel::host::fs::{self, File, Whence};
 use kernel::host::os;
 use kernel::prelude::*;
@@ -22,6 +26,11 @@ fn main() {
             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 <file>)"),
         None => panic!("fix: expected a command (init | eval <file>)"),
     }
@@ -123,3 +132,94 @@ fn eval(evaluator: &Evaluator<FixOnArca>, e: &Expr, ctx: &mut BTreeMap<String, H
         Expr::Group(x) => 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 and how its passed in
+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);
+                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}");
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn eval_parallel(
+    evaluator: &parallel_evaluator::Evaluator<FixOnArca>,
+    e: &Expr,
+    ctx: &mut BTreeMap<String, Handle>,
+) -> Handle {
+    match e {
+        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 arg_handles: Vec<Handle> = 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();
+                    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_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:?}"),
+            }
+        }
+        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),
+    }
+}

fix/src/parallel_evaluator.rs

@@ -0,0 +1,185 @@
+extern crate alloc;
+use crate::scheduler::{Scheduler, Task};
+use alloc::sync::Arc;
+use kernel::{coreid, kthread};
+
+use crate::handle::*;
+use crate::runtime::Runtime;
+use crate::storage::Storage;
+use kernel::prelude::*;
+
+const NUM_WORKERS: usize = 19;
+#[derive(Clone, Copy)]
+enum EvalType {
+    Parallel,
+    Serial,
+}
+pub struct Evaluator<R: Runtime> {
+    runtime: R,
+    scheduler: Scheduler,
+}
+
+impl<R: Runtime> Evaluator<R> {
+    pub fn new(runtime: R) -> Arc<Self> {
+        let evaluator = Arc::new(Self {
+            runtime,
+            scheduler: Scheduler::new(),
+        });
+
+        evaluator.start_workers(NUM_WORKERS);
+        evaluator
+    }
+
+    pub fn start_workers(self: &Arc<Self>, 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<Self>, _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<Task>) -> 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 storage(&self) -> &dyn Storage {
+        self.runtime.storage()
+    }
+
+    fn apply(&self, combination: Tree) -> Handle {
+        self.runtime.execute(combination)
+    }
+
+    pub 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,
+        }
+    }
+
+    pub 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(reference) => self.lift(Handle::Ref(reference)),
+            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<Handle> = 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() <= 3 {
+            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
+    fn eval_test(&self, handle: Handle, eval_mode: EvalType) -> Handle {
+        //println!("evaluating {handle}");
+        match handle {
+            Handle::Ref(reference) => self.eval(self.lift(Handle::Ref(reference))),
+            Handle::Thunk(_) => todo!(),
+            Handle::Object(obj) => match obj {
+                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),
+        }
+    }
+
+    pub fn eval(&self, handle: Handle) -> Handle {
+        self.eval_test(handle, EvalType::Parallel)
+    }
+}