Register denali on yellowstone, handle read requests.

rust/lib/mammoth/src/task/mod.rs

@@ -26,9 +26,7 @@ impl TaskId {
 
 pub struct Task {
     id: TaskId,
-    // FIXME: This only needs to be sync because of the CURRENT_EXECUTOR
-    // needing to be shared between threads.
-    future: Pin<Box<dyn Future<Output = ()> + Sync + Send>>,
+    future: Pin<Box<dyn Future<Output = ()> + Send>>,
 }
 
 impl Task {
@@ -72,7 +70,7 @@ impl Wake for TaskWaker {
 }
 
 pub struct Executor {
-    tasks: BTreeMap<TaskId, Task>,
+    tasks: Arc<Mutex<BTreeMap<TaskId, Task>>>,
     // TODO: Consider a better datastructure for this.
     task_queue: Arc<Mutex<VecDeque<TaskId>>>,
     waker_cache: BTreeMap<TaskId, Waker>,
@@ -81,7 +79,7 @@ pub struct Executor {
 impl Executor {
     pub fn new() -> Executor {
         Executor {
-            tasks: BTreeMap::new(),
+            tasks: Arc::new(Mutex::new(BTreeMap::new())),
             task_queue: Arc::new(Mutex::new(VecDeque::new())),
             waker_cache: BTreeMap::new(),
         }
@@ -89,7 +87,7 @@ impl Executor {
 
     pub fn spawn(&mut self, task: Task) {
         let task_id = task.id;
-        if self.tasks.insert(task_id, task).is_some() {
+        if self.tasks.lock().insert(task_id, task).is_some() {
             panic!("Task is already existed in executor map");
         }
         self.task_queue.lock().push_back(task_id);
@@ -97,7 +95,8 @@ impl Executor {
 
     fn run_ready_tasks(&mut self) {
         while let Some(task_id) = self.task_queue.lock().pop_front() {
-            let task = self.tasks.get_mut(&task_id).unwrap();
+            let mut tasks = self.tasks.lock();
+            let task = tasks.get_mut(&task_id).unwrap();
             let waker = self
                 .waker_cache
                 .entry(task_id)
@@ -105,7 +104,7 @@ impl Executor {
             let mut ctx = Context::from_waker(waker);
             match task.poll(&mut ctx) {
                 Poll::Ready(()) => {
-                    self.tasks.remove(&task_id);
+                    tasks.remove(&task_id);
                     self.waker_cache.remove(&task_id);
                 }
                 Poll::Pending => {}
@@ -120,6 +119,30 @@ impl Executor {
             syscall::thread_sleep(50).unwrap();
         }
     }
+
+    pub fn new_spawner(&self) -> Spawner {
+        Spawner::new(self.tasks.clone(), self.task_queue.clone())
+    }
 }
 
-static CURRENT_EXECUTOR: Option<Executor> = None;
+pub struct Spawner {
+    tasks: Arc<Mutex<BTreeMap<TaskId, Task>>>,
+    task_queue: Arc<Mutex<VecDeque<TaskId>>>,
+}
+
+impl Spawner {
+    fn new(
+        tasks: Arc<Mutex<BTreeMap<TaskId, Task>>>,
+        task_queue: Arc<Mutex<VecDeque<TaskId>>>,
+    ) -> Self {
+        Spawner { tasks, task_queue }
+    }
+
+    pub fn spawn(&self, task: Task) {
+        let task_id = task.id;
+        if self.tasks.lock().insert(task_id, task).is_some() {
+            panic!("Task is already existed in executor map");
+        }
+        self.task_queue.lock().push_back(task_id);
+    }
+}

rust/lib/yunq/src/server.rs

@@ -4,9 +4,8 @@ use crate::buffer::ByteBuffer;
 use alloc::sync::Arc;
 use alloc::vec::Vec;
 use mammoth::cap::Capability;
-use mammoth::sync::Mutex;
 use mammoth::syscall;
-use mammoth::task::Executor;
+use mammoth::task::Spawner;
 use mammoth::task::Task;
 use mammoth::thread;
 use mammoth::thread::JoinHandle;
@@ -70,7 +69,7 @@ pub trait AsyncYunqServer
 where
     Self: Send + Sync + 'static,
 {
-    fn server_loop(self: Arc<Self>, executor: Arc<Mutex<Executor>>) {
+    fn server_loop(self: Arc<Self>, spawner: Spawner) {
         loop {
             let mut byte_buffer = ByteBuffer::<1024>::new();
             let mut cap_buffer = vec![0; 10];
@@ -85,7 +84,7 @@ where
                 .at::<u64>(8)
                 .expect("Failed to access request length.");
             let self_clone = self.clone();
-            executor.lock().spawn(Task::new((async move || {
+            spawner.spawn(Task::new((async move || {
                 self_clone
                     .handle_request_and_response(method, byte_buffer, cap_buffer, reply_port_cap)
                     .await;
@@ -137,11 +136,11 @@ where
     ) -> impl Future<Output = Result<usize, ZError>> + Sync + Send;
 }
 
-pub fn spawn_async_server_thread<T>(server: Arc<T>, executor: Arc<Mutex<Executor>>) -> JoinHandle
+pub fn spawn_async_server_thread<T>(server: Arc<T>, spawner: Spawner) -> JoinHandle
 where
     T: AsyncYunqServer + Send + Sync + 'static,
 {
     thread::spawn(move || {
-        server.server_loop(executor);
+        server.server_loop(spawner);
     })
 }