diff --git a/src/runtime/src/comms.rs b/src/runtime/src/comms.rs index 0e21c98..5d459b1 100644 --- a/src/runtime/src/comms.rs +++ b/src/runtime/src/comms.rs @@ -696,27 +696,6 @@ async fn handle_connection( } } -async fn load_and_run_idle_kernel( - buffer: &Vec, - control: &Rc>, - up_destinations: &Rc>, - aux_mutex: &Rc>, - routing_table: &drtio_routing::RoutingTable, - timer: GlobalTimer, -) { - info!("Loading idle kernel"); - let res = handle_flash_kernel(buffer, control, up_destinations, aux_mutex, routing_table, timer).await; - match res { - Err(_) => warn!("error loading idle kernel"), - _ => (), - } - info!("Running idle kernel"); - let _ = handle_run_kernel(None, control, up_destinations, aux_mutex, routing_table, timer) - .await - .map_err(|_| warn!("error running idle kernel")); - info!("Idle kernel terminated"); -} - pub fn main(timer: GlobalTimer, cfg: Config) { let net_addresses = net_settings::get_addresses(&cfg); info!("network addresses: {}", net_addresses); @@ -777,7 +756,6 @@ pub fn main(timer: GlobalTimer, cfg: Config) { moninj::start(timer, &aux_mutex, &drtio_routing_table); let control: Rc> = Rc::new(RefCell::new(kernel::Control::start())); - let idle_kernel = Rc::new(cfg.read("idle_kernel").ok()); if let Ok(buffer) = cfg.read("startup_kernel") { info!("Loading startup kernel..."); let routing_table = drtio_routing_table.borrow(); @@ -803,36 +781,27 @@ pub fn main(timer: GlobalTimer, cfg: Config) { error!("Error loading startup kernel!"); } } - - mgmt::start(cfg); + + let cfg = Rc::new(cfg); + let restart_idle = Rc::new(Semaphore::new(1, 1)); + mgmt::start(cfg.clone(), restart_idle.clone()); task::spawn(async move { - let connection = Rc::new(Semaphore::new(0, 1)); + let connection = Rc::new(Semaphore::new(1, 1)); let terminate = Rc::new(Semaphore::new(0, 1)); - { - let control = control.clone(); - let idle_kernel = idle_kernel.clone(); - let connection = connection.clone(); - let terminate = terminate.clone(); - let up_destinations = up_destinations.clone(); - let aux_mutex = aux_mutex.clone(); - let routing_table = drtio_routing_table.clone(); - task::spawn(async move { - let routing_table = routing_table.borrow(); - select_biased! { - _ = (async { - if let Some(buffer) = &*idle_kernel { - load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await; - } - }).fuse() => (), - _ = terminate.async_wait().fuse() => () - } - connection.signal(); - }); - } - + let can_restart_idle = Rc::new(Semaphore::new(1, 1)); + let restart_idle = restart_idle.clone(); loop { - let mut stream = TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap(); + let control = control.clone(); + let mut maybe_stream = select_biased! { + s = (async { + TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap() + }).fuse() => Some(s), + _ = (async { + restart_idle.async_wait().await; + can_restart_idle.async_wait().await; + }).fuse() => None + }; if connection.try_wait().is_none() { // there is an existing connection @@ -840,32 +809,56 @@ pub fn main(timer: GlobalTimer, cfg: Config) { connection.async_wait().await; } + let maybe_idle_kernel = cfg.read("idle_kernel").ok(); + if maybe_idle_kernel.is_none() && maybe_stream.is_none() { + control.borrow_mut().restart(); // terminate idle kernel if running + } + let control = control.clone(); - let idle_kernel = idle_kernel.clone(); let connection = connection.clone(); let terminate = terminate.clone(); + let can_restart_idle = can_restart_idle.clone(); let up_destinations = up_destinations.clone(); let aux_mutex = aux_mutex.clone(); let routing_table = drtio_routing_table.clone(); // we make sure the value of terminate is 0 before we start let _ = terminate.try_wait(); + let _ = can_restart_idle.try_wait(); task::spawn(async move { 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)); - if let Some(buffer) = &*idle_kernel { - load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await; + if let Some(stream) = &mut maybe_stream { + let _ = handle_connection(stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer) + .await + .map_err(|e| warn!("connection terminated: {}", e)); + } + can_restart_idle.signal(); + match maybe_idle_kernel { + Some(buffer) => { + info!("loading idle kernel"); + match handle_flash_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await { + Ok(_) => { + info!("running idle kernel"); + match handle_run_kernel(None, &control, &up_destinations, &aux_mutex, &routing_table, timer).await { + Ok(_) => info!("idle kernel finished"), + Err(_) => warn!("idle kernel running error") + } + }, + Err(_) => warn!("idle kernel loading error") + } + }, + None => info!("no idle kernel found") } }).fuse() => (), _ = terminate.async_wait().fuse() => () } connection.signal(); - let _ = stream.flush().await; - let _ = stream.abort().await; + if let Some(stream) = maybe_stream { + let _ = stream.flush().await; + let _ = stream.abort().await; + } }); } }); @@ -913,8 +906,9 @@ pub fn soft_panic_main(timer: GlobalTimer, cfg: Config) -> ! { }; Sockets::init(32); - - mgmt::start(cfg); + + let dummy = Rc::new(Semaphore::new(0, 1)); + mgmt::start(Rc::new(cfg), dummy); // getting eth settings disables the LED as it resets GPIO // need to re-enable it here diff --git a/src/runtime/src/mgmt.rs b/src/runtime/src/mgmt.rs index bfeed9b..e802316 100644 --- a/src/runtime/src/mgmt.rs +++ b/src/runtime/src/mgmt.rs @@ -6,6 +6,7 @@ use libasync::{smoltcp::TcpStream, task}; use libboard_artiq::logger::{BufferLogger, LogBufferRef}; use libboard_zynq::{slcr, smoltcp}; use libconfig::Config; +use libcortex_a9::semaphore::Semaphore; use log::{self, debug, error, info, warn, LevelFilter}; use num_derive::FromPrimitive; use num_traits::FromPrimitive; @@ -111,7 +112,7 @@ async fn read_key(stream: &mut TcpStream) -> Result { Ok(String::from_utf8(buffer).unwrap()) } -async fn handle_connection(stream: &mut TcpStream, pull_id: Rc>, cfg: Rc) -> Result<()> { +async fn handle_connection(stream: &mut TcpStream, pull_id: Rc>, cfg: Rc, restart_idle: Rc) -> Result<()> { if !expect(&stream, b"ARTIQ management\n").await? { return Err(Error::UnexpectedPattern); } @@ -200,6 +201,9 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc>, cf let value = cfg.write(&key, buffer); if value.is_ok() { debug!("write success"); + if key == "idle_kernel" { + restart_idle.signal(); + } write_i8(stream, Reply::Success as i8).await?; } else { // this is an error because we do not expect write to fail @@ -213,6 +217,9 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc>, cf let value = cfg.remove(&key); if value.is_ok() { debug!("erase success"); + if key == "idle_kernel" { + restart_idle.signal(); + } write_i8(stream, Reply::Success as i8).await?; } else { warn!("erase failed"); @@ -229,17 +236,17 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc>, cf } } -pub fn start(cfg: Config) { +pub fn start(cfg: Rc, restart_idle: Rc) { task::spawn(async move { let pull_id = Rc::new(RefCell::new(0u32)); - let cfg = Rc::new(cfg); loop { let mut stream = TcpStream::accept(1380, 2048, 2048).await.unwrap(); let pull_id = pull_id.clone(); let cfg = cfg.clone(); + let restart_idle = restart_idle.clone(); task::spawn(async move { info!("received connection"); - let _ = handle_connection(&mut stream, pull_id, cfg) + let _ = handle_connection(&mut stream, pull_id, cfg, restart_idle) .await .map_err(|e| warn!("connection terminated: {:?}", e)); let _ = stream.flush().await;