From ee0387962073af4fd7a92dca5d30b8b45466b997 Mon Sep 17 00:00:00 2001 From: Simon Renblad Date: Fri, 18 Oct 2024 14:16:48 +0800 Subject: [PATCH] fix satman error traits --- src/libboard_artiq/src/drtioaux.rs | 28 +-- src/libboard_artiq/src/drtioaux_async.rs | 16 +- src/libksupport/src/rpc.rs | 12 +- src/satman/src/main.rs | 4 +- src/satman/src/repeater.rs | 20 +- src/satman/src/routing.rs | 2 +- src/satman/src/subkernel.rs | 277 +++++++++++++++-------- 7 files changed, 224 insertions(+), 135 deletions(-) diff --git a/src/libboard_artiq/src/drtioaux.rs b/src/libboard_artiq/src/drtioaux.rs index 4bb12cc..dce496b 100644 --- a/src/libboard_artiq/src/drtioaux.rs +++ b/src/libboard_artiq/src/drtioaux.rs @@ -9,7 +9,7 @@ pub use crate::drtioaux_proto::Packet; use crate::{drtioaux_proto::Error as ProtocolError, mem::mem::DRTIOAUX_MEM, pl::csr::DRTIOAUX}; #[derive(Debug)] -pub enum Error { +pub enum Error { GatewareError, CorruptedPacket, @@ -19,17 +19,17 @@ pub enum Error { RoutingError, - Protocol(ProtocolError), + Protocol(ProtocolError), } -impl From for Error { - fn from(value: ProtocolError) -> Error { +impl From> for Error { + fn from(value: ProtocolError) -> Error { Error::Protocol(value) } } -impl From for Error { - fn from(value: IoError) -> Error { +impl From> for Error { + fn from(value: IoError) -> Error { Error::Protocol(ProtocolError::Io(value)) } } @@ -56,8 +56,8 @@ pub fn has_rx_error(linkno: u8) -> bool { } } -fn receive(linkno: u8, f: F) -> Result, Error> -where F: FnOnce(&[u8]) -> Result { +fn receive(linkno: u8, f: F) -> Result, Error> +where F: FnOnce(&[u8]) -> Result> { let linkidx = linkno as usize; unsafe { if (DRTIOAUX[linkidx].aux_rx_present_read)() == 1 { @@ -72,14 +72,14 @@ where F: FnOnce(&[u8]) -> Result { } } -pub fn recv(linkno: u8) -> Result, Error> { +pub fn recv(linkno: u8) -> Result, Error> { if has_rx_error(linkno) { return Err(Error::GatewareError); } receive(linkno, |buffer| { if buffer.len() < 8 { - return Err(IoError::UnexpectedEnd).into()); + return Err(IoError::UnexpectedEnd.into()); } let mut reader = Cursor::new(buffer); @@ -96,7 +96,7 @@ pub fn recv(linkno: u8) -> Result, Error> { }) } -pub fn recv_timeout(linkno: u8, timeout_ms: Option, timer: GlobalTimer) -> Result { +pub fn recv_timeout(linkno: u8, timeout_ms: Option, timer: GlobalTimer) -> Result> { let timeout_ms = Milliseconds(timeout_ms.unwrap_or(10)); let limit = timer.get_time() + timeout_ms; while timer.get_time() < limit { @@ -108,8 +108,8 @@ pub fn recv_timeout(linkno: u8, timeout_ms: Option, timer: GlobalTimer) -> Err(Error::TimedOut) } -fn transmit(linkno: u8, f: F) -> Result<(), Error> -where F: FnOnce(&mut [u8]) -> Result { +fn transmit(linkno: u8, f: F) -> Result<(), Error> +where F: FnOnce(&mut [u8]) -> Result> { let linkno = linkno as usize; unsafe { while (DRTIOAUX[linkno].aux_tx_read)() != 0 {} @@ -121,7 +121,7 @@ where F: FnOnce(&mut [u8]) -> Result { } } -pub fn send(linkno: u8, packet: &Packet) -> Result<(), Error> { +pub fn send(linkno: u8, packet: &Packet) -> Result<(), Error> { transmit(linkno, |buffer| { let mut writer = Cursor::new(buffer); diff --git a/src/libboard_artiq/src/drtioaux_async.rs b/src/libboard_artiq/src/drtioaux_async.rs index 3244432..e86b5fb 100644 --- a/src/libboard_artiq/src/drtioaux_async.rs +++ b/src/libboard_artiq/src/drtioaux_async.rs @@ -34,8 +34,8 @@ fn tx_ready(linkno: usize) -> nb::Result<(), Void> { } } -async fn receive(linkno: u8, f: F) -> Result, Error> -where F: FnOnce(&[u8]) -> Result { +async fn receive(linkno: u8, f: F) -> Result, Error> +where F: FnOnce(&[u8]) -> Result> { let linkidx = linkno as usize; unsafe { if (DRTIOAUX[linkidx].aux_rx_present_read)() == 1 { @@ -50,14 +50,14 @@ where F: FnOnce(&[u8]) -> Result { } } -pub async fn recv(linkno: u8) -> Result, Error> { +pub async fn recv(linkno: u8) -> Result, Error> { if has_rx_error(linkno) { return Err(Error::GatewareError); } receive(linkno, |buffer| { if buffer.len() < 8 { - return Err(IoError::UnexpectedEnd).into()); + return Err(IoError::UnexpectedEnd.into()); } let mut reader = Cursor::new(buffer); @@ -75,7 +75,7 @@ pub async fn recv(linkno: u8) -> Result, Error> { .await } -pub async fn recv_timeout(linkno: u8, timeout_ms: Option, timer: GlobalTimer) -> Result { +pub async fn recv_timeout(linkno: u8, timeout_ms: Option, timer: GlobalTimer) -> Result> { let timeout_ms = Milliseconds(timeout_ms.unwrap_or(10)); let limit = timer.get_time() + timeout_ms; let mut would_block = false; @@ -95,8 +95,8 @@ pub async fn recv_timeout(linkno: u8, timeout_ms: Option, timer: GlobalTime Err(Error::TimedOut) } -async fn transmit(linkno: u8, f: F) -> Result<(), Error> -where F: FnOnce(&mut [u8]) -> Result { +async fn transmit(linkno: u8, f: F) -> Result<(), Error> +where F: FnOnce(&mut [u8]) -> Result> { let linkno = linkno as usize; unsafe { let _ = block_async!(tx_ready(linkno)).await; @@ -108,7 +108,7 @@ where F: FnOnce(&mut [u8]) -> Result { } } -pub async fn send(linkno: u8, packet: &Packet) -> Result<(), Error> { +pub async fn send(linkno: u8, packet: &Packet) -> Result<(), Error> { transmit(linkno, |buffer| { let mut writer = Cursor::new(buffer); diff --git a/src/libksupport/src/rpc.rs b/src/libksupport/src/rpc.rs index 4bfa425..85825ec 100644 --- a/src/libksupport/src/rpc.rs +++ b/src/libksupport/src/rpc.rs @@ -44,7 +44,7 @@ unsafe fn recv_elements( alloc: &mut F, ) -> Result<(), E> where - F: FnMut(usize) -> *mut (), + F: FnMut(usize) -> Result<*mut (), E>, R: Read + ?Sized, E: From>, { @@ -81,7 +81,7 @@ where unsafe fn recv_value(reader: &mut R, tag: Tag, data: &mut *mut (), alloc: &mut F) -> Result<(), E> where - F: FnMut(usize) -> *mut (), + F: FnMut(usize) -> Result<*mut (), E>, R: Read + ?Sized, E: From>, { @@ -110,7 +110,7 @@ where Tag::String | Tag::Bytes | Tag::ByteArray => { consume_value!(CMutSlice, |ptr| { let length = reader.read_u32()? as usize; - *ptr = CMutSlice::new(alloc(length) as *mut u8, length); + *ptr = CMutSlice::new(alloc(length)? as *mut u8, length); reader.read_exact((*ptr).as_mut())?; Ok(()) }) @@ -140,7 +140,7 @@ where let storage_offset = round_up(list_size, tag.alignment()); let storage_size = tag.size() * length; - let allocation = alloc(storage_offset + storage_size) as *mut u8; + let allocation = alloc(storage_offset + storage_size)? as *mut u8; *ptr_to_list = allocation as *mut List; let storage = allocation.offset(storage_offset as isize) as *mut (); @@ -159,7 +159,7 @@ where } let elt_tag = it.clone().next().expect("truncated tag"); - *buffer = alloc(elt_tag.size() * total_len); + *buffer = alloc(elt_tag.size() * total_len)?; recv_elements(reader, elt_tag, total_len, *buffer, alloc) }) } @@ -183,7 +183,7 @@ pub fn recv_return<'a, F, R, E>( alloc: &mut F, ) -> Result<&'a [u8], E> where - F: FnMut(usize) -> *mut (), + F: FnMut(usize) -> Result<*mut (), E>, R: Read + ?Sized, E: From>, { diff --git a/src/satman/src/main.rs b/src/satman/src/main.rs index d77cff1..dd63c5f 100644 --- a/src/satman/src/main.rs +++ b/src/satman/src/main.rs @@ -1,6 +1,6 @@ #![no_std] #![no_main] -#![feature(alloc_error_handler, try_trait, never_type, panic_info_message)] +#![feature(alloc_error_handler, never_type)] #[macro_use] extern crate log; @@ -149,7 +149,7 @@ fn process_aux_packet( analyzer: &mut Analyzer, kernel_manager: &mut KernelManager, router: &mut Router, -) -> Result<(), drtioaux::Error> { +) -> Result<(), drtioaux::Error> { // In the code below, *_chan_sel_write takes an u8 if there are fewer than 256 channels, // and u16 otherwise; hence the `as _` conversion. match packet { diff --git a/src/satman/src/repeater.rs b/src/satman/src/repeater.rs index 7834007..8481aa3 100644 --- a/src/satman/src/repeater.rs +++ b/src/satman/src/repeater.rs @@ -191,7 +191,7 @@ impl Repeater { } } - fn recv_aux_timeout(&self, timeout: u32, timer: &mut GlobalTimer) -> Result { + fn recv_aux_timeout(&self, timeout: u32, timer: &mut GlobalTimer) -> Result> { let max_time = timer.get_time() + Milliseconds(timeout.into()); loop { if !rep_link_rx_up(self.repno) { @@ -216,7 +216,7 @@ impl Repeater { rank: u8, self_destination: u8, timer: &mut GlobalTimer, - ) -> Result<(), drtioaux::Error> { + ) -> Result<(), drtioaux::Error> { self.aux_send(request)?; loop { let reply = self.recv_aux_timeout(200, timer)?; @@ -242,14 +242,14 @@ impl Repeater { Ok(()) } - pub fn aux_send(&self, request: &drtioaux::Packet) -> Result<(), drtioaux::Error> { + pub fn aux_send(&self, request: &drtioaux::Packet) -> Result<(), drtioaux::Error> { if self.state != RepeaterState::Up { return Err(drtioaux::Error::LinkDown); } drtioaux::send(self.auxno, request) } - pub fn sync_tsc(&self, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { + pub fn sync_tsc(&self, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { if self.state != RepeaterState::Up { return Ok(()); } @@ -275,7 +275,7 @@ impl Repeater { destination: u8, hops: &[u8; drtio_routing::MAX_HOPS], timer: &mut GlobalTimer, - ) -> Result<(), drtioaux::Error> { + ) -> Result<(), drtioaux::Error> { if self.state != RepeaterState::Up { return Ok(()); } @@ -299,14 +299,14 @@ impl Repeater { &self, routing_table: &drtio_routing::RoutingTable, timer: &mut GlobalTimer, - ) -> Result<(), drtioaux::Error> { + ) -> Result<(), drtioaux::Error> { for i in 0..drtio_routing::DEST_COUNT { self.set_path(i as u8, &routing_table.0[i], timer)?; } Ok(()) } - pub fn set_rank(&self, rank: u8, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { + pub fn set_rank(&self, rank: u8, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { if self.state != RepeaterState::Up { return Ok(()); } @@ -318,7 +318,7 @@ impl Repeater { Ok(()) } - pub fn rtio_reset(&self, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { + pub fn rtio_reset(&self, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { let repno = self.repno as usize; unsafe { (csr::DRTIOREP[repno].reset_write)(1); @@ -361,11 +361,11 @@ impl Repeater { ) { } - pub fn sync_tsc(&self, _timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { + pub fn sync_tsc(&self, _timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { Ok(()) } - pub fn rtio_reset(&self, _timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { + pub fn rtio_reset(&self, _timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> { Ok(()) } } diff --git a/src/satman/src/routing.rs b/src/satman/src/routing.rs index 87d5f09..de2ad32 100644 --- a/src/satman/src/routing.rs +++ b/src/satman/src/routing.rs @@ -129,7 +129,7 @@ impl Router { _routing_table: &drtio_routing::RoutingTable, _rank: u8, _destination: u8, - ) -> Result<(), drtioaux::Error> { + ) -> Result<(), drtioaux::Error> { #[cfg(has_drtio_routing)] { let destination = packet.routable_destination(); diff --git a/src/satman/src/subkernel.rs b/src/satman/src/subkernel.rs index b418363..9b836e0 100644 --- a/src/satman/src/subkernel.rs +++ b/src/satman/src/subkernel.rs @@ -1,17 +1,21 @@ -use alloc::{collections::BTreeMap, - format, - string::{String, ToString}, - vec::Vec}; -use core::{option::NoneError, slice, str}; +use alloc::{ + collections::BTreeMap, + format, + string::{String, ToString}, + vec::Vec, +}; +use core::{slice, str}; use cslice::AsCSlice; use dma::{Error as DmaError, Manager as DmaManager}; use io::{Cursor, Error as IoError, ProtoWrite, Write}; use ksupport::{eh_artiq, kernel, rpc}; -use libboard_artiq::{drtio_routing::RoutingTable, - drtioaux, - drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE}, - pl::csr}; +use libboard_artiq::{ + drtio_routing::RoutingTable, + drtioaux, + drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE}, + pl::csr, +}; use libboard_zynq::{time::Milliseconds, timer::GlobalTimer}; use libcortex_a9::sync_channel::Receiver; use log::warn; @@ -51,6 +55,7 @@ enum KernelState { }, } +#[allow(dead_code)] #[derive(Debug)] pub enum Error { Load(String), @@ -64,14 +69,8 @@ pub enum Error { DmaError(DmaError), } -impl From for Error { - fn from(_: NoneError) -> Error { - Error::KernelNotFound - } -} - -impl From for Error { - fn from(_value: IoError) -> Error { +impl From> for Error { + fn from(_value: IoError) -> Error { Error::SubkernelIoError } } @@ -88,8 +87,8 @@ impl From<()> for Error { } } -impl From for Error { - fn from(_value: drtioaux::Error) -> Error { +impl From> for Error { + fn from(_value: drtioaux::Error) -> Error { Error::DrtioError } } @@ -125,7 +124,7 @@ struct MessageManager { struct Session { id: u32, kernel_state: KernelState, - last_exception: Option, // exceptions raised locally + last_exception: Option, // exceptions raised locally external_exception: Option>, // exceptions from sub-subkernels messages: MessageManager, source: u8, // which destination requested running the kernel @@ -221,7 +220,10 @@ impl MessageManager { } } - pub fn get_outgoing_slice(&mut self, data_slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE]) -> Option { + pub fn get_outgoing_slice( + &mut self, + data_slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE], + ) -> Option { if self.out_state != OutMessageState::MessageBeingSent { return None; } @@ -302,7 +304,13 @@ impl<'a> Manager<'_> { } } - pub fn add(&mut self, id: u32, status: PayloadStatus, data: &[u8], data_len: usize) -> Result<(), Error> { + pub fn add( + &mut self, + id: u32, + status: PayloadStatus, + data: &[u8], + data_len: usize, + ) -> Result<(), Error> { let kernel = match self.kernels.get_mut(&id) { Some(kernel) => { if kernel.complete || status.is_first() { @@ -315,7 +323,7 @@ impl<'a> Manager<'_> { complete: false, }, ); - self.kernels.get_mut(&id)? + self.kernels.get_mut(&id).ok_or(Error::KernelNotFound)? } else { kernel } @@ -328,7 +336,7 @@ impl<'a> Manager<'_> { complete: false, }, ); - self.kernels.get_mut(&id)? + self.kernels.get_mut(&id).ok_or(Error::KernelNotFound)? } }; kernel.library.extend(&data[0..data_len]); @@ -372,10 +380,15 @@ impl<'a> Manager<'_> { if !self.running() { return; } - self.session.messages.handle_incoming(status, id, length, slice); + self.session + .messages + .handle_incoming(status, id, length, slice); } - pub fn message_get_slice(&mut self, slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE]) -> Option { + pub fn message_get_slice( + &mut self, + slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE], + ) -> Option { if !self.running() { return None; } @@ -394,15 +407,19 @@ impl<'a> Manager<'_> { if self.session.id == id && self.session.kernel_state == KernelState::Loaded { return Ok(()); } - if !self.kernels.get(&id)?.complete { + if !self.kernels.get(&id).ok_or(Error::KernelNotFound)?.complete { return Err(Error::KernelNotFound); } self.session = Session::new(id); self.control.restart(); - self.control - .tx - .send(kernel::Message::LoadRequest(self.kernels.get(&id)?.library.clone())); + self.control.tx.send(kernel::Message::LoadRequest( + self.kernels + .get(&id) + .ok_or(Error::KernelNotFound)? + .library + .clone(), + )); let reply = self.control.rx.recv(); match reply { kernel::Message::LoadCompleted => Ok(()), @@ -414,7 +431,10 @@ impl<'a> Manager<'_> { } } - pub fn exception_get_slice(&mut self, data_slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE]) -> SliceMeta { + pub fn exception_get_slice( + &mut self, + data_slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE], + ) -> SliceMeta { match self.session.last_exception.as_mut() { Some(exception) => exception.get_slice_master(data_slice), None => SliceMeta { @@ -568,7 +588,14 @@ impl<'a> Manager<'_> { } } - match self.process_kern_message(router, routing_table, rank, destination, dma_manager, timer) { + match self.process_kern_message( + router, + routing_table, + rank, + destination, + dma_manager, + timer, + ) { Ok(true) => { self.last_finished = Some(SubkernelFinished { id: self.session.id, @@ -611,7 +638,9 @@ impl<'a> Manager<'_> { for (i, (status, exception_source)) in self.session.subkernels_finished.iter().enumerate() { if *status == id { if exception_source.is_none() { - self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply); + self.control + .tx + .send(kernel::Message::SubkernelAwaitFinishReply); self.session.kernel_state = KernelState::Running; self.session.subkernels_finished.swap_remove(i); } else { @@ -639,15 +668,26 @@ impl<'a> Manager<'_> { if self.session.kernel_state == KernelState::SubkernelAwaitLoad { self.control .tx - .send(kernel::Message::SubkernelLoadRunReply { succeeded: succeeded }); + .send(kernel::Message::SubkernelLoadRunReply { + succeeded: succeeded, + }); 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) { - let exception_src = if with_exception { Some(exception_source) } else { None }; + pub fn remote_subkernel_finished( + &mut self, + id: u32, + with_exception: bool, + exception_source: u8, + ) { + let exception_src = if with_exception { + Some(exception_source) + } else { + None + }; self.session.subkernels_finished.push((id, exception_src)); } @@ -660,18 +700,19 @@ impl<'a> Manager<'_> { rank: u8, self_destination: u8, ) { - if let KernelState::SubkernelRetrievingException { destination } = self.session.kernel_state { + if let KernelState::SubkernelRetrievingException { destination } = self.session.kernel_state + { self.session .external_exception .as_mut() .unwrap() .extend_from_slice(exception_data); if last { - self.control - .tx - .send(kernel::Message::SubkernelError(kernel::SubkernelStatus::Exception( + self.control.tx.send(kernel::Message::SubkernelError( + kernel::SubkernelStatus::Exception( self.session.external_exception.take().unwrap(), - ))); + ), + )); self.session.kernel_state = KernelState::Running; } else { /* fetch another slice */ @@ -706,7 +747,12 @@ impl<'a> Manager<'_> { dma_manager.cleanup(router, rank, self_destination, routing_table); return Ok(true); } - kernel::Message::KernelException(exceptions, stack_pointers, backtrace, async_errors) => { + kernel::Message::KernelException( + exceptions, + stack_pointers, + backtrace, + async_errors, + ) => { error!("exception in kernel"); for exception in exceptions { error!("{:?}", exception.unwrap()); @@ -715,12 +761,21 @@ impl<'a> Manager<'_> { error!("backtrace: {:?}", backtrace); let buf: Vec = Vec::new(); let mut writer = Cursor::new(buf); - match write_exception(&mut writer, exceptions, stack_pointers, backtrace, async_errors) { + match write_exception( + &mut writer, + exceptions, + stack_pointers, + backtrace, + async_errors, + ) { Ok(()) => (), Err(_) => error!("Error writing exception data"), } self.kernel_stop(); - return Err(Error::KernelException(Sliceable::new(0, writer.into_inner()))); + return Err(Error::KernelException(Sliceable::new( + 0, + writer.into_inner(), + ))); } kernel::Message::CachePutRequest(key, value) => { self.cache.insert(key, value); @@ -768,11 +823,13 @@ impl<'a> Manager<'_> { let max_time = timer.get_time() + Milliseconds(10000); self.session.kernel_state = match self.session.kernel_state { // if we are still waiting for the traces to be uploaded, extend the state by timeout - KernelState::DmaPendingPlayback { id, timestamp } => KernelState::DmaPendingAwait { - id: id, - timestamp: timestamp, - max_time: max_time, - }, + KernelState::DmaPendingPlayback { id, timestamp } => { + KernelState::DmaPendingAwait { + id: id, + timestamp: timestamp, + max_time: max_time, + } + } _ => KernelState::DmaAwait { max_time: max_time }, }; } @@ -843,7 +900,10 @@ impl<'a> Manager<'_> { )); } _ => { - unexpected!("unexpected message from core1 while kernel was running: {:?}", reply); + unexpected!( + "unexpected message from core1 while kernel was running: {:?}", + reply + ); } } Ok(false) @@ -861,9 +921,9 @@ impl<'a> Manager<'_> { KernelState::MsgAwait { max_time, id, tags } => { if let Some(max_time) = *max_time { if timer.get_time() > max_time { - self.control - .tx - .send(kernel::Message::SubkernelError(kernel::SubkernelStatus::Timeout)); + self.control.tx.send(kernel::Message::SubkernelError( + kernel::SubkernelStatus::Timeout, + )); self.session.kernel_state = KernelState::Running; return Ok(()); } @@ -871,7 +931,9 @@ impl<'a> Manager<'_> { if let Some(message) = self.session.messages.get_incoming(*id) { self.control .tx - .send(kernel::Message::SubkernelMsgRecvReply { count: message.count }); + .send(kernel::Message::SubkernelMsgRecvReply { + count: message.count, + }); let tags = tags.clone(); self.session.kernel_state = KernelState::Running; self.pass_message_to_kernel(&message, tags, timer) @@ -893,9 +955,9 @@ impl<'a> Manager<'_> { KernelState::SubkernelAwaitFinish { max_time, id } => { if let Some(max_time) = *max_time { if timer.get_time() > max_time { - self.control - .tx - .send(kernel::Message::SubkernelError(kernel::SubkernelStatus::Timeout)); + self.control.tx.send(kernel::Message::SubkernelError( + kernel::SubkernelStatus::Timeout, + )); self.session.kernel_state = KernelState::Running; return Ok(()); } @@ -921,7 +983,12 @@ impl<'a> Manager<'_> { } } - fn pass_message_to_kernel(&mut self, message: &Message, tags: Vec, timer: &GlobalTimer) -> Result<(), Error> { + fn pass_message_to_kernel( + &mut self, + message: &Message, + tags: Vec, + timer: &GlobalTimer, + ) -> Result<(), Error> { let mut reader = Cursor::new(&message.data); let mut current_tags: &[u8] = &tags; let mut i = message.count; @@ -932,33 +999,54 @@ impl<'a> Manager<'_> { }; let mut exception: Option = None; let mut unexpected: Option = None; - let remaining_tags = rpc::recv_return(&mut reader, current_tags, slot, &mut |size| { - if size == 0 { - 0 as *mut () - } else { - self.control.tx.send(kernel::Message::RpcRecvReply(Ok(size))); - match recv_w_timeout(&mut self.control.rx, timer, 100) { - Ok(kernel::Message::RpcRecvRequest(slot)) => slot, - Ok(kernel::Message::KernelException(exceptions, stack_pointers, backtrace, async_errors)) => { - let buf: Vec = Vec::new(); - let mut writer = Cursor::new(buf); - match write_exception(&mut writer, exceptions, stack_pointers, backtrace, async_errors) { - Ok(()) => { - exception = Some(Sliceable::new(0, writer.into_inner())); - } - Err(_) => { - unexpected = Some("Error writing exception data".to_string()); - } - }; - 0 as *mut () - } - other => { - unexpected = Some(format!("expected nested value slot from kernel CPU, not {:?}", other)); - 0 as *mut () + let remaining_tags = + rpc::recv_return(&mut reader, current_tags, slot, &mut |size| -> Result< + _, + Error, + > { + if size == 0 { + Ok(0 as *mut ()) + } else { + self.control + .tx + .send(kernel::Message::RpcRecvReply(Ok(size))); + match recv_w_timeout(&mut self.control.rx, timer, 100) { + Ok(kernel::Message::RpcRecvRequest(slot)) => Ok(slot), + Ok(kernel::Message::KernelException( + exceptions, + stack_pointers, + backtrace, + async_errors, + )) => { + let buf: Vec = Vec::new(); + let mut writer = Cursor::new(buf); + match write_exception( + &mut writer, + exceptions, + stack_pointers, + backtrace, + async_errors, + ) { + Ok(()) => { + exception = Some(Sliceable::new(0, writer.into_inner())); + } + Err(_) => { + unexpected = + Some("Error writing exception data".to_string()); + } + }; + Ok(0 as *mut ()) + } + other => { + unexpected = Some(format!( + "expected nested value slot from kernel CPU, not {:?}", + other + )); + Ok(0 as *mut ()) + } } } - } - })?; + })?; if let Some(exception) = exception { self.kernel_stop(); return Err(Error::KernelException(exception)); @@ -1000,17 +1088,18 @@ where writer.write_u32(u32::MAX)?; writer.write_u32(exception.message.as_ptr() as u32)?; } else { - let msg = - str::from_utf8(unsafe { slice::from_raw_parts(exception.message.as_ptr(), exception.message.len()) }) - .unwrap() - .replace( - "{rtio_channel_info:0}", - &format!( - "0x{:04x}:{}", - exception.param[0], - ksupport::resolve_channel_name(exception.param[0] as u32) - ), - ); + let msg = str::from_utf8(unsafe { + slice::from_raw_parts(exception.message.as_ptr(), exception.message.len()) + }) + .unwrap() + .replace( + "{rtio_channel_info:0}", + &format!( + "0x{:04x}:{}", + exception.param[0], + ksupport::resolve_channel_name(exception.param[0] as u32) + ), + ); writer.write_string(&msg)?; } writer.write_u64(exception.param[0] as u64)?;