satellite: process kernel requests more often

src/libio/proto.rs

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

src/runtime/Cargo.toml

@@ -38,6 +38,6 @@ dyld = { path = "../libdyld" }
 dwarf = { path = "../libdwarf" }
 unwind = { path = "../libunwind" }
 libc = { path = "../libc" }
-io = { path = "../libio" }
+io = { path = "../libio", features = ["alloc"] }
 ksupport = { path = "../libksupport" }
 libboard_artiq = { path = "../libboard_artiq" }

src/satman/src/main.rs

@@ -126,6 +126,7 @@ fn process_aux_packet(
             #[cfg(not(has_drtio_routing))]
             let hop = 0;
 
+            if hop == 0 {
                 if let Some(status) = dma_manager.check_state() {
                     info!(
                         "playback done, error: {}, channel: {}, timestamp: {}",
@@ -134,14 +135,14 @@ fn process_aux_packet(
                     drtioaux::send(
                         0,
                         &drtioaux::Packet::DmaPlaybackStatus {
-                        destination: rank,
+                            destination: *_rank,
                             id: status.id,
                             error: status.error,
                             channel: status.channel,
                             timestamp: status.timestamp,
                         },
-                )
+                    )?;
-            } else if let Some(subkernel_finished) = kernel_manager.process_kern_requests(rank, timer) {
+                } else if let Some(subkernel_finished) = kernel_manager.get_last_finished() {
                     info!(
                         "subkernel {} finished, with exception: {}",
                         subkernel_finished.id, subkernel_finished.with_exception
@@ -152,42 +153,22 @@ fn process_aux_packet(
                             id: subkernel_finished.id,
                             with_exception: subkernel_finished.with_exception,
                         },
-                )
+                    )?;
                 } else if kernel_manager.message_is_ready() {
                     let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
-                if let Some(meta) = kernel_manager.message_get_slice(&mut data_slice) {
+                    match kernel_manager.message_get_slice(&mut data_slice) {
-                    drtioaux::send(
+                        Some(meta) => drtioaux::send(
                             0,
                             &drtioaux::Packet::SubkernelMessage {
-                            destination: rank,
+                                destination: *_rank,
                                 id: kernel_manager.get_current_id().unwrap(),
                                 last: meta.last,
                                 length: meta.len as u16,
                                 data: data_slice,
                             },
-                    )
+                        )?,
-                } else {
+                        None => warn!("subkernel message is ready but no message is present"),
-                    warn!("subkernel message is ready but no message is present");
-                    Ok(())
                     }
-            }
-
-            if hop == 0 {
-                if let Some(status) = dma_manager.check_state() {
-                    info!(
-                        "playback done, error: {}, channel: {}, timestamp: {}",
-                        status.error, status.channel, status.timestamp
-                    );
-                    drtioaux::send(
-                        0,
-                        &drtioaux::Packet::DmaPlaybackStatus {
-                            destination: _destination,
-                            id: status.id,
-                            error: status.error,
-                            channel: status.channel,
-                            timestamp: status.timestamp,
-                        },
-                    )?;
                 } else {
                     let errors;
                     unsafe {
@@ -897,6 +878,7 @@ pub extern "C" fn main_core0() -> i32 {
                     error!("aux packet error: {:?}", e);
                 }
             }
+            kernel_manager.process_kern_requests(rank, timer);
         }
 
         drtiosat_reset_phy(true);

src/satman/src/subkernel.rs

@@ -123,6 +123,7 @@ pub struct Manager<'a> {
     session: Session,
     control: &'a mut kernel::Control,
     cache: BTreeMap<String, Vec<i32>>,
+    last_finished: Option<SubkernelFinished>,
 }
 
 pub struct SubkernelFinished {
@@ -242,8 +243,8 @@ impl MessageManager {
     }
 
     pub fn accept_outgoing(&mut self, message: Vec<u8>) -> Result<(), Error> {
-        // skip service tag
+        // service tag skipped in kernel
-        self.out_message = Some(Sliceable::new(message[4..].to_vec()));
+        self.out_message = Some(Sliceable::new(message));
         self.out_state = OutMessageState::MessageReady;
         Ok(())
     }
@@ -260,6 +261,7 @@ impl<'a> Manager<'_> {
             session: Session::new(0),
             control: control,
             cache: BTreeMap::new(),
+            last_finished: None,
         }
     }
 
@@ -380,6 +382,10 @@ impl<'a> Manager<'_> {
         }
     }
 
+    pub fn get_last_finished(&mut self) -> Option<SubkernelFinished> {
+        self.last_finished.take()
+    }
+
     fn kernel_stop(&mut self) {
         self.session.kernel_state = KernelState::Absent;
         unsafe {
@@ -415,17 +421,17 @@ impl<'a> Manager<'_> {
         self.kernel_stop();
     }
 
-    pub fn process_kern_requests(&mut self, rank: u8, timer: GlobalTimer) -> Option<SubkernelFinished> {
+    pub fn process_kern_requests(&mut self, rank: u8, timer: GlobalTimer) {
         if !self.running() {
-            return None;
+            return;
         }
 
         match self.process_external_messages(timer) {
             Ok(()) => (),
-            Err(Error::AwaitingMessage) => return None, // kernel still waiting, do not process kernel messages
+            Err(Error::AwaitingMessage) => return, // kernel still waiting, do not process kernel messages
             Err(Error::KernelException(exception)) => {
                 self.session.last_exception = Some(exception);
-                return Some(SubkernelFinished {
+                self.last_finished = Some(SubkernelFinished {
                     id: self.session.id,
                     with_exception: true,
                 });
@@ -433,7 +439,7 @@ impl<'a> Manager<'_> {
             Err(e) => {
                 error!("Error while running processing external messages: {:?}", e);
                 self.runtime_exception(e);
-                return Some(SubkernelFinished {
+                self.last_finished = Some(SubkernelFinished {
                     id: self.session.id,
                     with_exception: true,
                 });
@@ -441,14 +447,16 @@ impl<'a> Manager<'_> {
         }
 
         match self.process_kern_message(rank, timer) {
-            Ok(true) => Some(SubkernelFinished {
+            Ok(true) => {
+                self.last_finished = Some(SubkernelFinished {
                     id: self.session.id,
                     with_exception: false,
-            }),
+                });
-            Ok(false) | Err(Error::NoMessage) => None,
+            }
+            Ok(false) | Err(Error::NoMessage) => (),
             Err(Error::KernelException(exception)) => {
                 self.session.last_exception = Some(exception);
-                return Some(SubkernelFinished {
+                self.last_finished = Some(SubkernelFinished {
                     id: self.session.id,
                     with_exception: true,
                 });
@@ -456,10 +464,10 @@ impl<'a> Manager<'_> {
             Err(e) => {
                 error!("Error while running kernel: {:?}", e);
                 self.runtime_exception(e);
-                Some(SubkernelFinished {
+                self.last_finished = Some(SubkernelFinished {
                     id: self.session.id,
                     with_exception: true,
-                })
+                });
             }
         }
     }