Stack representation

bastion-executor/Cargo.toml

@@ -9,6 +9,10 @@ edition = "2018"
 
 
 [dependencies]
+crossbeam-utils = "0.6"
+crossbeam-epoch = "0.7"
 lazy_static = "1.4"
 libc = "0.2"
 num_cpus = "1.10"
+
+lightproc = { "path" = "../lightproc" }

bastion-executor/src/distributor.rs

@@ -1,6 +1,8 @@
-use crate::placement;
-use crate::placement::CoreId;
+use super::placement;
+use super::placement::CoreId;
 use std::thread;
+use super::run_queue::{Worker, Stealer};
+use lightproc::prelude::*;
 
 pub(crate) struct Distributor {
     pub round: usize,
@@ -17,13 +19,17 @@ impl Distributor {
         }
     }
 
-    pub fn assign<P>(mut self, thunk: P)
+    pub fn assign<P>(mut self, thunk: P) -> Vec<Stealer<LightProc>>
     where
         P: Fn() + Send + Sync + Copy + 'static,
     {
+        let mut stealers = Vec::<Stealer<LightProc>>::new();
         for core in self.cores {
             self.round = core.id;
 
+            let worker = Worker::new_fifo();
+            stealers.push(worker.stealer());
+
             thread::Builder::new()
                 .name("bastion-async-thread".to_string())
                 .spawn(move || {
@@ -35,5 +41,7 @@ impl Distributor {
                 })
                 .expect("cannot start the thread for running proc");
         }
+
+        stealers
     }
 }

bastion-executor/src/lib.rs

@@ -3,3 +3,7 @@ pub mod distributor;
 pub mod placement;
 pub mod pool;
 pub mod thread_recovery;
+pub mod load_balancer;
+pub mod run_queue;
+pub mod sleepers;
+pub mod worker;

bastion-executor/src/load_balancer.rs

@@ -0,0 +1,27 @@
+use lazy_static::*;
+use std::thread;
+
+pub struct LoadBalancer();
+
+impl LoadBalancer {
+    pub fn trigger() {
+        unimplemented!()
+    }
+}
+
+#[inline]
+pub(crate) fn launch() -> &'static LoadBalancer {
+    lazy_static! {
+        static ref LOAD_BALANCER: LoadBalancer = {
+            thread::Builder::new()
+                .name("load-balancer-thread".to_string())
+                .spawn(|| {
+
+                })
+                .expect("load-balancer couldn't start");
+
+            LoadBalancer()
+        };
+    }
+    &*LOAD_BALANCER
+}

bastion-executor/src/placement.rs

@@ -288,30 +288,30 @@ mod macos {
 
     use super::CoreId;
 
-    type kern_return_t = c_int;
-    type integer_t = c_int;
-    type natural_t = c_uint;
-    type thread_t = c_uint;
-    type thread_policy_flavor_t = natural_t;
-    type mach_msg_type_number_t = natural_t;
+    type KernReturnT = c_int;
+    type IntegerT = c_int;
+    type NaturalT = c_uint;
+    type ThreadT = c_uint;
+    type ThreadPolicyFlavorT = NaturalT;
+    type MachMsgTypeNumberT = NaturalT;
 
     #[repr(C)]
-    struct thread_affinity_policy_data_t {
-        affinity_tag: integer_t,
+    struct ThreadAffinityPolicyDataT {
+        affinity_tag: IntegerT,
     }
 
-    type thread_policy_t = *mut thread_affinity_policy_data_t;
+    type ThreadPolicyT = *mut ThreadAffinityPolicyDataT;
 
-    const THREAD_AFFINITY_POLICY: thread_policy_flavor_t = 4;
+    const THREAD_AFFINITY_POLICY: ThreadPolicyFlavorT = 4;
 
     #[link(name = "System", kind = "framework")]
     extern "C" {
         fn thread_policy_set(
-            thread: thread_t,
-            flavor: thread_policy_flavor_t,
-            policy_info: thread_policy_t,
-            count: mach_msg_type_number_t,
-        ) -> kern_return_t;
+            thread: ThreadT,
+            flavor: ThreadPolicyFlavorT,
+            policy_info: ThreadPolicyT,
+            count: MachMsgTypeNumberT,
+        ) -> KernReturnT;
     }
 
     pub fn get_core_ids() -> Option<Vec<CoreId>> {
@@ -324,20 +324,20 @@ mod macos {
     }
 
     pub fn set_for_current(core_id: CoreId) {
-        let THREAD_AFFINITY_POLICY_COUNT: mach_msg_type_number_t =
-            mem::size_of::<thread_affinity_policy_data_t>() as mach_msg_type_number_t
-                / mem::size_of::<integer_t>() as mach_msg_type_number_t;
+        let thread_affinity_policy_count: MachMsgTypeNumberT =
+            mem::size_of::<ThreadAffinityPolicyDataT>() as MachMsgTypeNumberT
+                / mem::size_of::<IntegerT>() as MachMsgTypeNumberT;
 
-        let mut info = thread_affinity_policy_data_t {
-            affinity_tag: core_id.id as integer_t,
+        let mut info = ThreadAffinityPolicyDataT {
+            affinity_tag: core_id.id as IntegerT,
         };
 
         unsafe {
             thread_policy_set(
-                pthread_self() as thread_t,
+                pthread_self() as ThreadT,
                 THREAD_AFFINITY_POLICY,
-                &mut info as thread_policy_t,
-                THREAD_AFFINITY_POLICY_COUNT,
+                &mut info as ThreadPolicyT,
+                thread_affinity_policy_count,
             );
         }
     }

bastion-executor/src/pool.rs

@@ -1,19 +1,41 @@
-use crate::distributor::Distributor;
+use super::distributor::Distributor;
+use super::run_queue::{Worker, Injector, Stealer};
 use lazy_static::*;
+use lightproc::prelude::*;
+use super::sleepers::Sleepers;
 
-pub struct Pool {}
+pub struct Pool {
+    pub injector: Injector<LightProc>,
+    pub stealers: Vec<Stealer<LightProc>>,
+    pub sleepers: Sleepers,
+}
+
+impl Pool {
+    /// Error recovery for the fallen threads
+    pub fn recover_async_thread() {
+        unimplemented!()
+    }
+
+    pub fn fetch_proc(&self, local: &Worker<LightProc>) -> Option<LightProc> {
+        // Pop only from the local queue with full trust
+        local.pop()
+    }
+}
 
 #[inline]
 pub fn get() -> &'static Pool {
     lazy_static! {
         static ref POOL: Pool = {
             let distributor = Distributor::new();
-
-            distributor.assign(|| {
+            let stealers = distributor.assign(|| {
                 println!("1,2,3");
             });
 
-            Pool {}
+            Pool {
+                injector: Injector::new(),
+                stealers,
+                sleepers: Sleepers::new(),
+            }
         };
     }
     &*POOL