diff --git a/Svc/ActiveRateGroup/ActiveRateGroup.fpp b/Svc/ActiveRateGroup/ActiveRateGroup.fpp index 3ee1488e375..c60ad6d0d1e 100644 --- a/Svc/ActiveRateGroup/ActiveRateGroup.fpp +++ b/Svc/ActiveRateGroup/ActiveRateGroup.fpp @@ -15,7 +15,7 @@ module Svc { output port RateGroupMemberOut: [ActiveRateGroupOutputPorts] Sched @ Ping input port for health - async input port PingIn: Ping + async input port PingIn: Ping drop @ Ping output port for health output port PingOut: Ping diff --git a/Svc/BufferLogger/BufferLogger.fpp b/Svc/BufferLogger/BufferLogger.fpp index 6da2ffd6930..e045b185625 100644 --- a/Svc/BufferLogger/BufferLogger.fpp +++ b/Svc/BufferLogger/BufferLogger.fpp @@ -16,12 +16,12 @@ module Svc { async input port comIn: Fw.Com @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping - async input port schedIn: Svc.Sched + async input port schedIn: Svc.Sched drop # ---------------------------------------------------------------------- # Special ports diff --git a/Svc/CmdDispatcher/CmdDispatcher.fpp b/Svc/CmdDispatcher/CmdDispatcher.fpp index 66343b886e8..2f77328966e 100644 --- a/Svc/CmdDispatcher/CmdDispatcher.fpp +++ b/Svc/CmdDispatcher/CmdDispatcher.fpp @@ -24,10 +24,10 @@ module Svc { async input port seqCmdBuff: [CmdDispatcherSequencePorts] Fw.Com hook @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Run port used to emit telemetry - async input port run: Svc.Sched + async input port run: Svc.Sched drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/CmdSequencer/CmdSequencer.fpp b/Svc/CmdSequencer/CmdSequencer.fpp index f1c4012784f..ebc38165324 100644 --- a/Svc/CmdSequencer/CmdSequencer.fpp +++ b/Svc/CmdSequencer/CmdSequencer.fpp @@ -68,7 +68,7 @@ module Svc { async input port cmdResponseIn: Fw.CmdResponse @ Ping in port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping out port output port pingOut: Svc.Ping @@ -86,7 +86,7 @@ module Svc { output port comCmdOut: Fw.Com @ Schedule in port - async input port schedIn: Svc.Sched + async input port schedIn: Svc.Sched drop @ Notifies that a sequence has started running output port seqStartOut: Svc.CmdSeqIn diff --git a/Svc/ComAggregator/ComAggregator.cpp b/Svc/ComAggregator/ComAggregator.cpp index c4edf6d8be7..2e313f99600 100644 --- a/Svc/ComAggregator/ComAggregator.cpp +++ b/Svc/ComAggregator/ComAggregator.cpp @@ -8,6 +8,12 @@ namespace Svc { +namespace { +//! Queue slots that must remain free for a timeout signal to be enqueued: the timeout itself plus one +//! in-flight 'fill' and one in-flight 'status' signal (each bounded to one message by the com protocol). +constexpr FwSizeType TIMEOUT_QUEUE_HEADROOM = 3; +} // namespace + // ---------------------------------------------------------------------- // Component construction and destruction // ---------------------------------------------------------------------- @@ -55,7 +61,16 @@ void ComAggregator ::timeout_handler(FwIndexType portNum, U32 context) { // // Behaviorally, this solution will work exactly like the naive implementation with an infinite queue depth, but // prevents queue overflow when using finite queues. - if (this->m_allow_timeout) { + // + // Even so, timeout remains the only signal source without flow control: 'fill' and 'status' are each bounded + // to one in-flight message by the com protocol, but the rate group keeps delivering ticks while this + // component's dispatch thread is stalled (downstream backpressure, thread starvation). In the FILL state + // (m_allow_timeout true) such a stall would still fill the queue with timeout signals and trip the queue-full + // assertion in the autocoded signal send. Ticks are periodic and idempotent, so additionally skip the signal + // unless the queue has headroom for it plus the in-flight flow-controlled signals; a skipped tick is simply + // retried on the next cycle. + if (this->m_allow_timeout && + (this->m_queue.getMessagesAvailable() + TIMEOUT_QUEUE_HEADROOM <= this->m_queue.getDepth())) { this->aggregationMachine_sendSignal_timeout(); } } diff --git a/Svc/ComAggregator/test/ut/ComAggregatorTestMain.cpp b/Svc/ComAggregator/test/ut/ComAggregatorTestMain.cpp index 5c118ae0087..44bd5914aca 100644 --- a/Svc/ComAggregator/test/ut/ComAggregatorTestMain.cpp +++ b/Svc/ComAggregator/test/ut/ComAggregatorTestMain.cpp @@ -63,6 +63,12 @@ TEST(OffNominal, TimeoutOverflowPrevention) { tester.test_timeout(); } +TEST(OffNominal, TimeoutOverflowPreventionFillState) { + Svc::ComAggregatorTester tester; + tester.test_initial(); + tester.test_timeout_overflow_prevention_fill_state(); +} + TEST(Nominal, HoldWhileWaiting) { Svc::ComAggregatorTester tester; tester.test_initial(); diff --git a/Svc/ComAggregator/test/ut/ComAggregatorTester.cpp b/Svc/ComAggregator/test/ut/ComAggregatorTester.cpp index 98c2055fab5..b41ee15dd82 100644 --- a/Svc/ComAggregator/test/ut/ComAggregatorTester.cpp +++ b/Svc/ComAggregator/test/ut/ComAggregatorTester.cpp @@ -266,6 +266,48 @@ void ComAggregatorTester ::test_timeout_overflow_prevention() { this->clearHistory(); } +//! Tests that a stalled dispatch thread in the FILL state cannot flood the queue with timeout +//! signals past its depth, and that the component recovers cleanly once draining resumes +void ComAggregatorTester ::test_timeout_overflow_prevention_fill_state() { + // Precondition: initial has run. The state machine is now in the FILL state + // (m_allow_timeout == true) with an empty aggregation buffer and an empty queue. Unlike + // test_timeout_overflow_prevention (which drives m_allow_timeout to false via a comStatusIn + // failure before flooding, exercising the WAIT_STATUS-adjacent gate from the pre-existing + // fix), this test floods while m_allow_timeout stays true, exercising the FILL-state gap. + ASSERT_EQ(this->component.m_queue.getMessagesAvailable(), 0); + + // Simulate a stalled dispatch thread: the rate group keeps delivering 'timeout' ticks (this + // port is invoked directly here, exactly as a rate-group member call would) while nothing + // drains the component's message queue via dispatchOne. Loop well past the queue depth -- + // without the FILL-state headroom check, this would keep enqueueing right up to depth and + // trip the autocoded queue-full assertion inside aggregationMachine_sendSignal_timeout. + for (FwSizeType i = 0; i < static_cast(TEST_INSTANCE_QUEUE_DEPTH) * 2; i++) { + this->invoke_to_timeout(0, 0); + // The queue must never be allowed to reach its full depth: headroom is reserved for the + // (at most one each) in-flight 'fill'/'status' signals, so FILL-state flooding cannot trip + // the queue-full assert no matter how long the simulated stall runs. + ASSERT_LT(this->component.m_queue.getMessagesAvailable(), static_cast(TEST_INSTANCE_QUEUE_DEPTH)); + } + // With the aggregation buffer empty, isNotEmpty is false, so every dispatched timeout is a + // no-op transition (see test_timeout_zero); flooding has no side effects of its own. + ASSERT_from_dataOut_SIZE(0); + + // Confirm recovery once the (simulated) dispatch stall clears and draining resumes: every + // queued signal is still processable without error, and the queue returns to empty. + // (dispatchOne's underlying doDispatch() issues a BLOCKING queue receive, so drain exactly the + // snapshotted number of queued messages via dispatchCurrentMessages rather than looping until + // an "empty" status that a blocking dispatch will never produce.) + ASSERT_EQ(this->dispatchCurrentMessages(this->component), + Svc::ComAggregatorComponentBase::MsgDispatchStatus::MSG_DISPATCH_OK); + ASSERT_EQ(this->component.m_queue.getMessagesAvailable(), 0); + ASSERT_from_dataOut_SIZE(0); + this->clearHistory(); + + // Normal fill/timeout operation resumes cleanly after the flood and drain. + this->test_fill_multi(); + this->test_timeout(); +} + void ComAggregatorTester ::test_timeout_zero() { // Precondition: initialize has run this->invoke_to_timeout(0, 0); diff --git a/Svc/ComAggregator/test/ut/ComAggregatorTester.hpp b/Svc/ComAggregator/test/ut/ComAggregatorTester.hpp index 1ea0a1dcc23..856a323f455 100644 --- a/Svc/ComAggregator/test/ut/ComAggregatorTester.hpp +++ b/Svc/ComAggregator/test/ut/ComAggregatorTester.hpp @@ -65,6 +65,10 @@ class ComAggregatorTester final : public ComAggregatorGTestBase { //! Tests timeout operation void test_timeout_overflow_prevention(); + //! Tests that a stalled dispatch thread in the FILL state cannot flood the queue with timeout + //! signals past its depth, and that the component recovers cleanly once draining resumes + void test_timeout_overflow_prevention_fill_state(); + //! Tests timeout operation sends no empty buffer void test_timeout_zero(); diff --git a/Svc/ComLogger/ComLogger.fpp b/Svc/ComLogger/ComLogger.fpp index c4327b7ddb7..cf4bc223871 100644 --- a/Svc/ComLogger/ComLogger.fpp +++ b/Svc/ComLogger/ComLogger.fpp @@ -11,7 +11,7 @@ module Svc { async input port comIn: Fw.Com @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/DpCatalog/DpCatalog.fpp b/Svc/DpCatalog/DpCatalog.fpp index e1fd468dd2a..59ffc363f75 100644 --- a/Svc/DpCatalog/DpCatalog.fpp +++ b/Svc/DpCatalog/DpCatalog.fpp @@ -32,7 +32,7 @@ module Svc { # Component specific ports @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/DpManager/DpManager.fpp b/Svc/DpManager/DpManager.fpp index ebbf5a46fb4..74574f97f36 100644 --- a/Svc/DpManager/DpManager.fpp +++ b/Svc/DpManager/DpManager.fpp @@ -8,7 +8,7 @@ module Svc { # ---------------------------------------------------------------------- @ Schedule in port - async input port schedIn: Svc.Sched + async input port schedIn: Svc.Sched drop # ---------------------------------------------------------------------- # Ports for handling buffer requests diff --git a/Svc/DpWriter/DpWriter.fpp b/Svc/DpWriter/DpWriter.fpp index f24594f0925..152be91baf4 100644 --- a/Svc/DpWriter/DpWriter.fpp +++ b/Svc/DpWriter/DpWriter.fpp @@ -8,7 +8,7 @@ module Svc { # ---------------------------------------------------------------------- @ Schedule in port - async input port schedIn: Svc.Sched + async input port schedIn: Svc.Sched drop # ---------------------------------------------------------------------- # Ports for handling data products diff --git a/Svc/EventManager/EventManager.fpp b/Svc/EventManager/EventManager.fpp index b97a8d807e3..2ea88960ae9 100644 --- a/Svc/EventManager/EventManager.fpp +++ b/Svc/EventManager/EventManager.fpp @@ -53,7 +53,7 @@ module Svc { output port FatalAnnounce: Svc.FatalEvent @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/FileDownlink/FileDownlink.fpp b/Svc/FileDownlink/FileDownlink.fpp index eed9376aa41..5e710a3f41f 100644 --- a/Svc/FileDownlink/FileDownlink.fpp +++ b/Svc/FileDownlink/FileDownlink.fpp @@ -8,7 +8,7 @@ module Svc { # ---------------------------------------------------------------------- @ Run input port - async input port Run: Svc.Sched + async input port Run: Svc.Sched drop @ Mutexed Sendfile input port guarded input port SendFile: Svc.SendFileRequest @@ -23,7 +23,7 @@ module Svc { output port bufferSendOut: Fw.BufferSend @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/FileManager/FileManager.fpp b/Svc/FileManager/FileManager.fpp index bef777c7133..d85f42002f7 100644 --- a/Svc/FileManager/FileManager.fpp +++ b/Svc/FileManager/FileManager.fpp @@ -8,7 +8,7 @@ module Svc { # ---------------------------------------------------------------------- @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Scheduler input port for rate group operations sync input port schedIn: Sched diff --git a/Svc/FileUplink/FileUplink.fpp b/Svc/FileUplink/FileUplink.fpp index 90ef4e9a821..b2b67bd5c39 100644 --- a/Svc/FileUplink/FileUplink.fpp +++ b/Svc/FileUplink/FileUplink.fpp @@ -14,7 +14,7 @@ module Svc { output port bufferSendOut: Fw.BufferSend @ Ping in - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping out output port pingOut: Svc.Ping diff --git a/Svc/FpySequencer/FpySequencer.fpp b/Svc/FpySequencer/FpySequencer.fpp index 8af40d3d7ce..b9e80e1af03 100644 --- a/Svc/FpySequencer/FpySequencer.fpp +++ b/Svc/FpySequencer/FpySequencer.fpp @@ -38,7 +38,7 @@ module Svc { @ Ping in port # TODO should ping have highest prio? or lowest? - async input port pingIn: Svc.Ping priority 10 assert + async input port pingIn: Svc.Ping priority 10 drop @ port to trigger a wakeup or timeout check. increase frequency @ to increase temporal resolution of sequencer diff --git a/Svc/PrmDb/PrmDb.fpp b/Svc/PrmDb/PrmDb.fpp index bca4f172faa..b2a4b95b1e1 100644 --- a/Svc/PrmDb/PrmDb.fpp +++ b/Svc/PrmDb/PrmDb.fpp @@ -69,7 +69,7 @@ module Svc { async input port setPrm: Fw.PrmSet @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/TlmChan/TlmChan.fpp b/Svc/TlmChan/TlmChan.fpp index d15072ed08a..52b73b2d818 100644 --- a/Svc/TlmChan/TlmChan.fpp +++ b/Svc/TlmChan/TlmChan.fpp @@ -10,13 +10,13 @@ module Svc { guarded input port TlmGet: Fw.TlmGet @ Run port for starting packet send cycle - async input port Run: Svc.Sched + async input port Run: Svc.Sched drop @ Packet send port output port PktSend: Fw.Com @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping diff --git a/Svc/TlmPacketizer/TlmPacketizer.cpp b/Svc/TlmPacketizer/TlmPacketizer.cpp index 89af22e1d01..ba1c685558c 100644 --- a/Svc/TlmPacketizer/TlmPacketizer.cpp +++ b/Svc/TlmPacketizer/TlmPacketizer.cpp @@ -89,10 +89,10 @@ void TlmPacketizer::setPacketList(const TlmPacketizerPacketList& packetList, entry.ignored = false; entry.channelSize = packetList.list[pktEntry]->list[tlmEntry].size; // the offset into the buffer will be the current packet length - // the offset must fit within FwSignedSizeType to allow for negative values - FW_ASSERT(packetLen <= static_cast(std::numeric_limits::max()), + // the offset must fit within I16 to allow for the -1 sentinel value + FW_ASSERT(packetLen <= static_cast(std::numeric_limits::max()), static_cast(packetLen)); - entry.packetOffset[pktEntry] = static_cast(packetLen); + entry.packetOffset[pktEntry] = static_cast(packetLen); packetLen += entry.channelSize; diff --git a/Svc/TlmPacketizer/TlmPacketizer.fpp b/Svc/TlmPacketizer/TlmPacketizer.fpp index 1655c3e2eb7..4125fdac9ca 100644 --- a/Svc/TlmPacketizer/TlmPacketizer.fpp +++ b/Svc/TlmPacketizer/TlmPacketizer.fpp @@ -28,13 +28,13 @@ module Svc { async input port controlIn: EnableSection @ Ping input port - async input port pingIn: Svc.Ping + async input port pingIn: Svc.Ping drop @ Ping output port output port pingOut: Svc.Ping @ Run port for starting packet send cycle - async input port Run: Svc.Sched + async input port Run: Svc.Sched drop @ Input configuration port async input port configureSectionGroupRate: ConfigureGroupRate diff --git a/Svc/TlmPacketizer/TlmPacketizer.hpp b/Svc/TlmPacketizer/TlmPacketizer.hpp index 7f80d76d265..ffb58fc5d17 100644 --- a/Svc/TlmPacketizer/TlmPacketizer.hpp +++ b/Svc/TlmPacketizer/TlmPacketizer.hpp @@ -172,7 +172,7 @@ class TlmPacketizer final : public TlmPacketizerComponentBase, public Fw::ParamE FwChanIdType id; //!< telemetry id stored in slot // Offsets into packet buffers. // -1 means that channel is not in that packet - FwSignedSizeType packetOffset[MAX_PACKETIZER_PACKETS]; + I16 packetOffset[MAX_PACKETIZER_PACKETS]; FwSizeType channelSize; //!< max serialized size of the channel in bytes bool ignored; //!< ignored channel id bool hasValue; //!< if the entry has received a value at least once