src/Cargo.lock

@@ -218,6 +218,33 @@ dependencies = [
  "libsupport_zynq",
 ]
 
+[[package]]
+name = "ksupport"
+version = "0.1.0"
+dependencies = [
+ "build_zynq",
+ "byteorder",
+ "core_io",
+ "cslice",
+ "dwarf",
+ "dyld",
+ "io",
+ "libasync",
+ "libboard_artiq",
+ "libboard_zynq",
+ "libc",
+ "libconfig",
+ "libcortex_a9",
+ "libm",
+ "libregister",
+ "libsupport_zynq",
+ "log",
+ "log_buffer",
+ "nb 0.1.3",
+ "unwind",
+ "void",
+]
+
 [[package]]
 name = "libasync"
 version = "0.0.0"
@@ -438,18 +465,17 @@ dependencies = [
  "embedded-hal",
  "futures",
  "io",
+ "ksupport",
  "libasync",
  "libboard_artiq",
  "libboard_zynq",
  "libc",
  "libconfig",
  "libcortex_a9",
- "libm",
  "libregister",
  "libsupport_zynq",
  "log",
  "log_buffer",
- "nb 0.1.3",
  "num-derive",
  "num-traits",
  "unwind",
@@ -471,7 +497,11 @@ name = "satman"
 version = "0.0.0"
 dependencies = [
  "build_zynq",
+ "core_io",
+ "cslice",
  "embedded-hal",
+ "io",
+ "ksupport",
  "libasync",
  "libboard_artiq",
  "libboard_zynq",

src/Cargo.toml

@@ -5,6 +5,7 @@ members = [
   "libdwarf",
   "libio",
   "libunwind",
+  "libksupport",
   "runtime",
   "satman"
 ]

src/gateware/kasli_soc.py

@@ -470,7 +470,7 @@ class GenericSatellite(SoCCore):
         self.comb += self.drtiosat.async_errors.eq(self.local_io.async_errors)
 
         self.submodules.cri_con = rtio.CRIInterconnectShared(
-            [self.drtiosat.cri, self.rtio_dma.cri],
+            [self.drtiosat.cri, self.rtio_dma.cri, self.rtio.cri],
             [self.local_io.cri] + self.drtio_cri,
             enable_routing=True)
         self.csr_devices.append("cri_con")

src/gateware/zc706.py

@@ -476,7 +476,7 @@ class _SatelliteBase(SoCCore):
 
         self.submodules.local_io = SyncRTIO(self.rtio_tsc, rtio_channels)
         self.submodules.cri_con = rtio.CRIInterconnectShared(
-            [self.drtiosat.cri, self.rtio_dma.cri],
+            [self.drtiosat.cri, self.rtio_dma.cri, self.rtio.cri],
             [self.local_io.cri] + self.drtio_cri,
             enable_routing=True)
         self.csr_devices.append("cri_con")

src/libio/cursor.rs

@@ -1,3 +1,6 @@
+#[cfg(feature = "alloc")]
+use alloc::vec::Vec;
+
 use core_io::{Error as IoError, Read, Write};
 
 #[derive(Debug, Clone)]
@@ -64,7 +67,7 @@ impl Write for Cursor<&mut [u8]> {
 }
 
 #[cfg(feature = "alloc")]
-impl Write for Cursor<::alloc::Vec<u8>> {
+impl Write for Cursor<Vec<u8>> {
     #[inline]
     fn write(&mut self, buf: &[u8]) -> Result<usize, IoError> {
         self.inner.extend_from_slice(buf);

src/libio/lib.rs

@@ -1,13 +1,10 @@
 #![no_std]
 #![feature(never_type)]
-#![cfg_attr(feature = "alloc", feature(alloc))]
 
+#[cfg(feature = "alloc")]
 extern crate alloc;
 extern crate core_io;
 
-#[cfg(feature = "alloc")]
-#[macro_use]
-use alloc;
 #[cfg(feature = "byteorder")]
 extern crate byteorder;
 

src/libio/proto.rs

@@ -1,3 +1,4 @@
+#[cfg(feature = "alloc")]
 use alloc::{string::String, vec};
 use core::str::Utf8Error;
 
@@ -50,7 +51,8 @@ pub trait ProtoRead {
     }
 
     #[inline]
-    fn read_bytes(&mut self) -> Result<::alloc::vec::Vec<u8>, Self::ReadError> {
+    #[cfg(feature = "alloc")]
+    fn read_bytes(&mut self) -> Result<vec::Vec<u8>, Self::ReadError> {
         let length = self.read_u32()?;
         let mut value = vec![0; length as usize];
         self.read_exact(&mut value)?;
@@ -58,7 +60,8 @@ pub trait ProtoRead {
     }
 
     #[inline]
-    fn read_string(&mut self) -> Result<::alloc::string::String, ReadStringError<Self::ReadError>> {
+    #[cfg(feature = "alloc")]
+    fn read_string(&mut self) -> Result<String, ReadStringError<Self::ReadError>> {
         let bytes = self.read_bytes().map_err(ReadStringError::Other)?;
         String::from_utf8(bytes).map_err(|err| ReadStringError::Utf8(err.utf8_error()))
     }
@@ -135,6 +138,7 @@ pub trait ProtoWrite {
     }
 
     #[inline]
+    #[cfg(feature = "alloc")]
     fn write_string(&mut self, value: &str) -> Result<(), Self::WriteError> {
         self.write_bytes(value.as_bytes())
     }

src/libksupport/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "ksupport"
+description = "Kernel support for Zynq-based platforms"
+version = "0.1.0"
+authors = ["M-Labs"]
+edition = "2018"
+
+[build-dependencies]
+build_zynq = { path = "../libbuild_zynq" }
+
+[dependencies]
+cslice = "0.3"
+log = "0.4"
+nb = "0.1"
+core_io = { version = "0.1", features = ["collections"] }
+byteorder = { version = "1.3", default-features = false }
+void = { version = "1", default-features = false }
+log_buffer = { version = "1.2" }
+libm = { version = "0.2", features = ["unstable"] }
+
+libboard_zynq = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git", features = ["ipv6"]}
+libsupport_zynq = { default-features = false, features = ["alloc_core"], git = "https://git.m-labs.hk/M-Labs/zynq-rs.git" }
+libcortex_a9 = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git" }
+libasync = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git" }
+libregister = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git" }
+libconfig = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git", features = ["fat_lfn", "ipv6"] }
+
+dyld = { path = "../libdyld" }
+dwarf = { path = "../libdwarf" }
+unwind = { path = "../libunwind" }
+libc = { path = "../libc" }
+io = { path = "../libio" }
+libboard_artiq = { path = "../libboard_artiq" }

src/libksupport/build.rs

@@ -0,0 +1,5 @@
+extern crate build_zynq;
+
+fn main() {
+    build_zynq::cfg();
+}

src/libksupport/src/kernel/api.rs

@@ -5,12 +5,12 @@ use libc::{c_char, c_int, size_t};
 use libm;
 use log::{info, warn};
 
+#[cfg(has_drtio)]
+use super::subkernel;
 use super::{cache,
             core1::rtio_get_destination_status,
             dma,
             rpc::{rpc_recv, rpc_send, rpc_send_async}};
-#[cfg(has_drtio)]
-use super::subkernel;
 use crate::{eh_artiq, i2c, rtio};
 
 extern "C" {

src/libksupport/src/kernel/mod.rs

@@ -3,7 +3,7 @@ use core::ptr;
 
 use libcortex_a9::{mutex::Mutex, semaphore::Semaphore, sync_channel};
 
-use crate::eh_artiq;
+use crate::{eh_artiq, RPCException};
 
 mod control;
 pub use control::Control;
@@ -16,17 +16,6 @@ mod cache;
 #[cfg(has_drtio)]
 mod subkernel;
 
-#[derive(Debug, Clone)]
-pub struct RPCException {
-    pub id: u32,
-    pub message: u32,
-    pub param: [i64; 3],
-    pub file: u32,
-    pub line: i32,
-    pub column: i32,
-    pub function: u32,
-}
-
 #[cfg(has_drtio)]
 #[derive(Debug, Clone)]
 pub enum SubkernelStatus {
@@ -109,6 +98,8 @@ pub enum Message {
         data: Vec<u8>,
     },
     #[cfg(has_drtio)]
+    SubkernelMsgSent,
+    #[cfg(has_drtio)]
     SubkernelMsgRecvRequest {
         id: u32,
         timeout: u64,

src/libksupport/src/kernel/subkernel.rs

@@ -62,6 +62,10 @@ pub extern "C" fn send_message(id: u32, count: u8, tag: &CSlice<u8>, data: *cons
             data: buffer[3..].to_vec(),
         });
     }
+    match unsafe { KERNEL_CHANNEL_0TO1.as_mut().unwrap() }.recv() {
+        Message::SubkernelMsgSent => (),
+        _ => panic!("expected SubkernelMsgSent after SubkernelMsgSend"),
+    }
 }
 
 pub extern "C" fn await_message(id: u32, timeout: u64, min: u8, max: u8) {

src/libksupport/src/lib.rs

@@ -0,0 +1,163 @@
+#![no_std]
+#![feature(c_variadic)]
+#![feature(const_btree_new)]
+#![feature(const_in_array_repeat_expressions)]
+#![feature(naked_functions)]
+#![feature(asm)]
+
+#[macro_use]
+extern crate alloc;
+
+use alloc::{collections::BTreeMap, string::String};
+
+use io::{Cursor, ProtoRead};
+use libasync::block_async;
+use libconfig::Config;
+use log::{error, warn};
+#[cfg(has_drtiosat)]
+pub use pl::csr::drtiosat as rtio_core;
+#[cfg(has_rtio_core)]
+pub use pl::csr::rtio_core;
+use void::Void;
+
+pub mod eh_artiq;
+pub mod i2c;
+pub mod irq;
+pub mod kernel;
+pub mod rpc;
+#[cfg(ki_impl = "csr")]
+#[path = "rtio_csr.rs"]
+pub mod rtio;
+#[cfg(ki_impl = "acp")]
+#[path = "rtio_acp.rs"]
+pub mod rtio;
+#[rustfmt::skip]
+#[path = "../../../build/pl.rs"]
+pub mod pl;
+
+
+#[derive(Debug, Clone)]
+pub struct RPCException {
+    pub id: u32,
+    pub message: u32,
+    pub param: [i64; 3],
+    pub file: u32,
+    pub line: i32,
+    pub column: i32,
+    pub function: u32,
+}
+
+pub static mut SEEN_ASYNC_ERRORS: u8 = 0;
+
+pub const ASYNC_ERROR_COLLISION: u8 = 1 << 0;
+pub const ASYNC_ERROR_BUSY: u8 = 1 << 1;
+pub const ASYNC_ERROR_SEQUENCE_ERROR: u8 = 1 << 2;
+
+pub unsafe fn get_async_errors() -> u8 {
+    let errors = SEEN_ASYNC_ERRORS;
+    SEEN_ASYNC_ERRORS = 0;
+    errors
+}
+
+fn wait_for_async_rtio_error() -> nb::Result<(), Void> {
+    unsafe {
+        #[cfg(has_rtio_core)]
+        let errors = rtio_core::async_error_read();
+        #[cfg(has_drtiosat)]
+        let errors = rtio_core::protocol_error_read();
+        if errors != 0 {
+            Ok(())
+        } else {
+            Err(nb::Error::WouldBlock)
+        }
+    }
+}
+
+pub async fn report_async_rtio_errors() {
+    loop {
+        let _ = block_async!(wait_for_async_rtio_error()).await;
+        unsafe {
+            #[cfg(has_rtio_core)]
+            let errors = rtio_core::async_error_read();
+            #[cfg(has_drtiosat)]
+            let errors = rtio_core::protocol_error_read();
+            if errors & ASYNC_ERROR_COLLISION != 0 {
+                let channel = rtio_core::collision_channel_read();
+                error!(
+                    "RTIO collision involving channel 0x{:04x}:{}",
+                    channel,
+                    resolve_channel_name(channel as u32)
+                );
+            }
+            if errors & ASYNC_ERROR_BUSY != 0 {
+                let channel = rtio_core::busy_channel_read();
+                error!(
+                    "RTIO busy error involving channel 0x{:04x}:{}",
+                    channel,
+                    resolve_channel_name(channel as u32)
+                );
+            }
+            if errors & ASYNC_ERROR_SEQUENCE_ERROR != 0 {
+                let channel = rtio_core::sequence_error_channel_read();
+                error!(
+                    "RTIO sequence error involving channel 0x{:04x}:{}",
+                    channel,
+                    resolve_channel_name(channel as u32)
+                );
+            }
+            SEEN_ASYNC_ERRORS = errors;
+            #[cfg(has_rtio_core)]
+            rtio_core::async_error_write(errors);
+            #[cfg(has_drtiosat)]
+            rtio_core::protocol_error_write(errors);
+        }
+    }
+}
+
+static mut RTIO_DEVICE_MAP: BTreeMap<u32, String> = BTreeMap::new();
+
+fn read_device_map(cfg: &Config) -> BTreeMap<u32, String> {
+    let mut device_map: BTreeMap<u32, String> = BTreeMap::new();
+    let _ = cfg
+        .read("device_map")
+        .and_then(|raw_bytes| {
+            let mut bytes_cr = Cursor::new(raw_bytes);
+            let size = bytes_cr.read_u32().unwrap();
+            for _ in 0..size {
+                let channel = bytes_cr.read_u32().unwrap();
+                let device_name = bytes_cr.read_string().unwrap();
+                if let Some(old_entry) = device_map.insert(channel, device_name.clone()) {
+                    warn!(
+                        "conflicting device map entries for RTIO channel {}: '{}' and '{}'",
+                        channel, old_entry, device_name
+                    );
+                }
+            }
+            Ok(())
+        })
+        .or_else(|err| {
+            warn!(
+                "error reading device map ({}), device names will not be available in RTIO error messages",
+                err
+            );
+            Err(err)
+        });
+    device_map
+}
+
+fn _resolve_channel_name(channel: u32, device_map: &BTreeMap<u32, String>) -> String {
+    match device_map.get(&channel) {
+        Some(val) => val.clone(),
+        None => String::from("unknown"),
+    }
+}
+
+pub fn resolve_channel_name(channel: u32) -> String {
+    _resolve_channel_name(channel, unsafe { &RTIO_DEVICE_MAP })
+}
+
+pub fn setup_device_map(cfg: &Config) {
+    unsafe {
+        RTIO_DEVICE_MAP = read_device_map(cfg);
+    }
+}

src/libksupport/src/rpc.rs

@@ -1,256 +1,62 @@
-use alloc::boxed::Box;
-#[cfg(has_drtio)]
-use alloc::vec::Vec;
-use core::{future::Future, str};
+use core::str;
 
-use async_recursion::async_recursion;
 use byteorder::{ByteOrder, NativeEndian};
-use core_io::{Error, Write};
+use core_io::{Error, Read, Write};
 use cslice::{CMutSlice, CSlice};
-#[cfg(has_drtio)]
-use io::{Cursor, ProtoRead};
-use io::ProtoWrite;
-use libasync::smoltcp::TcpStream;
-use libboard_zynq::smoltcp;
+use io::{ProtoRead, ProtoWrite};
 use log::trace;
 
 use self::tag::{split_tag, Tag, TagIterator};
-use crate::proto_async;
 
 #[inline]
-fn round_up(val: usize, power_of_two: usize) -> usize {
+pub fn round_up(val: usize, power_of_two: usize) -> usize {
     assert!(power_of_two.is_power_of_two());
     let max_rem = power_of_two - 1;
     (val + max_rem) & (!max_rem)
 }
 
 #[inline]
-unsafe fn round_up_mut<T>(ptr: *mut T, power_of_two: usize) -> *mut T {
+pub unsafe fn round_up_mut<T>(ptr: *mut T, power_of_two: usize) -> *mut T {
     round_up(ptr as usize, power_of_two) as *mut T
 }
 
 #[inline]
-unsafe fn round_up_const<T>(ptr: *const T, power_of_two: usize) -> *const T {
+pub unsafe fn round_up_const<T>(ptr: *const T, power_of_two: usize) -> *const T {
     round_up(ptr as usize, power_of_two) as *const T
 }
 
 #[inline]
-unsafe fn align_ptr<T>(ptr: *const ()) -> *const T {
+pub unsafe fn align_ptr<T>(ptr: *const ()) -> *const T {
     round_up_const(ptr, core::mem::align_of::<T>()) as *const T
 }
 
 #[inline]
-unsafe fn align_ptr_mut<T>(ptr: *mut ()) -> *mut T {
+pub unsafe fn align_ptr_mut<T>(ptr: *mut ()) -> *mut T {
     round_up_mut(ptr, core::mem::align_of::<T>()) as *mut T
 }
 
-/// Reads (deserializes) `length` array or list elements of type `tag` from `stream`,
-/// writing them into the buffer given by `storage`.
-///
-/// `alloc` is used for nested allocations (if elements themselves contain
-/// lists/arrays), see [recv_value].
-#[async_recursion(?Send)]
-async unsafe fn recv_elements<F>(
-    stream: &TcpStream,
-    elt_tag: Tag<'async_recursion>,
-    length: usize,
-    storage: *mut (),
-    alloc: &(impl Fn(usize) -> F + 'async_recursion),
-) -> Result<(), smoltcp::Error>
-where
-    F: Future<Output = *mut ()>,
-{
-    // List of simple types are special-cased in the protocol for performance.
-    match elt_tag {
-        Tag::Bool => {
-            let dest = core::slice::from_raw_parts_mut(storage as *mut u8, length);
-            proto_async::read_chunk(stream, dest).await?;
-        }
-        Tag::Int32 => {
-            let ptr = storage as *mut u32;
-            let dest = core::slice::from_raw_parts_mut(ptr as *mut u8, length * 4);
-            proto_async::read_chunk(stream, dest).await?;
-            drop(dest);
-            let dest = core::slice::from_raw_parts_mut(ptr, length);
-            NativeEndian::from_slice_u32(dest);
-        }
-        Tag::Int64 | Tag::Float64 => {
-            let ptr = storage as *mut u64;
-            let dest = core::slice::from_raw_parts_mut(ptr as *mut u8, length * 8);
-            proto_async::read_chunk(stream, dest).await?;
-            drop(dest);
-            let dest = core::slice::from_raw_parts_mut(ptr, length);
-            NativeEndian::from_slice_u64(dest);
-        }
-        _ => {
-            let mut data = storage;
-            for _ in 0..length {
-                recv_value(stream, elt_tag, &mut data, alloc).await?
-            }
-        }
-    }
-    Ok(())
-}
-
-/// Reads (deserializes) a value of type `tag` from `stream`, writing the results to
-/// the kernel-side buffer `data` (the passed pointer to which is incremented to point
-/// past the just-received data). For nested allocations (lists/arrays), `alloc` is
-/// invoked any number of times with the size of the required allocation as a parameter
-/// (which is assumed to be correctly aligned for all payload types).
-#[async_recursion(?Send)]
-async unsafe fn recv_value<F>(
-    stream: &TcpStream,
-    tag: Tag<'async_recursion>,
-    data: &mut *mut (),
-    alloc: &(impl Fn(usize) -> F + 'async_recursion),
-) -> Result<(), smoltcp::Error>
-where
-    F: Future<Output = *mut ()>,
-{
-    macro_rules! consume_value {
-        ($ty:ty, | $ptr:ident | $map:expr) => {{
-            let $ptr = align_ptr_mut::<$ty>(*data);
-            *data = $ptr.offset(1) as *mut ();
-            $map
-        }};
-    }
-
-    match tag {
-        Tag::None => Ok(()),
-        Tag::Bool => consume_value!(i8, |ptr| {
-            *ptr = proto_async::read_i8(stream).await?;
-            Ok(())
-        }),
-        Tag::Int32 => consume_value!(i32, |ptr| {
-            *ptr = proto_async::read_i32(stream).await?;
-            Ok(())
-        }),
-        Tag::Int64 | Tag::Float64 => consume_value!(i64, |ptr| {
-            *ptr = proto_async::read_i64(stream).await?;
-            Ok(())
-        }),
-        Tag::String | Tag::Bytes | Tag::ByteArray => {
-            consume_value!(CMutSlice<u8>, |ptr| {
-                let length = proto_async::read_i32(stream).await? as usize;
-                *ptr = CMutSlice::new(alloc(length).await as *mut u8, length);
-                proto_async::read_chunk(stream, (*ptr).as_mut()).await?;
-                Ok(())
-            })
-        }
-        Tag::Tuple(it, arity) => {
-            let alignment = tag.alignment();
-            *data = round_up_mut(*data, alignment);
-            let mut it = it.clone();
-            for _ in 0..arity {
-                let tag = it.next().expect("truncated tag");
-                recv_value(stream, tag, data, alloc).await?
-            }
-            // Take into account any tail padding (if element(s) with largest alignment
-            // are not at the end).
-            *data = round_up_mut(*data, alignment);
-            Ok(())
-        }
-        Tag::List(it) => {
-            #[repr(C)]
-            struct List {
-                elements: *mut (),
-                length: usize,
-            }
-            consume_value!(*mut List, |ptr_to_list| {
-                let tag = it.clone().next().expect("truncated tag");
-                let length = proto_async::read_i32(stream).await? as usize;
-
-                // To avoid multiple kernel CPU roundtrips, use a single allocation for
-                // both the pointer/length List (slice) and the backing storage for the
-                // elements. We can assume that alloc() is aligned suitably, so just
-                // need to take into account any extra padding required.
-                // (Note: At the time of writing, there will never actually be any types
-                // with alignment larger than 8 bytes, so storage_offset == 0 always.)
-                let list_size = 4 + 4;
-                let storage_offset = round_up(list_size, tag.alignment());
-                let storage_size = tag.size() * length;
-
-                let allocation = alloc(storage_offset + storage_size).await as *mut u8;
-                *ptr_to_list = allocation as *mut List;
-                let storage = allocation.offset(storage_offset as isize) as *mut ();
-
-                (**ptr_to_list).length = length;
-                (**ptr_to_list).elements = storage;
-                recv_elements(stream, tag, length, storage, alloc).await
-            })
-        }
-        Tag::Array(it, num_dims) => {
-            consume_value!(*mut (), |buffer| {
-                // Deserialize length along each dimension and compute total number of
-                // elements.
-                let mut total_len: usize = 1;
-                for _ in 0..num_dims {
-                    let len = proto_async::read_i32(stream).await? as usize;
-                    total_len *= len;
-                    consume_value!(usize, |ptr| *ptr = len)
-                }
-
-                // Allocate backing storage for elements; deserialize them.
-                let elt_tag = it.clone().next().expect("truncated tag");
-                *buffer = alloc(elt_tag.size() * total_len).await;
-                recv_elements(stream, elt_tag, total_len, *buffer, alloc).await
-            })
-        }
-        Tag::Range(it) => {
-            *data = round_up_mut(*data, tag.alignment());
-            let tag = it.clone().next().expect("truncated tag");
-            recv_value(stream, tag, data, alloc).await?;
-            recv_value(stream, tag, data, alloc).await?;
-            recv_value(stream, tag, data, alloc).await?;
-            Ok(())
-        }
-        Tag::Keyword(_) => unreachable!(),
-        Tag::Object => unreachable!(),
-    }
-}
-
-pub async fn recv_return<F>(
-    stream: &TcpStream,
-    tag_bytes: &[u8],
-    data: *mut (),
-    alloc: &impl Fn(usize) -> F,
-) -> Result<(), smoltcp::Error>
-where
-    F: Future<Output = *mut ()>,
-{
-    let mut it = TagIterator::new(tag_bytes);
-    trace!("recv ...->{}", it);
-
-    let tag = it.next().expect("truncated tag");
-    let mut data = data;
-    unsafe { recv_value(stream, tag, &mut data, alloc).await? };
-
-    Ok(())
-}
-
-// versions for Cursor rather than TcpStream
+// versions for reader rather than TcpStream
 // they will be made into sync for satellite subkernels later
-#[cfg(has_drtio)]
-#[async_recursion(?Send)]
-async unsafe fn recv_elements_cursor<F>(
-    cursor: &mut Cursor<Vec<u8>>,
-    elt_tag: Tag<'async_recursion>,
+unsafe fn recv_elements<F, R>(
+    reader: &mut R,
+    elt_tag: Tag,
     length: usize,
     storage: *mut (),
-    alloc: &(impl Fn(usize) -> F + 'async_recursion),
+    alloc: &mut F,
 ) -> Result<(), Error>
 where
-    F: Future<Output = *mut ()>,
+    F: FnMut(usize) -> *mut (),
+    R: Read + ?Sized,
 {
     match elt_tag {
         Tag::Bool => {
             let dest = core::slice::from_raw_parts_mut(storage as *mut u8, length);
-            cursor.read_exact(dest)?;
+            reader.read_exact(dest)?;
         }
         Tag::Int32 => {
             let ptr = storage as *mut u32;
             let dest = core::slice::from_raw_parts_mut(ptr as *mut u8, length * 4);
-            cursor.read_exact(dest)?;
+            reader.read_exact(dest)?;
             drop(dest);
             let dest = core::slice::from_raw_parts_mut(ptr, length);
             NativeEndian::from_slice_u32(dest);
@@ -258,7 +64,7 @@ where
         Tag::Int64 | Tag::Float64 => {
             let ptr = storage as *mut u64;
             let dest = core::slice::from_raw_parts_mut(ptr as *mut u8, length * 8);
-            cursor.read_exact(dest)?;
+            reader.read_exact(dest)?;
             drop(dest);
             let dest = core::slice::from_raw_parts_mut(ptr, length);
             NativeEndian::from_slice_u64(dest);
@@ -266,23 +72,17 @@ where
         _ => {
             let mut data = storage;
             for _ in 0..length {
-                recv_value_cursor(cursor, elt_tag, &mut data, alloc).await?
+                recv_value(reader, elt_tag, &mut data, alloc)?
             }
         }
     }
     Ok(())
 }
 
-#[cfg(has_drtio)]
-#[async_recursion(?Send)]
-async unsafe fn recv_value_cursor<F>(
-    cursor: &mut Cursor<Vec<u8>>,
-    tag: Tag<'async_recursion>,
-    data: &mut *mut (),
-    alloc: &(impl Fn(usize) -> F + 'async_recursion),
-) -> Result<(), Error>
+unsafe fn recv_value<F, R>(reader: &mut R, tag: Tag, data: &mut *mut (), alloc: &mut F) -> Result<(), Error>
 where
-    F: Future<Output = *mut ()>,
+    F: FnMut(usize) -> *mut (),
+    R: Read + ?Sized,
 {
     macro_rules! consume_value {
         ($ty:ty, | $ptr:ident | $map:expr) => {{
@@ -295,22 +95,22 @@ where
     match tag {
         Tag::None => Ok(()),
         Tag::Bool => consume_value!(i8, |ptr| {
-            *ptr = cursor.read_u8()? as i8;
+            *ptr = reader.read_u8()? as i8;
             Ok(())
         }),
         Tag::Int32 => consume_value!(i32, |ptr| {
-            *ptr = cursor.read_u32()? as i32;
+            *ptr = reader.read_u32()? as i32;
             Ok(())
         }),
         Tag::Int64 | Tag::Float64 => consume_value!(i64, |ptr| {
-            *ptr = cursor.read_u64()? as i64;
+            *ptr = reader.read_u64()? as i64;
             Ok(())
         }),
         Tag::String | Tag::Bytes | Tag::ByteArray => {
             consume_value!(CMutSlice<u8>, |ptr| {
-                let length = cursor.read_u32()? as usize;
-                *ptr = CMutSlice::new(alloc(length).await as *mut u8, length);
-                cursor.read_exact((*ptr).as_mut())?;
+                let length = reader.read_u32()? as usize;
+                *ptr = CMutSlice::new(alloc(length) as *mut u8, length);
+                reader.read_exact((*ptr).as_mut())?;
                 Ok(())
             })
         }
@@ -320,7 +120,7 @@ where
             let mut it = it.clone();
             for _ in 0..arity {
                 let tag = it.next().expect("truncated tag");
-                recv_value_cursor(cursor, tag, data, alloc).await?
+                recv_value(reader, tag, data, alloc)?
             }
             *data = round_up_mut(*data, alignment);
             Ok(())
@@ -333,41 +133,41 @@ where
             }
             consume_value!(*mut List, |ptr_to_list| {
                 let tag = it.clone().next().expect("truncated tag");
-                let length = cursor.read_u32()? as usize;
+                let length = reader.read_u32()? as usize;
 
                 let list_size = 4 + 4;
                 let storage_offset = round_up(list_size, tag.alignment());
                 let storage_size = tag.size() * length;
 
-                let allocation = alloc(storage_offset + storage_size).await as *mut u8;
+                let allocation = alloc(storage_offset + storage_size) as *mut u8;
                 *ptr_to_list = allocation as *mut List;
                 let storage = allocation.offset(storage_offset as isize) as *mut ();
 
                 (**ptr_to_list).length = length;
                 (**ptr_to_list).elements = storage;
-                recv_elements_cursor(cursor, tag, length, storage, alloc).await
+                recv_elements(reader, tag, length, storage, alloc)
             })
         }
         Tag::Array(it, num_dims) => {
             consume_value!(*mut (), |buffer| {
                 let mut total_len: usize = 1;
                 for _ in 0..num_dims {
-                    let len = cursor.read_u32()? as usize;
+                    let len = reader.read_u32()? as usize;
                     total_len *= len;
                     consume_value!(usize, |ptr| *ptr = len)
                 }
 
                 let elt_tag = it.clone().next().expect("truncated tag");
-                *buffer = alloc(elt_tag.size() * total_len).await;
-                recv_elements_cursor(cursor, elt_tag, total_len, *buffer, alloc).await
+                *buffer = alloc(elt_tag.size() * total_len);
+                recv_elements(reader, elt_tag, total_len, *buffer, alloc)
             })
         }
         Tag::Range(it) => {
             *data = round_up_mut(*data, tag.alignment());
             let tag = it.clone().next().expect("truncated tag");
-            recv_value_cursor(cursor, tag, data, alloc).await?;
-            recv_value_cursor(cursor, tag, data, alloc).await?;
-            recv_value_cursor(cursor, tag, data, alloc).await?;
+            recv_value(reader, tag, data, alloc)?;
+            recv_value(reader, tag, data, alloc)?;
+            recv_value(reader, tag, data, alloc)?;
             Ok(())
         }
         Tag::Keyword(_) => unreachable!(),
@@ -375,22 +175,17 @@ where
     }
 }
 
-#[cfg(has_drtio)]
-pub async fn recv_return_cursor<F>(
-    cursor: &mut Cursor<Vec<u8>>,
-    tag_bytes: &[u8],
-    data: *mut (),
-    alloc: &impl Fn(usize) -> F,
-) -> Result<(), Error>
+pub fn recv_return<F, R>(reader: &mut R, tag_bytes: &[u8], data: *mut (), alloc: &mut F) -> Result<(), Error>
 where
-    F: Future<Output = *mut ()>,
+    F: FnMut(usize) -> *mut (),
+    R: Read + ?Sized,
 {
     let mut it = TagIterator::new(tag_bytes);
     trace!("recv ...->{}", it);
 
     let tag = it.next().expect("truncated tag");
     let mut data = data;
-    unsafe { recv_value_cursor(cursor, tag, &mut data, alloc).await? };
+    unsafe { recv_value(reader, tag, &mut data, alloc)? };
 
     Ok(())
 }
@@ -538,7 +333,7 @@ where W: Write + ?Sized {
     Ok(())
 }
 
-mod tag {
+pub mod tag {
     use core::fmt;
 
     pub fn split_tag(tag_bytes: &[u8]) -> (&[u8], &[u8]) {

src/libksupport/src/rtio_acp.rs

@@ -4,7 +4,7 @@ use cslice::CSlice;
 use libcortex_a9::asm;
 use vcell::VolatileCell;
 
-use crate::{artiq_raise, pl::csr, rtio_mgt::resolve_channel_name};
+use crate::{artiq_raise, pl::csr, resolve_channel_name, rtio_core};
 
 pub const RTIO_O_STATUS_WAIT: i32 = 1;
 pub const RTIO_O_STATUS_UNDERFLOW: i32 = 2;
@@ -52,7 +52,7 @@ static mut TRANSACTION_BUFFER: Transaction = Transaction {
 
 pub extern "C" fn init() {
     unsafe {
-        csr::rtio_core::reset_write(1);
+        rtio_core::reset_write(1);
         csr::rtio::engine_addr_base_write(&TRANSACTION_BUFFER as *const Transaction as u32);
         csr::rtio::enable_write(1);
     }

src/libksupport/src/rtio_csr.rs

@@ -2,7 +2,7 @@ use core::ptr::{read_volatile, write_volatile};
 
 use cslice::CSlice;
 
-use crate::{artiq_raise, pl::csr, rtio_mgt::resolve_channel_name};
+use crate::{artiq_raise, pl::csr, resolve_channel_name, rtio_core};
 
 pub const RTIO_O_STATUS_WAIT: u8 = 1;
 pub const RTIO_O_STATUS_UNDERFLOW: u8 = 2;
@@ -20,7 +20,7 @@ pub struct TimestampedData {
 
 pub extern "C" fn init() {
     unsafe {
-        csr::rtio_core::reset_write(1);
+        rtio_core::reset_write(1);
     }
 }
 

src/runtime/Cargo.toml

@@ -18,7 +18,6 @@ num-traits = { version = "0.2", default-features = false }
 num-derive = "0.3"
 cslice = "0.3"
 log = "0.4"
-nb = "0.1"
 embedded-hal = "0.2"
 core_io = { version = "0.1", features = ["collections"] }
 byteorder = { version = "1.3", default-features = false }
@@ -26,7 +25,6 @@ void = { version = "1", default-features = false }
 futures = { version = "0.3", default-features = false, features = ["async-await"] }
 async-recursion = "0.3"
 log_buffer = { version = "1.2" }
-libm = { version = "0.2", features = ["unstable"] }
 vcell = "0.1"
 
 libboard_zynq = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git", features = ["ipv6"]}
@@ -40,5 +38,6 @@ dyld = { path = "../libdyld" }
 dwarf = { path = "../libdwarf" }
 unwind = { path = "../libunwind" }
 libc = { path = "../libc" }
-io = { path = "../libio" }
+io = { path = "../libio", features = ["alloc"] }
+ksupport = { path = "../libksupport" }
 libboard_artiq = { path = "../libboard_artiq" }

src/runtime/src/comms.rs

@@ -6,6 +6,9 @@ use cslice::CSlice;
 use futures::{future::FutureExt, select_biased};
 #[cfg(has_drtio)]
 use io::{Cursor, ProtoRead};
+#[cfg(has_drtio)]
+use ksupport::rpc;
+use ksupport::{kernel, resolve_channel_name};
 use libasync::{smoltcp::{Sockets, TcpStream},
                task};
 use libboard_artiq::drtio_routing;
@@ -27,10 +30,7 @@ use num_traits::{FromPrimitive, ToPrimitive};
 
 #[cfg(has_drtio)]
 use crate::pl;
-use crate::{analyzer, kernel, mgmt, moninj,
-            proto_async::*,
-            rpc, rtio_dma,
-            rtio_mgt::{self, resolve_channel_name}};
+use crate::{analyzer, mgmt, moninj, proto_async::*, rpc_async, rtio_dma, rtio_mgt};
 #[cfg(has_drtio)]
 use crate::{subkernel, subkernel::Error as SubkernelError};
 
@@ -207,7 +207,7 @@ async fn handle_run_kernel(
                                 kernel::Message::RpcRecvRequest(slot) => slot,
                                 other => panic!("expected root value slot from core1, not {:?}", other),
                             };
-                            rpc::recv_return(stream, &tag, slot, &|size| {
+                            rpc_async::recv_return(stream, &tag, slot, &|size| {
                                 let control = control.clone();
                                 async move {
                                     if size == 0 {
@@ -252,7 +252,7 @@ async fn handle_run_kernel(
                             let function = read_i32(stream).await? as u32;
                             control
                                 .tx
-                                .async_send(kernel::Message::RpcRecvReply(Err(kernel::RPCException {
+                                .async_send(kernel::Message::RpcRecvReply(Err(ksupport::RPCException {
                                     id,
                                     message,
                                     param,
@@ -444,6 +444,11 @@ async fn handle_run_kernel(
                         error!("error sending subkernel message: {:?}", e)
                     }
                 };
+                control
+                    .borrow_mut()
+                    .tx
+                    .async_send(kernel::Message::SubkernelMsgSent)
+                    .await;
             }
             #[cfg(has_drtio)]
             kernel::Message::SubkernelMsgRecvRequest { id, timeout } => {
@@ -471,24 +476,20 @@ async fn handle_run_kernel(
                             kernel::Message::RpcRecvRequest(slot) => slot,
                             other => panic!("expected root value slot from core1, not {:?}", other),
                         };
-                        rpc::recv_return_cursor(&mut reader, &tag, slot, &|size| {
-                            let control = control.clone();
-                            async move {
-                                if size == 0 {
-                                    0 as *mut ()
-                                } else {
-                                    let mut control = control.borrow_mut();
-                                    fast_send(&mut control.tx, kernel::Message::RpcRecvReply(Ok(size))).await;
-                                    match fast_recv(&mut control.rx).await {
-                                        kernel::Message::RpcRecvRequest(slot) => slot,
-                                        other => {
-                                            panic!("expected nested value slot from kernel CPU, not {:?}", other)
-                                        }
+                        rpc::recv_return(&mut reader, &tag, slot, &mut |size| {
+                            if size == 0 {
+                                0 as *mut ()
+                            } else {
+                                let mut control = control.borrow_mut();
+                                control.tx.send(kernel::Message::RpcRecvReply(Ok(size)));
+                                match control.rx.recv() {
+                                    kernel::Message::RpcRecvRequest(slot) => slot,
+                                    other => {
+                                        panic!("expected nested value slot from kernel CPU, not {:?}", other)
                                     }
                                 }
                             }
-                        })
-                        .await?;
+                        })?;
                         control
                             .borrow_mut()
                             .tx
@@ -619,7 +620,7 @@ async fn handle_connection(
                         }
                     }
                 }
-                #[cfg(not(has_drtio))] 
+                #[cfg(not(has_drtio))]
                 {
                     write_header(stream, Reply::LoadFailed).await?;
                     write_chunk(stream, b"No DRTIO on this system, subkernels are not supported").await?;
@@ -688,7 +689,8 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
     #[cfg(has_drtio_routing)]
     drtio_routing::interconnect_disable_all();
 
-    rtio_mgt::startup(&aux_mutex, &drtio_routing_table, &up_destinations, timer, &cfg);
+    rtio_mgt::startup(&aux_mutex, &drtio_routing_table, &up_destinations, timer);
+    ksupport::setup_device_map(&cfg);
 
     analyzer::start(&aux_mutex, &drtio_routing_table, &up_destinations, timer);
     moninj::start(timer, &aux_mutex, &drtio_routing_table);

src/runtime/src/main.rs

@@ -2,12 +2,8 @@
 #![no_main]
 #![recursion_limit = "1024"] // for futures_util::select!
 #![feature(alloc_error_handler)]
-#![feature(panic_info_message)]
-#![feature(c_variadic)]
 #![feature(const_btree_new)]
-#![feature(const_in_array_repeat_expressions)]
-#![feature(naked_functions)]
-#![feature(asm)]
+#![feature(panic_info_message)]
 
 #[macro_use]
 extern crate alloc;
@@ -15,7 +11,8 @@ extern crate alloc;
 #[cfg(feature = "target_kasli_soc")]
 use core::cell::RefCell;
 
-use libasync::{block_async, task};
+use ksupport;
+use libasync::task;
 #[cfg(feature = "target_kasli_soc")]
 use libboard_artiq::io_expander;
 use libboard_artiq::{identifier_read, logger, pl};
@@ -23,90 +20,22 @@ use libboard_zynq::{gic, mpcore, timer::GlobalTimer};
 use libconfig::Config;
 use libcortex_a9::l2c::enable_l2_cache;
 use libsupport_zynq::ram;
-use log::{error, info, warn};
-use nb;
-use void::Void;
-
-const ASYNC_ERROR_COLLISION: u8 = 1 << 0;
-const ASYNC_ERROR_BUSY: u8 = 1 << 1;
-const ASYNC_ERROR_SEQUENCE_ERROR: u8 = 1 << 2;
+use log::{info, warn};
 
 mod analyzer;
 mod comms;
-mod eh_artiq;
-mod i2c;
-mod irq;
-mod kernel;
+
 mod mgmt;
 mod moninj;
 mod panic;
 mod proto_async;
-mod rpc;
-#[cfg(ki_impl = "csr")]
-#[path = "rtio_csr.rs"]
-mod rtio;
-#[cfg(ki_impl = "acp")]
-#[path = "rtio_acp.rs"]
-mod rtio;
+mod rpc_async;
 mod rtio_clocking;
 mod rtio_dma;
 mod rtio_mgt;
 #[cfg(has_drtio)]
 mod subkernel;
 
-static mut SEEN_ASYNC_ERRORS: u8 = 0;
-
-pub unsafe fn get_async_errors() -> u8 {
-    let errors = SEEN_ASYNC_ERRORS;
-    SEEN_ASYNC_ERRORS = 0;
-    errors
-}
-
-fn wait_for_async_rtio_error() -> nb::Result<(), Void> {
-    unsafe {
-        if pl::csr::rtio_core::async_error_read() != 0 {
-            Ok(())
-        } else {
-            Err(nb::Error::WouldBlock)
-        }
-    }
-}
-
-async fn report_async_rtio_errors() {
-    loop {
-        let _ = block_async!(wait_for_async_rtio_error()).await;
-        unsafe {
-            let errors = pl::csr::rtio_core::async_error_read();
-            if errors & ASYNC_ERROR_COLLISION != 0 {
-                let channel = pl::csr::rtio_core::collision_channel_read();
-                error!(
-                    "RTIO collision involving channel 0x{:04x}:{}",
-                    channel,
-                    rtio_mgt::resolve_channel_name(channel as u32)
-                );
-            }
-            if errors & ASYNC_ERROR_BUSY != 0 {
-                let channel = pl::csr::rtio_core::busy_channel_read();
-                error!(
-                    "RTIO busy error involving channel 0x{:04x}:{}",
-                    channel,
-                    rtio_mgt::resolve_channel_name(channel as u32)
-                );
-            }
-            if errors & ASYNC_ERROR_SEQUENCE_ERROR != 0 {
-                let channel = pl::csr::rtio_core::sequence_error_channel_read();
-                error!(
-                    "RTIO sequence error involving channel 0x{:04x}:{}",
-                    channel,
-                    rtio_mgt::resolve_channel_name(channel as u32)
-                );
-            }
-            SEEN_ASYNC_ERRORS = errors;
-            pl::csr::rtio_core::async_error_write(errors);
-        }
-    }
-}
-
 #[cfg(feature = "target_kasli_soc")]
 async fn io_expanders_service(
     i2c_bus: RefCell<&mut libboard_zynq::i2c::I2c>,
@@ -145,9 +74,9 @@ pub fn main_core0() {
 
     info!("gateware ident: {}", identifier_read(&mut [0; 64]));
 
-    i2c::init();
+    ksupport::i2c::init();
     #[cfg(feature = "target_kasli_soc")]
-    let i2c_bus = unsafe { (i2c::I2C_BUS).as_mut().unwrap() };
+    let i2c_bus = unsafe { (ksupport::i2c::I2C_BUS).as_mut().unwrap() };
 
     #[cfg(feature = "target_kasli_soc")]
     let (mut io_expander0, mut io_expander1);
@@ -180,7 +109,7 @@ pub fn main_core0() {
 
     rtio_clocking::init(&mut timer, &cfg);
 
-    task::spawn(report_async_rtio_errors());
+    task::spawn(ksupport::report_async_rtio_errors());
 
     #[cfg(feature = "target_kasli_soc")]
     task::spawn(io_expanders_service(

src/runtime/src/rpc_async.rs

@@ -0,0 +1,197 @@
+use alloc::boxed::Box;
+use core::future::Future;
+
+use async_recursion::async_recursion;
+use byteorder::{ByteOrder, NativeEndian};
+use cslice::CMutSlice;
+use ksupport::rpc::{tag::{Tag, TagIterator},
+                    *};
+use libasync::smoltcp::TcpStream;
+use libboard_zynq::smoltcp;
+use log::trace;
+
+use crate::proto_async;
+
+/// Reads (deserializes) `length` array or list elements of type `tag` from `stream`,
+/// writing them into the buffer given by `storage`.
+///
+/// `alloc` is used for nested allocations (if elements themselves contain
+/// lists/arrays), see [recv_value].
+#[async_recursion(?Send)]
+async unsafe fn recv_elements<F>(
+    stream: &TcpStream,
+    elt_tag: Tag<'async_recursion>,
+    length: usize,
+    storage: *mut (),
+    alloc: &(impl Fn(usize) -> F + 'async_recursion),
+) -> Result<(), smoltcp::Error>
+where
+    F: Future<Output = *mut ()>,
+{
+    // List of simple types are special-cased in the protocol for performance.
+    match elt_tag {
+        Tag::Bool => {
+            let dest = core::slice::from_raw_parts_mut(storage as *mut u8, length);
+            proto_async::read_chunk(stream, dest).await?;
+        }
+        Tag::Int32 => {
+            let ptr = storage as *mut u32;
+            let dest = core::slice::from_raw_parts_mut(ptr as *mut u8, length * 4);
+            proto_async::read_chunk(stream, dest).await?;
+            drop(dest);
+            let dest = core::slice::from_raw_parts_mut(ptr, length);
+            NativeEndian::from_slice_u32(dest);
+        }
+        Tag::Int64 | Tag::Float64 => {
+            let ptr = storage as *mut u64;
+            let dest = core::slice::from_raw_parts_mut(ptr as *mut u8, length * 8);
+            proto_async::read_chunk(stream, dest).await?;
+            drop(dest);
+            let dest = core::slice::from_raw_parts_mut(ptr, length);
+            NativeEndian::from_slice_u64(dest);
+        }
+        _ => {
+            let mut data = storage;
+            for _ in 0..length {
+                recv_value(stream, elt_tag, &mut data, alloc).await?
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Reads (deserializes) a value of type `tag` from `stream`, writing the results to
+/// the kernel-side buffer `data` (the passed pointer to which is incremented to point
+/// past the just-received data). For nested allocations (lists/arrays), `alloc` is
+/// invoked any number of times with the size of the required allocation as a parameter
+/// (which is assumed to be correctly aligned for all payload types).
+#[async_recursion(?Send)]
+async unsafe fn recv_value<F>(
+    stream: &TcpStream,
+    tag: Tag<'async_recursion>,
+    data: &mut *mut (),
+    alloc: &(impl Fn(usize) -> F + 'async_recursion),
+) -> Result<(), smoltcp::Error>
+where
+    F: Future<Output = *mut ()>,
+{
+    macro_rules! consume_value {
+        ($ty:ty, | $ptr:ident | $map:expr) => {{
+            let $ptr = align_ptr_mut::<$ty>(*data);
+            *data = $ptr.offset(1) as *mut ();
+            $map
+        }};
+    }
+
+    match tag {
+        Tag::None => Ok(()),
+        Tag::Bool => consume_value!(i8, |ptr| {
+            *ptr = proto_async::read_i8(stream).await?;
+            Ok(())
+        }),
+        Tag::Int32 => consume_value!(i32, |ptr| {
+            *ptr = proto_async::read_i32(stream).await?;
+            Ok(())
+        }),
+        Tag::Int64 | Tag::Float64 => consume_value!(i64, |ptr| {
+            *ptr = proto_async::read_i64(stream).await?;
+            Ok(())
+        }),
+        Tag::String | Tag::Bytes | Tag::ByteArray => {
+            consume_value!(CMutSlice<u8>, |ptr| {
+                let length = proto_async::read_i32(stream).await? as usize;
+                *ptr = CMutSlice::new(alloc(length).await as *mut u8, length);
+                proto_async::read_chunk(stream, (*ptr).as_mut()).await?;
+                Ok(())
+            })
+        }
+        Tag::Tuple(it, arity) => {
+            let alignment = tag.alignment();
+            *data = round_up_mut(*data, alignment);
+            let mut it = it.clone();
+            for _ in 0..arity {
+                let tag = it.next().expect("truncated tag");
+                recv_value(stream, tag, data, alloc).await?
+            }
+            // Take into account any tail padding (if element(s) with largest alignment
+            // are not at the end).
+            *data = round_up_mut(*data, alignment);
+            Ok(())
+        }
+        Tag::List(it) => {
+            #[repr(C)]
+            struct List {
+                elements: *mut (),
+                length: usize,
+            }
+            consume_value!(*mut List, |ptr_to_list| {
+                let tag = it.clone().next().expect("truncated tag");
+                let length = proto_async::read_i32(stream).await? as usize;
+
+                // To avoid multiple kernel CPU roundtrips, use a single allocation for
+                // both the pointer/length List (slice) and the backing storage for the
+                // elements. We can assume that alloc() is aligned suitably, so just
+                // need to take into account any extra padding required.
+                // (Note: At the time of writing, there will never actually be any types
+                // with alignment larger than 8 bytes, so storage_offset == 0 always.)
+                let list_size = 4 + 4;
+                let storage_offset = round_up(list_size, tag.alignment());
+                let storage_size = tag.size() * length;
+
+                let allocation = alloc(storage_offset + storage_size).await as *mut u8;
+                *ptr_to_list = allocation as *mut List;
+                let storage = allocation.offset(storage_offset as isize) as *mut ();
+
+                (**ptr_to_list).length = length;
+                (**ptr_to_list).elements = storage;
+                recv_elements(stream, tag, length, storage, alloc).await
+            })
+        }
+        Tag::Array(it, num_dims) => {
+            consume_value!(*mut (), |buffer| {
+                // Deserialize length along each dimension and compute total number of
+                // elements.
+                let mut total_len: usize = 1;
+                for _ in 0..num_dims {
+                    let len = proto_async::read_i32(stream).await? as usize;
+                    total_len *= len;
+                    consume_value!(usize, |ptr| *ptr = len)
+                }
+
+                // Allocate backing storage for elements; deserialize them.
+                let elt_tag = it.clone().next().expect("truncated tag");
+                *buffer = alloc(elt_tag.size() * total_len).await;
+                recv_elements(stream, elt_tag, total_len, *buffer, alloc).await
+            })
+        }
+        Tag::Range(it) => {
+            *data = round_up_mut(*data, tag.alignment());
+            let tag = it.clone().next().expect("truncated tag");
+            recv_value(stream, tag, data, alloc).await?;
+            recv_value(stream, tag, data, alloc).await?;
+            recv_value(stream, tag, data, alloc).await?;
+            Ok(())
+        }
+        Tag::Keyword(_) => unreachable!(),
+        Tag::Object => unreachable!(),
+    }
+}
+
+pub async fn recv_return<F>(
+    stream: &TcpStream,
+    tag_bytes: &[u8],
+    data: *mut (),
+    alloc: &impl Fn(usize) -> F,
+) -> Result<(), smoltcp::Error>
+where
+    F: Future<Output = *mut ()>,
+{
+    let mut it = TagIterator::new(tag_bytes);
+    trace!("recv ...->{}", it);
+
+    let tag = it.next().expect("truncated tag");
+    let mut data = data;
+    unsafe { recv_value(stream, tag, &mut data, alloc).await? };
+
+    Ok(())
+}

src/runtime/src/rtio_clocking.rs

@@ -1,4 +1,6 @@
 use embedded_hal::blocking::delay::DelayMs;
+#[cfg(has_si5324)]
+use ksupport::i2c;
 use libboard_artiq::pl;
 #[cfg(has_si5324)]
 use libboard_artiq::si5324;
@@ -8,9 +10,6 @@ use libboard_zynq::timer::GlobalTimer;
 use libconfig::Config;
 use log::{info, warn};
 
-#[cfg(has_si5324)]
-use crate::i2c;
-
 #[derive(Debug, PartialEq, Copy, Clone)]
 #[allow(non_camel_case_types)]
 pub enum RtioClock {

src/runtime/src/rtio_dma.rs

@@ -2,14 +2,13 @@ use alloc::{collections::BTreeMap, rc::Rc, string::String, vec::Vec};
 #[cfg(has_drtio)]
 use core::mem;
 
+use ksupport::kernel::DmaRecorder;
 #[cfg(has_drtio)]
 use libasync::task;
 use libboard_artiq::drtio_routing::RoutingTable;
 use libboard_zynq::timer::GlobalTimer;
 use libcortex_a9::{cache::dcci_slice, mutex::Mutex};
 
-use crate::kernel::DmaRecorder;
-
 const ALIGNMENT: usize = 16 * 8;
 
 static DMA_RECORD_STORE: Mutex<BTreeMap<String, (u32, Vec<u8>, i64)>> = Mutex::new(BTreeMap::new());

src/runtime/src/rtio_mgt.rs

@@ -1,20 +1,17 @@
-use alloc::{collections::BTreeMap, rc::Rc, string::String};
+use alloc::rc::Rc;
 use core::cell::RefCell;
 
-use io::{Cursor, ProtoRead};
 use libboard_artiq::{drtio_routing, pl::csr};
 use libboard_zynq::timer::GlobalTimer;
-use libconfig::Config;
 use libcortex_a9::mutex::Mutex;
-use log::warn;
-
-static mut RTIO_DEVICE_MAP: BTreeMap<u32, String> = BTreeMap::new();
 
 #[cfg(has_drtio)]
 pub mod drtio {
     use alloc::vec::Vec;
 
     use embedded_hal::blocking::delay::DelayMs;
+    use ksupport::{resolve_channel_name, ASYNC_ERROR_BUSY, ASYNC_ERROR_COLLISION, ASYNC_ERROR_SEQUENCE_ERROR,
+                   SEEN_ASYNC_ERRORS};
     use libasync::{delay, task};
     use libboard_artiq::{drtioaux::Error, drtioaux_async, drtioaux_async::Packet,
                          drtioaux_proto::MASTER_PAYLOAD_MAX_SIZE};
@@ -22,8 +19,7 @@ pub mod drtio {
     use log::{error, info, warn};
 
     use super::*;
-    use crate::{analyzer::remote_analyzer::RemoteBuffer, rtio_dma::remote_dma, subkernel, ASYNC_ERROR_BUSY,
-                ASYNC_ERROR_COLLISION, ASYNC_ERROR_SEQUENCE_ERROR, SEEN_ASYNC_ERRORS};
+    use crate::{analyzer::remote_analyzer::RemoteBuffer, rtio_dma::remote_dma, subkernel};
 
     pub fn startup(
         aux_mutex: &Rc<Mutex<bool>>,
@@ -756,46 +752,6 @@ pub mod drtio {
     }
 }
 
-fn read_device_map(cfg: &Config) -> BTreeMap<u32, String> {
-    let mut device_map: BTreeMap<u32, String> = BTreeMap::new();
-    let _ = cfg
-        .read("device_map")
-        .and_then(|raw_bytes| {
-            let mut bytes_cr = Cursor::new(raw_bytes);
-            let size = bytes_cr.read_u32().unwrap();
-            for _ in 0..size {
-                let channel = bytes_cr.read_u32().unwrap();
-                let device_name = bytes_cr.read_string().unwrap();
-                if let Some(old_entry) = device_map.insert(channel, device_name.clone()) {
-                    warn!(
-                        "conflicting device map entries for RTIO channel {}: '{}' and '{}'",
-                        channel, old_entry, device_name
-                    );
-                }
-            }
-            Ok(())
-        })
-        .or_else(|err| {
-            warn!(
-                "error reading device map ({}), device names will not be available in RTIO error messages",
-                err
-            );
-            Err(err)
-        });
-    device_map
-}
-
-fn _resolve_channel_name(channel: u32, device_map: &BTreeMap<u32, String>) -> String {
-    match device_map.get(&channel) {
-        Some(val) => val.clone(),
-        None => String::from("unknown"),
-    }
-}
-
-pub fn resolve_channel_name(channel: u32) -> String {
-    _resolve_channel_name(channel, unsafe { &RTIO_DEVICE_MAP })
-}
-
 #[cfg(not(has_drtio))]
 pub mod drtio {
     use super::*;
@@ -817,11 +773,7 @@ pub fn startup(
     routing_table: &Rc<RefCell<drtio_routing::RoutingTable>>,
     up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
     timer: GlobalTimer,
-    cfg: &Config,
 ) {
-    unsafe {
-        RTIO_DEVICE_MAP = read_device_map(cfg);
-    }
     drtio::startup(aux_mutex, routing_table, up_destinations, timer);
     unsafe {
         csr::rtio_core::reset_phy_write(1);

src/satman/Cargo.toml

@@ -14,6 +14,8 @@ build_zynq = { path = "../libbuild_zynq" }
 
 [dependencies]
 log = { version = "0.4", default-features = false }
+core_io = { version = "0.1", features = ["collections"] }
+cslice = "0.3"
 embedded-hal = "0.2"
 
 libboard_zynq = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git", features = ["ipv6"]}
@@ -25,4 +27,6 @@ libconfig = { git = "https://git.m-labs.hk/M-Labs/zynq-rs.git", features = ["fat
 
 libboard_artiq = { path = "../libboard_artiq" }
 unwind = { path = "../libunwind" }
-libc = { path = "../libc" }
+libc = { path = "../libc" }
+io = { path = "../libio", features = ["alloc"] }
+ksupport = { path = "../libksupport" }

src/satman/src/dma.rs

@@ -170,4 +170,8 @@ impl Manager {
             }
         }
     }
+
+    pub fn running(&self) -> bool {
+        self.state == ManagerState::Playback
+    }
 }

src/satman/src/main.rs

@@ -1,13 +1,15 @@
 #![no_std]
 #![no_main]
-#![feature(never_type, panic_info_message, asm, naked_functions)]
-#![feature(alloc_error_handler)]
+#![feature(alloc_error_handler, try_trait, never_type, panic_info_message)]
 
 #[macro_use]
 extern crate log;
-
+extern crate core_io;
+extern crate cslice;
 extern crate embedded_hal;
 
+extern crate io;
+extern crate ksupport;
 extern crate libboard_artiq;
 extern crate libboard_zynq;
 extern crate libcortex_a9;
@@ -18,8 +20,6 @@ extern crate unwind;
 
 extern crate alloc;
 
-use core::sync::atomic::{AtomicBool, Ordering};
-
 use analyzer::Analyzer;
 use dma::Manager as DmaManager;
 use embedded_hal::blocking::delay::DelayUs;
@@ -27,21 +27,22 @@ use embedded_hal::blocking::delay::DelayUs;
 use libboard_artiq::io_expander;
 #[cfg(has_si5324)]
 use libboard_artiq::si5324;
-use libboard_artiq::{drtio_routing, drtioaux, drtioaux_proto::SAT_PAYLOAD_MAX_SIZE, identifier_read, logger, pl::csr};
+use libboard_artiq::{drtio_routing, drtioaux,
+                     drtioaux_proto::{MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE},
+                     identifier_read, logger,
+                     pl::csr};
 #[cfg(feature = "target_kasli_soc")]
 use libboard_zynq::error_led::ErrorLED;
-use libboard_zynq::{gic, i2c::I2c, mpcore, print, println, stdio, time::Milliseconds, timer::GlobalTimer};
-use libcortex_a9::{asm, interrupt_handler,
-                   l2c::enable_l2_cache,
-                   notify_spin_lock,
-                   regs::{MPIDR, SP},
-                   spin_lock_yield};
-use libregister::{RegisterR, RegisterW};
+use libboard_zynq::{i2c::I2c, print, println, time::Milliseconds, timer::GlobalTimer};
+use libcortex_a9::{l2c::enable_l2_cache, regs::MPIDR};
+use libregister::RegisterR;
 use libsupport_zynq::ram;
+use subkernel::Manager as KernelManager;
 
 mod analyzer;
 mod dma;
 mod repeater;
+mod subkernel;
 
 fn drtiosat_reset(reset: bool) {
     unsafe {
@@ -98,6 +99,7 @@ fn process_aux_packet(
     i2c: &mut I2c,
     dma_manager: &mut DmaManager,
     analyzer: &mut Analyzer,
+    kernel_manager: &mut KernelManager,
 ) -> Result<(), drtioaux::Error> {
     // In the code below, *_chan_sel_write takes an u8 if there are fewer than 256 channels,
     // and u16 otherwise; hence the `as _` conversion.
@@ -117,10 +119,10 @@ fn process_aux_packet(
         }
 
         drtioaux::Packet::DestinationStatusRequest {
-            destination: _destination,
+            destination,
         } => {
             #[cfg(has_drtio_routing)]
-            let hop = _routing_table.0[_destination as usize][*_rank as usize];
+            let hop = _routing_table.0[destination as usize][*_rank as usize];
             #[cfg(not(has_drtio_routing))]
             let hop = 0;
 
@@ -133,13 +135,40 @@ fn process_aux_packet(
                     drtioaux::send(
                         0,
                         &drtioaux::Packet::DmaPlaybackStatus {
-                            destination: _destination,
+                            destination: destination,
                             id: status.id,
                             error: status.error,
                             channel: status.channel,
                             timestamp: status.timestamp,
                         },
                     )?;
+                } else if let Some(subkernel_finished) = kernel_manager.get_last_finished() {
+                    info!(
+                        "subkernel {} finished, with exception: {}",
+                        subkernel_finished.id, subkernel_finished.with_exception
+                    );
+                    drtioaux::send(
+                        0,
+                        &drtioaux::Packet::SubkernelFinished {
+                            id: subkernel_finished.id,
+                            with_exception: subkernel_finished.with_exception,
+                        },
+                    )?;
+                } else if kernel_manager.message_is_ready() {
+                    let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+                    match kernel_manager.message_get_slice(&mut data_slice) {
+                        Some(meta) => drtioaux::send(
+                            0,
+                            &drtioaux::Packet::SubkernelMessage {
+                                destination: destination,
+                                id: kernel_manager.get_current_id().unwrap(),
+                                last: meta.last,
+                                length: meta.len as u16,
+                                data: data_slice,
+                            },
+                        )?,
+                        None => warn!("subkernel message is ready but no message is present"),
+                    }
                 } else {
                     let errors;
                     unsafe {
@@ -180,7 +209,7 @@ fn process_aux_packet(
                         let repno = hop - 1;
                         match _repeaters[repno].aux_forward(
                             &drtioaux::Packet::DestinationStatusRequest {
-                                destination: _destination,
+                                destination: destination,
                             },
                             timer,
                         ) {
@@ -487,10 +516,98 @@ fn process_aux_packet(
             timestamp,
         } => {
             forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
-            let succeeded = dma_manager.playback(id, timestamp).is_ok();
+            let succeeded = if !kernel_manager.running() {
+                dma_manager.playback(id, timestamp).is_ok()
+            } else {
+                false
+            };
             drtioaux::send(0, &drtioaux::Packet::DmaPlaybackReply { succeeded: succeeded })
         }
 
+        drtioaux::Packet::SubkernelAddDataRequest {
+            destination: _destination,
+            id,
+            last,
+            length,
+            data,
+        } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
+            let succeeded = kernel_manager.add(id, last, &data, length as usize).is_ok();
+            drtioaux::send(0, &drtioaux::Packet::SubkernelAddDataReply { succeeded: succeeded })
+        }
+        drtioaux::Packet::SubkernelLoadRunRequest {
+            destination: _destination,
+            id,
+            run,
+        } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
+            let mut succeeded = kernel_manager.load(id).is_ok();
+            // allow preloading a kernel with delayed run
+            if run {
+                if dma_manager.running() {
+                    // cannot run kernel while DDMA is running
+                    succeeded = false;
+                } else {
+                    succeeded |= kernel_manager.run(id).is_ok();
+                }
+            }
+            drtioaux::send(0, &drtioaux::Packet::SubkernelLoadRunReply { succeeded: succeeded })
+        }
+        drtioaux::Packet::SubkernelExceptionRequest {
+            destination: _destination,
+        } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
+            let mut data_slice: [u8; SAT_PAYLOAD_MAX_SIZE] = [0; SAT_PAYLOAD_MAX_SIZE];
+            let meta = kernel_manager.exception_get_slice(&mut data_slice);
+            drtioaux::send(
+                0,
+                &drtioaux::Packet::SubkernelException {
+                    last: meta.last,
+                    length: meta.len,
+                    data: data_slice,
+                },
+            )
+        }
+        drtioaux::Packet::SubkernelMessage {
+            destination,
+            id: _id,
+            last,
+            length,
+            data,
+        } => {
+            forward!(_routing_table, destination, *_rank, _repeaters, &packet, timer);
+            kernel_manager.message_handle_incoming(last, length as usize, &data);
+            drtioaux::send(
+                0,
+                &drtioaux::Packet::SubkernelMessageAck {
+                    destination: destination,
+                },
+            )
+        }
+        drtioaux::Packet::SubkernelMessageAck {
+            destination: _destination,
+        } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
+            if kernel_manager.message_ack_slice() {
+                let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+                if let Some(meta) = kernel_manager.message_get_slice(&mut data_slice) {
+                    drtioaux::send(
+                        0,
+                        &drtioaux::Packet::SubkernelMessage {
+                            destination: *_rank,
+                            id: kernel_manager.get_current_id().unwrap(),
+                            last: meta.last,
+                            length: meta.len as u16,
+                            data: data_slice,
+                        },
+                    )?;
+                } else {
+                    error!("Error receiving message slice");
+                }
+            }
+            Ok(())
+        }
+
         _ => {
             warn!("received unexpected aux packet");
             Ok(())
@@ -506,6 +623,7 @@ fn process_aux_packets(
     i2c: &mut I2c,
     dma_manager: &mut DmaManager,
     analyzer: &mut Analyzer,
+    kernel_manager: &mut KernelManager,
 ) {
     let result = drtioaux::recv(0).and_then(|packet| {
         if let Some(packet) = packet {
@@ -518,6 +636,7 @@ fn process_aux_packets(
                 i2c,
                 dma_manager,
                 analyzer,
+                kernel_manager,
             )
         } else {
             Ok(())
@@ -626,8 +745,8 @@ pub extern "C" fn main_core0() -> i32 {
 
     ram::init_alloc_core0();
 
-    let mut i2c = I2c::i2c0();
-    i2c.init().expect("I2C initialization failed");
+    ksupport::i2c::init();
+    let mut i2c = unsafe { (ksupport::i2c::I2C_BUS).as_mut().unwrap() };
 
     #[cfg(feature = "target_kasli_soc")]
     let (mut io_expander0, mut io_expander1);
@@ -682,6 +801,8 @@ pub extern "C" fn main_core0() -> i32 {
 
     let mut hardware_tick_ts = 0;
 
+    let mut control = ksupport::kernel::Control::start();
+
     loop {
         while !drtiosat_link_rx_up() {
             drtiosat_process_errors();
@@ -709,12 +830,12 @@ pub extern "C" fn main_core0() -> i32 {
             si5324::siphaser::calibrate_skew(&mut timer).expect("failed to calibrate skew");
         }
 
-        // DMA manager created here, so when link is dropped, all DMA traces
+        // Various managers created here, so when link is dropped, all DMA traces
         // are cleared out for a clean slate on subsequent connections,
         // without a manual intervention.
         let mut dma_manager = DmaManager::new();
-        // same for RTIO Analyzer
         let mut analyzer = Analyzer::new();
+        let mut kernel_manager = KernelManager::new(&mut control);
 
         drtioaux::reset(0);
         drtiosat_reset(false);
@@ -730,6 +851,7 @@ pub extern "C" fn main_core0() -> i32 {
                 &mut i2c,
                 &mut dma_manager,
                 &mut analyzer,
+                &mut kernel_manager,
             );
             #[allow(unused_mut)]
             for mut rep in repeaters.iter_mut() {
@@ -756,6 +878,7 @@ pub extern "C" fn main_core0() -> i32 {
                     error!("aux packet error: {:?}", e);
                 }
             }
+            kernel_manager.process_kern_requests(rank, timer);
         }
 
         drtiosat_reset_phy(true);
@@ -771,46 +894,8 @@ extern "C" {
     static mut __stack1_start: u32;
 }
 
-interrupt_handler!(IRQ, irq, __irq_stack0_start, __irq_stack1_start, {
-    if MPIDR.read().cpu_id() == 1 {
-        let mpcore = mpcore::RegisterBlock::mpcore();
-        let mut gic = gic::InterruptController::gic(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);
-            notify_spin_lock();
-            main_core1();
-        }
-        stdio::drop_uart();
-    }
-    loop {}
-});
-
 static mut PANICKED: [bool; 2] = [false; 2];
 
-static CORE1_RESTART: AtomicBool = AtomicBool::new(false);
-
-pub fn restart_core1() {
-    let mut interrupt_controller = gic::InterruptController::gic(mpcore::RegisterBlock::mpcore());
-    CORE1_RESTART.store(true, Ordering::Relaxed);
-    interrupt_controller.send_sgi(gic::InterruptId(0), gic::CPUCore::Core1.into());
-    while CORE1_RESTART.load(Ordering::Relaxed) {
-        spin_lock_yield();
-    }
-}
-
-#[no_mangle]
-pub fn main_core1() {
-    let mut interrupt_controller = gic::InterruptController::gic(mpcore::RegisterBlock::mpcore());
-    interrupt_controller.enable_interrupts();
-
-    loop {}
-}
-
 #[no_mangle]
 pub extern "C" fn exception(_vect: u32, _regs: *const u32, pc: u32, ea: u32) {
     fn hexdump(addr: u32) {
@@ -866,23 +951,3 @@ pub fn panic_fmt(info: &core::panic::PanicInfo) -> ! {
 
     loop {}
 }
-
-// linker symbols
-extern "C" {
-    static __text_start: u32;
-    static __text_end: u32;
-    static __exidx_start: u32;
-    static __exidx_end: u32;
-}
-
-#[no_mangle]
-extern "C" fn dl_unwind_find_exidx(_pc: *const u32, len_ptr: *mut u32) -> *const u32 {
-    let length;
-    let start: *const u32;
-    unsafe {
-        length = (&__exidx_end as *const u32).offset_from(&__exidx_start) as u32;
-        start = &__exidx_start;
-        *len_ptr = length;
-    }
-    start
-}

src/satman/src/subkernel.rs

@@ -0,0 +1,692 @@
+use alloc::{collections::{BTreeMap, VecDeque},
+            format,
+            string::{String, ToString},
+            vec::Vec};
+use core::{cmp::min, option::NoneError, slice, str};
+
+use core_io::{Error as IoError, Write};
+use cslice::AsCSlice;
+use io::{Cursor, ProtoRead, ProtoWrite};
+use ksupport::{eh_artiq, kernel, rpc};
+use libboard_artiq::{drtioaux_proto::{MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE},
+                     pl::csr};
+use libboard_zynq::{time::Milliseconds, timer::GlobalTimer};
+use libcortex_a9::sync_channel::Receiver;
+use log::warn;
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+enum KernelState {
+    Absent,
+    Loaded,
+    Running,
+    MsgAwait(Milliseconds),
+    MsgSending,
+}
+
+#[derive(Debug)]
+pub enum Error {
+    Load(String),
+    KernelNotFound,
+    Unexpected(String),
+    NoMessage,
+    AwaitingMessage,
+    SubkernelIoError,
+    KernelException(Sliceable),
+}
+
+impl From<NoneError> for Error {
+    fn from(_: NoneError) -> Error {
+        Error::KernelNotFound
+    }
+}
+
+impl From<IoError> for Error {
+    fn from(_value: IoError) -> Error {
+        Error::SubkernelIoError
+    }
+}
+
+impl From<()> for Error {
+    fn from(_: ()) -> Error {
+        Error::NoMessage
+    }
+}
+
+macro_rules! unexpected {
+    ($($arg:tt)*) => (return Err(Error::Unexpected(format!($($arg)*))));
+}
+
+/* represents data that has to be sent to Master */
+#[derive(Debug)]
+pub struct Sliceable {
+    it: usize,
+    data: Vec<u8>,
+}
+
+/* represents interkernel messages */
+struct Message {
+    count: u8,
+    tag: u8,
+    data: Vec<u8>,
+}
+
+#[derive(PartialEq)]
+enum OutMessageState {
+    NoMessage,
+    MessageReady,
+    MessageBeingSent,
+    MessageSent,
+    MessageAcknowledged,
+}
+
+/* for dealing with incoming and outgoing interkernel messages */
+struct MessageManager {
+    out_message: Option<Sliceable>,
+    out_state: OutMessageState,
+    in_queue: VecDeque<Message>,
+    in_buffer: Option<Message>,
+}
+
+// Per-run state
+struct Session {
+    id: u32,
+    kernel_state: KernelState,
+    last_exception: Option<Sliceable>,
+    messages: MessageManager,
+}
+
+impl Session {
+    pub fn new(id: u32) -> Session {
+        Session {
+            id: id,
+            kernel_state: KernelState::Absent,
+            last_exception: None,
+            messages: MessageManager::new(),
+        }
+    }
+
+    fn running(&self) -> bool {
+        match self.kernel_state {
+            KernelState::Absent | KernelState::Loaded => false,
+            KernelState::Running | KernelState::MsgAwait { .. } | KernelState::MsgSending => true,
+        }
+    }
+}
+
+#[derive(Debug)]
+struct KernelLibrary {
+    library: Vec<u8>,
+    complete: bool,
+}
+
+pub struct Manager<'a> {
+    kernels: BTreeMap<u32, KernelLibrary>,
+    session: Session,
+    control: &'a mut kernel::Control,
+    cache: BTreeMap<String, Vec<i32>>,
+    last_finished: Option<SubkernelFinished>,
+}
+
+pub struct SubkernelFinished {
+    pub id: u32,
+    pub with_exception: bool,
+}
+
+pub struct SliceMeta {
+    pub len: u16,
+    pub last: bool,
+}
+
+macro_rules! get_slice_fn {
+    ($name:tt, $size:expr) => {
+        pub fn $name(&mut self, data_slice: &mut [u8; $size]) -> SliceMeta {
+            if self.data.len() == 0 {
+                return SliceMeta { len: 0, last: true };
+            }
+            let len = min($size, self.data.len() - self.it);
+            let last = self.it + len == self.data.len();
+
+            data_slice[..len].clone_from_slice(&self.data[self.it..self.it + len]);
+            self.it += len;
+
+            SliceMeta {
+                len: len as u16,
+                last: last,
+            }
+        }
+    };
+}
+
+impl Sliceable {
+    pub fn new(data: Vec<u8>) -> Sliceable {
+        Sliceable { it: 0, data: data }
+    }
+
+    get_slice_fn!(get_slice_sat, SAT_PAYLOAD_MAX_SIZE);
+    get_slice_fn!(get_slice_master, MASTER_PAYLOAD_MAX_SIZE);
+}
+
+impl MessageManager {
+    pub fn new() -> MessageManager {
+        MessageManager {
+            out_message: None,
+            out_state: OutMessageState::NoMessage,
+            in_queue: VecDeque::new(),
+            in_buffer: None,
+        }
+    }
+
+    pub fn handle_incoming(&mut self, last: bool, length: usize, data: &[u8; MASTER_PAYLOAD_MAX_SIZE]) {
+        // called when receiving a message from master
+        match self.in_buffer.as_mut() {
+            Some(message) => message.data.extend(&data[..length]),
+            None => {
+                self.in_buffer = Some(Message {
+                    count: data[0],
+                    tag: data[1],
+                    data: data[2..length].to_vec(),
+                });
+            }
+        };
+        if last {
+            // when done, remove from working queue
+            self.in_queue.push_back(self.in_buffer.take().unwrap());
+        }
+    }
+
+    pub fn is_outgoing_ready(&mut self) -> bool {
+        // called by main loop, to see if there's anything to send, will send it afterwards
+        match self.out_state {
+            OutMessageState::MessageReady => {
+                self.out_state = OutMessageState::MessageBeingSent;
+                true
+            }
+            _ => false,
+        }
+    }
+
+    pub fn was_message_acknowledged(&mut self) -> bool {
+        match self.out_state {
+            OutMessageState::MessageAcknowledged => {
+                self.out_state = OutMessageState::NoMessage;
+                true
+            }
+            _ => false,
+        }
+    }
+
+    pub fn get_outgoing_slice(&mut self, data_slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE]) -> Option<SliceMeta> {
+        if self.out_state != OutMessageState::MessageBeingSent {
+            return None;
+        }
+        let meta = self.out_message.as_mut()?.get_slice_master(data_slice);
+        if meta.last {
+            // clear the message slot
+            self.out_message = None;
+            // notify kernel with a flag that message is sent
+            self.out_state = OutMessageState::MessageSent;
+        }
+        Some(meta)
+    }
+
+    pub fn ack_slice(&mut self) -> bool {
+        // returns whether or not there's more to be sent
+        match self.out_state {
+            OutMessageState::MessageBeingSent => true,
+            OutMessageState::MessageSent => {
+                self.out_state = OutMessageState::MessageAcknowledged;
+                false
+            }
+            _ => {
+                warn!("received unsolicited SubkernelMessageAck");
+                false
+            }
+        }
+    }
+
+    pub fn accept_outgoing(&mut self, message: Vec<u8>) -> Result<(), Error> {
+        // service tag skipped in kernel
+        self.out_message = Some(Sliceable::new(message));
+        self.out_state = OutMessageState::MessageReady;
+        Ok(())
+    }
+
+    pub fn get_incoming(&mut self) -> Option<Message> {
+        self.in_queue.pop_front()
+    }
+}
+
+impl<'a> Manager<'_> {
+    pub fn new(control: &mut kernel::Control) -> Manager {
+        Manager {
+            kernels: BTreeMap::new(),
+            session: Session::new(0),
+            control: control,
+            cache: BTreeMap::new(),
+            last_finished: None,
+        }
+    }
+
+    pub fn add(&mut self, id: u32, last: bool, data: &[u8], data_len: usize) -> Result<(), Error> {
+        let kernel = match self.kernels.get_mut(&id) {
+            Some(kernel) => {
+                if kernel.complete {
+                    // replace entry
+                    self.kernels.remove(&id);
+                    self.kernels.insert(
+                        id,
+                        KernelLibrary {
+                            library: Vec::new(),
+                            complete: false,
+                        },
+                    );
+                    self.kernels.get_mut(&id)?
+                } else {
+                    kernel
+                }
+            }
+            None => {
+                self.kernels.insert(
+                    id,
+                    KernelLibrary {
+                        library: Vec::new(),
+                        complete: false,
+                    },
+                );
+                self.kernels.get_mut(&id)?
+            }
+        };
+        kernel.library.extend(&data[0..data_len]);
+
+        kernel.complete = last;
+        Ok(())
+    }
+
+    pub fn running(&self) -> bool {
+        self.session.running()
+    }
+
+    pub fn get_current_id(&self) -> Option<u32> {
+        match self.running() {
+            true => Some(self.session.id),
+            false => None,
+        }
+    }
+
+    pub fn run(&mut self, id: u32) -> Result<(), Error> {
+        info!("starting subkernel #{}", id);
+        if self.session.kernel_state != KernelState::Loaded || self.session.id != id {
+            self.load(id)?;
+        }
+        self.session.kernel_state = KernelState::Running;
+        unsafe {
+            csr::cri_con::selected_write(2);
+        }
+
+        self.control.tx.send(kernel::Message::StartRequest);
+        Ok(())
+    }
+
+    pub fn message_handle_incoming(&mut self, last: bool, length: usize, slice: &[u8; MASTER_PAYLOAD_MAX_SIZE]) {
+        if !self.running() {
+            return;
+        }
+        self.session.messages.handle_incoming(last, length, slice);
+    }
+
+    pub fn message_get_slice(&mut self, slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE]) -> Option<SliceMeta> {
+        if !self.running() {
+            return None;
+        }
+        self.session.messages.get_outgoing_slice(slice)
+    }
+
+    pub fn message_ack_slice(&mut self) -> bool {
+        if !self.running() {
+            warn!("received unsolicited SubkernelMessageAck");
+            return false;
+        }
+        self.session.messages.ack_slice()
+    }
+
+    pub fn message_is_ready(&mut self) -> bool {
+        self.session.messages.is_outgoing_ready()
+    }
+
+    pub fn load(&mut self, id: u32) -> Result<(), Error> {
+        if self.session.id == id && self.session.kernel_state == KernelState::Loaded {
+            return Ok(());
+        }
+        if !self.kernels.get(&id)?.complete {
+            return Err(Error::KernelNotFound);
+        }
+        self.session = Session::new(id);
+        self.control.restart();
+
+        self.control
+            .tx
+            .send(kernel::Message::LoadRequest(self.kernels.get(&id)?.library.clone()));
+        let reply = self.control.rx.recv();
+        match reply {
+            kernel::Message::LoadCompleted => Ok(()),
+            kernel::Message::LoadFailed => Err(Error::Load("kernel load failed".to_string())),
+            _ => Err(Error::Load(format!(
+                "unexpected kernel CPU reply to load request: {:?}",
+                reply
+            ))),
+        }
+    }
+
+    pub fn exception_get_slice(&mut self, data_slice: &mut [u8; SAT_PAYLOAD_MAX_SIZE]) -> SliceMeta {
+        match self.session.last_exception.as_mut() {
+            Some(exception) => exception.get_slice_sat(data_slice),
+            None => SliceMeta { len: 0, last: true },
+        }
+    }
+
+    pub fn get_last_finished(&mut self) -> Option<SubkernelFinished> {
+        self.last_finished.take()
+    }
+
+    fn kernel_stop(&mut self) {
+        self.session.kernel_state = KernelState::Absent;
+        unsafe {
+            csr::cri_con::selected_write(0);
+        }
+    }
+
+    fn runtime_exception(&mut self, cause: Error) {
+        let raw_exception: Vec<u8> = Vec::new();
+        let mut writer = Cursor::new(raw_exception);
+        match write_exception(
+            &mut writer,
+            &[Some(eh_artiq::Exception {
+                id: 11, // SubkernelError, defined in ksupport
+                message: format!("in subkernel id {}: {:?}", self.session.id, cause).as_c_slice(),
+                param: [0, 0, 0],
+                file: file!().as_c_slice(),
+                line: line!(),
+                column: column!(),
+                function: format!("subkernel id {}", self.session.id).as_c_slice(),
+            })],
+            &[eh_artiq::StackPointerBacktrace {
+                stack_pointer: 0,
+                initial_backtrace_size: 0,
+                current_backtrace_size: 0,
+            }],
+            &[],
+            0,
+        ) {
+            Ok(_) => self.session.last_exception = Some(Sliceable::new(writer.into_inner())),
+            Err(_) => error!("Error writing exception data"),
+        }
+        self.kernel_stop();
+    }
+
+    pub fn process_kern_requests(&mut self, rank: u8, timer: GlobalTimer) {
+        if !self.running() {
+            return;
+        }
+
+        match self.process_external_messages(timer) {
+            Ok(()) => (),
+            Err(Error::AwaitingMessage) => return, // kernel still waiting, do not process kernel messages
+            Err(Error::KernelException(exception)) => {
+                self.session.last_exception = Some(exception);
+                self.last_finished = Some(SubkernelFinished {
+                    id: self.session.id,
+                    with_exception: true,
+                });
+            }
+            Err(e) => {
+                error!("Error while running processing external messages: {:?}", e);
+                self.runtime_exception(e);
+                self.last_finished = Some(SubkernelFinished {
+                    id: self.session.id,
+                    with_exception: true,
+                });
+            }
+        }
+
+        match self.process_kern_message(rank, timer) {
+            Ok(true) => {
+                self.last_finished = Some(SubkernelFinished {
+                    id: self.session.id,
+                    with_exception: false,
+                });
+            }
+            Ok(false) | Err(Error::NoMessage) => (),
+            Err(Error::KernelException(exception)) => {
+                self.session.last_exception = Some(exception);
+                self.last_finished = Some(SubkernelFinished {
+                    id: self.session.id,
+                    with_exception: true,
+                });
+            }
+            Err(e) => {
+                error!("Error while running kernel: {:?}", e);
+                self.runtime_exception(e);
+                self.last_finished = Some(SubkernelFinished {
+                    id: self.session.id,
+                    with_exception: true,
+                });
+            }
+        }
+    }
+
+    fn process_kern_message(&mut self, rank: u8, timer: GlobalTimer) -> Result<bool, Error> {
+        let reply = self.control.rx.try_recv()?;
+        match reply {
+            kernel::Message::KernelFinished(_async_errors) => {
+                self.kernel_stop();
+                return Ok(true);
+            }
+            kernel::Message::KernelException(exceptions, stack_pointers, backtrace, async_errors) => {
+                error!("exception in kernel");
+                for exception in exceptions {
+                    error!("{:?}", exception.unwrap());
+                }
+                error!("stack pointers: {:?}", stack_pointers);
+                error!("backtrace: {:?}", backtrace);
+                let buf: Vec<u8> = Vec::new();
+                let mut writer = Cursor::new(buf);
+                match write_exception(&mut writer, exceptions, stack_pointers, backtrace, async_errors) {
+                    Ok(()) => (),
+                    Err(_) => error!("Error writing exception data"),
+                }
+                self.kernel_stop();
+                return Err(Error::KernelException(Sliceable::new(writer.into_inner())));
+            }
+            kernel::Message::CachePutRequest(key, value) => {
+                self.cache.insert(key, value);
+            }
+            kernel::Message::CacheGetRequest(key) => {
+                const DEFAULT: Vec<i32> = Vec::new();
+                let value = self.cache.get(&key).unwrap_or(&DEFAULT).clone();
+                self.control.tx.send(kernel::Message::CacheGetReply(value));
+            }
+            kernel::Message::SubkernelMsgSend { id: _, data } => {
+                self.session.messages.accept_outgoing(data)?;
+                self.session.kernel_state = KernelState::MsgSending;
+            }
+            kernel::Message::SubkernelMsgRecvRequest { id: _, timeout } => {
+                let max_time = timer.get_time() + Milliseconds(timeout);
+                self.session.kernel_state = KernelState::MsgAwait(max_time);
+            }
+            kernel::Message::UpDestinationsRequest(destination) => {
+                self.control
+                    .tx
+                    .send(kernel::Message::UpDestinationsReply(destination == (rank as i32)));
+            }
+            _ => {
+                unexpected!("unexpected message from core1 while kernel was running: {:?}", reply);
+            }
+        }
+        Ok(false)
+    }
+
+    fn process_external_messages(&mut self, timer: GlobalTimer) -> Result<(), Error> {
+        match self.session.kernel_state {
+            KernelState::MsgAwait(timeout) => {
+                if timer.get_time() > timeout {
+                    self.control.tx.send(kernel::Message::SubkernelMsgRecvReply {
+                        status: kernel::SubkernelStatus::Timeout,
+                        count: 0,
+                    });
+                    self.session.kernel_state = KernelState::Running;
+                    return Ok(());
+                }
+                if let Some(message) = self.session.messages.get_incoming() {
+                    self.control.tx.send(kernel::Message::SubkernelMsgRecvReply {
+                        status: kernel::SubkernelStatus::NoError,
+                        count: message.count,
+                    });
+                    self.session.kernel_state = KernelState::Running;
+                    self.pass_message_to_kernel(&message, timer)
+                } else {
+                    Err(Error::AwaitingMessage)
+                }
+            }
+            KernelState::MsgSending => {
+                if self.session.messages.was_message_acknowledged() {
+                    self.session.kernel_state = KernelState::Running;
+                    self.control.tx.send(kernel::Message::SubkernelMsgSent);
+                    Ok(())
+                } else {
+                    Err(Error::AwaitingMessage)
+                }
+            }
+            _ => Ok(()),
+        }
+    }
+
+    fn pass_message_to_kernel(&mut self, message: &Message, timer: GlobalTimer) -> Result<(), Error> {
+        let mut reader = Cursor::new(&message.data);
+        let mut tag: [u8; 1] = [message.tag];
+        let mut i = message.count;
+        loop {
+            let slot = match recv_w_timeout(&mut self.control.rx, timer, 100)? {
+                kernel::Message::RpcRecvRequest(slot) => slot,
+                other => unexpected!("expected root value slot from core1, not {:?}", other),
+            };
+            let mut exception: Option<Sliceable> = None;
+            let mut unexpected: Option<String> = None;
+            rpc::recv_return(&mut reader, &tag, slot, &mut |size| {
+                if size == 0 {
+                    0 as *mut ()
+                } else {
+                    self.control.tx.send(kernel::Message::RpcRecvReply(Ok(size)));
+                    match recv_w_timeout(&mut self.control.rx, timer, 100) {
+                        Ok(kernel::Message::RpcRecvRequest(slot)) => slot,
+                        Ok(kernel::Message::KernelException(exceptions, stack_pointers, backtrace, async_errors)) => {
+                            let buf: Vec<u8> = Vec::new();
+                            let mut writer = Cursor::new(buf);
+                            match write_exception(&mut writer, exceptions, stack_pointers, backtrace, async_errors) {
+                                Ok(()) => {
+                                    exception = Some(Sliceable::new(writer.into_inner()));
+                                }
+                                Err(_) => {
+                                    unexpected = Some("Error writing exception data".to_string());
+                                }
+                            };
+                            0 as *mut ()
+                        }
+                        other => {
+                            unexpected = Some(format!("expected nested value slot from kernel CPU, not {:?}", other));
+                            0 as *mut ()
+                        }
+                    }
+                }
+            })?;
+            if let Some(exception) = exception {
+                self.kernel_stop();
+                return Err(Error::KernelException(exception));
+            } else if let Some(unexpected) = unexpected {
+                self.kernel_stop();
+                unexpected!("{}", unexpected);
+            }
+            self.control.tx.send(kernel::Message::RpcRecvReply(Ok(0)));
+            i -= 1;
+            if i == 0 {
+                break;
+            } else {
+                // update the tag for next read
+                tag[0] = reader.read_u8()?;
+            }
+        }
+        Ok(())
+    }
+}
+
+fn write_exception<W>(
+    writer: &mut W,
+    exceptions: &[Option<eh_artiq::Exception>],
+    stack_pointers: &[eh_artiq::StackPointerBacktrace],
+    backtrace: &[(usize, usize)],
+    async_errors: u8,
+) -> Result<(), Error>
+where
+    W: Write + ?Sized,
+{
+    /* header */
+    writer.write_bytes(&[0x5a, 0x5a, 0x5a, 0x5a, /*Reply::KernelException*/ 9])?;
+    writer.write_u32(exceptions.len() as u32)?;
+    for exception in exceptions.iter() {
+        let exception = exception.as_ref().unwrap();
+        writer.write_u32(exception.id)?;
+
+        if exception.message.len() == usize::MAX {
+            // exception with host string
+            writer.write_u32(u32::MAX)?;
+            writer.write_u32(exception.message.as_ptr() as u32)?;
+        } else {
+            let msg =
+                str::from_utf8(unsafe { slice::from_raw_parts(exception.message.as_ptr(), exception.message.len()) })
+                    .unwrap()
+                    .replace(
+                        "{rtio_channel_info:0}",
+                        &format!(
+                            "0x{:04x}:{}",
+                            exception.param[0],
+                            ksupport::resolve_channel_name(exception.param[0] as u32)
+                        ),
+                    );
+            writer.write_string(&msg)?;
+        }
+        writer.write_u64(exception.param[0] as u64)?;
+        writer.write_u64(exception.param[1] as u64)?;
+        writer.write_u64(exception.param[2] as u64)?;
+        writer.write_bytes(exception.file.as_ref())?;
+        writer.write_u32(exception.line)?;
+        writer.write_u32(exception.column)?;
+        writer.write_bytes(exception.function.as_ref())?;
+    }
+
+    for sp in stack_pointers.iter() {
+        writer.write_u32(sp.stack_pointer as u32)?;
+        writer.write_u32(sp.initial_backtrace_size as u32)?;
+        writer.write_u32(sp.current_backtrace_size as u32)?;
+    }
+    writer.write_u32(backtrace.len() as u32)?;
+    for &(addr, sp) in backtrace {
+        writer.write_u32(addr as u32)?;
+        writer.write_u32(sp as u32)?;
+    }
+    writer.write_u8(async_errors as u8)?;
+    Ok(())
+}
+
+fn recv_w_timeout(
+    rx: &mut Receiver<'_, kernel::Message>,
+    timer: GlobalTimer,
+    timeout: u64,
+) -> Result<kernel::Message, Error> {
+    let max_time = timer.get_time() + Milliseconds(timeout);
+    while timer.get_time() < max_time {
+        match rx.try_recv() {
+            Err(_) => (),
+            Ok(message) => return Ok(message),
+        }
+    }
+    Err(Error::NoMessage)
+}