Add ConfigBuilder, rename and refactor setup structs

benches/batch_mode.rs

@@ -10,7 +10,7 @@ use noir_compute::RuntimeConfig;
 use noir_compute::StreamContext;
 
 fn batch_mode(batch_mode: BatchMode, dataset: &'static [u32]) {
-    let config = RuntimeConfig::local(4);
+    let config = RuntimeConfig::local(4).unwrap();
     let env = StreamContext::new(config);
 
     let source = IteratorSource::new(dataset.iter().cloned());

benches/collatz.rs

@@ -16,7 +16,7 @@ fn bench_main(c: &mut Criterion) {
         g.throughput(Throughput::Elements(size));
         g.bench_with_input(BenchmarkId::new("collatz", size), &size, |b, n| {
             b.iter(|| {
-                let env = StreamContext::default();
+                let env = StreamContext::new_local();
                 env.stream_par_iter(0..*n)
                     .batch_mode(BatchMode::fixed(1024))
                     .map(move |n| {

benches/common.rs

@@ -1,7 +1,7 @@
 #![allow(unused)]
 
 use criterion::{black_box, Bencher};
-use noir_compute::config::{HostConfig, RemoteConfig, RuntimeConfig};
+use noir_compute::config::{ConfigBuilder, HostConfig, RemoteConfig, RuntimeConfig};
 use std::marker::PhantomData;
 use std::sync::atomic::{AtomicU16, Ordering};
 use std::sync::Arc;
@@ -36,12 +36,11 @@ pub fn remote_loopback_deploy(
     let mut join_handles = vec![];
     let body = Arc::new(body);
     for host_id in 0..num_hosts {
-        let config = RuntimeConfig::Remote(RemoteConfig {
-            host_id: Some(host_id),
-            hosts: hosts.clone(),
-            tracing_dir: None,
-            cleanup_executable: false,
-        });
+        let config = ConfigBuilder::new_remote()
+            .add_hosts(&hosts)
+            .host_id(host_id)
+            .build()
+            .unwrap();
 
         let body = body.clone();
         join_handles.push(
@@ -100,6 +99,6 @@ where
 }
 
 pub fn noir_bench_default(b: &mut Bencher, logic: impl Fn(&StreamContext)) {
-    let builder = NoirBenchBuilder::new(StreamContext::default, logic);
+    let builder = NoirBenchBuilder::new(StreamContext::new_local, logic);
     b.iter_custom(|n| builder.bench(n));
 }

benches/connected.rs

@@ -2,7 +2,6 @@ use criterion::BenchmarkId;
 use criterion::{criterion_group, criterion_main, Criterion, Throughput};
 use fxhash::FxHashMap;
 use noir_compute::operator::Operator;
-use noir_compute::RuntimeConfig;
 use noir_compute::Stream;
 use noir_compute::StreamContext;
 use rand::prelude::*;
@@ -109,7 +108,7 @@ fn bench_main(c: &mut Criterion) {
         g.throughput(Throughput::Elements(size));
         g.bench_with_input(BenchmarkId::new("connected", size), &size, |b, size| {
             b.iter(|| {
-                let env = StreamContext::new(RuntimeConfig::local(4));
+                let env = StreamContext::new_local();
                 let edges = *size;
                 let nodes = ((edges as f32).sqrt() * 25.) as u64 + 1;
 

benches/fold_vs_reduce.rs

@@ -8,7 +8,7 @@ use noir_compute::RuntimeConfig;
 use noir_compute::StreamContext;
 
 fn fold(dataset: &'static [u32]) {
-    let config = RuntimeConfig::local(4);
+    let config = RuntimeConfig::default();
     let env = StreamContext::new(config);
 
     let source = IteratorSource::new(dataset.iter().cloned());
@@ -21,7 +21,7 @@ fn fold(dataset: &'static [u32]) {
 }
 
 fn reduce(dataset: &'static [u32]) {
-    let config = RuntimeConfig::local(4);
+    let config = RuntimeConfig::default();
     let env = StreamContext::new(config);
 
     let source = IteratorSource::new(dataset.iter().cloned());
@@ -34,7 +34,7 @@ fn reduce(dataset: &'static [u32]) {
 }
 
 fn fold_assoc(dataset: &'static [u32]) {
-    let config = RuntimeConfig::local(4);
+    let config = RuntimeConfig::default();
     let env = StreamContext::new(config);
 
     let source = IteratorSource::new(dataset.iter().cloned());
@@ -51,7 +51,7 @@ fn fold_assoc(dataset: &'static [u32]) {
 }
 
 fn reduce_assoc(dataset: &'static [u32]) {
-    let config = RuntimeConfig::local(4);
+    let config = RuntimeConfig::default();
     let env = StreamContext::new(config);
 
     let source = IteratorSource::new(dataset.iter().cloned());

benches/nexmark.rs

@@ -365,7 +365,7 @@ fn bench_main(c: &mut Criterion) {
         ($q:expr, $n:expr) => {{
             g.bench_with_input(BenchmarkId::new($q, $n), &$n, |b, size| {
                 b.iter(|| {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     run_query(&env, $q, *size);
                     env.execute_blocking();
                 })

benches/shuffle.rs

@@ -8,7 +8,7 @@ use noir_compute::BatchMode;
 use noir_compute::StreamContext;
 
 fn shuffle(dataset: &'static [u32]) {
-    let env = StreamContext::default();
+    let env = StreamContext::new_local();
 
     let source = IteratorSource::new(dataset.iter().cloned());
     let stream = env

benches/wordcount.rs

@@ -58,7 +58,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_fold(&env, path);
                     env.execute_blocking();
                 })
@@ -70,7 +70,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_fold_assoc(&env, path);
                     env.execute_blocking();
                 })
@@ -82,7 +82,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_count_assoc(&env, path);
                     env.execute_blocking();
                 })
@@ -94,7 +94,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_reduce(&env, path);
                     env.execute_blocking();
                 })
@@ -106,7 +106,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_reduce_assoc(&env, path);
                     env.execute_blocking();
                 })
@@ -118,7 +118,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_fast(&env, path);
                     env.execute_blocking();
                 })
@@ -130,7 +130,7 @@ fn wordcount_bench(c: &mut Criterion) {
             file.path(),
             |b, path| {
                 b.iter(move || {
-                    let env = StreamContext::default();
+                    let env = StreamContext::new_local();
                     wc_fast_kstring(&env, path);
                     env.execute_blocking();
                 })

config.toml

@@ -0,0 +1,10 @@
+
+[[host]]
+address = "host1"
+base_port = 9500
+num_cores = 16
+
+[[host]]
+address = "host2"
+base_port = 9500
+num_cores = 24