Executor work

bastion-executor/Cargo.toml

@@ -1,7 +1,33 @@
 [package]
 name = "bastion-executor"
 version = "0.2.1-alpha.0"
-description = "Fault-tolerant Runtime for Rust applications"
+description = "Cache affine NUMA-aware executor for Rust"
 authors = ["Mahmut Bulut <vertexclique@gmail.com>"]
 keywords = ["fault-tolerant", "runtime", "actor", "system"]
-categories = []
+categories = ["concurrency", "asynchronous"]
+homepage = "https://github.com/bastion-rs/bastion"
+repository = "https://github.com/bastion-rs/bastion"
+documentation = "https://docs.rs/bastion"
+readme = "README.md"
+license = "Apache-2.0/MIT"
+edition = "2018"
+
+[features]
+unstable = ["numanji", "allocator-suite"]
+
+[dependencies]
+crossbeam-utils = "0.6"
+crossbeam-epoch = "0.7"
+lazy_static = "1.4"
+libc = "0.2"
+num_cpus = "1.10"
+rustc-hash = "1.0.1"
+pin-utils = "0.1.0-alpha.4"
+lightproc = { "path" = "../lightproc" }
+
+# Allocator
+numanji = { version = "^0.1", optional = true, default-features = false}
+allocator-suite = { version = "^0.1", optional = true, default-features = false}
+
+[target.'cfg(not(any(target_os = "android", target_os = "linux")))'.dependencies]
+jemallocator = "^0.3"

bastion-executor/README.md

@@ -0,0 +1,3 @@
+# Bastion Executor
+
+SMP enabled NUMA-aware executor for Rust

bastion-executor/benches/perf.rs

@@ -0,0 +1,20 @@
+#![feature(test)]
+
+extern crate test;
+
+use bastion_executor::prelude::*;
+use test::{black_box, Bencher};
+use lightproc::proc_stack::ProcStack;
+
+#[bench]
+fn increment(b: &mut Bencher) {
+    let mut sum = 0;
+
+    b.iter( || {
+        run(async {
+            (0..10_000_000).for_each(|_| { sum += 1; });
+        }, ProcStack::default());
+    });
+
+    black_box(sum);
+}

bastion-executor/examples/blocking_run.rs

@@ -0,0 +1,12 @@
+use bastion_executor::prelude::*;
+use lightproc::proc_stack::ProcStack;
+
+fn main() {
+    run(
+        async {
+            println!("Example execution");
+            panic!("fault");
+        },
+        ProcStack::default().with_after_panic(|| println!("after panic")),
+    );
+}

bastion-executor/examples/spawn_async.rs

@@ -0,0 +1,26 @@
+use bastion_executor::prelude::*;
+use lightproc::proc_stack::ProcStack;
+
+fn main() {
+    let pid = 1;
+    let stack = ProcStack::default()
+        .with_pid(pid)
+        .with_after_panic(move || println!("after panic {}", pid.clone()));
+
+    let handle = spawn(
+        async {
+            panic!("test");
+        },
+        stack,
+    );
+
+    let pid = 2;
+    let stack = ProcStack::default().with_pid(pid);
+
+    run(
+        async {
+            handle.await;
+        },
+        stack.clone(),
+    );
+}

bastion-executor/src/allocator.rs

@@ -0,0 +1,7 @@
+unstable_api! {
+    // Allocation selector import
+    use numanji::*;
+
+    // Drive selection of allocator here
+    autoselect!();
+}

bastion-executor/src/blocking_pool.rs

@@ -0,0 +1 @@
+// FIXME: Blocking pool implementation goes here.

bastion-executor/src/distributor.rs

@@ -0,0 +1,48 @@
+use super::placement;
+use super::placement::CoreId;
+use super::run_queue::{Stealer, Worker};
+
+use lightproc::prelude::*;
+
+use crate::worker;
+use std::thread;
+
+pub(crate) struct Distributor {
+    pub round: usize,
+    pub last_dead: usize,
+    pub cores: Vec<CoreId>,
+}
+
+impl Distributor {
+    pub fn new() -> Self {
+        Distributor {
+            round: 0_usize,
+            last_dead: usize::max_value(),
+            cores: placement::get_core_ids().expect("Core mapping couldn't be fetched"),
+        }
+    }
+
+    pub fn assign(mut self) -> Vec<Stealer<LightProc>> {
+        let mut stealers = Vec::<Stealer<LightProc>>::new();
+
+        for core in self.cores {
+            self.round = core.id;
+
+            let wrk = Worker::new_fifo();
+            stealers.push(wrk.stealer());
+
+            thread::Builder::new()
+                .name("bastion-async-thread".to_string())
+                .spawn(move || {
+                    // affinity assignment
+                    placement::set_for_current(core);
+
+                    // actual execution
+                    worker::main_loop(core.id.clone(), wrk);
+                })
+                .expect("cannot start the thread for running proc");
+        }
+
+        stealers
+    }
+}

bastion-executor/src/lib.rs

@@ -1,2 +1,30 @@
-mod pool;
-mod blocking_pool;
+//!
+//!
+//!
+//! NUMA-aware SMP based Fault-tolerant Executor
+//!
+//!
+
+// Force missing implementations
+#![warn(missing_docs)]
+#![warn(missing_debug_implementations)]
+
+#[macro_use]
+mod macros;
+
+pub mod allocator;
+pub mod blocking_pool;
+pub mod distributor;
+pub mod load_balancer;
+pub mod placement;
+pub mod pool;
+pub mod run;
+pub mod run_queue;
+pub mod sleepers;
+pub mod thread_recovery;
+pub mod worker;
+
+pub mod prelude {
+    pub use crate::pool::*;
+    pub use crate::run::*;
+}

bastion-executor/src/load_balancer.rs

@@ -0,0 +1,67 @@
+use super::placement;
+use lazy_static::*;
+
+use std::thread;
+
+use super::load_balancer;
+use crossbeam_utils::sync::ShardedLock;
+use rustc_hash::FxHashMap;
+
+pub struct LoadBalancer();
+
+impl LoadBalancer {
+    pub fn sample() {
+        thread::Builder::new()
+            .name("load-balancer-thread".to_string())
+            .spawn(move || {
+                loop {
+                    let mut m = 0_usize;
+                    if let Ok(stats) = load_balancer::stats().try_read() {
+                        m = stats
+                            .smp_queues
+                            .values()
+                            .sum::<usize>()
+                            .wrapping_div(placement::get_core_ids().unwrap().len());
+                    }
+
+                    if let Ok(mut stats) = load_balancer::stats().try_write() {
+                        stats.mean_level = m;
+                    }
+                }
+            })
+            .expect("load-balancer couldn't start");
+    }
+}
+
+#[derive(Clone)]
+pub struct Stats {
+    pub(crate) global_run_queue: usize,
+    pub(crate) mean_level: usize,
+    pub(crate) smp_queues: FxHashMap<usize, usize>,
+}
+
+unsafe impl Send for Stats {}
+unsafe impl Sync for Stats {}
+
+#[inline]
+pub fn stats() -> &'static ShardedLock<Stats> {
+    lazy_static! {
+        static ref LB_STATS: ShardedLock<Stats> = {
+            let stats = Stats {
+                global_run_queue: 0,
+                mean_level: 0,
+                smp_queues: FxHashMap::with_capacity_and_hasher(
+                    placement::get_core_ids().unwrap().len(),
+                    Default::default()
+                )
+            };
+
+            // Start sampler
+            LoadBalancer::sample();
+
+            // Return stats
+            ShardedLock::new(stats)
+        };
+    }
+    &*LB_STATS
+}

bastion-executor/src/macros.rs

@@ -0,0 +1,9 @@
+#[doc(hidden)]
+macro_rules! unstable_api {
+    ($($block:item)*) => {
+        $(
+            #[cfg(feature = "unstable")]
+            $block
+        )*
+    }
+}