forked from M-Labs/artiq-zynq
Compare commits
No commits in common. "7c2cc51ae3d493e97fbbe8d03062accbf5d0ee71" and "90312e2fb795bde167074be3d321878374d4b4af" have entirely different histories.
7c2cc51ae3
...
90312e2fb7
|
|
@ -4,7 +4,9 @@ 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}; #[cfg(has_drtio)] use io::Cursor;
|
||||
use futures::{future::FutureExt, select_biased};
|
||||
#[cfg(has_drtio)]
|
||||
use io::Cursor;
|
||||
#[cfg(has_drtio)]
|
||||
use ksupport::rpc;
|
||||
use ksupport::{kernel, resolve_channel_name};
|
||||
|
|
@ -805,82 +807,89 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
|
|||
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() => (),
|
||||
}
|
||||
}
|
||||
});
|
||||
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();
|
||||
|
||||
// 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() => (),
|
||||
}
|
||||
}
|
||||
});
|
||||
select_biased! {
|
||||
_ = (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();
|
||||
|
||||
task::spawn(async move {
|
||||
loop {
|
||||
select_biased! {
|
||||
_ = (async {
|
||||
let temp_s = Some(TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap()); // ALT: implement dummy
|
||||
// tcp stream that is empty (another reason the priv construct is annoying)
|
||||
if let Some(&stream) = stream_opt {
|
||||
let _ = stream.flush().await;
|
||||
let _ = stream.abort().await;
|
||||
}
|
||||
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
|
||||
}
|
||||
_ = (async {
|
||||
restart_idle.wait_async().await;
|
||||
}).fuse() => {
|
||||
if semaphore.3.try_wait().is_none() {
|
||||
semaphore.3.signal(); // STOP IDLE
|
||||
semaphore.2.signal(); // NEW IDLE
|
||||
info!("awaiting connection");
|
||||
if let Ok(mut stream) = TcpStream::accept(1381, 0x10_000, 0x10_000).await {
|
||||
info!("accepted connection");
|
||||
task::spawn(async move {
|
||||
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() => (),
|
||||
}
|
||||
});
|
||||
|
||||
|
|
@ -941,15 +950,3 @@ 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),+)
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -201,11 +201,10 @@ 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");
|
||||
if key == "idle_kernel" {
|
||||
restart_idle.1.async_wait().await?;
|
||||
restart_idle.0.signal();
|
||||
}
|
||||
write_i8(stream, Reply::Success as i8).await?;
|
||||
if key == "idle_kernel" {
|
||||
restart_idle.signal();
|
||||
}
|
||||
} else {
|
||||
// this is an error because we do not expect write to fail
|
||||
error!("failed to write: {:?}", value);
|
||||
|
|
|
|||
Loading…
Reference in New Issue