bsp_q7s/em/emObjectFactory.cpp

@@ -163,9 +163,11 @@ void ObjectFactory::produce(void* args) {
                                &ipCoreHandler);
   createCcsdsComponents(ccsdsArgs);
 #if OBSW_TM_TO_PTME == 1
-  if (ccsdsArgs.liveDestination != nullptr) {
+  if (ccsdsArgs.normalLiveTmDest != MessageQueueIF::NO_QUEUE) {
-    pusFunnel->addLiveDestination("VC0 LIVE TM", *ccsdsArgs.liveDestination, 0);
+    pusFunnel->addLiveDestinationByRawId("VC0 NORMAL LIVE TM", ccsdsArgs.normalLiveTmDest, 0);
-    cfdpFunnel->addLiveDestination("VC0 LIVE TM", *ccsdsArgs.liveDestination, 0);
+  }
+  if (ccsdsArgs.cfdpLiveTmDest != MessageQueueIF::NO_QUEUE) {
+    cfdpFunnel->addLiveDestinationByRawId("VC0 CFDP LIVE TM", ccsdsArgs.cfdpLiveTmDest, 0);
   }
 #endif
 #endif /* OBSW_ADD_CCSDS_IP_CORES == 1 */

bsp_q7s/objectFactory.cpp

@@ -777,12 +777,13 @@ ReturnValue_t ObjectFactory::createCcsdsComponents(CcsdsComponentArgs& args) {
       new CcsdsIpCoreHandler(objects::CCSDS_HANDLER, objects::CCSDS_PACKET_DISTRIBUTOR, *ptmeConfig,
                              LINK_STATE, &args.gpioComIF, gpios, PTME_LOCKED);
   // This VC will receive all live TM
-  auto* vcWithQueue =
+  auto* vcWithQueue = new VirtualChannel(objects::PTME_VC0_LIVE_TM, ccsds::VC0, "PTME VC0 LIVE TM",
-      new VirtualChannelWithQueue(objects::PTME_VC0_LIVE_TM, ccsds::VC0, "PTME VC0 LIVE TM", *ptme,
+                                         *ptme, LINK_STATE);
-                                  LINK_STATE, args.tmStore, 500);
-  args.liveDestination = vcWithQueue;
   auto* liveTask = new LiveTmTask(objects::LIVE_TM_TASK, args.pusFunnel, args.cfdpFunnel,
-                                  *vcWithQueue, PTME_LOCKED);
+                                  *vcWithQueue, PTME_LOCKED, config::LIVE_CHANNEL_NORMAL_QUEUE_SIZE,
+                                  config::LIVE_CHANNEL_CFDP_QUEUE_SIZE);
+  args.normalLiveTmDest = liveTask->getNormalLiveQueueId();
+  args.cfdpLiveTmDest = liveTask->getCfdpLiveQueueId();
   liveTask->connectModeTreeParent(satsystem::com::SUBSYSTEM);
 
   // Set up log store.

bsp_q7s/objectFactory.h

@@ -46,7 +46,8 @@ struct CcsdsComponentArgs {
   PusTmFunnel& pusFunnel;
   CfdpTmFunnel& cfdpFunnel;
   CcsdsIpCoreHandler** ipCoreHandler;
-  AcceptsTelemetryIF* liveDestination = nullptr;
+  MessageQueueId_t normalLiveTmDest = MessageQueueIF::NO_QUEUE;
+  MessageQueueId_t cfdpLiveTmDest = MessageQueueIF::NO_QUEUE;
 };
 
 void setStatics();

common/config/eive/definitions.h

@@ -58,6 +58,9 @@ static constexpr uint32_t NOK_STORE_QUEUE_SIZE = 350;
 static constexpr uint32_t HK_STORE_QUEUE_SIZE = 300;
 static constexpr uint32_t CFDP_STORE_QUEUE_SIZE = 300;
 
+static constexpr uint32_t LIVE_CHANNEL_NORMAL_QUEUE_SIZE = 300;
+static constexpr uint32_t LIVE_CHANNEL_CFDP_QUEUE_SIZE = 300;
+
 static constexpr uint32_t CFDP_MAX_FSM_CALL_COUNT_SRC_HANDLER = 50;
 static constexpr uint32_t CFDP_MAX_FSM_CALL_COUNT_DEST_HANDLER = 300;
 static constexpr uint32_t CFDP_SHORT_DELAY_MS = 50;

mission/cfdp/CfdpHandler.cpp

@@ -10,6 +10,7 @@
 #include "fsfw/ipc/QueueFactory.h"
 #include "fsfw/tasks/TaskFactory.h"
 #include "fsfw/tmtcservices/TmTcMessage.h"
+#include "mission/sysDefs.h"
 
 using namespace returnvalue;
 using namespace cfdp;
@@ -68,20 +69,38 @@ ReturnValue_t CfdpHandler::initialize() {
       fsmCount++;
     }
     fsmCount = 0;
-    const SourceHandler::FsmResult& srcResult = srcHandler.stateMachine();
+    if (signals::CFDP_CHANNEL_THROTTLE_SIGNAL) {
-    while (srcResult.callStatus == CallStatus::CALL_AGAIN) {
+      throttlePeriodSourceHandler.resetTimer();
-      // Limit number of messages.
+      throttlePeriodOngoing = true;
-      if (fsmCount == config::CFDP_MAX_FSM_CALL_COUNT_SRC_HANDLER) {
+      signals::CFDP_CHANNEL_THROTTLE_SIGNAL = false;
+    }
+
+    if (throttlePeriodOngoing) {
+      if (throttlePeriodSourceHandler.hasTimedOut()) {
+        throttlePeriodOngoing = false;
+
+      } else {
         shortDelay = true;
-        break;
       }
-      srcHandler.stateMachine();
+    }
-      if (srcResult.result == cfdp::TM_STORE_FULL) {
+    // CFDP can be throttled by the slowest live TM handler to handle back pressure in a sensible
-        sif::warning << "CFDP Source Handler: TM store is full" << std::endl;
+    // way without requiring huge amounts of memory for large files.
-      } else if (srcResult.result == cfdp::TARGET_MSG_QUEUE_FULL) {
+    if (!throttlePeriodOngoing) {
-        sif::warning << "CFDP Source Handler: TM queue is full" << std::endl;
+      const SourceHandler::FsmResult& srcResult = srcHandler.stateMachine();
+      while (srcResult.callStatus == CallStatus::CALL_AGAIN) {
+        // Limit number of messages.
+        if (fsmCount == config::CFDP_MAX_FSM_CALL_COUNT_SRC_HANDLER) {
+          shortDelay = true;
+          break;
+        }
+        srcHandler.stateMachine();
+        if (srcResult.result == cfdp::TM_STORE_FULL) {
+          sif::warning << "CFDP Source Handler: TM store is full" << std::endl;
+        } else if (srcResult.result == cfdp::TARGET_MSG_QUEUE_FULL) {
+          sif::warning << "CFDP Source Handler: TM queue is full" << std::endl;
+        }
+        fsmCount++;
       }
-      fsmCount++;
     }
     if (shortDelay) {
       TaskFactory::delayTask(config::CFDP_SHORT_DELAY_MS);

mission/cfdp/CfdpHandler.h

@@ -4,6 +4,7 @@
 #include <etl/queue.h>
 #include <fsfw/cfdp/handler/SourceHandler.h>
 #include <fsfw/ipc/CommandMessage.h>
+#include <fsfw/timemanager/Countdown.h>
 
 #include <utility>
 
@@ -73,6 +74,9 @@ class CfdpHandler : public SystemObject, public ExecutableObjectIF, public Accep
  private:
   MessageQueueIF& pduQueue;
   MessageQueueIF& cfdpRequestQueue;
+  Countdown throttlePeriodSourceHandler = Countdown(80);
+  bool throttlePeriodOngoing = false;
+
   cfdp::LocalEntityCfg localCfg;
   cfdp::RemoteConfigTableIF& remoteCfgProvider;
   cfdp::FsfwParams fsfwParams;

mission/com/LiveTmTask.cpp

@@ -5,8 +5,13 @@
 #include <fsfw/tasks/TaskFactory.h>
 #include <fsfw/timemanager/Stopwatch.h>
 
+#include "mission/sysDefs.h"
+
+std::atomic_bool signals::CFDP_CHANNEL_THROTTLE_SIGNAL = false;
+
 LiveTmTask::LiveTmTask(object_id_t objectId, PusTmFunnel& pusFunnel, CfdpTmFunnel& cfdpFunnel,
-                       VirtualChannelWithQueue& channel, const std::atomic_bool& ptmeLocked)
+                       VirtualChannel& channel, const std::atomic_bool& ptmeLocked,
+                       uint32_t regularTmQueueDepth, uint32_t cfdpQueueDepth)
     : SystemObject(objectId),
       modeHelper(this),
       pusFunnel(pusFunnel),
@@ -14,17 +19,34 @@ LiveTmTask::LiveTmTask(object_id_t objectId, PusTmFunnel& pusFunnel, CfdpTmFunne
       channel(channel),
       ptmeLocked(ptmeLocked) {
   requestQueue = QueueFactory::instance()->createMessageQueue();
+  cfdpTmQueue = QueueFactory::instance()->createMessageQueue(cfdpQueueDepth);
+  regularTmQueue = QueueFactory::instance()->createMessageQueue(regularTmQueueDepth);
 }
 
 ReturnValue_t LiveTmTask::performOperation(uint8_t opCode) {
   readCommandQueue();
+  bool handledTm;
+  ReturnValue_t result;
   while (true) {
-    // The funnel tasks are scheduled here directly as well.
+    // TODO: Must read CFDP TM queue and regular TM queue and forward them. Handle regular queue
-    ReturnValue_t result = channel.handleNextTm(!ptmeLocked);
+    //       first.
-    if (result == DirectTmSinkIF::IS_BUSY) {
+    handledTm = false;
-      sif::error << "Lost live TM, PAPB busy" << std::endl;
+    if (!channel.isBusy()) {
+      result = handleRegularTmQueue();
+      if (result == MessageQueueIF::EMPTY) {
+        result = handleCfdpTmQueue();
+      }
+      if (result == returnvalue::OK) {
+        handledTm = true;
+      }
     }
-    if (result == MessageQueueIF::EMPTY) {
+    if (channel.isBusy()) {
+      // Throttle CFDP packet creator. It is by far the most relevant data creator, so throttling
+      // it is the easiest way to handle back pressure for now in a sensible way. It is cleared
+      // by the data creator.
+      signals::CFDP_CHANNEL_THROTTLE_SIGNAL = true;
+    }
+    if (!handledTm) {
       if (tmFunnelCd.hasTimedOut()) {
         pusFunnel.performOperation(0);
         cfdpFunnel.performOperation(0);
@@ -94,9 +116,17 @@ void LiveTmTask::readCommandQueue(void) {
   }
 }
 
+ReturnValue_t LiveTmTask::handleRegularTmQueue() { return returnvalue::OK; }
+
+ReturnValue_t LiveTmTask::handleCfdpTmQueue() { return returnvalue::OK; }
+
 ModeTreeChildIF& LiveTmTask::getModeTreeChildIF() { return *this; }
 
 ReturnValue_t LiveTmTask::initialize() {
   modeHelper.initialize();
   return returnvalue::OK;
 }
+
+MessageQueueId_t LiveTmTask::getNormalLiveQueueId() const { return regularTmQueue->getId(); }
+
+MessageQueueId_t LiveTmTask::getCfdpLiveQueueId() const { return cfdpTmQueue->getId(); }

mission/com/LiveTmTask.h

@@ -18,25 +18,33 @@ class LiveTmTask : public SystemObject,
                    public ModeTreeConnectionIF {
  public:
   LiveTmTask(object_id_t objectId, PusTmFunnel& pusFunnel, CfdpTmFunnel& cfdpFunnel,
-             VirtualChannelWithQueue& channel, const std::atomic_bool& ptmeLocked);
+             VirtualChannel& channel, const std::atomic_bool& ptmeLocked,
+             uint32_t regularTmQueueDepth, uint32_t cfdpQueueDepth);
 
+  MessageQueueId_t getNormalLiveQueueId() const;
+  MessageQueueId_t getCfdpLiveQueueId() const;
   ReturnValue_t performOperation(uint8_t opCode) override;
   ReturnValue_t initialize() override;
   ReturnValue_t connectModeTreeParent(HasModeTreeChildrenIF& parent) override;
 
  private:
   MessageQueueIF* requestQueue;
+  MessageQueueIF* cfdpTmQueue;
+  MessageQueueIF* regularTmQueue;
   ModeHelper modeHelper;
   Mode_t mode = HasModesIF::MODE_OFF;
   Countdown tmFunnelCd = Countdown(100);
   PusTmFunnel& pusFunnel;
   CfdpTmFunnel& cfdpFunnel;
-  VirtualChannelWithQueue& channel;
+  VirtualChannel& channel;
   uint32_t packetCounter = 0;
   const std::atomic_bool& ptmeLocked;
 
   void readCommandQueue(void);
 
+  ReturnValue_t handleRegularTmQueue();
+  ReturnValue_t handleCfdpTmQueue();
+
   MessageQueueId_t getCommandQueue() const override;
 
   void getMode(Mode_t* mode, Submode_t* submode) override;

mission/tmtc/PusLiveDemux.cpp

@@ -57,9 +57,7 @@ ReturnValue_t PusLiveDemux::demultiplexPackets(StorageManagerIF& tmStore,
 
 uint32_t PusLiveDemux::addDestination(const char* name,
                                       const AcceptsTelemetryIF& downlinkDestination, uint8_t vcid) {
-  auto queueId = downlinkDestination.getReportReceptionQueue(vcid);
+  return addDestinationByRawId(name, downlinkDestination.getReportReceptionQueue(vcid), vcid);
-  destinations.emplace_back(name, queueId, vcid);
-  return destinations.size() - 1;
 }
 
 void PusLiveDemux::setDestFull(uint32_t listIndex) {
@@ -73,3 +71,9 @@ void PusLiveDemux::setDestAvailable(uint32_t listIndex) {
     destinations[listIndex].isFull = false;
   }
 }
+
+uint32_t PusLiveDemux::addDestinationByRawId(const char* name, MessageQueueId_t downlinkDestination,
+                                             uint8_t vcid) {
+  destinations.emplace_back(name, downlinkDestination, vcid);
+  return destinations.size() - 1;
+}

mission/tmtc/PusLiveDemux.h

@@ -14,6 +14,8 @@ class PusLiveDemux {
   ReturnValue_t demultiplexPackets(StorageManagerIF& tmStore, store_address_t origStoreId,
                                    const uint8_t* tmData, size_t tmSize);
 
+  uint32_t addDestinationByRawId(const char* name, MessageQueueId_t downlinkDestination,
+                                 uint8_t vcid = 0);
   uint32_t addDestination(const char* name, const AcceptsTelemetryIF& downlinkDestination,
                           uint8_t vcid = 0);
   void setDestFull(uint32_t listIndex);

mission/tmtc/TmFunnelBase.cpp

@@ -70,3 +70,9 @@ ReturnValue_t TmFunnelBase::saveSequenceCountToFile() {
   ofile << sourceSequenceCount << "\n";
   return returnvalue::OK;
 }
+
+uint32_t TmFunnelBase::addLiveDestinationByRawId(const char *name,
+                                                 MessageQueueId_t downlinkDestination,
+                                                 uint8_t vcid) {
+  return liveDemux.addDestinationByRawId(name, downlinkDestination, vcid);
+}

mission/tmtc/TmFunnelBase.h

@@ -37,6 +37,8 @@ class TmFunnelBase : public AcceptsTelemetryIF, public SystemObject {
   };
   explicit TmFunnelBase(FunnelCfg cfg);
   [[nodiscard]] MessageQueueId_t getReportReceptionQueue(uint8_t virtualChannel) const override;
+  virtual uint32_t addLiveDestinationByRawId(const char* name, MessageQueueId_t downlinkDestination,
+                                             uint8_t vcid = 0);
   virtual uint32_t addLiveDestination(const char* name,
                                       const AcceptsTelemetryIF& downlinkDestination,
                                       uint8_t vcid = 0);