PUS packets arriving now

embedded-examples/stm32h7-rtic/Cargo.lock

@@ -251,6 +251,31 @@ name = "embedded-hal"
 version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "361a90feb7004eca4019fb28352a9465666b24f840f5c3cddf0ff13920590b89"
+dependencies = [
+ "defmt",
+]
+
+[[package]]
+name = "embedded-hal-async"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c4c685bbef7fe13c3c6dd4da26841ed3980ef33e841cddfa15ce8a8fb3f1884"
+dependencies = [
+ "defmt",
+ "embedded-hal 1.0.0",
+]
+
+[[package]]
+name = "embedded-hal-bus"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57b4e6ede84339ebdb418cd986e6320a34b017cdf99b5cc3efceec6450b06886"
+dependencies = [
+ "critical-section",
+ "defmt",
+ "embedded-hal 1.0.0",
+ "embedded-hal-async",
+]
 
 [[package]]
 name = "embedded-storage"
@@ -455,6 +480,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
+[[package]]
+name = "portable-atomic"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7170ef9988bc169ba16dd36a7fa041e5c4cbeb6a35b76d4c03daded371eae7c0"
+
 [[package]]
 name = "proc-macro-error"
 version = "1.0.4"
@@ -553,6 +584,22 @@ dependencies = [
  "rtic-time",
 ]
 
+[[package]]
+name = "rtic-sync"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49b1200137ccb2bf272a1801fa6e27264535facd356cb2c1d5bc8e12aa211bad"
+dependencies = [
+ "critical-section",
+ "defmt",
+ "embedded-hal 1.0.0",
+ "embedded-hal-async",
+ "embedded-hal-bus",
+ "heapless 0.8.0",
+ "portable-atomic",
+ "rtic-common",
+]
+
 [[package]]
 name = "rtic-time"
 version = "1.3.0"
@@ -623,6 +670,7 @@ dependencies = [
  "panic-probe",
  "rtic",
  "rtic-monotonics",
+ "rtic-sync",
  "satrs",
  "smoltcp",
  "stm32h7xx-hal",

embedded-examples/stm32h7-rtic/Cargo.toml

@@ -22,6 +22,7 @@ panic-probe = { version = "0.3", features = ["print-defmt"] }
 cortex-m-semihosting = "0.5.0"
 stm32h7xx-hal = { version="0.16", features= ["stm32h743v", "ethernet"] }
 embedded-alloc = "0.5"
+rtic-sync = { version = "1", features = ["defmt-03"] }
 
 [dependencies.smoltcp]
 version = "0.11.0"

embedded-examples/stm32h7-rtic/pyclient/def_tmtc_conf.json

@@ -1,4 +1,4 @@
 {
-    "com_if": "serial_cobs",
+    "com_if": "udp",
-    "serial_baudrate": 115200
+    "tcpip_udp_port": 7301
 }

embedded-examples/stm32h7-rtic/src/main.rs

@@ -5,10 +5,11 @@ extern crate alloc;
 use rtic::app;
 use rtic_monotonics::systick::Systick;
 use rtic_monotonics::Monotonic;
-use satrs::pool::StaticHeaplessMemoryPool;
+use satrs::pool::{PoolAddr, PoolProvider, StaticHeaplessMemoryPool};
 use satrs::static_subpool;
 // global logger + panicking-behavior + memory layout
 use satrs_stm32h7_nucleo_rtic as _;
+use smoltcp::socket::udp::UdpMetadata;
 use smoltcp::socket::{dhcpv4, udp};
 
 use core::mem::MaybeUninit;
@@ -22,6 +23,12 @@ const PORT: u16 = 7301;
 
 const HEAP_SIZE: usize = 131_072;
 
+const TC_SOURCE_CHANNEL_DEPTH: usize = 16;
+pub type SharedPool = StaticHeaplessMemoryPool<3>;
+pub type TcSourceChannel = rtic_sync::channel::Channel<PoolAddr, TC_SOURCE_CHANNEL_DEPTH>;
+pub type TcSourceTx = rtic_sync::channel::Sender<'static, PoolAddr, TC_SOURCE_CHANNEL_DEPTH>;
+pub type TcSourceRx = rtic_sync::channel::Receiver<'static, PoolAddr, TC_SOURCE_CHANNEL_DEPTH>;
+
 #[global_allocator]
 static HEAP: Heap = Heap::empty();
 
@@ -41,25 +48,23 @@ pub struct NetStorageStatic<'a> {
 // initialised by the runtime
 static mut STORE: MaybeUninit<NetStorageStatic> = MaybeUninit::uninit();
 
-static mut UDP_RX_META: [udp::PacketMetadata; 4] = [udp::PacketMetadata::EMPTY; 4];
+static mut UDP_RX_META: [udp::PacketMetadata; 12] = [udp::PacketMetadata::EMPTY; 12];
 static mut UDP_RX: [u8; 2048] = [0; 2048];
-static mut UDP_TX_META: [udp::PacketMetadata; 4] = [udp::PacketMetadata::EMPTY; 4];
+static mut UDP_TX_META: [udp::PacketMetadata; 12] = [udp::PacketMetadata::EMPTY; 12];
 static mut UDP_TX: [u8; 2048] = [0; 2048];
 
 /// Locally administered MAC address
 const MAC_ADDRESS: [u8; 6] = [0x02, 0x00, 0x11, 0x22, 0x33, 0x44];
 
-pub struct Net<'a> {
+pub struct Net {
     iface: Interface,
     ethdev: ethernet::EthernetDMA<4, 4>,
-    sockets: SocketSet<'a>,
     dhcp_handle: SocketHandle,
-    udp_handle: SocketHandle,
 }
 
-impl<'a> Net<'a> {
+impl Net {
     pub fn new(
-        store: &'a mut NetStorageStatic<'a>,
+        sockets: &mut SocketSet<'static>,
         mut ethdev: ethernet::EthernetDMA<4, 4>,
         ethernet_addr: HardwareAddress,
     ) -> Self {
@@ -69,51 +74,31 @@ impl<'a> Net<'a> {
             &mut ethdev,
             smoltcp::time::Instant::from_millis((Systick::now() - Systick::ZERO).to_millis()),
         );
-        // SAFETY: The RX and TX buffers are passed here and not used anywhere else.
-        let udp_rx_buffer =
-            smoltcp::socket::udp::PacketBuffer::new(unsafe { &mut UDP_RX_META[..] }, unsafe {
-                &mut UDP_RX[..]
-            });
-        let udp_tx_buffer =
-            smoltcp::socket::udp::PacketBuffer::new(unsafe { &mut UDP_TX_META[..] }, unsafe {
-                &mut UDP_TX[..]
-            });
-        let udp_socket = smoltcp::socket::udp::Socket::new(udp_rx_buffer, udp_tx_buffer);
-
         // Create sockets
         let dhcp_socket = dhcpv4::Socket::new();
-
         iface.update_ip_addrs(|addrs| {
             let _ = addrs.push(IpCidr::new(IpAddress::v4(192, 168, 1, 99), 0));
         });
 
-        let mut sockets = SocketSet::new(&mut store.socket_storage[..]);
         let dhcp_handle = sockets.add(dhcp_socket);
-        let udp_handle = sockets.add(udp_socket);
+        Net {
-        Net::<'a> {
             iface,
             ethdev,
-            sockets,
             dhcp_handle,
-            udp_handle,
         }
     }
 
     /// Polls on the ethernet interface. You should refer to the smoltcp
     /// documentation for poll() to understand how to call poll efficiently
-    pub fn poll(&mut self) -> bool {
+    pub fn poll<'a>(&mut self, sockets: &'a mut SocketSet) -> bool {
         let uptime = Systick::now() - Systick::ZERO;
         let timestamp = smoltcp::time::Instant::from_millis(uptime.to_millis());
 
-        self.iface
+        self.iface.poll(timestamp, &mut self.ethdev, sockets)
-            .poll(timestamp, &mut self.ethdev, &mut self.sockets)
     }
 
-    pub fn poll_dhcp(&mut self) -> Option<dhcpv4::Event> {
+    pub fn poll_dhcp<'a>(&mut self, sockets: &'a mut SocketSet) -> Option<dhcpv4::Event<'a>> {
-        let opt_event = self
+        let opt_event = sockets.get_mut::<dhcpv4::Socket>(self.dhcp_handle).poll();
-            .sockets
-            .get_mut::<dhcpv4::Socket>(self.dhcp_handle)
-            .poll();
         if let Some(event) = &opt_event {
             match event {
                 dhcpv4::Event::Deconfigured => {
@@ -144,25 +129,61 @@ impl<'a> Net<'a> {
         }
         opt_event
     }
+}
 
-    pub fn poll_udp(&mut self) {
+pub struct UdpNet {
-        let socket = self.sockets.get_mut::<udp::Socket>(self.udp_handle);
+    udp_handle: SocketHandle,
+    last_client: Option<UdpMetadata>,
+    tc_source_tx: TcSourceTx,
+}
+
+impl UdpNet {
+    pub fn new<'sockets>(sockets: &mut SocketSet<'sockets>, tc_source_tx: TcSourceTx) -> Self {
+        // SAFETY: The RX and TX buffers are passed here and not used anywhere else.
+        let udp_rx_buffer =
+            smoltcp::socket::udp::PacketBuffer::new(unsafe { &mut UDP_RX_META[..] }, unsafe {
+                &mut UDP_RX[..]
+            });
+        let udp_tx_buffer =
+            smoltcp::socket::udp::PacketBuffer::new(unsafe { &mut UDP_TX_META[..] }, unsafe {
+                &mut UDP_TX[..]
+            });
+        let udp_socket = smoltcp::socket::udp::Socket::new(udp_rx_buffer, udp_tx_buffer);
+
+        let udp_handle = sockets.add(udp_socket);
+        Self {
+            udp_handle,
+            last_client: None,
+            tc_source_tx,
+        }
+    }
+
+    pub fn poll<'sockets>(
+        &mut self,
+        sockets: &'sockets mut SocketSet,
+        shared_pool: &mut SharedPool,
+    ) {
+        let socket = sockets.get_mut::<udp::Socket>(self.udp_handle);
         if !socket.is_open() {
             if let Err(e) = socket.bind(PORT) {
-                defmt::warn!("binding UDP socket failed");
+                defmt::warn!("binding UDP socket failed: {}", e);
             }
         }
         loop {
             match socket.recv() {
                 Ok((data, client)) => {
-                    /*defmt::info!("UDP: rx {} bytes from {}", data.len(), endpoint);
+                    match shared_pool.add(data) {
-                    if let Ok(recv_str) = str::from_utf8(data) {
+                        Ok(store_addr) => {
-                        defmt::info!("recv: {}", recv_str);
+                            if let Err(e) = self.tc_source_tx.try_send(store_addr) {
+                                defmt::warn!("TC source channel is full: {}", e);
                             }
-                    */
+                        }
+                        Err(e) => {
+                            defmt::warn!("could not add UDP packet to shared pool: {}", e);
+                        }
+                    }
+                    self.last_client = Some(client);
                     // TODO: Implement packet wiretapping.
-                    // TODO: Store last endpoint.
-                    // TODO: Send packet to PUS/CCSDS distributor via message queue.
                 }
                 Err(e) => match e {
                     udp::RecvError::Exhausted => {
@@ -184,29 +205,35 @@ mod app {
     use super::*;
     use rtic_monotonics::systick::fugit::MillisDurationU32;
     use rtic_monotonics::systick::Systick;
+    use satrs::spacepackets::ecss::tc::PusTcReader;
     use stm32h7xx_hal::ethernet::{EthernetMAC, PHY};
     use stm32h7xx_hal::gpio::{Output, Pin};
     use stm32h7xx_hal::prelude::*;
     use stm32h7xx_hal::stm32::Interrupt;
 
-    #[shared]
-    struct Shared {
-        blink_freq: MillisDurationU32,
-        eth_link_up: bool,
-    }
-
     struct BlinkyLeds {
         led1: Pin<'B', 7, Output>,
         led2: Pin<'B', 14, Output>,
     }
+
     #[local]
     struct Local {
         leds: BlinkyLeds,
         link_led: Pin<'B', 0, Output>,
-        net: Net<'static>,
+        net: Net,
+        udp: UdpNet,
+        tc_source_rx: TcSourceRx,
         phy: ethernet::phy::LAN8742A<EthernetMAC>,
     }
 
+    #[shared]
+    struct Shared {
+        blink_freq: MillisDurationU32,
+        eth_link_up: bool,
+        sockets: SocketSet<'static>,
+        shared_pool: SharedPool,
+    }
+
     #[init]
     fn init(mut cx: init::Context) -> (Shared, Local) {
         defmt::println!("Starting sat-rs demo application for the STM32H743ZIT");
@@ -318,9 +345,14 @@ mod app {
             STORE.assume_init_mut()
         };
 
-        let net = Net::new(store, eth_dma, mac_addr.into());
+        let (tc_source_tx, tc_source_rx) =
+            rtic_sync::make_channel!(PoolAddr, TC_SOURCE_CHANNEL_DEPTH);
 
-        let mut heapless_pool: StaticHeaplessMemoryPool<3> = StaticHeaplessMemoryPool::new(true);
+        let mut sockets = SocketSet::new(&mut store.socket_storage[..]);
+        let net = Net::new(&mut sockets, eth_dma, mac_addr.into());
+        let udp = UdpNet::new(&mut sockets, tc_source_tx);
+
+        let mut shared_pool: SharedPool = StaticHeaplessMemoryPool::new(true);
         static_subpool!(
             SUBPOOL_SMALL,
             SUBPOOL_SMALL_SIZES,
@@ -343,28 +375,28 @@ mod app {
             link_section = ".axisram"
         );
 
-        heapless_pool
+        shared_pool
             .grow(
                 unsafe { SUBPOOL_SMALL.assume_init_mut() },
                 unsafe { SUBPOOL_SMALL_SIZES.assume_init_mut() },
                 SUBPOOL_SMALL_NUM_BLOCKS,
-                false,
+                true,
             )
             .expect("growing heapless memory pool failed");
-        heapless_pool
+        shared_pool
             .grow(
                 unsafe { SUBPOOL_MEDIUM.assume_init_mut() },
                 unsafe { SUBPOOL_MEDIUM_SIZES.assume_init_mut() },
                 SUBPOOL_MEDIUM_NUM_BLOCKS,
-                false,
+                true,
             )
             .expect("growing heapless memory pool failed");
-        heapless_pool
+        shared_pool
             .grow(
                 unsafe { SUBPOOL_LARGE.assume_init_mut() },
                 unsafe { SUBPOOL_LARGE_SIZES.assume_init_mut() },
                 SUBPOOL_LARGE_NUM_BLOCKS,
-                false,
+                true,
             )
             .expect("growing heapless memory pool failed");
 
@@ -374,23 +406,30 @@ mod app {
         unsafe { HEAP.init(HEAP_MEM.as_ptr() as usize, HEAP_SIZE) }
 
         eth_link_check::spawn().expect("eth link check failed");
-        blink::spawn().expect("spawning blink task failed");
+        blinky::spawn().expect("spawning blink task failed");
+        udp_task::spawn().expect("spawning UDP task failed");
+        tc_source_task::spawn().expect("spawning TC source task failed");
+
         (
             Shared {
                 blink_freq: MillisDurationU32::from_ticks(DEFAULT_BLINK_FREQ_MS),
                 eth_link_up: false,
+                sockets,
+                shared_pool,
             },
             Local {
                 link_led,
                 leds,
                 net,
+                udp,
+                tc_source_rx,
                 phy: lan8742a,
             },
         )
     }
 
     #[task(local = [leds], shared=[blink_freq])]
-    async fn blink(mut cx: blink::Context) {
+    async fn blinky(mut cx: blinky::Context) {
         let leds = cx.local.leds;
         loop {
             leds.led1.toggle();
@@ -400,6 +439,7 @@ mod app {
         }
     }
 
+    /// This task checks for the network link.
     #[task(local=[link_led, phy], shared=[eth_link_up])]
     async fn eth_link_check(mut cx: eth_link_check::Context) {
         let phy = cx.local.phy;
@@ -421,16 +461,68 @@ mod app {
         }
     }
 
-    #[task(binds=ETH, local=[net])]
+    #[task(binds=ETH, local=[net], shared=[sockets])]
-    fn eth_isr(cx: eth_isr::Context) {
+    fn eth_isr(mut cx: eth_isr::Context) {
         // SAFETY: We do not write the register mentioned inside the docs anywhere else.
         unsafe {
             ethernet::interrupt_handler();
         }
-        // TODO: I am not fully sure whether we should do everything here. Mabye we should
+        // Check and process ETH frames and DHCP. UDP is checked in a different task.
-        // offload this to a regular task?
+        cx.shared.sockets.lock(|sockets| {
-        cx.local.net.poll();
+            cx.local.net.poll(sockets);
-        cx.local.net.poll_dhcp();
+            cx.local.net.poll_dhcp(sockets);
-        cx.local.net.poll_udp();
+        });
+    }
+
+    /// This task routes UDP packets.
+    #[task(local=[udp], shared=[sockets, shared_pool])]
+    async fn udp_task(mut cx: udp_task::Context) {
+        loop {
+            cx.shared.sockets.lock(|sockets| {
+                cx.shared.shared_pool.lock(|pool| {
+                    cx.local.udp.poll(sockets, pool);
+                })
+            });
+            Systick::delay(40.millis()).await;
+        }
+    }
+
+    /// This task handles all the incoming telecommands.
+    #[task(local=[read_buf: [u8; 1024] = [0; 1024], tc_source_rx], shared=[shared_pool])]
+    async fn tc_source_task(mut cx: tc_source_task::Context) {
+        loop {
+            let recv_result = cx.local.tc_source_rx.recv().await;
+            match recv_result {
+                Ok(pool_addr) => {
+                    cx.shared.shared_pool.lock(|pool| {
+                        match pool.read(&pool_addr, cx.local.read_buf.as_mut()) {
+                            Ok(packet_len) => {
+                                defmt::info!("received {} bytes in the TC source task", packet_len);
+                                match PusTcReader::new(&cx.local.read_buf[0..packet_len]) {
+                                    Ok((packet, _tc_len)) => {
+                                        // TODO: Handle packet here or dispatch to dedicated PUS
+                                        // handler? Dispatching could simplify some things and make
+                                        // the software more scalable..
+                                        defmt::info!("received PUS packet: {}", packet);
+                                    }
+                                    Err(e) => {
+                                        defmt::info!("invalid TC format, not a PUS packet: {}", e);
+                                    }
+                                }
+                                if let Err(e) = pool.delete(pool_addr) {
+                                    defmt::warn!("deleting TC data failed: {}", e);
+                                }
+                            }
+                            Err(e) => {
+                                defmt::warn!("TC packet read failed: {}", e);
+                            }
+                        }
+                    });
+                }
+                Err(e) => {
+                    defmt::warn!("TC source reception error: {}", e);
+                }
+            };
+        }
     }
 }

satrs/src/pool.rs

@@ -133,6 +133,7 @@ impl Display for StaticPoolAddr {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
 pub enum StoreIdError {
     InvalidSubpool(u16),
     InvalidPacketIdx(u16),
@@ -156,6 +157,7 @@ impl Error for StoreIdError {}
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
 pub enum PoolError {
     /// Requested data block is too large
     DataTooLarge(usize),
@@ -396,7 +398,7 @@ pub mod heapless_mod {
                 core::mem::MaybeUninit::new([0; $num_blocks * $block_size]);
             #[$meta_data]
             static mut $sizes_list_name: core::mem::MaybeUninit<[usize; $num_blocks]> =
-                core::mem::MaybeUninit::new([satrs::pool::STORE_FREE; $num_blocks]);
+                core::mem::MaybeUninit::new([$crate::pool::STORE_FREE; $num_blocks]);
         };
     }