Parallel evaluator - #48
Conversation
|
Do you have approximate performance numbers? e.g., time taken on 1 core vs. 16 cores |
I've been timing the work on larger workloads and its 6.2× speedup with eight task executors and 9.4x speedup with 16 task executors. The workload contains 128 expensive slowaddblob operations arranged into eight independent subtrees, along with eight expensive eight-input additions and a final fast eight-input reduction. My results were: Serial: 10.53 seconds I removed the println! statements from both the serial and parallel evaluators so that they don't affect timing and I also changed the parallel evaluator so that application trees with only two arguments are evaluated serially, avoiding the overhead of creating parallel tasks for very small trees. Are there any optimizations you might suggest to make it even faster? |
|
For next steps, I’m considering:
I’d appreciate any high-level feedback on which of these would be the most useful next step to prioritize! |
Akshay-Srivatsan
left a comment
Akshay-Srivatsan
left a comment
There was a problem hiding this comment.
This is a great start! There's a few cleanup things to simplify the PR, and then I think this'll be ready to merge.
For next steps, I think any of your three ideas is viable/we will eventually need all three. I'd probably lean towards either doing the condition variable (which should be quick hopefully?) or designing the API (which is more involved/conceptual), depending on which makes more sense for your poster/what you want to do.
| 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::*; |
There was a problem hiding this comment.
Can you remove the unused imports here?
| 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? |
There was a problem hiding this comment.
You will most likely have to implement a condition variable using atomics, which I think should be a separate follow-up PR.
| /* | ||
| pub fn runtime(&self) -> &R { | ||
| &self.runtime | ||
| } | ||
| */ |
| /* | ||
| 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<Handle> = 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 | ||
| } | ||
| */ |
| // 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: ¶llel_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::Group(x) => eval_parallel(evaluator, x, ctx), | ||
| } | ||
| } |
There was a problem hiding this comment.
Would it make sense to rebase this on top of Haibib's changes, which I think simplify this code?
@Haibib thoughts?
There was a problem hiding this comment.
I'll wait to rebase until Habib's new PR is approved and I'll push again!
There was a problem hiding this comment.
I don't think we necessarily need a separate serial and parallel evaluator? If we have them we should probably try to deduplicate the code between them, but I think we'll eventually only want the parallel one so we might we well just replace the serial one.
There was a problem hiding this comment.
I think this is my fault -- I had thought that (especially when we move the evaluators to user-space) we probably want to keep a simple single-threaded evaluator around so we can run it for simplicity and debugging and have for teaching/explanatory purposes...
There was a problem hiding this comment.
For now, I’ll keep the serial and parallel evaluators separate, but let me know if you’d prefer that I merge them instead.
… in main.rs for processing
…ove eval_parallel handling in main.rs
52c0018 to
cdf4249
Compare
Akshay-Srivatsan
left a comment
Akshay-Srivatsan
left a comment
There was a problem hiding this comment.
I think once you rebase/test that things still work, this is good to merge. (Or you could try to race Haibib to merge first and let him rebase, I guess...)
| o = 15; | ||
| p = 16; | ||
|
|
||
| add = create_blob("./target/x86_64-unknown-none/slowaddblob"); |
There was a problem hiding this comment.
Just a note, I had to cargo clean and rebuild before it found this file.
3 participants
This PR adds initial support for parallel Fix evaluation using persistent worker threads and a shared scheduler.
Changes