drtio-proto: add coremgmt-over-drtio messages

src/libboard_artiq/src/drtioaux_proto.rs

@@ -287,6 +287,62 @@ pub enum Packet {
     SubkernelMessageAck {
         destination: u8,
     },
+
+    CoreMgmtGetLogRequest {
+        destination: u8,
+        clear: bool,
+    },
+    CoreMgmtClearLogRequest {
+        destination: u8,
+    },
+    CoreMgmtSetLogLevelRequest {
+        destination: u8,
+        log_level: u8,
+    },
+    CoreMgmtSetUartLogLevelRequest {
+        destination: u8,
+        log_level: u8,
+    },
+    CoreMgmtConfigReadRequest {
+        destination: u8,
+        length: u16,
+        key: [u8; MASTER_PAYLOAD_MAX_SIZE],
+    },
+    CoreMgmtConfigReadContinue {
+        destination: u8,
+    },
+    CoreMgmtConfigWriteRequest {
+        destination: u8,
+        last: bool,
+        length: u16,
+        data: [u8; MASTER_PAYLOAD_MAX_SIZE],
+    },
+    CoreMgmtConfigRemoveRequest {
+        destination: u8,
+        length: u16,
+        key: [u8; MASTER_PAYLOAD_MAX_SIZE],
+    },
+    CoreMgmtConfigEraseRequest {
+        destination: u8,
+    },
+    CoreMgmtRebootRequest {
+        destination: u8,
+    },
+    CoreMgmtAllocatorDebugRequest {
+        destination: u8,
+    },
+    CoreMgmtGetLogReply {
+        last: bool,
+        length: u16,
+        data: [u8; SAT_PAYLOAD_MAX_SIZE],
+    },
+    CoreMgmtConfigReadReply {
+        last: bool,
+        length: u16,
+        value: [u8; SAT_PAYLOAD_MAX_SIZE],
+    },
+    CoreMgmtAck,
+    CoreMgmtNack,
 }
 
 impl Packet {
@@ -563,6 +619,93 @@ impl Packet {
                 destination: reader.read_u8()?,
             },
 
+            0xd0 => Packet::CoreMgmtGetLogRequest {
+                destination: reader.read_u8()?,
+                clear: reader.read_bool()?,
+            },
+            0xd1 => Packet::CoreMgmtClearLogRequest {
+                destination: reader.read_u8()?,
+            },
+            0xd2 => Packet::CoreMgmtSetLogLevelRequest {
+                destination: reader.read_u8()?,
+                log_level: reader.read_u8()?,
+            },
+            0xd3 => Packet::CoreMgmtSetUartLogLevelRequest {
+                destination: reader.read_u8()?,
+                log_level: reader.read_u8()?,
+            },
+            0xd4 => {
+                let destination = reader.read_u8()?;
+                let length = reader.read_u16()?;
+                let mut key: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+                reader.read_exact(&mut key[0..length as usize])?;
+                Packet::CoreMgmtConfigReadRequest {
+                    destination: destination,
+                    length: length,
+                    key: key,
+                }
+            }
+            0xd5 => Packet::CoreMgmtConfigReadContinue {
+                destination: reader.read_u8()?,
+            },
+            0xd6 => {
+                let destination = reader.read_u8()?;
+                let last = reader.read_bool()?;
+                let length = reader.read_u16()?;
+                let mut data: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+                reader.read_exact(&mut data[0..length as usize])?;
+                Packet::CoreMgmtConfigWriteRequest {
+                    destination: destination,
+                    last: last,
+                    length: length,
+                    data: data,
+                }
+            }
+            0xd7 => {
+                let destination = reader.read_u8()?;
+                let length = reader.read_u16()?;
+                let mut key: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+                reader.read_exact(&mut key[0..length as usize])?;
+                Packet::CoreMgmtConfigRemoveRequest {
+                    destination: destination,
+                    length: length,
+                    key: key,
+                }
+            }
+            0xd8 => Packet::CoreMgmtConfigEraseRequest {
+                destination: reader.read_u8()?,
+            },
+            0xd9 => Packet::CoreMgmtRebootRequest {
+                destination: reader.read_u8()?,
+            },
+            0xda => Packet::CoreMgmtAllocatorDebugRequest {
+                destination: reader.read_u8()?,
+            },
+            0xdb => {
+                let last = reader.read_bool()?;
+                let length = reader.read_u16()?;
+                let mut data: [u8; SAT_PAYLOAD_MAX_SIZE] = [0; SAT_PAYLOAD_MAX_SIZE];
+                reader.read_exact(&mut data[0..length as usize])?;
+                Packet::CoreMgmtGetLogReply {
+                    last: last,
+                    length: length,
+                    data: data,
+                }
+            }
+            0xdc => {
+                let last = reader.read_bool()?;
+                let length = reader.read_u16()?;
+                let mut value: [u8; SAT_PAYLOAD_MAX_SIZE] = [0; SAT_PAYLOAD_MAX_SIZE];
+                reader.read_exact(&mut value[0..length as usize])?;
+                Packet::CoreMgmtConfigReadReply {
+                    last: last,
+                    length: length,
+                    value: value,
+                }
+            }
+            0xdd => Packet::CoreMgmtAck,
+            0xde => Packet::CoreMgmtNack,
+
             ty => return Err(Error::UnknownPacket(ty)),
         })
     }
@@ -938,6 +1081,88 @@ impl Packet {
                 writer.write_u8(0xcc)?;
                 writer.write_u8(destination)?;
             }
+
+            Packet::CoreMgmtGetLogRequest { destination, clear } => {
+                writer.write_u8(0xd0)?;
+                writer.write_u8(destination)?;
+                writer.write_bool(clear)?;
+            }
+            Packet::CoreMgmtClearLogRequest { destination } => {
+                writer.write_u8(0xd1)?;
+                writer.write_u8(destination)?;
+            }
+            Packet::CoreMgmtSetLogLevelRequest { destination, log_level } => {
+                writer.write_u8(0xd2)?;
+                writer.write_u8(destination)?;
+                writer.write_u8(log_level)?;
+            }
+            Packet::CoreMgmtSetUartLogLevelRequest { destination, log_level } => {
+                writer.write_u8(0xd3)?;
+                writer.write_u8(destination)?;
+                writer.write_u8(log_level)?;
+            }
+            Packet::CoreMgmtConfigReadRequest {
+                destination,
+                length,
+                key,
+            } => {
+                writer.write_u8(0xd4)?;
+                writer.write_u8(destination)?;
+                writer.write_u16(length)?;
+                writer.write_all(&key[0..length as usize])?;
+            }
+            Packet::CoreMgmtConfigReadContinue { destination } => {
+                writer.write_u8(0xd5)?;
+                writer.write_u8(destination)?;
+            }
+            Packet::CoreMgmtConfigWriteRequest {
+                destination,
+                last,
+                length,
+                data,
+            } => {
+                writer.write_u8(0xd6)?;
+                writer.write_u8(destination)?;
+                writer.write_bool(last)?;
+                writer.write_u16(length)?;
+                writer.write_all(&data[0..length as usize])?;
+            }
+            Packet::CoreMgmtConfigRemoveRequest {
+                destination,
+                length,
+                key,
+            } => {
+                writer.write_u8(0xd7)?;
+                writer.write_u8(destination)?;
+                writer.write_u16(length)?;
+                writer.write_all(&key[0..length as usize])?;
+            }
+            Packet::CoreMgmtConfigEraseRequest { destination } => {
+                writer.write_u8(0xd8)?;
+                writer.write_u8(destination)?;
+            }
+            Packet::CoreMgmtRebootRequest { destination } => {
+                writer.write_u8(0xd9)?;
+                writer.write_u8(destination)?;
+            }
+            Packet::CoreMgmtAllocatorDebugRequest { destination } => {
+                writer.write_u8(0xda)?;
+                writer.write_u8(destination)?;
+            }
+            Packet::CoreMgmtGetLogReply { last, length, data } => {
+                writer.write_u8(0xdb)?;
+                writer.write_bool(last)?;
+                writer.write_u16(length)?;
+                writer.write_all(&data[0..length as usize])?;
+            }
+            Packet::CoreMgmtConfigReadReply { last, length, value } => {
+                writer.write_u8(0xdc)?;
+                writer.write_bool(last)?;
+                writer.write_u16(length)?;
+                writer.write_all(&value[0..length as usize])?;
+            }
+            Packet::CoreMgmtAck => writer.write_u8(0xdd)?,
+            Packet::CoreMgmtNack => writer.write_u8(0xde)?,
         }
         Ok(())
     }

src/runtime/src/comms.rs

@@ -783,7 +783,7 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
         }
     }
 
-    mgmt::start(cfg);
+    mgmt::start(cfg, Some((&aux_mutex, &drtio_routing_table, timer)));
 
     task::spawn(async move {
         let connection = Rc::new(Semaphore::new(1, 1));
@@ -886,7 +886,7 @@ pub fn soft_panic_main(timer: GlobalTimer, cfg: Config) -> ! {
 
     Sockets::init(32);
 
-    mgmt::start(cfg);
+    mgmt::start(cfg, None);
 
     // getting eth settings disables the LED as it resets GPIO
     // need to re-enable it here

src/runtime/src/mgmt.rs

@@ -1,16 +1,21 @@
 use alloc::{rc::Rc, string::String, vec::Vec};
-use core::cell::RefCell;
+use core::{cell::RefCell, ops::Deref};
 
 use futures::{future::poll_fn, task::Poll};
 use libasync::{smoltcp::TcpStream, task};
-use libboard_artiq::logger::{BufferLogger, LogBufferRef};
+use libboard_artiq::{drtio_routing,
-use libboard_zynq::{slcr, smoltcp};
+                     drtio_routing::RoutingTable,
+                     logger::{BufferLogger, LogBufferRef}};
+use libboard_zynq::{slcr, smoltcp, timer::GlobalTimer};
 use libconfig::Config;
+use libcortex_a9::mutex::Mutex;
 use log::{self, debug, error, info, warn, LevelFilter};
 use num_derive::FromPrimitive;
 use num_traits::FromPrimitive;
 
 use crate::proto_async::*;
+#[cfg(has_drtio)]
+use crate::rtio_mgt::*;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Error {
@@ -18,6 +23,8 @@ pub enum Error {
     UnknownLogLevel(u8),
     UnexpectedPattern,
     UnrecognizedPacket,
+    #[cfg(has_drtio)]
+    DrtioError(drtio::Error),
 }
 
 type Result<T> = core::result::Result<T, Error>;
@@ -29,6 +36,8 @@ impl core::fmt::Display for Error {
             &Error::UnknownLogLevel(lvl) => write!(f, "unknown log level {}", lvl),
             &Error::UnexpectedPattern => write!(f, "unexpected pattern"),
             &Error::UnrecognizedPacket => write!(f, "unrecognized packet"),
+            #[cfg(has_drtio)]
+            &Error::DrtioError(error) => write!(f, "drtio error: {}", error),
         }
     }
 }
@@ -39,6 +48,13 @@ impl From<smoltcp::Error> for Error {
     }
 }
 
+#[cfg(has_drtio)]
+impl From<drtio::Error> for Error {
+    fn from(error: drtio::Error) -> Self {
+        Error::DrtioError(error)
+    }
+}
+
 #[derive(Debug, FromPrimitive)]
 pub enum Request {
     GetLog = 1,
@@ -51,6 +67,9 @@ pub enum Request {
     ConfigRead = 12,
     ConfigWrite = 13,
     ConfigRemove = 14,
+    ConfigErase = 15,
+
+    DebugAllocator = 8,
 }
 
 #[repr(i8)]
@@ -111,30 +130,510 @@ async fn read_key(stream: &mut TcpStream) -> Result<String> {
     Ok(String::from_utf8(buffer).unwrap())
 }
 
-async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cfg: Rc<Config>) -> Result<()> {
+#[cfg(has_drtio)]
-    if !expect(&stream, b"ARTIQ management\n").await? {
+mod remote_coremgmt {
-        return Err(Error::UnexpectedPattern);
+    use io::{Cursor, ProtoWrite};
-    }
+    use libboard_artiq::drtioaux_proto::{Packet, MASTER_PAYLOAD_MAX_SIZE};
-    stream.send_slice("e".as_bytes()).await?;
 
+    use super::*;
+    use crate::rtio_mgt::drtio;
+
+    pub async fn get_log(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+    ) -> Result<()> {
+        let mut buffer = Vec::new();
         loop {
-        let msg = read_i8(stream).await;
+            let reply = drtio::aux_transact(
-        if let Err(smoltcp::Error::Finished) = msg {
+                aux_mutex,
+                linkno,
+                routing_table,
+                &Packet::CoreMgmtGetLogRequest {
+                    destination,
+                    clear: false,
+                },
+                timer,
+            )
+            .await;
+
+            match reply {
+                Ok(Packet::CoreMgmtGetLogReply { last, length, data }) => {
+                    buffer.extend(&data[..length as usize]);
+                    if last {
+                        write_i8(stream, Reply::LogContent as i8).await?;
+                        write_chunk(stream, &buffer).await?;
                         return Ok(());
                     }
-        let msg: Request = FromPrimitive::from_i8(msg?).ok_or(Error::UnrecognizedPacket)?;
+                }
-        match msg {
+                Ok(packet) => {
-            Request::GetLog => {
+                    error!("received unexpected aux packet: {:?}", packet);
+                    write_i8(stream, Reply::Error as i8).await?;
+                    return Err(drtio::Error::UnexpectedReply.into());
+                }
+                Err(e) => {
+                    error!("aux packet error ({})", e);
+                    write_i8(stream, Reply::Error as i8).await?;
+                    return Err(e.into());
+                }
+            }
+        }
+    }
+
+    pub async fn clear_log(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+    ) -> Result<()> {
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtClearLogRequest { destination },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn pull_log(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+        pull_id: &Rc<RefCell<u32>>,
+    ) -> Result<()> {
+        let id = {
+            let mut guard = pull_id.borrow_mut();
+            *guard += 1;
+            *guard
+        };
+        loop {
+            if id != *pull_id.borrow() {
+                // another connection attempts to pull the log...
+                // abort this connection...
+                break;
+            }
+
+            let reply = drtio::aux_transact(
+                aux_mutex,
+                linkno,
+                routing_table,
+                &Packet::CoreMgmtGetLogRequest {
+                    destination,
+                    clear: true,
+                },
+                timer,
+            )
+            .await;
+
+            match reply {
+                Ok(Packet::CoreMgmtGetLogReply { last: _, length, data }) => {
+                    write_chunk(stream, &data[..length as usize]).await?;
+                }
+                Ok(packet) => {
+                    error!("received unexpected aux packet: {:?}", packet);
+                    return Err(drtio::Error::UnexpectedReply.into());
+                }
+                Err(e) => {
+                    error!("aux packet error ({})", e);
+                    return Err(e.into());
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    pub async fn set_log_filter(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+        level: LevelFilter,
+    ) -> Result<()> {
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtSetLogLevelRequest {
+                destination,
+                log_level: level as u8,
+            },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn set_uart_log_filter(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+        level: LevelFilter,
+    ) -> Result<()> {
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtSetUartLogLevelRequest {
+                destination,
+                log_level: level as u8,
+            },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn config_read(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+        _cfg: &Rc<Config>,
+        key: &String,
+    ) -> Result<()> {
+        let mut config_key: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+        let len = key.len();
+        config_key[..len].clone_from_slice(key.as_bytes());
+
+        let mut reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtConfigReadRequest {
+                destination: destination,
+                length: len as u16,
+                key: config_key,
+            },
+            timer,
+        )
+        .await;
+
+        let mut buffer = Vec::<u8>::new();
+        loop {
+            match reply {
+                Ok(Packet::CoreMgmtConfigReadReply { last, length, value }) => {
+                    buffer.extend(&value[..length as usize]);
+
+                    if last {
+                        write_i8(stream, Reply::ConfigData as i8).await?;
+                        write_chunk(stream, &buffer).await?;
+                        return Ok(());
+                    }
+
+                    reply = drtio::aux_transact(
+                        aux_mutex,
+                        linkno,
+                        routing_table,
+                        &Packet::CoreMgmtConfigReadContinue {
+                            destination: destination,
+                        },
+                        timer,
+                    )
+                    .await;
+                }
+                Ok(packet) => {
+                    error!("received unexpected aux packet: {:?}", packet);
+                    write_i8(stream, Reply::Error as i8).await?;
+                    return Err(drtio::Error::UnexpectedReply.into());
+                }
+                Err(e) => {
+                    error!("aux packet error ({})", e);
+                    write_i8(stream, Reply::Error as i8).await?;
+                    return Err(e.into());
+                }
+            }
+        }
+    }
+
+    pub async fn config_write(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+        _cfg: &Rc<Config>,
+        key: &String,
+        value: Vec<u8>,
+    ) -> Result<()> {
+        let mut message = Cursor::new(Vec::with_capacity(key.len() + value.len() + 4 * 2));
+        message.write_string(key).unwrap();
+        message.write_bytes(&value).unwrap();
+
+        match drtio::partition_data(
+            linkno,
+            aux_mutex,
+            routing_table,
+            timer,
+            message.get_ref(),
+            |slice, status, len: usize| Packet::CoreMgmtConfigWriteRequest {
+                destination: destination,
+                last: status.is_last(),
+                length: len as u16,
+                data: *slice,
+            },
+            |reply| match reply {
+                Packet::CoreMgmtAck => Ok(()),
+                packet => {
+                    error!("received unexpected aux packet: {:?}", packet);
+                    Err(drtio::Error::UnexpectedReply)
+                }
+            },
+        )
+        .await
+        {
+            Ok(()) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn config_remove(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+        _cfg: &Rc<Config>,
+        key: &String,
+    ) -> Result<()> {
+        let mut config_key: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
+        let len = key.len();
+        config_key[..len].clone_from_slice(key.as_bytes());
+
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtConfigRemoveRequest {
+                destination: destination,
+                length: len as u16,
+                key: config_key,
+            },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn config_erase(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+    ) -> Result<()> {
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtConfigEraseRequest {
+                destination: destination,
+            },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn reboot(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+    ) -> Result<()> {
+        info!("initited reboot request to satellite destination {}", destination);
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtRebootRequest {
+                destination: destination,
+            },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::RebootImminent as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                write_i8(stream, Reply::Error as i8).await?;
+                Err(e.into())
+            }
+        }
+    }
+
+    pub async fn debug_allocator(
+        stream: &mut TcpStream,
+        aux_mutex: &Rc<Mutex<bool>>,
+        routing_table: &drtio_routing::RoutingTable,
+        timer: GlobalTimer,
+        linkno: u8,
+        destination: u8,
+    ) -> Result<()> {
+        let reply = drtio::aux_transact(
+            aux_mutex,
+            linkno,
+            routing_table,
+            &Packet::CoreMgmtAllocatorDebugRequest {
+                destination: destination,
+            },
+            timer,
+        )
+        .await;
+
+        match reply {
+            Ok(Packet::CoreMgmtAck) => {
+                write_i8(stream, Reply::Success as i8).await?;
+                Ok(())
+            }
+            Ok(packet) => {
+                error!("received unexpected aux packet: {:?}", packet);
+                Err(drtio::Error::UnexpectedReply.into())
+            }
+            Err(e) => {
+                error!("aux packet error ({})", e);
+                Err(e.into())
+            }
+        }
+    }
+}
+
+mod local_coremgmt {
+    use super::*;
+
+    pub async fn get_log(stream: &mut TcpStream) -> Result<()> {
         let buffer = get_logger_buffer().await.extract().as_bytes().to_vec();
         write_i8(stream, Reply::LogContent as i8).await?;
         write_chunk(stream, &buffer).await?;
+        Ok(())
     }
-            Request::ClearLog => {
+
+    pub async fn clear_log(stream: &mut TcpStream) -> Result<()> {
         let mut buffer = get_logger_buffer().await;
         buffer.clear();
         write_i8(stream, Reply::Success as i8).await?;
+        Ok(())
     }
-            Request::PullLog => {
+
+    pub async fn pull_log(stream: &mut TcpStream, pull_id: &Rc<RefCell<u32>>) -> Result<()> {
         let id = {
             let mut guard = pull_id.borrow_mut();
             *guard += 1;
@@ -159,24 +658,26 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
                 logger.set_buffer_log_level(LevelFilter::Trace);
             }
         }
+        Ok(())
     }
-            Request::SetLogFilter => {
+
-                let lvl = read_log_level_filter(stream).await?;
+    pub async fn set_log_filter(stream: &mut TcpStream, lvl: LevelFilter) -> Result<()> {
         info!("Changing log level to {}", lvl);
         log::set_max_level(lvl);
         write_i8(stream, Reply::Success as i8).await?;
+        Ok(())
     }
-            Request::SetUartLogFilter => {
+
-                let lvl = read_log_level_filter(stream).await?;
+    pub async fn set_uart_log_filter(stream: &mut TcpStream, lvl: LevelFilter) -> Result<()> {
         info!("Changing UART log level to {}", lvl);
         unsafe {
             BufferLogger::get_logger().as_ref().unwrap().set_uart_log_level(lvl);
         }
         write_i8(stream, Reply::Success as i8).await?;
+        Ok(())
     }
-            Request::ConfigRead => {
+
-                let key = read_key(stream).await?;
+    pub async fn config_read(stream: &mut TcpStream, cfg: &Rc<Config>, key: &String) -> Result<()> {
-                debug!("read key: {}", key);
         let value = cfg.read(&key);
         if let Ok(value) = value {
             debug!("got value");
@@ -186,18 +687,11 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
             warn!("read error: no such key");
             write_i8(stream, Reply::Error as i8).await?;
         }
+        Ok(())
     }
-            Request::ConfigWrite => {
+
-                let key = read_key(stream).await?;
+    pub async fn config_write(stream: &mut TcpStream, cfg: &Rc<Config>, key: &String, value: Vec<u8>) -> Result<()> {
-                debug!("write key: {}", key);
+        let value = cfg.write(&key, value);
-                let len = read_i32(stream).await?;
-                let len = if len <= 0 { 0 } else { len as usize };
-                let mut buffer = Vec::with_capacity(len);
-                unsafe {
-                    buffer.set_len(len);
-                }
-                read_chunk(stream, &mut buffer).await?;
-                let value = cfg.write(&key, buffer);
         if value.is_ok() {
             debug!("write success");
             write_i8(stream, Reply::Success as i8).await?;
@@ -206,10 +700,10 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
             error!("failed to write: {:?}", value);
             write_i8(stream, Reply::Error as i8).await?;
         }
+        Ok(())
     }
-            Request::ConfigRemove => {
+
-                let key = read_key(stream).await?;
+    pub async fn config_remove(stream: &mut TcpStream, cfg: &Rc<Config>, key: &String) -> Result<()> {
-                debug!("erase key: {}", key);
         let value = cfg.remove(&key);
         if value.is_ok() {
             debug!("erase success");
@@ -218,18 +712,172 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
             warn!("erase failed");
             write_i8(stream, Reply::Error as i8).await?;
         }
+        Ok(())
     }
-            Request::Reboot => {
+
+    pub async fn config_erase(stream: &mut TcpStream) -> Result<()> {
+        error!("zynq device does not support config erase");
+        write_i8(stream, Reply::Error as i8).await?;
+        Ok(())
+    }
+
+    pub async fn reboot(stream: &mut TcpStream) -> Result<()> {
         info!("rebooting");
         write_i8(stream, Reply::RebootImminent as i8).await?;
         stream.flush().await?;
         slcr::reboot();
+
+        unreachable!()
     }
-        }
+
+    pub async fn debug_allocator(_stream: &mut TcpStream) -> Result<()> {
+        error!("zynq device does not support allocator debug print");
+        Ok(())
     }
 }
 
-pub fn start(cfg: Config) {
+#[cfg(has_drtio)]
+macro_rules! process {
+    ($stream: ident, $drtio_tuple:ident, $destination:expr, $func:ident $(, $param:expr)*) => {{
+        if $destination == 0 {
+            local_coremgmt::$func($stream, $($param, )*).await
+        } else if let Some((aux_mutex, routing_table, timer)) = $drtio_tuple {
+            let linkno = routing_table.0[$destination as usize][0] - 1 as u8;
+            remote_coremgmt::$func($stream, aux_mutex, routing_table, timer, linkno, $destination, $($param, )*).await
+        } else {
+            error!("coremgmt-over-drtio not supported for panicked device, please reboot");
+            write_i8($stream, Reply::Error as i8).await?;
+            Err(drtio::Error::LinkDown.into())
+        }
+    }}
+}
+
+#[cfg(not(has_drtio))]
+macro_rules! process {
+    ($stream: ident, $drtio_tuple:ident, $destination:expr, $func:ident $(, $param:expr)*) => {{
+        local_coremgmt::$func($stream, $($param, )*).await
+    }}
+}
+
+async fn handle_connection(
+    stream: &mut TcpStream,
+    pull_id: Rc<RefCell<u32>>,
+    cfg: Rc<Config>,
+    _drtio_tuple: Option<(&Rc<Mutex<bool>>, &RoutingTable, GlobalTimer)>,
+) -> Result<()> {
+    if !expect(&stream, b"ARTIQ management\n").await? {
+        return Err(Error::UnexpectedPattern);
+    }
+
+    let _destination: u8 = read_i8(stream).await? as u8;
+    stream.send_slice("e".as_bytes()).await?;
+
+    loop {
+        let msg = read_i8(stream).await;
+        if let Err(smoltcp::Error::Finished) = msg {
+            return Ok(());
+        }
+        let msg: Request = FromPrimitive::from_i8(msg?).ok_or(Error::UnrecognizedPacket)?;
+        match msg {
+            Request::GetLog => process!(stream, _drtio_tuple, _destination, get_log),
+            Request::ClearLog => process!(stream, _drtio_tuple, _destination, clear_log),
+            Request::PullLog => process!(
+                stream,
+                _drtio_tuple,
+                _destination,
+                pull_log,
+                &pull_id
+            ),
+            Request::SetLogFilter => {
+                let lvl = read_log_level_filter(stream).await?;
+                process!(
+                    stream,
+                    _drtio_tuple,
+                    _destination,
+                    set_log_filter,
+                    lvl
+                )
+            }
+            Request::SetUartLogFilter => {
+                let lvl = read_log_level_filter(stream).await?;
+                process!(
+                    stream,
+                    _drtio_tuple,
+                    _destination,
+                    set_uart_log_filter,
+                    lvl
+                )
+            }
+            Request::ConfigRead => {
+                let key = read_key(stream).await?;
+                process!(
+                    stream,
+                    _drtio_tuple,
+                    _destination,
+                    config_read,
+                    &cfg,
+                    &key
+                )
+            }
+            Request::ConfigWrite => {
+                let key = read_key(stream).await?;
+                let len = read_i32(stream).await?;
+                let len = if len <= 0 { 0 } else { len as usize };
+                let mut buffer = Vec::with_capacity(len);
+                unsafe {
+                    buffer.set_len(len);
+                }
+                read_chunk(stream, &mut buffer).await?;
+                process!(
+                    stream,
+                    _drtio_tuple,
+                    _destination,
+                    config_write,
+                    &cfg,
+                    &key,
+                    buffer
+                )
+            }
+            Request::ConfigRemove => {
+                let key = read_key(stream).await?;
+                process!(
+                    stream,
+                    _drtio_tuple,
+                    _destination,
+                    config_remove,
+                    &cfg,
+                    &key
+                )
+            }
+            Request::Reboot => {
+                process!(stream, _drtio_tuple, _destination, reboot)
+            }
+            Request::ConfigErase => {
+                process!(stream, _drtio_tuple, _destination, config_erase)
+            }
+            Request::DebugAllocator => {
+                process!(
+                    stream,
+                    _drtio_tuple,
+                    _destination,
+                    debug_allocator
+                )
+            }
+        }?;
+    }
+}
+
+pub fn start(
+    cfg: Config,
+    drtio_tuple: Option<(
+        &Rc<Mutex<bool>>,
+        &Rc<RefCell<drtio_routing::RoutingTable>>,
+        GlobalTimer,
+    )>,
+) {
+    let drtio_tuple = drtio_tuple.map(
+        |(aux_mutex, routing_table, timer)| (aux_mutex.clone(), routing_table.clone(), timer)
+    );
     task::spawn(async move {
         let pull_id = Rc::new(RefCell::new(0u32));
         let cfg = Rc::new(cfg);
@@ -237,9 +885,19 @@ pub fn start(cfg: Config) {
             let mut stream = TcpStream::accept(1380, 2048, 2048).await.unwrap();
             let pull_id = pull_id.clone();
             let cfg = cfg.clone();
+
+            let drtio_tuple = drtio_tuple.clone();
             task::spawn(async move {
                 info!("received connection");
-                let _ = handle_connection(&mut stream, pull_id, cfg)
+                // Avoid consuming the tuple
+                // Keep the borrowed value on stack
+                let drtio_tuple = drtio_tuple.as_ref().map(
+                    |(aux_mutex, routing_table, timer)| (aux_mutex, routing_table.borrow(), *timer)
+                );
+                let drtio_tuple = drtio_tuple.as_ref().map(
+                    |(aux_mutex, routing_table, timer)| (*aux_mutex, routing_table.deref(), *timer)
+                );
+                let _ = handle_connection(&mut stream, pull_id, cfg, drtio_tuple)
                     .await
                     .map_err(|e| warn!("connection terminated: {:?}", e));
                 let _ = stream.flush().await;

src/runtime/src/rtio_mgt.rs

@@ -540,7 +540,7 @@ pub mod drtio {
         }
     }
 
-    async fn partition_data<PacketF, HandlerF>(
+    pub async fn partition_data<PacketF, HandlerF>(
         linkno: u8,
         aux_mutex: &Rc<Mutex<bool>>,
         routing_table: &RoutingTable,

src/satman/src/main.rs

@@ -38,16 +38,18 @@ use libboard_artiq::{drtio_routing, drtioaux,
                      pl::csr};
 #[cfg(feature = "target_kasli_soc")]
 use libboard_zynq::error_led::ErrorLED;
-use libboard_zynq::{i2c::I2c, print, println, time::Milliseconds, timer::GlobalTimer};
+use libboard_zynq::{i2c::I2c, print, println, slcr, time::Milliseconds, timer::GlobalTimer};
 use libconfig::Config;
 use libcortex_a9::{l2c::enable_l2_cache, regs::MPIDR};
 use libregister::RegisterR;
 use libsupport_zynq::{exception_vectors, ram};
+use mgmt::Manager as CoreManager;
 use routing::Router;
 use subkernel::Manager as KernelManager;
 
 mod analyzer;
 mod dma;
+mod mgmt;
 mod repeater;
 mod routing;
 mod subkernel;
@@ -149,6 +151,7 @@ fn process_aux_packet(
     dma_manager: &mut DmaManager,
     analyzer: &mut Analyzer,
     kernel_manager: &mut KernelManager,
+    core_manager: &mut CoreManager,
     router: &mut Router,
 ) -> Result<(), drtioaux::Error> {
     // In the code below, *_chan_sel_write takes an u8 if there are fewer than 256 channels,
@@ -1010,6 +1013,279 @@ fn process_aux_packet(
             }
             Ok(())
         }
+        drtioaux::Packet::CoreMgmtGetLogRequest {
+            destination: _destination,
+            clear,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+            let mut data_slice = [0; SAT_PAYLOAD_MAX_SIZE];
+            let meta = core_manager.log_get_slice(&mut data_slice);
+            if clear && meta.status.is_first() {
+                mgmt::clear_log();
+            }
+            drtioaux::send(
+                0,
+                &drtioaux::Packet::CoreMgmtGetLogReply {
+                    last: meta.status.is_last(),
+                    length: meta.len as u16,
+                    data: data_slice,
+                },
+            )
+        }
+        drtioaux::Packet::CoreMgmtClearLogRequest {
+            destination: _destination,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+            mgmt::clear_log();
+            drtioaux::send(0, &drtioaux::Packet::CoreMgmtAck)
+        }
+        drtioaux::Packet::CoreMgmtSetLogLevelRequest {
+            destination: _destination,
+            log_level,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            if let Ok(level_filter) = mgmt::byte_to_level_filter(log_level) {
+                info!("Changing log level to {}", level_filter);
+                log::set_max_level(level_filter);
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtAck)
+            } else {
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+            }
+        }
+        drtioaux::Packet::CoreMgmtSetUartLogLevelRequest {
+            destination: _destination,
+            log_level,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            if let Ok(level_filter) = mgmt::byte_to_level_filter(log_level) {
+                info!("Changing log level to {}", level_filter);
+                unsafe {
+                    logger::BufferLogger::get_logger()
+                        .as_ref()
+                        .unwrap()
+                        .set_uart_log_level(level_filter);
+                }
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtAck)
+            } else {
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+            }
+        }
+        drtioaux::Packet::CoreMgmtConfigReadRequest {
+            destination: _destination,
+            length,
+            key,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            let mut value_slice = [0; SAT_PAYLOAD_MAX_SIZE];
+
+            let key_slice = &key[..length as usize];
+            if !key_slice.is_ascii() {
+                error!("invalid key");
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+            } else {
+                let key = core::str::from_utf8(key_slice).unwrap();
+                if core_manager.fetch_config_value(key).is_ok() {
+                    let meta = core_manager.get_config_value_slice(&mut value_slice);
+                    drtioaux::send(
+                        0,
+                        &drtioaux::Packet::CoreMgmtConfigReadReply {
+                            last: meta.status.is_last(),
+                            length: meta.len as u16,
+                            value: value_slice,
+                        },
+                    )
+                } else {
+                    drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+                }
+            }
+        }
+        drtioaux::Packet::CoreMgmtConfigReadContinue {
+            destination: _destination,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            let mut value_slice = [0; SAT_PAYLOAD_MAX_SIZE];
+            let meta = core_manager.get_config_value_slice(&mut value_slice);
+            drtioaux::send(
+                0,
+                &drtioaux::Packet::CoreMgmtConfigReadReply {
+                    last: meta.status.is_last(),
+                    length: meta.len as u16,
+                    value: value_slice,
+                },
+            )
+        }
+        drtioaux::Packet::CoreMgmtConfigWriteRequest {
+            destination: _destination,
+            last,
+            length,
+            data,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            core_manager.add_data(&data, length as usize);
+
+            let mut succeeded = true;
+            if last {
+                succeeded = core_manager.write_config().is_ok();
+                core_manager.clear_data();
+            }
+
+            if succeeded {
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtAck)
+            } else {
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+            }
+        }
+        drtioaux::Packet::CoreMgmtConfigRemoveRequest {
+            destination: _destination,
+            length,
+            key,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            let key_slice = &key[..length as usize];
+            if !key_slice.is_ascii() {
+                error!("invalid key");
+                drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+            } else {
+                let key = core::str::from_utf8(key_slice).unwrap();
+                if core_manager.remove_config(key).is_ok() {
+                    drtioaux::send(0, &drtioaux::Packet::CoreMgmtAck)
+                } else {
+                    drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+                }
+            }
+        }
+        drtioaux::Packet::CoreMgmtConfigEraseRequest {
+            destination: _destination,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            error!("config erase not supported on zynq device");
+            drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+        }
+        drtioaux::Packet::CoreMgmtRebootRequest {
+            destination: _destination,
+        } => {
+            info!("received reboot request");
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            drtioaux::send(0, &drtioaux::Packet::CoreMgmtAck)?;
+            info!("reboot imminent");
+            slcr::reboot();
+            Ok(())
+        }
+        drtioaux::Packet::CoreMgmtAllocatorDebugRequest {
+            destination: _destination,
+        } => {
+            forward!(
+                router,
+                _routing_table,
+                _destination,
+                *rank,
+                *self_destination,
+                _repeaters,
+                &packet,
+                timer
+            );
+
+            error!("debug allocator not supported on zynq device");
+            drtioaux::send(0, &drtioaux::Packet::CoreMgmtNack)
+        }
 
         p => {
             warn!("received unexpected aux packet: {:?}", p);
@@ -1028,6 +1304,7 @@ fn process_aux_packets(
     dma_manager: &mut DmaManager,
     analyzer: &mut Analyzer,
     kernel_manager: &mut KernelManager,
+    core_manager: &mut CoreManager,
     router: &mut Router,
 ) {
     let result = drtioaux::recv(0).and_then(|packet| {
@@ -1043,6 +1320,7 @@ fn process_aux_packets(
                 dma_manager,
                 analyzer,
                 kernel_manager,
+                core_manager,
                 router,
             )
         } else {
@@ -1239,7 +1517,7 @@ pub extern "C" fn main_core0() -> i32 {
     #[cfg(has_si549)]
     si549::helper_setup(&mut timer, &SI549_SETTINGS).expect("cannot initialize helper Si549");
 
-    let cfg = match Config::new() {
+    let mut cfg = match Config::new() {
         Ok(cfg) => cfg,
         Err(err) => {
             warn!("config initialization failed: {}", err);
@@ -1314,6 +1592,7 @@ pub extern "C" fn main_core0() -> i32 {
         let mut dma_manager = DmaManager::new();
         let mut analyzer = Analyzer::new();
         let mut kernel_manager = KernelManager::new(&mut control);
+        let mut core_manager = CoreManager::new(&mut cfg);
 
         drtioaux::reset(0);
         drtiosat_reset(false);
@@ -1331,6 +1610,7 @@ pub extern "C" fn main_core0() -> i32 {
                 &mut dma_manager,
                 &mut analyzer,
                 &mut kernel_manager,
+                &mut core_manager,
                 &mut router,
             );
             #[allow(unused_mut)]

src/satman/src/mgmt.rs

@@ -0,0 +1,116 @@
+use alloc::vec::Vec;
+
+use io::{Cursor, ProtoRead, ProtoWrite};
+use libboard_artiq::{drtioaux_proto::SAT_PAYLOAD_MAX_SIZE,
+                     logger::{BufferLogger, LogBufferRef}};
+use libconfig::Config;
+use log::{self, debug, error, info, warn, LevelFilter};
+
+use crate::routing::{SliceMeta, Sliceable};
+
+type Result<T> = core::result::Result<T, ()>;
+
+pub fn byte_to_level_filter(level_byte: u8) -> Result<log::LevelFilter> {
+    Ok(match level_byte {
+        0 => log::LevelFilter::Off,
+        1 => log::LevelFilter::Error,
+        2 => log::LevelFilter::Warn,
+        3 => log::LevelFilter::Info,
+        4 => log::LevelFilter::Debug,
+        5 => log::LevelFilter::Trace,
+        lv => {
+            error!("unknown log level: {}", lv);
+            return Err(());
+        }
+    })
+}
+
+fn get_logger_buffer_pred() -> LogBufferRef<'static> {
+    let logger = unsafe { BufferLogger::get_logger().as_mut().unwrap() };
+    loop {
+        if let Some(buffer_ref) = logger.buffer() {
+            return buffer_ref;
+        }
+    }
+}
+
+fn get_logger_buffer() -> LogBufferRef<'static> {
+    get_logger_buffer_pred()
+}
+
+pub fn clear_log() {
+    let mut buffer = get_logger_buffer();
+    buffer.clear();
+}
+
+pub struct Manager<'a> {
+    cfg: &'a mut Config,
+    last_log: Sliceable,
+    current_payload: Cursor<Vec<u8>>,
+    last_value: Sliceable,
+}
+
+impl<'a> Manager<'_> {
+    pub fn new(cfg: &mut Config) -> Manager {
+        Manager {
+            cfg: cfg,
+            last_log: Sliceable::new(0, Vec::new()),
+            current_payload: Cursor::new(Vec::new()),
+            last_value: Sliceable::new(0, Vec::new()),
+        }
+    }
+
+    pub fn log_get_slice(&mut self, data_slice: &mut [u8; SAT_PAYLOAD_MAX_SIZE]) -> SliceMeta {
+        // Populate buffer if depleted
+        if self.last_log.at_end() {
+            self.last_log.extend(get_logger_buffer().extract().as_bytes());
+        }
+
+        self.last_log.get_slice_satellite(data_slice)
+    }
+
+    pub fn fetch_config_value(&mut self, key: &str) -> Result<()> {
+        self.cfg
+            .read(&key)
+            .map(|value| {
+                debug!("got value");
+                self.last_value = Sliceable::new(0, value)
+            })
+            .map_err(|_| warn!("read error: no such key"))
+    }
+
+    pub fn get_config_value_slice(&mut self, data_slice: &mut [u8; SAT_PAYLOAD_MAX_SIZE]) -> SliceMeta {
+        self.last_value.get_slice_satellite(data_slice)
+    }
+
+    pub fn add_data(&mut self, data: &[u8], data_len: usize) {
+        self.current_payload.write_all(&data[..data_len]).unwrap();
+    }
+
+    pub fn clear_data(&mut self) {
+        self.current_payload.get_mut().clear();
+        self.current_payload.set_position(0);
+    }
+
+    pub fn write_config(&mut self) -> Result<()> {
+        let key = self
+            .current_payload
+            .read_string()
+            .map_err(|_err| error!("error on reading key"))?;
+        debug!("write key: {}", key);
+        let value = self.current_payload.read_bytes().unwrap();
+
+        self.cfg
+            .write(&key, value)
+            .map(|()| debug!("write success"))
+            .map_err(|err| error!("failed to write: {:?}", err))
+    }
+
+    pub fn remove_config(&mut self, key: &str) -> Result<()> {
+        debug!("erase key: {}", key);
+        self.cfg
+            .remove(&key)
+            .map(|()| debug!("erase success"))
+            .map_err(|err| warn!("failed to erase: {:?}", err))
+    }
+}

src/satman/src/routing.rs

@@ -4,7 +4,7 @@ use core::cmp::min;
 #[cfg(has_drtio_routing)]
 use libboard_artiq::pl::csr;
 use libboard_artiq::{drtio_routing, drtioaux,
-                     drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE}};
+                     drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE}};
 
 pub struct SliceMeta {
     pub destination: u8,
@@ -58,6 +58,7 @@ impl Sliceable {
     }
 
     get_slice_fn!(get_slice_master, MASTER_PAYLOAD_MAX_SIZE);
+    get_slice_fn!(get_slice_satellite, SAT_PAYLOAD_MAX_SIZE);
 }
 
 // Packets from downstream (further satellites) are received and routed appropriately.