runtime/comms: limited concurrent connections to 1 only.

examples/dma.py

@@ -8,11 +8,12 @@ class DMAPulses(EnvExperiment):
 
     @kernel
     def record(self):
-        with self.core_dma.record("pulses"):
+        with self.core_dma.record("pulse"):
-            # all RTIO operations now go to the "pulses"
+            delay(200*ms)
+            # all RTIO operations now go to the "pulse"
             # DMA buffer, instead of being executed immediately.
-            self.led0.pulse(100*ns)
+            self.led0.pulse(500*ms)
-            delay(100*ns)
+
 
     @kernel
     def run(self):
@@ -20,7 +21,6 @@ class DMAPulses(EnvExperiment):
         self.record()
         # prefetch the address of the DMA buffer
         # for faster playback trigger
-        pulses_handle = self.core_dma.get_handle("pulses")
+        pulse_handle = self.core_dma.get_handle("pulse")
         self.core.break_realtime()
-        self.core_dma.playback_handle(pulses_handle)
+        self.core_dma.playback_handle(pulse_handle)
-

src/Cargo.lock

@@ -201,7 +201,7 @@ dependencies = [
 [[package]]
 name = "libasync"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zc706.git#0aa75d3544a2092a0ba6ce689c3f025f22ec30e4"
+source = "git+https://git.m-labs.hk/M-Labs/zc706.git#b65606f2d02fab273645835a102048b23c3394f7"
 dependencies = [
  "embedded-hal",
  "libcortex_a9",
@@ -213,7 +213,7 @@ dependencies = [
 [[package]]
 name = "libboard_zynq"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zc706.git#0aa75d3544a2092a0ba6ce689c3f025f22ec30e4"
+source = "git+https://git.m-labs.hk/M-Labs/zc706.git#b65606f2d02fab273645835a102048b23c3394f7"
 dependencies = [
  "bit_field",
  "embedded-hal",
@@ -237,7 +237,7 @@ dependencies = [
 [[package]]
 name = "libcortex_a9"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zc706.git#0aa75d3544a2092a0ba6ce689c3f025f22ec30e4"
+source = "git+https://git.m-labs.hk/M-Labs/zc706.git#b65606f2d02fab273645835a102048b23c3394f7"
 dependencies = [
  "bit_field",
  "libregister",
@@ -252,7 +252,7 @@ checksum = "c7d73b3f436185384286bd8098d17ec07c9a7d2388a6599f824d8502b529702a"
 [[package]]
 name = "libregister"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zc706.git#0aa75d3544a2092a0ba6ce689c3f025f22ec30e4"
+source = "git+https://git.m-labs.hk/M-Labs/zc706.git#b65606f2d02fab273645835a102048b23c3394f7"
 dependencies = [
  "bit_field",
  "vcell",
@@ -262,7 +262,7 @@ dependencies = [
 [[package]]
 name = "libsupport_zynq"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zc706.git#0aa75d3544a2092a0ba6ce689c3f025f22ec30e4"
+source = "git+https://git.m-labs.hk/M-Labs/zc706.git#b65606f2d02fab273645835a102048b23c3394f7"
 dependencies = [
  "compiler_builtins",
  "libboard_zynq",

src/runtime/Cargo.toml

@@ -26,7 +26,7 @@ log_buffer = { version = "1.2" }
 libm = { version = "0.2", features = ["unstable"] }
 
 libboard_zynq = { git = "https://git.m-labs.hk/M-Labs/zc706.git" }
-libsupport_zynq = { default-features = false, git = "https://git.m-labs.hk/M-Labs/zc706.git" }
+libsupport_zynq = { default-features = false, features = ["alloc_core"], git = "https://git.m-labs.hk/M-Labs/zc706.git" }
 libcortex_a9 = { git = "https://git.m-labs.hk/M-Labs/zc706.git" }
 libasync = { git = "https://git.m-labs.hk/M-Labs/zc706.git" }
 libregister = { git = "https://git.m-labs.hk/M-Labs/zc706.git" }

src/runtime/link.x

@@ -48,9 +48,12 @@ SECTIONS
 
     .heap (NOLOAD) : ALIGN(8)
     {
-        __heap_start = .;
+        __heap0_start = .;
-        . += 0x1000000;
+        . += 0x800000;
-        __heap_end = .;
+        __heap0_end = .;
+        __heap1_start = .;
+        . += 0x800000;
+        __heap1_end = .;
     } > SDRAM
 
     .stack1 (NOLOAD) : ALIGN(8)

src/runtime/src/comms.rs

@@ -2,7 +2,6 @@ use core::fmt;
 use core::cell::RefCell;
 use core::str::Utf8Error;
 use alloc::rc::Rc;
-use alloc::sync::Arc;
 use alloc::{vec, vec::Vec, string::String};
 use log::{info, warn, error};
 
@@ -19,6 +18,8 @@ use libboard_zynq::{
     },
     timer::GlobalTimer,
 };
+use libcortex_a9::sync_channel;
+use futures::{select_biased, future::FutureExt};
 use libasync::{smoltcp::{Sockets, TcpStream}, task};
 
 use crate::config;
@@ -126,7 +127,7 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
     control.borrow_mut().tx.async_send(kernel::Message::StartRequest).await;
     loop {
         let reply = control.borrow_mut().rx.async_recv().await;
-        match *reply {
+        match reply {
             kernel::Message::RpcSend { is_async, data } => {
                 if stream.is_none() {
                     error!("Unexpected RPC from startup/idle kernel!");
@@ -141,7 +142,7 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
                     match host_request {
                         Request::RPCReply => {
                             let tag = read_bytes(stream, 512).await?;
-                            let slot = match *control.borrow_mut().rx.async_recv().await {
+                            let slot = match control.borrow_mut().rx.async_recv().await {
                                 kernel::Message::RpcRecvRequest(slot) => slot,
                                 other => panic!("expected root value slot from core1, not {:?}", other),
                             };
@@ -155,7 +156,7 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
                                     } else {
                                         let mut control = control.borrow_mut();
                                         control.tx.async_send(kernel::Message::RpcRecvReply(Ok(size))).await;
-                                        match *control.rx.async_recv().await {
+                                        match control.rx.async_recv().await {
                                             kernel::Message::RpcRecvRequest(slot) => slot,
                                             other => panic!("expected nested value slot from kernel CPU, not {:?}", other),
                                         }
@@ -166,7 +167,7 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
                         },
                         Request::RPCException => {
                             let mut control = control.borrow_mut();
-                            match *control.rx.async_recv().await {
+                            match control.rx.async_recv().await {
                                 kernel::Message::RpcRecvRequest(_) => (),
                                 other => panic!("expected (ignored) root value slot from kernel CPU, not {:?}", other),
                             }
@@ -231,11 +232,12 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
 }
 
 
-async fn load_kernel(buffer: Vec<u8>, control: &Rc<RefCell<kernel::Control>>, stream: Option<&TcpStream>) -> Result<()> {
+async fn load_kernel(buffer: &Vec<u8>, control: &Rc<RefCell<kernel::Control>>, stream: Option<&TcpStream>) -> Result<()> {
     let mut control = control.borrow_mut();
-    control.tx.async_send(kernel::Message::LoadRequest(Arc::new(buffer))).await;
+    control.restart();
+    control.tx.async_send(kernel::Message::LoadRequest(buffer.to_vec())).await;
     let reply = control.rx.async_recv().await;
-    match *reply {
+    match reply {
         kernel::Message::LoadCompleted => {
             if let Some(stream) = stream {
                 write_header(stream, Reply::LoadCompleted).await?;
@@ -279,7 +281,7 @@ async fn handle_connection(stream: &TcpStream, control: Rc<RefCell<kernel::Contr
             },
             Request::LoadKernel => {
                 let buffer = read_bytes(stream, 1024*1024).await?;
-                load_kernel(buffer, &control, Some(stream)).await?;
+                load_kernel(&buffer, &control, Some(stream)).await?;
             },
             Request::RunKernel => {
                 handle_run_kernel(Some(stream), &control).await?;
@@ -338,9 +340,10 @@ pub fn main(timer: GlobalTimer, cfg: &config::Config) {
     moninj::start(timer);
 
     let control: Rc<RefCell<kernel::Control>> = Rc::new(RefCell::new(kernel::Control::start()));
+    let idle_kernel = Rc::new(cfg.read("idle").ok());
     if let Ok(buffer) = cfg.read("startup") {
         info!("Loading startup kernel...");
-        if let Ok(()) = task::block_on(load_kernel(buffer, &control, None)) {
+        if let Ok(()) = task::block_on(load_kernel(&buffer, &control, None)) {
             info!("Starting startup kernel...");
             let _ = task::block_on(handle_run_kernel(None, &control));
             info!("Startup kernel finished!");
@@ -350,14 +353,53 @@ pub fn main(timer: GlobalTimer, cfg: &config::Config) {
     }
 
     task::spawn(async move {
+        let (tx, rx) = sync_channel!(u32, 1);
+        let tx = RefCell::new(tx);
+        let rx = Rc::new(RefCell::new(rx));
+        let has_connection = Rc::new(RefCell::new(false));
         loop {
             let stream = TcpStream::accept(1381, 2048, 2048).await.unwrap();
+            let has_connection = has_connection.clone();
+
+            if *has_connection.borrow() {
+                let mut tx = tx.borrow_mut();
+                tx.async_send(42).await;
+                // the second send is used to block until another connection received the abort
+                // request.
+                tx.async_send(42).await;
+            }
+
+            *has_connection.borrow_mut() = true;
             let control = control.clone();
-            task::spawn(async {
+            let idle_kernel = idle_kernel.clone();
-                info!("received connection");
+            let _ = rx.borrow_mut().try_recv();
-                let _ = handle_connection(&stream, control)
+
-                    .await
+            let new_rx = rx.clone();
-                    .map_err(|e| warn!("connection terminated: {}", e));
+            task::spawn(async move {
+                let mut new_rx = new_rx.borrow_mut();
+                select_biased! {
+                    _ = (async {
+                        let _ = handle_connection(&stream, control.clone())
+                            .await
+                            .map_err(|e| warn!("connection terminated: {}", e));
+                        if let Some(buffer) = &*idle_kernel {
+                            info!("Loading idle kernel");
+                            let _ = load_kernel(&buffer, &control, None)
+                                .await.map_err(|e| warn!("error loading idle kernel"));
+                            info!("Running idle kernel");
+                            let _ = handle_run_kernel(None, &control)
+                                .await.map_err(|e| warn!("error running idle kernel"));
+                            info!("Idle kernel terminated");
+                        }
+                    }).fuse() => (),
+                    _ = new_rx.async_recv().fuse() => ()
+                }
+                *has_connection.borrow_mut() = false;
+                // it is possible that when `handle_connection` is terminating,
+                // another connection sent an abort request and get blocked,
+                // so we try_recv here to unblock in that case.
+                let _ = new_rx.try_recv();
+                core::mem::drop(new_rx);
                 let _ = stream.flush().await;
                 let _ = stream.abort().await;
             });

src/runtime/src/irq.rs

@@ -0,0 +1,46 @@
+use libboard_zynq::{gic, mpcore, println, stdio};
+use libcortex_a9::{
+    asm,
+    regs::{MPIDR, SP},
+};
+use libregister::{RegisterR, RegisterW};
+use core::sync::atomic::{AtomicBool, Ordering};
+
+extern "C" {
+    static mut __stack1_start: u32;
+    fn main_core1() -> !;
+}
+
+static CORE1_RESTART: AtomicBool = AtomicBool::new(false);
+
+#[link_section = ".text.boot"]
+#[no_mangle]
+#[naked]
+pub unsafe extern "C" fn IRQ() {
+    if MPIDR.read().cpu_id() == 1 {
+        let mpcore = mpcore::RegisterBlock::new();
+        let mut gic = gic::InterruptController::new(mpcore);
+        let id = gic.get_interrupt_id();
+        if id.0 == 0 {
+            gic.end_interrupt(id);
+            asm::exit_irq();
+            SP.write(&mut __stack1_start as *mut _ as u32);
+            asm::enable_irq();
+            CORE1_RESTART.store(false, Ordering::Relaxed);
+            asm::sev();
+            main_core1();
+        }
+    }
+    stdio::drop_uart();
+    println!("IRQ");
+    loop {}
+}
+
+pub fn restart_core1() {
+    let mut interrupt_controller = gic::InterruptController::new(mpcore::RegisterBlock::new());
+    CORE1_RESTART.store(true, Ordering::Relaxed);
+    interrupt_controller.send_sgi(gic::InterruptId(0), gic::CPUCore::Core1.into());
+    while CORE1_RESTART.load(Ordering::Relaxed) {
+        asm::wfe();
+    }
+}

src/runtime/src/kernel/cache.rs

@@ -3,7 +3,6 @@ use libcortex_a9::mutex::Mutex;
 use cslice::{CSlice, AsCSlice};
 use core::mem::transmute;
 use core::str;
-use log::debug;
 
 use crate::artiq_raise;
 

src/runtime/src/kernel/control.rs

@@ -1,25 +1,49 @@
-use libcortex_a9::sync_channel::{self, sync_channel};
+use libcortex_a9::sync_channel::{Sender, Receiver};
 use libsupport_zynq::boot::Core1;
 
 use super::{CHANNEL_0TO1, CHANNEL_1TO0, Message};
+use crate::irq::restart_core1;
+
+use core::mem::{forget, replace};
 
 pub struct Control {
-    pub tx: sync_channel::Sender<Message>,
+    pub tx: Sender<'static, Message>,
-    pub rx: sync_channel::Receiver<Message>,
+    pub rx: Receiver<'static, Message>,
+}
+
+fn get_channels() -> (Sender<'static, Message>, Receiver<'static, Message>) {
+    let mut core0_tx = None;
+    while core0_tx.is_none() {
+        core0_tx = CHANNEL_0TO1.lock().take();
+    }
+    let core0_tx = core0_tx.unwrap();
+
+    let mut core0_rx = None;
+    while core0_rx.is_none() {
+        core0_rx = CHANNEL_1TO0.lock().take();
+    }
+    let core0_rx = core0_rx.unwrap();
+
+    (core0_tx, core0_rx)
 }
 
 impl Control {
     pub fn start() -> Self {
         Core1::start(true);
-
+        let (core0_tx, core0_rx) = get_channels();
-        let (core0_tx, core1_rx) = sync_channel(4);
-        let (core1_tx, core0_rx) = sync_channel(4);
-        *CHANNEL_0TO1.lock() = Some(core1_rx);
-        *CHANNEL_1TO0.lock() = Some(core1_tx);
 
         Control {
             tx: core0_tx,
             rx: core0_rx,
         }
     }
+
+    pub fn restart(&mut self) {
+        restart_core1();
+        let (core0_tx, core0_rx) = get_channels();
+        // dangling pointer here, so we forget it
+        forget(replace(&mut self.tx, core0_tx));
+        forget(replace(&mut self.rx, core0_rx));
+    }
 }
+

src/runtime/src/kernel/core1.rs

@@ -9,7 +9,10 @@ use libcortex_a9::{
     enable_fpu,
     cache::{dcci_slice, iciallu, bpiall},
     asm::{dsb, isb},
+    sync_channel,
 };
+use libboard_zynq::{mpcore, gic};
+use libsupport_zynq::ram;
 use dyld::{self, Library};
 use crate::eh_artiq;
 use super::{
@@ -138,23 +141,23 @@ pub fn main_core1() {
     enable_fpu();
     debug!("FPU enabled on Core1");
 
-    let mut core1_tx = None;
+    ram::init_alloc_core1();
-    while core1_tx.is_none() {
+    gic::InterruptController::new(mpcore::RegisterBlock::new()).enable_interrupts();
-        core1_tx = CHANNEL_1TO0.lock().take();
-    }
-    let mut core1_tx = core1_tx.unwrap();
 
-    let mut core1_rx = None;
+    let (mut core0_tx, mut core1_rx) = sync_channel!(Message, 4);
-    while core1_rx.is_none() {
+    let (mut core1_tx, core0_rx) = sync_channel!(Message, 4);
-        core1_rx = CHANNEL_0TO1.lock().take();
+    unsafe {
+        core0_tx.reset();
+        core1_tx.reset();
     }
-    let mut core1_rx = core1_rx.unwrap();
+    *CHANNEL_0TO1.lock() = Some(core0_tx);
+    *CHANNEL_1TO0.lock() = Some(core0_rx);
 
     // set on load, cleared on start
     let mut loaded_kernel = None;
     loop {
         let message = core1_rx.recv();
-        match *message {
+        match message {
             Message::LoadRequest(data) => {
                 let result = dyld::load(&data, &resolve)
                     .and_then(KernelImage::new);
@@ -213,8 +216,6 @@ pub fn terminate(exception: &'static eh_artiq::Exception<'static>, backtrace: &'
         let mut core1_tx = KERNEL_CHANNEL_1TO0.lock();
         core1_tx.as_mut().unwrap().send(Message::KernelException(exception, &backtrace[..cursor]));
     }
-    // TODO: remove after implementing graceful kernel termination.
-    error!("Core1 uncaught exception");
     loop {}
 }
 

src/runtime/src/kernel/dma.rs

@@ -258,7 +258,7 @@ pub extern fn dma_retrieve(name: CSlice<u8>) -> DmaTrace {
 pub extern fn dma_playback(timestamp: i64, ptr: i32) {
     assert!(ptr % ALIGNMENT as i32 == 0);
 
-    debug!("DMA Playback");
+    debug!("DMA playback started");
     unsafe {
         csr::rtio_dma::base_address_write(ptr as u32);
         csr::rtio_dma::time_offset_write(timestamp as u64);
@@ -268,6 +268,8 @@ pub extern fn dma_playback(timestamp: i64, ptr: i32) {
         while csr::rtio_dma::enable_read() != 0 {}
         csr::cri_con::selected_write(0);
 
+        debug!("DMA playback finished");
+
         let error = csr::rtio_dma::error_read();
         if error != 0 {
             let timestamp = csr::rtio_dma::error_timestamp_read();

src/runtime/src/kernel/mod.rs

@@ -1,5 +1,5 @@
 use core::ptr;
-use alloc::{vec::Vec, sync::Arc, string::String};
+use alloc::{vec::Vec, string::String};
 
 use libcortex_a9::{mutex::Mutex, sync_channel};
 use crate::eh_artiq;
@@ -12,7 +12,7 @@ mod rpc;
 mod dma;
 mod cache;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct RPCException {
     pub name: String,
     pub message: String,
@@ -23,24 +23,24 @@ pub struct RPCException {
     pub function: String
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum Message {
-    LoadRequest(Arc<Vec<u8>>),
+    LoadRequest(Vec<u8>),
     LoadCompleted,
     LoadFailed,
     StartRequest,
     KernelFinished,
     KernelException(&'static eh_artiq::Exception<'static>, &'static [usize]),
-    RpcSend { is_async: bool, data: Arc<Vec<u8>> },
+    RpcSend { is_async: bool, data: Vec<u8> },
     RpcRecvRequest(*mut ()),
     RpcRecvReply(Result<usize, RPCException>),
 }
 
-static CHANNEL_0TO1: Mutex<Option<sync_channel::Receiver<Message>>> = Mutex::new(None);
+static CHANNEL_0TO1: Mutex<Option<sync_channel::Sender<'static, Message>>> = Mutex::new(None);
-static CHANNEL_1TO0: Mutex<Option<sync_channel::Sender<Message>>> = Mutex::new(None);
+static CHANNEL_1TO0: Mutex<Option<sync_channel::Receiver<'static, Message>>> = Mutex::new(None);
 
-static KERNEL_CHANNEL_0TO1: Mutex<Option<sync_channel::Receiver<Message>>> = Mutex::new(None);
+static KERNEL_CHANNEL_0TO1: Mutex<Option<sync_channel::Receiver<'static, Message>>> = Mutex::new(None);
-static KERNEL_CHANNEL_1TO0: Mutex<Option<sync_channel::Sender<Message>>> = Mutex::new(None);
+static KERNEL_CHANNEL_1TO0: Mutex<Option<sync_channel::Sender<'static, Message>>> = Mutex::new(None);
 
 static mut KERNEL_IMAGE: *const core1::KernelImage = ptr::null();
 

src/runtime/src/kernel/rpc.rs

@@ -1,6 +1,6 @@
 //! Kernel-side RPC API
 
-use alloc::{vec::Vec, sync::Arc};
+use alloc::vec::Vec;
 use cslice::{CSlice, AsCSlice};
 
 use crate::eh_artiq;
@@ -14,7 +14,7 @@ fn rpc_send_common(is_async: bool, service: u32, tag: &CSlice<u8>, data: *const
     let mut core1_tx = KERNEL_CHANNEL_1TO0.lock();
     let mut buffer = Vec::<u8>::new();
     send_args(&mut buffer, service, tag.as_ref(), data).expect("RPC encoding failed");
-    core1_tx.as_mut().unwrap().send(Message::RpcSend { is_async, data: Arc::new(buffer) });
+    core1_tx.as_mut().unwrap().send(Message::RpcSend { is_async, data: buffer });
 }
 
 pub extern fn rpc_send(service: u32, tag: &CSlice<u8>, data: *const *const ()) {
@@ -32,7 +32,7 @@ pub extern fn rpc_recv(slot: *mut ()) -> usize {
         core1_tx.as_mut().unwrap().send(Message::RpcRecvRequest(slot));
         core1_rx.as_mut().unwrap().recv()
     };
-    match *reply {
+    match reply {
         Message::RpcRecvReply(Ok(alloc_size)) => alloc_size,
         Message::RpcRecvReply(Err(exception)) => unsafe {
             eh_artiq::raise(&eh_artiq::Exception {

src/runtime/src/main.rs

@@ -6,13 +6,15 @@
 #![feature(c_variadic)]
 #![feature(const_btree_new)]
 #![feature(ptr_offset_from)]
+#![feature(const_in_array_repeat_expressions)]
+#![feature(naked_functions)]
 
 extern crate alloc;
 
 use core::{cmp, str};
 use log::{info, warn, error};
 
-use libboard_zynq::{timer::GlobalTimer, devc, slcr};
+use libboard_zynq::{timer::GlobalTimer, devc, slcr, mpcore, gic};
 use libasync::{task, block_async};
 use libsupport_zynq::ram;
 use libregister::RegisterW;
@@ -38,6 +40,7 @@ mod panic;
 mod logger;
 mod mgmt;
 mod analyzer;
+mod irq;
 
 fn init_gateware() {
     // Set up PS->PL clocks
@@ -183,7 +186,8 @@ pub fn main_core0() {
 
     info!("NAR3/Zynq7000 starting...");
 
-    ram::init_alloc_linker();
+    ram::init_alloc_core0();
+    gic::InterruptController::new(mpcore::RegisterBlock::new()).enable_interrupts();
 
     init_gateware();
     info!("detected gateware: {}", identifier_read(&mut [0; 64]));

src/szl/link.x

@@ -34,6 +34,12 @@ SECTIONS
         __bss_end = .;
     } > OCM3
 
+    .heap (NOLOAD) : ALIGN(8)
+    {
+        __heap0_start = .;
+        __heap0_end = .;
+    } > OCM3
+
     .stack1 (NOLOAD) : ALIGN(8)
     {
         __stack1_end = .;