changing to semaphore heavy solution

src/runtime/src/comms.rs

@@ -4,9 +4,7 @@ use core::{cell::RefCell, fmt, slice, str};
 use core_io::Error as IoError;
 use cslice::CSlice;
 use dyld::elf;
-use futures::{future::FutureExt, select_biased};
+use futures::{future::FutureExt, select_biased}; #[cfg(has_drtio)] use io::Cursor;
-#[cfg(has_drtio)]
-use io::Cursor;
 #[cfg(has_drtio)]
 use ksupport::rpc;
 use ksupport::{kernel, resolve_channel_name};
@@ -806,90 +804,83 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
     let cfg = Rc::new(cfg);
     let restart_idle = Rc::new(Semaphore::new(0, 1));
     mgmt::start(cfg.clone(), restart_idle.clone());
+    
+    let semaphores = Rc::new(
+        (Semaphore::new(0, 1), 
+         Semaphore::new(1, 1),
+         Semaphore::new(1, 1),
+         Semaphore::new(1, 1)));
+    
+    // handle connects
+    task::spawn(async move {
+        loop {
+            clone_mult!(semaphores, control, up_destinations, aux_mutex, routing_table);
+            semaphores.0.async_wait().await;
+            semaphores.1.try_wait();
+            let _ = task_lock.async_lock().await;
+            select_biased! {
+                _ = (async {
+                    // run connection
+                    let mut stream = stream_opt.borrow_mut()?;
+                    let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
+                        .await
+                        .map_err(|e| warn!("connection terminated: {}", e));
+                }).fuse() => {
+                    semaphores.2.signal();
+                },
+                _ = semaphores.1.async_wait().fuse() => (),
+            }
+        }
+    });
+    
+    // handle idle
+    task::spawn(async move {
+        loop {
+            clone_mult!(semaphores, control, up_destinations, aux_mutex, routing_table);
+            semaphores.2.async_wait().await;
+            semaphores.3.try_wait();
+            let _ = task_lock.async_lock().await;
+            select_biased! {
+                _ = (async {
+                    if let Some(buffer) = cfg.read("idle_kernel").ok() {
+                        load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await;
+                    }
+                }).fuse() => (),
+                _ = semaphores.3.async_wait().fuse() => (),
+            }
+        }
+    });
 
     task::spawn(async move {
-        let terminate = Rc::new(Semaphore::new(0, 1));
+        loop {
-        let running = Rc::new(Mutex::new(false));
+            select_biased! {
-        let repeat_lock = Rc::new(Mutex::new(false));
+                _ = (async {
-        let restart_idle = restart_idle.clone();
+                        let temp_s = Some(TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap()); // ALT: implement dummy
-        let cfg = cfg.clone();
+                    // tcp stream that is empty (another reason the priv construct is annoying)
-        let control = control.clone();
+                        if let Some(&stream) = stream_opt {
-        let up_destinations = up_destinations.clone();
+                           let _ = stream.flush().await;
-        let aux_mutex = aux_mutex.clone();
+                           let _ = stream.abort().await;
-        let routing_table = drtio_routing_table.clone();
+                        }
-
+                        stream_opt.replace(temp_s);
-        select_biased! {
+                    }).fuse() => {
-            _ = (async {
+                        if semaphore.1.try_wait().is_none() {
-                    loop {
+                            semaphore.1.signal(); // STOP KERNEL
-                        let cfg = cfg.clone();
+                        }
-                        let control = control.clone();
+                        if semaphore.3.try_wait().is_none() {
-                        let up_destinations = up_destinations.clone();
+                            semaphore.3.signal(); // STOP IDLE
-                        let aux_mutex = aux_mutex.clone();
+                        }
-                        let routing_table = drtio_routing_table.clone();
+                        // if the other stream exists -> flush it and replace with the new stream
-                        let terminate = terminate.clone();
+                        semaphore.0.signal(); // NEW KERNEL
-                        let running = running.clone();
+                    }
-                        let repeat_lock = repeat_lock.clone();
+                _ = (async {
-
+                        restart_idle.wait_async().await;
-                        info!("awaiting connection");
+                    }).fuse() => {
-                        if let Ok(mut stream) = TcpStream::accept(1381, 0x10_000, 0x10_000).await {
+                        if semaphore.3.try_wait().is_none() {
-                            info!("accepted connection");
+                            semaphore.3.signal(); // STOP IDLE
-                            task::spawn(async move {
+                            semaphore.2.signal(); // NEW IDLE
-                                info!("awaiting lock");
-                                let _rep_lock = repeat_lock.async_lock().await;
-                                terminate.signal();
-                                let _lock = running.async_lock().await;
-                                drop(_rep_lock);
-                                info!("lock achieved");
-                                let routing_table = routing_table.borrow();
-                                select_biased! {
-                                    _ = (async {
-                                        info!("pre handling connection");
-                                        let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
-                                            .await
-                                            .map_err(|e| warn!("connection terminated: {}", e));
-                                    }).fuse() => (),
-                                    _ = terminate.async_wait().fuse() => ()
-                                }
-                                drop(_lock);
-                                let _ = stream.flush().await;
-                                let _ = stream.abort().await;
-                            });
-                        } else {
-                            info!("failed to accept connection");
                         }
                     }
-            }).fuse() => (),
+            }
-            _ = (async {
-                    loop {
-                        let cfg = cfg.clone();
-                        let control = control.clone();
-                        let up_destinations = up_destinations.clone();
-                        let aux_mutex = aux_mutex.clone();
-                        let routing_table = drtio_routing_table.clone();
-                        let terminate = terminate.clone();
-                        let running = running.clone();
-                        let repeat_lock = repeat_lock.clone();
-                        let restart_idle = restart_idle.clone();
-                        
-                        let _rep_lock = repeat_lock.async_lock().await;
-                        drop(_rep_lock);
-                        info!("getting lock idle kernel");
-                        let _lock = running.async_lock().await;
-                        info!("locked idle kernel");
-                        let routing_table = routing_table.borrow();
-                        select_biased! {
-                                _ = (async {
-                                    if let Some(buffer) = cfg.read("idle_kernel").ok() {
-                                        load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await;
-                                    }
-                                }).fuse() => (),
-                                _ = terminate.async_wait().fuse() => (),
-                                _ = restart_idle.async_wait().fuse() => ()
-                        }
-                        info!("supposed to drop idle kernel");
-                        drop(_lock);
-                    }
-            }).fuse() => (),
         }
     });
 
@@ -950,3 +941,15 @@ pub fn soft_panic_main(timer: GlobalTimer, cfg: Config) -> ! {
 
     Sockets::run(&mut iface, || Instant::from_millis(timer.get_time().0 as i32));
 }
+
+macro_rules! clone_mult {
+    ($id:ident) => {
+        let $id = $id.clone();
+    };
+
+    // Decompose multiple `eval`s recursively
+    ($id:ident, $($ids:ident),+) => {
+        clone_mult!($id)
+        clone_mult!($($ids),+)
+    };
+}

src/runtime/src/mgmt.rs

@@ -201,10 +201,11 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
                 let value = cfg.write(&key, buffer);
                 if value.is_ok() {
                     debug!("write success");
-                    write_i8(stream, Reply::Success as i8).await?;
                     if key == "idle_kernel" {
-                        restart_idle.signal();
+                        restart_idle.1.async_wait().await?;
+                        restart_idle.0.signal();
                     }
+                    write_i8(stream, Reply::Success as i8).await?;
                 } else {
                     // this is an error because we do not expect write to fail
                     error!("failed to write: {:?}", value);