Finish PUS service optimizations

- Better naming for pool abstractions - Added last unittests for PUS helper services - Introduce new abstraction for PUS schedulers - `StoreAddr` is now a generic u64 - `spacepackets` points to 0.7.0 release
2024-02-03 13:41:51 +01:00
parent 6152c834d4
commit a891b947c7
22 changed files with 1669 additions and 874 deletions
--- a/satrs-example/src/main.rs
+++ b/satrs-example/src/main.rs
@ -29,7 +29,7 @@ use satrs_core::event_man::{
 };
 use satrs_core::events::EventU32;
 use satrs_core::hk::HkRequest;
-use satrs_core::pool::{LocalPool, PoolCfg};
+use satrs_core::pool::{PoolProviderMemInPlace, StaticMemoryPool, StaticPoolConfig};
 use satrs_core::pus::event_man::{
    DefaultPusMgmtBackendProvider, EventReporter, EventRequest, EventRequestWithToken,
    PusEventDispatcher,
@ -42,7 +42,9 @@ use satrs_core::pus::test::PusService17TestHandler;
 use satrs_core::pus::verification::{
    TcStateStarted, VerificationReporterCfg, VerificationReporterWithSender, VerificationToken,
 };
-use satrs_core::pus::{EcssTcInStoreConverter, MpscTcInStoreReceiver, MpscTmInStoreSender};
+use satrs_core::pus::{
+    EcssTcInSharedStoreConverter, MpscTcReceiver, MpscTmInStoreSender, PusServiceHelper,
+};
 use satrs_core::seq_count::{CcsdsSimpleSeqCountProvider, SequenceCountProviderCore};
 use satrs_core::spacepackets::ecss::tm::{PusTmCreator, PusTmZeroCopyWriter};
 use satrs_core::spacepackets::{
@ -66,7 +68,7 @@ use std::time::Duration;
 fn main() {
    setup_logger().expect("setting up logging with fern failed");
    println!("Running OBSW example");
-    let tm_pool = LocalPool::new(PoolCfg::new(vec![
+    let tm_pool = StaticMemoryPool::new(StaticPoolConfig::new(vec![
        (30, 32),
        (15, 64),
        (15, 128),
@ -74,9 +76,9 @@ fn main() {
        (15, 1024),
        (15, 2048),
    ]));
-    let shared_tm_store = SharedTmStore::new(Box::new(tm_pool));
+    let shared_tm_store = SharedTmStore::new(tm_pool);
    let tm_store_event = shared_tm_store.clone();
-    let tc_pool = LocalPool::new(PoolCfg::new(vec![
+    let tc_pool = StaticMemoryPool::new(StaticPoolConfig::new(vec![
        (30, 32),
        (15, 64),
        (15, 128),
@ -85,8 +87,16 @@ fn main() {
        (15, 2048),
    ]));
    let tc_store = TcStore {
-        pool: Arc::new(RwLock::new(Box::new(tc_pool))),
+        pool: Arc::new(RwLock::new(tc_pool)),
    };
+    let sched_tc_pool = StaticMemoryPool::new(StaticPoolConfig::new(vec![
+        (30, 32),
+        (15, 64),
+        (15, 128),
+        (15, 256),
+        (15, 1024),
+        (15, 2048),
+    ]));

    let seq_count_provider = CcsdsSimpleSeqCountProvider::new();
    let mut msg_counter_map: HashMap<u8, u16> = HashMap::new();
@ -172,18 +182,18 @@ fn main() {
        shared_tm_store.clone(),
        tm_funnel_tx.clone(),
    );
-    let test_srv_receiver = MpscTcInStoreReceiver::new(
+    let test_srv_receiver = MpscTcReceiver::new(
        TcReceiverId::PusTest as ChannelId,
        "PUS_17_TC_RECV",
        pus_test_rx,
    );
-    let pus17_handler = PusService17TestHandler::new(
+    let pus17_handler = PusService17TestHandler::new(PusServiceHelper::new(
        Box::new(test_srv_receiver),
        Box::new(test_srv_tm_sender),
        PUS_APID,
        verif_reporter.clone(),
-        EcssTcInStoreConverter::new(tc_store.pool.clone(), 2048),
-    );
+        EcssTcInSharedStoreConverter::new(tc_store.pool.clone(), 2048),
+    ));
    let mut pus_17_wrapper = Service17CustomWrapper {
        pus17_handler,
        test_srv_event_sender,
@ -195,7 +205,7 @@ fn main() {
        shared_tm_store.clone(),
        tm_funnel_tx.clone(),
    );
-    let sched_srv_receiver = MpscTcInStoreReceiver::new(
+    let sched_srv_receiver = MpscTcReceiver::new(
        TcReceiverId::PusSched as ChannelId,
        "PUS_11_TC_RECV",
        pus_sched_rx,
@ -203,16 +213,18 @@ fn main() {
    let scheduler = PusScheduler::new_with_current_init_time(Duration::from_secs(5))
        .expect("Creating PUS Scheduler failed");
    let pus_11_handler = PusService11SchedHandler::new(
-        Box::new(sched_srv_receiver),
-        Box::new(sched_srv_tm_sender),
-        PUS_APID,
-        verif_reporter.clone(),
-        EcssTcInStoreConverter::new(tc_store.pool.clone(), 2048),
-        tc_store.pool.clone(),
+        PusServiceHelper::new(
+            Box::new(sched_srv_receiver),
+            Box::new(sched_srv_tm_sender),
+            PUS_APID,
+            verif_reporter.clone(),
+            EcssTcInSharedStoreConverter::new(tc_store.pool.clone(), 2048),
+        ),
        scheduler,
    );
    let mut pus_11_wrapper = Pus11Wrapper {
        pus_11_handler,
+        sched_tc_pool,
        tc_source_wrapper,
    };

@ -222,17 +234,19 @@ fn main() {
        shared_tm_store.clone(),
        tm_funnel_tx.clone(),
    );
-    let event_srv_receiver = MpscTcInStoreReceiver::new(
+    let event_srv_receiver = MpscTcReceiver::new(
        TcReceiverId::PusEvent as ChannelId,
        "PUS_5_TC_RECV",
        pus_event_rx,
    );
    let pus_5_handler = PusService5EventHandler::new(
-        Box::new(event_srv_receiver),
-        Box::new(event_srv_tm_sender),
-        PUS_APID,
-        verif_reporter.clone(),
-        EcssTcInStoreConverter::new(tc_store.pool.clone(), 2048),
+        PusServiceHelper::new(
+            Box::new(event_srv_receiver),
+            Box::new(event_srv_tm_sender),
+            PUS_APID,
+            verif_reporter.clone(),
+            EcssTcInSharedStoreConverter::new(tc_store.pool.clone(), 2048),
+        ),
        event_request_tx,
    );
    let mut pus_5_wrapper = Pus5Wrapper { pus_5_handler };
@ -243,7 +257,7 @@ fn main() {
        shared_tm_store.clone(),
        tm_funnel_tx.clone(),
    );
-    let action_srv_receiver = MpscTcInStoreReceiver::new(
+    let action_srv_receiver = MpscTcReceiver::new(
        TcReceiverId::PusAction as ChannelId,
        "PUS_8_TC_RECV",
        pus_action_rx,
@ -253,7 +267,7 @@ fn main() {
        Box::new(action_srv_tm_sender),
        PUS_APID,
        verif_reporter.clone(),
-        EcssTcInStoreConverter::new(tc_store.pool.clone(), 2048),
+        EcssTcInSharedStoreConverter::new(tc_store.pool.clone(), 2048),
        request_map.clone(),
    );
    let mut pus_8_wrapper = Pus8Wrapper { pus_8_handler };
@ -265,13 +279,13 @@ fn main() {
        tm_funnel_tx.clone(),
    );
    let hk_srv_receiver =
-        MpscTcInStoreReceiver::new(TcReceiverId::PusHk as ChannelId, "PUS_8_TC_RECV", pus_hk_rx);
+        MpscTcReceiver::new(TcReceiverId::PusHk as ChannelId, "PUS_8_TC_RECV", pus_hk_rx);
    let pus_3_handler = PusService3HkHandler::new(
        Box::new(hk_srv_receiver),
        Box::new(hk_srv_tm_sender),
        PUS_APID,
        verif_reporter.clone(),
-        EcssTcInStoreConverter::new(tc_store.pool.clone(), 2048),
+        EcssTcInSharedStoreConverter::new(tc_store.pool.clone(), 2048),
        request_map,
    );
    let mut pus_3_wrapper = Pus3Wrapper { pus_3_handler };
--- a/satrs-example/src/pus/action.rs
+++ b/satrs-example/src/pus/action.rs
@ -4,8 +4,8 @@ use satrs_core::pus::verification::{
    FailParams, TcStateAccepted, VerificationReporterWithSender, VerificationToken,
 };
 use satrs_core::pus::{
-    EcssTcInMemConverter, EcssTcInStoreConverter, EcssTcReceiver, EcssTmSender,
-    PusPacketHandlerResult, PusPacketHandlingError, PusServiceBase, PusServiceHandler,
+    EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcReceiver, EcssTmSender,
+    PusPacketHandlerResult, PusPacketHandlingError, PusServiceBase, PusServiceHelper,
 };
 use satrs_core::spacepackets::ecss::tc::PusTcReader;
 use satrs_core::spacepackets::ecss::PusPacket;
@ -14,7 +14,7 @@ use std::collections::HashMap;
 use std::sync::mpsc::Sender;

 pub struct PusService8ActionHandler<TcInMemConverter: EcssTcInMemConverter> {
-    psb: PusServiceHandler<TcInMemConverter>,
+    service_helper: PusServiceHelper<TcInMemConverter>,
    request_handlers: HashMap<TargetIdWithApid, Sender<RequestWithToken>>,
 }

@ -28,7 +28,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService8ActionHandler<TcInMemCon
        request_handlers: HashMap<TargetIdWithApid, Sender<RequestWithToken>>,
    ) -> Self {
        Self {
-            psb: PusServiceHandler::new(
+            service_helper: PusServiceHelper::new(
                tc_receiver,
                tm_sender,
                tm_apid,
@ -47,7 +47,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService8ActionHandler<TcInMemCon
    ) -> Result<(), PusPacketHandlingError> {
        let user_data = tc.user_data();
        if user_data.len() < 8 {
-            self.psb
+            self.service_helper
                .common
                .verification_handler
                .borrow_mut()
@ -77,7 +77,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService8ActionHandler<TcInMemCon
        } else {
            let mut fail_data: [u8; 4] = [0; 4];
            fail_data.copy_from_slice(&target_id.target.to_be_bytes());
-            self.psb
+            self.service_helper
                .common
                .verification_handler
                .borrow_mut()
@ -98,15 +98,15 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService8ActionHandler<TcInMemCon
    }

    fn handle_one_tc(&mut self) -> Result<PusPacketHandlerResult, PusPacketHandlingError> {
-        let possible_packet = self.psb.retrieve_and_accept_next_packet()?;
+        let possible_packet = self.service_helper.retrieve_and_accept_next_packet()?;
        if possible_packet.is_none() {
            return Ok(PusPacketHandlerResult::Empty);
        }
        let ecss_tc_and_token = possible_packet.unwrap();
-        self.psb
+        self.service_helper
            .tc_in_mem_converter
-            .cache_ecss_tc_in_memory(&ecss_tc_and_token)?;
-        let tc = PusTcReader::new(self.psb.tc_in_mem_converter.tc_slice_raw())?.0;
+            .cache_ecss_tc_in_memory(&ecss_tc_and_token.tc_in_memory)?;
+        let tc = PusTcReader::new(self.service_helper.tc_in_mem_converter.tc_slice_raw())?.0;
        let subservice = tc.subservice();
        let mut partial_error = None;
        let time_stamp = PusServiceBase::get_current_timestamp(&mut partial_error);
@ -116,7 +116,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService8ActionHandler<TcInMemCon
            }
            _ => {
                let fail_data = [subservice];
-                self.psb
+                self.service_helper
                    .common
                    .verification_handler
                    .get_mut()
@ -142,7 +142,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService8ActionHandler<TcInMemCon
 }

 pub struct Pus8Wrapper {
-    pub(crate) pus_8_handler: PusService8ActionHandler<EcssTcInStoreConverter>,
+    pub(crate) pus_8_handler: PusService8ActionHandler<EcssTcInSharedStoreConverter>,
 }

 impl Pus8Wrapper {
--- a/satrs-example/src/pus/event.rs
+++ b/satrs-example/src/pus/event.rs
@ -1,9 +1,9 @@
 use log::{error, warn};
 use satrs_core::pus::event_srv::PusService5EventHandler;
-use satrs_core::pus::{EcssTcInStoreConverter, PusPacketHandlerResult};
+use satrs_core::pus::{EcssTcInSharedStoreConverter, PusPacketHandlerResult};

 pub struct Pus5Wrapper {
-    pub pus_5_handler: PusService5EventHandler<EcssTcInStoreConverter>,
+    pub pus_5_handler: PusService5EventHandler<EcssTcInSharedStoreConverter>,
 }

 impl Pus5Wrapper {
--- a/satrs-example/src/pus/hk.rs
+++ b/satrs-example/src/pus/hk.rs
@ -3,8 +3,8 @@ use log::{error, warn};
 use satrs_core::hk::{CollectionIntervalFactor, HkRequest};
 use satrs_core::pus::verification::{FailParams, StdVerifReporterWithSender};
 use satrs_core::pus::{
-    EcssTcInMemConverter, EcssTcInStoreConverter, EcssTcReceiver, EcssTmSender,
-    PusPacketHandlerResult, PusPacketHandlingError, PusServiceBase, PusServiceHandler,
+    EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcReceiver, EcssTmSender,
+    PusPacketHandlerResult, PusPacketHandlingError, PusServiceBase, PusServiceHelper,
 };
 use satrs_core::spacepackets::ecss::{hk, PusPacket};
 use satrs_example::{hk_err, tmtc_err, TargetIdWithApid};
@ -12,7 +12,7 @@ use std::collections::HashMap;
 use std::sync::mpsc::Sender;

 pub struct PusService3HkHandler<TcInMemConverter: EcssTcInMemConverter> {
-    psb: PusServiceHandler<TcInMemConverter>,
+    psb: PusServiceHelper<TcInMemConverter>,
    request_handlers: HashMap<TargetIdWithApid, Sender<RequestWithToken>>,
 }

@ -26,7 +26,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService3HkHandler<TcInMemConvert
        request_handlers: HashMap<TargetIdWithApid, Sender<RequestWithToken>>,
    ) -> Self {
        Self {
-            psb: PusServiceHandler::new(
+            psb: PusServiceHelper::new(
                tc_receiver,
                tm_sender,
                tm_apid,
@ -46,7 +46,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService3HkHandler<TcInMemConvert
        let tc = self
            .psb
            .tc_in_mem_converter
-            .convert_ecss_tc_in_memory_to_reader(&ecss_tc_and_token)?;
+            .convert_ecss_tc_in_memory_to_reader(&ecss_tc_and_token.tc_in_memory)?;
        let subservice = tc.subservice();
        let mut partial_error = None;
        let time_stamp = PusServiceBase::get_current_timestamp(&mut partial_error);
@ -148,7 +148,7 @@ impl<TcInMemConverter: EcssTcInMemConverter> PusService3HkHandler<TcInMemConvert
 }

 pub struct Pus3Wrapper {
-    pub(crate) pus_3_handler: PusService3HkHandler<EcssTcInStoreConverter>,
+    pub(crate) pus_3_handler: PusService3HkHandler<EcssTcInSharedStoreConverter>,
 }

 impl Pus3Wrapper {
--- a/satrs-example/src/pus/scheduler.rs
+++ b/satrs-example/src/pus/scheduler.rs
@ -1,50 +1,54 @@
 use crate::tmtc::PusTcSource;
 use log::{error, info, warn};
-use satrs_core::pus::scheduler::TcInfo;
+use satrs_core::pool::{PoolProviderMemInPlace, StaticMemoryPool};
+use satrs_core::pus::scheduler::{PusScheduler, TcInfo};
 use satrs_core::pus::scheduler_srv::PusService11SchedHandler;
-use satrs_core::pus::{EcssTcInStoreConverter, PusPacketHandlerResult};
+use satrs_core::pus::{EcssTcInSharedStoreConverter, PusPacketHandlerResult};

 pub struct Pus11Wrapper {
-    pub pus_11_handler: PusService11SchedHandler<EcssTcInStoreConverter>,
+    pub pus_11_handler: PusService11SchedHandler<EcssTcInSharedStoreConverter, PusScheduler>,
+    pub sched_tc_pool: StaticMemoryPool,
    pub tc_source_wrapper: PusTcSource,
 }

 impl Pus11Wrapper {
    pub fn release_tcs(&mut self) {
-        let releaser = |enabled: bool, info: &TcInfo| -> bool {
+        let releaser = |enabled: bool, _info: &TcInfo, tc: &[u8]| -> bool {
            if enabled {
+                // Transfer TC from scheduler TC pool to shared TC pool.
+                let released_tc_addr = self
+                    .tc_source_wrapper
+                    .tc_store
+                    .pool
+                    .write()
+                    .expect("locking pool failed")
+                    .add(tc)
+                    .expect("adding TC to shared pool failed");
+
                self.tc_source_wrapper
                    .tc_source
-                    .send(info.addr())
+                    .send(released_tc_addr)
                    .expect("sending TC to TC source failed");
            }
            true
        };

-        let mut pool = self
-            .tc_source_wrapper
-            .tc_store
-            .pool
-            .write()
-            .expect("error locking pool");
-
        self.pus_11_handler
            .scheduler_mut()
            .update_time_from_now()
            .unwrap();
-        if let Ok(released_tcs) = self
+        let released_tcs = self
            .pus_11_handler
            .scheduler_mut()
-            .release_telecommands(releaser, pool.as_mut())
-        {
-            if released_tcs > 0 {
-                info!("{released_tcs} TC(s) released from scheduler");
-            }
+            .release_telecommands(releaser, &mut self.sched_tc_pool)
+            .expect("releasing TCs failed");
+        if released_tcs > 0 {
+            info!("{released_tcs} TC(s) released from scheduler");
        }
    }

    pub fn handle_next_packet(&mut self) -> bool {
-        match self.pus_11_handler.handle_one_tc() {
+        match self.pus_11_handler.handle_one_tc(&mut self.sched_tc_pool) {
            Ok(result) => match result {
                PusPacketHandlerResult::RequestHandled => {}
                PusPacketHandlerResult::RequestHandledPartialSuccess(e) => {
--- a/satrs-example/src/pus/test.rs
+++ b/satrs-example/src/pus/test.rs
@ -7,12 +7,12 @@ use satrs_core::spacepackets::ecss::tc::PusTcReader;
 use satrs_core::spacepackets::ecss::PusPacket;
 use satrs_core::spacepackets::time::cds::TimeProvider;
 use satrs_core::spacepackets::time::TimeWriter;
-use satrs_core::{events::EventU32, pus::EcssTcInStoreConverter};
+use satrs_core::{events::EventU32, pus::EcssTcInSharedStoreConverter};
 use satrs_example::{tmtc_err, TEST_EVENT};
 use std::sync::mpsc::Sender;

 pub struct Service17CustomWrapper {
-    pub pus17_handler: PusService17TestHandler<EcssTcInStoreConverter>,
+    pub pus17_handler: PusService17TestHandler<EcssTcInSharedStoreConverter>,
    pub test_srv_event_sender: Sender<(EventU32, Option<Params>)>,
 }

@ -38,9 +38,13 @@ impl Service17CustomWrapper {
                warn!("PUS17: Subservice {subservice} not implemented")
            }
            PusPacketHandlerResult::CustomSubservice(subservice, token) => {
-                let (tc, _) =
-                    PusTcReader::new(self.pus17_handler.psb.tc_in_mem_converter.tc_slice_raw())
-                        .unwrap();
+                let (tc, _) = PusTcReader::new(
+                    self.pus17_handler
+                        .service_helper
+                        .tc_in_mem_converter
+                        .tc_slice_raw(),
+                )
+                .unwrap();
                let time_stamper = TimeProvider::from_now_with_u16_days().unwrap();
                let mut stamp_buf: [u8; 7] = [0; 7];
                time_stamper.write_to_bytes(&mut stamp_buf).unwrap();
@ -51,14 +55,14 @@ impl Service17CustomWrapper {
                        .expect("Sending test event failed");
                    let start_token = self
                        .pus17_handler
-                        .psb
+                        .service_helper
                        .common
                        .verification_handler
                        .get_mut()
                        .start_success(token, Some(&stamp_buf))
                        .expect("Error sending start success");
                    self.pus17_handler
-                        .psb
+                        .service_helper
                        .common
                        .verification_handler
                        .get_mut()
@ -67,7 +71,7 @@ impl Service17CustomWrapper {
                } else {
                    let fail_data = [tc.subservice()];
                    self.pus17_handler
-                        .psb
+                        .service_helper
                        .common
                        .verification_handler
                        .get_mut()
--- a/satrs-example/src/tmtc.rs
+++ b/satrs-example/src/tmtc.rs
@ -5,7 +5,7 @@ use std::sync::mpsc::{Receiver, SendError, Sender, TryRecvError};
 use thiserror::Error;

 use crate::pus::PusReceiver;
-use satrs_core::pool::{SharedPool, StoreAddr, StoreError};
+use satrs_core::pool::{PoolProviderMemInPlace, SharedStaticMemoryPool, StoreAddr, StoreError};
 use satrs_core::spacepackets::ecss::tc::PusTcReader;
 use satrs_core::spacepackets::ecss::PusPacket;
 use satrs_core::tmtc::tm_helper::SharedTmStore;
@ -41,7 +41,7 @@ pub enum MpscStoreAndSendError {

 #[derive(Clone)]
 pub struct TcStore {
-    pub pool: SharedPool,
+    pub pool: SharedStaticMemoryPool,
 }

 impl TcStore {
--- a/satrs-example/src/udp.rs
+++ b/satrs-example/src/udp.rs
@ -3,7 +3,7 @@ use std::{net::SocketAddr, sync::mpsc::Receiver};
 use log::{info, warn};
 use satrs_core::{
    hal::std::udp_server::{ReceiveResult, UdpTcServer},
-    pool::{SharedPool, StoreAddr},
+    pool::{PoolProviderMemInPlaceWithGuards, SharedStaticMemoryPool, StoreAddr},
    tmtc::CcsdsError,
 };

@ -12,7 +12,7 @@ use crate::tmtc::MpscStoreAndSendError;
 pub struct UdpTmtcServer {
    pub udp_tc_server: UdpTcServer<CcsdsError<MpscStoreAndSendError>>,
    pub tm_rx: Receiver<StoreAddr>,
-    pub tm_store: SharedPool,
+    pub tm_store: SharedStaticMemoryPool,
 }
 impl UdpTmtcServer {
    pub fn periodic_operation(&mut self) {