working on the executor

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>

src/executor/mod.rs

@@ -1,5 +1,6 @@
 use std::io;
 
+pub mod slot_list;
 pub mod thread_pool;
 
 pub fn num_cpus() -> io::Result<usize> {

src/executor/slot_list.rs

@@ -0,0 +1,31 @@
+pub struct SlotList<T> {
+    tasks: Vec<Option<T>>,
+    free_slots: Vec<usize>,
+}
+
+impl<T> SlotList<T> {
+    pub fn new() -> Self {
+        Self {
+            tasks: Vec::new(),
+            free_slots: Vec::new(),
+        }
+    }
+
+    pub fn add(&mut self, data: T) -> usize {
+        if let Some(id) = self.free_slots.pop() {
+            let old = self.tasks[id].replace(data);
+            assert!(old.is_none());
+            id
+        } else {
+            let id = self.tasks.len();
+            self.tasks.push(Some(data));
+            id
+        }
+    }
+
+    pub fn remove(&mut self, id: usize) -> T {
+        let entry = self.tasks[id].take().unwrap();
+        self.free_slots.push(id);
+        entry
+    }
+}

src/executor/thread_pool.rs

@@ -1,28 +1,41 @@
+use std::future::Future;
 use std::io;
-use std::sync::{Arc, Mutex};
+use std::sync::{mpsc, Arc, Mutex};
 use std::thread::JoinHandle;
 
 use super::num_cpus;
+use super::slot_list::SlotList;
+
+type BoxFut = Box<dyn Future<Output = ()> + Send + 'static>;
 
 pub struct ThreadPool {
-    inner: Arc<Inner>,
+    inner: Arc<ThreadPoolInner>,
 }
 
-pub struct Inner {
+pub struct ThreadPoolInner {
     threads: Mutex<Vec<Thread>>,
+    tasks: Mutex<SlotList<BoxFut>>,
 }
 
 pub struct Thread {
     handle: JoinHandle<()>,
+    inner: Arc<ThreadInner>,
+    queue_sender: mpsc::Sender<Work>,
+}
+
+pub struct ThreadInner {
     id: usize,
 }
 
+pub struct Work {}
+
 impl ThreadPool {
     pub fn new() -> io::Result<Self> {
         let count = num_cpus()?;
 
-        let inner = Arc::new(Inner {
+        let inner = Arc::new(ThreadPoolInner {
             threads: Mutex::new(Vec::new()),
+            tasks: Mutex::new(SlotList::new()),
         });
 
         let mut threads = Vec::with_capacity(count);
@@ -34,16 +47,46 @@ impl ThreadPool {
 
         Ok(ThreadPool { inner })
     }
+
+    pub fn spawn<T>(&self, future: T)
+    where
+        T: Future<Output = ()> + Send + 'static,
+    {
+        self.inner.spawn(Box::new(future))
+    }
+}
+
+impl ThreadPoolInner {
+    fn spawn(&self, future: BoxFut) {
+        let mut tasks = self.tasks.lock().unwrap();
+        self.queue_task(tasks.add(future));
+    }
+
+    fn queue_task(&self, task: usize) {
+        //
+    }
 }
 
 impl Thread {
-    fn new(pool: Arc<Inner>, id: usize) -> Self {
-        let handle = std::thread::spawn(move || Self::thread_main(pool, id));
-        Self { handle, id }
-    }
+    fn new(pool: Arc<ThreadPoolInner>, id: usize) -> Self {
+        let (queue_sender, queue_receiver) = mpsc::channel();
 
-    fn thread_main(pool: Arc<Inner>, thread_id: usize) {
-        let _ = pool;
-        let _ = thread_id;
+        let inner = Arc::new(ThreadInner { id });
+
+        let handle = std::thread::spawn({
+            let inner = Arc::clone(&inner);
+            move || inner.thread_main(queue_receiver)
+        });
+        Thread {
+            handle,
+            inner,
+            queue_sender,
+        }
+    }
+}
+
+impl ThreadInner {
+    fn thread_main(self: Arc<Self>, queue: mpsc::Receiver<Work>) {
+        let _ = queue;
     }
 }