satman: support sub-subkernels, routing

src/libboard_artiq/src/drtioaux_proto.rs

@@ -935,7 +935,7 @@ impl Packet {
             Packet::SubkernelLoadRunRequest { destination, .. } => Some(*destination),
             Packet::SubkernelLoadRunReply { destination, .. } => Some(*destination),
             Packet::SubkernelMessage { destination, .. } => Some(*destination),
-            Packet::SubkernelMessageAck { destination, .. } => Some(*destination),
+            Packet::SubkernelMessageAck { destination } => Some(*destination),
             Packet::DmaPlaybackStatus { destination, .. } => Some(*destination),
             Packet::SubkernelFinished { destination, .. } => Some(*destination),
             _ => None,

src/satman/src/main.rs

@@ -489,7 +489,6 @@ fn process_aux_packet(
                 _routing_table,
                 *rank,
                 *self_destination,
-                _repeaters,
             )
         }
         drtioaux::Packet::DmaRemoveTraceRequest {
@@ -507,7 +506,6 @@ fn process_aux_packet(
                 _routing_table,
                 *rank,
                 *self_destination,
-                _repeaters,
             )
         }
         drtioaux::Packet::DmaPlaybackRequest {
@@ -530,7 +528,6 @@ fn process_aux_packet(
                 _routing_table,
                 *rank,
                 *self_destination,
-                _repeaters,
             )
         }
 
@@ -571,7 +568,6 @@ fn process_aux_packet(
                 _routing_table,
                 *rank,
                 *self_destination,
-                _repeaters,
             )
         }
         drtioaux::Packet::SubkernelLoadRunReply {
@@ -618,7 +614,10 @@ 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 })
+            router.send(drtioaux::Packet::SubkernelMessageAck { destination: source },
+                        _routing_table,
+                        *rank,
+                        *self_destination)
         }
         drtioaux::Packet::SubkernelMessageAck {
             destination: _destination,

src/satman/src/routing.rs

@@ -1,7 +1,6 @@
 use alloc::collections::vec_deque::VecDeque;
 
 use libboard_artiq::{drtio_routing, drtioaux, pl::csr};
-use repeater::Repeater;
 
 // Packets from downstream (further satellites) are received and routed appropriately.
 // they're passed as soon as possible downstream (within the subtree), or sent upstream,
@@ -68,7 +67,6 @@ impl Router {
         _routing_table: &drtio_routing::RoutingTable,
         _rank: u8,
         _destination: u8,
-        _repeaters: &[Repeater],
     ) -> Result<(), drtioaux::Error> {
         #[cfg(has_drtio_routing)]
         {

src/satman/src/subkernel.rs

@@ -104,8 +104,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: Vec<u32>,
 }
 
 impl Session {
@@ -116,7 +116,7 @@ impl Session {
             last_exception: None,
             messages: MessageManager::new(),
             source: 0,
-            subkernels_finished: VecDeque::new(),
+            subkernels_finished: Vec::new(),
         }
     }
 
@@ -474,6 +474,24 @@ impl<'a> Manager<'_> {
         destination: u8,
         timer: &GlobalTimer,
     ) {
+        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: subkernel_finished.exception_source,
+                },
+                &routing_table,
+                rank,
+                destination,
+            );
+        }
+
         if !self.running() {
             return;
         }
@@ -532,29 +550,13 @@ 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,
-            );
-        }
     }
 
     pub fn subkernel_load_run_reply(&mut self, succeeded: bool) {
         if self.session.kernel_state == KernelState::SubkernelAwaitLoad {
-            self.control.tx.send(kernel::Message::SubkernelLoadRunReply { succeeded: succeeded });
+            self.control
+                .tx
+                .send(kernel::Message::SubkernelLoadRunReply { succeeded: succeeded });
             self.session.kernel_state = KernelState::Running;
         } else {
             warn!("received unsolicited SubkernelLoadRunReply");
@@ -562,9 +564,17 @@ 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));
+        if with_exception {
+            self.kernel_stop();
+            self.last_finished = Some(SubkernelFinished {
+                source: self.session.source,
+                id: self.session.id,
+                with_exception: true,
+                exception_source: exception_source,
+            })
+        } else {
+            self.session.subkernels_finished.push(id);
+        }
     }
 
     fn process_kern_message(
@@ -704,27 +714,16 @@ impl<'a> Manager<'_> {
                     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 {
-                            self.kernel_stop();
-                            self.last_finished = Some(SubkernelFinished {
-                                source: self.session.source,
-                                id: self.session.id,
-                                with_exception: true,
-                                exception_source: finish_status.2,
-                            })
-                        } else {
+                    for status in &self.session.subkernels_finished {
+                        if *status == *id {
                             self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply {
                                 status: kernel::SubkernelStatus::NoError,
                             });
                             self.session.kernel_state = KernelState::Running;
+                            self.session.subkernels_finished.swap_remove(i);
+                            break;
                         }
+                        i += 1;
                     }
                 }
                 Ok(())