satman: support sub-subkernels, routing

src/libboard_artiq/src/drtioaux_proto.rs

@@ -926,19 +926,19 @@ impl Packet {
     pub fn routable_destination(&self) -> Option<u8> {
         // only for packets that could be re-routed, not only forwarded
         match self {
-            Packet::DmaAddTraceRequest      { destination, .. } => Some(*destination),
-            Packet::DmaAddTraceReply        { destination, .. } => Some(*destination),
-            Packet::DmaRemoveTraceRequest   { destination, .. } => Some(*destination),
-            Packet::DmaRemoveTraceReply     { destination, .. } => Some(*destination),
-            Packet::DmaPlaybackRequest      { destination, .. } => Some(*destination),
-            Packet::DmaPlaybackReply        { destination, .. } => Some(*destination),
+            Packet::DmaAddTraceRequest { destination, .. } => Some(*destination),
+            Packet::DmaAddTraceReply { destination, .. } => Some(*destination),
+            Packet::DmaRemoveTraceRequest { destination, .. } => Some(*destination),
+            Packet::DmaRemoveTraceReply { destination, .. } => Some(*destination),
+            Packet::DmaPlaybackRequest { destination, .. } => Some(*destination),
+            Packet::DmaPlaybackReply { destination, .. } => Some(*destination),
             Packet::SubkernelLoadRunRequest { destination, .. } => Some(*destination),
-            Packet::SubkernelLoadRunReply   { destination, .. } => Some(*destination),
-            Packet::SubkernelMessage        { destination, .. } => Some(*destination),
-            Packet::SubkernelMessageAck     { destination, .. } => Some(*destination),
-            Packet::DmaPlaybackStatus       { destination, .. } => Some(*destination),
-            Packet::SubkernelFinished       { destination, .. } => Some(*destination),
-            _ => None
+            Packet::SubkernelLoadRunReply { destination, .. } => Some(*destination),
+            Packet::SubkernelMessage { destination, .. } => Some(*destination),
+            Packet::SubkernelMessageAck { destination, .. } => Some(*destination),
+            Packet::DmaPlaybackStatus { destination, .. } => Some(*destination),
+            Packet::SubkernelFinished { destination, .. } => Some(*destination),
+            _ => None,
         }
     }
 
@@ -947,11 +947,14 @@ impl Packet {
         // e.g. reply, ACK packets end a conversation,
         // and firmware should not wait for response
         match self {
-            Packet::DmaAddTraceReply { .. } | Packet::DmaRemoveTraceReply { .. } |
-                Packet::DmaPlaybackReply { .. } | Packet::SubkernelLoadRunReply { .. } |
-                Packet::SubkernelMessageAck { .. } | Packet::DmaPlaybackStatus { .. } |
-                Packet::SubkernelFinished { .. } => false,
-            _ => true
+            Packet::DmaAddTraceReply { .. }
+            | Packet::DmaRemoveTraceReply { .. }
+            | Packet::DmaPlaybackReply { .. }
+            | Packet::SubkernelLoadRunReply { .. }
+            | Packet::SubkernelMessageAck { .. }
+            | Packet::DmaPlaybackStatus { .. }
+            | Packet::SubkernelFinished { .. } => false,
+            _ => true,
         }
     }
 }

src/runtime/src/rtio_mgt.rs

@@ -53,7 +53,12 @@ pub mod drtio {
         async_ready
     }
 
-    async fn process_async_packets(aux_mutex: &Mutex<bool>, linkno: u8, routing_table: &drtio_routing::RoutingTable, timer: GlobalTimer) {
+    async fn process_async_packets(
+        aux_mutex: &Mutex<bool>,
+        linkno: u8,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+    ) {
         if link_has_async_ready(linkno).await {
             loop {
                 let reply = aux_transact(aux_mutex, linkno, &Packet::RoutingRetrievePackets, timer).await;

src/satman/src/dma.rs

@@ -53,7 +53,14 @@ impl Manager {
         }
     }
 
-    pub fn add(&mut self, source: u8, id: u32, status: PayloadStatus, trace: &[u8], trace_len: usize) -> Result<(), Error> {
+    pub fn add(
+        &mut self,
+        source: u8,
+        id: u32,
+        status: PayloadStatus,
+        trace: &[u8],
+        trace_len: usize,
+    ) -> Result<(), Error> {
         let entry = match self.entries.get_mut(&(source, id)) {
             Some(entry) => {
                 if entry.complete || status.is_first() {

src/satman/src/main.rs

@@ -39,13 +39,13 @@ use libboard_zynq::{i2c::I2c, print, println, time::Milliseconds, timer::GlobalT
 use libcortex_a9::{l2c::enable_l2_cache, regs::MPIDR};
 use libregister::RegisterR;
 use libsupport_zynq::ram;
-use subkernel::Manager as KernelManager;
 use routing::Router;
+use subkernel::Manager as KernelManager;
 
-mod routing;
 mod analyzer;
 mod dma;
 mod repeater;
+mod routing;
 mod subkernel;
 
 fn drtiosat_reset(reset: bool) {
@@ -141,35 +141,35 @@ fn process_aux_packet(
             let hop = 0;
 
             if hop == 0 {
-                    *self_destination = destination;
-                    let errors;
+                *self_destination = destination;
+                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 {
-                        let channel;
-                        unsafe {
-                            channel = csr::drtiosat::sequence_error_channel_read();
-                            csr::drtiosat::rtio_error_write(1);
-                        }
-                        drtioaux::send(0, &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::Packet::DestinationOkReply)?;
+                    drtioaux::send(0, &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::Packet::DestinationOkReply)?;
+                }
             }
 
             #[cfg(has_drtio_routing)]
@@ -241,7 +241,7 @@ fn process_aux_packet(
         drtioaux::Packet::RoutingRetrievePackets => {
             let packet = match router.get_upstream_packet() {
                 Some(packet) => packet,
-                None => drtioaux::Packet::RoutingNoPackets
+                None => drtioaux::Packet::RoutingNoPackets,
             };
             drtioaux::send(0, &packet)
         }
@@ -481,9 +481,16 @@ fn process_aux_packet(
             forward!(_routing_table, destination, *rank, _repeaters, &packet, timer);
             *self_destination = destination;
             let succeeded = dma_manager.add(source, id, status, &trace, length as usize).is_ok();
-            router.send(drtioaux::Packet::DmaAddTraceReply { 
-                destination: source, succeeded: succeeded 
-            }, _routing_table, *rank, *self_destination, _repeaters)
+            router.send(
+                drtioaux::Packet::DmaAddTraceReply {
+                    destination: source,
+                    succeeded: succeeded,
+                },
+                _routing_table,
+                *rank,
+                *self_destination,
+                _repeaters,
+            )
         }
         drtioaux::Packet::DmaRemoveTraceRequest {
             source,
@@ -492,9 +499,16 @@ fn process_aux_packet(
         } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
             let succeeded = dma_manager.erase(source, id).is_ok();
-            router.send(drtioaux::Packet::DmaRemoveTraceReply { 
-                destination: source, succeeded: succeeded 
-            }, _routing_table, *rank, *self_destination, _repeaters)
+            router.send(
+                drtioaux::Packet::DmaRemoveTraceReply {
+                    destination: source,
+                    succeeded: succeeded,
+                },
+                _routing_table,
+                *rank,
+                *self_destination,
+                _repeaters,
+            )
         }
         drtioaux::Packet::DmaPlaybackRequest {
             source,
@@ -508,9 +522,16 @@ fn process_aux_packet(
             } else {
                 false
             };
-            router.send(drtioaux::Packet::DmaPlaybackReply { 
-                destination: source, succeeded: succeeded
-            }, _routing_table, *rank, *self_destination, _repeaters)
+            router.send(
+                drtioaux::Packet::DmaPlaybackReply {
+                    destination: source,
+                    succeeded: succeeded,
+                },
+                _routing_table,
+                *rank,
+                *self_destination,
+                _repeaters,
+            )
         }
 
         drtioaux::Packet::SubkernelAddDataRequest {
@@ -542,19 +563,33 @@ fn process_aux_packet(
                     succeeded |= kernel_manager.run(source, id).is_ok();
                 }
             }
-            router.send(drtioaux::Packet::SubkernelLoadRunReply { 
-                destination: source, succeeded: succeeded 
-            }, 
-            _routing_table, *rank, *self_destination, _repeaters)
+            router.send(
+                drtioaux::Packet::SubkernelLoadRunReply {
+                    destination: source,
+                    succeeded: succeeded,
+                },
+                _routing_table,
+                *rank,
+                *self_destination,
+                _repeaters,
+            )
         }
-        drtioaux::Packet::SubkernelLoadRunReply { destination: _destination, succeeded } => {
+        drtioaux::Packet::SubkernelLoadRunReply {
+            destination: _destination,
+            succeeded,
+        } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
             // received if local subkernel started another, remote subkernel
             kernel_manager.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 } => {
+        drtioaux::Packet::SubkernelFinished {
+            destination: _destination,
+            id,
+            with_exception,
+            exception_src,
+        } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
             kernel_manager.remote_subkernel_finished(id, with_exception, exception_src);
             Ok(())
@@ -584,12 +619,7 @@ fn process_aux_packet(
         } => {
             forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
             kernel_manager.message_handle_incoming(status, length as usize, &data);
-            drtioaux::send(
-                0,
-                &drtioaux::Packet::SubkernelMessageAck {
-                    destination: source,
-                },
-            )
+            drtioaux::send(0, &drtioaux::Packet::SubkernelMessageAck { destination: source })
         }
         drtioaux::Packet::SubkernelMessageAck {
             destination: _destination,
@@ -598,10 +628,20 @@ fn process_aux_packet(
             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) {
-        // route and not send immediately as ACKs are not a beginning of a transaction
-                router.route(drtioaux::Packet::SubkernelMessage {
-                    source: *self_destination, destination: meta.destination, id: kernel_manager.get_current_id().unwrap(),
-                    status: meta.status, length: meta.len as u16, data: data_slice}, _routing_table, *rank, *self_destination);
+                    // route and not send immediately as ACKs are not a beginning of a transaction
+                    router.route(
+                        drtioaux::Packet::SubkernelMessage {
+                            source: *self_destination,
+                            destination: meta.destination,
+                            id: kernel_manager.get_current_id().unwrap(),
+                            status: meta.status,
+                            length: meta.len as u16,
+                            data: data_slice,
+                        },
+                        _routing_table,
+                        *rank,
+                        *self_destination,
+                    );
                 } else {
                     error!("Error receiving message slice");
                 }
@@ -641,7 +681,7 @@ fn process_aux_packets(
                 dma_manager,
                 analyzer,
                 kernel_manager,
-                router
+                router,
             )
         } else {
             Ok(())
@@ -811,7 +851,6 @@ pub extern "C" fn main_core0() -> i32 {
     let mut control = ksupport::kernel::Control::start();
 
     loop {
-
         let mut router = Router::new();
 
         while !drtiosat_link_rx_up() {
@@ -903,7 +942,11 @@ pub extern "C" fn main_core0() -> i32 {
                         error: status.error,
                         channel: status.channel,
                         timestamp: status.timestamp,
-                    },  &routing_table, rank, destination);
+                    },
+                    &routing_table,
+                    rank,
+                    destination,
+                );
             }
 
             kernel_manager.process_kern_requests(&mut router, &routing_table, rank, destination, &timer);

src/satman/src/repeater.rs

@@ -6,7 +6,6 @@ use libboard_artiq::{drtio_routing, drtioaux};
 #[cfg(has_drtio_routing)]
 use libboard_zynq::time::Milliseconds;
 use libboard_zynq::timer::GlobalTimer;
-
 use routing::Router;
 
 #[cfg(has_drtio_routing)]
@@ -55,7 +54,14 @@ impl Repeater {
         self.state == RepeaterState::Up
     }
 
-    pub fn service(&mut self, routing_table: &drtio_routing::RoutingTable, rank: u8, destination: u8, router: &mut Router, timer: &mut GlobalTimer) {
+    pub fn service(
+        &mut self,
+        routing_table: &drtio_routing::RoutingTable,
+        rank: u8,
+        destination: u8,
+        router: &mut Router,
+        timer: &mut GlobalTimer,
+    ) {
         self.process_local_errors();
 
         match self.state {
@@ -182,21 +188,27 @@ impl Repeater {
 
     fn async_messages_ready(&self) -> bool {
         let async_rdy;
-        unsafe { 
+        unsafe {
             async_rdy = (csr::DRTIOREP[self.repno as usize].async_messages_ready_read)();
             (csr::DRTIOREP[self.repno as usize].async_messages_ready_write)(0);
         }
         async_rdy == 1
     }
 
-    fn handle_async(&self, routing_table: &drtio_routing::RoutingTable, rank: u8, self_destination: u8, router: &mut Router, timer: &mut GlobalTimer
+    fn handle_async(
+        &self,
+        routing_table: &drtio_routing::RoutingTable,
+        rank: u8,
+        self_destination: u8,
+        router: &mut Router,
+        timer: &mut GlobalTimer,
     ) -> Result<(), drtioaux::Error> {
         loop {
             drtioaux::send(self.auxno, &drtioaux::Packet::RoutingRetrievePackets).unwrap();
             let reply = self.recv_aux_timeout(200, timer)?;
             match reply {
                 drtioaux::Packet::RoutingNoPackets => break,
-                packet => router.route(packet, routing_table, rank, self_destination)
+                packet => router.route(packet, routing_table, rank, self_destination),
             }
         }
         Ok(())

src/satman/src/routing.rs

@@ -1,6 +1,6 @@
 use alloc::collections::vec_deque::VecDeque;
-use libboard_artiq::{drtioaux, drtio_routing, pl::csr};
 
+use libboard_artiq::{drtio_routing, drtioaux, pl::csr};
 use repeater::Repeater;
 
 // Packets from downstream (further satellites) are received and routed appropriately.
@@ -13,7 +13,7 @@ use repeater::Repeater;
 // from both master and satellite, e.g. DDMA and Subkernel.
 
 pub struct Router {
-    upstream_queue: VecDeque<drtioaux::Packet>, 
+    upstream_queue: VecDeque<drtioaux::Packet>,
     local_queue: VecDeque<drtioaux::Packet>,
     downstream_queue: VecDeque<(usize, drtioaux::Packet)>,
     upstream_notified: bool,
@@ -31,9 +31,12 @@ impl Router {
 
     // called by local sources (DDMA, kernel) and by repeaters on receiving async data
     // messages are always buffered for both upstream and downstream
-    pub fn route(&mut self, packet: drtioaux::Packet,
-        _routing_table: &drtio_routing::RoutingTable, _rank: u8,
-        _self_destination: u8
+    pub fn route(
+        &mut self,
+        packet: drtioaux::Packet,
+        _routing_table: &drtio_routing::RoutingTable,
+        _rank: u8,
+        _self_destination: u8,
     ) {
         #[cfg(has_drtio_routing)]
         {
@@ -59,9 +62,13 @@ 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,
-        _destination: u8, _repeaters: &[Repeater]
+    pub fn send(
+        &mut self,
+        packet: drtioaux::Packet,
+        _routing_table: &drtio_routing::RoutingTable,
+        _rank: u8,
+        _destination: u8,
+        _repeaters: &[Repeater],
     ) -> Result<(), drtioaux::Error> {
         #[cfg(has_drtio_routing)]
         {
@@ -116,4 +123,4 @@ impl Router {
     pub fn get_local_packet(&mut self) -> Option<drtioaux::Packet> {
         self.local_queue.pop_front()
     }
-}
+}

src/satman/src/subkernel.rs

@@ -8,12 +8,13 @@ use core_io::{Error as IoError, Write};
 use cslice::AsCSlice;
 use io::{Cursor, ProtoWrite};
 use ksupport::{eh_artiq, kernel, rpc};
-use libboard_artiq::{drtioaux, drtio_routing::RoutingTable, drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE},
+use libboard_artiq::{drtio_routing::RoutingTable,
+                     drtioaux,
+                     drtioaux_proto::{PayloadStatus, 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;
-
 use routing::Router;
 
 #[derive(Debug, Clone, PartialEq)]
@@ -104,8 +105,8 @@ struct Session {
     kernel_state: KernelState,
     last_exception: Option<Sliceable>,
     messages: MessageManager,
-    source: u8, // which destination requested running the kernel
-    subkernels_finished: VecDeque<(u32, bool, u8)> // tuple of id, with_exception, exception_source
+    source: u8,                                     // which destination requested running the kernel
+    subkernels_finished: VecDeque<(u32, bool, u8)>, // tuple of id, with_exception, exception_source
 }
 
 impl Session {
@@ -116,15 +117,18 @@ impl Session {
             last_exception: None,
             messages: MessageManager::new(),
             source: 0,
-            subkernels_finished: VecDeque::new()
+            subkernels_finished: VecDeque::new(),
         }
     }
 
     fn running(&self) -> bool {
         match self.kernel_state {
             KernelState::Absent | KernelState::Loaded => false,
-            KernelState::Running | KernelState::MsgAwait { .. } | KernelState::MsgSending | KernelState::SubkernelAwaitLoad |
-            KernelState::SubkernelAwaitFinish { .. } => true
+            KernelState::Running
+            | KernelState::MsgAwait { .. }
+            | KernelState::MsgSending
+            | KernelState::SubkernelAwaitLoad
+            | KernelState::SubkernelAwaitFinish { .. } => true,
         }
     }
 }
@@ -178,7 +182,11 @@ macro_rules! get_slice_fn {
 
 impl Sliceable {
     pub fn new(destination: u8, data: Vec<u8>) -> Sliceable {
-        Sliceable { it: 0, data: data, destination: destination }
+        Sliceable {
+            it: 0,
+            data: data,
+            destination: destination,
+        }
     }
 
     get_slice_fn!(get_slice_sat, SAT_PAYLOAD_MAX_SIZE);
@@ -254,18 +262,34 @@ impl MessageManager {
         }
     }
 
-    pub fn accept_outgoing(&mut self, id: u32, self_destination: u8, destination: u8, 
-        message: Vec<u8>, routing_table: &RoutingTable, rank: u8, router: &mut Router
-    ) -> Result<(), Error>  {
+    pub fn accept_outgoing(
+        &mut self,
+        id: u32,
+        self_destination: u8,
+        destination: u8,
+        message: Vec<u8>,
+        routing_table: &RoutingTable,
+        rank: u8,
+        router: &mut Router,
+    ) -> Result<(), Error> {
         self.out_message = Some(Sliceable::new(destination, message));
 
         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();
-        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);
+        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,
+        );
         Ok(())
     }
 
@@ -443,7 +467,14 @@ impl<'a> Manager<'_> {
         self.kernel_stop();
     }
 
-    pub fn process_kern_requests(&mut self, router: &mut Router, routing_table: &RoutingTable, rank: u8, destination: u8, timer: &GlobalTimer) {
+    pub fn process_kern_requests(
+        &mut self,
+        router: &mut Router,
+        routing_table: &RoutingTable,
+        rank: u8,
+        destination: u8,
+        timer: &GlobalTimer,
+    ) {
         if !self.running() {
             return;
         }
@@ -504,11 +535,21 @@ impl<'a> Manager<'_> {
         }
 
         if let Some(subkernel_finished) = self.last_finished.take() {
-            info!("subkernel {} finished, with exception: {}", subkernel_finished.id, subkernel_finished.with_exception);
-            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);
+            info!(
+                "subkernel {} finished, with exception: {}",
+                subkernel_finished.id, subkernel_finished.with_exception
+            );
+            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,
+            );
         }
     }
 
@@ -521,11 +562,19 @@ impl<'a> Manager<'_> {
     }
 
     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));
+        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, self_destination: u8, timer: &GlobalTimer) -> Result<bool, Error> {
+    fn process_kern_message(
+        &mut self,
+        router: &mut Router,
+        routing_table: &RoutingTable,
+        rank: u8,
+        self_destination: u8,
+        timer: &GlobalTimer,
+    ) -> Result<bool, Error> {
         let reply = self.control.rx.try_recv()?;
         match reply {
             kernel::Message::KernelFinished(_async_errors) => {
@@ -556,30 +605,57 @@ impl<'a> Manager<'_> {
                 let value = self.cache.get(&key).unwrap_or(&DEFAULT).clone();
                 self.control.tx.send(kernel::Message::CacheGetReply(value));
             }
-            kernel::Message::SubkernelMsgSend { id, destination: msg_dest, data } => {
+            kernel::Message::SubkernelMsgSend {
+                id,
+                destination: msg_dest,
+                data,
+            } => {
                 let msg_dest = msg_dest.or(Some(self.session.source)).unwrap();
-                self.session.messages.accept_outgoing(id, self_destination, msg_dest, data, routing_table, rank, router)?;
+                self.session.messages.accept_outgoing(
+                    id,
+                    self_destination,
+                    msg_dest,
+                    data,
+                    routing_table,
+                    rank,
+                    router,
+                )?;
                 self.session.kernel_state = KernelState::MsgSending;
             }
             kernel::Message::SubkernelMsgRecvRequest { id: _, timeout, tags } => {
                 let max_time = timer.get_time() + Milliseconds(timeout);
                 self.session.kernel_state = KernelState::MsgAwait(max_time, tags);
             }
-            kernel::Message::SubkernelLoadRunRequest { id, destination: sk_destination, run } => {
+            kernel::Message::SubkernelLoadRunRequest {
+                id,
+                destination: sk_destination,
+                run,
+            } => {
                 self.session.kernel_state = KernelState::SubkernelAwaitLoad;
-                router.route(drtioaux::Packet::SubkernelLoadRunRequest { 
-                    source: self_destination, destination: sk_destination, id: id, run: run 
-                }, routing_table, rank, self_destination);
+                router.route(
+                    drtioaux::Packet::SubkernelLoadRunRequest {
+                        source: self_destination,
+                        destination: sk_destination,
+                        id: id,
+                        run: run,
+                    },
+                    routing_table,
+                    rank,
+                    self_destination,
+                );
             }
 
-            kernel::Message::SubkernelAwaitFinishRequest{ id, timeout } => {
+            kernel::Message::SubkernelAwaitFinishRequest { id, timeout } => {
                 let max_time = timer.get_time() + Milliseconds(timeout);
-                self.session.kernel_state = KernelState::SubkernelAwaitFinish { max_time: max_time, id: id };
+                self.session.kernel_state = KernelState::SubkernelAwaitFinish {
+                    max_time: max_time,
+                    id: id,
+                };
             }
             kernel::Message::UpDestinationsRequest(destination) => {
-                self.control
-                    .tx
-                    .send(kernel::Message::UpDestinationsReply(destination == (self_destination as i32)));
+                self.control.tx.send(kernel::Message::UpDestinationsReply(
+                    destination == (self_destination as i32),
+                ));
             }
             _ => {
                 unexpected!("unexpected message from core1 while kernel was running: {:?}", reply);
@@ -622,7 +698,9 @@ impl<'a> Manager<'_> {
             }
             KernelState::SubkernelAwaitFinish { max_time, id } => {
                 if timer.get_time() > *max_time {
-                    self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply { status: kernel::SubkernelStatus::Timeout });
+                    self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply {
+                        status: kernel::SubkernelStatus::Timeout,
+                    });
                     self.session.kernel_state = KernelState::Running;
                 } else {
                     let mut i = 0;
@@ -635,12 +713,16 @@ impl<'a> Manager<'_> {
                     if let Some(finish_status) = self.session.subkernels_finished.remove(i) {
                         if finish_status.1 {
                             self.kernel_stop();
-                            self.last_finished = Some(SubkernelFinished { 
-                                source: self.session.source, id: self.session.id,
-                                with_exception: true, exception_source: finish_status.2
+                            self.last_finished = Some(SubkernelFinished {
+                                source: self.session.source,
+                                id: self.session.id,
+                                with_exception: true,
+                                exception_source: finish_status.2,
                             })
                         } else {
-                            self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply { status: kernel::SubkernelStatus::NoError });
+                            self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply {
+                                status: kernel::SubkernelStatus::NoError,
+                            });
                             self.session.kernel_state = KernelState::Running;
                         }
                     }