Svc/GenericHub: drop message with event and counter on allocation failure instead of asserting

Svc/ComQueue/docs/sdd.md

@@ -121,6 +121,11 @@ It does the following:
 In the case where the component is already in `READY` state, this will process the queue immediately after the buffer
 is added to the queue.
 
+`Svc::GenericHub` is a `bufferQueueIn` producer in the recommended hub wiring (see the
+[GenericHub SDD](../../GenericHub/docs/sdd.md#transport-formations)); its frames start with a `U32` message type rather
+than a packet descriptor, so the APID read into the frame context is 0 with the default `U16` `FwPacketDescriptorType` and is not
+consumed downstream.
+
 #### 4.5.2 comPacketQueueIn
 The `comPacketQueueIn` port handler receives an `Fw::ComBuffer` data type and a port number. 
 It does the following:

Svc/GenericHub/GenericHub.cpp

@@ -12,6 +12,7 @@
 
 #include <Fw/FPrimeBasicTypes.hpp>
 #include <Svc/GenericHub/GenericHub.hpp>
+#include "Fw/Cmd/CmdPacket.hpp"
 #include "Fw/Logger/Logger.hpp"
 #include "Fw/Types/Assert.hpp"
 
@@ -24,15 +25,35 @@ namespace Svc {
 // Construction, initialization, and destruction
 // ----------------------------------------------------------------------
 
-GenericHub::GenericHub(const char* const compName) : GenericHubComponentBase(compName) {}
+GenericHub::GenericHub(const char* const compName)
+    : GenericHubComponentBase(compName), m_numAllocationFailures(0), m_reportingFailure(false) {}
 
 GenericHub::~GenericHub() {}
 
-void GenericHub::send_data(const HubType type, const FwIndexType port, const U8* data, const FwSizeType size) {
+Fw::Success GenericHub::send_data(const GenericHub_HubType::T type,
+                                  const FwIndexType port,
+                                  const U8* data,
+                                  const FwSizeType size) {
     FW_ASSERT(data != nullptr);
     Fw::SerializeStatus status;
+    const FwSizeType needed = size + sizeof(U32) + sizeof(U32) + sizeof(FwBuffSizeType);
     // Buffer to send and a buffer used to write to it
-    Fw::Buffer outgoing = allocate_out(0, static_cast<U32>(size + sizeof(U32) + sizeof(U32) + sizeof(FwBuffSizeType)));
+    Fw::Buffer outgoing = allocate_out(0, needed);
+    // Allocation may fail (e.g. pool exhausted): drop the message rather than assert
+    if ((outgoing.getData() == nullptr) || (outgoing.getSize() < needed)) {
+        if (outgoing.getData() != nullptr) {
+            deallocate_out(0, outgoing);
+        }
+        this->m_numAllocationFailures++;
+        // Report once per failure episode (re-armed by the next successful send); later drops are only counted
+        if (!this->m_reportingFailure.exchange(true)) {
+            Fw::Logger::log("GenericHub: buffer allocation failed for message type %" PRI_U32
+                            " on port %" PRI_FwIndexType ": requested %" PRI_FwSizeType " bytes\n",
+                            static_cast<U32>(type), port, needed);
+            this->log_WARNING_HI_BufferAllocationFailed(type, port, needed);
+        }
+        return Fw::Success::FAILURE;
+    }
     auto serialize = outgoing.getSerializer();
     // Write data to our buffer
     status = serialize.serializeFrom(static_cast<U32>(type));
@@ -43,14 +64,17 @@ void GenericHub::send_data(const HubType type, const FwIndexType port, const U8*
     FW_ASSERT(status == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(status));
     outgoing.setSize(serialize.getSize());
     toBufferDriver_out(0, outgoing);
+    this->m_reportingFailure.store(false);
+    return Fw::Success::SUCCESS;
 }
 
 // ----------------------------------------------------------------------
 // Handler implementations for user-defined typed input ports
 // ----------------------------------------------------------------------
 
 void GenericHub::bufferIn_handler(const FwIndexType portNum, Fw::Buffer& fwBuffer) {
-    this->send_data(HUB_TYPE_BUFFER, portNum, fwBuffer.getData(), fwBuffer.getSize());
+    // A drop is counted and evented in send_data; this port has no status to return
+    (void)this->send_data(GenericHub_HubType::HUB_TYPE_BUFFER, portNum, fwBuffer.getData(), fwBuffer.getSize());
     bufferInReturn_out(portNum, fwBuffer);
 }
 
@@ -74,7 +98,17 @@ void GenericHub ::cmdDispIn_handler(FwIndexType portNum, Fw::ComBuffer& data, U3
     FW_ASSERT(status == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(status));
     FwSizeType size = serializer.getSize();
 
-    this->send_data(HUB_TYPE_CMD_DISP, portNum, buffer, size);
+    if (this->send_data(GenericHub_HubType::HUB_TYPE_CMD_DISP, portNum, buffer, size) != Fw::Success::SUCCESS) {
+        // The command was dropped: fail it back to the caller so it is not left waiting for a response
+        // Read the opcode with a fresh cursor: an upstream CmdSplitter may already have consumed `data`
+        Fw::ExternalSerializeBuffer reader(data.getBuffAddr(), data.getSize());
+        (void)reader.setBuffLen(data.getSize());
+        Fw::CmdPacket cmdPkt;
+        (void)cmdPkt.deserializeFrom(reader);  // opcode is 0 if the packet is malformed, as in CmdDispatcher
+        if (this->isConnected_cmdRespOut_OutputPort(portNum)) {
+            this->cmdRespOut_out(portNum, cmdPkt.getOpCode(), context, Fw::CmdResponse::EXECUTION_ERROR);
+        }
+    }
 }
 
 void GenericHub ::cmdRespIn_handler(FwIndexType portNum,
@@ -94,11 +128,12 @@ void GenericHub ::cmdRespIn_handler(FwIndexType portNum,
     FW_ASSERT(status == Fw::SerializeStatus::FW_SERIALIZE_OK);
     FwSizeType size = serializer.getSize();
 
-    this->send_data(HUB_TYPE_CMD_RESP, portNum, buffer, size);
+    // A dropped response cannot be reported to the far side (see SDD, "Allocation failure")
+    (void)this->send_data(GenericHub_HubType::HUB_TYPE_CMD_RESP, portNum, buffer, size);
 }
 
 void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer& fwBuffer) {
-    HubType type = HUB_TYPE_MAX;
+    GenericHub_HubType::T type = GenericHub_HubType::HUB_TYPE_PORT;  // Set once type_in is validated below
     U32 type_in = 0;
     U32 port = 0;
     FwBuffSizeType size = 0;
@@ -113,12 +148,12 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
         status = incoming.deserializeTo(type_in);
     }
     // If the deserialization was good, but the type is invalid, set invalid data
-    if ((status == Fw::FW_SERIALIZE_OK) && (type_in >= HUB_TYPE_MAX)) {
+    if ((status == Fw::FW_SERIALIZE_OK) && (type_in >= static_cast<U32>(GenericHub_HubType::NUM_CONSTANTS))) {
         status = Fw::FW_DESERIALIZE_INVALID_DATA;
     }
     // If the deserialization was good and the type was valid, set the type and move onto port deserialization
     else if (status == Fw::FW_SERIALIZE_OK) {
-        type = static_cast<HubType>(type_in);
+        type = static_cast<GenericHub_HubType::T>(type_in);
         status = incoming.deserializeTo(port);
     }
     // If the deserialization was good, move onto size deserialization
@@ -131,7 +166,7 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
         // invokeSerial deserializes arguments before calling a normal invoke, this will return ownership immediately
         U8* rawData = fwBuffer.getData() + sizeof(U32) + sizeof(U32) + sizeof(FwBuffSizeType);
         FwSizeType rawSize = fwBuffer.getSize() - sizeof(U32) - sizeof(U32) - sizeof(FwBuffSizeType);
-        if (type == HUB_TYPE_PORT) {
+        if (type == GenericHub_HubType::HUB_TYPE_PORT) {
             // Com buffer representations should be copied before the call returns, so we need not "allocate" new data
             Fw::ExternalSerializeBuffer wrapper(rawData, rawSize);
             status = wrapper.setBuffLen(rawSize);
@@ -143,7 +178,7 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
             }
             // Deallocate the existing buffer
             fromBufferDriverReturn_out(0, fwBuffer);
-        } else if (type == HUB_TYPE_BUFFER) {
+        } else if (type == GenericHub_HubType::HUB_TYPE_BUFFER) {
             // Fw::Buffers can reuse the existing data buffer as the storage type!  No deallocation done.
             fwBuffer.set(rawData, rawSize, fwBuffer.getContext());
             // Confirm that the port is valid and connected before calling out
@@ -154,7 +189,7 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
                 // Return the buffer if the port is invalid or not connected to avoid leaks
                 fromBufferDriverReturn_out(0, fwBuffer);
             }
-        } else if (type == HUB_TYPE_EVENT) {
+        } else if (type == GenericHub_HubType::HUB_TYPE_EVENT) {
             FwEventIdType id;
             Fw::Time timeTag;
             Fw::LogSeverity severity;
@@ -179,7 +214,7 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
             }
             // Deallocate the existing buffer
             fromBufferDriverReturn_out(0, fwBuffer);
-        } else if (type == HUB_TYPE_CHANNEL) {
+        } else if (type == GenericHub_HubType::HUB_TYPE_CHANNEL) {
             FwChanIdType id;
             Fw::Time timeTag;
             Fw::TlmBuffer val;
@@ -200,7 +235,7 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
 
             // Return the received buffer
             fromBufferDriverReturn_out(0, fwBuffer);
-        } else if (type == HUB_TYPE_CMD_DISP) {
+        } else if (type == GenericHub_HubType::HUB_TYPE_CMD_DISP) {
             U32 context;
             // Check that the size is sufficient for the context
             // The wrapped command must fit the ComBuffer capacity; SERIALIZED_SIZE is the capacity
@@ -225,7 +260,7 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
             }
             // Deallocate the existing buffer
             fromBufferDriverReturn_out(0, fwBuffer);
-        } else if (type == HUB_TYPE_CMD_RESP) {
+        } else if (type == GenericHub_HubType::HUB_TYPE_CMD_RESP) {
             FwOpcodeType opCode;
             U32 cmdSeq;
             Fw::CmdResponse response;
@@ -252,6 +287,10 @@ void GenericHub::fromBufferDriver_handler(const FwIndexType portNum, Fw::Buffer&
     }
 }
 
+void GenericHub::schedIn_handler(FwIndexType portNum, U32 context) {
+    this->tlmWrite_NumAllocationFailures(this->m_numAllocationFailures);
+}
+
 void GenericHub::toBufferDriverReturn_handler(FwIndexType portNum, Fw::Buffer& fwBuffer) {
     // Deallocate the existing buffer
     deallocate_out(portNum, fwBuffer);
@@ -276,7 +315,8 @@ void GenericHub::eventIn_handler(const FwIndexType portNum,
     status = serializer.serializeFrom(args);
     FW_ASSERT(status == Fw::SerializeStatus::FW_SERIALIZE_OK);
     FwSizeType size = serializer.getSize();
-    this->send_data(HubType::HUB_TYPE_EVENT, portNum, buffer, size);
+    // A drop is counted and evented in send_data; this port has no status to return
+    (void)this->send_data(GenericHub_HubType::HUB_TYPE_EVENT, portNum, buffer, size);
 }
 
 void GenericHub::tlmIn_handler(const FwIndexType portNum, FwChanIdType id, Fw::Time& timeTag, Fw::TlmBuffer& val) {
@@ -291,7 +331,8 @@ void GenericHub::tlmIn_handler(const FwIndexType portNum, FwChanIdType id, Fw::T
     status = serializer.serializeFrom(val);
     FW_ASSERT(status == Fw::SerializeStatus::FW_SERIALIZE_OK);
     FwSizeType size = serializer.getSize();
-    this->send_data(HubType::HUB_TYPE_CHANNEL, portNum, buffer, size);
+    // A drop is counted and evented in send_data; this port has no status to return
+    (void)this->send_data(GenericHub_HubType::HUB_TYPE_CHANNEL, portNum, buffer, size);
 }
 
 // ----------------------------------------------------------------------
@@ -301,7 +342,8 @@ void GenericHub::tlmIn_handler(const FwIndexType portNum, FwChanIdType id, Fw::T
 void GenericHub::serialIn_handler(FwIndexType portNum,         /*!< The port number*/
                                   Fw::LinearBufferBase& Buffer /*!< The serialization buffer*/
 ) {
-    send_data(HUB_TYPE_PORT, portNum, Buffer.getBuffAddr(), Buffer.getSize());
+    // A drop is counted and evented in send_data; this port has no status to return
+    (void)send_data(GenericHub_HubType::HUB_TYPE_PORT, portNum, Buffer.getBuffAddr(), Buffer.getSize());
 }
 
 }  // end namespace Svc

Svc/GenericHub/GenericHub.fpp

@@ -42,6 +42,16 @@ module Svc {
   @| ----------------------------------------------------------------------
   passive component GenericHub {
 
+    @ Type of serialized data on the wire, allowing the receiving hub to expand it
+    enum HubType : U32 {
+      HUB_TYPE_PORT      @< Port type transmission
+      HUB_TYPE_BUFFER    @< Buffer type transmission
+      HUB_TYPE_EVENT     @< Event transmission
+      HUB_TYPE_CHANNEL   @< Telemetry channel type
+      HUB_TYPE_CMD_DISP  @< Command dispatch type
+      HUB_TYPE_CMD_RESP  @< Command response type
+    }
+
     # ----------------------------------------------------------------------
     # Ports for sending data from FSW to the hub
     # ----------------------------------------------------------------------
@@ -51,7 +61,7 @@ module Svc {
     # 1. Invoke allocate to allocate a buffer B.
     # 2. Serialize the hub message type (event, telemetry, serial, buffer),
     #    the port number, and the data into B.
-    # 3. Emit B on toBufferDriver.
+    # 3. Emit B on toBufferDriver (or drop the message if the allocation failed).
     #
     # Sample connections:
     #
@@ -106,7 +116,7 @@ module Svc {
     @ for sending remote commands 
     sync input port cmdDispIn: [CmdDispatcherSequencePorts] Fw.Com
 
-    @ Command response back from remote hub
+    @ Command response back from remote hub, or EXECUTION_ERROR for a command dropped on allocation failure
     output port cmdRespOut: [CmdDispatcherSequencePorts] Fw.CmdResponse
 
     # ----------------------------------------------------------------------
@@ -120,6 +130,9 @@ module Svc {
     #    genericHub.toBufferDriver -> bufferDriver.bufferIn
     #    bufferDriver.bufferInReturn -> genericHub.toBufferDriverReturn
     #    genericHub.deallocate -> bufferManager.bufferSendIn
+    #
+    # The recommended formation routes toBufferDriver through Svc.ComQueue and
+    # Svc.ComStub instead of a ByteStreamBufferAdapter; see docs/sdd.md.
     # ----------------------------------------------------------------------
 
     @ This interface provides ports allocate and deallocate
@@ -209,6 +222,37 @@ module Svc {
     @ remote command response 
     sync input port cmdRespIn: [CmdDispatcherSequencePorts] Fw.CmdResponse
 
+    # ----------------------------------------------------------------------
+    # Scheduling, events, and telemetry
+    # ----------------------------------------------------------------------
+
+    @ Schedule input port, used to emit telemetry
+    sync input port schedIn: Svc.Sched
+
+    @ Event port
+    event port Log
+
+    @ Text event port
+    text event port LogText
+
+    @ Time get port
+    time get port Time
+
+    @ Telemetry port
+    telemetry port Tlm
+
+    @ Buffer allocation failed; the outgoing message was dropped
+    event BufferAllocationFailed(
+                                  msgType: HubType @< The hub message type of the dropped message
+                                  $port: FwIndexType @< The input port index of the dropped message
+                                  $size: FwSizeType @< The requested allocation size
+                                ) \
+      severity warning high \
+      format "Buffer allocation failed for message type {} on port {}: requested {} bytes"
+
+    @ The number of failed buffer allocations
+    telemetry NumAllocationFailures: U32
+
   }
 
 }

Svc/GenericHub/GenericHub.hpp

@@ -13,27 +13,14 @@
 #ifndef Svc_GenericHub_HPP
 #define Svc_GenericHub_HPP
 
+#include <atomic>
+#include "Fw/Types/SuccessEnumAc.hpp"
 #include "Svc/GenericHub/GenericHubComponentAc.hpp"
 
 namespace Svc {
 
 class GenericHub final : public GenericHubComponentBase {
   public:
-    /**
-     * HubType:
-     *
-     * Type of serialized data on the wire. Allows for expanding them on the opposing end.
-     */
-    enum HubType {
-        HUB_TYPE_PORT,      //!< Port type transmission
-        HUB_TYPE_BUFFER,    //!< Buffer type transmission
-        HUB_TYPE_EVENT,     //!< Event transmission
-        HUB_TYPE_CHANNEL,   //!< Telemetry channel type
-        HUB_TYPE_CMD_DISP,  //!< Command dispatch type
-        HUB_TYPE_CMD_RESP,  //!< Command response type
-        HUB_TYPE_MAX
-    };
-
     // ----------------------------------------------------------------------
     // Construction, initialization, and destruction
     // ----------------------------------------------------------------------
@@ -86,6 +73,13 @@ class GenericHub final : public GenericHubComponentBase {
     void fromBufferDriver_handler(const FwIndexType portNum, /*!< The port number*/
                                   Fw::Buffer& fwBuffer) override;
 
+    //! Handler implementation for schedIn
+    //!
+    //! Schedule input port, used to emit telemetry
+    void schedIn_handler(FwIndexType portNum,  //!< The port number
+                         U32 context           //!< The call order
+                         ) override;
+
     //! Handler implementation for toBufferDriverReturn
     //!
     //! Port for receiving buffers sent on toBufferDriver and then returned
@@ -121,7 +115,16 @@ class GenericHub final : public GenericHubComponentBase {
                           ) override;
 
     // Helpers and members
-    void send_data(const HubType type, const FwIndexType port, const U8* data, const FwSizeType size);
+
+    //! Frame and send data on toBufferDriver. Returns Fw::Success::FAILURE if the message was dropped because
+    //! no sufficiently sized buffer could be allocated.
+    Fw::Success send_data(const GenericHub_HubType::T type,
+                          const FwIndexType port,
+                          const U8* data,
+                          const FwSizeType size);
+
+    std::atomic<U32> m_numAllocationFailures;  //!< Count of dropped messages; incremented from any caller thread
+    std::atomic<bool> m_reportingFailure;  //!< Set by the first drop of a failure episode, cleared by a successful send
 };
 
 }  // end namespace Svc