cargo fmt

Co-authored-by: mwojcik <mw@m-labs.hk>
Co-committed-by: mwojcik <mw@m-labs.hk>

README.md

@@ -4,8 +4,8 @@ ARTIQ on Zynq
 How to use
 ----------
 
-1. Install ARTIQ-7 or newer.
+1. Install the ARTIQ version that corresponds to the artiq-zynq version are targeting.
-2. Select the latest successful build on Hydra: https://nixbld.m-labs.hk/jobset/artiq/zynq
+2. Select the latest successful build on [Hydra](https://nixbld.m-labs.hk/) for the targeted artiq-zynq version, or use AFWS to obtain firmware binaries.
 3. Search for the job named ``<board>-<variant>-sd`` (for example: ``zc706-nist_clock-sd`` or ``zc706-nist_qc2-sd``).
 4. Download the ``boot.bin`` "binary distribution" and place it at the root of a FAT-formatted SD card.
 5. Optionally, create a ``config.txt`` configuration file at the root of the SD card containing ``key=value`` pairs on each line. Use the ``ip``, ``ip6`` and ``mac`` keys to respectively set the IPv4, IPv6 and MAC address of the board. Configuring an IPv6 address is entirely optional. If these keys are not found, the firmware will use default values that may or may not be compatible with your network.

examples/device_db.py

@@ -59,6 +59,14 @@ device_db["ad9914dds1"] = {
     "arguments": {"sysclk": 3e9, "bus_channel": 50, "channel": 1},
 }
 
+for i in range(4):
+    device_db["ttl"+str(i)+"_counter"] = {
+        "type": "local",
+        "module": "artiq.coredevice.edge_counter",
+        "class": "EdgeCounter",
+        "arguments": {"channel": 52+i}
+    }
+
 # for ARTIQ test suite
 device_db.update(
     loop_out="ttl0",

flake.lock

@@ -11,11 +11,11 @@
         "src-pythonparser": "src-pythonparser"
       },
       "locked": {
-        "lastModified": 1677033865,
+        "lastModified": 1680398257,
-        "narHash": "sha256-9w9V+B6vMl4I2uX5k6lJc1FXhf4PeTCZXrq5Tdq2qHc=",
+        "narHash": "sha256-ANODm+Xjx/OYHkmQpybDuIPXiu22IkoQ0wdI0ReQsNk=",
         "ref": "refs/heads/master",
-        "rev": "d0437f5672b7dd5f898a73469d730bc46f058ecc",
+        "rev": "7ba06bfe61cad4ad41f478188917cac6571a6875",
-        "revCount": 8310,
+        "revCount": 8323,
         "type": "git",
         "url": "https://github.com/m-labs/artiq.git"
       },
@@ -84,11 +84,11 @@
     "mozilla-overlay_2": {
       "flake": false,
       "locked": {
-        "lastModified": 1675354105,
+        "lastModified": 1677493379,
-        "narHash": "sha256-ZAJGIZ7TjOCU7302lSUabNDz+rxM4If0l8/ZbE/7R5U=",
+        "narHash": "sha256-A1gO8zlWLv3+tZ3cGVB1WYvvoN9pbFyv0xIJHcTsckw=",
         "owner": "mozilla",
         "repo": "nixpkgs-mozilla",
-        "rev": "85eb0ba7d8e5d6d4b79e5b0180aadbdd25d76404",
+        "rev": "78e723925daf5c9e8d0a1837ec27059e61649cb6",
         "type": "github"
       },
       "original": {

src/gateware/zc706.py

@@ -14,7 +14,7 @@ from misoc.integration import cpu_interface
 from misoc.cores import gpio
 
 from artiq.gateware import rtio, nist_clock, nist_qc2
-from artiq.gateware.rtio.phy import ttl_simple, ttl_serdes_7series, dds, spi2
+from artiq.gateware.rtio.phy import ttl_simple, ttl_serdes_7series, dds, spi2, edge_counter
 from artiq.gateware.rtio.xilinx_clocking import fix_serdes_timing_path
 from artiq.gateware.drtio.transceiver import gtx_7series
 from artiq.gateware.drtio.siphaser import SiPhaser7Series
@@ -577,12 +577,16 @@ class _NIST_QC2_RTIO:
         platform.add_extension(pmod1_33)
 
         rtio_channels = []
+        edge_counter_phy = []
 
         # All TTL channels are In+Out capable
         for i in range(40):
             phy = ttl_serdes_7series.InOut_8X(platform.request("ttl", i))
             self.submodules += phy
             rtio_channels.append(rtio.Channel.from_phy(phy, ififo_depth=512))
+            # first four TTLs will also have edge counters
+            if i < 4:
+                edge_counter_phy.append(phy)
 
         # no SMA GPIO, replaced with PMOD1_0
         phy = ttl_serdes_7series.InOut_8X(platform.request("pmod1_33", 0))
@@ -621,6 +625,11 @@ class _NIST_QC2_RTIO:
                 platform.request("dds", backplane_offset), 12, onehot=True)
             self.submodules += phy
             rtio_channels.append(rtio.Channel.from_phy(phy, ififo_depth=4))
+        
+        for phy in edge_counter_phy:
+            counter = edge_counter.SimpleEdgeCounter(phy.input_state)
+            self.submodules += counter
+            rtio_channels.append(rtio.Channel.from_phy(counter))
 
         self.config["RTIO_LOG_CHANNEL"] = len(rtio_channels)
         rtio_channels.append(rtio.LogChannel())

src/libboard_artiq/src/drtioaux_proto.rs

@@ -136,37 +136,37 @@ pub enum Packet {
         succeeded: bool,
     },
 
-    DmaAddTraceRequest { 
+    DmaAddTraceRequest {
-        destination: u8, 
+        destination: u8,
-        id: u32, 
+        id: u32,
-        last: bool, 
+        last: bool,
-        length: u16, 
+        length: u16,
-        trace: [u8; DMA_TRACE_MAX_SIZE] 
+        trace: [u8; DMA_TRACE_MAX_SIZE],
     },
-    DmaAddTraceReply { 
+    DmaAddTraceReply {
-        succeeded: bool 
+        succeeded: bool,
     },
-    DmaRemoveTraceRequest { 
+    DmaRemoveTraceRequest {
-        destination: u8, 
+        destination: u8,
-        id: u32 
+        id: u32,
     },
-    DmaRemoveTraceReply { 
+    DmaRemoveTraceReply {
-        succeeded: bool 
+        succeeded: bool,
     },
-    DmaPlaybackRequest { 
+    DmaPlaybackRequest {
-        destination: u8, 
+        destination: u8,
-        id: u32, 
+        id: u32,
-        timestamp: u64 
+        timestamp: u64,
     },
-    DmaPlaybackReply { 
+    DmaPlaybackReply {
-        succeeded: bool 
+        succeeded: bool,
     },
-    DmaPlaybackStatus { 
+    DmaPlaybackStatus {
-        destination: u8, 
+        destination: u8,
-        id: u32, 
+        id: u32,
-        error: u8, 
+        error: u8,
-        channel: u32, 
+        channel: u32,
-        timestamp: u64 
+        timestamp: u64,
     },
 }
 
@@ -298,7 +298,7 @@ impl Packet {
                 succeeded: reader.read_bool()?,
             },
 
-            0xb0 => { 
+            0xb0 => {
                 let destination = reader.read_u8()?;
                 let id = reader.read_u32()?;
                 let last = reader.read_bool()?;
@@ -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)),
@@ -526,12 +526,12 @@ impl Packet {
                 writer.write_bool(succeeded)?;
             }
 
-            Packet::DmaAddTraceRequest { 
+            Packet::DmaAddTraceRequest {
-                destination, 
+                destination,
-                id, 
+                id,
-                last, 
+                last,
-                trace, 
+                trace,
-                length 
+                length,
             } => {
                 writer.write_u8(0xb0)?;
                 writer.write_u8(destination)?;
@@ -555,10 +555,10 @@ impl Packet {
                 writer.write_u8(0xb3)?;
                 writer.write_bool(succeeded)?;
             }
-            Packet::DmaPlaybackRequest { 
+            Packet::DmaPlaybackRequest {
-                destination, 
+                destination,
-                id, 
+                id,
-                timestamp 
+                timestamp,
             } => {
                 writer.write_u8(0xb4)?;
                 writer.write_u8(destination)?;
@@ -569,12 +569,12 @@ impl Packet {
                 writer.write_u8(0xb5)?;
                 writer.write_bool(succeeded)?;
             }
-            Packet::DmaPlaybackStatus { 
+            Packet::DmaPlaybackStatus {
-                destination, 
+                destination,
-                id, 
+                id,
-                error, 
+                error,
-                channel, 
+                channel,
-                timestamp 
+                timestamp,
             } => {
                 writer.write_u8(0xb6)?;
                 writer.write_u8(destination)?;

src/runtime/src/comms.rs

@@ -26,7 +26,7 @@ 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}};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -79,7 +79,6 @@ enum Reply {
 }
 
 static CACHE_STORE: Mutex<BTreeMap<String, Vec<i32>>> = Mutex::new(BTreeMap::new());
-static DMA_RECORD_STORE: Mutex<BTreeMap<String, (Vec<u8>, i64)>> = Mutex::new(BTreeMap::new());
 
 async fn write_header(stream: &TcpStream, reply: Reply) -> Result<()> {
     stream
@@ -157,6 +156,9 @@ async fn handle_run_kernel(
     stream: Option<&TcpStream>,
     control: &Rc<RefCell<kernel::Control>>,
     _up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+    aux_mutex: &Rc<Mutex<bool>>,
+    routing_table: &drtio_routing::RoutingTable,
+    timer: GlobalTimer,
 ) -> Result<()> {
     control.borrow_mut().tx.async_send(kernel::Message::StartRequest).await;
     loop {
@@ -319,16 +321,16 @@ async fn handle_run_kernel(
                     .await;
             }
             kernel::Message::DmaPutRequest(recorder) => {
-                DMA_RECORD_STORE
+                let _id = rtio_dma::put_record(aux_mutex, routing_table, timer, recorder).await;
-                    .lock()
+                #[cfg(has_drtio)]
-                    .insert(recorder.name, (recorder.buffer, recorder.duration));
+                rtio_dma::remote_dma::upload_traces(aux_mutex, routing_table, timer, _id).await;
             }
             kernel::Message::DmaEraseRequest(name) => {
                 // prevent possible OOM when we have large DMA record replacement.
-                DMA_RECORD_STORE.lock().remove(&name);
+                rtio_dma::erase(name, aux_mutex, routing_table, timer).await;
             }
             kernel::Message::DmaGetRequest(name) => {
-                let result = DMA_RECORD_STORE.lock().get(&name).map(|v| v.clone());
+                let result = rtio_dma::retrieve(name);
                 control
                     .borrow_mut()
                     .tx
@@ -336,6 +338,33 @@ async fn handle_run_kernel(
                     .await;
             }
             #[cfg(has_drtio)]
+            kernel::Message::DmaStartRemoteRequest { id, timestamp } => {
+                rtio_dma::remote_dma::playback(aux_mutex, routing_table, timer, id as u32, timestamp as u64).await;
+            }
+            #[cfg(has_drtio)]
+            kernel::Message::DmaAwaitRemoteRequest(id) => {
+                let result = rtio_dma::remote_dma::await_done(id as u32, Some(10_000), timer).await;
+                let reply = match result {
+                    Ok(rtio_dma::remote_dma::RemoteState::PlaybackEnded {
+                        error,
+                        channel,
+                        timestamp,
+                    }) => kernel::Message::DmaAwaitRemoteReply {
+                        timeout: false,
+                        error: error,
+                        channel: channel,
+                        timestamp: timestamp,
+                    },
+                    _ => kernel::Message::DmaAwaitRemoteReply {
+                        timeout: true,
+                        error: 0,
+                        channel: 0,
+                        timestamp: 0,
+                    },
+                };
+                control.borrow_mut().tx.async_send(reply).await;
+            }
+            #[cfg(has_drtio)]
             kernel::Message::UpDestinationsRequest(destination) => {
                 let result = _up_destinations.borrow()[destination as usize];
                 control
@@ -395,6 +424,9 @@ async fn handle_connection(
     stream: &mut TcpStream,
     control: Rc<RefCell<kernel::Control>>,
     up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+    aux_mutex: &Rc<Mutex<bool>>,
+    routing_table: &drtio_routing::RoutingTable,
+    timer: GlobalTimer,
 ) -> Result<()> {
     stream.set_ack_delay(None);
 
@@ -418,7 +450,15 @@ async fn handle_connection(
                 load_kernel(&buffer, &control, Some(stream)).await?;
             }
             Request::RunKernel => {
-                handle_run_kernel(Some(stream), &control, &up_destinations).await?;
+                handle_run_kernel(
+                    Some(stream),
+                    &control,
+                    &up_destinations,
+                    aux_mutex,
+                    routing_table,
+                    timer,
+                )
+                .await?;
             }
             _ => {
                 error!("unexpected request from host: {:?}", request);
@@ -485,7 +525,7 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
     rtio_mgt::startup(&aux_mutex, &drtio_routing_table, &up_destinations, timer, &cfg);
 
     analyzer::start();
-    moninj::start(timer, aux_mutex, drtio_routing_table);
+    moninj::start(timer, &aux_mutex, &drtio_routing_table);
 
     let control: Rc<RefCell<kernel::Control>> = Rc::new(RefCell::new(kernel::Control::start()));
     let idle_kernel = Rc::new(cfg.read("idle_kernel").ok());
@@ -493,7 +533,15 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
         info!("Loading startup kernel...");
         if let Ok(()) = task::block_on(load_kernel(&buffer, &control, None)) {
             info!("Starting startup kernel...");
-            let _ = task::block_on(handle_run_kernel(None, &control, &up_destinations));
+            let routing_table = drtio_routing_table.borrow();
+            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!");
@@ -519,13 +567,16 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
             let connection = connection.clone();
             let terminate = terminate.clone();
             let up_destinations = up_destinations.clone();
+            let aux_mutex = aux_mutex.clone();
+            let routing_table = drtio_routing_table.clone();
 
             // we make sure the value of terminate is 0 before we start
             let _ = terminate.try_wait();
             task::spawn(async move {
+                let routing_table = routing_table.borrow();
                 select_biased! {
                     _ = (async {
-                        let _ = handle_connection(&mut stream, control.clone(), &up_destinations)
+                        let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
                             .await
                             .map_err(|e| warn!("connection terminated: {}", e));
                         if let Some(buffer) = &*idle_kernel {
@@ -533,7 +584,7 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
                             let _ = load_kernel(&buffer, &control, None)
                                 .await.map_err(|_| warn!("error loading idle kernel"));
                             info!("Running idle kernel");
-                            let _ = handle_run_kernel(None, &control, &up_destinations)
+                            let _ = handle_run_kernel(None, &control, &up_destinations, &aux_mutex, &routing_table, timer)
                                 .await.map_err(|_| warn!("error running idle kernel"));
                             info!("Idle kernel terminated");
                         }

src/runtime/src/kernel/dma.rs

@@ -1,14 +1,11 @@
-use alloc::{boxed::Box, string::String, vec::Vec};
+use alloc::{string::String, vec::Vec};
 use core::mem;
 
 use cslice::CSlice;
-use libcortex_a9::cache::dcci_slice;
 
 use super::{Message, KERNEL_CHANNEL_0TO1, KERNEL_CHANNEL_1TO0, KERNEL_IMAGE};
 use crate::{artiq_raise, pl::csr, rtio};
 
-const ALIGNMENT: usize = 16 * 8;
-
 #[repr(C)]
 pub struct DmaTrace {
     duration: i64,
@@ -20,6 +17,7 @@ pub struct DmaRecorder {
     pub name: String,
     pub buffer: Vec<u8>,
     pub duration: i64,
+    pub enable_ddma: bool,
 }
 
 static mut RECORDER: Option<DmaRecorder> = None;
@@ -53,11 +51,12 @@ pub extern "C" fn dma_record_start(name: CSlice<u8>) {
             name,
             buffer: Vec::new(),
             duration: 0,
+            enable_ddma: false,
         });
     }
 }
 
-pub extern "C" fn dma_record_stop(duration: i64) {
+pub extern "C" fn dma_record_stop(duration: i64, enable_ddma: bool) {
     unsafe {
         if RECORDER.is_none() {
             artiq_raise!("DMAError", "DMA is not recording")
@@ -71,6 +70,7 @@ pub extern "C" fn dma_record_stop(duration: i64) {
 
         let mut recorder = RECORDER.take().unwrap();
         recorder.duration = duration;
+        recorder.enable_ddma = enable_ddma;
         KERNEL_CHANNEL_1TO0
             .as_mut()
             .unwrap()
@@ -151,20 +151,7 @@ pub extern "C" fn dma_retrieve(name: CSlice<u8>) -> DmaTrace {
     }
     match unsafe { KERNEL_CHANNEL_0TO1.as_mut().unwrap() }.recv() {
         Message::DmaGetReply(None) => (),
-        Message::DmaGetReply(Some((mut v, duration))) => {
+        Message::DmaGetReply(Some((address, duration))) => {
-            v.reserve(ALIGNMENT - 1);
-            let original_length = v.len();
-            let padding = ALIGNMENT - v.as_ptr() as usize % ALIGNMENT;
-            let padding = if padding == ALIGNMENT { 0 } else { padding };
-            for _ in 0..padding {
-                v.push(0);
-            }
-            // trailing zero to indicate end of buffer
-            v.push(0);
-            v.copy_within(0..original_length, padding);
-            dcci_slice(&v);
-            let v = Box::new(v);
-            let address = Box::into_raw(v) as *mut Vec<u8> as i32;
             return DmaTrace { address, duration };
         }
         _ => panic!("Expected DmaGetReply after DmaGetRequest!"),
@@ -175,22 +162,22 @@ pub extern "C" fn dma_retrieve(name: CSlice<u8>) -> DmaTrace {
 
 pub extern "C" fn dma_playback(timestamp: i64, ptr: i32) {
     unsafe {
-        let v = Box::from_raw(ptr as *mut Vec<u8>);
-        let padding = ALIGNMENT - v.as_ptr() as usize % ALIGNMENT;
-        let padding = if padding == ALIGNMENT { 0 } else { padding };
-        let ptr = v.as_ptr().add(padding) as i32;
-
         csr::rtio_dma::base_address_write(ptr as u32);
         csr::rtio_dma::time_offset_write(timestamp as u64);
 
         csr::cri_con::selected_write(1);
         csr::rtio_dma::enable_write(1);
+        #[cfg(has_drtio)]
+        KERNEL_CHANNEL_1TO0
+            .as_mut()
+            .unwrap()
+            .send(Message::DmaStartRemoteRequest {
+                id: ptr,
+                timestamp: timestamp,
+            });
         while csr::rtio_dma::enable_read() != 0 {}
         csr::cri_con::selected_write(0);
 
-        // leave the handle as we may try to do playback for another time.
-        mem::forget(v);
-
         let error = csr::rtio_dma::error_read();
         if error != 0 {
             let timestamp = csr::rtio_dma::error_timestamp_read();
@@ -215,5 +202,46 @@ pub extern "C" fn dma_playback(timestamp: i64, ptr: i32) {
                 );
             }
         }
+        #[cfg(has_drtio)]
+        {
+            KERNEL_CHANNEL_1TO0
+                .as_mut()
+                .unwrap()
+                .send(Message::DmaAwaitRemoteRequest(ptr));
+            match KERNEL_CHANNEL_0TO1.as_mut().unwrap().recv() {
+                Message::DmaAwaitRemoteReply {
+                    timeout,
+                    error,
+                    channel,
+                    timestamp,
+                } => {
+                    if timeout {
+                        artiq_raise!(
+                            "DMAError",
+                            "Error running DMA on satellite device, timed out waiting for results"
+                        );
+                    }
+                    if error & 1 != 0 {
+                        artiq_raise!(
+                            "RTIOUnderflow",
+                            "RTIO underflow at {1} mu, channel {rtio_channel_info:0}",
+                            channel as i64,
+                            timestamp as i64,
+                            0
+                        );
+                    }
+                    if error & 2 != 0 {
+                        artiq_raise!(
+                            "RTIODestinationUnreachable",
+                            "RTIO destination unreachable, output, at {1} mu, channel {rtio_channel_info:0}",
+                            channel as i64,
+                            timestamp as i64,
+                            0
+                        );
+                    }
+                }
+                _ => panic!("Expected DmaAwaitRemoteReply after DmaAwaitRemoteRequest!"),
+            }
+        }
     }
 }

src/runtime/src/kernel/mod.rs

@@ -52,7 +52,21 @@ pub enum Message {
     DmaPutRequest(DmaRecorder),
     DmaEraseRequest(String),
     DmaGetRequest(String),
-    DmaGetReply(Option<(Vec<u8>, i64)>),
+    DmaGetReply(Option<(i32, i64)>),
+    #[cfg(has_drtio)]
+    DmaStartRemoteRequest {
+        id: i32,
+        timestamp: i64,
+    },
+    #[cfg(has_drtio)]
+    DmaAwaitRemoteRequest(i32),
+    #[cfg(has_drtio)]
+    DmaAwaitRemoteReply {
+        timeout: bool,
+        error: u8,
+        channel: u32,
+        timestamp: u64,
+    },
 
     #[cfg(has_drtio)]
     UpDestinationsRequest(i32),

src/runtime/src/main.rs

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

src/runtime/src/moninj.rs

@@ -298,7 +298,13 @@ 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 {
         loop {
             let aux_mutex = aux_mutex.clone();

src/runtime/src/rtio_dma.rs

@@ -0,0 +1,349 @@
+use alloc::{collections::BTreeMap, rc::Rc, string::String, vec::Vec};
+#[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;
+
+static DMA_RECORD_STORE: Mutex<BTreeMap<String, (u32, Vec<u8>, i64)>> = Mutex::new(BTreeMap::new());
+
+#[cfg(has_drtio)]
+pub mod remote_dma {
+    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 },
+    }
+    #[derive(Debug, Clone)]
+    struct RemoteTrace {
+        trace: Vec<u8>,
+        pub state: RemoteState,
+    }
+
+    impl From<Vec<u8>> for RemoteTrace {
+        fn from(trace: Vec<u8>) -> Self {
+            RemoteTrace {
+                trace: trace,
+                state: RemoteState::NotLoaded,
+            }
+        }
+    }
+
+    impl RemoteTrace {
+        pub fn get_trace(&self) -> &Vec<u8> {
+            &self.trace
+        }
+    }
+
+    // represents all traces for a given ID
+    struct TraceSet {
+        id: u32,
+        done_count: Mutex<usize>,
+        traces: Mutex<BTreeMap<u8, RemoteTrace>>,
+    }
+
+    impl TraceSet {
+        pub fn new(id: u32, traces: BTreeMap<u8, Vec<u8>>) -> TraceSet {
+            let mut trace_map: BTreeMap<u8, RemoteTrace> = BTreeMap::new();
+            for (destination, trace) in traces {
+                trace_map.insert(destination, trace.into());
+            }
+            TraceSet {
+                id: id,
+                done_count: Mutex::new(0),
+                traces: Mutex::new(trace_map),
+            }
+        }
+
+        pub async fn await_done(&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)
+                & (*(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 lock = self.traces.async_lock().await;
+            let trace_iter = lock.iter_mut();
+            for (_dest, trace) in trace_iter {
+                match trace.state {
+                    RemoteState::PlaybackEnded {
+                        error: e,
+                        channel: _c,
+                        timestamp: _ts,
+                    } => {
+                        if e != 0 {
+                            playback_state = trace.state.clone();
+                        }
+                    }
+                    _ => (),
+                }
+                trace.state = RemoteState::Loaded;
+            }
+            Ok(playback_state)
+        }
+
+        pub async fn upload_traces(
+            &mut self,
+            aux_mutex: &Rc<Mutex<bool>>,
+            routing_table: &RoutingTable,
+            timer: GlobalTimer,
+        ) {
+            let mut lock = self.traces.async_lock().await;
+            let trace_iter = lock.iter_mut();
+            for (destination, trace) in trace_iter {
+                match drtio::ddma_upload_trace(
+                    aux_mutex,
+                    routing_table,
+                    timer,
+                    self.id,
+                    *destination,
+                    trace.get_trace(),
+                )
+                .await
+                {
+                    Ok(_) => trace.state = RemoteState::Loaded,
+                    Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
+                }
+            }
+            *(self.done_count.async_lock().await) = 0;
+        }
+
+        pub async fn erase(&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(aux_mutex, routing_table, timer, self.id, *destination).await {
+                    Ok(_) => (),
+                    Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
+                }
+            }
+        }
+
+        pub async fn playback_done(&mut self, destination: u8, error: u8, channel: u32, timestamp: u64) {
+            let mut traces_locked = self.traces.async_lock().await;
+            let mut trace = traces_locked.get_mut(&destination).unwrap();
+            trace.state = RemoteState::PlaybackEnded {
+                error: error,
+                channel: channel,
+                timestamp: timestamp,
+            };
+            *(self.done_count.async_lock().await) += 1;
+        }
+
+        pub async fn playback(
+            &self,
+            aux_mutex: &Rc<Mutex<bool>>,
+            routing_table: &RoutingTable,
+            timer: GlobalTimer,
+            timestamp: u64,
+        ) {
+            let mut dest_list: Vec<u8> = Vec::new();
+            {
+                let lock = self.traces.async_lock().await;
+                let trace_iter = lock.iter();
+                for (dest, trace) in trace_iter {
+                    if trace.state != RemoteState::Loaded {
+                        error!("Destination {} not ready for DMA, state: {:?}", dest, trace.state);
+                        continue;
+                    }
+                    dest_list.push(dest.clone());
+                }
+            }
+            // 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(aux_mutex, routing_table, timer, self.id, destination, timestamp).await
+                {
+                    Ok(_) => (),
+                    Err(e) => error!("Error during remote DMA playback: {}", e),
+                }
+            }
+        }
+
+        pub async fn destination_changed(
+            &mut self,
+            aux_mutex: &Rc<Mutex<bool>>,
+            routing_table: &RoutingTable,
+            timer: GlobalTimer,
+            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) {
+                if up {
+                    match drtio::ddma_upload_trace(
+                        aux_mutex,
+                        routing_table,
+                        timer,
+                        self.id,
+                        destination,
+                        trace.get_trace(),
+                    )
+                    .await
+                    {
+                        Ok(_) => trace.state = RemoteState::Loaded,
+                        Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
+                    }
+                } else {
+                    trace.state = RemoteState::NotLoaded;
+                }
+            }
+        }
+    }
+
+    static mut TRACES: BTreeMap<u32, TraceSet> = BTreeMap::new();
+
+    pub fn add_traces(id: u32, traces: BTreeMap<u8, Vec<u8>>) {
+        unsafe { TRACES.insert(id, TraceSet::new(id, traces)) };
+    }
+
+    pub async fn await_done(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(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);
+        }
+    }
+
+    pub async fn upload_traces(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;
+    }
+
+    pub async fn playback(
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &RoutingTable,
+        timer: GlobalTimer,
+        id: u32,
+        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(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;
+    }
+
+    pub async fn destination_changed(
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &RoutingTable,
+        timer: GlobalTimer,
+        destination: u8,
+        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;
+        }
+    }
+}
+
+pub async fn put_record(
+    _aux_mutex: &Rc<Mutex<bool>>,
+    _routing_table: &RoutingTable,
+    _timer: GlobalTimer,
+    mut recorder: DmaRecorder,
+) -> u32 {
+    #[cfg(has_drtio)]
+    let mut remote_traces: BTreeMap<u8, Vec<u8>> = BTreeMap::new();
+
+    #[cfg(has_drtio)]
+    if recorder.enable_ddma {
+        let mut local_trace: Vec<u8> = Vec::new();
+        // analyze each entry and put in proper buckets, as the kernel core
+        // sends whole chunks, to limit comms/kernel CPU communication,
+        // and as only comms core has access to varios DMA buffers.
+        let mut ptr = 0;
+        recorder.buffer.push(0);
+        while recorder.buffer[ptr] != 0 {
+            // ptr + 3 = tgt >> 24 (destination)
+            let len = recorder.buffer[ptr] as usize;
+            let destination = recorder.buffer[ptr + 3];
+            if destination == 0 {
+                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());
+                }
+            }
+            // and jump to the next event
+            ptr += len;
+        }
+        mem::swap(&mut recorder.buffer, &mut local_trace);
+    }
+    // trailing zero to indicate end of buffer
+    recorder.buffer.push(0);
+    recorder.buffer.reserve(ALIGNMENT - 1);
+    let original_length = recorder.buffer.len();
+    let padding = ALIGNMENT - recorder.buffer.as_ptr() as usize % ALIGNMENT;
+    let padding = if padding == ALIGNMENT { 0 } else { padding };
+    for _ in 0..padding {
+        recorder.buffer.push(0);
+    }
+    recorder.buffer.copy_within(0..original_length, padding);
+    dcci_slice(&recorder.buffer);
+
+    let ptr = recorder.buffer[padding..].as_ptr() as u32;
+
+    let _old_record = DMA_RECORD_STORE
+        .lock()
+        .insert(recorder.name, (ptr, recorder.buffer, recorder.duration));
+
+    #[cfg(has_drtio)]
+    {
+        if let Some((old_id, _v, _d)) = _old_record {
+            remote_dma::erase(_aux_mutex, _routing_table, _timer, old_id).await;
+        }
+        remote_dma::add_traces(ptr, remote_traces);
+    }
+
+    ptr
+}
+
+pub async fn erase(name: String, _aux_mutex: &Rc<Mutex<bool>>, _routing_table: &RoutingTable, _timer: GlobalTimer) {
+    let _entry = DMA_RECORD_STORE.lock().remove(&name);
+    #[cfg(has_drtio)]
+    if let Some((id, _v, _d)) = _entry {
+        remote_dma::erase(_aux_mutex, _routing_table, _timer, id).await;
+    }
+}
+
+pub fn retrieve(name: String) -> Option<(i32, i64)> {
+    let (ptr, _v, duration) = DMA_RECORD_STORE.lock().get(&name)?.clone();
+    Some((ptr as i32, duration))
+}

src/runtime/src/rtio_mgt.rs

@@ -6,20 +6,23 @@ use libboard_artiq::{drtio_routing, pl::csr};
 use libboard_zynq::timer::GlobalTimer;
 use libconfig::Config;
 use libcortex_a9::mutex::Mutex;
-use log::error;
+use log::warn;
 
 static mut RTIO_DEVICE_MAP: BTreeMap<u32, String> = BTreeMap::new();
 
 #[cfg(has_drtio)]
 pub mod drtio {
+    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_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,
+                SEEN_ASYNC_ERRORS};
 
     pub fn startup(
         aux_mutex: &Rc<Mutex<bool>>,
@@ -63,7 +66,22 @@ pub mod drtio {
         }
         let _lock = aux_mutex.async_lock().await;
         drtioaux_async::send(linkno, request).await.unwrap();
-        recv_aux_timeout(linkno, 200, timer).await
+        loop {
+            let reply = recv_aux_timeout(linkno, 200, timer).await;
+            match reply {
+                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),
+            }
+        }
     }
 
     async fn drain_buffer(linkno: u8, draining_time: Milliseconds, timer: GlobalTimer) {
@@ -172,6 +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,
+            })) => 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),
@@ -252,7 +277,9 @@ pub mod drtio {
                         .await;
                         match reply {
                             Ok(Packet::DestinationDownReply) => {
-                                destination_set_up(routing_table, up_destinations, destination, false).await
+                                destination_set_up(routing_table, up_destinations, destination, false).await;
+                                remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, false)
+                                    .await;
                             }
                             Ok(Packet::DestinationOkReply) => (),
                             Ok(Packet::DestinationSequenceErrorReply { channel }) => {
@@ -287,6 +314,7 @@ pub mod drtio {
                         }
                     } else {
                         destination_set_up(routing_table, up_destinations, destination, false).await;
+                        remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, false).await;
                     }
                 } else {
                     if up_links[linkno as usize] {
@@ -304,6 +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;
                             }
                             Ok(packet) => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet),
                             Err(e) => error!("[DEST#{}] communication failed ({})", destination, e),
@@ -389,6 +419,109 @@ pub mod drtio {
             }
         }
     }
+
+    pub async fn ddma_upload_trace(
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        id: u32,
+        destination: 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 last = i + len == trace.len();
+            trace_slice[..len].clone_from_slice(&trace[i..i + len]);
+            i += len;
+            let reply = aux_transact(
+                aux_mutex,
+                linkno,
+                &Packet::DmaAddTraceRequest {
+                    id: id,
+                    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");
+                }
+                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,
+    ) -> Result<(), &'static str> {
+        let linkno = routing_table.0[destination as usize][0] - 1;
+        let reply = aux_transact(
+            aux_mutex,
+            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"),
+        }
+    }
+
+    pub async fn ddma_send_playback(
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        id: u32,
+        destination: u8,
+        timestamp: u64,
+    ) -> Result<(), &'static str> {
+        let linkno = routing_table.0[destination as usize][0] - 1;
+        let reply = aux_transact(
+            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"),
+        }
+    }
 }
 
 fn read_device_map(cfg: &Config) -> BTreeMap<u32, String> {
@@ -402,8 +535,8 @@ fn read_device_map(cfg: &Config) -> BTreeMap<u32, String> {
                 let channel = bytes_cr.read_u32().unwrap();
                 let device_name = bytes_cr.read_string().unwrap();
                 if let Some(old_entry) = device_map.insert(channel, device_name.clone()) {
-                    error!(
+                    warn!(
-                        "read_device_map: conflicting entries for channel {}: `{}` and `{}`",
+                        "conflicting device map entries for RTIO channel {}: '{}' and '{}'",
                         channel, old_entry, device_name
                     );
                 }
@@ -411,7 +544,10 @@ fn read_device_map(cfg: &Config) -> BTreeMap<u32, String> {
             Ok(())
         })
         .or_else(|err| {
-            error!("read_device_map: error reading `device_map` from config: {}", err);
+            warn!(
+                "error reading device map ({}), device names will not be available in RTIO error messages",
+                err
+            );
             Err(err)
         });
     device_map

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 id: u32,
-    pub error: u8, 
+    pub error: u8,
-    pub channel: u32, 
+    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,29 +56,37 @@ impl Manager {
                 if entry.complete {
                     // replace entry
                     self.entries.remove(&id);
-                    self.entries.insert(id, Entry {
+                    self.entries.insert(
-                        trace: Vec::new(),
+                        id,
-                        padding_len: 0,
+                        Entry {
-                        complete: false });
+                            trace: Vec::new(),
+                            padding_len: 0,
+                            complete: false,
+                        },
+                    );
                     self.entries.get_mut(&id).unwrap()
                 } else {
                     entry
                 }
-            },
+            }
             None => {
-                self.entries.insert(id, Entry {
+                self.entries.insert(
-                    trace: Vec::new(),
+                    id,
-                    padding_len: 0,
+                    Entry {
-                    complete: false });
+                        trace: Vec::new(),
+                        padding_len: 0,
+                        complete: false,
+                    },
+                );
                 self.entries.get_mut(&id).unwrap()
-            },
+            }
         };
         entry.trace.extend(&trace[0..trace_len]);
 
         if last {
             entry.trace.push(0);
             let data_len = entry.trace.len();
-    
+
             // Realign.
             entry.trace.reserve(ALIGNMENT - 1);
             let padding = ALIGNMENT - entry.trace.as_ptr() as usize % ALIGNMENT;
@@ -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);
@@ -126,7 +135,7 @@ impl Manager {
         unsafe {
             csr::rtio_dma::base_address_write(ptr as u32);
             csr::rtio_dma::time_offset_write(timestamp as u64);
-    
+
             csr::cri_con::selected_write(1);
             csr::rtio_dma::enable_write(1);
             // playback has begun here, for status call check_state
@@ -144,21 +153,21 @@ impl Manager {
             return None;
         } else {
             self.state = ManagerState::Idle;
-            unsafe { 
+            unsafe {
                 csr::cri_con::selected_write(0);
-                let error =  csr::rtio_dma::error_read();
+                let error = csr::rtio_dma::error_read();
                 let channel = csr::rtio_dma::error_channel_read();
                 let timestamp = csr::rtio_dma::error_timestamp_read();
                 if error != 0 {
                     csr::rtio_dma::error_write(1);
                 }
-                return Some(RtioStatus { 
+                return Some(RtioStatus {
-                    id: self.currentid, 
+                    id: self.currentid,
                     error: error,
-                    channel: channel, 
+                    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.
@@ -412,36 +412,33 @@ fn process_aux_packet(
             )
         }
 
-        drtioaux::Packet::DmaAddTraceRequest { 
+        drtioaux::Packet::DmaAddTraceRequest {
-            destination: _destination, 
+            destination: _destination,
-            id, 
+            id,
-            last, 
+            last,
-            length, 
+            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 { 
+        drtioaux::Packet::DmaRemoveTraceRequest {
-            destination: _destination, 
+            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 { 
+        drtioaux::Packet::DmaPlaybackRequest {
-            destination: _destination, 
+            destination: _destination,
-            id, 
+            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: {}",
-                    destination: rank, 
+                    status.error, status.channel, status.timestamp
-                    id: status.id, 
+                );
-                    error: status.error, 
+                if let Err(e) = drtioaux::send(
-                    channel: status.channel, 
+                    0,
-                    timestamp: status.timestamp 
+                    &drtioaux::Packet::DmaPlaybackStatus {
-                }) {
+                        destination: rank,
+                        id: status.id,
+                        error: status.error,
+                        channel: status.channel,
+                        timestamp: status.timestamp,
+                    },
+                ) {
                     error!("error sending DMA playback status: {:?}", e);
                 }
             }