Add improvements to ComQueue (nasa#4190)

* Move dequeued ComBuffer to persistent memory

* Add ordering notes to com interface docs

* Add buffer ownership tracking

* Assert dequeue status for serialized buffer

Added status check for dequeue operation to ensure successful serialization.

* Fix static_assert -> static_cast

* Fix format

Author: LeStarch

Svc/ComQueue/ComQueue.cpp

@@ -31,6 +31,7 @@ ComQueue ::QueueConfigurationTable ::QueueConfigurationTable() {
 ComQueue ::ComQueue(const char* const compName)
     : ComQueueComponentBase(compName),
       m_state(WAITING),
+      m_buffer_state(OWNED),
       m_allocationId(static_cast<FwEnumStoreType>(-1)),
       m_allocator(nullptr),
       m_allocation(nullptr) {
@@ -191,6 +192,8 @@ void ComQueue::run_handler(const FwIndexType portNum, U32 context) {
 
 void ComQueue ::dataReturnIn_handler(FwIndexType portNum, Fw::Buffer& data, const ComCfg::FrameContext& context) {
     static_assert(std::numeric_limits<FwIndexType>::is_signed, "FwIndexType must be signed");
+    FW_ASSERT(this->m_buffer_state == UNOWNED);
+    this->m_buffer_state = OWNED;
     // For the buffer queues, the index of the queue is portNum offset by COM_PORT_COUNT since
     // the first COM_PORT_COUNT queues are for ComBuffer. So we have for buffer queues:
     // queueNum = portNum + COM_PORT_COUNT
@@ -259,7 +262,8 @@ void ComQueue::sendComBuffer(Fw::ComBuffer& comBuffer, FwIndexType queueIndex) {
     FW_ASSERT(status == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(status));
     context.set_apid(static_cast<ComCfg::Apid::T>(descriptor));
     context.set_comQueueIndex(queueIndex);
-
+    FW_ASSERT(this->m_buffer_state == OWNED);
+    this->m_buffer_state = UNOWNED;
     this->dataOut_out(0, outBuffer, context);
     // Set state to WAITING for the status to come back
     this->m_state = WAITING;
@@ -276,7 +280,8 @@ void ComQueue::sendBuffer(Fw::Buffer& buffer, FwIndexType queueIndex) {
     FW_ASSERT(status == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(status));
     context.set_apid(static_cast<ComCfg::Apid::T>(descriptor));
     context.set_comQueueIndex(queueIndex);
-
+    FW_ASSERT(this->m_buffer_state == OWNED);
+    this->m_buffer_state = UNOWNED;
     this->dataOut_out(0, buffer, context);
     // Set state to WAITING for the status to come back
     this->m_state = WAITING;
@@ -301,12 +306,19 @@ void ComQueue::processQueue() {
 
         // Send out the message based on the type
         if (entry.index < COM_PORT_COUNT) {
-            Fw::ComBuffer comBuffer;
-            queue.dequeue(reinterpret_cast<U8*>(&comBuffer), sizeof(comBuffer));
-            this->sendComBuffer(comBuffer, entry.index);
+            // Dequeue is reading the whole persisted Fw::ComBuffer object from the queue's storage.
+            // thus it takes an address to the object to fill and the size of the actual object.
+            FW_ASSERT(this->m_buffer_state == OWNED);
+            auto dequeue_status =
+                queue.dequeue(reinterpret_cast<U8*>(&this->m_dequeued_com_buffer), sizeof(this->m_dequeued_com_buffer));
+            FW_ASSERT(dequeue_status == Fw::SerializeStatus::FW_SERIALIZE_OK,
+                      static_cast<FwAssertArgType>(dequeue_status));
+            this->sendComBuffer(this->m_dequeued_com_buffer, entry.index);
         } else {
             Fw::Buffer buffer;
-            queue.dequeue(reinterpret_cast<U8*>(&buffer), sizeof(buffer));
+            auto dequeue_status = queue.dequeue(reinterpret_cast<U8*>(&buffer), sizeof(buffer));
+            FW_ASSERT(dequeue_status == Fw::SerializeStatus::FW_SERIALIZE_OK,
+                      static_cast<FwAssertArgType>(dequeue_status));
             this->sendBuffer(buffer, entry.index);
         }
 

Svc/ComQueue/ComQueue.hpp

@@ -22,6 +22,9 @@ namespace Svc {
 // ----------------------------------------------------------------------
 
 class ComQueue final : public ComQueueComponentBase {
+    //! State of the currently transmitted buffer
+    enum BufferState { OWNED, UNOWNED };
+
   public:
     //!< Count of Fw::Com input ports and thus Fw::Com queues
     static const FwIndexType COM_PORT_COUNT = ComQueueComponentBase::NUM_COMPACKETQUEUEIN_INPUT_PORTS;
@@ -197,10 +200,12 @@ class ComQueue final : public ComQueueComponentBase {
     // ----------------------------------------------------------------------
     // Member variables
     // ----------------------------------------------------------------------
+    Fw::ComBuffer m_dequeued_com_buffer;                //!< Store a dequeued com buffer so it does not leave scope
     Types::Queue m_queues[TOTAL_PORT_COUNT];            //!< Stores queued data waiting for transmission
     QueueMetadata m_prioritizedList[TOTAL_PORT_COUNT];  //!< Priority sorted list of queue metadata
     bool m_throttle[TOTAL_PORT_COUNT];                  //!< Per-queue EVR throttles
     SendState m_state;                                  //!< State of the component
+    BufferState m_buffer_state;                         //!< Ownership state of buffer
 
     // Storage for Fw::MemAllocator properties
     FwEnumStoreType m_allocationId;  //!< Component's allocation ID

Svc/ComQueue/test/ut/ComQueueTestMain.cpp

@@ -39,11 +39,6 @@ TEST(Nominal, ContextData) {
     tester.testContextData();
 }
 
-TEST(Nominal, testBufferQueueReturn) {
-    Svc::ComQueueTester tester;
-    tester.testBufferQueueReturn();
-}
-
 int main(int argc, char** argv) {
     ::testing::InitGoogleTest(&argc, argv);
     return RUN_ALL_TESTS();

Svc/ComQueue/test/ut/ComQueueTester.cpp

@@ -92,7 +92,9 @@ void ComQueueTester ::testQueueSend() {
     for (FwIndexType portNum = 0; portNum < ComQueue::BUFFER_PORT_COUNT; portNum++) {
         invoke_to_bufferQueueIn(portNum, buffer);
         emitOneAndCheck(portNum, buffer.getData(), buffer.getSize());
+        ASSERT_from_bufferReturnOut(portNum, buffer);
     }
+    ASSERT_from_bufferReturnOut_SIZE(ComQueue::BUFFER_PORT_COUNT);
     clearFromPortHistory();
     component.cleanup();
 }
@@ -122,7 +124,9 @@ void ComQueueTester ::testQueuePause() {
         invoke_to_comStatusIn(0, state);
         invoke_to_comStatusIn(0, state);
         emitOneAndCheck(portNum, buffer.getData(), buffer.getSize());
+        ASSERT_from_bufferReturnOut(portNum, buffer);
     }
+    ASSERT_from_bufferReturnOut_SIZE(ComQueue::BUFFER_PORT_COUNT);
     clearFromPortHistory();
     component.cleanup();
 }
@@ -226,9 +230,9 @@ void ComQueueTester::testExternalQueueOverflow() {
         dispatchAll();
 
         if (QueueType::BUFFER_QUEUE == overflow_type) {
-            // Third message overflowed, so third bufferReturnOut
-            ASSERT_from_bufferReturnOut_SIZE(3);
-            ASSERT_from_bufferReturnOut(2, buffer);
+            // Third message overflowed, emitOne yielded a return, so fourth bufferReturnOut
+            ASSERT_from_bufferReturnOut_SIZE(4);
+            ASSERT_from_bufferReturnOut(3, buffer);
         }
 
         // emitOne() reset the throttle, then overflow again. So expect a second overflow event
@@ -344,26 +348,9 @@ void ComQueueTester ::testContextData() {
     component.cleanup();
 }
 
-void ComQueueTester ::testBufferQueueReturn() {
-    U8 data[BUFFER_LENGTH] = BUFFER_DATA;
-    Fw::Buffer buffer(&data[0], sizeof(data));
-    ComCfg::FrameContext context;
-    configure();
-
-    for (FwIndexType portNum = 0; portNum < ComQueue::TOTAL_PORT_COUNT; portNum++) {
-        clearFromPortHistory();
-        context.set_comQueueIndex(portNum);
-        invoke_to_dataReturnIn(0, buffer, context);
-        // APIDs that correspond to an buffer originating from a Fw.Com port
-        // do no get deallocated – APIDs that correspond to a Fw.Buffer do
-        if (portNum < ComQueue::COM_PORT_COUNT) {
-            ASSERT_from_bufferReturnOut_SIZE(0);
-        } else {
-            ASSERT_from_bufferReturnOut_SIZE(1);
-            ASSERT_from_bufferReturnOut(0, buffer);
-        }
-    }
-    component.cleanup();
+void ComQueueTester ::from_dataOut_handler(FwIndexType portNum, Fw::Buffer& data, const ComCfg::FrameContext& context) {
+    this->pushFromPortEntry_dataOut(data, context);
+    this->invoke_to_dataReturnIn(0, data, context);
 }
 
 }  // end namespace Svc

Svc/ComQueue/test/ut/ComQueueTester.hpp

@@ -77,8 +77,6 @@ class ComQueueTester : public ComQueueGTestBase {
 
     void testContextData();
 
-    void testBufferQueueReturn();
-
   private:
     // ----------------------------------------------------------------------
     // Helper methods
@@ -92,6 +90,9 @@ class ComQueueTester : public ComQueueGTestBase {
     //!
     void initComponents();
 
+    //! Intercept from data out to return the call
+    void from_dataOut_handler(FwIndexType portNum, Fw::Buffer& data, const ComCfg::FrameContext& context) override;
+
   private:
     // ----------------------------------------------------------------------
     // Variables

docs/reference/communication-adapter-interface.md

@@ -65,7 +65,10 @@ This port is used to receive a callback returning ownership of the `Fw::Buffer`
 This port carries a status of `Fw::Success::SUCCESS` or `Fw::Success::FAILURE` typically in response to a call to the `dataIn` port described above. 
 
 > [!NOTE]
-> it is critical to obey the protocol as described in the protocol section below.
+> It is critical to obey the protocol as described in the protocol section below.
+
+> [!CAUTION]
+> Calls to `comStatusOut` must happen after calls to `dataReturnOut` returning the data the com status applies to.
 
 ## Communication Queue Protocol
 
@@ -110,5 +113,8 @@ startup to indicate communication is initially ready and once after each Fw::Suc
 communication has been restored. By emitting Fw::Success::SUCCESS after any failure, the communication adapter ensures
 that each received message eventually results in a Fw::Success::SUCCESS.
 
+Since the communication status reflects the status of specific data transmission it must be sent after the return (deallocation) of that data.
+
+
 > [!NOTE]
 > It is imperative that *Communication Adapters* implement the `comStatusOut` protocol correctly.