Merge pull request 'prep next release' (#171) from pre-v0.2.0-rc.4 into main

Reviewed-on: #171

satrs-example/pytmtc/main.py

@@ -103,7 +103,9 @@ class PusHandler(GenericApidHandlerBase):
 
     def handle_tm(self, apid: int, packet: bytes, _user_args: Any):
         try:
-            pus_tm = PusTelemetry.unpack(packet, time_reader=CdsShortTimestamp.empty())
+            pus_tm = PusTelemetry.unpack(
+                packet, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
+            )
         except ValueError as e:
             _LOGGER.warning("Could not generate PUS TM object from raw data")
             _LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}")
@@ -111,7 +113,7 @@ class PusHandler(GenericApidHandlerBase):
         service = pus_tm.service
         if service == 1:
             tm_packet = Service1Tm.unpack(
-                data=packet, params=UnpackParams(CdsShortTimestamp.empty(), 1, 2)
+                data=packet, params=UnpackParams(CdsShortTimestamp.TIMESTAMP_SIZE, 1, 2)
             )
             res = self.verif_wrapper.add_tm(tm_packet)
             if res is None:
@@ -128,7 +130,9 @@ class PusHandler(GenericApidHandlerBase):
         elif service == 3:
             _LOGGER.info("No handling for HK packets implemented")
             _LOGGER.info(f"Raw packet: 0x[{packet.hex(sep=',')}]")
-            pus_tm = PusTelemetry.unpack(packet, time_reader=CdsShortTimestamp.empty())
+            pus_tm = PusTelemetry.unpack(
+                packet, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
+            )
             if pus_tm.subservice == 25:
                 if len(pus_tm.source_data) < 8:
                     raise ValueError("No addressable ID in HK packet")
@@ -136,7 +140,7 @@ class PusHandler(GenericApidHandlerBase):
                 _LOGGER.info(json_str)
         elif service == 5:
             tm_packet = PusTelemetry.unpack(
-                packet, time_reader=CdsShortTimestamp.empty()
+                packet, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
             )
             src_data = tm_packet.source_data
             event_u32 = EventU32.unpack(src_data)
@@ -145,7 +149,7 @@ class PusHandler(GenericApidHandlerBase):
                 _LOGGER.info("Received test event")
         elif service == 17:
             tm_packet = Service17Tm.unpack(
-                packet, time_reader=CdsShortTimestamp.empty()
+                packet, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
             )
             if tm_packet.subservice == 2:
                 self.file_logger.info("Received Ping Reply TM[17,2]")
@@ -162,7 +166,7 @@ class PusHandler(GenericApidHandlerBase):
                 f"The service {service} is not implemented in Telemetry Factory"
             )
             tm_packet = PusTelemetry.unpack(
-                packet, time_reader=CdsShortTimestamp.empty()
+                packet, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
             )
         self.raw_logger.log_tm(pus_tm)
 
@@ -197,15 +201,15 @@ class TcHandler(TcHandlerBase):
             _LOGGER.info(log_entry.log_str)
 
     def queue_finished_cb(self, info: ProcedureWrapper):
-        if info.proc_type == TcProcedureType.DEFAULT:
-            def_proc = info.to_def_procedure()
+        if info.proc_type == TcProcedureType.TREE_COMMANDING:
+            def_proc = info.to_tree_commanding_procedure()
             _LOGGER.info(f"Queue handling finished for command {def_proc.cmd_path}")
 
     def feed_cb(self, info: ProcedureWrapper, wrapper: FeedWrapper):
         q = self.queue_helper
         q.queue_wrapper = wrapper.queue_wrapper
-        if info.proc_type == TcProcedureType.DEFAULT:
-            def_proc = info.to_def_procedure()
+        if info.proc_type == TcProcedureType.TREE_COMMANDING:
+            def_proc = info.to_tree_commanding_procedure()
             assert def_proc.cmd_path is not None
             pus_tc.pack_pus_telecommands(q, def_proc.cmd_path)
 
@@ -256,6 +260,7 @@ def main():
         while True:
             state = tmtc_backend.periodic_op(None)
             if state.request == BackendRequest.TERMINATION_NO_ERROR:
+                tmtc_backend.close_com_if()
                 sys.exit(0)
             elif state.request == BackendRequest.DELAY_IDLE:
                 _LOGGER.info("TMTC Client in IDLE mode")
@@ -270,6 +275,7 @@ def main():
             elif state.request == BackendRequest.CALL_NEXT:
                 pass
     except KeyboardInterrupt:
+        tmtc_backend.close_com_if()
         sys.exit(0)
 
 

satrs-example/pytmtc/requirements.txt

@@ -1,2 +1,2 @@
-tmtccmd == 8.0.0rc1
+tmtccmd == 8.0.0rc2
 # -e git+https://github.com/robamu-org/tmtccmd@97e5e51101a08b21472b3ddecc2063359f7e307a#egg=tmtccmd

satrs-example/src/interface/tcp.rs

@@ -1,3 +1,4 @@
+use std::time::Duration;
 use std::{
     collections::{HashSet, VecDeque},
     fmt::Debug,
@@ -139,7 +140,9 @@ impl<TcSender: PacketSenderRaw<Error = SendError>, SendError: Debug + 'static>
 
     pub fn periodic_operation(&mut self) {
         loop {
-            let result = self.0.handle_all_connections(None);
+            let result = self
+                .0
+                .handle_all_connections(Some(Duration::from_millis(400)));
             match result {
                 Ok(_conn_result) => (),
                 Err(e) => {

satrs-example/src/interface/udp.rs

@@ -114,6 +114,7 @@ impl<
 
 #[cfg(test)]
 mod tests {
+    use std::net::Ipv4Addr;
     use std::{
         cell::RefCell,
         collections::VecDeque,
@@ -182,7 +183,7 @@ mod tests {
 
     #[test]
     fn test_transactions() {
-        let sock_addr = SocketAddr::new(IpAddr::V4(OBSW_SERVER_ADDR), 0);
+        let sock_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);
         let test_receiver = TestSender::default();
         // let tc_queue = test_receiver.tc_vec.clone();
         let udp_tc_server =
@@ -200,8 +201,8 @@ mod tests {
             .unwrap();
         let client = UdpSocket::bind("127.0.0.1:0").expect("Connecting to UDP server failed");
         let client_addr = client.local_addr().unwrap();
-        client.connect(server_addr).unwrap();
-        client.send(&ping_tc).unwrap();
+        println!("{}", server_addr);
+        client.send_to(&ping_tc, server_addr).unwrap();
         udp_dyn_server.periodic_operation();
         {
             let mut queue = udp_dyn_server.udp_tc_server.tc_sender.tc_vec.borrow_mut();

satrs/CHANGELOG.md

@@ -8,6 +8,23 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 # [unreleased]
 
+# [v0.2.0-rc.4] 2024-04-23
+
+## Changed
+
+- The `parse_for_ccsds_space_packets` method now expects a non-mutable slice and does not copy
+  broken tail packets anymore. It also does not expect a mutable `next_write_idx` argument anymore.
+  Instead, a `ParseResult` structure is returned which contains the `packets_found` and an
+  optional `incomplete_tail_start` value.
+
+## Fixed
+
+- `parse_for_ccsds_space_packets` did not detect CCSDS space packets at the buffer end with the
+  smallest possible size of 7 bytes.
+- TCP server component now re-registers the internal `mio::Poll` object if the client reset
+  the connection unexpectedly. Not doing so prevented the server from functioning properly
+  after a re-connect.
+
 # [v0.2.0-rc.3] 2024-04-17
 
 docs-rs hotfix 2

satrs/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "satrs"
-version = "0.2.0-rc.3"
+version = "0.2.0-rc.4"
 edition = "2021"
 rust-version = "1.71.1"
 authors = ["Robin Mueller <muellerr@irs.uni-stuttgart.de>"]

satrs/src/encoding/ccsds.rs

@@ -21,6 +21,13 @@ pub trait SpacePacketValidator {
     fn validate(&self, sp_header: &SpHeader, raw_buf: &[u8]) -> SpValidity;
 }
 
+#[derive(Default, Debug, PartialEq, Eq)]
+pub struct ParseResult {
+    pub packets_found: u32,
+    /// If an incomplete space packet was found, its start index is indicated by this value.
+    pub incomplete_tail_start: Option<usize>,
+}
+
 /// This function parses a given buffer for tightly packed CCSDS space packets. It uses the
 /// [spacepackets::SpHeader] of the CCSDS packets and a user provided [SpacePacketValidator]
 /// to check whether a received space packet is relevant for processing.
@@ -41,35 +48,29 @@ pub trait SpacePacketValidator {
 ///  3. [SpValidity::Skip]: The parser skips the packet using the packet length determined from the
 ///     space packet header.
 pub fn parse_buffer_for_ccsds_space_packets<SendError>(
-    buf: &mut [u8],
+    buf: &[u8],
     packet_validator: &(impl SpacePacketValidator + ?Sized),
     sender_id: ComponentId,
     packet_sender: &(impl PacketSenderRaw<Error = SendError> + ?Sized),
-    next_write_idx: &mut usize,
-) -> Result<u32, SendError> {
-    *next_write_idx = 0;
-    let mut packets_found = 0;
+) -> Result<ParseResult, SendError> {
+    let mut parse_result = ParseResult::default();
     let mut current_idx = 0;
     let buf_len = buf.len();
     loop {
-        if current_idx + 7 >= buf.len() {
+        if current_idx + 7 > buf.len() {
             break;
         }
         let sp_header = SpHeader::from_be_bytes(&buf[current_idx..]).unwrap().0;
-        // let packet_id = u16::from_be_bytes(buf[current_idx..current_idx + 2].try_into().unwrap());
         match packet_validator.validate(&sp_header, &buf[current_idx..]) {
             SpValidity::Valid => {
                 let packet_size = sp_header.total_len();
                 if (current_idx + packet_size) <= buf_len {
                     packet_sender
                         .send_packet(sender_id, &buf[current_idx..current_idx + packet_size])?;
-                    packets_found += 1;
+                    parse_result.packets_found += 1;
                 } else {
                     // Move packet to start of buffer if applicable.
-                    if current_idx > 0 {
-                        buf.copy_within(current_idx.., 0);
-                        *next_write_idx = buf.len() - current_idx;
-                    }
+                    parse_result.incomplete_tail_start = Some(current_idx);
                 }
                 current_idx += packet_size;
                 continue;
@@ -83,14 +84,14 @@ pub fn parse_buffer_for_ccsds_space_packets<SendError>(
             }
         }
     }
-    Ok(packets_found)
+    Ok(parse_result)
 }
 
 #[cfg(test)]
 mod tests {
     use spacepackets::{
         ecss::{tc::PusTcCreator, WritablePusPacket},
-        CcsdsPacket, PacketId, SpHeader,
+        CcsdsPacket, PacketId, PacketSequenceCtrl, PacketType, SequenceFlags, SpHeader,
     };
 
     use crate::{encoding::tests::TcCacher, ComponentId};
@@ -136,17 +137,15 @@ mod tests {
             .write_to_bytes(&mut buffer)
             .expect("writing packet failed");
         let tc_cacher = TcCacher::default();
-        let mut next_write_idx = 0;
         let parse_result = parse_buffer_for_ccsds_space_packets(
-            &mut buffer,
+            &buffer,
             &SimpleVerificator::default(),
             PARSER_ID,
             &tc_cacher,
-            &mut next_write_idx,
         );
         assert!(parse_result.is_ok());
-        let parsed_packets = parse_result.unwrap();
-        assert_eq!(parsed_packets, 1);
+        let parse_result = parse_result.unwrap();
+        assert_eq!(parse_result.packets_found, 1);
         let mut queue = tc_cacher.tc_queue.borrow_mut();
         assert_eq!(queue.len(), 1);
         let packet_with_sender = queue.pop_front().unwrap();
@@ -167,17 +166,15 @@ mod tests {
             .write_to_bytes(&mut buffer[packet_len_ping..])
             .expect("writing packet failed");
         let tc_cacher = TcCacher::default();
-        let mut next_write_idx = 0;
         let parse_result = parse_buffer_for_ccsds_space_packets(
-            &mut buffer,
+            &buffer,
             &SimpleVerificator::default(),
             PARSER_ID,
             &tc_cacher,
-            &mut next_write_idx,
         );
         assert!(parse_result.is_ok());
-        let parsed_packets = parse_result.unwrap();
-        assert_eq!(parsed_packets, 2);
+        let parse_result = parse_result.unwrap();
+        assert_eq!(parse_result.packets_found, 2);
         let mut queue = tc_cacher.tc_queue.borrow_mut();
         assert_eq!(queue.len(), 2);
         let packet_with_addr = queue.pop_front().unwrap();
@@ -205,18 +202,12 @@ mod tests {
             .write_to_bytes(&mut buffer[packet_len_ping..])
             .expect("writing packet failed");
         let tc_cacher = TcCacher::default();
-        let mut next_write_idx = 0;
         let verificator = SimpleVerificator::new_with_second_id();
-        let parse_result = parse_buffer_for_ccsds_space_packets(
-            &mut buffer,
-            &verificator,
-            PARSER_ID,
-            &tc_cacher,
-            &mut next_write_idx,
-        );
+        let parse_result =
+            parse_buffer_for_ccsds_space_packets(&buffer, &verificator, PARSER_ID, &tc_cacher);
         assert!(parse_result.is_ok());
-        let parsed_packets = parse_result.unwrap();
-        assert_eq!(parsed_packets, 2);
+        let parse_result = parse_result.unwrap();
+        assert_eq!(parse_result.packets_found, 2);
         let mut queue = tc_cacher.tc_queue.borrow_mut();
         assert_eq!(queue.len(), 2);
         let packet_with_addr = queue.pop_front().unwrap();
@@ -242,24 +233,24 @@ mod tests {
             .write_to_bytes(&mut buffer[packet_len_ping..])
             .expect("writing packet failed");
         let tc_cacher = TcCacher::default();
-        let mut next_write_idx = 0;
         let verificator = SimpleVerificator::new_with_second_id();
         let parse_result = parse_buffer_for_ccsds_space_packets(
-            &mut buffer[..packet_len_ping + packet_len_action - 4],
+            &buffer[..packet_len_ping + packet_len_action - 4],
             &verificator,
             PARSER_ID,
             &tc_cacher,
-            &mut next_write_idx,
         );
         assert!(parse_result.is_ok());
-        let parsed_packets = parse_result.unwrap();
-        assert_eq!(parsed_packets, 1);
+        let parse_result = parse_result.unwrap();
+        assert_eq!(parse_result.packets_found, 1);
+        assert!(parse_result.incomplete_tail_start.is_some());
+        let incomplete_tail_idx = parse_result.incomplete_tail_start.unwrap();
+        assert_eq!(incomplete_tail_idx, packet_len_ping);
 
         let queue = tc_cacher.tc_queue.borrow();
         assert_eq!(queue.len(), 1);
         // The broken packet was moved to the start, so the next write index should be after the
         // last segment missing 4 bytes.
-        assert_eq!(next_write_idx, packet_len_action - 4);
     }
 
     #[test]
@@ -273,19 +264,36 @@ mod tests {
         let tc_cacher = TcCacher::default();
 
         let verificator = SimpleVerificator::new_with_second_id();
-        let mut next_write_idx = 0;
         let parse_result = parse_buffer_for_ccsds_space_packets(
-            &mut buffer[..packet_len_ping - 4],
+            &buffer[..packet_len_ping - 4],
             &verificator,
             PARSER_ID,
             &tc_cacher,
-            &mut next_write_idx,
         );
-        assert_eq!(next_write_idx, 0);
         assert!(parse_result.is_ok());
-        let parsed_packets = parse_result.unwrap();
-        assert_eq!(parsed_packets, 0);
+        let parse_result = parse_result.unwrap();
+        assert_eq!(parse_result.packets_found, 0);
         let queue = tc_cacher.tc_queue.borrow();
         assert_eq!(queue.len(), 0);
     }
+
+    #[test]
+    fn test_smallest_packet() {
+        let ccsds_header_only = SpHeader::new(
+            PacketId::new(PacketType::Tc, true, TEST_APID_0),
+            PacketSequenceCtrl::new(SequenceFlags::Unsegmented, 0),
+            0,
+        );
+        let mut buf: [u8; 7] = [0; 7];
+        ccsds_header_only
+            .write_to_be_bytes(&mut buf)
+            .expect("writing failed");
+        let verificator = SimpleVerificator::default();
+        let tc_cacher = TcCacher::default();
+        let parse_result =
+            parse_buffer_for_ccsds_space_packets(&buf, &verificator, PARSER_ID, &tc_cacher);
+        assert!(parse_result.is_ok());
+        let parse_result = parse_result.unwrap();
+        assert_eq!(parse_result.packets_found, 1);
+    }
 }

satrs/src/hal/std/tcp_server.rs

@@ -9,8 +9,6 @@ use mio::{Events, Interest, Poll, Token};
 use socket2::{Domain, Socket, Type};
 use std::io::{self, Read};
 use std::net::SocketAddr;
-// use std::net::TcpListener;
-// use std::net::{SocketAddr, TcpStream};
 use std::thread;
 
 use crate::tmtc::{PacketSenderRaw, PacketSource};
@@ -244,13 +242,16 @@ impl<
         // Create a poll instance.
         let poll = Poll::new()?;
         // Create storage for events.
-        let events = Events::with_capacity(10);
+        let events = Events::with_capacity(32);
         let listener: std::net::TcpListener = socket.into();
         let mut mio_listener = TcpListener::from_std(listener);
 
         // Start listening for incoming connections.
-        poll.registry()
-            .register(&mut mio_listener, Token(0), Interest::READABLE)?;
+        poll.registry().register(
+            &mut mio_listener,
+            Token(0),
+            Interest::READABLE | Interest::WRITABLE,
+        )?;
 
         Ok(Self {
             id: cfg.id,
@@ -280,11 +281,11 @@ impl<
         self.listener.local_addr()
     }
 
-    /// This call is used to handle the next connection to a client. Right now, it performs
+    /// This call is used to handle all connection from clients. Right now, it performs
     /// the following steps:
     ///
-    /// 1. It calls the [std::net::TcpListener::accept] method internally using the blocking API
-    ///    until a client connects.
+    /// 1. It calls the [std::net::TcpListener::accept] method until a client connects. An optional
+    ///    timeout can be specified for non-blocking acceptance.
     /// 2. It reads all the telecommands from the client and parses all received data using the
     ///    user specified [TcpTcParser].
     /// 3. After reading and parsing all telecommands, it sends back all telemetry using the
@@ -317,11 +318,17 @@ impl<
             loop {
                 match self.listener.accept() {
                     Ok((stream, addr)) => {
-                        self.handle_accepted_connection(stream, addr)?;
+                        if let Err(e) = self.handle_accepted_connection(stream, addr) {
+                            self.reregister_poll_interest()?;
+                            return Err(e);
+                        }
                         handled_connections += 1;
                     }
                     Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break,
-                    Err(err) => return Err(TcpTmtcError::Io(err)),
+                    Err(err) => {
+                        self.reregister_poll_interest()?;
+                        return Err(TcpTmtcError::Io(err));
+                    }
                 }
             }
         }
@@ -331,6 +338,14 @@ impl<
         Ok(ConnectionResult::AcceptTimeout)
     }
 
+    fn reregister_poll_interest(&mut self) -> io::Result<()> {
+        self.poll.registry().reregister(
+            &mut self.listener,
+            Token(0),
+            Interest::READABLE | Interest::WRITABLE,
+        )
+    }
+
     fn handle_accepted_connection(
         &mut self,
         mut stream: TcpStream,

satrs/src/hal/std/tcp_spacepackets_server.rs

@@ -26,14 +26,19 @@ impl<T: SpacePacketValidator, TmError, TcError: 'static> TcpTcParser<TmError, Tc
         next_write_idx: &mut usize,
     ) -> Result<(), TcpTmtcError<TmError, TcError>> {
         // Reader vec full, need to parse for packets.
-        conn_result.num_received_tcs += parse_buffer_for_ccsds_space_packets(
-            &mut tc_buffer[..current_write_idx],
+        let parse_result = parse_buffer_for_ccsds_space_packets(
+            &tc_buffer[..current_write_idx],
             self,
             sender_id,
             tc_sender,
-            next_write_idx,
         )
         .map_err(|e| TcpTmtcError::TcError(e))?;
+        if let Some(broken_tail_start) = parse_result.incomplete_tail_start {
+            // Copy broken tail to front of buffer.
+            tc_buffer.copy_within(broken_tail_start..current_write_idx, 0);
+            *next_write_idx = current_write_idx - broken_tail_start;
+        }
+        conn_result.num_received_tcs += parse_result.packets_found;
         Ok(())
     }
 }