satman: allow subkernels start subkernels

artiq/compiler/embedding.py

@@ -48,7 +48,7 @@ class SpecializedFunction:
 
 
 class EmbeddingMap:
-    def __init__(self):
+    def __init__(self, subkernels={}):
         self.object_current_key = 0
         self.object_forward_map = {}
         self.object_reverse_map = {}
@@ -65,6 +65,13 @@ class EmbeddingMap:
         self.str_forward_map = {}
         self.str_reverse_map = {}
         
+        # subkernels: dict of ID: function, just like object_forward_map
+        # allow the embedding map to be aware of subkernels from other kernels
+        for key, obj_ref in subkernels.items():
+            self.object_forward_map[key] = obj_ref
+            obj_id = id(obj_ref)
+            self.object_reverse_map[obj_id] = key
+
         self.preallocate_runtime_exception_names(["RuntimeError",
                                                   "RTIOUnderflow",
                                                   "RTIOOverflow",
@@ -165,6 +172,11 @@ class EmbeddingMap:
             return self.object_reverse_map[obj_id]
 
         self.object_current_key += 1
+        while self.object_forward_map.get(self.object_current_key):
+            # make sure there's no collisions with previously inserted subkernels
+            # their identifiers must be consistent between kernels/subkernels
+            self.object_current_key += 1
+        
         self.object_forward_map[self.object_current_key] = obj_ref
         self.object_reverse_map[obj_id] = self.object_current_key
         return self.object_current_key
@@ -200,10 +212,6 @@ class EmbeddingMap:
                 self.object_forward_map.values()
             ))
 
-    def has_rpc_or_subkernel(self):
-        return any(filter(lambda x: inspect.isfunction(x) or inspect.ismethod(x),
-                          self.object_forward_map.values()))
-
 
 class ASTSynthesizer:
     def __init__(self, embedding_map, value_map, quote_function=None, expanded_from=None):
@@ -794,7 +802,7 @@ class TypedtreeHasher(algorithm.Visitor):
         return hash(tuple(freeze(getattr(node, field_name)) for field_name in fields))
 
 class Stitcher:
-    def __init__(self, core, dmgr, engine=None, print_as_rpc=True, destination=0, subkernel_arg_types=[]):
+    def __init__(self, core, dmgr, engine=None, print_as_rpc=True, destination=0, subkernel_arg_types=[], subkernels={}):
         self.core = core
         self.dmgr = dmgr
         if engine is None:
@@ -816,7 +824,7 @@ class Stitcher:
 
         self.functions = {}
 
-        self.embedding_map = EmbeddingMap()
+        self.embedding_map = EmbeddingMap(subkernels)
         self.value_map = defaultdict(lambda: [])
         self.definitely_changed = False
 

artiq/compiler/transforms/artiq_ir_generator.py

@@ -2557,7 +2557,8 @@ class ARTIQIRGenerator(algorithm.Visitor):
             if types.is_method(fn):
                 fn = types.get_method_function(fn)
             sid = ir.Constant(fn.sid, builtins.TInt32())
-            return self.append(ir.Builtin("subkernel_preload", [sid], builtins.TNone()))
+            dest = ir.Constant(fn.destination, builtins.TInt32())
+            return self.append(ir.Builtin("subkernel_preload", [sid, dest], builtins.TNone()))
         elif types.is_exn_constructor(typ):
             return self.alloc_exn(node.type, *[self.visit(arg_node) for arg_node in node.args])
         elif types.is_constructor(typ):

artiq/compiler/transforms/llvm_ir_generator.py

@@ -399,9 +399,9 @@ class LLVMIRGenerator:
             llty = ll.FunctionType(lli32, [llptr])
 
         elif name == "subkernel_send_message":
-            llty = ll.FunctionType(llvoid, [lli32, lli8, llsliceptr, llptrptr])
+            llty = ll.FunctionType(llvoid, [lli32, lli1, lli8, lli8, llsliceptr, llptrptr])
         elif name == "subkernel_load_run":
-            llty = ll.FunctionType(llvoid, [lli32, lli1])
+            llty = ll.FunctionType(llvoid, [lli32, lli8, lli1])
         elif name == "subkernel_await_finish":
             llty = ll.FunctionType(llvoid, [lli32, lli64])
         elif name == "subkernel_await_message":
@@ -1417,7 +1417,8 @@ class LLVMIRGenerator:
             return self._build_rpc_recv(insn.type, llstackptr)
         elif insn.op == "subkernel_preload":
             llsid = self.map(insn.operands[0])
-            return self.llbuilder.call(self.llbuiltin("subkernel_load_run"), [llsid, ll.Constant(lli1, 0)], 
+            lldest = ll.Constant(lli8, insn.operands[1].value)
+            return self.llbuilder.call(self.llbuiltin("subkernel_load_run"), [llsid, lldest, ll.Constant(lli1, 0)], 
                                 name="subkernel.preload")
         else:
             assert False
@@ -1660,6 +1661,7 @@ class LLVMIRGenerator:
 
     def _build_subkernel_call(self, fun_loc, fun_type, args):
         llsid = ll.Constant(lli32, fun_type.sid)
+        lldest = ll.Constant(lli8, fun_type.destination)
         tag = b""
 
         for arg in args:
@@ -1678,7 +1680,7 @@ class LLVMIRGenerator:
         tag += b":"
 
         # run the kernel first
-        self.llbuilder.call(self.llbuiltin("subkernel_load_run"), [llsid, ll.Constant(lli1, 1)])
+        self.llbuilder.call(self.llbuiltin("subkernel_load_run"), [llsid, lldest, ll.Constant(lli1, 1)])
 
         # arg sent in the same vein as RPC
         llstackptr = self.llbuilder.call(self.llbuiltin("llvm.stacksave"), [],
@@ -1708,8 +1710,10 @@ class LLVMIRGenerator:
 
             llargcount = ll.Constant(lli8, len(args))
 
+            llisreturn = ll.Constant(lli1, False)
+
             self.llbuilder.call(self.llbuiltin("subkernel_send_message"),
-                                [llsid, llargcount, lltagptr, llargs])
+                                [llsid, llisreturn, lldest, llargcount, lltagptr, llargs])
             self.llbuilder.call(self.llbuiltin("llvm.stackrestore"), [llstackptr])
 
         return llsid
@@ -1746,10 +1750,12 @@ class LLVMIRGenerator:
         llretslot = self.llbuilder.bitcast(llretslot, llptr)
         self.llbuilder.store(llretslot, llrets)
 
-        llsid = ll.Constant(lli32, 0)  # return goes back to master, sid is ignored
+        llsid = ll.Constant(lli32, 0)  # return goes back to the caller, sid is ignored
         lltagcount = ll.Constant(lli8, 1)  # only one thing is returned
+        llisreturn = ll.Constant(lli1, True)  # it's a return, so destination is ignored
+        lldest = ll.Constant(lli8, 0)
         self.llbuilder.call(self.llbuiltin("subkernel_send_message"),
-                            [llsid, lltagcount, lltagptr, llrets])
+                            [llsid, llisreturn, lldest, lltagcount, lltagptr, llrets])
 
     def process_Call(self, insn):
         functiontyp = insn.target_function().type

artiq/coredevice/core.py

@@ -120,13 +120,15 @@ class Core:
 
     def compile(self, function, args, kwargs, set_result=None,
                 attribute_writeback=True, print_as_rpc=True,
-                target=None, destination=0, subkernel_arg_types=[]):
+                target=None, destination=0, subkernel_arg_types=[],
+                subkernels={}):
         try:
             engine = _DiagnosticEngine(all_errors_are_fatal=True)
 
             stitcher = Stitcher(engine=engine, core=self, dmgr=self.dmgr,
                                 print_as_rpc=print_as_rpc,
-                                destination=destination, subkernel_arg_types=subkernel_arg_types)
+                                destination=destination, subkernel_arg_types=subkernel_arg_types,
+                                subkernels=subkernels)
             stitcher.stitch_call(function, args, kwargs, set_result)
             stitcher.finalize()
 
@@ -165,7 +167,7 @@ class Core:
         self._run_compiled(kernel_library, embedding_map, symbolizer, demangler)
         return result
 
-    def compile_subkernel(self, sid, subkernel_fn, embedding_map, args, subkernel_arg_types):
+    def compile_subkernel(self, sid, subkernel_fn, embedding_map, args, subkernel_arg_types, subkernels):
         # pass self to subkernels (if applicable)
         # assuming the first argument is self
         subkernel_args = getfullargspec(subkernel_fn.artiq_embedded.function)
@@ -179,17 +181,30 @@ class Core:
         object_map, kernel_library, _, _, _ = \
             self.compile(subkernel_fn, self_arg, {}, attribute_writeback=False,
                         print_as_rpc=False, target=target, destination=destination, 
-                        subkernel_arg_types=subkernel_arg_types.get(sid, []))
+                        subkernel_arg_types=subkernel_arg_types.get(sid, []),
-        if object_map.has_rpc_or_subkernel():
+                        subkernels=subkernels)
-            raise ValueError("Subkernel must not use RPC or subkernels in other destinations")
+        if object_map.has_rpc():
-        return destination, kernel_library
+            raise ValueError("Subkernel must not use RPC")
+        return destination, kernel_library, object_map
 
     def compile_and_upload_subkernels(self, embedding_map, args, subkernel_arg_types):
-        for sid, subkernel_fn in embedding_map.subkernels().items():
+        subkernels = embedding_map.subkernels()
-            destination, kernel_library = \
+        subkernels_compiled = []
-                self.compile_subkernel(sid, subkernel_fn, embedding_map,
+        while True:
-                                       args, subkernel_arg_types)
+            new_subkernels = {}
-            self.comm.upload_subkernel(kernel_library, sid, destination)
+            for sid, subkernel_fn in subkernels.items():
+                if sid in subkernels_compiled:
+                    continue
+                destination, kernel_library, sub_embedding_map = \
+                    self.compile_subkernel(sid, subkernel_fn, embedding_map,
+                                        args, subkernel_arg_types, subkernels)
+                self.comm.upload_subkernel(kernel_library, sid, destination)
+                new_subkernels.update(sub_embedding_map.subkernels())
+                subkernels_compiled.append(sid)
+            if new_subkernels == subkernels:
+                break
+            subkernels.update(new_subkernels)
+
 
     def precompile(self, function, *args, **kwargs):
         """Precompile a kernel and return a callable that executes it on the core device

artiq/firmware/ksupport/lib.rs

@@ -460,8 +460,8 @@ extern fn dma_playback(_timestamp: i64, _ptr: i32, _uses_ddma: bool) {
 }
 
 #[unwind(allowed)]
-extern fn subkernel_load_run(id: u32, run: bool) {
+extern fn subkernel_load_run(id: u32, destination: u8, run: bool) {
-    send(&SubkernelLoadRunRequest { id: id, run: run });
+    send(&SubkernelLoadRunRequest { id: id, destination: destination, run: run });
     recv!(&SubkernelLoadRunReply { succeeded } => {
         if !succeeded {
             raise!("SubkernelError",
@@ -489,9 +489,11 @@ extern fn subkernel_await_finish(id: u32, timeout: u64) {
 }
 
 #[unwind(aborts)]
-extern fn subkernel_send_message(id: u32, count: u8, tag: &CSlice<u8>, data: *const *const ()) {
+extern fn subkernel_send_message(id: u32, is_return: bool, destination: u8, 
+    count: u8, tag: &CSlice<u8>, data: *const *const ()) {
     send(&SubkernelMsgSend { 
         id: id,
+        destination: if is_return { None } else { Some(destination) },
         count: count,
         tag: tag.as_ref(),
         data: data 

artiq/firmware/libproto_artiq/drtioaux_proto.rs

@@ -116,7 +116,7 @@ pub enum Packet {
     DmaRemoveTraceReply { destination: u8, succeeded: bool },
     DmaPlaybackRequest { source: u8, destination: u8, id: u32, timestamp: u64 },
     DmaPlaybackReply { destination: u8, succeeded: bool },
-    DmaPlaybackStatus { destination: u8, id: u32, error: u8, channel: u32, timestamp: u64 },
+    DmaPlaybackStatus { source: u8, destination: u8, id: u32, error: u8, channel: u32, timestamp: u64 },
 
     SubkernelAddDataRequest { destination: u8, id: u32, status: PayloadStatus, length: u16, data: [u8; MASTER_PAYLOAD_MAX_SIZE] },
     SubkernelAddDataReply { succeeded: bool },
@@ -322,6 +322,7 @@ impl Packet {
                 succeeded: reader.read_bool()?
             },
             0xb6 => Packet::DmaPlaybackStatus {
+                source: reader.read_u8()?,
                 destination: reader.read_u8()?,
                 id: reader.read_u32()?,
                 error: reader.read_u8()?,
@@ -617,8 +618,9 @@ impl Packet {
                 writer.write_u8(destination)?;
                 writer.write_bool(succeeded)?;
             },
-            Packet::DmaPlaybackStatus { destination, id, error, channel, timestamp } => {
+            Packet::DmaPlaybackStatus { source, destination, id, error, channel, timestamp } => {
                 writer.write_u8(0xb6)?;
+                writer.write_u8(source)?;
                 writer.write_u8(destination)?;
                 writer.write_u32(id)?;
                 writer.write_u8(error)?;

artiq/firmware/libproto_artiq/kernel_proto.rs

@@ -103,11 +103,11 @@ pub enum Message<'a> {
     SpiReadReply { succeeded: bool, data: u32 },
     SpiBasicReply { succeeded: bool },
 
-    SubkernelLoadRunRequest { id: u32, run: bool },
+    SubkernelLoadRunRequest { id: u32, destination: u8, run: bool },
     SubkernelLoadRunReply { succeeded: bool },
     SubkernelAwaitFinishRequest { id: u32, timeout: u64 },
     SubkernelAwaitFinishReply { status: SubkernelStatus },
-    SubkernelMsgSend { id: u32, count: u8, tag: &'a [u8], data: *const *const () },
+    SubkernelMsgSend { id: u32, destination: Option<u8>, count: u8, tag: &'a [u8], data: *const *const () },
     SubkernelMsgRecvRequest { id: u32, timeout: u64, tags: &'a [u8] },
     SubkernelMsgRecvReply { status: SubkernelStatus, count: u8 },
 

artiq/firmware/runtime/rtio_mgt.rs

@@ -65,7 +65,7 @@ pub mod drtio {
         let up_destinations = up_destinations.clone();
         let ddma_mutex = ddma_mutex.clone();
         let subkernel_mutex = subkernel_mutex.clone();
-        io.spawn(8192, move |io| {
+        io.spawn(10240, move |io| {
             let routing_table = routing_table.borrow();
             link_thread(io, &aux_mutex, &routing_table, &up_destinations, &ddma_mutex, &subkernel_mutex);
         });
@@ -96,17 +96,57 @@ pub mod drtio {
         }
     }
 
-    fn process_async_packets(io: &Io, ddma_mutex: &Mutex, subkernel_mutex: &Mutex, 
+    fn process_async_packets(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, subkernel_mutex: &Mutex, 
-            routing_table: &drtio_routing::RoutingTable, linkno: u8, packet: drtioaux::Packet
+            routing_table: &drtio_routing::RoutingTable, linkno: u8)
-    ) -> Option<drtioaux::Packet> {
+    {
-        // returns None if a packet has been consumed or re-routed
+        if link_has_async_ready(linkno) {
-        macro_rules! route_packet {
+            loop {
-            ($dest:ident) => {{
+                let reply = aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::RoutingRetrievePackets);
-                let dest_link = routing_table.0[$dest as usize][0] - 1; 
+                if let Ok(packet) = reply {
-                if dest_link == linkno {
+                    match packet {
-                    warn!("[LINK#{}] Re-routed packet would return to the same link, dropping: {:?}", linkno, packet);
+                        // packets to be consumed locally
-                } else if $dest == 0 {
+                        drtioaux::Packet::DmaPlaybackStatus { id, source, destination: 0, error, channel, timestamp } => {
-                    warn!("[LINK#{}] Received invalid routable packet: {:?}", linkno, packet)
+                            remote_dma::playback_done(io, ddma_mutex, id, source, error, channel, timestamp);
+                        },
+                        drtioaux::Packet::SubkernelFinished { id, destination: 0, with_exception, exception_src } => {
+                            subkernel::subkernel_finished(io, subkernel_mutex, id, with_exception, exception_src);
+                        },
+                        drtioaux::Packet::SubkernelMessage { id, source: from, destination: 0, status, length, data } => {
+                            subkernel::message_handle_incoming(io, subkernel_mutex, id, status, length as usize, &data);
+                            // acknowledge receiving part of the message
+                            drtioaux::send(linkno, 
+                                &drtioaux::Packet::SubkernelMessageAck { destination: from }
+                            ).unwrap();
+                        },
+                        // routable packets
+                        drtioaux::Packet::DmaAddTraceRequest      { destination, .. } |
+                            drtioaux::Packet::DmaAddTraceReply        { destination, .. } |
+                            drtioaux::Packet::DmaRemoveTraceRequest   { destination, .. } |
+                            drtioaux::Packet::DmaRemoveTraceReply     { destination, .. } |
+                            drtioaux::Packet::DmaPlaybackRequest      { destination, .. } |
+                            drtioaux::Packet::DmaPlaybackReply        { destination, .. } |
+                            drtioaux::Packet::SubkernelLoadRunRequest { destination, .. } |
+                            drtioaux::Packet::SubkernelLoadRunReply   { destination, .. } |
+                            drtioaux::Packet::SubkernelMessage        { destination, .. } |
+                            drtioaux::Packet::SubkernelMessageAck     { destination, .. } |
+                            drtioaux::Packet::DmaPlaybackStatus       { destination, .. } |
+                            drtioaux::Packet::SubkernelFinished       { destination, .. } => {
+                            let dest_link = routing_table.0[destination as usize][0] - 1; 
+                            if dest_link == linkno {
+                                warn!("[LINK#{}] Re-routed packet would return to the same link, dropping: {:?}", linkno, packet);
+                            } else if destination == 0 {
+                                warn!("[LINK#{}] Received invalid routable packet: {:?}", linkno, packet)
+                            } else {
+                                drtioaux::send(dest_link, &packet).unwrap();
+                            }
+                        }
+
+                        drtioaux::Packet::RoutingNoPackets => break,
+
+                        other => warn!("[LINK#{}] Received an unroutable packet: {:?}", linkno, other)
+                    }
+                } else {
+                    warn!("[LINK#{}] Error handling async packets ({})", linkno, reply.unwrap_err());
                 }
                 else {
                     drtioaux::send(dest_link, &packet).unwrap();

artiq/firmware/runtime/session.rs

@@ -631,6 +631,8 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
                 unsafe { kernel::stop() }
                 session.kernel_state = KernelState::Absent;
                 unsafe { session.congress.cache.unborrow() }
+                #[cfg(has_drtio)]
+                subkernel::clear_subkernels(io, _subkernel_mutex)?;
 
                 match stream {
                     None => return Ok(true),
@@ -668,7 +670,7 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
                 }
             }
             #[cfg(has_drtio)]
-            &kern::SubkernelLoadRunRequest { id, run } => {
+            &kern::SubkernelLoadRunRequest { id, destination: _, run } => {
                 let succeeded = match subkernel::load(
                     io, aux_mutex, _subkernel_mutex, routing_table, id, run) {
                         Ok(()) => true,
@@ -699,7 +701,7 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
                 kern_send(io, &kern::SubkernelAwaitFinishReply { status: status })
             }
             #[cfg(has_drtio)]
-            &kern::SubkernelMsgSend { id, count, tag, data } => {
+            &kern::SubkernelMsgSend { id, destination: _, count, tag, data } => {
                 subkernel::message_send(io, aux_mutex, _subkernel_mutex, routing_table, id, count, tag, data)?;
                 kern_acknowledge()
             }

artiq/firmware/satman/kernel.rs

@@ -2,7 +2,7 @@ use core::{mem, option::NoneError, cmp::min};
 use alloc::{string::String, format, vec::Vec, collections::{btree_map::BTreeMap, vec_deque::VecDeque}};
 use cslice::AsCSlice;
 
-use board_artiq::{mailbox, spi};
+use board_artiq::{drtioaux, drtio_routing::RoutingTable, mailbox, spi};
 use board_misoc::{csr, clock, i2c};
 use proto_artiq::{
     drtioaux_proto::PayloadStatus,
@@ -15,6 +15,7 @@ use kernel::eh_artiq::StackPointerBacktrace;
 
 use ::{cricon_select, RtioMaster};
 use cache::Cache;
+use routing::Router;
 use SAT_PAYLOAD_MAX_SIZE;
 use MASTER_PAYLOAD_MAX_SIZE;
 
@@ -62,7 +63,9 @@ enum KernelState {
     Loaded,
     Running,
     MsgAwait { max_time: u64, tags: Vec<u8> },
-    MsgSending
+    MsgSending,
+    SubkernelAwaitLoad,
+    SubkernelAwaitFinish { max_time: u64, id: u32 }
 }
 
 #[derive(Debug)]
@@ -74,6 +77,7 @@ pub enum Error {
     NoMessage,
     AwaitingMessage,
     SubkernelIoError,
+    DrtioError,
     KernelException(Sliceable)
 }
 
@@ -89,6 +93,12 @@ impl From<io::Error<!>> for Error {
     }
 }
 
+impl From<drtioaux::Error<!>> for Error {
+    fn from(_value: drtioaux::Error<!>) -> Error {
+        Error::DrtioError
+    }
+}
+
 macro_rules! unexpected {
     ($($arg:tt)*) => (return Err(Error::Unexpected(format!($($arg)*))));
 }
@@ -97,7 +107,8 @@ macro_rules! unexpected {
 #[derive(Debug)]
 pub struct Sliceable {
     it: usize,
-    data: Vec<u8>
+    data: Vec<u8>,
+    destination: u8
 }
 
 /* represents interkernel messages */
@@ -109,7 +120,6 @@ struct Message {
 #[derive(PartialEq)]
 enum OutMessageState {
     NoMessage,
-    MessageReady,
     MessageBeingSent,
     MessageSent,
     MessageAcknowledged
@@ -129,7 +139,8 @@ struct Session {
     log_buffer: String,
     last_exception: Option<Sliceable>,
     source: u8, // which destination requested running the kernel
-    messages: MessageManager
+    messages: MessageManager,
+    subkernels_finished: VecDeque<(u32, bool, u8)> // tuple of id, with_exception, exception_source
 }
 
 #[derive(Debug)]
@@ -154,6 +165,7 @@ pub struct SubkernelFinished {
 }
 
 pub struct SliceMeta {
+    pub destination: u8,
     pub len: u16,
     pub status: PayloadStatus
 }
@@ -169,6 +181,7 @@ macro_rules! get_slice_fn {
             self.it += len;
     
             SliceMeta {
+                destination: self.destination,
                 len: len as u16,
                 status: status
             }
@@ -177,10 +190,11 @@ macro_rules! get_slice_fn {
 }
 
 impl Sliceable {
-    pub fn new(data: Vec<u8>) -> Sliceable {
+    pub fn new(destination: u8, data: Vec<u8>) -> Sliceable {
         Sliceable {
             it: 0,
-            data: data
+            data: data,
+            destination: destination
         }
     }
 
@@ -219,17 +233,6 @@ impl MessageManager {
         }
     }
 
-    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 => {
@@ -269,14 +272,27 @@ impl MessageManager {
         }
     }
 
-    pub fn accept_outgoing(&mut self, count: u8, tag: &[u8], data: *const *const ()) -> Result<(), Error>  {
+    pub fn accept_outgoing(&mut self, id: u32, self_destination: u8, destination: u8, 
+        count: u8, tag: &[u8], data: *const *const (), 
+        routing_table: &RoutingTable, rank: u8, router: &mut Router
+    ) -> Result<(), Error>  {
         let mut writer = Cursor::new(Vec::new());
         rpc::send_args(&mut writer, 0, tag, data, false)?;
         // skip service tag, but write the count
         let mut data = writer.into_inner().split_off(3);
         data[0] = count;
-        self.out_message = Some(Sliceable::new(data));
+        self.out_message = Some(Sliceable::new(destination, data));
-        self.out_state = OutMessageState::MessageReady;
+
+        let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+        self.out_state = OutMessageState::MessageBeingSent;
+        let meta = self.get_outgoing_slice(&mut data_slice).unwrap();
+        let res = router.route(drtioaux::Packet::SubkernelMessage {
+                source: self_destination, destination: destination, id: id,
+                status: meta.status, length: meta.len as u16, data: data_slice
+        }, routing_table, rank, self_destination);
+        if let Err(e) = res {
+            warn!("error sending SubkernelMessage: {}", e);
+        }
         Ok(())
     }
 
@@ -292,7 +308,8 @@ impl Session {
             log_buffer: String::new(),
             last_exception: None,
             source: 0,
-            messages: MessageManager::new()
+            messages: MessageManager::new(),
+            subkernels_finished: VecDeque::new()
         }
     }
 
@@ -300,7 +317,8 @@ impl Session {
         match self.kernel_state {
             KernelState::Absent  | KernelState::Loaded  => false,
             KernelState::Running | KernelState::MsgAwait { .. } |
-                KernelState::MsgSending => true
+                KernelState::MsgSending | KernelState::SubkernelAwaitLoad |
+                KernelState::SubkernelAwaitFinish { .. } => true
         }
     }
 
@@ -408,14 +426,6 @@ impl Manager {
         self.session.messages.ack_slice()
     }
 
-    pub fn message_is_ready(&mut self) -> bool {
-        self.session.messages.is_outgoing_ready()
-    }
-
-    pub fn get_last_finished(&mut self) -> Option<SubkernelFinished> {
-        self.last_finished.take()
-    }
-
     pub fn load(&mut self, id: u32) -> Result<(), Error> {
         if self.current_id == id && self.session.kernel_state == KernelState::Loaded {
             return Ok(())
@@ -439,6 +449,7 @@ impl Manager {
                     }
                     kern::LoadReply(Err(error)) => {
                         kernel_cpu::stop();
+                        error!("load error: {:?}", error);
                         Err(Error::Load(format!("{}", error)))
                     }
                     other => {
@@ -452,7 +463,7 @@ impl Manager {
     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, status: PayloadStatus::FirstAndLast }
+            None => SliceMeta { destination: 0, len: 0, status: PayloadStatus::FirstAndLast }
         }
     }
 
@@ -477,12 +488,18 @@ impl Manager {
             backtrace: &[],
             async_errors: 0
         }).write_to(&mut writer) {
-            Ok(_) => self.session.last_exception = Some(Sliceable::new(writer.into_inner())),
+            Ok(_) => self.session.last_exception = Some(Sliceable::new(0, writer.into_inner())),
             Err(_) => error!("Error writing exception data")
         }
     }
 
-    pub fn process_kern_requests(&mut self, destination: u8) {
+    pub fn process_kern_requests(&mut self, router: &mut Router, routing_table: &RoutingTable, rank: u8, destination: u8) {
+        macro_rules! finished {
+            ($with_exception:expr) => {{ Some(SubkernelFinished { 
+                source: self.session.source, id: self.current_id, 
+                with_exception: $with_exception, exception_source: destination 
+            }) }}
+        }
         if !self.is_running() {
             return;
         }
@@ -495,34 +512,37 @@ impl Manager {
                 self.session.kernel_state = KernelState::Absent;
                 unsafe { self.cache.unborrow() }
                 self.session.last_exception = Some(exception);
-                self.last_finished = Some(SubkernelFinished { 
+                self.last_finished = finished!(true);
-                    source: self.session.source, id: self.current_id, with_exception: true, exception_source: destination 
-                })
             },
             Err(e) => { 
                 error!("Error while running processing external messages: {:?}", e);
                 self.stop();
                 self.runtime_exception(e);
-                self.last_finished = Some(SubkernelFinished { 
+                self.last_finished = finished!(true);
-                    source: self.session.source, id: self.current_id, with_exception: true, exception_source: destination 
-                })
              }
         }
 
-        match self.process_kern_message(destination) {
+        match self.process_kern_message(router, routing_table, rank, destination) {
             Ok(Some(with_exception)) => {
-                self.last_finished = Some(SubkernelFinished { 
+                self.last_finished = finished!(with_exception)
-                    source: self.session.source, id: self.current_id, with_exception: with_exception, exception_source: destination 
-                })
             },
             Ok(None) | Err(Error::NoMessage) => (),
             Err(e) => { 
                 error!("Error while running kernel: {:?}", e); 
                 self.stop(); 
                 self.runtime_exception(e);
-                self.last_finished = Some(SubkernelFinished { 
+                self.last_finished = finished!(true);
-                    source: self.session.source, id: self.current_id, with_exception: true, exception_source: destination 
+            }
-                })
+        }
+
+        if let Some(subkernel_finished) = self.last_finished.take() {
+            info!("subkernel {} finished, with exception: {}", subkernel_finished.id, subkernel_finished.with_exception);
+            let res = router.route(drtioaux::Packet::SubkernelFinished {
+                destination: subkernel_finished.source, id: subkernel_finished.id, 
+                with_exception: subkernel_finished.with_exception, exception_src: destination
+            }, &routing_table, rank, destination);
+            if let Err(e) = res {
+                warn!("error sending SubkernelFinished: {}", e);
             }
         }
     }
@@ -552,11 +572,64 @@ impl Manager {
                     Err(Error::AwaitingMessage)
                 }
             },
+            KernelState::SubkernelAwaitFinish { max_time, id } => {
+                if clock::get_ms() > *max_time {
+                    kern_send(&kern::SubkernelAwaitFinishReply { status: kern::SubkernelStatus::Timeout })?;
+                    self.session.kernel_state = KernelState::Running;
+                } else {
+                    let mut i = 0;
+                    for status in self.session.subkernels_finished.iter() {
+                        if status.0 == *id {
+                            break;
+                        }
+                        i += 1;
+                    }
+                    if let Some(finish_status) = self.session.subkernels_finished.remove(i) {
+                        if finish_status.1 {
+                            unsafe { kernel_cpu::stop() }
+                            self.session.kernel_state = KernelState::Absent;
+                            unsafe { self.cache.unborrow() }
+                            self.last_finished = Some(SubkernelFinished { 
+                                source: self.session.source, id: self.current_id,
+                                with_exception: true, exception_source: finish_status.2
+                            })
+                        } else {
+                            kern_send(&kern::SubkernelAwaitFinishReply { status: kern::SubkernelStatus::NoError })?;
+                            self.session.kernel_state = KernelState::Running;
+                        }
+                    }
+                }
+                Ok(())
+            }
             _ => Ok(())
         }
     }
 
-    fn process_kern_message(&mut self, destination: u8) -> Result<Option<bool>, Error> {
+    pub fn subkernel_load_run_reply(&mut self, succeeded: bool, self_destination: u8) {
+        if self.session.kernel_state == KernelState::SubkernelAwaitLoad {
+            if let Err(e) = kern_send(&kern::SubkernelLoadRunReply { succeeded: succeeded }) {
+                self.stop(); 
+                self.runtime_exception(e);
+                self.last_finished = Some(SubkernelFinished { 
+                    source: self.session.source, id: self.current_id, 
+                    with_exception: true, exception_source: self_destination 
+                })
+            } else {
+                self.session.kernel_state = KernelState::Running;
+            }
+        } else {
+            warn!("received unsolicited SubkernelLoadRunReply");
+        }
+    }
+
+    pub fn remote_subkernel_finished(&mut self, id: u32, with_exception: bool, exception_source: u8) {
+        self.session.subkernels_finished.push_back((id, with_exception, exception_source));
+    }
+
+    fn process_kern_message(&mut self, router: &mut Router, 
+        routing_table: &RoutingTable,
+        rank: u8, destination: u8
+    ) -> Result<Option<bool>, Error> {
         // returns Ok(with_exception) on finish
         // None if the kernel is still running
         kern_recv(|request| {
@@ -626,8 +699,14 @@ impl Manager {
                     return Ok(Some(true))
                 }
 
-                &kern::SubkernelMsgSend { id: _, count, tag, data } => {
+                &kern::SubkernelMsgSend { id: _, destination: msg_dest, count, tag, data } => {
-                    self.session.messages.accept_outgoing(count, tag, data)?;
+                    let dest = match msg_dest {
+                        Some(dest) => dest,
+                        None => self.session.source
+                    };
+                    self.session.messages.accept_outgoing(self.current_id, destination,
+                        dest, count, tag, data, 
+                        routing_table, rank, router)?;
                     // acknowledge after the message is sent
                     self.session.kernel_state = KernelState::MsgSending;
                     Ok(())
@@ -639,6 +718,20 @@ impl Manager {
                     Ok(())
                 },
 
+                &kern::SubkernelLoadRunRequest { id, destination: sk_destination, run } => {
+                    self.session.kernel_state = KernelState::SubkernelAwaitLoad;
+                    router.route(drtioaux::Packet::SubkernelLoadRunRequest { 
+                        source: destination, destination: sk_destination, id: id, run: run 
+                    }, routing_table, rank, destination).map_err(|_| Error::DrtioError)?;
+                    kern_acknowledge()
+                }
+
+                &kern::SubkernelAwaitFinishRequest{ id, timeout } => {
+                    let max_time = clock::get_ms() + timeout as u64;
+                    self.session.kernel_state = KernelState::SubkernelAwaitFinish { max_time: max_time, id: id };
+                    kern_acknowledge()
+                }
+
                 request => unexpected!("unexpected request {:?} from kernel CPU", request)
             }.and(Ok(None))
         })
@@ -712,7 +805,7 @@ fn slice_kernel_exception(exceptions: &[Option<eh_artiq::Exception>],
         async_errors: 0
     }).write_to(&mut writer) {
         // save last exception data to be received by master
-        Ok(_) => Ok(Sliceable::new(writer.into_inner())),
+        Ok(_) => Ok(Sliceable::new(0, writer.into_inner())),
         Err(_) => Err(Error::SubkernelIoError)
     }
 }

artiq/firmware/satman/main.rs

@@ -132,46 +132,46 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
             let hop = 0;
 
             if hop == 0 {
+                *self_destination = destination;
                 // async messages
                 if *rank == 1 {
                     if let Some(packet) = router.get_upstream_packet(*rank) {
                         // pass any async or routed packets to master
                         // this does mean that DDMA/SK packets to master will "trickle down" to higher rank
-                        drtioaux::send(0, &packet)?;
+                        return drtioaux::send(0, &packet)
                     }
-                } else {
+                }
-                    let errors;
+                let errors;
+                unsafe {
+                    errors = csr::drtiosat::rtio_error_read();
+                }
+                if errors & 1 != 0 {
+                    let channel;
                     unsafe {
-                        errors = csr::drtiosat::rtio_error_read();
+                        channel = csr::drtiosat::sequence_error_channel_read();
+                        csr::drtiosat::rtio_error_write(1);
                     }
-                    if errors & 1 != 0 {
+                    drtioaux::send(0,
-                        let channel;
+                        &drtioaux::Packet::DestinationSequenceErrorReply { channel })?;
-                        unsafe {
+                } else if errors & 2 != 0 {
-                            channel = csr::drtiosat::sequence_error_channel_read();
+                    let channel;
-                            csr::drtiosat::rtio_error_write(1);
+                    unsafe {
-                        }
+                        channel = csr::drtiosat::collision_channel_read();
-                        drtioaux::send(0,
+                        csr::drtiosat::rtio_error_write(2);
-                            &drtioaux::Packet::DestinationSequenceErrorReply { channel })?;
-                    } else if errors & 2 != 0 {
-                        let channel;
-                        unsafe {
-                            channel = csr::drtiosat::collision_channel_read();
-                            csr::drtiosat::rtio_error_write(2);
-                        }
-                        drtioaux::send(0,
-                            &drtioaux::Packet::DestinationCollisionReply { channel })?;
-                    } else if errors & 4 != 0 {
-                        let channel;
-                        unsafe {
-                            channel = csr::drtiosat::busy_channel_read();
-                            csr::drtiosat::rtio_error_write(4);
-                        }
-                        drtioaux::send(0,
-                            &drtioaux::Packet::DestinationBusyReply { channel })?;
                     }
-                    else {
+                    drtioaux::send(0,
-                        drtioaux::send(0, &drtioaux::Packet::DestinationOkReply)?;
+                        &drtioaux::Packet::DestinationCollisionReply { channel })?;
+                } else if errors & 4 != 0 {
+                    let channel;
+                    unsafe {
+                        channel = csr::drtiosat::busy_channel_read();
+                        csr::drtiosat::rtio_error_write(4);
                     }
+                    drtioaux::send(0,
+                        &drtioaux::Packet::DestinationBusyReply { channel })?;
+                }
+                else {
+                    drtioaux::send(0, &drtioaux::Packet::DestinationOkReply)?;
                 }
             }
 
@@ -196,7 +196,6 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
                     }
                 }
             }
-
             Ok(())
         }
 
@@ -378,14 +377,14 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
             let succeeded = dmamgr.add(source, id, status, &trace, length as usize).is_ok();
             router.send(drtioaux::Packet::DmaAddTraceReply { 
                 destination: source, succeeded: succeeded 
-            }, _routing_table, *rank)
+            }, _routing_table, *rank, *self_destination)
         }
         drtioaux::Packet::DmaRemoveTraceRequest { source, destination: _destination, id } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet);
             let succeeded = dmamgr.erase(source, id).is_ok();
             router.send(drtioaux::Packet::DmaRemoveTraceReply { 
                 destination: source, succeeded: succeeded 
-            }, _routing_table, *rank)
+            }, _routing_table, *rank, *self_destination)
         }
         drtioaux::Packet::DmaPlaybackRequest { source, destination: _destination, id, timestamp } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet);
@@ -393,7 +392,7 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
             let succeeded = !kernelmgr.is_running() && dmamgr.playback(source, id, timestamp).is_ok();
             router.send(drtioaux::Packet::DmaPlaybackReply { 
                 destination: source, succeeded: succeeded
-            }, _routing_table, *rank)
+            }, _routing_table, *rank, *self_destination)
         }
 
         drtioaux::Packet::SubkernelAddDataRequest { destination, id, status, length, data } => {
@@ -418,7 +417,19 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
             router.send(drtioaux::Packet::SubkernelLoadRunReply { 
                     destination: source, succeeded: succeeded 
                 }, 
-            _routing_table, *rank)
+            _routing_table, *rank, *self_destination)
+        }
+        drtioaux::Packet::SubkernelLoadRunReply { destination: _destination, succeeded } => {
+            forward!(_routing_table, _destination, *rank, _repeaters, &packet);
+            // received if local subkernel started another, remote subkernel
+            kernelmgr.subkernel_load_run_reply(succeeded, *self_destination);
+            Ok(())
+        }
+        // { destination: u8, id: u32, with_exception: bool, exception_src: u8 },
+        drtioaux::Packet::SubkernelFinished { destination: _destination, id, with_exception, exception_src } => {
+            forward!(_routing_table, _destination, *rank, _repeaters, &packet);
+            kernelmgr.remote_subkernel_finished(id, with_exception, exception_src);
+            Ok(())
         }
         drtioaux::Packet::SubkernelExceptionRequest { destination: _destination } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet);
@@ -435,7 +446,7 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
             kernelmgr.message_handle_incoming(status, length as usize, &data);
             router.send(drtioaux::Packet::SubkernelMessageAck {
                     destination: source
-                }, _routing_table, *rank)
+                }, _routing_table, *rank, *self_destination)
         }
         drtioaux::Packet::SubkernelMessageAck { destination: _destination } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet);
@@ -443,10 +454,9 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg
                 let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
                 if let Some(meta) = kernelmgr.message_get_slice(&mut data_slice) {
                     router.send(drtioaux::Packet::SubkernelMessage {
-                        source: *self_destination, destination: 0, id: kernelmgr.get_current_id().unwrap(),
+                        source: *self_destination, destination: meta.destination, id: kernelmgr.get_current_id().unwrap(),
                         status: meta.status, length: meta.len as u16, data: data_slice
-                    }, 
+                    }, _routing_table, *rank, *self_destination)?;
-                    _routing_table, *rank)?;
                 } else {
                     error!("Error receiving message slice");
                 }
@@ -694,7 +704,7 @@ pub extern fn main() -> i32 {
         while !drtiosat_link_rx_up() {
             drtiosat_process_errors();
             for rep in repeaters.iter_mut() {
-                rep.service(&routing_table, rank, &mut router);
+                rep.service(&routing_table, rank, destination, &mut router);
             }
             #[cfg(all(soc_platform = "kasli", hw_rev = "v2.0"))]
             {
@@ -731,7 +741,7 @@ pub extern fn main() -> i32 {
                 &mut kernelmgr, &mut repeaters, &mut routing_table,
                 &mut rank, &mut router, &mut destination);
             for rep in repeaters.iter_mut() {
-                rep.service(&routing_table, rank, &mut router);
+                rep.service(&routing_table, rank, destination, &mut router);
             }
             #[cfg(all(soc_platform = "kasli", hw_rev = "v2.0"))]
             {
@@ -754,34 +764,21 @@ pub extern fn main() -> i32 {
             }
             if let Some(status) = dma_manager.get_status() {
                 info!("playback done, error: {}, channel: {}, timestamp: {}", status.error, status.channel, status.timestamp);
-                let res = router.route(drtioaux::Packet::DmaPlaybackStatus { 
+                router.route(drtioaux::Packet::DmaPlaybackStatus { 
-                    destination: status.source, id: status.id, error: status.error, 
+                    source: destination, destination: status.source, id: status.id, 
-                    channel: status.channel, timestamp: status.timestamp 
+                    error: status.error, channel: status.channel, timestamp: status.timestamp 
-                }, &routing_table, rank);
+                }, &routing_table, rank, destination);
-                if let Err(e) = res {
-                    warn!("error sending DmaPlaybackStatus: {}", e);
-                }
             }
-            kernelmgr.process_kern_requests(destination);
+
-            if let Some(subkernel_finished) = kernelmgr.get_last_finished() {
+            kernelmgr.process_kern_requests(&mut router, &routing_table, rank, destination);
-                info!("subkernel {} finished, with exception: {}", subkernel_finished.id, subkernel_finished.with_exception);
+
-                let res = router.route(drtioaux::Packet::SubkernelFinished {
+            if rank > 1 {
-                    destination: subkernel_finished.source, id: subkernel_finished.id, 
+                if let Some(packet) = router.get_upstream_packet(rank) {
-                    with_exception: subkernel_finished.with_exception, exception_src: destination
+                    // in sat-sat communications, it can be async
-                }, &routing_table, rank);
+                    let res = drtioaux::send(0, &packet);
-                if let Err(e) = res {
+                    if let Err(e) = res {
-                    warn!("error sending SubkernelFinished: {}", e);
+                        warn!("error routing packet: {}", e);
-                }
+                    }
-            }
-            if kernelmgr.message_is_ready() {
-                let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
-                let meta = kernelmgr.message_get_slice(&mut data_slice).unwrap();
-                let res = router.route(drtioaux::Packet::SubkernelMessage {
-                    source: destination, destination: 0, id: kernelmgr.get_current_id().unwrap(),
-                    status: meta.status, length: meta.len as u16, data: data_slice
-                }, &routing_table, rank);
-                if let Err(e) = res {
-                    warn!("error sending SubkernelMessage: {}", e);
                 }
             }
         }

artiq/firmware/satman/repeater.rs

@@ -49,7 +49,7 @@ impl Repeater {
         self.state == RepeaterState::Up
     }
 
-    pub fn service(&mut self, routing_table: &drtio_routing::RoutingTable, rank: u8, router: &mut Router) {
+    pub fn service(&mut self, routing_table: &drtio_routing::RoutingTable, rank: u8, destination: u8, router: &mut Router) {
         self.process_local_errors();
 
         match self.state {
@@ -107,7 +107,7 @@ impl Repeater {
                 }
             }
             RepeaterState::Up => {
-                self.process_unsolicited_aux(routing_table, rank, router);
+                self.process_unsolicited_aux(routing_table, rank, destination, router);
                 if !rep_link_rx_up(self.repno) {
                     info!("[REP#{}] link is down", self.repno);
                     self.state = RepeaterState::Down;
@@ -122,7 +122,7 @@ impl Repeater {
         }
     }
 
-    fn process_unsolicited_aux(&self, routing_table: &drtio_routing::RoutingTable, rank: u8, router: &mut Router) {
+    fn process_unsolicited_aux(&self, routing_table: &drtio_routing::RoutingTable, rank: u8, self_destination: u8, router: &mut Router) {
         match drtioaux::recv(self.auxno) {
             Ok(Some(packet)) => {
                 let destination = get_routable_packet_destination(&packet);
@@ -130,7 +130,7 @@ impl Repeater {
                     warn!("[REP#{}] unsolicited aux packet: {:?}", self.repno, packet);
                 } else {
                     // routable packet
-                    let res = router.route(packet, routing_table, rank);
+                    let res = router.route(packet, routing_table, rank, self_destination);
                     match res {
                         Ok(()) => drtioaux::send(self.auxno, &drtioaux::Packet::RoutingAck).unwrap(),
                         Err(e) => warn!("[REP#{}] Error routing packet: {:?}", self.repno, e),
@@ -212,9 +212,8 @@ impl Repeater {
             (csr::DRTIOREP[repno].set_time_write)(1);
             while (csr::DRTIOREP[repno].set_time_read)() == 1 {}
         }
-
+        // TSCAck is sent spontaneously by the satellite,
-        // TSCAck is the only aux packet that is sent spontaneously
+        // in response to a TSC set on the RT link.
-        // by the satellite, in response to a TSC set on the RT link.
         let reply = self.recv_aux_timeout(10000)?;
         if reply == drtioaux::Packet::TSCAck {
             return Ok(());
@@ -288,7 +287,7 @@ pub struct Repeater {
 impl Repeater {
     pub fn new(_repno: u8) -> Repeater { Repeater::default() }
 
-    pub fn service(&self, _routing_table: &drtio_routing::RoutingTable, _rank: u8) { }
+    pub fn service(&self, _routing_table: &drtio_routing::RoutingTable, _rank: u8, _destination: u8, _router: &mut Router) { }
 
     pub fn sync_tsc(&self) -> Result<(), drtioaux::Error<!>> { Ok(()) }
 

artiq/firmware/satman/routing.rs

@@ -1,6 +1,5 @@
 use alloc::collections::vec_deque::VecDeque;
 use board_artiq::{drtioaux, drtio_routing};
-use board_misoc::csr;
 
 // Packets from downstream (further satellites) are received and routed appropriately.
 // they're passed immediately if it's possible (within the subtree), or sent upstream.
@@ -51,7 +50,8 @@ impl Router {
     // called by local sources (DDMA, kernel) and by repeaters on receiving unsolicited data
     // messages are always buffered for upstream, or passed downstream directly
     pub fn route(&mut self, packet: drtioaux::Packet,
-        _routing_table: &drtio_routing::RoutingTable, _rank: u8
+        _routing_table: &drtio_routing::RoutingTable, _rank: u8,
+        _self_destination: u8
     ) -> Result<(), drtioaux::Error<!>>  {
         #[cfg(has_drtio_routing)]
         {
@@ -59,14 +59,12 @@ impl Router {
             if let Some(destination) = destination {
                 let hop = _routing_table.0[destination as usize][_rank as usize];
                 let auxno = if destination == 0 { 0 } else { hop };
-                if hop != 0 {
+                if destination == _self_destination {
-                    if hop as usize <= csr::DRTIOREP.len() {
-                        drtioaux::send(auxno, &packet)?;
-                    } else {
-                        self.out_messages.push_back(packet);
-                    }
-                } else {
                     self.local_messages.push_back(packet);
+                } else if _rank > 1 {
+                    drtioaux::send(auxno, &packet)?;
+                } else {
+                    self.out_messages.push_back(packet);
                 }
             } else {
                 return Err(drtioaux::Error::RoutingError);
@@ -81,7 +79,7 @@ impl Router {
 
     // Sends a packet to a required destination, routing if it's necessary
     pub fn send(&mut self, packet: drtioaux::Packet,
-        _routing_table: &drtio_routing::RoutingTable, _rank: u8) -> Result<(), drtioaux::Error<!>> {
+        _routing_table: &drtio_routing::RoutingTable, _rank: u8, _destination: u8) -> Result<(), drtioaux::Error<!>> {
         #[cfg(has_drtio_routing)]
         {
             let destination = get_routable_packet_destination(&packet);
@@ -89,7 +87,7 @@ impl Router {
                 // send upstream directly (response to master)
                 drtioaux::send(0, &packet)
             } else {
-                self.route(packet, _routing_table, _rank)
+                self.route(packet, _routing_table, _rank, _destination)
             }
         }
         #[cfg(not(has_drtio_routing))]

artiq/frontend/artiq_compile.py

@@ -67,12 +67,21 @@ def main():
                 core.compile(exp.run, [exp_inst], {},
                              attribute_writeback=False, print_as_rpc=False)
 
-            subkernels = {}
+            subkernels = object_map.subkernels()
-            for sid, subkernel_fn in object_map.subkernels().items():
+            compiled_subkernels = {}
-                destination, subkernel_library = core.compile_subkernel(
+            while True:
-                    sid, subkernel_fn, object_map, 
+                new_subkernels = {}
-                    [exp_inst], subkernel_arg_types)
+                for sid, subkernel_fn in subkernels.items():
-                subkernels[sid] = (destination, subkernel_library)
+                    if sid in compiled_subkernels.keys():
+                        continue
+                    destination, subkernel_library, embedding_map = core.compile_subkernel(
+                        sid, subkernel_fn, object_map, 
+                        [exp_inst], subkernel_arg_types, subkernels)
+                    compiled_subkernels[sid] = (destination, subkernel_library)
+                    new_subkernels.update(embedding_map.subkernels())
+                if new_subkernels == subkernels:
+                    break
+                subkernels.update(new_subkernels)
         except CompileError as error:
             return
         finally:
@@ -107,7 +116,7 @@ def main():
             tar.addfile(main_kernel_info, fileobj=main_kernel_fileobj)
 
             # subkernels as "<sid> <destination>.elf"
-            for sid, (destination, subkernel_library) in subkernels.items():
+            for sid, (destination, subkernel_library) in compiled_subkernels.items():
                 subkernel_fileobj = io.BytesIO(subkernel_library)
                 subkernel_info = tarfile.TarInfo(name="{} {}.elf".format(sid, destination))
                 subkernel_info.size = len(subkernel_library)

artiq/test/lit/embedding/subkernel_no_arg.py

@@ -6,13 +6,13 @@ from artiq.language.types import *
 
 @kernel
 def entrypoint():
-    # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+    # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
     # CHECK-NOT: call void @subkernel_send_message\(.*\), !dbg !.
     no_arg()
 
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-NOT-L: declare void @subkernel_send_message(i32, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-NOT-L: declare void @subkernel_send_message(i32, i1, i8, { i8*, i32 }*, i8**) local_unnamed_addr
 @subkernel(destination=1)
 def no_arg() -> TStr:
     pass

artiq/test/lit/embedding/subkernel_return.py

@@ -6,15 +6,15 @@ from artiq.language.types import *
 
 @kernel
 def entrypoint():
-    # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+    # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
     # CHECK-NOT: call void @subkernel_send_message\(.*\), !dbg !.
     returning()
     # CHECK: call i8 @subkernel_await_message\(i32 1, i64 10000, { i8\*, i32 }\* nonnull .*, i8 1, i8 1\), !dbg !.
     # CHECK: call void @subkernel_await_finish\(i32 1, i64 10000\), !dbg !.
     subkernel_await(returning)
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-NOT-L: declare void @subkernel_send_message(i32, i8, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-NOT-L: declare void @subkernel_send_message(i32, i1, i8, i8, { i8*, i32 }*, i8**) local_unnamed_addr
 # CHECK-L: declare i8 @subkernel_await_message(i32, i64, { i8*, i32 }*, i8, i8) local_unnamed_addr
 # CHECK-L: declare void @subkernel_await_finish(i32, i64) local_unnamed_addr
 @subkernel(destination=1)

artiq/test/lit/embedding/subkernel_return_none.py

@@ -6,15 +6,15 @@ from artiq.language.types import *
 
 @kernel
 def entrypoint():
-    # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+    # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
     # CHECK-NOT: call void @subkernel_send_message\(.*\), !dbg !.
     returning_none()
     # CHECK: call void @subkernel_await_finish\(i32 1, i64 10000\), !dbg !.
     # CHECK-NOT: call i8 @subkernel_await_message\(i32 1, i64 10000\, .*\), !dbg !.
     subkernel_await(returning_none)
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-NOT-L: declare void @subkernel_send_message(i32, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-NOT-L: declare void @subkernel_send_message(i32, i1, i8, { i8*, i32 }*, i8**) local_unnamed_addr
 # CHECK-L: declare void @subkernel_await_finish(i32, i64) local_unnamed_addr
 # CHECK-NOT-L: declare i8 @subkernel_await_message(i32, i64, { i8*, i32 }*, i8, i8) local_unnamed_addr
 @subkernel(destination=1)

artiq/test/lit/embedding/subkernel_self.py

@@ -11,7 +11,7 @@ class A:
 
     @kernel
     def kernel_entrypoint(self):
-        # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+        # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
         # CHECK-NOT: call void @subkernel_send_message\(.*\), !dbg !.
         self.sk()
 
@@ -21,5 +21,5 @@ a = A()
 def entrypoint():
     a.kernel_entrypoint()
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-NOT-L: declare void @subkernel_send_message(i32, i8, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-NOT-L: declare void @subkernel_send_message(i32, i1, i8, i8, { i8*, i32 }*, i8**) local_unnamed_addr

artiq/test/lit/embedding/subkernel_self_args.py

@@ -11,8 +11,8 @@ class A:
 
     @kernel
     def kernel_entrypoint(self):
-        # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+        # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
-        # CHECK: call void @subkernel_send_message\(i32 1, i8 1, .*\), !dbg !.
+        # CHECK: call void @subkernel_send_message\(i32 1, i1 false, i8 1, i8 1, .*\), !dbg !.
         self.sk(1)
 
 a = A()
@@ -21,5 +21,5 @@ a = A()
 def entrypoint():
     a.kernel_entrypoint()
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-L: declare void @subkernel_send_message(i32, i8, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-L: declare void @subkernel_send_message(i32, i1, i8, i8, { i8*, i32 }*, i8**) local_unnamed_addr

artiq/test/lit/embedding/subkernel_with_arg.py

@@ -6,13 +6,13 @@ from artiq.language.types import *
 
 @kernel
 def entrypoint():
-    # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+    # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
-    # CHECK: call void @subkernel_send_message\(i32 ., i8 1, .*\), !dbg !.
+    # CHECK: call void @subkernel_send_message\(i32 ., i1 false, i8 1, i8 1, .*\), !dbg !.
     accept_arg(1)
 
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-L: declare void @subkernel_send_message(i32, i8, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-L: declare void @subkernel_send_message(i32, i1, i8, i8, { i8*, i32 }*, i8**) local_unnamed_addr
 @subkernel(destination=1)
 def accept_arg(arg: TInt32) -> TNone:
     pass

artiq/test/lit/embedding/subkernel_with_opt_arg.py

@@ -6,16 +6,16 @@ from artiq.language.types import *
 
 @kernel
 def entrypoint():
-    # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+    # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
-    # CHECK: call void @subkernel_send_message\(i32 ., i8 1, .*\), !dbg !.
+    # CHECK: call void @subkernel_send_message\(i32 ., i1 false, i8 1, i8 1, .*\), !dbg !.
     accept_arg(1)
-    # CHECK: call void @subkernel_load_run\(i32 1, i1 true\), !dbg !.
+    # CHECK: call void @subkernel_load_run\(i32 1, i8 1, i1 true\), !dbg !.
-    # CHECK: call void @subkernel_send_message\(i32 ., i8 2, .*\), !dbg !.
+    # CHECK: call void @subkernel_send_message\(i32 ., i1 false, i8 1, i8 2, .*\), !dbg !.
     accept_arg(1, 2)
 
 
-# CHECK-L: declare void @subkernel_load_run(i32, i1) local_unnamed_addr
+# CHECK-L: declare void @subkernel_load_run(i32, i8, i1) local_unnamed_addr
-# CHECK-L: declare void @subkernel_send_message(i32, i8, { i8*, i32 }*, i8**) local_unnamed_addr
+# CHECK-L: declare void @subkernel_send_message(i32, i1, i8, i8, { i8*, i32 }*, i8**) local_unnamed_addr
 @subkernel(destination=1)
 def accept_arg(arg_a, arg_b=5) -> TNone:
     pass