satman: support sub-subkernels, routing

src/libboard_artiq/src/drtioaux_proto.rs

@@ -938,7 +938,7 @@ impl Packet {
             Packet::SubkernelMessageAck { destination, .. } => Some(*destination),
             Packet::DmaPlaybackStatus { destination, .. } => Some(*destination),
             Packet::SubkernelFinished { destination, .. } => Some(*destination),
-            _ => None
+            _ => 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::DmaAddTraceReply { .. }
-                Packet::DmaPlaybackReply { .. } | Packet::SubkernelLoadRunReply { .. } |
+            | Packet::DmaRemoveTraceReply { .. }
-                Packet::SubkernelMessageAck { .. } | Packet::DmaPlaybackStatus { .. } |
+            | Packet::DmaPlaybackReply { .. }
-                Packet::SubkernelFinished { .. } => false,
+            | Packet::SubkernelLoadRunReply { .. }
-            _ => true
+            | 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) {
@@ -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 { 
+            router.send(
-                destination: source, succeeded: succeeded 
+                drtioaux::Packet::DmaAddTraceReply {
-            }, _routing_table, *rank, *self_destination, _repeaters)
+                    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 { 
+            router.send(
-                destination: source, succeeded: succeeded 
+                drtioaux::Packet::DmaRemoveTraceReply {
-            }, _routing_table, *rank, *self_destination, _repeaters)
+                    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 { 
+            router.send(
-                destination: source, succeeded: succeeded
+                drtioaux::Packet::DmaPlaybackReply {
-            }, _routing_table, *rank, *self_destination, _repeaters)
+                    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 { 
+            router.send(
-                destination: source, succeeded: succeeded 
+                drtioaux::Packet::SubkernelLoadRunReply {
+                    destination: source,
+                    succeeded: succeeded,
                 },
-            _routing_table, *rank, *self_destination, _repeaters)
+                _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(
+            drtioaux::send(0, &drtioaux::Packet::SubkernelMessageAck { destination: source })
-                0,
-                &drtioaux::Packet::SubkernelMessageAck {
-                    destination: source,
-                },
-            )
         }
         drtioaux::Packet::SubkernelMessageAck {
             destination: _destination,
@@ -599,9 +629,19 @@ fn process_aux_packet(
                 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 {
+                    router.route(
-                    source: *self_destination, destination: meta.destination, id: kernel_manager.get_current_id().unwrap(),
+                        drtioaux::Packet::SubkernelMessage {
-                    status: meta.status, length: meta.len as u16, data: data_slice}, _routing_table, *rank, *self_destination);
+                            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 {
@@ -189,14 +195,20 @@ impl Repeater {
         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.
@@ -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,
+    pub fn route(
-        _routing_table: &drtio_routing::RoutingTable, _rank: u8,
+        &mut self,
-        _self_destination: u8
+        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,
+    pub fn send(
-        _routing_table: &drtio_routing::RoutingTable, _rank: u8,
+        &mut self,
-        _destination: u8, _repeaters: &[Repeater]
+        packet: drtioaux::Packet,
+        _routing_table: &drtio_routing::RoutingTable,
+        _rank: u8,
+        _destination: u8,
+        _repeaters: &[Repeater],
     ) -> Result<(), drtioaux::Error> {
         #[cfg(has_drtio_routing)]
         {

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)]
@@ -105,7 +106,7 @@ struct Session {
     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
+    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::Running
-            KernelState::SubkernelAwaitFinish { .. } => true
+            | 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, 
+    pub fn accept_outgoing(
-        message: Vec<u8>, routing_table: &RoutingTable, rank: u8, router: &mut Router
+        &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 {
+        router.route(
-                source: self_destination, destination: destination, id: id,
+            drtioaux::Packet::SubkernelMessage {
-                status: meta.status, length: meta.len as u16, data: data_slice
+                source: self_destination,
-        }, routing_table, rank, 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);
+            info!(
-            router.route(drtioaux::Packet::SubkernelFinished {
+                "subkernel {} finished, with exception: {}",
-                destination: subkernel_finished.source, id: subkernel_finished.id, 
+                subkernel_finished.id, subkernel_finished.with_exception
-                with_exception: subkernel_finished.with_exception, exception_src: destination
+            );
-            }, &routing_table, rank, destination);
+            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, 
+    fn process_kern_message(
-        routing_table: &RoutingTable, rank: u8, self_destination: u8, timer: &GlobalTimer) -> Result<bool, Error> {
+        &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 { 
+                router.route(
-                    source: self_destination, destination: sk_destination, id: id, run: run 
+                    drtioaux::Packet::SubkernelLoadRunRequest {
-                }, routing_table, rank, self_destination);
+                        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
+                self.control.tx.send(kernel::Message::UpDestinationsReply(
-                    .tx
+                    destination == (self_destination as i32),
-                    .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;
@@ -636,11 +714,15 @@ impl<'a> Manager<'_> {
                         if finish_status.1 {
                             self.kernel_stop();
                             self.last_finished = Some(SubkernelFinished {
-                                source: self.session.source, id: self.session.id,
+                                source: self.session.source,
-                                with_exception: true, exception_source: finish_status.2
+                                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;
                         }
                     }