DpWriter shrink container buffer after processing (#4904)

* Add unit tests for shrinking processed buffer

* Update DpWriter to shrink processed buffers after processing

* Formatting

* Updated DpWriter to use the dataSize in the container header to shrink the container size

* Update DpWriter SDD with container shrinking steps

* Re-run formatting

* Reset container structure iff the container was processed. Remove cast comment

* Fixed DpWriter formatting

---------

Co-authored-by: M Starch <LeStarch@googlemail.com>
Co-authored-by: Thomas Boyer-Chammard <49786685+thomas-bc@users.noreply.github.com>

Author: 3 people

Fw/Dp/DpContainer.cpp

@@ -142,6 +142,22 @@ void DpContainer::setBuffer(const Buffer& buffer) {
     this->m_dataSize = 0;
 }
 
+void DpContainer::shrinkBufferSize() {
+    // Calculate the assumed size for the Fw::Buffer
+    const FwSizeType newSize = this->getPacketSize();
+
+    // Check that the buffer can still store a data product
+    // AND
+    // That the update is a shrink operation. Growing an
+    // Fw::Buffer is not safe
+    FW_ASSERT(newSize >= MIN_PACKET_SIZE, static_cast<FwAssertArgType>(newSize));
+    FW_ASSERT(newSize <= this->m_buffer.getSize(), static_cast<FwAssertArgType>(newSize),
+              static_cast<FwAssertArgType>(this->m_buffer.getSize()));
+
+    // Shrink the Fw::Buffer
+    this->m_buffer.setSize(newSize);
+}
+
 Utils::HashBuffer DpContainer::getHeaderHash() const {
     const FwSizeType bufferSize = this->m_buffer.getSize();
     const FwSizeType minBufferSize = HEADER_HASH_OFFSET + HASH_DIGEST_LENGTH;

Fw/Dp/DpContainer.hpp

@@ -177,6 +177,10 @@ class DpContainer {
     void setBuffer(const Buffer& buffer  //!< The packet buffer
     );
 
+    //! Shrink the Fw::Buffer size to match the
+    //! DataSize in the container header
+    void shrinkBufferSize();
+
     //! Invalidate the packet buffer
     void invalidateBuffer() {
         this->m_buffer = Fw::Buffer();

Svc/DpWriter/DpWriter.cpp

@@ -158,17 +158,38 @@ Fw::Success::T DpWriter::deserializePacketHeader(Fw::Buffer& buffer, Fw::DpConta
     return status;
 }
 
-void DpWriter::performProcessing(const Fw::DpContainer& container) {
+void DpWriter::performProcessing(Fw::DpContainer& container) {
     // Get the buffer
     Fw::Buffer buffer = container.getBuffer();
     // Get the bit mask for the processing types
     const Fw::DpCfg::ProcType::SerialType procTypes = container.getProcTypes();
     // Do the processing
+    bool did_process = false;
     for (FwIndexType portNum = 0; portNum < NUM_PROCBUFFERSENDOUT_OUTPUT_PORTS; ++portNum) {
         if ((procTypes & (1 << portNum)) != 0) {
             this->procBufferSendOut_out(portNum, buffer);
+            did_process = true;
         }
     }
+
+    if (did_process) {
+        // Updated DpContainer object state with the returned value in the
+        // container buffer
+        Fw::SerializeStatus stat = container.deserializeHeader();
+        FW_ASSERT(stat == Fw::FW_SERIALIZE_OK, stat);
+
+        // Check that the buffer size is compatible with the data size in
+        // the container header
+        FW_ASSERT(container.getDataSize() <= buffer.getSize(), static_cast<FwAssertArgType>(container.getDataSize()),
+                  static_cast<FwAssertArgType>(buffer.getSize()));
+
+        // Re-compute and serialize the container header into the buffer
+        container.updateHeaderHash();
+        container.serializeHeader();
+
+        // Shrink internal Fw::Buffer
+        container.shrinkBufferSize();
+    }
 }
 
 Fw::Success::T DpWriter::writeFile(const Fw::DpContainer& container,

Svc/DpWriter/DpWriter.hpp

@@ -79,7 +79,7 @@ class DpWriter final : public DpWriterComponentBase {
     );
 
     //! Perform processing on a packet buffer
-    void performProcessing(const Fw::DpContainer& container  //!< The container
+    void performProcessing(Fw::DpContainer& container  //!< The container
     );
 
     //! Write the file

Svc/DpWriter/docs/sdd.md

@@ -35,6 +35,7 @@ SVC-DPWRITER-003 | On receiving a data product container _C_, `Svc::DpWriter` sh
 SVC-DPWRITER-004 | On receiving an `Fw::Buffer` _B_, and after performing any requested processing on _B_, `Svc::DpWriter` shall write _B_ to disk. | The purpose of `DpWriter` is to write data products to the disk. | Unit Test
 SVC-DPWRITER-005 | `Svc::DpWriter` shall provide a port for notifying other components that data products have been written. | This requirement allows `Svc::DpCatalog` or a similar component to update its catalog in real time. | Unit Test
 SVC-DPWRITER-006 | `Svc::DpManager` shall provide telemetry that reports the number of buffers received, the number of data products written, the number of bytes written, the number of failed writes, and the number of errors. | This requirement establishes the telemetry interface for the component. | Unit test
+SVC-DPWRITER-007 | On receiving an `Fw::Buffer` _B_, and after performing any requested processing on _B_, `Svc::DpWriter` shall re-parse the container header and shrink the size of the product. | Allows processing interfaces to compress data products and communicate that compressed state back to `Svc::DpWriter`. | Unit Test
 
 ## 3. Design
 
@@ -125,6 +126,10 @@ It does the following:
       `procBufferSendOut` at port number `N`, passing in `B`.
       This step updates the memory pointed to by `B` in place.
 
+   1. Re-parse the container header pointed to by `B`. If necessary,
+      shrink the size of the `B` buffer to be consistent with the data
+      size in the updated container header.
+
    1. Write `B` to a file, using the format described in the [**File
       Format**](#file_format) section. For the time stamp, use the time
       provided by `timeGetOut`.

Svc/DpWriter/test/ut/AbstractState.cpp

@@ -24,6 +24,18 @@ namespace Svc {
 //! Get a data product buffer backed by m_bufferData
 //! \return The buffer
 Fw::Buffer AbstractState::getDpBuffer() {
+    Fw::DpCfg::ProcType::SerialType procTypes = 0;
+    for (FwIndexType i = 0; i < Fw::DpCfg::ProcType::NUM_CONSTANTS; i++) {
+        const bool selector = static_cast<bool>(STest::Pick::lowerUpper(0, 1));
+        if (selector) {
+            procTypes = static_cast<Fw::DpCfg::ProcType::SerialType>(procTypes | (1 << i));
+        }
+    }
+
+    return getDpBufferWithProc(procTypes);
+}
+
+Fw::Buffer AbstractState::getDpBufferWithProc(Fw::DpCfg::ProcType::SerialType procTypes) {
     // Generate the ID
     const FwDpIdType id = static_cast<FwDpIdType>(STest::Pick::lowerUpper(
         std::numeric_limits<FwDpIdType>::min(), static_cast<U32>(std::numeric_limits<FwDpIdType>::max())));
@@ -45,13 +57,6 @@ Fw::Buffer AbstractState::getDpBuffer() {
     const U32 microseconds = STest::Pick::startLength(0, 1000000);
     container.setTimeTag(Fw::Time(seconds, microseconds));
     // Update the processing types
-    Fw::DpCfg::ProcType::SerialType procTypes = 0;
-    for (FwIndexType i = 0; i < Fw::DpCfg::ProcType::NUM_CONSTANTS; i++) {
-        const bool selector = static_cast<bool>(STest::Pick::lowerUpper(0, 1));
-        if (selector) {
-            procTypes = static_cast<Fw::DpCfg::ProcType::SerialType>(procTypes | (1 << i));
-        }
-    }
     container.setProcTypes(procTypes);
     // Update the data size
     container.setDataSize(dataSize);

Svc/DpWriter/test/ut/AbstractState.hpp

@@ -51,7 +51,8 @@ class AbstractState {
           m_NumFailedWrites(0),
           m_NumSuccessfulWrites(0),
           m_NumErrors(0),
-          m_procTypes(0) {}
+          m_procTypes(0),
+          m_procShrinkDataSizeOpt() {}
 
   public:
     // ----------------------------------------------------------------------
@@ -66,10 +67,16 @@ class AbstractState {
     //! Set the data size
     void setDataSize(FwSizeType dataSize) { this->m_dataSizeOpt.set(dataSize); }
 
+    void clearDataSize() { this->m_dataSizeOpt.clear(); }
+
     //! Get a data product buffer backed by bufferData
     //! \return The buffer
     Fw::Buffer getDpBuffer();
 
+    //! Get a data product buffer backed by bufferData
+    //! \return The buffer
+    Fw::Buffer getDpBufferWithProc(Fw::DpCfg::ProcType::SerialType procTypes);
+
   private:
     // ----------------------------------------------------------------------
     // Private state variables
@@ -127,6 +134,8 @@ class AbstractState {
 
     //! Bit mask for processing out port calls
     Fw::DpCfg::ProcType::SerialType m_procTypes;
+
+    TestUtils::Option<FwSizeType> m_procShrinkDataSizeOpt;
 };
 
 }  // namespace Svc

Svc/DpWriter/test/ut/DpWriterTestMain.cpp

@@ -71,6 +71,16 @@ TEST(BufferSendIn, OK) {
     tester.OK();
 }
 
+TEST(BufferSendIn, OKProcShrink) {
+    COMMENT("Invoke bufferSendIn with nominal input. Shrink the buffer in processing");
+    REQUIREMENT("SVC-DPMANAGER-002");
+    REQUIREMENT("SVC-DPMANAGER-003");
+    REQUIREMENT("SVC-DPMANAGER-004");
+    REQUIREMENT("SVC-DPMANAGER-007");
+    BufferSendIn::Tester tester;
+    tester.OKProcShrink();
+}
+
 TEST(CLEAR_EVENT_THROTTLE, OK) {
     COMMENT("Test the CLEAR_EVENT_THROTTLE command.");
     REQUIREMENT("SVC-DPMANAGER-006");

Svc/DpWriter/test/ut/DpWriterTester.cpp

@@ -35,6 +35,15 @@ void DpWriterTester::from_procBufferSendOut_handler(FwIndexType portNum, Fw::Buf
     this->pushFromPortEntry_procBufferSendOut(buffer);
     this->abstractState.m_procTypes =
         static_cast<Fw::DpCfg::ProcType::SerialType>(this->abstractState.m_procTypes | (1 << portNum));
+
+    if (this->abstractState.m_procShrinkDataSizeOpt.hasValue()) {
+        // Update the data size in the container
+        const FwSizeType newSize = this->abstractState.m_procShrinkDataSizeOpt.get();
+        Fw::DpContainer c(0, buffer);
+        c.deserializeHeader();
+        c.setDataSize(newSize);
+        c.serializeHeader();
+    }
 }
 
 // ----------------------------------------------------------------------

Svc/DpWriter/test/ut/Rules/BufferSendIn.cpp

@@ -79,6 +79,72 @@ void TestState ::action__BufferSendIn__OK() {
     this->abstractState.m_NumSuccessfulWrites.value++;
 }
 
+bool TestState ::precondition__BufferSendIn__OKProcShrink() const {
+    const auto& fileData = Os::Stub::File::Test::StaticData::data;
+    bool result = true;
+    result &= (fileData.openStatus == Os::File::Status::OP_OK);
+    result &= (fileData.writeStatus == Os::File::Status::OP_OK);
+    return result;
+}
+
+void TestState ::action__BufferSendIn__OKProcShrink() {
+    // Clear the history
+    this->clearHistory();
+    // Reset the saved proc types
+    // These are updated in the from_procBufferSendOut handler
+    this->abstractState.m_procTypes = 0;
+    // Reset the file pointer in the stub file implementation
+    auto& fileData = Os::Stub::File::Test::StaticData::data;
+    fileData.pointer = 0;
+    // Update m_NumBuffersReceived
+    this->abstractState.m_NumBuffersReceived.value++;
+    // Construct a random buffer
+    const FwSizeType buffer_data_size =
+        STest::Pick::lowerUpper(AbstractState::MAX_DATA_SIZE / 2, AbstractState::MAX_DATA_SIZE);
+    const FwSizeType shrink_data_size = buffer_data_size / 2;
+    this->abstractState.setDataSize(buffer_data_size);
+    Fw::Buffer buffer = this->abstractState.getDpBufferWithProc(1);
+    // Instruct the proc handler to shrink the buffer
+    this->abstractState.m_procShrinkDataSizeOpt.set(shrink_data_size);
+    const FwSizeType exp_buffer_size = Fw::DpContainer::MIN_PACKET_SIZE + shrink_data_size;
+    // Send the buffer
+    this->invoke_to_bufferSendIn(0, buffer);
+    this->doDispatch();
+    // Deserialize the container header
+    Fw::DpContainer container;
+    container.setBuffer(buffer);
+    const Fw::SerializeStatus status = container.deserializeHeader();
+    ASSERT_EQ(status, Fw::FW_SERIALIZE_OK);
+    // Check events
+    ASSERT_EVENTS_SIZE(1);
+    ASSERT_EVENTS_FileWritten_SIZE(1);
+    Fw::FileNameString fileName;
+    this->constructDpFileName(container.getId(), container.getTimeTag(), fileName);
+    ASSERT_EVENTS_FileWritten(0, static_cast<U32>(exp_buffer_size), fileName.toChar());
+    // Check processing types
+    this->checkProcTypes(container);
+    // Check DP notification
+    ASSERT_from_dpWrittenOut_SIZE(1);
+    ASSERT_from_dpWrittenOut(0, fileName, container.getPriority(), exp_buffer_size);
+    // Check deallocation
+    ASSERT_from_deallocBufferSendOut_SIZE(1);
+    ASSERT_from_deallocBufferSendOut(0, buffer);
+    // Check file write
+    ASSERT_EQ(exp_buffer_size, fileData.pointer);
+    ASSERT_EQ(0, ::memcmp(buffer.getData(), fileData.writeResult, exp_buffer_size));
+    // Check data checksum is valid for the container buffer
+    Utils::HashBuffer storedHash;
+    Utils::HashBuffer computedHash;
+    ASSERT_EQ(Fw::Success::SUCCESS, container.checkDataHash(storedHash, computedHash));
+    // Update m_NumBytesWritten
+    this->abstractState.m_NumBytesWritten.value += exp_buffer_size;
+    // Update m_NumSuccessfulWrites
+    this->abstractState.m_NumSuccessfulWrites.value++;
+
+    this->abstractState.clearDataSize();
+    this->abstractState.m_procShrinkDataSizeOpt.clear();
+}
+
 bool TestState ::precondition__BufferSendIn__InvalidBuffer() const {
     bool result = true;
     return result;
@@ -444,6 +510,11 @@ void Tester::OK() {
     this->testState.printEvents();
 }
 
+void Tester::OKProcShrink() {
+    this->ruleOKProcShrink.apply(this->testState);
+    this->testState.printEvents();
+}
+
 }  // namespace BufferSendIn
 
 }  // namespace Svc