M-Labs/artiq-zynq
10
1
Fork 19
Code Issues 12 Pull Requests Releases Wiki Activity

good progress on restart_idle

Browse Source
This commit is contained in:
Simon Renblad 2024-09-06 17:31:37 +08:00
parent 7c2cc51ae3
commit d1dce8ede9
1 changed files with 38 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -805,82 +805,59 @@ 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( let terminate = Rc::new(Semaphore::new(0, 1));
(Semaphore::new(0, 1), let connection = Rc::new(Semaphore::new(1, 1));
Semaphore::new(1, 1), let finish = Rc::new(Semaphore::new(1, 1));
Semaphore::new(1, 1), let start = Rc::new(Semaphore::new(1, 1)); // run idle kernel once without needing restart
Semaphore::new(1, 1))); // turn the above into a control obj
//
// handle connects let stream_refcell = Rc::new(RefCell::new(None));
task::spawn(async move { task::spawn(async move {
loop { loop {
clone_mult!(semaphores, control, up_destinations, aux_mutex, routing_table); clone_mult!(semaphores, control, up_destinations, aux_mutex, routing_table);
semaphores.0.async_wait().await; start.async_wait().await;
semaphores.1.try_wait(); let stream_opt = stream_refcell.borrow_mut();
let _ = task_lock.async_lock().await; let _ = connection.try_wait();
select_biased! { let _ = finish.try_wait();
_ = (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! { select_biased! {
_ = (async { _ = (async {
if let Some(&mut stream) = stream_opt {
let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
.await
.map_err(|e| warn!("connection terminated: {}", e));
}
connection.signal();
if let Some(buffer) = cfg.read("idle_kernel").ok() { if let Some(buffer) = cfg.read("idle_kernel").ok() {
load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await; load_and_run_idle_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await;
} }
}).fuse() => (), }).fuse() => (), // TODO: on clean exit with existing idle -> run again?
_ = semaphores.3.async_wait().fuse() => (), _ = terminate.async_wait().fuse() => (),
} }
if let Some(&stream) = stream_opt {
stream.flush().await();
stream.abort().await();
}
finish.signal();
} }
}); });
task::spawn(async move { task::spawn(async move {
loop { loop {
select_biased! { clone_mult!(semaphores, stream_refcell);
let stream_opt = select_biased! {
temp_s = (async {
Some(Rc::new(TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap()))
}).fuse() => temp_s,
_ = (async { _ = (async {
let temp_s = Some(TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap()); // ALT: implement dummy restart_idle.async_wait().await;
// tcp stream that is empty (another reason the priv construct is annoying) connection.async_wait().await;
if let Some(&stream) = stream_opt { }).fuse() => None
let _ = stream.flush().await; };
let _ = stream.abort().await; terminate.signal();
} finish.async_wait().await;
stream_opt.replace(temp_s); stream_refcell.replace(stream_opt);
}).fuse() => { start.signal();
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
}
}
}
} }
}); });