apply rustfmt policies to ddma code

src/libboard_artiq/src/drtioaux_proto.rs

@@ -141,32 +141,32 @@ pub enum Packet {
         id: u32,
         last: bool,
         length: u16,
-        trace: [u8; DMA_TRACE_MAX_SIZE] 
+        trace: [u8; DMA_TRACE_MAX_SIZE],
     },
     DmaAddTraceReply {
-        succeeded: bool 
+        succeeded: bool,
     },
     DmaRemoveTraceRequest {
         destination: u8,
-        id: u32 
+        id: u32,
     },
     DmaRemoveTraceReply {
-        succeeded: bool 
+        succeeded: bool,
     },
     DmaPlaybackRequest {
         destination: u8,
         id: u32,
-        timestamp: u64 
+        timestamp: u64,
     },
     DmaPlaybackReply {
-        succeeded: bool 
+        succeeded: bool,
     },
     DmaPlaybackStatus {
         destination: u8,
         id: u32,
         error: u8,
         channel: u32,
-        timestamp: u64 
+        timestamp: u64,
     },
 }
 
@@ -312,31 +312,31 @@ impl Packet {
                     length: length as u16,
                     trace: trace,
                 }
-            },
+            }
             0xb1 => Packet::DmaAddTraceReply {
-                succeeded: reader.read_bool()?
+                succeeded: reader.read_bool()?,
             },
             0xb2 => Packet::DmaRemoveTraceRequest {
                 destination: reader.read_u8()?,
-                id: reader.read_u32()?
+                id: reader.read_u32()?,
             },
             0xb3 => Packet::DmaRemoveTraceReply {
-                succeeded: reader.read_bool()?
+                succeeded: reader.read_bool()?,
             },
             0xb4 => Packet::DmaPlaybackRequest {
                 destination: reader.read_u8()?,
                 id: reader.read_u32()?,
-                timestamp: reader.read_u64()?
+                timestamp: reader.read_u64()?,
             },
             0xb5 => Packet::DmaPlaybackReply {
-                succeeded: reader.read_bool()?
+                succeeded: reader.read_bool()?,
             },
             0xb6 => Packet::DmaPlaybackStatus {
                 destination: reader.read_u8()?,
                 id: reader.read_u32()?,
                 error: reader.read_u8()?,
                 channel: reader.read_u32()?,
-                timestamp: reader.read_u64()?
+                timestamp: reader.read_u64()?,
             },
 
             ty => return Err(Error::UnknownPacket(ty)),
@@ -531,7 +531,7 @@ impl Packet {
                 id,
                 last,
                 trace,
-                length 
+                length,
             } => {
                 writer.write_u8(0xb0)?;
                 writer.write_u8(destination)?;
@@ -558,7 +558,7 @@ impl Packet {
             Packet::DmaPlaybackRequest {
                 destination,
                 id,
-                timestamp 
+                timestamp,
             } => {
                 writer.write_u8(0xb4)?;
                 writer.write_u8(destination)?;
@@ -574,7 +574,7 @@ impl Packet {
                 id,
                 error,
                 channel,
-                timestamp 
+                timestamp,
             } => {
                 writer.write_u8(0xb6)?;
                 writer.write_u8(destination)?;

src/runtime/src/comms.rs

@@ -26,9 +26,8 @@ use num_traits::{FromPrimitive, ToPrimitive};
 use crate::pl;
 use crate::{analyzer, kernel, mgmt, moninj,
             proto_async::*,
-            rpc,
+            rpc, rtio_dma,
-            rtio_mgt::{self, resolve_channel_name},
+            rtio_mgt::{self, resolve_channel_name}};
-            rtio_dma};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Error {
@@ -81,7 +80,6 @@ enum Reply {
 
 static CACHE_STORE: Mutex<BTreeMap<String, Vec<i32>>> = Mutex::new(BTreeMap::new());
 
-
 async fn write_header(stream: &TcpStream, reply: Reply) -> Result<()> {
     stream
         .send_slice(&[0x5a, 0x5a, 0x5a, 0x5a, reply.to_u8().unwrap()])
@@ -160,7 +158,7 @@ async fn handle_run_kernel(
     _up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
     aux_mutex: &Rc<Mutex<bool>>,
     routing_table: &drtio_routing::RoutingTable,
-    timer: GlobalTimer
+    timer: GlobalTimer,
 ) -> Result<()> {
     control.borrow_mut().tx.async_send(kernel::Message::StartRequest).await;
     loop {
@@ -350,19 +348,19 @@ async fn handle_run_kernel(
                     Ok(rtio_dma::remote_dma::RemoteState::PlaybackEnded {
                         error,
                         channel,
-                        timestamp
+                        timestamp,
                     }) => kernel::Message::DmaAwaitRemoteReply {
                         timeout: false,
                         error: error,
                         channel: channel,
-                        timestamp: timestamp 
+                        timestamp: timestamp,
                     },
                     _ => kernel::Message::DmaAwaitRemoteReply {
                         timeout: true,
                         error: 0,
                         channel: 0,
-                        timestamp: 0 
+                        timestamp: 0,
-                    }
+                    },
                 };
                 control.borrow_mut().tx.async_send(reply).await;
             }
@@ -428,7 +426,7 @@ async fn handle_connection(
     up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
     aux_mutex: &Rc<Mutex<bool>>,
     routing_table: &drtio_routing::RoutingTable,
-    timer: GlobalTimer
+    timer: GlobalTimer,
 ) -> Result<()> {
     stream.set_ack_delay(None);
 
@@ -452,7 +450,15 @@ async fn handle_connection(
                 load_kernel(&buffer, &control, Some(stream)).await?;
             }
             Request::RunKernel => {
-                handle_run_kernel(Some(stream), &control, &up_destinations, aux_mutex, routing_table, timer).await?;
+                handle_run_kernel(
+                    Some(stream),
+                    &control,
+                    &up_destinations,
+                    aux_mutex,
+                    routing_table,
+                    timer,
+                )
+                .await?;
             }
             _ => {
                 error!("unexpected request from host: {:?}", request);
@@ -528,7 +534,14 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
         if let Ok(()) = task::block_on(load_kernel(&buffer, &control, None)) {
             info!("Starting startup kernel...");
             let routing_table = drtio_routing_table.borrow();
-            let _ = task::block_on(handle_run_kernel(None, &control, &up_destinations, &aux_mutex, &routing_table, timer));
+            let _ = task::block_on(handle_run_kernel(
+                None,
+                &control,
+                &up_destinations,
+                &aux_mutex,
+                &routing_table,
+                timer,
+            ));
             info!("Startup kernel finished!");
         } else {
             error!("Error loading startup kernel!");

src/runtime/src/kernel/dma.rs

@@ -17,7 +17,7 @@ pub struct DmaRecorder {
     pub name: String,
     pub buffer: Vec<u8>,
     pub duration: i64,
-    pub enable_ddma: bool
+    pub enable_ddma: bool,
 }
 
 static mut RECORDER: Option<DmaRecorder> = None;
@@ -51,7 +51,7 @@ pub extern "C" fn dma_record_start(name: CSlice<u8>) {
             name,
             buffer: Vec::new(),
             duration: 0,
-            enable_ddma: false
+            enable_ddma: false,
         });
     }
 }
@@ -168,8 +168,13 @@ pub extern "C" fn dma_playback(timestamp: i64, ptr: i32) {
         csr::cri_con::selected_write(1);
         csr::rtio_dma::enable_write(1);
         #[cfg(has_drtio)]
-        KERNEL_CHANNEL_1TO0.as_mut().unwrap().send(
+        KERNEL_CHANNEL_1TO0
-            Message::DmaStartRemoteRequest{ id: ptr, timestamp: timestamp });
+            .as_mut()
+            .unwrap()
+            .send(Message::DmaStartRemoteRequest {
+                id: ptr,
+                timestamp: timestamp,
+            });
         while csr::rtio_dma::enable_read() != 0 {}
         csr::cri_con::selected_write(0);
 
@@ -199,10 +204,17 @@ pub extern "C" fn dma_playback(timestamp: i64, ptr: i32) {
         }
         #[cfg(has_drtio)]
         {
-            KERNEL_CHANNEL_1TO0.as_mut().unwrap().send(
+            KERNEL_CHANNEL_1TO0
-                Message::DmaAwaitRemoteRequest(ptr));
+                .as_mut()
+                .unwrap()
+                .send(Message::DmaAwaitRemoteRequest(ptr));
             match KERNEL_CHANNEL_0TO1.as_mut().unwrap().recv() {
-                Message::DmaAwaitRemoteReply { timeout, error, channel, timestamp } => {
+                Message::DmaAwaitRemoteReply {
+                    timeout,
+                    error,
+                    channel,
+                    timestamp,
+                } => {
                     if timeout {
                         artiq_raise!(
                             "DMAError",

src/runtime/src/kernel/mod.rs

@@ -56,7 +56,7 @@ pub enum Message {
     #[cfg(has_drtio)]
     DmaStartRemoteRequest {
         id: i32,
-        timestamp: i64
+        timestamp: i64,
     },
     #[cfg(has_drtio)]
     DmaAwaitRemoteRequest(i32),
@@ -65,7 +65,7 @@ pub enum Message {
         timeout: bool,
         error: u8,
         channel: u32,
-        timestamp: u64
+        timestamp: u64,
     },
 
     #[cfg(has_drtio)]

src/runtime/src/main.rs

@@ -46,8 +46,8 @@ mod rtio;
 #[path = "rtio_acp.rs"]
 mod rtio;
 mod rtio_clocking;
-mod rtio_mgt;
 mod rtio_dma;
+mod rtio_mgt;
 
 static mut SEEN_ASYNC_ERRORS: u8 = 0;
 

src/runtime/src/moninj.rs

@@ -298,7 +298,11 @@ async fn handle_connection(
     }
 }
 
-pub fn start(timer: GlobalTimer, aux_mutex: &Rc<Mutex<bool>>, routing_table: &Rc<RefCell<drtio_routing::RoutingTable>>) {
+pub fn start(
+    timer: GlobalTimer,
+    aux_mutex: &Rc<Mutex<bool>>,
+    routing_table: &Rc<RefCell<drtio_routing::RoutingTable>>,
+) {
     let aux_mutex = aux_mutex.clone();
     let routing_table = routing_table.clone();
     task::spawn(async move {

src/runtime/src/rtio_dma.rs

@@ -1,12 +1,13 @@
-
+use alloc::{collections::BTreeMap, rc::Rc, string::String, vec::Vec};
-use alloc::{collections::BTreeMap, string::String, vec::Vec, rc::Rc};
-use libcortex_a9::{mutex::Mutex, cache::dcci_slice};
-use libboard_zynq::timer::GlobalTimer;
-use libboard_artiq::drtio_routing::RoutingTable;
-#[cfg(has_drtio)]
-use libasync::task;
 #[cfg(has_drtio)]
 use core::mem;
+
+#[cfg(has_drtio)]
+use libasync::task;
+use libboard_artiq::drtio_routing::RoutingTable;
+use libboard_zynq::timer::GlobalTimer;
+use libcortex_a9::{cache::dcci_slice, mutex::Mutex};
+
 use crate::kernel::DmaRecorder;
 
 const ALIGNMENT: usize = 16 * 8;
@@ -15,28 +16,29 @@ static DMA_RECORD_STORE: Mutex<BTreeMap<String, (u32, Vec<u8>, i64)>> = Mutex::n
 
 #[cfg(has_drtio)]
 pub mod remote_dma {
-    use super::*;
     use libboard_zynq::time::Milliseconds;
     use log::error;
+
+    use super::*;
     use crate::rtio_mgt::drtio;
 
     #[derive(Debug, PartialEq, Clone)]
     pub enum RemoteState {
         NotLoaded,
         Loaded,
-        PlaybackEnded { error: u8, channel: u32, timestamp: u64 }
+        PlaybackEnded { error: u8, channel: u32, timestamp: u64 },
     }
     #[derive(Debug, Clone)]
     struct RemoteTrace {
         trace: Vec<u8>,
-        pub state: RemoteState
+        pub state: RemoteState,
     }
 
     impl From<Vec<u8>> for RemoteTrace {
         fn from(trace: Vec<u8>) -> Self {
             RemoteTrace {
                 trace: trace,
-                state: RemoteState::NotLoaded
+                state: RemoteState::NotLoaded,
             }
         }
     }
@@ -51,7 +53,7 @@ pub mod remote_dma {
     struct TraceSet {
         id: u32,
         done_count: Mutex<usize>,
-        traces: Mutex<BTreeMap<u8, RemoteTrace>>
+        traces: Mutex<BTreeMap<u8, RemoteTrace>>,
     }
 
     impl TraceSet {
@@ -63,26 +65,27 @@ pub mod remote_dma {
             TraceSet {
                 id: id,
                 done_count: Mutex::new(0),
-                traces: Mutex::new(trace_map)
+                traces: Mutex::new(trace_map),
             }
         }
 
-        pub async fn await_done(
+        pub async fn await_done(&self, timeout: Option<u64>, timer: GlobalTimer) -> Result<RemoteState, &'static str> {
-            &self, 
-            timeout: Option<u64>, 
-            timer: GlobalTimer
-        ) -> Result<RemoteState, &'static str> {
             let timeout_ms = Milliseconds(timeout.unwrap_or(10_000));
             let limit = timer.get_time() + timeout_ms;
-            while (timer.get_time() < limit) & 
+            while (timer.get_time() < limit)
-                (*(self.done_count.async_lock().await) < self.traces.async_lock().await.len()) {
+                & (*(self.done_count.async_lock().await) < self.traces.async_lock().await.len())
+            {
                 task::r#yield().await;
             }
             if timer.get_time() >= limit {
                 error!("Remote DMA await done timed out");
                 return Err("Timed out waiting for results.");
             }
-            let mut playback_state: RemoteState = RemoteState::PlaybackEnded { error: 0, channel: 0, timestamp: 0 };
+            let mut playback_state: RemoteState = RemoteState::PlaybackEnded {
+                error: 0,
+                channel: 0,
+                timestamp: 0,
+            };
             let mut lock = self.traces.async_lock().await;
             let trace_iter = lock.iter_mut();
             for (_dest, trace) in trace_iter {
@@ -90,8 +93,12 @@ pub mod remote_dma {
                     RemoteState::PlaybackEnded {
                         error: e,
                         channel: _c,
-                        timestamp: _ts 
+                        timestamp: _ts,
-                    } => if e != 0 { playback_state = trace.state.clone(); },
+                    } => {
+                        if e != 0 {
+                            playback_state = trace.state.clone();
+                        }
+                    }
                     _ => (),
                 }
                 trace.state = RemoteState::Loaded;
@@ -103,7 +110,7 @@ pub mod remote_dma {
             &mut self,
             aux_mutex: &Rc<Mutex<bool>>,
             routing_table: &RoutingTable,
-            timer: GlobalTimer
+            timer: GlobalTimer,
         ) {
             let mut lock = self.traces.async_lock().await;
             let trace_iter = lock.iter_mut();
@@ -114,33 +121,24 @@ pub mod remote_dma {
                     timer,
                     self.id,
                     *destination,
-                        trace.get_trace()
+                    trace.get_trace(),
-                ).await {
+                )
+                .await
+                {
                     Ok(_) => trace.state = RemoteState::Loaded,
-                    Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e)
+                    Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
                 }
             }
             *(self.done_count.async_lock().await) = 0;
         }
 
-        pub async fn erase(
+        pub async fn erase(&mut self, aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, timer: GlobalTimer) {
-            &mut self, 
-            aux_mutex: &Rc<Mutex<bool>>, 
-            routing_table: &RoutingTable, 
-            timer: GlobalTimer
-        ) {
             let lock = self.traces.async_lock().await;
             let trace_iter = lock.keys();
             for destination in trace_iter {
-                match drtio::ddma_send_erase(
+                match drtio::ddma_send_erase(aux_mutex, routing_table, timer, self.id, *destination).await {
-                        aux_mutex, 
-                        routing_table,
-                        timer,
-                        self.id,
-                        *destination
-                ).await {
                     Ok(_) => (),
-                    Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e)
+                    Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
                 }
             }
         }
@@ -151,7 +149,7 @@ pub mod remote_dma {
             trace.state = RemoteState::PlaybackEnded {
                 error: error,
                 channel: channel,
-                timestamp: timestamp 
+                timestamp: timestamp,
             };
             *(self.done_count.async_lock().await) += 1;
         }
@@ -161,7 +159,7 @@ pub mod remote_dma {
             aux_mutex: &Rc<Mutex<bool>>,
             routing_table: &RoutingTable,
             timer: GlobalTimer,
-            timestamp: u64
+            timestamp: u64,
         ) {
             let mut dest_list: Vec<u8> = Vec::new();
             {
@@ -178,16 +176,10 @@ pub mod remote_dma {
             // mutex lock must be dropped before sending a playback request to avoid a deadlock,
             // if PlaybackStatus is sent from another satellite and the state must be updated.
             for destination in dest_list {
-                match drtio::ddma_send_playback(
+                match drtio::ddma_send_playback(aux_mutex, routing_table, timer, self.id, destination, timestamp).await
-                    aux_mutex, 
+                {
-                    routing_table,
-                    timer,
-                    self.id, 
-                    destination, 
-                    timestamp
-                ).await {
                     Ok(_) => (),
-                    Err(e) => error!("Error during remote DMA playback: {}", e)
+                    Err(e) => error!("Error during remote DMA playback: {}", e),
                 }
             }
         }
@@ -197,7 +189,8 @@ pub mod remote_dma {
             aux_mutex: &Rc<Mutex<bool>>,
             routing_table: &RoutingTable,
             timer: GlobalTimer,
-            destination: u8, up: bool
+            destination: u8,
+            up: bool,
         ) {
             // update state of the destination, resend traces if it's up
             if let Some(trace) = self.traces.lock().get_mut(&destination) {
@@ -208,10 +201,12 @@ pub mod remote_dma {
                         timer,
                         self.id,
                         destination,
-                        trace.get_trace()
+                        trace.get_trace(),
-                    ).await {
+                    )
+                    .await
+                    {
                         Ok(_) => trace.state = RemoteState::Loaded,
-                        Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e)
+                        Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
                     }
                 } else {
                     trace.state = RemoteState::NotLoaded;
@@ -226,32 +221,20 @@ pub mod remote_dma {
         unsafe { TRACES.insert(id, TraceSet::new(id, traces)) };
     }
 
-    pub async fn await_done(
+    pub async fn await_done(id: u32, timeout: Option<u64>, timer: GlobalTimer) -> Result<RemoteState, &'static str> {
-        id: u32,
-        timeout: Option<u64>,
-        timer: GlobalTimer
-    ) -> Result<RemoteState, &'static str> {
         let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
         trace_set.await_done(timeout, timer).await
     }
 
-    pub async fn erase(
+    pub async fn erase(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, timer: GlobalTimer, id: u32) {
-        aux_mutex: &Rc<Mutex<bool>>, 
-        routing_table: &RoutingTable, 
-        timer: GlobalTimer,
-        id: u32
-    ) {
         let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
         trace_set.erase(aux_mutex, routing_table, timer).await;
-        unsafe { TRACES.remove(&id); }
+        unsafe {
+            TRACES.remove(&id);
+        }
     }
 
-    pub async fn upload_traces(
+    pub async fn upload_traces(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, timer: GlobalTimer, id: u32) {
-        aux_mutex: &Rc<Mutex<bool>>, 
-        routing_table: &RoutingTable,
-        timer: GlobalTimer,
-        id: u32
-    ) {
         let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
         trace_set.upload_traces(aux_mutex, routing_table, timer).await;
     }
@@ -261,19 +244,13 @@ pub mod remote_dma {
         routing_table: &RoutingTable,
         timer: GlobalTimer,
         id: u32,
-        timestamp: u64
+        timestamp: u64,
     ) {
         let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
         trace_set.playback(aux_mutex, routing_table, timer, timestamp).await;
     }
 
-    pub async fn playback_done(
+    pub async fn playback_done(id: u32, destination: u8, error: u8, channel: u32, timestamp: u64) {
-        id: u32, 
-        destination: u8, 
-        error: u8, 
-        channel: u32, 
-        timestamp: u64
-    ) {
         let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
         trace_set.playback_done(destination, error, channel, timestamp).await;
     }
@@ -283,17 +260,19 @@ pub mod remote_dma {
         routing_table: &RoutingTable,
         timer: GlobalTimer,
         destination: u8,
-        up: bool
+        up: bool,
     ) {
         let trace_iter = unsafe { TRACES.values_mut() };
         for trace_set in trace_iter {
-            trace_set.destination_changed(aux_mutex, routing_table, timer, destination, up).await;
+            trace_set
+                .destination_changed(aux_mutex, routing_table, timer, destination, up)
+                .await;
         }
     }
 }
 
-
+pub async fn put_record(
-pub async fn put_record(_aux_mutex: &Rc<Mutex<bool>>, 
+    _aux_mutex: &Rc<Mutex<bool>>,
     _routing_table: &RoutingTable,
     _timer: GlobalTimer,
     mut recorder: DmaRecorder,
@@ -312,15 +291,14 @@ pub async fn put_record(_aux_mutex: &Rc<Mutex<bool>>,
         while recorder.buffer[ptr] != 0 {
             // ptr + 3 = tgt >> 24 (destination)
             let len = recorder.buffer[ptr] as usize;
-            let destination = recorder.buffer[ptr+3];
+            let destination = recorder.buffer[ptr + 3];
             if destination == 0 {
-                local_trace.extend(&recorder.buffer[ptr..ptr+len]);
+                local_trace.extend(&recorder.buffer[ptr..ptr + len]);
-            }
-            else {
-                if let Some(remote_trace) = remote_traces.get_mut(&destination) {
-                    remote_trace.extend(&recorder.buffer[ptr..ptr+len]);
             } else {
-                    remote_traces.insert(destination, recorder.buffer[ptr..ptr+len].to_vec());
+                if let Some(remote_trace) = remote_traces.get_mut(&destination) {
+                    remote_trace.extend(&recorder.buffer[ptr..ptr + len]);
+                } else {
+                    remote_traces.insert(destination, recorder.buffer[ptr..ptr + len].to_vec());
                 }
             }
             // and jump to the next event
@@ -357,9 +335,7 @@ pub async fn put_record(_aux_mutex: &Rc<Mutex<bool>>,
     ptr
 }
 
-pub async fn erase(name: String, _aux_mutex: &Rc<Mutex<bool>>, 
+pub async fn erase(name: String, _aux_mutex: &Rc<Mutex<bool>>, _routing_table: &RoutingTable, _timer: GlobalTimer) {
-    _routing_table: &RoutingTable, _timer: GlobalTimer
-) {
     let _entry = DMA_RECORD_STORE.lock().remove(&name);
     #[cfg(has_drtio)]
     if let Some((id, _v, _d)) = _entry {

src/runtime/src/rtio_mgt.rs

@@ -12,17 +12,17 @@ static mut RTIO_DEVICE_MAP: BTreeMap<u32, String> = BTreeMap::new();
 
 #[cfg(has_drtio)]
 pub mod drtio {
-    use embedded_hal::blocking::delay::DelayMs;
     use alloc::vec::Vec;
+
+    use embedded_hal::blocking::delay::DelayMs;
     use libasync::{delay, task};
-    use libboard_artiq::{drtioaux::Error, drtioaux_async, drtioaux_async::Packet};
+    use libboard_artiq::{drtioaux::Error, drtioaux_async, drtioaux_async::Packet, drtioaux_proto::DMA_TRACE_MAX_SIZE};
-    use libboard_artiq::drtioaux_proto::DMA_TRACE_MAX_SIZE;
     use libboard_zynq::time::Milliseconds;
     use log::{error, info, warn};
 
     use super::*;
-    use crate::{ASYNC_ERROR_BUSY, ASYNC_ERROR_COLLISION, ASYNC_ERROR_SEQUENCE_ERROR, SEEN_ASYNC_ERRORS};
+    use crate::{rtio_dma::remote_dma, ASYNC_ERROR_BUSY, ASYNC_ERROR_COLLISION, ASYNC_ERROR_SEQUENCE_ERROR,
-    use crate::rtio_dma::remote_dma;
+                SEEN_ASYNC_ERRORS};
 
     pub fn startup(
         aux_mutex: &Rc<Mutex<bool>>,
@@ -69,11 +69,17 @@ pub mod drtio {
         loop {
             let reply = recv_aux_timeout(linkno, 200, timer).await;
             match reply {
-                Ok(Packet::DmaPlaybackStatus { id, destination, error, channel, timestamp }) => {
+                Ok(Packet::DmaPlaybackStatus {
+                    id,
+                    destination,
+                    error,
+                    channel,
+                    timestamp,
+                }) => {
                     remote_dma::playback_done(id, destination, error, channel, timestamp).await;
-                },
+                }
                 Ok(packet) => return Ok(packet),
-                Err(e) => return Err(e)
+                Err(e) => return Err(e),
             }
         }
     }
@@ -184,8 +190,13 @@ pub mod drtio {
     async fn process_unsolicited_aux(aux_mutex: &Rc<Mutex<bool>>, linkno: u8) {
         let _lock = aux_mutex.async_lock().await;
         match drtioaux_async::recv(linkno).await {
-            Ok(Some(Packet::DmaPlaybackStatus { id, destination, error, channel, timestamp })
+            Ok(Some(Packet::DmaPlaybackStatus {
-            ) => remote_dma::playback_done(id, destination, error, channel, timestamp).await,
+                id,
+                destination,
+                error,
+                channel,
+                timestamp,
+            })) => remote_dma::playback_done(id, destination, error, channel, timestamp).await,
             Ok(Some(packet)) => warn!("[LINK#{}] unsolicited aux packet: {:?}", linkno, packet),
             Ok(None) => (),
             Err(_) => warn!("[LINK#{}] aux packet error", linkno),
@@ -267,7 +278,8 @@ pub mod drtio {
                         match reply {
                             Ok(Packet::DestinationDownReply) => {
                                 destination_set_up(routing_table, up_destinations, destination, false).await;
-                                remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, false).await;
+                                remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, false)
+                                    .await;
                             }
                             Ok(Packet::DestinationOkReply) => (),
                             Ok(Packet::DestinationSequenceErrorReply { channel }) => {
@@ -320,7 +332,8 @@ pub mod drtio {
                             Ok(Packet::DestinationOkReply) => {
                                 destination_set_up(routing_table, up_destinations, destination, true).await;
                                 init_buffer_space(destination as u8, linkno).await;
-                                remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, true).await;
+                                remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, true)
+                                    .await;
                             }
                             Ok(packet) => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet),
                             Err(e) => error!("[DEST#{}] communication failed ({})", destination, e),
@@ -413,48 +426,72 @@ pub mod drtio {
         timer: GlobalTimer,
         id: u32,
         destination: u8,
-        trace: &Vec<u8>
+        trace: &Vec<u8>,
     ) -> Result<(), &'static str> {
         let linkno = routing_table.0[destination as usize][0] - 1;
         let mut i = 0;
         while i < trace.len() {
             let mut trace_slice: [u8; DMA_TRACE_MAX_SIZE] = [0; DMA_TRACE_MAX_SIZE];
-            let len: usize = if i + DMA_TRACE_MAX_SIZE < trace.len() { DMA_TRACE_MAX_SIZE } else { trace.len() - i } as usize;
+            let len: usize = if i + DMA_TRACE_MAX_SIZE < trace.len() {
+                DMA_TRACE_MAX_SIZE
+            } else {
+                trace.len() - i
+            } as usize;
             let last = i + len == trace.len();
-            trace_slice[..len].clone_from_slice(&trace[i..i+len]);
+            trace_slice[..len].clone_from_slice(&trace[i..i + len]);
             i += len;
-            let reply = aux_transact(aux_mutex, linkno, 
+            let reply = aux_transact(
+                aux_mutex,
+                linkno,
                 &Packet::DmaAddTraceRequest {
-                    id: id, destination: destination, last: last, length: len as u16, trace: trace_slice},
+                    id: id,
-                timer).await;
+                    destination: destination,
+                    last: last,
+                    length: len as u16,
+                    trace: trace_slice,
+                },
+                timer,
+            )
+            .await;
             match reply {
                 Ok(Packet::DmaAddTraceReply { succeeded: true }) => (),
                 Ok(Packet::DmaAddTraceReply { succeeded: false }) => {
-                    return Err("error adding trace on satellite"); },
+                    return Err("error adding trace on satellite");
-                Ok(_) => { return Err("adding DMA trace failed, unexpected aux packet"); },
+                }
-                Err(_) => { return Err("adding DMA trace failed, aux error"); }
+                Ok(_) => {
+                    return Err("adding DMA trace failed, unexpected aux packet");
+                }
+                Err(_) => {
+                    return Err("adding DMA trace failed, aux error");
+                }
             }
         }
         Ok(())
     }
 
-
     pub async fn ddma_send_erase(
         aux_mutex: &Rc<Mutex<bool>>,
         routing_table: &drtio_routing::RoutingTable,
         timer: GlobalTimer,
         id: u32,
-        destination: u8
+        destination: u8,
     ) -> Result<(), &'static str> {
         let linkno = routing_table.0[destination as usize][0] - 1;
-        let reply = aux_transact(aux_mutex, linkno, 
+        let reply = aux_transact(
-            &Packet::DmaRemoveTraceRequest { id: id, destination: destination },
+            aux_mutex,
-            timer).await;
+            linkno,
+            &Packet::DmaRemoveTraceRequest {
+                id: id,
+                destination: destination,
+            },
+            timer,
+        )
+        .await;
         match reply {
             Ok(Packet::DmaRemoveTraceReply { succeeded: true }) => Ok(()),
             Ok(Packet::DmaRemoveTraceReply { succeeded: false }) => Err("satellite DMA erase error"),
             Ok(_) => Err("adding trace failed, unexpected aux packet"),
-            Err(_) => Err("erasing trace failed, aux error")
+            Err(_) => Err("erasing trace failed, aux error"),
         }
     }
 
@@ -464,19 +501,27 @@ pub mod drtio {
         timer: GlobalTimer,
         id: u32,
         destination: u8,
-        timestamp: u64
+        timestamp: u64,
     ) -> Result<(), &'static str> {
         let linkno = routing_table.0[destination as usize][0] - 1;
-        let reply = aux_transact(aux_mutex, linkno, &Packet::DmaPlaybackRequest{
+        let reply = aux_transact(
-            id: id, destination: destination, timestamp: timestamp }, timer).await;
+            aux_mutex,
+            linkno,
+            &Packet::DmaPlaybackRequest {
+                id: id,
+                destination: destination,
+                timestamp: timestamp,
+            },
+            timer,
+        )
+        .await;
         match reply {
             Ok(Packet::DmaPlaybackReply { succeeded: true }) => Ok(()),
             Ok(Packet::DmaPlaybackReply { succeeded: false }) => Err("error on DMA playback request"),
             Ok(_) => Err("received unexpected aux packet during DMA playback"),
-            Err(_) => Err("aux error on DMA playback")
+            Err(_) => Err("aux error on DMA playback"),
         }
     }
-
 }
 
 fn read_device_map(cfg: &Config) -> BTreeMap<u32, String> {

src/satman/src/dma.rs

@@ -1,49 +1,48 @@
-use libcortex_a9::cache::dcci_slice;
+use alloc::{collections::btree_map::BTreeMap, vec::Vec};
+
 use libboard_artiq::pl::csr;
-use alloc::{vec::Vec, collections::btree_map::BTreeMap};
+use libcortex_a9::cache::dcci_slice;
 
 const ALIGNMENT: usize = 64;
 
 #[derive(Debug, PartialEq)]
 enum ManagerState {
     Idle,
-    Playback
+    Playback,
 }
 
 pub struct RtioStatus {
     pub id: u32,
     pub error: u8,
     pub channel: u32,
-    pub timestamp: u64
+    pub timestamp: u64,
 }
 
 pub enum Error {
     IdNotFound,
     PlaybackInProgress,
-    EntryNotComplete
+    EntryNotComplete,
 }
 
 #[derive(Debug)]
 struct Entry {
     trace: Vec<u8>,
     padding_len: usize,
-    complete: bool
+    complete: bool,
 }
 
 #[derive(Debug)]
 pub struct Manager {
     entries: BTreeMap<u32, Entry>,
     state: ManagerState,
-    currentid: u32
+    currentid: u32,
 }
 
 impl Manager {
     pub fn new() -> Manager {
         // in case Manager is created during a DMA in progress
         // wait for it to end
-        unsafe {
+        unsafe { while csr::rtio_dma::enable_read() != 0 {} }
-            while csr::rtio_dma::enable_read() != 0 {} 
-        }
         Manager {
             entries: BTreeMap::new(),
             currentid: 0,
@@ -57,22 +56,30 @@ impl Manager {
                 if entry.complete {
                     // replace entry
                     self.entries.remove(&id);
-                    self.entries.insert(id, Entry {
+                    self.entries.insert(
+                        id,
+                        Entry {
                             trace: Vec::new(),
                             padding_len: 0,
-                        complete: false });
+                            complete: false,
+                        },
+                    );
                     self.entries.get_mut(&id).unwrap()
                 } else {
                     entry
                 }
-            },
+            }
             None => {
-                self.entries.insert(id, Entry {
+                self.entries.insert(
+                    id,
+                    Entry {
                         trace: Vec::new(),
                         padding_len: 0,
-                    complete: false });
+                        complete: false,
-                self.entries.get_mut(&id).unwrap()
                     },
+                );
+                self.entries.get_mut(&id).unwrap()
+            }
         };
         entry.trace.extend(&trace[0..trace_len]);
 
@@ -101,7 +108,7 @@ impl Manager {
     pub fn erase(&mut self, id: u32) -> Result<(), Error> {
         match self.entries.remove(&id) {
             Some(_) => Ok(()),
-            None => Err(Error::IdNotFound)
+            None => Err(Error::IdNotFound),
         }
     }
 
@@ -110,9 +117,11 @@ impl Manager {
             return Err(Error::PlaybackInProgress);
         }
 
-        let entry = match self.entries.get(&id){
+        let entry = match self.entries.get(&id) {
             Some(entry) => entry,
-            None => { return Err(Error::IdNotFound); }
+            None => {
+                return Err(Error::IdNotFound);
+            }
         };
         if !entry.complete {
             return Err(Error::EntryNotComplete);
@@ -156,9 +165,9 @@ impl Manager {
                     id: self.currentid,
                     error: error,
                     channel: channel,
-                    timestamp: timestamp });
+                    timestamp: timestamp,
+                });
             }
         }
     }
-
 }

src/satman/src/main.rs

@@ -20,6 +20,7 @@ extern crate alloc;
 
 use core::sync::atomic::{AtomicBool, Ordering};
 
+use dma::Manager as DmaManager;
 use embedded_hal::blocking::delay::DelayUs;
 #[cfg(feature = "target_kasli_soc")]
 use libboard_artiq::io_expander;
@@ -36,10 +37,9 @@ use libcortex_a9::{asm, interrupt_handler,
                    spin_lock_yield};
 use libregister::{RegisterR, RegisterW};
 use libsupport_zynq::ram;
-use dma::Manager as DmaManager;
 
-mod repeater;
 mod dma;
+mod repeater;
 
 fn drtiosat_reset(reset: bool) {
     unsafe {
@@ -94,7 +94,7 @@ fn process_aux_packet(
     packet: drtioaux::Packet,
     timer: &mut GlobalTimer,
     i2c: &mut I2c,
-    dma_manager: &mut DmaManager
+    dma_manager: &mut DmaManager,
 ) -> 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.
@@ -417,31 +417,28 @@ fn process_aux_packet(
             id,
             last,
             length,
-            trace 
+            trace,
         } => {
             forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
             let succeeded = dma_manager.add(id, last, &trace, length as usize).is_ok();
-            drtioaux::send(0,
+            drtioaux::send(0, &drtioaux::Packet::DmaAddTraceReply { succeeded: succeeded })
-                &drtioaux::Packet::DmaAddTraceReply { succeeded: succeeded })
         }
         drtioaux::Packet::DmaRemoveTraceRequest {
             destination: _destination,
-            id 
+            id,
         } => {
             forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
             let succeeded = dma_manager.erase(id).is_ok();
-            drtioaux::send(0,
+            drtioaux::send(0, &drtioaux::Packet::DmaRemoveTraceReply { succeeded: succeeded })
-                &drtioaux::Packet::DmaRemoveTraceReply { succeeded: succeeded })
         }
         drtioaux::Packet::DmaPlaybackRequest {
             destination: _destination,
             id,
-            timestamp 
+            timestamp,
         } => {
             forward!(_routing_table, _destination, *_rank, _repeaters, &packet, timer);
             let succeeded = dma_manager.playback(id, timestamp).is_ok();
-            drtioaux::send(0,
+            drtioaux::send(0, &drtioaux::Packet::DmaPlaybackReply { succeeded: succeeded })
-                &drtioaux::Packet::DmaPlaybackReply { succeeded: succeeded })
         }
 
         _ => {
@@ -457,7 +454,7 @@ fn process_aux_packets(
     rank: &mut u8,
     timer: &mut GlobalTimer,
     i2c: &mut I2c,
-    dma_manager: &mut DmaManager
+    dma_manager: &mut DmaManager,
 ) {
     let result = drtioaux::recv(0).and_then(|packet| {
         if let Some(packet) = packet {
@@ -645,7 +642,14 @@ pub extern "C" fn main_core0() -> i32 {
 
         while drtiosat_link_rx_up() {
             drtiosat_process_errors();
-            process_aux_packets(&mut repeaters, &mut routing_table, &mut rank, &mut timer, &mut i2c, &mut dma_manager);
+            process_aux_packets(
+                &mut repeaters,
+                &mut routing_table,
+                &mut rank,
+                &mut timer,
+                &mut i2c,
+                &mut dma_manager,
+            );
             #[allow(unused_mut)]
             for mut rep in repeaters.iter_mut() {
                 rep.service(&routing_table, rank, &mut timer);
@@ -663,14 +667,20 @@ pub extern "C" fn main_core0() -> i32 {
                 }
             }
             if let Some(status) = dma_manager.check_state() {
-                info!("playback done, error: {}, channel: {}, timestamp: {}", status.error, status.channel, status.timestamp);
+                info!(
-                if let Err(e) = drtioaux::send(0, &drtioaux::Packet::DmaPlaybackStatus { 
+                    "playback done, error: {}, channel: {}, timestamp: {}",
+                    status.error, status.channel, status.timestamp
+                );
+                if let Err(e) = drtioaux::send(
+                    0,
+                    &drtioaux::Packet::DmaPlaybackStatus {
                         destination: rank,
                         id: status.id,
                         error: status.error,
                         channel: status.channel,
-                    timestamp: status.timestamp 
+                        timestamp: status.timestamp,
-                }) {
+                    },
+                ) {
                     error!("error sending DMA playback status: {:?}", e);
                 }
             }