Merge branch 'master' into nac3

RELEASE_NOTES.rst

@@ -23,6 +23,9 @@ Highlights:
   support legacy installations, but may be removed in a future release.
 * Added channel names to RTIO errors.
 * Full Python 3.10 support.
+* Distributed DMA is now supported, allowing DMA to be run directly on satellites for corresponding
+  RTIO events, increasing bandwidth in scenarios with heavy satellite usage.
+* Persistent datasets are now stored in a LMDB database for improved performance.
 
 ARTIQ-7
 -------

artiq/afws.pem

@@ -1,23 +0,0 @@
------BEGIN CERTIFICATE-----
-MIID0zCCArugAwIBAgIUPkNfEUx/uau3z8SD4mgMbCK/DEgwDQYJKoZIhvcNAQEL
-BQAweTELMAkGA1UEBhMCSEsxEzARBgNVBAgMClNvbWUtU3RhdGUxFzAVBgNVBAoM
-Dk0tTGFicyBMaW1pdGVkMRkwFwYDVQQDDBBuaXhibGQubS1sYWJzLmhrMSEwHwYJ
-KoZIhvcNAQkBFhJoZWxwZGVza0BtLWxhYnMuaGswHhcNMjIwMjA2MTA1ODQ0WhcN
-MjUwMjA1MTA1ODQ0WjB5MQswCQYDVQQGEwJISzETMBEGA1UECAwKU29tZS1TdGF0
-ZTEXMBUGA1UECgwOTS1MYWJzIExpbWl0ZWQxGTAXBgNVBAMMEG5peGJsZC5tLWxh
-YnMuaGsxITAfBgkqhkiG9w0BCQEWEmhlbHBkZXNrQG0tbGFicy5oazCCASIwDQYJ
-KoZIhvcNAQEBBQADggEPADCCAQoCggEBAPWetZhoggPR2ae7waGzv1AQ8NQO3noW
-8DofVjusNpX5i/YB0waAr1bm1tALLJoHV2r/gTxujlXCe/L/WG1DLseCf6NO9sHg
-t0FLhDpF9kPMWBgauVVLepd2Y2yU1G8eFuEVGnsiQSu0IzsZP5FQBJSyxvxJ+V/L
-EW9ox91VGOP9VZR9jqdlYjGhcwClHA/nHe0q1fZq42+9rG466I5yIlNSoa7ilhTU
-2C2doxy6Sr6VJYnLEMQqoIF65t3MkKi9iaqN7az0OCrj6XR0P5iKBzUhIgMUd2qs
-7Id0XUdbQvaoaRI67vhGkNr+f4rdAUNCDGcbbokuBnmE7/gva6BAABUCAwEAAaNT
-MFEwHQYDVR0OBBYEFM2e2FmcytXhKyfC1KEjVJ2mPSy3MB8GA1UdIwQYMBaAFM2e
-2FmcytXhKyfC1KEjVJ2mPSy3MA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEL
-BQADggEBAKH0z5vlbfTghjYWwd2yEEFBbZx5XxaLHboFQpFpxu9sZoidVs047tco
-MOr1py9juiNGGM8G35sw9306f+thDFwqlQfSExUwp5pRQNq+mxglMSF05HWDqBwb
-wnItKi/WXpkMQXgpQJFVeflz4B4ZFNlH1UQl5bwacXOM9NM9zO7duCjVXmGE0yxi
-VQyApfPQYu9whCSowDYYaA0toJeikMzGfWxhlAH79/2Qmit8KcSCbX1fK/QoRZLa
-5NeUi/OlJbBpkgTrfzfMLphmsPWPAVMeUKzqd/vXfG6ZBOZZm6e6sl8RBycBezII
-15WekikTE5+T54/E0xiu+zIW/Xhhk14=
------END CERTIFICATE-----

artiq/coredevice/ad9914.py

@@ -92,8 +92,11 @@ class AD9914:
         self.exit_x_duration_mu    = 3 * self.write_duration_mu
 
     @staticmethod
-    def get_rtio_channels(bus_channel, **kwargs):
+    def get_rtio_channels(bus_channel, channel, **kwargs):
+        # return only first entry, as there are several devices with the same RTIO channel
+        if channel == 0:
             return [(bus_channel, None)]
+        return []
 
     @kernel
     def write(self, addr: int32, data: int32):

artiq/coredevice/adf5356.py

@@ -84,8 +84,8 @@ class ADF5356:
         self._init_registers()
 
     @staticmethod
-    def get_rtio_channels(channel, **kwargs):
-        return [(channel, None)]
+    def get_rtio_channels(**kwargs):
+        return []
 
     @kernel
     def init(self, blind: bool = False):

artiq/coredevice/comm_analyzer.py

@@ -102,15 +102,15 @@ def decode_dump(data):
     # messages are big endian
     parts = struct.unpack(endian + "IQbbb", data[:15])
     (sent_bytes, total_byte_count,
-     error_occured, log_channel, dds_onehot_sel) = parts
+     error_occurred, log_channel, dds_onehot_sel) = parts
 
     expected_len = sent_bytes + 15
     if expected_len != len(data):
         raise ValueError("analyzer dump has incorrect length "
                          "(got {}, expected {})".format(
                             len(data), expected_len))
-    if error_occured:
-        logger.warning("error occured within the analyzer, "
+    if error_occurred:
+        logger.warning("error occurred within the analyzer, "
                        "data may be corrupted")
     if total_byte_count > sent_bytes:
         logger.info("analyzer ring buffer has wrapped %d times",

artiq/coredevice/coredevice_generic.schema.json

@@ -339,7 +339,8 @@
                             "maxItems": 2
                         },
                         "sampler_hw_rev": {
-                            "type": "string"
+                            "type": "string",
+                            "pattern": "^v[0-9]+\\.[0-9]+"
                         },
                         "urukul0_ports": {
                             "type": "array",

artiq/coredevice/dma.py

@@ -18,7 +18,7 @@ def dma_record_start(name: str):
     raise NotImplementedError("syscall not simulated")
 
 @extern
-def dma_record_stop(duration: int64):
+def dma_record_stop(duration: int64, enable_ddma: bool):
     raise NotImplementedError("syscall not simulated")
 
 @extern
@@ -26,11 +26,11 @@ def dma_erase(name: str):
     raise NotImplementedError("syscall not simulated")
 
 @extern
-def dma_retrieve(name: str) -> tuple[int64, int32]:
+def dma_retrieve(name: str) -> tuple[int64, int32, bool]:
     raise NotImplementedError("syscall not simulated")
 
 @extern
-def dma_playback(timestamp: int64, ptr: int32):
+def dma_playback(timestamp: int64, ptr: int32, enable_ddma: bool):
     raise NotImplementedError("syscall not simulated")
 
 
@@ -52,6 +52,7 @@ class DMARecordContextManager:
     def __init__(self):
         self.name = ""
         self.saved_now_mu = int64(0)
+        self.enable_ddma = False
 
     @kernel
     def __enter__(self):
@@ -61,7 +62,7 @@ class DMARecordContextManager:
 
     @kernel
     def __exit__(self):
-        dma_record_stop(now_mu()) # see above
+        dma_record_stop(now_mu(), self.enable_ddma) # see above
         at_mu(self.saved_now_mu)
 
 
@@ -82,12 +83,20 @@ class CoreDMA:
         self.epoch    = 0
 
     @kernel
-    def record(self, name: str) -> DMARecordContextManager:
+    def record(self, name: str, enable_ddma: bool = False) -> DMARecordContextManager:
         """Returns a context manager that will record a DMA trace called ``name``.
         Any previously recorded trace with the same name is overwritten.
-        The trace will persist across kernel switches."""
+        The trace will persist across kernel switches.
+
+        In DRTIO context, distributed DMA can be toggled with ``enable_ddma``.
+        Enabling it allows running DMA on satellites, rather than sending all
+        events from the master.
+
+        Keeping it disabled it may improve performance in some scenarios, 
+        e.g. when there are many small satellite buffers."""
         self.epoch += 1
         self.recorder.name = name
+        self.recorder.enable_ddma = enable_ddma
         return self.recorder
 
     @kernel
@@ -100,24 +109,24 @@ class CoreDMA:
     def playback(self, name: str):
         """Replays a previously recorded DMA trace. This function blocks until
         the entire trace is submitted to the RTIO FIFOs."""
-        (advance_mu, ptr) = dma_retrieve(name)
-        dma_playback(now_mu(), ptr)
+        (advance_mu, ptr, uses_ddma) = dma_retrieve(name)
+        dma_playback(now_mu(), ptr, uses_ddma)
         delay_mu(advance_mu)
 
     @kernel
     def get_handle(self, name: str) -> tuple[int32, int64, int32]:
         """Returns a handle to a previously recorded DMA trace. The returned handle
         is only valid until the next call to :meth:`record` or :meth:`erase`."""
-        (advance_mu, ptr) = dma_retrieve(name)
-        return (self.epoch, advance_mu, ptr)
+        (advance_mu, ptr, uses_ddma) = dma_retrieve(name)
+        return (self.epoch, advance_mu, ptr, uses_ddma)
 
     @kernel
     def playback_handle(self, handle: tuple[int32, int64, int32]):
         """Replays a handle obtained with :meth:`get_handle`. Using this function
         is much faster than :meth:`playback` for replaying a set of traces repeatedly,
         but incurs the overhead of managing the handles onto the programmer."""
-        (epoch, advance_mu, ptr) = handle
+        (epoch, advance_mu, ptr, uses_ddma) = handle
         if self.epoch != epoch:
             raise DMAError("Invalid handle")
-        dma_playback(now_mu(), ptr)
+        dma_playback(now_mu(), ptr, uses_ddma)
         delay_mu(advance_mu)

artiq/firmware/ksupport/lib.rs

@@ -269,7 +269,7 @@ extern fn dma_record_start(name: &CSlice<u8>) {
 }
 
 #[unwind(allowed)]
-extern fn dma_record_stop(duration: i64) {
+extern fn dma_record_stop(duration: i64, enable_ddma: bool) {
     unsafe {
         dma_record_flush();
 
@@ -285,7 +285,8 @@ extern fn dma_record_stop(duration: i64) {
 
         DMA_RECORDER.active = false;
         send(&DmaRecordStop {
-            duration: duration as u64
+            duration: duration as u64,
+            enable_ddma: enable_ddma
         });
     }
 }
@@ -371,6 +372,7 @@ extern fn dma_erase(name: &CSlice<u8>) {
 struct DmaTrace {
     duration: i64,
     address:  i32,
+    uses_ddma: bool,
 }
 
 #[unwind(allowed)]
@@ -378,11 +380,12 @@ extern fn dma_retrieve(name: &CSlice<u8>) -> DmaTrace {
     let name = str::from_utf8(name.as_ref()).unwrap();
 
     send(&DmaRetrieveRequest { name: name });
-    recv!(&DmaRetrieveReply { trace, duration } => {
+    recv!(&DmaRetrieveReply { trace, duration, uses_ddma } => {
         match trace {
             Some(bytes) => Ok(DmaTrace {
                 address:  bytes.as_ptr() as i32,
-                duration: duration as i64
+                duration: duration as i64,
+                uses_ddma: uses_ddma,
             }),
             None => Err(())
         }
@@ -395,7 +398,7 @@ extern fn dma_retrieve(name: &CSlice<u8>) -> DmaTrace {
 
 #[cfg(has_rtio_dma)]
 #[unwind(allowed)]
-extern fn dma_playback(timestamp: i64, ptr: i32) {
+extern fn dma_playback(timestamp: i64, ptr: i32, _uses_ddma: bool) {
     assert!(ptr % 64 == 0);
 
     unsafe {
@@ -404,6 +407,10 @@ extern fn dma_playback(timestamp: i64, ptr: i32) {
 
         csr::cri_con::selected_write(1);
         csr::rtio_dma::enable_write(1);
+        #[cfg(has_drtio)]
+        if _uses_ddma {
+            send(&DmaStartRemoteRequest { id: ptr as i32, timestamp: timestamp });
+        }
         while csr::rtio_dma::enable_read() != 0 {}
         csr::cri_con::selected_write(0);
 
@@ -424,11 +431,32 @@ extern fn dma_playback(timestamp: i64, ptr: i32) {
             }
         }
     }
+
+    #[cfg(has_drtio)]
+    if _uses_ddma {
+        send(&DmaAwaitRemoteRequest { id: ptr as i32 });
+        recv!(&DmaAwaitRemoteReply { timeout, error, channel, timestamp } => {
+            if timeout {
+                raise!("DMAError",
+                    "Error running DMA on satellite device, timed out waiting for results");
+            }
+            if error & 1 != 0 {
+                raise!("RTIOUnderflow",
+                    "RTIO underflow at channel {rtio_channel_info:0}, {1} mu",
+                    channel as i64, timestamp as i64, 0);
+            }
+            if error & 2 != 0 {
+                raise!("RTIODestinationUnreachable",
+                    "RTIO destination unreachable, output, at channel {rtio_channel_info:0}, {1} mu",
+                    channel as i64, timestamp as i64, 0);
+            }
+        });
+    }
 }
 
 #[cfg(not(has_rtio_dma))]
 #[unwind(allowed)]
-extern fn dma_playback(_timestamp: i64, _ptr: i32) {
+extern fn dma_playback(_timestamp: i64, _ptr: i32, _uses_ddma: bool) {
     unimplemented!("not(has_rtio_dma)")
 }
 

artiq/firmware/libproto_artiq/drtioaux_proto.rs

@@ -14,8 +14,8 @@ impl<T> From<IoError<T>> for Error<T> {
     }
 }
 
-/* 512 (max size) - 4 (CRC) - 1 (packet ID) - 4 (trace ID) - 1 (last) - 2 (length) */
-const DMA_TRACE_MAX_SIZE: usize = 500;
+/* 512 (max size) - 4 (CRC) - 1 (packet ID) - 1 (destination) - 4 (trace ID) - 1 (last) - 2 (length) */
+pub const DMA_TRACE_MAX_SIZE: usize = 499;
 
 #[derive(PartialEq, Debug)]
 pub enum Packet {
@@ -58,13 +58,13 @@ pub enum Packet {
     SpiReadReply { succeeded: bool, data: u32 },
     SpiBasicReply { succeeded: bool },
 
-    DmaAddTraceRequest { id: u32, last: bool, length: u16, trace: [u8; DMA_TRACE_MAX_SIZE] },
+    DmaAddTraceRequest { destination: u8, id: u32, last: bool, length: u16, trace: [u8; DMA_TRACE_MAX_SIZE] },
     DmaAddTraceReply { succeeded: bool },
-    DmaRemoveTraceRequest { id: u32 },
+    DmaRemoveTraceRequest { destination: u8, id: u32 },
     DmaRemoveTraceReply { succeeded: bool },
-    DmaPlaybackRequest { id: u32, timestamp: u64 },
+    DmaPlaybackRequest { destination: u8, id: u32, timestamp: u64 },
     DmaPlaybackReply { succeeded: bool },
-    DmaPlaybackStatus { id: u32, error: u8, channel: u32, timestamp: u64 }
+    DmaPlaybackStatus { destination: u8, id: u32, error: u8, channel: u32, timestamp: u64 }
 
 }
 
@@ -198,12 +198,14 @@ impl Packet {
             },
 
             0xb0 => { 
+                let destination = reader.read_u8()?;
                 let id = reader.read_u32()?;
                 let last = reader.read_bool()?;
                 let length = reader.read_u16()?;
                 let mut trace: [u8; DMA_TRACE_MAX_SIZE] = [0; DMA_TRACE_MAX_SIZE];
                 reader.read_exact(&mut trace[0..length as usize])?;
                 Packet::DmaAddTraceRequest {
+                    destination: destination,
                     id: id,
                     last: last,
                     length: length as u16,
@@ -214,12 +216,14 @@ impl Packet {
                 succeeded: reader.read_bool()?
             },
             0xb2 => Packet::DmaRemoveTraceRequest {
+                destination: reader.read_u8()?,
                 id: reader.read_u32()?
             },
             0xb3 => Packet::DmaRemoveTraceReply {
                 succeeded: reader.read_bool()?
             },
             0xb4 => Packet::DmaPlaybackRequest {
+                destination: reader.read_u8()?,
                 id: reader.read_u32()?,
                 timestamp: reader.read_u64()?
             },
@@ -227,6 +231,7 @@ impl Packet {
                 succeeded: reader.read_bool()?
             },
             0xb6 => Packet::DmaPlaybackStatus {
+                destination: reader.read_u8()?,
                 id: reader.read_u32()?,
                 error: reader.read_u8()?,
                 channel: reader.read_u32()?,
@@ -392,8 +397,9 @@ impl Packet {
                 writer.write_bool(succeeded)?;
             },
 
-            Packet::DmaAddTraceRequest { id, last, trace, length } => {
+            Packet::DmaAddTraceRequest { destination, id, last, trace, length } => {
                 writer.write_u8(0xb0)?;
+                writer.write_u8(destination)?;
                 writer.write_u32(id)?;
                 writer.write_bool(last)?;
                 // trace may be broken down to fit within drtio aux memory limit
@@ -405,16 +411,18 @@ impl Packet {
                 writer.write_u8(0xb1)?;
                 writer.write_bool(succeeded)?;
             },
-            Packet::DmaRemoveTraceRequest { id } => {
+            Packet::DmaRemoveTraceRequest { destination, id } => {
                 writer.write_u8(0xb2)?;
+                writer.write_u8(destination)?;
                 writer.write_u32(id)?;
             },
             Packet::DmaRemoveTraceReply { succeeded } => {
                 writer.write_u8(0xb3)?;
                 writer.write_bool(succeeded)?;
             },
-            Packet::DmaPlaybackRequest { id, timestamp } => {
+            Packet::DmaPlaybackRequest { destination, id, timestamp } => {
                 writer.write_u8(0xb4)?;
+                writer.write_u8(destination)?;
                 writer.write_u32(id)?;
                 writer.write_u64(timestamp)?;
             },
@@ -422,8 +430,9 @@ impl Packet {
                 writer.write_u8(0xb5)?;
                 writer.write_bool(succeeded)?;
             },
-            Packet::DmaPlaybackStatus { id, error, channel, timestamp } => {
+            Packet::DmaPlaybackStatus { destination, id, error, channel, timestamp } => {
                 writer.write_u8(0xb6)?;
+                writer.write_u8(destination)?;
                 writer.write_u32(id)?;
                 writer.write_u8(error)?;
                 writer.write_u32(channel)?;

artiq/firmware/libproto_artiq/kernel_proto.rs

@@ -20,7 +20,8 @@ pub enum Message<'a> {
     DmaRecordStart(&'a str),
     DmaRecordAppend(&'a [u8]),
     DmaRecordStop {
-        duration:  u64
+        duration:  u64,
+        enable_ddma: bool
     },
 
     DmaEraseRequest {
@@ -32,9 +33,25 @@ pub enum Message<'a> {
     },
     DmaRetrieveReply {
         trace:    Option<&'a [u8]>,
-        duration: u64
+        duration: u64,
+        uses_ddma: bool,
     },
 
+    DmaStartRemoteRequest {
+        id: i32,
+        timestamp: i64,
+    },
+    DmaAwaitRemoteRequest {
+        id: i32
+    },
+    DmaAwaitRemoteReply {
+        timeout: bool,
+        error: u8,
+        channel: u32,
+        timestamp: u64
+    },
+
+
     RunFinished,
     RunException {
         exceptions: &'a [Option<eh::eh_artiq::Exception<'a>>],

artiq/firmware/libproto_artiq/session_proto.rs

@@ -1,10 +1,14 @@
-use core::str::Utf8Error;
-use alloc::vec::Vec;
+use core::{str, str::Utf8Error, slice};
+use alloc::{vec::Vec, format, collections::BTreeMap, string::String};
 use eh::eh_artiq::{Exception, StackPointerBacktrace};
 use cslice::CSlice;
 
 use io::{Read, ProtoRead, Write, ProtoWrite, Error as IoError, ReadStringError};
 
+pub type DeviceMap = BTreeMap<u32, String>;
+
+static mut RTIO_DEVICE_MAP: Option<DeviceMap> = None;
+
 #[derive(Fail, Debug)]
 pub enum Error<T> {
     #[fail(display = "incorrect magic")]
@@ -190,7 +194,14 @@ impl<'a> Reply<'a> {
                 for exception in exceptions.iter() {
                     let exception = exception.as_ref().unwrap();
                     writer.write_u32(exception.id as u32)?;
+                    if exception.message.len() == usize::MAX {
+                        // exception with host string
                         write_exception_string(writer, &exception.message)?;
+                    } else {
+                        let msg = str::from_utf8(unsafe { slice::from_raw_parts(exception.message.as_ptr(), exception.message.len()) }).unwrap()
+                          .replace("{rtio_channel_info:0}", &format!("0x{:04x}:{}", exception.param[0], resolve_channel_name(exception.param[0] as u32)));
+                        write_exception_string(writer, unsafe { &CSlice::new(msg.as_ptr(), msg.len()) })?;
+                    }
                     writer.write_u64(exception.param[0] as u64)?;
                     writer.write_u64(exception.param[1] as u64)?;
                     writer.write_u64(exception.param[2] as u64)?;
@@ -226,3 +237,20 @@ impl<'a> Reply<'a> {
         Ok(())
     }
 }
+
+pub fn set_device_map(device_map: DeviceMap) {
+    unsafe { RTIO_DEVICE_MAP = Some(device_map); }
+}
+
+fn _resolve_channel_name(channel: u32, device_map: &Option<DeviceMap>) -> String {
+    if let Some(dev_map) = device_map {
+        match dev_map.get(&channel) {
+            Some(val) => val.clone(),
+            None => String::from("unknown")
+        }
+    } else { String::from("unknown") }
+}
+
+pub fn resolve_channel_name(channel: u32) -> String {
+    _resolve_channel_name(channel, unsafe{&RTIO_DEVICE_MAP})
+}

artiq/firmware/runtime/kern_hwreq.rs

@@ -14,8 +14,8 @@ mod remote_i2c {
     use rtio_mgt::drtio;
     use sched::{Io, Mutex};
 
-    pub fn start(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<(), &'static str> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::I2cStartRequest {
+    pub fn start(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<(), &'static str> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::I2cStartRequest {
             destination: destination,
             busno: busno
         });
@@ -34,8 +34,8 @@ mod remote_i2c {
         }
     }
 
-    pub fn restart(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<(), &'static str> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::I2cRestartRequest {
+    pub fn restart(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<(), &'static str> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::I2cRestartRequest {
             destination: destination,
             busno: busno
         });
@@ -54,8 +54,8 @@ mod remote_i2c {
         }
     }
 
-    pub fn stop(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<(), &'static str> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::I2cStopRequest  {
+    pub fn stop(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<(), &'static str> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::I2cStopRequest  {
             destination: destination,
             busno: busno
         });
@@ -74,8 +74,8 @@ mod remote_i2c {
         }
     }
 
-    pub fn write(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, data: u8) -> Result<bool, &'static str> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::I2cWriteRequest {
+    pub fn write(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, data: u8) -> Result<bool, &'static str> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::I2cWriteRequest {
             destination: destination,
             busno: busno,
             data: data
@@ -95,8 +95,8 @@ mod remote_i2c {
         }
     }
 
-    pub fn read(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, ack: bool) -> Result<u8, &'static str> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::I2cReadRequest {
+    pub fn read(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, ack: bool) -> Result<u8, &'static str> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::I2cReadRequest {
             destination: destination,
             busno: busno,
             ack: ack
@@ -116,8 +116,8 @@ mod remote_i2c {
         }
     }
 
-    pub fn switch_select(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, address: u8, mask: u8) -> Result<(), &'static str> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::I2cSwitchSelectRequest {
+    pub fn switch_select(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, address: u8, mask: u8) -> Result<(), &'static str> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::I2cSwitchSelectRequest {
             destination: destination,
             busno: busno,
             address: address,
@@ -145,8 +145,8 @@ mod remote_spi {
     use rtio_mgt::drtio;
     use sched::{Io, Mutex};
 
-    pub fn set_config(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, flags: u8, length: u8, div: u8, cs: u8) -> Result<(), ()> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::SpiSetConfigRequest {
+    pub fn set_config(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, flags: u8, length: u8, div: u8, cs: u8) -> Result<(), ()> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::SpiSetConfigRequest {
             destination: destination,
             busno: busno,
             flags: flags,
@@ -169,8 +169,8 @@ mod remote_spi {
         }
     }
 
-    pub fn write(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, data: u32) -> Result<(), ()> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::SpiWriteRequest {
+    pub fn write(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8, data: u32) -> Result<(), ()> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::SpiWriteRequest {
             destination: destination,
             busno: busno,
             data: data
@@ -190,8 +190,8 @@ mod remote_spi {
         }
     }
 
-    pub fn read(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<u32, ()> {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::SpiReadRequest {
+    pub fn read(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, busno: u8) -> Result<u32, ()> {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::SpiReadRequest {
             destination: destination,
             busno: busno
         });
@@ -214,7 +214,7 @@ mod remote_spi {
 
 #[cfg(has_drtio)]
 macro_rules! dispatch {
-    ($io:ident, $aux_mutex:ident, $mod_local:ident, $mod_remote:ident, $routing_table:ident, $busno:expr, $func:ident $(, $param:expr)*) => {{
+    ($io:ident, $aux_mutex:ident, $ddma_mutex:ident, $mod_local:ident, $mod_remote:ident, $routing_table:ident, $busno:expr, $func:ident $(, $param:expr)*) => {{
         let destination = ($busno >> 16) as u8;
         let busno = $busno as u8;
         let hop = $routing_table.0[destination as usize][0];
@@ -222,27 +222,28 @@ macro_rules! dispatch {
             $mod_local::$func(busno, $($param, )*)
         } else {
             let linkno = hop - 1;
-            $mod_remote::$func($io, $aux_mutex, linkno, destination, busno, $($param, )*)
+            $mod_remote::$func($io, $aux_mutex, $ddma_mutex, linkno, destination, busno, $($param, )*)
         }
     }}
 }
 
 #[cfg(not(has_drtio))]
 macro_rules! dispatch {
-    ($io:ident, $aux_mutex:ident,$mod_local:ident, $mod_remote:ident, $routing_table:ident, $busno:expr, $func:ident $(, $param:expr)*) => {{
+    ($io:ident, $aux_mutex:ident, $ddma_mutex:ident, $mod_local:ident, $mod_remote:ident, $routing_table:ident, $busno:expr, $func:ident $(, $param:expr)*) => {{
         let busno = $busno as u8;
         $mod_local::$func(busno, $($param, )*)
     }}
 }
 
 pub fn process_kern_hwreq(io: &Io, aux_mutex: &Mutex,
+        ddma_mutex: &Mutex,
         _routing_table: &drtio_routing::RoutingTable,
         _up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
         request: &kern::Message) -> Result<bool, Error<SchedError>> {
     match request {
         &kern::RtioInitRequest => {
             info!("resetting RTIO");
-            rtio_mgt::reset(io, aux_mutex);
+            rtio_mgt::reset(io, aux_mutex, ddma_mutex);
             kern_acknowledge()
         }
 
@@ -258,47 +259,47 @@ pub fn process_kern_hwreq(io: &Io, aux_mutex: &Mutex,
         }
 
         &kern::I2cStartRequest { busno } => {
-            let succeeded = dispatch!(io, aux_mutex, local_i2c, remote_i2c, _routing_table, busno, start).is_ok();
+            let succeeded = dispatch!(io, aux_mutex, ddma_mutex, local_i2c, remote_i2c, _routing_table, busno, start).is_ok();
             kern_send(io, &kern::I2cBasicReply { succeeded: succeeded })
         }
         &kern::I2cRestartRequest { busno } => {
-            let succeeded = dispatch!(io, aux_mutex, local_i2c, remote_i2c, _routing_table, busno, restart).is_ok();
+            let succeeded = dispatch!(io, aux_mutex, ddma_mutex, local_i2c, remote_i2c, _routing_table, busno, restart).is_ok();
             kern_send(io, &kern::I2cBasicReply { succeeded: succeeded })
         }
         &kern::I2cStopRequest { busno } => {
-            let succeeded = dispatch!(io, aux_mutex, local_i2c, remote_i2c, _routing_table, busno, stop).is_ok();
+            let succeeded = dispatch!(io, aux_mutex, ddma_mutex, local_i2c, remote_i2c, _routing_table, busno, stop).is_ok();
             kern_send(io, &kern::I2cBasicReply { succeeded: succeeded })
         }
         &kern::I2cWriteRequest { busno, data } => {
-            match dispatch!(io, aux_mutex, local_i2c, remote_i2c, _routing_table, busno, write, data) {
+            match dispatch!(io, aux_mutex, ddma_mutex, local_i2c, remote_i2c, _routing_table, busno, write, data) {
                 Ok(ack) => kern_send(io, &kern::I2cWriteReply { succeeded: true, ack: ack }),
                 Err(_) => kern_send(io, &kern::I2cWriteReply { succeeded: false, ack: false })
             }
         }
         &kern::I2cReadRequest { busno, ack } => {
-            match dispatch!(io, aux_mutex, local_i2c, remote_i2c, _routing_table, busno, read, ack) {
+            match dispatch!(io, aux_mutex, ddma_mutex, local_i2c, remote_i2c, _routing_table, busno, read, ack) {
                 Ok(data) => kern_send(io, &kern::I2cReadReply { succeeded: true, data: data }),
                 Err(_) => kern_send(io, &kern::I2cReadReply { succeeded: false, data: 0xff })
             }
         }
         &kern::I2cSwitchSelectRequest { busno, address, mask } => {
-            let succeeded = dispatch!(io, aux_mutex, local_i2c, remote_i2c, _routing_table, busno,
+            let succeeded = dispatch!(io, aux_mutex, ddma_mutex, local_i2c, remote_i2c, _routing_table, busno,
                 switch_select, address, mask).is_ok();
             kern_send(io, &kern::I2cBasicReply { succeeded: succeeded })
         }
 
         &kern::SpiSetConfigRequest { busno, flags, length, div, cs } => {
-            let succeeded = dispatch!(io, aux_mutex, local_spi, remote_spi, _routing_table, busno,
+            let succeeded = dispatch!(io, aux_mutex, ddma_mutex, local_spi, remote_spi, _routing_table, busno,
                 set_config, flags, length, div, cs).is_ok();
             kern_send(io, &kern::SpiBasicReply { succeeded: succeeded })
         },
         &kern::SpiWriteRequest { busno, data } => {
-            let succeeded = dispatch!(io, aux_mutex, local_spi, remote_spi, _routing_table, busno,
+            let succeeded = dispatch!(io, aux_mutex, ddma_mutex, local_spi, remote_spi, _routing_table, busno,
                 write, data).is_ok();
             kern_send(io, &kern::SpiBasicReply { succeeded: succeeded })
         }
         &kern::SpiReadRequest { busno } => {
-            match dispatch!(io, aux_mutex, local_spi, remote_spi, _routing_table, busno, read) {
+            match dispatch!(io, aux_mutex, ddma_mutex, local_spi, remote_spi, _routing_table, busno, read) {
                 Ok(data) => kern_send(io, &kern::SpiReadReply { succeeded: true, data: data }),
                 Err(_) => kern_send(io, &kern::SpiReadReply { succeeded: false, data: 0 })
             }

artiq/firmware/runtime/main.rs

@@ -182,6 +182,8 @@ fn startup() {
     drtio_routing::interconnect_disable_all();
     let aux_mutex = sched::Mutex::new();
 
+    let ddma_mutex = sched::Mutex::new();
+
     let mut scheduler = sched::Scheduler::new(interface);
     let io = scheduler.io();
 
@@ -189,20 +191,22 @@ fn startup() {
         io.spawn(4096, dhcp::dhcp_thread);
     }
 
-    rtio_mgt::startup(&io, &aux_mutex, &drtio_routing_table, &up_destinations);
+    rtio_mgt::startup(&io, &aux_mutex, &drtio_routing_table, &up_destinations, &ddma_mutex);
 
     io.spawn(4096, mgmt::thread);
     {
         let aux_mutex = aux_mutex.clone();
         let drtio_routing_table = drtio_routing_table.clone();
         let up_destinations = up_destinations.clone();
-        io.spawn(16384, move |io| { session::thread(io, &aux_mutex, &drtio_routing_table, &up_destinations) });
+        let ddma_mutex = ddma_mutex.clone();
+        io.spawn(16384, move |io| { session::thread(io, &aux_mutex, &drtio_routing_table, &up_destinations, &ddma_mutex) });
     }
     #[cfg(any(has_rtio_moninj, has_drtio))]
     {
         let aux_mutex = aux_mutex.clone();
+        let ddma_mutex = ddma_mutex.clone();
         let drtio_routing_table = drtio_routing_table.clone();
-        io.spawn(4096, move |io| { moninj::thread(io, &aux_mutex, &drtio_routing_table) });
+        io.spawn(4096, move |io| { moninj::thread(io, &aux_mutex, &ddma_mutex, &drtio_routing_table) });
     }
     #[cfg(has_rtio_analyzer)]
     io.spawn(4096, analyzer::thread);

artiq/firmware/runtime/moninj.rs

@@ -53,8 +53,8 @@ mod remote_moninj {
     use rtio_mgt::drtio;
     use sched::{Io, Mutex};
 
-    pub fn read_probe(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, channel: u16, probe: u8) -> u64 {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::MonitorRequest { 
+    pub fn read_probe(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, channel: u16, probe: u8) -> u64 {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::MonitorRequest { 
             destination: destination,
             channel: channel,
             probe: probe
@@ -67,7 +67,7 @@ mod remote_moninj {
         0
     }
 
-    pub fn inject(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, channel: u16, overrd: u8, value: u8) {
+    pub fn inject(io: &Io, aux_mutex: &Mutex, _ddma_mutex: &Mutex, linkno: u8, destination: u8, channel: u16, overrd: u8, value: u8) {
         let _lock = aux_mutex.lock(io).unwrap();
         drtioaux::send(linkno, &drtioaux::Packet::InjectionRequest {
             destination: destination,
@@ -77,8 +77,8 @@ mod remote_moninj {
         }).unwrap();
     }
 
-    pub fn read_injection_status(io: &Io, aux_mutex: &Mutex, linkno: u8, destination: u8, channel: u16, overrd: u8) -> u8 {
-        let reply = drtio::aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::InjectionStatusRequest {
+    pub fn read_injection_status(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, destination: u8, channel: u16, overrd: u8) -> u8 {
+        let reply = drtio::aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::InjectionStatusRequest {
             destination: destination,
             channel: channel,
             overrd: overrd
@@ -94,7 +94,7 @@ mod remote_moninj {
 
 #[cfg(has_drtio)]
 macro_rules! dispatch {
-    ($io:ident, $aux_mutex:ident, $routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
+    ($io:ident, $aux_mutex:ident, $ddma_mutex:ident, $routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
         let destination = ($channel >> 16) as u8;
         let channel = $channel as u16;
         let hop = $routing_table.0[destination as usize][0];
@@ -102,20 +102,20 @@ macro_rules! dispatch {
             local_moninj::$func(channel, $($param, )*)
         } else {
             let linkno = hop - 1;
-            remote_moninj::$func($io, $aux_mutex, linkno, destination, channel, $($param, )*)
+            remote_moninj::$func($io, $aux_mutex, $ddma_mutex, linkno, destination, channel, $($param, )*)
         }
     }}
 }
 
 #[cfg(not(has_drtio))]
 macro_rules! dispatch {
-    ($io:ident, $aux_mutex:ident, $routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
+    ($io:ident, $aux_mutex:ident, $ddma_mutex:ident, $routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
         let channel = $channel as u16;
         local_moninj::$func(channel, $($param, )*)
     }}
 }
 
-fn connection_worker(io: &Io, _aux_mutex: &Mutex, _routing_table: &drtio_routing::RoutingTable,
+fn connection_worker(io: &Io, _aux_mutex: &Mutex, _ddma_mutex: &Mutex, _routing_table: &drtio_routing::RoutingTable,
         mut stream: &mut TcpStream) -> Result<(), Error<SchedError>> {
     let mut probe_watch_list = BTreeMap::new();
     let mut inject_watch_list = BTreeMap::new();
@@ -144,9 +144,9 @@ fn connection_worker(io: &Io, _aux_mutex: &Mutex, _routing_table: &drtio_routing
                         let _ = inject_watch_list.remove(&(channel, overrd));
                     }
                 },
-                HostMessage::Inject { channel, overrd, value } => dispatch!(io, _aux_mutex, _routing_table, channel, inject, overrd, value),
+                HostMessage::Inject { channel, overrd, value } => dispatch!(io, _aux_mutex, _ddma_mutex, _routing_table, channel, inject, overrd, value),
                 HostMessage::GetInjectionStatus { channel, overrd } => {
-                    let value = dispatch!(io, _aux_mutex, _routing_table, channel, read_injection_status, overrd);
+                    let value = dispatch!(io, _aux_mutex, _ddma_mutex, _routing_table, channel, read_injection_status, overrd);
                     let reply = DeviceMessage::InjectionStatus {
                         channel: channel,
                         overrd: overrd,
@@ -163,7 +163,7 @@ fn connection_worker(io: &Io, _aux_mutex: &Mutex, _routing_table: &drtio_routing
 
         if clock::get_ms() > next_check {
             for (&(channel, probe), previous) in probe_watch_list.iter_mut() {
-                let current = dispatch!(io, _aux_mutex, _routing_table, channel, read_probe, probe);
+                let current = dispatch!(io, _aux_mutex, _ddma_mutex, _routing_table, channel, read_probe, probe);
                 if previous.is_none() || previous.unwrap() != current {
                     let message = DeviceMessage::MonitorStatus {
                         channel: channel,
@@ -178,7 +178,7 @@ fn connection_worker(io: &Io, _aux_mutex: &Mutex, _routing_table: &drtio_routing
                 }
             }
             for (&(channel, overrd), previous) in inject_watch_list.iter_mut() {
-                let current = dispatch!(io, _aux_mutex, _routing_table, channel, read_injection_status, overrd);
+                let current = dispatch!(io, _aux_mutex, _ddma_mutex, _routing_table, channel, read_injection_status, overrd);
                 if previous.is_none() || previous.unwrap() != current {
                     let message = DeviceMessage::InjectionStatus {
                         channel: channel,
@@ -199,18 +199,19 @@ fn connection_worker(io: &Io, _aux_mutex: &Mutex, _routing_table: &drtio_routing
     }
 }
 
-pub fn thread(io: Io, aux_mutex: &Mutex, routing_table: &Urc<RefCell<drtio_routing::RoutingTable>>) {
+pub fn thread(io: Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, routing_table: &Urc<RefCell<drtio_routing::RoutingTable>>) {
     let listener = TcpListener::new(&io, 2047);
     listener.listen(1383).expect("moninj: cannot listen");
 
     loop {
         let aux_mutex = aux_mutex.clone();
+        let ddma_mutex = ddma_mutex.clone();
         let routing_table = routing_table.clone();
         let stream = listener.accept().expect("moninj: cannot accept").into_handle();
         io.spawn(16384, move |io| {
             let routing_table = routing_table.borrow();
             let mut stream = TcpStream::from_handle(&io, stream);
-            match connection_worker(&io, &aux_mutex, &routing_table, &mut stream) {
+            match connection_worker(&io, &aux_mutex, &ddma_mutex, &routing_table, &mut stream) {
                 Ok(()) => {},
                 Err(err) => error!("moninj aborted: {}", err)
             }

artiq/firmware/runtime/rtio_clocking.rs

@@ -10,6 +10,7 @@ pub enum RtioClock {
     Int_100,
     Ext0_Bypass,
     Ext0_Synth0_10to125,
+    Ext0_Synth0_80to125,
     Ext0_Synth0_100to125,
     Ext0_Synth0_125to125,
 }
@@ -24,6 +25,7 @@ fn get_rtio_clock_cfg() -> RtioClock {
             Ok("ext0_bypass_125") => RtioClock::Ext0_Bypass,
             Ok("ext0_bypass_100") => RtioClock::Ext0_Bypass,
             Ok("ext0_synth0_10to125") => RtioClock::Ext0_Synth0_10to125,
+            Ok("ext0_synth0_80to125") => RtioClock::Ext0_Synth0_80to125,
             Ok("ext0_synth0_100to125") => RtioClock::Ext0_Synth0_100to125,
             Ok("ext0_synth0_125to125") => RtioClock::Ext0_Synth0_125to125,
             Ok("i") => {
@@ -44,6 +46,8 @@ fn get_rtio_clock_cfg() -> RtioClock {
             warn!("si5324_ext_ref and ext_ref_frequency compile-time options are deprecated. Please use the rtio_clock coreconfig settings instead.");
             #[cfg(all(rtio_frequency = "125.0", si5324_ext_ref, ext_ref_frequency = "10.0"))]
             return RtioClock::Ext0_Synth0_10to125;
+            #[cfg(all(rtio_frequency = "125.0", si5324_ext_ref, ext_ref_frequency = "80.0"))]
+            return RtioClock::Ext0_Synth0_80to125;
             #[cfg(all(rtio_frequency = "125.0", si5324_ext_ref, ext_ref_frequency = "100.0"))]
             return RtioClock::Ext0_Synth0_100to125;
             #[cfg(all(rtio_frequency = "125.0", si5324_ext_ref, ext_ref_frequency = "125.0"))]
@@ -110,6 +114,22 @@ fn setup_si5324_pll(cfg: RtioClock) {
                 SI5324_EXT_INPUT
             )
         },
+        RtioClock::Ext0_Synth0_80to125 => { // 125 MHz output from 80 MHz CLKINx reference, 611 Hz BW
+        info!("using 80MHz reference to make 125MHz RTIO clock with PLL");
+            (
+                si5324::FrequencySettings {
+                    n1_hs  : 4,
+                    nc1_ls : 10,
+                    n2_hs  : 10,
+                    n2_ls  : 250,
+                    n31    : 40,
+                    n32    : 40,
+                    bwsel  : 4,
+                    crystal_as_ckin2: false
+                },
+                SI5324_EXT_INPUT
+            )
+        },
         RtioClock::Ext0_Synth0_100to125 => { // 125MHz output, from 100MHz CLKINx reference, 586 Hz loop bandwidth
             info!("using 100MHz reference to make 125MHz RTIO clock with PLL");
             (

artiq/firmware/runtime/rtio_dma.rs

@@ -1,10 +1,188 @@
 use core::mem;
 use alloc::{vec::Vec, string::String, collections::btree_map::BTreeMap};
+use sched::{Io, Mutex};
 
 const ALIGNMENT: usize = 64;
 
+#[cfg(has_drtio)]
+pub mod remote_dma {
+    use super::*;
+    use board_artiq::drtio_routing::RoutingTable;
+    use rtio_mgt::drtio;
+    use board_misoc::clock;
+
+    #[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
+        }
+    }
+
+    // remote traces map. ID -> destination, trace pair
+    static mut TRACES: BTreeMap<u32, BTreeMap<u8, RemoteTrace>> = BTreeMap::new();
+
+    pub fn add_traces(io: &Io, ddma_mutex: &Mutex, id: u32, traces: BTreeMap<u8, Vec<u8>>) {
+        let _lock = ddma_mutex.lock(io);
+        let mut trace_map: BTreeMap<u8, RemoteTrace> = BTreeMap::new();
+        for (destination, trace) in traces {
+            trace_map.insert(destination, trace.into());
+        }
+        unsafe { TRACES.insert(id, trace_map); }
+    }
+
+    pub fn await_done(io: &Io, ddma_mutex: &Mutex, id: u32, timeout: u64) -> Result<RemoteState, &'static str> {
+        let max_time = clock::get_ms() + timeout as u64;
+        io.until(|| {
+            if clock::get_ms() > max_time {
+                return true;
+            }
+            if ddma_mutex.test_lock() {
+                // cannot lock again within io.until - scheduler guarantees
+                // that it will not be interrupted - so only test the lock
+                return false;
+            }
+            let traces = unsafe { TRACES.get(&id).unwrap() };
+            for (_dest, trace) in traces {
+                match trace.state {
+                    RemoteState::PlaybackEnded {error: _, channel: _, timestamp: _} => (),
+                    _ => return false
+                }
+            }
+            true
+        }).unwrap();
+        if clock::get_ms() > max_time {
+            error!("Remote DMA await done timed out");
+            return Err("Timed out waiting for results.");
+        }
+        // clear the internal state, and if there have been any errors, return one of them
+        let mut playback_state: RemoteState = RemoteState::PlaybackEnded { error: 0, channel: 0, timestamp: 0 };
+        {
+            let _lock = ddma_mutex.lock(io).unwrap();
+            let traces = unsafe { TRACES.get_mut(&id).unwrap() };
+            for (_dest, trace) in traces {
+                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 fn erase(io: &Io, aux_mutex: &Mutex, routing_table: &RoutingTable,
+            ddma_mutex: &Mutex, id: u32) {
+        let _lock = ddma_mutex.lock(io).unwrap();
+        let destinations = unsafe { TRACES.get(&id).unwrap() };
+        for destination in destinations.keys() {
+            match drtio::ddma_send_erase(io, aux_mutex, ddma_mutex, routing_table, id, *destination) {
+                Ok(_) => (),
+                Err(e) => error!("Error erasing trace on DMA: {}", e)
+            } 
+        }
+        unsafe { TRACES.remove(&id); }
+    }
+
+    pub fn upload_traces(io: &Io, aux_mutex: &Mutex, routing_table: &RoutingTable,
+            ddma_mutex: &Mutex, id: u32) {
+        let _lock = ddma_mutex.lock(io);
+        let traces = unsafe { TRACES.get_mut(&id).unwrap() };
+        for (destination, mut trace) in traces {
+            match drtio::ddma_upload_trace(io, aux_mutex, ddma_mutex, routing_table, id, *destination, trace.get_trace())
+            {
+                Ok(_) => trace.state = RemoteState::Loaded,
+                Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e)
+            }
+        }
+    }
+
+    pub fn playback(io: &Io, aux_mutex: &Mutex, routing_table: &RoutingTable,
+            ddma_mutex: &Mutex, id: u32, timestamp: u64) {
+        // triggers playback on satellites
+        let destinations = unsafe { 
+            let _lock = ddma_mutex.lock(io).unwrap();
+            TRACES.get(&id).unwrap() };
+        for (destination, trace) in destinations {
+            {
+                // need to drop the lock before sending the playback request to avoid a deadlock
+                // if a PlaybackStatus is returned from another satellite in the meanwhile.
+                let _lock = ddma_mutex.lock(io).unwrap();
+                if trace.state != RemoteState::Loaded {
+                    error!("Destination {} not ready for DMA, state: {:?}", *destination, trace.state);
+                    continue;
+                }
+            }
+            match drtio::ddma_send_playback(io, aux_mutex, ddma_mutex, routing_table, id, *destination, timestamp) {
+                Ok(_) => (),
+                Err(e) => error!("Error during remote DMA playback: {}", e)
+            }
+        }
+    }
+
+    pub fn playback_done(io: &Io, ddma_mutex: &Mutex, 
+            id: u32, destination: u8, error: u8, channel: u32, timestamp: u64) {
+        // called upon receiving PlaybackDone aux packet
+        let _lock = ddma_mutex.lock(io).unwrap();
+        let mut trace = unsafe { TRACES.get_mut(&id).unwrap().get_mut(&destination).unwrap() };
+        trace.state = RemoteState::PlaybackEnded {
+            error: error, 
+            channel: channel, 
+            timestamp: timestamp 
+        };
+    }
+
+    pub fn destination_changed(io: &Io, aux_mutex: &Mutex, routing_table: &RoutingTable,
+            ddma_mutex: &Mutex, destination: u8, up: bool) {
+        // update state of the destination, resend traces if it's up
+        let _lock = ddma_mutex.lock(io).unwrap();
+        let traces_iter = unsafe { TRACES.iter_mut() };
+        for (id, dest_traces) in traces_iter {
+            if let Some(trace) = dest_traces.get_mut(&destination) {
+                if up {
+                    match drtio::ddma_upload_trace(io, aux_mutex, ddma_mutex, routing_table, *id, destination, trace.get_trace())
+                    {
+                        Ok(_) => trace.state = RemoteState::Loaded,
+                        Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e)
+                    }
+                } else {
+                    trace.state = RemoteState::NotLoaded;
+                }
+            }
+        }
+    }
+
+    pub fn has_remote_traces(io: &Io, ddma_mutex: &Mutex, id: u32) -> bool {
+        let _lock = ddma_mutex.lock(io).unwrap();
+        let trace_list = unsafe { TRACES.get(&id).unwrap() };
+        !trace_list.is_empty()
+    }
+
+}
+
+
 #[derive(Debug)]
-struct Entry {
+struct LocalEntry {
     trace: Vec<u8>,
     padding_len: usize,
     duration: u64
@@ -12,7 +190,8 @@ struct Entry {
 
 #[derive(Debug)]
 pub struct Manager {
-    entries: BTreeMap<String, Entry>,
+    entries: BTreeMap<u32, LocalEntry>,
+    name_map: BTreeMap<String, u32>,
     recording_name: String,
     recording_trace: Vec<u8>
 }
@@ -21,59 +200,116 @@ impl Manager {
     pub fn new() -> Manager {
         Manager {
             entries: BTreeMap::new(),
-            recording_name: String::new(),
+            name_map: BTreeMap::new(),
             recording_trace: Vec::new(),
+            recording_name: String::new()
         }
     }
 
-    pub fn record_start(&mut self, name: &str) {
+    pub fn record_start(&mut self, name: &str) -> Option<u32> {
         self.recording_name = String::from(name);
         self.recording_trace = Vec::new();
-
-        // or we could needlessly OOM replacing a large trace
-        self.entries.remove(name);
+        if let Some(id) = self.name_map.get(&self.recording_name) {
+            // replacing a trace
+            let old_id = id.clone();
+            self.entries.remove(&id);
+            self.name_map.remove(&self.recording_name);
+            // return old ID
+            return Some(old_id);
+        }
+        return None;
     }
 
     pub fn record_append(&mut self, data: &[u8]) {
         self.recording_trace.extend_from_slice(data)
     }
 
-    pub fn record_stop(&mut self, duration: u64) {
+    pub fn record_stop(&mut self, duration: u64, _enable_ddma: bool,
+            _io: &Io, _ddma_mutex: &Mutex) -> u32 {
+        let mut local_trace = Vec::new();
+        let mut _remote_traces: BTreeMap<u8, Vec<u8>> = BTreeMap::new();
+
+        if _enable_ddma & cfg!(has_drtio) {
             let mut trace = Vec::new();
             mem::swap(&mut self.recording_trace, &mut trace);
             trace.push(0);
-        let data_len = trace.len();
+            // 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;
+            while trace[ptr] != 0 {
+                // ptr + 3 = tgt >> 24 (destination)
+                let len = trace[ptr] as usize;
+                let destination = trace[ptr+3];
+                if destination == 0 {
+                    local_trace.extend(&trace[ptr..ptr+len]);
+                }
+                else {
+                    if let Some(remote_trace) = _remote_traces.get_mut(&destination) {
+                        remote_trace.extend(&trace[ptr..ptr+len]);
+                    } else {
+                        _remote_traces.insert(destination, trace[ptr..ptr+len].to_vec());
+                    }
+                }
+                // and jump to the next event
+                ptr += len;
+            }
+        } else {
+            // with disabled DDMA, move the whole trace to local
+            mem::swap(&mut self.recording_trace, &mut local_trace);
+        }
 
-        // Realign.
-        trace.reserve(ALIGNMENT - 1);
-        let padding = ALIGNMENT - trace.as_ptr() as usize % ALIGNMENT;
+        local_trace.push(0);
+        let data_len = local_trace.len();
+        // Realign the local entry.
+        local_trace.reserve(ALIGNMENT - 1);
+        let padding = ALIGNMENT - local_trace.as_ptr() as usize % ALIGNMENT;
         let padding = if padding == ALIGNMENT { 0 } else { padding };
         for _ in 0..padding {
             // Vec guarantees that this will not reallocate
-            trace.push(0)
+            local_trace.push(0)
         }
         for i in 1..data_len + 1 {
-            trace[data_len + padding - i] = trace[data_len - i]
+            local_trace[data_len + padding - i] = local_trace[data_len - i]
         }
-
+        // trace ID is its pointer
+        let id = local_trace[padding..].as_ptr() as u32;
+        self.entries.insert(id, LocalEntry {
+            trace: local_trace,
+            padding_len: padding,
+            duration: duration,
+        });
         let mut name = String::new();
         mem::swap(&mut self.recording_name, &mut name);
-        self.entries.insert(name, Entry {
-            trace: trace,
-            padding_len: padding,
-            duration: duration
-        });
+        self.name_map.insert(name, id);
+
+        #[cfg(has_drtio)]
+        remote_dma::add_traces(_io, _ddma_mutex, id, _remote_traces);
+
+        id
     }
 
     pub fn erase(&mut self, name: &str) {
-        self.entries.remove(name);
+        if let Some(id) = self.name_map.get(name) {
+            self.entries.remove(&id);
+        }
+        self.name_map.remove(name);
+    }
+
+    #[cfg(has_drtio)]
+    pub fn get_id(&mut self, name: &str) -> Option<&u32> {
+        self.name_map.get(name)
     }
 
     pub fn with_trace<F, R>(&self, name: &str, f: F) -> R
             where F: FnOnce(Option<&[u8]>, u64) -> R {
-        match self.entries.get(name) {
+        if let Some(ptr) = self.name_map.get(name) {
+            match self.entries.get(ptr) {
                 Some(entry) => f(Some(&entry.trace[entry.padding_len..]), entry.duration),
                 None => f(None, 0)
             }
+        } else {
+            f(None, 0)
+        }
     }
 }

artiq/firmware/runtime/rtio_mgt.rs

@@ -1,5 +1,3 @@
-use alloc::collections::BTreeMap;
-use alloc::string::String;
 use core::cell::RefCell;
 use urc::Urc;
 use board_misoc::{csr, config};
@@ -9,26 +7,30 @@ use board_artiq::drtio_routing;
 use sched::Io;
 use sched::Mutex;
 use io::{Cursor, ProtoRead};
+use session_proto::{DeviceMap, resolve_channel_name, set_device_map};
 const ASYNC_ERROR_COLLISION: u8 = 1 << 0;
 const ASYNC_ERROR_BUSY: u8 = 1 << 1;
 const ASYNC_ERROR_SEQUENCE_ERROR: u8 = 1 << 2;
 
-static mut RTIO_DEVICE_MAP: BTreeMap<u32, String> = BTreeMap::new();
-
 #[cfg(has_drtio)]
 pub mod drtio {
     use super::*;
+    use alloc::vec::Vec;
     use drtioaux;
+    use proto_artiq::drtioaux_proto::DMA_TRACE_MAX_SIZE;
+    use rtio_dma::remote_dma;
 
     pub fn startup(io: &Io, aux_mutex: &Mutex,
             routing_table: &Urc<RefCell<drtio_routing::RoutingTable>>,
-            up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>) {
+            up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+            ddma_mutex: &Mutex) {
         let aux_mutex = aux_mutex.clone();
         let routing_table = routing_table.clone();
         let up_destinations = up_destinations.clone();
-        io.spawn(4096, move |io| {
+        let ddma_mutex = ddma_mutex.clone();
+        io.spawn(8192, move |io| {
             let routing_table = routing_table.borrow();
-            link_thread(io, &aux_mutex, &routing_table, &up_destinations);
+            link_thread(io, &aux_mutex, &routing_table, &up_destinations, &ddma_mutex);
         });
     }
 
@@ -57,14 +59,23 @@ pub mod drtio {
         }
     }
 
-    pub fn aux_transact(io: &Io, aux_mutex: &Mutex,
+    pub fn aux_transact(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex,
             linkno: u8, request: &drtioaux::Packet ) -> Result<drtioaux::Packet, &'static str> {
         let _lock = aux_mutex.lock(io).unwrap();
         drtioaux::send(linkno, request).unwrap();
-        recv_aux_timeout(io, linkno, 200)
+        loop {
+            let reply = recv_aux_timeout(io, linkno, 200);
+            match reply {
+                Ok(drtioaux::Packet::DmaPlaybackStatus { id, destination, error, channel, timestamp }) => {
+                    remote_dma::playback_done(io, ddma_mutex, id, destination, error, channel, timestamp);
+                },
+                Ok(packet) => return Ok(packet),
+                Err(e) => return Err(e)
+            }
+        }
     }
 
-    fn ping_remote(io: &Io, aux_mutex: &Mutex, linkno: u8) -> u32 {
+    fn ping_remote(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8) -> u32 {
         let mut count = 0;
         loop {
             if !link_rx_up(linkno) {
@@ -74,7 +85,7 @@ pub mod drtio {
             if count > 100 {
                 return 0;
             }
-            let reply = aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::EchoRequest);
+            let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::EchoRequest);
             match reply {
                 Ok(drtioaux::Packet::EchoReply) => {
                     // make sure receive buffer is drained
@@ -110,10 +121,10 @@ pub mod drtio {
         }
     }
 
-    fn load_routing_table(io: &Io, aux_mutex: &Mutex, linkno: u8, routing_table: &drtio_routing::RoutingTable)
-            -> Result<(), &'static str> {
+    fn load_routing_table(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, 
+        linkno: u8, routing_table: &drtio_routing::RoutingTable) -> Result<(), &'static str> {
         for i in 0..drtio_routing::DEST_COUNT {
-            let reply = aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::RoutingSetPath {
+            let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::RoutingSetPath {
                 destination: i as u8,
                 hops: routing_table.0[i]
             })?;
@@ -124,8 +135,8 @@ pub mod drtio {
         Ok(())
     }
 
-    fn set_rank(io: &Io, aux_mutex: &Mutex, linkno: u8, rank: u8) -> Result<(), &'static str> {
-        let reply = aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::RoutingSetRank {
+    fn set_rank(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex, linkno: u8, rank: u8) -> Result<(), &'static str> {
+        let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::RoutingSetRank {
             rank: rank
         })?;
         if reply != drtioaux::Packet::RoutingAck {
@@ -147,9 +158,12 @@ pub mod drtio {
         }
     }
 
-    fn process_unsolicited_aux(io: &Io, aux_mutex: &Mutex, linkno: u8) {
+    fn process_unsolicited_aux(io: &Io, aux_mutex: &Mutex, linkno: u8, ddma_mutex: &Mutex) {
         let _lock = aux_mutex.lock(io).unwrap();
         match drtioaux::recv(linkno) {
+            Ok(Some(drtioaux::Packet::DmaPlaybackStatus { id, destination, error, channel, timestamp })) => {
+                remote_dma::playback_done(io, ddma_mutex, id, destination, error, channel, timestamp);
+            }
             Ok(Some(packet)) => warn!("[LINK#{}] unsolicited aux packet: {:?}", linkno, packet),
             Ok(None) => (),
             Err(_) => warn!("[LINK#{}] aux packet error", linkno)
@@ -198,7 +212,8 @@ pub mod drtio {
 
     fn destination_survey(io: &Io, aux_mutex: &Mutex, routing_table: &drtio_routing::RoutingTable,
             up_links: &[bool],
-            up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>) {
+            up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+            ddma_mutex: &Mutex) {
         for destination in 0..drtio_routing::DEST_COUNT {
             let hop = routing_table.0[destination][0];
             let destination = destination as u8;
@@ -212,12 +227,14 @@ pub mod drtio {
                 let linkno = hop - 1;
                 if destination_up(up_destinations, destination) {
                     if up_links[linkno as usize] {
-                        let reply = aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::DestinationStatusRequest {
+                        let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::DestinationStatusRequest {
                             destination: destination
                         });
                         match reply {
-                            Ok(drtioaux::Packet::DestinationDownReply) =>
-                                destination_set_up(routing_table, up_destinations, destination, false),
+                            Ok(drtioaux::Packet::DestinationDownReply) => {
+                                destination_set_up(routing_table, up_destinations, destination, false);
+                                remote_dma::destination_changed(io, aux_mutex, routing_table, ddma_mutex, destination, false);
+                            }
                             Ok(drtioaux::Packet::DestinationOkReply) => (),
                             Ok(drtioaux::Packet::DestinationSequenceErrorReply { channel }) => {
                                 error!("[DEST#{}] RTIO sequence error involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32));
@@ -236,10 +253,11 @@ pub mod drtio {
                         }
                     } else {
                         destination_set_up(routing_table, up_destinations, destination, false);
+                        remote_dma::destination_changed(io, aux_mutex, routing_table, ddma_mutex, destination, false);
                     }
                 } else {
                     if up_links[linkno as usize] {
-                        let reply = aux_transact(io, aux_mutex, linkno, &drtioaux::Packet::DestinationStatusRequest {
+                        let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::DestinationStatusRequest {
                             destination: destination
                         });
                         match reply {
@@ -247,6 +265,7 @@ pub mod drtio {
                             Ok(drtioaux::Packet::DestinationOkReply) => {
                                 destination_set_up(routing_table, up_destinations, destination, true);
                                 init_buffer_space(destination as u8, linkno);
+                                remote_dma::destination_changed(io, aux_mutex, routing_table, ddma_mutex, destination, true);
                             },
                             Ok(packet) => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet),
                             Err(e) => error!("[DEST#{}] communication failed ({})", destination, e)
@@ -259,7 +278,8 @@ pub mod drtio {
 
     pub fn link_thread(io: Io, aux_mutex: &Mutex,
             routing_table: &drtio_routing::RoutingTable,
-            up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>) {
+            up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+            ddma_mutex: &Mutex) {
         let mut up_links = [false; csr::DRTIO.len()];
         loop {
             for linkno in 0..csr::DRTIO.len() {
@@ -267,7 +287,7 @@ pub mod drtio {
                 if up_links[linkno as usize] {
                     /* link was previously up */
                     if link_rx_up(linkno) {
-                        process_unsolicited_aux(&io, aux_mutex, linkno);
+                        process_unsolicited_aux(&io, aux_mutex, linkno, ddma_mutex);
                         process_local_errors(linkno);
                     } else {
                         info!("[LINK#{}] link is down", linkno);
@@ -277,17 +297,17 @@ pub mod drtio {
                     /* link was previously down */
                     if link_rx_up(linkno) {
                         info!("[LINK#{}] link RX became up, pinging", linkno);
-                        let ping_count = ping_remote(&io, aux_mutex, linkno);
+                        let ping_count = ping_remote(&io, aux_mutex, ddma_mutex, linkno);
                         if ping_count > 0 {
                             info!("[LINK#{}] remote replied after {} packets", linkno, ping_count);
                             up_links[linkno as usize] = true;
                             if let Err(e) = sync_tsc(&io, aux_mutex, linkno) {
                                 error!("[LINK#{}] failed to sync TSC ({})", linkno, e);
                             }
-                            if let Err(e) = load_routing_table(&io, aux_mutex, linkno, routing_table) {
+                            if let Err(e) = load_routing_table(&io, aux_mutex, ddma_mutex, linkno, routing_table) {
                                 error!("[LINK#{}] failed to load routing table ({})", linkno, e);
                             }
-                            if let Err(e) = set_rank(&io, aux_mutex, linkno, 1) {
+                            if let Err(e) = set_rank(&io, aux_mutex, ddma_mutex, linkno, 1) {
                                 error!("[LINK#{}] failed to set rank ({})", linkno, e);
                             }
                             info!("[LINK#{}] link initialization completed", linkno);
@@ -297,12 +317,12 @@ pub mod drtio {
                     }
                 }
             }
-            destination_survey(&io, aux_mutex, routing_table, &up_links, up_destinations);
+            destination_survey(&io, aux_mutex, routing_table, &up_links, up_destinations, ddma_mutex);
             io.sleep(200).unwrap();
         }
     }
 
-    pub fn reset(io: &Io, aux_mutex: &Mutex) {
+    pub fn reset(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex) {
         for linkno in 0..csr::DRTIO.len() {
             unsafe {
                 (csr::DRTIO[linkno].reset_write)(1);
@@ -318,7 +338,7 @@ pub mod drtio {
         for linkno in 0..csr::DRTIO.len() {
             let linkno = linkno as u8;
             if link_rx_up(linkno) {
-                let reply = aux_transact(io, aux_mutex, linkno,
+                let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno,
                     &drtioaux::Packet::ResetRequest);
                 match reply {
                     Ok(drtioaux::Packet::ResetAck) => (),
@@ -328,6 +348,62 @@ pub mod drtio {
             }
         }
     }
+
+    pub fn ddma_upload_trace(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex,
+            routing_table: &drtio_routing::RoutingTable,
+            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(io, aux_mutex, ddma_mutex, linkno, 
+                &drtioaux::Packet::DmaAddTraceRequest {
+                    id: id, destination: destination, last: last, length: len as u16, trace: trace_slice});
+            match reply {
+                Ok(drtioaux::Packet::DmaAddTraceReply { succeeded: true }) => (),
+                Ok(drtioaux::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 fn ddma_send_erase(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex,
+            routing_table: &drtio_routing::RoutingTable, 
+            id: u32, destination: u8) -> Result<(), &'static str> {
+        let linkno = routing_table.0[destination as usize][0] - 1;
+        let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, 
+            &drtioaux::Packet::DmaRemoveTraceRequest { id: id, destination: destination });
+        match reply {
+            Ok(drtioaux::Packet::DmaRemoveTraceReply { succeeded: true }) => Ok(()),
+            Ok(drtioaux::Packet::DmaRemoveTraceReply { succeeded: false }) => Err("satellite DMA erase error"),
+            Ok(_) => Err("erasing trace failed, unexpected aux packet"),
+            Err(_) => Err("erasing trace failed, aux error")
+        }
+    }
+
+    pub fn ddma_send_playback(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex,
+            routing_table: &drtio_routing::RoutingTable,
+            id: u32, destination: u8, timestamp: u64) -> Result<(), &'static str> {
+        let linkno = routing_table.0[destination as usize][0] - 1;
+        let reply = aux_transact(io, aux_mutex, ddma_mutex, linkno, &drtioaux::Packet::DmaPlaybackRequest{
+                id: id, destination: destination, timestamp: timestamp });
+        match reply {
+            Ok(drtioaux::Packet::DmaPlaybackReply { succeeded: true }) => return Ok(()),
+            Ok(drtioaux::Packet::DmaPlaybackReply { succeeded: false }) =>
+                    return Err("error on DMA playback request"),
+            Ok(_) => return Err("received unexpected aux packet during DMA playback"),
+            Err(_) => return Err("aux error on DMA playback")
+        }
+    }
+
 }
 
 #[cfg(not(has_drtio))]
@@ -336,8 +412,9 @@ pub mod drtio {
 
     pub fn startup(_io: &Io, _aux_mutex: &Mutex,
         _routing_table: &Urc<RefCell<drtio_routing::RoutingTable>>,
-        _up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>) {}
-    pub fn reset(_io: &Io, _aux_mutex: &Mutex) {}
+        _up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+        _ddma_mutex: &Mutex) {}
+    pub fn reset(_io: &Io, _aux_mutex: &Mutex, _ddma_mutex: &Mutex) {}
 }
 
 static mut SEEN_ASYNC_ERRORS: u8 = 0;
@@ -371,16 +448,16 @@ fn async_error_thread(io: Io) {
     }
 }
 
-fn read_device_map() -> BTreeMap<u32, String> {
-    let mut device_map: BTreeMap<u32, String> = BTreeMap::new();
+fn read_device_map() -> DeviceMap {
+    let mut device_map: DeviceMap = DeviceMap::new();
     config::read("device_map", |value: Result<&[u8], config::Error>| {
         let mut bytes = match value {
             Ok(val) => if val.len() > 0 { Cursor::new(val) } else {
-                error!("read_device_map: `device_map` was not found in the config");
+                warn!("device map not found in config, device names will not be available in RTIO error messages");
                 return;
             },
             Err(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);
                 return;
             }
         };
@@ -389,7 +466,7 @@ fn read_device_map() -> BTreeMap<u32, String> {
             let channel = bytes.read_u32().unwrap();
             let device_name= bytes.read_string().unwrap();
             if let Some(old_entry) = device_map.insert(channel, device_name.clone()) {
-                error!("conflicting entries for channel {}: `{}` and `{}`",
+                warn!("conflicting device map entries for RTIO channel {}: '{}' and '{}'",
                        channel, old_entry, device_name);
             }
         }
@@ -397,31 +474,21 @@ fn read_device_map() -> BTreeMap<u32, String> {
     device_map
 }
 
-fn _resolve_channel_name(channel: u32, device_map: &BTreeMap<u32, String>) -> String {
-    match device_map.get(&channel) {
-        Some(val) => val.clone(),
-        None => String::from("unknown")
-    }
-}
-
-pub fn resolve_channel_name(channel: u32) -> String {
-    _resolve_channel_name(channel, unsafe{&RTIO_DEVICE_MAP})
-}
-
 pub fn startup(io: &Io, aux_mutex: &Mutex,
         routing_table: &Urc<RefCell<drtio_routing::RoutingTable>>,
-        up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>) {
-    unsafe { RTIO_DEVICE_MAP = read_device_map(); }
-    drtio::startup(io, aux_mutex, routing_table, up_destinations);
+        up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+        ddma_mutex: &Mutex) {
+    set_device_map(read_device_map());
+    drtio::startup(io, aux_mutex, routing_table, up_destinations, ddma_mutex);
     unsafe {
         csr::rtio_core::reset_phy_write(1);
     }
     io.spawn(4096, async_error_thread);
 }
 
-pub fn reset(io: &Io, aux_mutex: &Mutex) {
+pub fn reset(io: &Io, aux_mutex: &Mutex, ddma_mutex: &Mutex) {
     unsafe {
         csr::rtio_core::reset_write(1);
     }
-    drtio::reset(io, aux_mutex)
+    drtio::reset(io, aux_mutex, ddma_mutex)
 }

artiq/firmware/runtime/sched.rs

@@ -301,6 +301,10 @@ impl Mutex {
         self.0.set(true);
         Ok(MutexGuard(&*self.0))
     }
+
+    pub fn test_lock<'a>(&'a self) -> bool {
+        self.0.get()
+    }
 }
 
 pub struct MutexGuard<'a>(&'a Cell<bool>);

artiq/firmware/runtime/session.rs

@@ -1,4 +1,4 @@
-use core::{mem, str, cell::{Cell, RefCell}, fmt::Write as FmtWrite, slice};
+use core::{mem, str, cell::{Cell, RefCell}, fmt::Write as FmtWrite};
 use alloc::{vec::Vec, string::String};
 use byteorder::{ByteOrder, NativeEndian};
 use cslice::CSlice;
@@ -10,7 +10,9 @@ use urc::Urc;
 use sched::{ThreadHandle, Io, Mutex, TcpListener, TcpStream, Error as SchedError};
 use rtio_clocking;
 use rtio_dma::Manager as DmaManager;
-use rtio_mgt::{get_async_errors, resolve_channel_name};
+#[cfg(has_drtio)]
+use rtio_dma::remote_dma;
+use rtio_mgt::get_async_errors;
 use cache::Cache;
 use kern_hwreq;
 use board_artiq::drtio_routing;
@@ -144,9 +146,9 @@ fn host_write<W>(writer: &mut W, reply: host::Reply) -> Result<(), IoError<W::Wr
 pub fn kern_send(io: &Io, request: &kern::Message) -> Result<(), Error<SchedError>> {
     match request {
         &kern::LoadRequest(_) => debug!("comm->kern LoadRequest(...)"),
-        &kern::DmaRetrieveReply { trace, duration } => {
+        &kern::DmaRetrieveReply { trace, duration, uses_ddma } => {
             if trace.map(|data| data.len() > 100).unwrap_or(false) {
-                debug!("comm->kern DmaRetrieveReply {{ trace: ..., duration: {:?} }}", duration)
+                debug!("comm->kern DmaRetrieveReply {{ trace: ..., duration: {:?}, uses_ddma: {} }}", duration, uses_ddma)
             } else {
                 debug!("comm->kern {:?}", request)
             }
@@ -329,7 +331,7 @@ fn process_host_message(io: &Io,
 fn process_kern_message(io: &Io, aux_mutex: &Mutex,
                         routing_table: &drtio_routing::RoutingTable,
                         up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
-                        mut stream: Option<&mut TcpStream>,
+                        ddma_mutex: &Mutex, mut stream: Option<&mut TcpStream>,
                         session: &mut Session) -> Result<bool, Error<SchedError>> {
     kern_recv_notrace(io, |request| {
         match (request, session.kernel_state) {
@@ -347,7 +349,7 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
 
         kern_recv_dotrace(request);
 
-        if kern_hwreq::process_kern_hwreq(io, aux_mutex, routing_table, up_destinations, request)? {
+        if kern_hwreq::process_kern_hwreq(io, aux_mutex, ddma_mutex, routing_table, up_destinations, request)? {
             return Ok(false)
         }
 
@@ -368,30 +370,71 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
             }
 
             &kern::DmaRecordStart(name) => {
-                session.congress.dma_manager.record_start(name);
+                if let Some(_id) = session.congress.dma_manager.record_start(name) {
+                    // replace the record
+                    #[cfg(has_drtio)]
+                    remote_dma::erase(io, aux_mutex, routing_table, ddma_mutex, _id);
+                }
                 kern_acknowledge()
             }
             &kern::DmaRecordAppend(data) => {
                 session.congress.dma_manager.record_append(data);
                 kern_acknowledge()
             }
-            &kern::DmaRecordStop { duration } => {
-                session.congress.dma_manager.record_stop(duration);
+            &kern::DmaRecordStop { duration, enable_ddma } => {
+                let _id = session.congress.dma_manager.record_stop(duration, enable_ddma, io, ddma_mutex);
+                #[cfg(has_drtio)]
+                if enable_ddma {
+                    remote_dma::upload_traces(io, aux_mutex, routing_table, ddma_mutex, _id);
+                }
                 cache::flush_l2_cache();
                 kern_acknowledge()
             }
             &kern::DmaEraseRequest { name } => {
+                #[cfg(has_drtio)]
+                if let Some(id) = session.congress.dma_manager.get_id(name) {
+                    remote_dma::erase(io, aux_mutex, routing_table, ddma_mutex, *id);
+                }
                 session.congress.dma_manager.erase(name);
                 kern_acknowledge()
             }
             &kern::DmaRetrieveRequest { name } => {
                 session.congress.dma_manager.with_trace(name, |trace, duration| {
+                    #[cfg(has_drtio)]
+                    let uses_ddma = match trace {
+                        Some(trace) => remote_dma::has_remote_traces(io, aux_mutex, trace.as_ptr() as u32),
+                        None => false
+                    };
+                    #[cfg(not(has_drtio))]
+                    let uses_ddma = false;
                     kern_send(io, &kern::DmaRetrieveReply {
                         trace:    trace,
-                        duration: duration
+                        duration: duration,
+                        uses_ddma: uses_ddma,
                     })
                 })
             }
+            &kern::DmaStartRemoteRequest { id: _id, timestamp: _timestamp } => {
+                #[cfg(has_drtio)]
+                remote_dma::playback(io, aux_mutex, routing_table, ddma_mutex, _id as u32, _timestamp as u64);
+                kern_acknowledge()
+            }
+            &kern::DmaAwaitRemoteRequest { id: _id } => {
+                #[cfg(has_drtio)]
+                let reply = match remote_dma::await_done(io, ddma_mutex, _id as u32, 10_000) {
+                    Ok(remote_dma::RemoteState::PlaybackEnded { error, channel, timestamp }) =>
+                        kern::DmaAwaitRemoteReply {
+                            timeout: false,
+                            error: error,
+                            channel: channel,
+                            timestamp: timestamp
+                        },
+                    _ => kern::DmaAwaitRemoteReply { timeout: true, error: 0, channel: 0, timestamp: 0},
+                };
+                #[cfg(not(has_drtio))]
+                let reply = kern::DmaAwaitRemoteReply { timeout: false, error: 0, channel: 0, timestamp: 0};
+                kern_send(io, &reply)
+            }
 
             &kern::RpcSend { async, service, tag, data } => {
                 match stream {
@@ -449,29 +492,6 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
                 session.kernel_state = KernelState::Absent;
                 unsafe { session.congress.cache.unborrow() }
 
-                let exceptions_with_channel: Vec<Option<eh::eh_artiq::Exception>> = exceptions.iter()
-                    .map(|exception| {
-                        if let Some(exn) = exception {
-                            if exn.message.len() == usize::MAX {  // host string
-                                Some(exn.clone())
-                            } else {
-                                let msg = str::from_utf8(unsafe { slice::from_raw_parts(exn.message.as_ptr(), exn.message.len()) })
-                                    .unwrap()
-                                    .replace("{rtio_channel_info:0}", &format!("0x{:04x}:{}", exn.param[0], resolve_channel_name(exn.param[0] as u32)));
-                                Some(eh::eh_artiq::Exception {
-                                    id: exn.id,
-                                    file: exn.file,
-                                    line: exn.line,
-                                    column: exn.column,
-                                    function: exn.function,
-                                    message: unsafe { CSlice::new(msg.as_ptr(), msg.len()) },
-                                    param: exn.param,
-                                })
-                            }
-                        } else { None }
-                    })
-                    .collect();
-
                 match stream {
                     None => {
                         error!("exception in flash kernel");
@@ -482,7 +502,7 @@ fn process_kern_message(io: &Io, aux_mutex: &Mutex,
                     },
                     Some(ref mut stream) => {
                         host_write(stream, host::Reply::KernelException {
-                            exceptions: &exceptions_with_channel,
+                            exceptions: exceptions,
                             stack_pointers: stack_pointers,
                             backtrace: backtrace,
                             async_errors: unsafe { get_async_errors() }
@@ -511,7 +531,7 @@ fn process_kern_queued_rpc(stream: &mut TcpStream,
 fn host_kernel_worker(io: &Io, aux_mutex: &Mutex,
                       routing_table: &drtio_routing::RoutingTable,
                       up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
-                      stream: &mut TcpStream,
+                      ddma_mutex: &Mutex, stream: &mut TcpStream,
                       congress: &mut Congress) -> Result<(), Error<SchedError>> {
     let mut session = Session::new(congress);
 
@@ -529,6 +549,7 @@ fn host_kernel_worker(io: &Io, aux_mutex: &Mutex,
         if mailbox::receive() != 0 {
             process_kern_message(io, aux_mutex,
                 routing_table, up_destinations,
+                ddma_mutex,
                 Some(stream), &mut session)?;
         }
 
@@ -546,7 +567,7 @@ fn host_kernel_worker(io: &Io, aux_mutex: &Mutex,
 fn flash_kernel_worker(io: &Io, aux_mutex: &Mutex,
                        routing_table: &drtio_routing::RoutingTable,
                        up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
-                       congress: &mut Congress,
+                       ddma_mutex: &Mutex, congress: &mut Congress,
                        config_key: &str) -> Result<(), Error<SchedError>> {
     let mut session = Session::new(congress);
 
@@ -568,7 +589,7 @@ fn flash_kernel_worker(io: &Io, aux_mutex: &Mutex,
         }
 
         if mailbox::receive() != 0 {
-            if process_kern_message(io, aux_mutex, routing_table, up_destinations, None, &mut session)? {
+            if process_kern_message(io, aux_mutex, routing_table, up_destinations, ddma_mutex, None, &mut session)? {
                 return Ok(())
             }
         }
@@ -598,7 +619,8 @@ fn respawn<F>(io: &Io, handle: &mut Option<ThreadHandle>, f: F)
 
 pub fn thread(io: Io, aux_mutex: &Mutex,
         routing_table: &Urc<RefCell<drtio_routing::RoutingTable>>,
-        up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>) {
+        up_destinations: &Urc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
+        ddma_mutex: &Mutex) {
     let listener = TcpListener::new(&io, 65535);
     listener.listen(1381).expect("session: cannot listen");
     info!("accepting network sessions");
@@ -609,7 +631,7 @@ pub fn thread(io: Io, aux_mutex: &Mutex,
     {
         let mut congress = congress.borrow_mut();
         info!("running startup kernel");
-        match flash_kernel_worker(&io, &aux_mutex, &routing_table.borrow(), &up_destinations, &mut congress, "startup_kernel") {
+        match flash_kernel_worker(&io, &aux_mutex, &routing_table.borrow(), &up_destinations, ddma_mutex, &mut congress, "startup_kernel") {
             Ok(()) =>
                 info!("startup kernel finished"),
             Err(Error::KernelNotFound) =>
@@ -649,12 +671,13 @@ pub fn thread(io: Io, aux_mutex: &Mutex,
             let routing_table = routing_table.clone();
             let up_destinations = up_destinations.clone();
             let congress = congress.clone();
+            let ddma_mutex = ddma_mutex.clone();
             let stream = stream.into_handle();
             respawn(&io, &mut kernel_thread, move |io| {
                 let routing_table = routing_table.borrow();
                 let mut congress = congress.borrow_mut();
                 let mut stream = TcpStream::from_handle(&io, stream);
-                match host_kernel_worker(&io, &aux_mutex, &routing_table, &up_destinations, &mut stream, &mut *congress) {
+                match host_kernel_worker(&io, &aux_mutex, &routing_table, &up_destinations, &ddma_mutex, &mut stream, &mut *congress) {
                     Ok(()) => (),
                     Err(Error::Protocol(host::Error::Io(IoError::UnexpectedEnd))) =>
                         info!("connection closed"),
@@ -677,10 +700,11 @@ pub fn thread(io: Io, aux_mutex: &Mutex,
             let routing_table = routing_table.clone();
             let up_destinations = up_destinations.clone();
             let congress = congress.clone();
+            let ddma_mutex = ddma_mutex.clone();
             respawn(&io, &mut kernel_thread, move |io| {
                 let routing_table = routing_table.borrow();
                 let mut congress = congress.borrow_mut();
-                match flash_kernel_worker(&io, &aux_mutex, &routing_table, &up_destinations, &mut *congress, "idle_kernel") {
+                match flash_kernel_worker(&io, &aux_mutex, &routing_table, &up_destinations, &ddma_mutex, &mut *congress, "idle_kernel") {
                     Ok(()) =>
                         info!("idle kernel finished, standing by"),
                     Err(Error::Protocol(host::Error::Io(

artiq/firmware/satman/dma.rs

@@ -1,4 +1,4 @@
-use board_misoc::csr;
+use board_misoc::{csr, cache::flush_l2_cache};
 use alloc::{vec::Vec, collections::btree_map::BTreeMap};
 
 const ALIGNMENT: usize = 64;
@@ -52,7 +52,19 @@ impl Manager {
 
     pub fn add(&mut self, id: u32, last: bool, trace: &[u8], trace_len: usize) -> Result<(), Error> {
         let entry = match self.entries.get_mut(&id) {
-            Some(entry) => entry,
+            Some(entry) => {
+                if entry.complete {
+                    // replace entry
+                    self.entries.remove(&id);
+                    self.entries.insert(id, Entry {
+                        trace: Vec::new(),
+                        padding_len: 0,
+                        complete: false });
+                    self.entries.get_mut(&id).unwrap()
+                } else {
+                    entry
+                }
+            },
             None => {
                 self.entries.insert(id, Entry {
                     trace: Vec::new(),
@@ -80,6 +92,7 @@ impl Manager {
             }
             entry.complete = true;
             entry.padding_len = padding;
+            flush_l2_cache();
         }
         Ok(())
     }

artiq/firmware/satman/main.rs

@@ -301,19 +301,22 @@ fn process_aux_packet(_manager: &mut DmaManager, _repeaters: &mut [repeater::Rep
             }
         }
         #[cfg(has_rtio_dma)]
-        drtioaux::Packet::DmaAddTraceRequest { id, last, length, trace } => {
+        drtioaux::Packet::DmaAddTraceRequest { destination: _destination, id, last, length, trace } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet);
             let succeeded = _manager.add(id, last, &trace, length as usize).is_ok();
             drtioaux::send(0,
                 &drtioaux::Packet::DmaAddTraceReply { succeeded: succeeded })
         }
         #[cfg(has_rtio_dma)]
-        drtioaux::Packet::DmaRemoveTraceRequest { id } => {
+        drtioaux::Packet::DmaRemoveTraceRequest { destination: _destination, id } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet);
             let succeeded = _manager.erase(id).is_ok();
             drtioaux::send(0,
                 &drtioaux::Packet::DmaRemoveTraceReply { succeeded: succeeded })
         }
         #[cfg(has_rtio_dma)]
-        drtioaux::Packet::DmaPlaybackRequest { id, timestamp } => {
+        drtioaux::Packet::DmaPlaybackRequest { destination: _destination, id, timestamp } => {
+            forward!(_routing_table, _destination, *_rank, _repeaters, &packet);
             let succeeded = _manager.playback(id, timestamp).is_ok();
             drtioaux::send(0,
                 &drtioaux::Packet::DmaPlaybackReply { succeeded: succeeded })
@@ -571,8 +574,9 @@ pub extern fn main() -> i32 {
                 }
             }
             if let Some(status) = dma_manager.check_state() {
+                info!("playback done, error: {}, channel: {}, timestamp: {}", status.error, status.channel, status.timestamp);
                 if let Err(e) = drtioaux::send(0, &drtioaux::Packet::DmaPlaybackStatus { 
-                    id: status.id, error: status.error, channel: status.channel, timestamp: status.timestamp }) {
+                    destination: rank, id: status.id, error: status.error, channel: status.channel, timestamp: status.timestamp }) {
                     error!("error sending DMA playback status: {}", e);
                 }
             }

artiq/firmware/satman/satman.ld

@@ -12,6 +12,7 @@ SECTIONS
     .text :
     {
         *(.text .text.*)
+        . = ALIGN(0x40000);
     } > main_ram
 
     .eh_frame :

artiq/frontend/afws_client.py

@@ -13,17 +13,6 @@ from getpass import getpass
 from tqdm import tqdm
 
 
-def get_artiq_cert():
-    try:
-        import artiq
-    except ImportError:
-        return None
-    filename = os.path.join(os.path.dirname(artiq.__file__), "afws.pem")
-    if not os.path.isfile(filename):
-        return None
-    return filename
-
-
 def get_artiq_rev():
     try:
         import artiq
@@ -54,6 +43,7 @@ class Client:
     def __init__(self, server, port, cafile):
         self.ssl_context = ssl.create_default_context(cafile=cafile)
         self.raw_socket = socket.create_connection((server, port))
+        self.init_websocket(server)
         try:
             self.socket = self.ssl_context.wrap_socket(self.raw_socket, server_hostname=server)
         except:
@@ -61,6 +51,19 @@ class Client:
             raise
         self.fsocket = self.socket.makefile("rwb")
 
+    def init_websocket(self, server):
+        self.raw_socket.sendall("GET / HTTP/1.1\r\nHost: {}\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"
+            .format(server).encode())
+        crlf_count = 0
+        while crlf_count < 4:
+            char = self.raw_socket.recv(1)
+            if not char:
+                return ValueError("Connection closed during WebSocket initialization")
+            if char == b"\r" or char == b"\n":
+                crlf_count += 1
+            else:
+                crlf_count = 0
+
     def close(self):
         self.socket.close()
         self.raw_socket.close()
@@ -160,9 +163,9 @@ class Client:
 
 def main():
     parser = argparse.ArgumentParser()
-    parser.add_argument("--server", default="nixbld.m-labs.hk", help="server to connect to (default: %(default)s)")
-    parser.add_argument("--port", default=7402, type=int, help="port to connect to (default: %(default)d)")
-    parser.add_argument("--cert", default=None, help="SSL certificate file used to authenticate server (default: afws.pem in ARTIQ)")
+    parser.add_argument("--server", default="afws.m-labs.hk", help="server to connect to (default: %(default)s)")
+    parser.add_argument("--port", default=80, type=int, help="port to connect to (default: %(default)d)")
+    parser.add_argument("--cert", default=None, help="SSL certificate file used to authenticate server (default: use system certificates)")
     parser.add_argument("username", help="user name for logging into AFWS")
     action = parser.add_subparsers(dest="action")
     action.required = True
@@ -181,36 +184,10 @@ def main():
     act_get_json.add_argument("-f", "--force", action="store_true", help="overwrite file if it already exists")
     args = parser.parse_args()
 
-    cert = args.cert
-    if cert is None:
-        cert = get_artiq_cert()
-    if cert is None:
-        print("SSL certificate not found in ARTIQ. Specify manually using --cert.")
-        sys.exit(1)
-
-    if args.action == "passwd":
-        password = getpass("Current password: ")
-    else:
-        password = getpass()
-
-    client = Client(args.server, args.port, cert)
+    client = Client(args.server, args.port, args.cert)
     try:
-        if not client.login(args.username, password):
-            print("Login failed")
-            sys.exit(1)
-        print("Logged in successfully.")
-        if args.action == "passwd":
-            print("Password must made of alphanumeric characters (a-z, A-Z, 0-9) and be at least 8 characters long.")
-            password = getpass("New password: ")
-            password_confirm = getpass("New password (again): ")
-            while password != password_confirm:
-                print("Passwords do not match")
-                password = getpass("New password: ")
-                password_confirm = getpass("New password (again): ")
-            if not client.passwd(password):
-                print("Failed to change password")
-                sys.exit(1)
-        elif args.action == "build":
+        if args.action == "build":
+            # do this before user enters password so errors are reported without unnecessary user action
             try:
                 os.mkdir(args.directory)
             except FileExistsError:
@@ -233,6 +210,29 @@ def main():
             if rev is None:
                 print("Unable to determine currently installed ARTIQ revision. Specify manually using --rev.")
                 sys.exit(1)
+
+        if args.action == "passwd":
+            password = getpass("Current password: ")
+        else:
+            password = getpass()
+        if not client.login(args.username, password):
+            print("Login failed")
+            sys.exit(1)
+
+        print("Logged in successfully.")
+        if args.action == "passwd":
+            print("Password must made of alphanumeric characters (a-z, A-Z, 0-9) and be at least 8 characters long.")
+            password = getpass("New password: ")
+            password_confirm = getpass("New password (again): ")
+            while password != password_confirm:
+                print("Passwords do not match")
+                password = getpass("New password: ")
+                password_confirm = getpass("New password (again): ")
+            if not client.passwd(password):
+                print("Failed to change password")
+                sys.exit(1)
+        elif args.action == "build":
+            # build dir and version variables set up above
             result, contents = client.build(major_ver, rev, args.variant, args.log, args.experimental)
             if result != "OK":
                 if result == "UNAUTHORIZED":

artiq/frontend/artiq_client.py

@@ -125,6 +125,9 @@ def get_argparser():
         "ls", help="list a directory on the master")
     parser_ls.add_argument("directory", default="", nargs="?")
 
+    subparsers.add_parser(
+        "terminate", help="terminate the ARTIQ master")
+
     common_args.verbosity_args(parser)
     return parser
 
@@ -195,6 +198,10 @@ def _action_ls(remote, args):
         print(name)
 
 
+def _action_terminate(remote, _args):
+    remote.terminate()
+
+
 def _show_schedule(schedule):
     clear_screen()
     if schedule:
@@ -319,7 +326,8 @@ def main():
             "del_dataset": "master_dataset_db",
             "scan_devices": "master_device_db",
             "scan_repository": "master_experiment_db",
-            "ls": "master_experiment_db"
+            "ls": "master_experiment_db",
+            "terminate": "master_terminate",
         }[action]
         remote = Client(args.server, port, target_name)
         try:

artiq/frontend/artiq_compile.py

@@ -24,7 +24,7 @@ def get_argparser():
     common_args.verbosity_args(parser)
     parser.add_argument("--device-db", default="device_db.py",
                         help="device database file (default: '%(default)s')")
-    parser.add_argument("--dataset-db", default="dataset_db.pyon",
+    parser.add_argument("--dataset-db", default="dataset_db.mdb",
                         help="dataset file (default: '%(default)s')")
 
     parser.add_argument("-c", "--class-name", default=None,
@@ -45,8 +45,10 @@ def main():
     common_args.init_logger_from_args(args)
 
     device_mgr = DeviceManager(DeviceDB(args.device_db))
-    dataset_mgr = DatasetManager(DatasetDB(args.dataset_db))
-
+    try:
+        dataset_db = DatasetDB(args.dataset_db)
+        try:
+            dataset_mgr = DatasetManager(dataset_db)
             embedding_map = EmbeddingMap()
 
             output = args.output
@@ -54,7 +56,6 @@ def main():
                 basename, ext = os.path.splitext(args.file)
                 output = "{}.elf".format(basename)
 
-    try:
             module = file_import(args.file, prefix="artiq_run_")
             exp = get_experiment(module, args.class_name)
             arguments = parse_arguments(args.arguments)
@@ -62,10 +63,11 @@ def main():
             exp_inst = exp((device_mgr, dataset_mgr, argument_mgr, {}))
             argument_mgr.check_unprocessed_arguments()
 
-
             if not getattr(exp.run, "__artiq_kernel__", False):
                 raise ValueError("Experiment entry point must be a kernel")
             exp_inst.core.compile(exp_inst.run, [], {}, embedding_map, file_output=output)
+        finally:
+            dataset_db.close_db()
     finally:
         device_mgr.close_devices()
 

artiq/frontend/artiq_master.py

@@ -4,6 +4,7 @@ import asyncio
 import argparse
 import atexit
 import logging
+from types import SimpleNamespace
 
 from sipyco.pc_rpc import Server as RPCServer
 from sipyco.sync_struct import Publisher
@@ -39,7 +40,7 @@ def get_argparser():
     group = parser.add_argument_group("databases")
     group.add_argument("--device-db", default="device_db.py",
                        help="device database file (default: '%(default)s')")
-    group.add_argument("--dataset-db", default="dataset_db.pyon",
+    group.add_argument("--dataset-db", default="dataset_db.mdb",
                        help="dataset file (default: '%(default)s')")
 
     group = parser.add_argument_group("repository")
@@ -100,6 +101,7 @@ def main():
 
     device_db = DeviceDB(args.device_db)
     dataset_db = DatasetDB(args.dataset_db)
+    atexit.register(dataset_db.close_db)
     dataset_db.start(loop=loop)
     atexit_register_coroutine(dataset_db.stop, loop=loop)
     worker_handlers = dict()
@@ -133,12 +135,16 @@ def main():
     })
     experiment_db.scan_repository_async(loop=loop)
 
+    signal_handler_task = loop.create_task(signal_handler.wait_terminate())
+    master_terminate = SimpleNamespace(terminate=lambda: signal_handler_task.cancel())
+
     server_control = RPCServer({
         "master_config": config,
         "master_device_db": device_db,
         "master_dataset_db": dataset_db,
         "master_schedule": scheduler,
-        "master_experiment_db": experiment_db
+        "master_experiment_db": experiment_db,
+        "master_terminate": master_terminate
     }, allow_parallel=True)
     loop.run_until_complete(server_control.start(
         bind, args.port_control))
@@ -161,7 +167,11 @@ def main():
     atexit_register_coroutine(server_logging.stop, loop=loop)
 
     print("ARTIQ master is now ready.")
-    loop.run_until_complete(signal_handler.wait_terminate())
+    try:
+        loop.run_until_complete(signal_handler_task)
+    except asyncio.CancelledError:
+        pass
+
 
 if __name__ == "__main__":
     main()

artiq/frontend/artiq_run.py

@@ -89,7 +89,7 @@ def get_argparser(with_file=True):
     common_args.verbosity_args(parser)
     parser.add_argument("--device-db", default="device_db.py",
                         help="device database file (default: '%(default)s')")
-    parser.add_argument("--dataset-db", default="dataset_db.pyon",
+    parser.add_argument("--dataset-db", default="dataset_db.mdb",
                         help="dataset file (default: '%(default)s')")
 
     parser.add_argument("-c", "--class-name", default=None,
@@ -142,12 +142,12 @@ def run(with_file=False):
     args = get_argparser(with_file).parse_args()
     common_args.init_logger_from_args(args)
 
+    dataset_db = DatasetDB(args.dataset_db)
+    try:
+        dataset_mgr = DatasetManager(dataset_db)
         device_mgr = DeviceManager(DeviceDB(args.device_db),
                                   virtual_devices={"scheduler": DummyScheduler(),
                                                    "ccb": DummyCCB()})
-    dataset_db = DatasetDB(args.dataset_db)
-    dataset_mgr = DatasetManager(dataset_db)
-
         try:
             exp_inst = _build_experiment(device_mgr, dataset_mgr, args)
             exp_inst.prepare()
@@ -166,7 +166,9 @@ def run(with_file=False):
         else:
             for k, v in sorted(dataset_mgr.local.items(), key=itemgetter(0)):
                 print("{}: {}".format(k, v))
+    finally:
         dataset_db.save()
+        dataset_db.close_db()
 
 
 def main():

artiq/gateware/rtio/core.py

@@ -60,17 +60,17 @@ class Core(Module, AutoCSR):
         # Outputs/Inputs
         quash_channels = [n for n, c in enumerate(channels) if isinstance(c, LogChannel)]
 
-        outputs = SED(channels, tsc.glbl_fine_ts_width,
+        outputs = ClockDomainsRenamer("rio")(SED(channels, tsc.glbl_fine_ts_width,
             quash_channels=quash_channels,
             lane_count=lane_count, fifo_depth=fifo_depth,
-            interface=self.cri)
+            interface=self.cri))
         self.submodules += outputs
         self.comb += outputs.coarse_timestamp.eq(tsc.coarse_ts)
         self.sync += outputs.minimum_coarse_timestamp.eq(tsc.coarse_ts + 12)
 
-        inputs = InputCollector(tsc, channels,
+        inputs = ClockDomainsRenamer("rio")(InputCollector(tsc, channels,
             quash_channels=quash_channels,
-            interface=self.cri)
+            interface=self.cri))
         self.submodules += inputs
 
         # Asychronous output errors

artiq/master/databases.py

@@ -1,11 +1,13 @@
 import asyncio
 
-from artiq.tools import file_import
+import lmdb
 
-from sipyco.sync_struct import Notifier, process_mod, update_from_dict
+from sipyco.sync_struct import Notifier, process_mod, ModAction, update_from_dict
 from sipyco import pyon
 from sipyco.asyncio_tools import TaskObject
 
+from artiq.tools import file_import
+
 
 def device_db_from_file(filename):
     mod = file_import(filename)
@@ -40,15 +42,25 @@ class DatasetDB(TaskObject):
         self.persist_file = persist_file
         self.autosave_period = autosave_period
 
-        try:
-            file_data = pyon.load_file(self.persist_file)
-        except FileNotFoundError:
-            file_data = dict()
-        self.data = Notifier({k: (True, v) for k, v in file_data.items()})
+        self.lmdb = lmdb.open(persist_file, subdir=False, map_size=2**30)
+        data = dict()
+        with self.lmdb.begin() as txn:
+            for key, value in txn.cursor():
+                data[key.decode()] = (True, pyon.decode(value.decode()))
+        self.data = Notifier(data)
+        self.pending_keys = set()
+
+    def close_db(self):
+        self.lmdb.close()
 
     def save(self):
-        data = {k: v[1] for k, v in self.data.raw_view.items() if v[0]}
-        pyon.store_file(self.persist_file, data)
+        with self.lmdb.begin(write=True) as txn:
+            for key in self.pending_keys:
+                if key not in self.data.raw_view or not self.data.raw_view[key][0]:
+                    txn.delete(key.encode())
+                else:
+                    txn.put(key.encode(), pyon.encode(self.data.raw_view[key][1]).encode())
+        self.pending_keys.clear()
 
     async def _do(self):
         try:
@@ -62,6 +74,12 @@ class DatasetDB(TaskObject):
         return self.data.raw_view[key][1]
 
     def update(self, mod):
+        if mod["path"]:
+            key = mod["path"][0]
+        else:
+            assert(mod["action"] == ModAction.setitem.value or mod["action"] == ModAction.delitem.value)
+            key = mod["key"]
+        self.pending_keys.add(key)
         process_mod(self.data, mod)
 
     # convenience functions (update() can be used instead)
@@ -72,7 +90,9 @@ class DatasetDB(TaskObject):
             else:
                 persist = False
         self.data[key] = (persist, value)
+        self.pending_keys.add(key)
 
     def delete(self, key):
         del self.data[key]
+        self.pending_keys.add(key)
     #

artiq/master/worker_db.py

@@ -119,6 +119,9 @@ class DatasetManager:
         if persist:
             broadcast = True
 
+        if not (broadcast or archive):
+            logger.warning(f"Dataset '{key}' will not be stored. Both 'broadcast' and 'archive' are set to False.")
+
         if broadcast:
             self._broadcaster[key] = persist, value
         elif key in self._broadcaster.raw_view:

artiq/master/worker_impl.py

@@ -186,10 +186,6 @@ class ExamineDatasetMgr:
     def get(key, archive=False):
         return ParentDatasetDB.get(key)
 
-    @staticmethod
-    def update(self, mod):
-        pass
-
 
 def examine(device_mgr, dataset_mgr, file):
     previous_keys = set(sys.modules.keys())

artiq/test/hardware_testbench.py

@@ -22,13 +22,14 @@ class ExperimentCase(unittest.TestCase):
     def setUp(self):
         self.device_db = DeviceDB(os.path.join(artiq_root, "device_db.py"))
         self.dataset_db = DatasetDB(
-            os.path.join(artiq_root, "dataset_db.pyon"))
+            os.path.join(artiq_root, "dataset_db.mdb"))
         self.device_mgr = DeviceManager(
             self.device_db, virtual_devices={"scheduler": DummyScheduler()})
         self.dataset_mgr = DatasetManager(self.dataset_db)
 
     def tearDown(self):
         self.device_mgr.close_devices()
+        self.dataset_db.close_db()
 
     def create(self, cls, *args, **kwargs):
         try:

artiq/test/test_client.py

@@ -0,0 +1,91 @@
+"""Tests for artiq_client functionality"""
+
+import subprocess
+import sys
+import os
+import unittest
+from tempfile import TemporaryDirectory
+from pygit2 import init_repository, Signature
+
+EXPERIMENT_CONTENT = """
+from artiq.experiment import *
+class EmptyExperiment(EnvExperiment):
+    def build(self):
+        pass
+    def run(self):
+        print("test content")
+"""
+
+DDB_CONTENT = """
+device_db = {}
+"""
+
+
+def get_env():
+    env = os.environ.copy()
+    env["PYTHONUNBUFFERED"] = "1"
+    return env
+
+
+class TestClient(unittest.TestCase):
+    def setUp(self):
+        self.tmp_dir = TemporaryDirectory(prefix="test")
+        self.tmp_empty_dir = TemporaryDirectory(prefix="empty_repo")
+        self.exp_name = "experiment.py"
+        self.exp_path = os.path.join(self.tmp_dir.name, self.exp_name)
+        self.device_db_path = os.path.join(self.tmp_dir.name, "device_db.py")
+        with open(self.exp_path, "w") as f:
+            f.write(EXPERIMENT_CONTENT)
+        with open(self.device_db_path, "w") as f:
+            f.write(DDB_CONTENT)
+
+    def start_master(self, *args):
+        self.master = subprocess.Popen([sys.executable, "-m", "artiq.frontend.artiq_master", "--device-db",
+                                        self.device_db_path, *args], encoding="utf8", env=get_env(),
+                                       text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        while self.master.stdout.readline().strip() != "ARTIQ master is now ready.":
+            pass
+
+    def check_and_terminate_master(self):
+        while not ("test content" in self.master.stdout.readline()):
+            pass
+        self.run_client("terminate")
+        self.assertEqual(self.master.wait(), 0)
+        self.master.stdout.close()
+
+    @staticmethod
+    def run_client(*args):
+        subprocess.run([sys.executable, "-m", "artiq.frontend.artiq_client", *args], check=True,
+                       capture_output=True, env=get_env(), text=True, encoding="utf8").check_returncode()
+
+    def test_submit_outside_repo(self):
+        self.start_master("-r", self.tmp_empty_dir.name)
+        self.run_client("submit", self.exp_path)
+        self.check_and_terminate_master()
+
+    def test_submit_by_content(self):
+        self.start_master("-r", self.tmp_empty_dir.name)
+        self.run_client("submit", self.exp_path, "--content")
+        self.check_and_terminate_master()
+
+    def test_submit_by_file_repo(self):
+        self.start_master("-r", self.tmp_dir.name)
+        self.run_client("submit", self.exp_name, "-R")
+        self.check_and_terminate_master()
+
+    def test_submit_by_git_repo(self):
+        repo = init_repository(self.tmp_dir.name)
+        repo.index.add_all()
+        repo.index.write()
+        tree = repo.index.write_tree()
+        signature = Signature("Test", "test@example.com")
+        commit_msg = "Commit message"
+        repo.create_commit("HEAD", signature, signature, commit_msg, tree, [])
+
+        self.start_master("-r", self.tmp_dir.name, "-g")
+        self.run_client("submit", self.exp_name, "-R")
+        self.check_and_terminate_master()
+
+    def tearDown(self):
+        self.tmp_dir.cleanup()
+        self.tmp_empty_dir.cleanup()

doc/manual/core_device.rst

@@ -175,6 +175,7 @@ KC705 in DRTIO variants and Kasli generates the RTIO clock using a PLL locked ei
     * ``int_100`` - internal crystal oscillator using PLL, 100 MHz output,
     * ``int_150`` - internal crystal oscillator using PLL, 150 MHz output,
     * ``ext0_synth0_10to125`` - external 10 MHz reference using PLL, 125 MHz output,
+    * ``ext0_synth0_80to125`` - external 80 MHz reference using PLL, 125 MHz output,
     * ``ext0_synth0_100to125`` - external 100 MHz reference using PLL, 125 MHz output,
     * ``ext0_synth0_125to125`` - external 125 MHz reference using PLL, 125 MHz output,
     * ``ext0_bypass``, ``ext0_bypass_125``, ``ext0_bypass_100`` - external clock - with explicit aliases available.

doc/manual/getting_started_core.rst

@@ -243,3 +243,27 @@ Try this: ::
                 # execute RTIO operations in the DMA buffer
                 # each playback advances the timeline by 50*(100+100) ns
                 self.core_dma.playback_handle(pulses_handle)
+
+Distributed Direct Memory Access (DDMA)
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+By default on DRTIO systems, all events recorded by the DMA core are kept and played back on the master.
+
+With distributed DMA, RTIO events that should be played back on remote destinations, are distributed to the corresponding satellites. In some cases (typically, large buffers on several satellites with high event throughput), it allows for better performance and higher bandwidth, as the RTIO events do not have to be sent over the DRTIO link(s) during playback.
+
+To enable distributed DMA, simply provide an ``enable_ddma=True`` argument for the :meth:`~artiq.coredevice.dma.CoreDMA.record` method - taking a snippet from the previous example: ::
+
+        @kernel
+        def record(self):
+            with self.core_dma.record("pulses", enable_ddma=True):
+                # all RTIO operations now go to the "pulses"
+                # DMA buffer, instead of being executed immediately.
+                for i in range(50):
+                    self.ttl0.pulse(100*ns)
+                    delay(100*ns)
+
+This argument is ignored on standalone systems, as it does not apply there.
+
+Enabling DDMA on a purely local sequence on a DRTIO system introduces an overhead during trace recording which comes from additional processing done on the record, so careful use is advised.
+
+Due to the extra time that communicating with relevant satellites takes, an additional delay before playback may be necessary to prevent a :exc:`~artiq.coredevice.exceptions.RTIOUnderflow` when playing back a DDMA-enabled sequence.

doc/manual/installing.rst

@@ -290,8 +290,9 @@ The KC705 may use either an external clock signal, or its internal clock with ex
 
 Other options include:
   - ``ext0_synth0_10to125`` - external 10MHz reference clock used by Si5324 to synthesize a 125MHz RTIO clock,
-  - ``ext0_synth0_100to125`` - exteral 100MHz reference clock used by Si5324 to synthesize a 125MHz RTIO clock,
-  - ``ext0_synth0_125to125`` - exteral 125MHz reference clock used by Si5324 to synthesize a 125MHz RTIO clock,
+  - ``ext0_synth0_80to125`` - external 80MHz reference clock used by Si5324 to synthesize a 125MHz RTIO clock,
+  - ``ext0_synth0_100to125`` - external 100MHz reference clock used by Si5324 to synthesize a 125MHz RTIO clock,
+  - ``ext0_synth0_125to125`` - external 125MHz reference clock used by Si5324 to synthesize a 125MHz RTIO clock,
   - ``int_100`` - internal crystal reference is used by Si5324 to synthesize a 100MHz RTIO clock,
   - ``int_150`` - internal crystal reference is used by Si5324 to synthesize a 150MHz RTIO clock.
   - ``ext0_bypass_125`` and ``ext0_bypass_100`` - explicit aliases for ``ext0_bypass``.

flake.nix

@@ -89,7 +89,7 @@
 
         # keep llvm_x in sync with nac3
         propagatedBuildInputs = [ pkgs.llvm_14 nac3.packages.x86_64-linux.nac3artiq-pgo sipyco.packages.x86_64-linux.sipyco artiq-comtools.packages.x86_64-linux.artiq-comtools ]
-          ++ (with pkgs.python3Packages; [ pyqtgraph pygit2 numpy dateutil scipy prettytable pyserial h5py pyqt5 qasync ]);
+          ++ (with pkgs.python3Packages; [ pyqtgraph pygit2 numpy dateutil scipy prettytable pyserial h5py pyqt5 qasync tqdm lmdb ]);
 
         dontWrapQtApps = true;
         postFixup = ''
@@ -106,7 +106,9 @@
         ];
 
         # FIXME: automatically propagate llvm_x dependency
-        checkInputs = [ pkgs.llvm_14 ];
+        # cacert is required in the check stage only, as certificates are to be
+        # obtained from system elsewhere
+        checkInputs = [ pkgs.llvm_14 pkgs.cacert ];
         checkPhase = ''
           python -m unittest discover -v artiq.test
           '';