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

restructure runkernel comms

This commit is contained in:
Simon Renblad 2024-08-09 11:33:53 +08:00
parent 43a6dc8838
commit 69e7d88dbc
1 changed files with 58 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -809,33 +809,71 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
task::spawn(async move { task::spawn(async move {
let terminate = Rc::new(Semaphore::new(0, 1)); let terminate = Rc::new(Semaphore::new(0, 1));
let running = Rc::new(Mutex::new(false)); let running = Rc::new(Mutex::new(false));
let repeat_lock = Rc::new(Mutex::new(false));
let cfg = cfg.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();
select_biased! { select_biased! {
_ = (async { _ = (async {
loop { loop {
let mut stream = TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap(); let cfg = cfg.clone();
terminate.signal(); 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();
info!("awaiting connection");
if let Ok(mut stream) = TcpStream::accept(1381, 0x10_000, 0x10_000).await {
info!("accepted connection");
task::spawn(async move { task::spawn(async move {
let _lock = running.lock(); 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(); let routing_table = routing_table.borrow();
select_biased! { select_biased! {
_ = (async { _ = (async {
info!("pre handling connection");
let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer) let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
.await .await
.map_err(|e| warn!("connection terminated: {}", e)); .map_err(|e| warn!("connection terminated: {}", e));
}).fuse() => (), }).fuse() => (),
_ = terminate.async_wait().fuse() => () _ = terminate.async_wait().fuse() => ()
} }
drop(_lock);
let _ = stream.flush().await; let _ = stream.flush().await;
let _ = stream.abort().await; let _ = stream.abort().await;
drop(_lock);
}); });
} else {
info!("failed to accept connection");
}
} }
}).fuse() => (), }).fuse() => (),
_ = (async { _ = (async {
loop { loop {
let _lock = running.lock(); 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 _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! { select_biased! {
_ = (async { _ = (async {
if let Some(buffer) = cfg.read("idle_kernel").ok() { if let Some(buffer) = cfg.read("idle_kernel").ok() {
@ -843,8 +881,8 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
} }
}).fuse() => (), }).fuse() => (),
_ = terminate.async_wait().fuse() => (), _ = terminate.async_wait().fuse() => (),
_ = update_idle.async_wait().fuse() => ()
} }
info!("supposed to drop idle kernel");
drop(_lock); drop(_lock);
} }
}).fuse() => (), }).fuse() => (),
@ -895,7 +933,7 @@ pub fn soft_panic_main(timer: GlobalTimer, cfg: Config) -> ! {
Sockets::init(32); Sockets::init(32);
mgmt::start(cfg); mgmt::start(Rc::new(cfg));
// getting eth settings disables the LED as it resets GPIO // getting eth settings disables the LED as it resets GPIO
// need to re-enable it here // need to re-enable it here