srenblad
/
artiq-zynq
forked from M-Labs/artiq-zynq
1
0
Fork 0
Code Pull Requests Activity

changing to semaphore heavy solution

This commit is contained in:
Simon Renblad 2024-09-05 17:20:20 +08:00
parent b485bf16a7
commit 7c2cc51ae3
1 changed files with 85 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

203
src/runtime/src/comms.rs
View File

@ -4,9 +4,7 @@ use core::{cell::RefCell, fmt, slice, str};
use core_io::Error as IoError; use core_io::Error as IoError;
use cslice::CSlice; use cslice::CSlice;
use dyld::elf; 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)] #[cfg(has_drtio)]
use ksupport::rpc; use ksupport::rpc;
use ksupport::{kernel, resolve_channel_name}; use ksupport::{kernel, resolve_channel_name};
@ -807,135 +805,82 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
let restart_idle = Rc::new(Semaphore::new(0, 1)); let restart_idle = Rc::new(Semaphore::new(0, 1));
mgmt::start(cfg.clone(), restart_idle.clone()); 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 { task::spawn(async move {
let terminate = Rc::new(Semaphore::new(0, 1));
let running = Rc::new(Mutex::new(false));
let repeat_lock = Rc::new(Mutex::new(false));
let restart_idle = restart_idle.clone();
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();
// copy same idea as artiq mainline
// loop
// check if socket.is_active() -> accept and run kernel task
// check if task is terminated -> run idle -> terminate idle on restart_idle
// yield and reloop (annoying thing is if async behaves as expected)
// start listening on a socket and return the socket
let listener = TcpListener::new(10_000);
listener.listen(1381);
loop { loop {
if listener.can_accept() { // ie can_accept (socket in state SYN-RECEIVED) clone_mult!(semaphores, control, up_destinations, aux_mutex, routing_table);
let mut stream = listener.accept().await; semaphores.0.async_wait().await;
if connection.try_wait().is_none() { semaphores.1.try_wait();
// there is an existing connection let _ = task_lock.async_lock().await;
terminate.signal(); select_biased! {
connection.async_wait().await; _ = (async {
} // run connection
let mut stream = stream_opt.borrow_mut()?;
let _ = terminate.try_wait(); let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
task::spawn(async move { .await
select_biased! { .map_err(|e| warn!("connection terminated: {}", e));
_ = (async { }).fuse() => {
semaphores.2.signal();
}) => (), },
_ = terminate.async_wait().fuse() => () _ = semaphores.1.async_wait().fuse() => (),
}
let _ = stream.flush().await;
let _ = stream.abort().await;
connection.signal();
});
} }
if connection.try_wait().is_some() {
let _ = terminate.try_wait(); // ensure terminate and restart_idle are always set
let _ = restart_idle.try_wait();
task::spawn(async move {
select_biased! {
_ = (async {
// needs to ensure that if there is no idle_kernel, enter idle loop and
// keep yielding
}) => (),
_ = terminate.async_wait().fuse() => (),
_ = async {
restart_idle.async_wait().fuse(),
}
}
connection.signal();
});
}
// yield and rerun
task::r#yield().await;
} }
});
select_biased! { // handle idle
_ = (async { task::spawn(async move {
loop { loop {
let cfg = cfg.clone(); clone_mult!(semaphores, control, up_destinations, aux_mutex, routing_table);
let control = control.clone(); semaphores.2.async_wait().await;
let up_destinations = up_destinations.clone(); semaphores.3.try_wait();
let aux_mutex = aux_mutex.clone(); let _ = task_lock.async_lock().await;
let routing_table = drtio_routing_table.clone(); select_biased! {
let terminate = terminate.clone(); _ = (async {
let running = running.clone(); if let Some(buffer) = cfg.read("idle_kernel").ok() {
let repeat_lock = repeat_lock.clone(); load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await;
let mut stream = TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap();
task::spawn(async move {
let _rep_lock = repeat_lock.async_lock().await;
terminate.signal();
let _lock = running.async_lock().await;
drop(_rep_lock);
let routing_table = routing_table.borrow();
select_biased! {
_ = (async {
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;
});
} }
}).fuse() => (), }).fuse() => (),
_ = (async { _ = semaphores.3.async_wait().fuse() => (),
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; task::spawn(async move {
drop(_rep_lock); loop {
let _lock = running.async_lock().await; select_biased! {
let routing_table = routing_table.borrow(); _ = (async {
select_biased! { let temp_s = Some(TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap()); // ALT: implement dummy
_ = (async { // tcp stream that is empty (another reason the priv construct is annoying)
if let Some(buffer) = cfg.read("idle_kernel").ok() { if let Some(&stream) = stream_opt {
load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await; let _ = stream.flush().await;
} let _ = stream.abort().await;
}).fuse() => (),
_ = (async {
restart_idle.0.async_wait().await;
restart_idle.1.signal();
}).fuse() => (),
} }
drop(_lock); stream_opt.replace(temp_s);
}).fuse() => {
if semaphore.1.try_wait().is_none() {
semaphore.1.signal(); // STOP KERNEL
}
if semaphore.3.try_wait().is_none() {
semaphore.3.signal(); // STOP IDLE
}
// if the other stream exists -> flush it and replace with the new stream
semaphore.0.signal(); // NEW KERNEL
} }
}).fuse() => (), _ = (async {
restart_idle.wait_async().await;
}).fuse() => {
if semaphore.3.try_wait().is_none() {
semaphore.3.signal(); // STOP IDLE
semaphore.2.signal(); // NEW IDLE
}
}
}
} }
}); });
@ -996,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)); 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),+)
};
}