eive/eive-obsw
13
4
Fork 0
Code Issues 4 Pull Requests 1 Releases 121 Wiki Activity

Merge remote-tracking branch 'origin/main' into str-extensions
All checks were successful
EIVE/eive-obsw/pipeline/pr-main This commit looks good
Details

Browse Source
This commit is contained in:
Robin Müller
2023-10-23 13:53:01 +02:00
parent feb6459878 4ee84c0a78
commit 3e5fae4458
56 changed files with 1469 additions and 798 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
mission/cfdp/CMakeLists.txt
View File

@@ -1 +1 @@
target_sources(${LIB_EIVE_MISSION} PRIVATE CfdpHandler.cpp)
target_sources(${LIB_EIVE_MISSION} PRIVATE CfdpHandler.cpp CfdpUser.cpp)

31
mission/cfdp/CfdpFaultHandler.h Normal file
View File

@@ -0,0 +1,31 @@
#ifndef MISSION_CFDP_CFDPFAULTHANDLER_H_
#define MISSION_CFDP_CFDPFAULTHANDLER_H_
#include "fsfw/cfdp.h"
namespace cfdp {
class EiveFaultHandler : public cfdp::FaultHandlerBase {
public:
void noticeOfSuspensionCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Notice of suspension detected for transaction " << id
<< " with condition code: " << cfdp::getConditionCodeString(code) << std::endl;
}
void noticeOfCancellationCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Notice of suspension detected for transaction " << id
<< " with condition code: " << cfdp::getConditionCodeString(code) << std::endl;
}
void abandonCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Transaction " << id
<< " was abandoned, condition code : " << cfdp::getConditionCodeString(code)
<< std::endl;
}
void ignoreCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Fault ignored for transaction " << id
<< ", condition code: " << cfdp::getConditionCodeString(code) << std::endl;
}
};
} // namespace cfdp
#endif /* MISSION_CFDP_CFDPFAULTHANDLER_H_ */

185
mission/cfdp/CfdpHandler.cpp
View File

@@ -1,25 +1,36 @@
#include "CfdpHandler.h"
#include <fsfw/cfdp/CfdpMessage.h>
#include <fsfw/ipc/CommandMessage.h>
#include "eive/definitions.h"
#include "fsfw/cfdp/pdu/AckPduReader.h"
#include "fsfw/cfdp/pdu/PduHeaderReader.h"
#include "fsfw/globalfunctions/arrayprinter.h"
#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;
CfdpHandler::CfdpHandler(const FsfwHandlerParams& fsfwParams, const CfdpHandlerCfg& cfdpCfg)
: SystemObject(fsfwParams.objectId),
msgQueue(fsfwParams.msgQueue),
destHandler(
DestHandlerParams(LocalEntityCfg(cfdpCfg.id, cfdpCfg.indicCfg, cfdpCfg.faultHandler),
cfdpCfg.userHandler, cfdpCfg.remoteCfgProvider, cfdpCfg.packetInfoList,
cfdpCfg.lostSegmentsList),
FsfwParams(fsfwParams.packetDest, nullptr, this, fsfwParams.tcStore,
fsfwParams.tmStore)) {
destHandler.setMsgQueue(msgQueue);
}
CfdpHandler::CfdpHandler(const FsfwHandlerParams& fsfwHandlerParams, const CfdpHandlerCfg& cfdpCfg,
const std::atomic_bool& throttleSignal)
: SystemObject(fsfwHandlerParams.objectId),
pduQueue(fsfwHandlerParams.tmtcQueue),
cfdpRequestQueue(fsfwHandlerParams.cfdpQueue),
localCfg(cfdpCfg.id, cfdpCfg.indicCfg, cfdpCfg.faultHandler),
remoteCfgProvider(cfdpCfg.remoteCfgProvider),
fsfwParams(fsfwHandlerParams.packetDest, &fsfwHandlerParams.tmtcQueue, this,
fsfwHandlerParams.tcStore, fsfwHandlerParams.tmStore),
destHandler(DestHandlerParams(localCfg, cfdpCfg.userHandler, cfdpCfg.remoteCfgProvider,
cfdpCfg.packetInfoList, cfdpCfg.lostSegmentsList),
this->fsfwParams),
srcHandler(SourceHandlerParams(localCfg, cfdpCfg.userHandler, seqCntProvider),
this->fsfwParams),
ipcStore(fsfwHandlerParams.ipcStore),
throttleSignal(throttleSignal) {}
[[nodiscard]] const char* CfdpHandler::getName() const { return "CFDP Handler"; }
@@ -27,7 +38,7 @@ CfdpHandler::CfdpHandler(const FsfwHandlerParams& fsfwParams, const CfdpHandlerC
return destHandler.getDestHandlerParams().cfg.localId.getValue();
}
[[nodiscard]] MessageQueueId_t CfdpHandler::getRequestQueue() const { return msgQueue.getId(); }
[[nodiscard]] MessageQueueId_t CfdpHandler::getRequestQueue() const { return pduQueue.getId(); }
ReturnValue_t CfdpHandler::initialize() {
ReturnValue_t result = destHandler.initialize();
@@ -40,29 +51,63 @@ ReturnValue_t CfdpHandler::initialize() {
return SystemObject::initialize();
}
ReturnValue_t CfdpHandler::performOperation(uint8_t operationCode) {
// TODO: Receive TC packets and route them to source and dest handler, depending on which is
// correct or more appropriate
ReturnValue_t status;
ReturnValue_t result = OK;
TmTcMessage tmtcMsg;
for (status = msgQueue.receiveMessage(&tmtcMsg); status == returnvalue::OK;
status = msgQueue.receiveMessage(&tmtcMsg)) {
result = handleCfdpPacket(tmtcMsg);
[[noreturn]] ReturnValue_t CfdpHandler::performOperation(uint8_t operationCode) {
while (true) {
bool shortDelay = false;
ReturnValue_t result = handlePduPacketMessages();
if (result != OK) {
status = result;
}
result = handleCfdpMessages();
if (result != OK) {
}
uint32_t fsmCount = 1;
const DestHandler::FsmResult& destResult = destHandler.stateMachine();
while (destResult.callStatus == CallStatus::CALL_AGAIN) {
if (fsmCount == config::CFDP_MAX_FSM_CALL_COUNT_DEST_HANDLER) {
shortDelay = true;
break;
}
destHandler.stateMachine();
fsmCount++;
}
fsmCount = 1;
throttlePeriodOngoing = throttleSignal;
// CFDP can be throttled by the slowest live TM handler to handle back pressure in a sensible
// way without requiring huge amounts of memory for large files.
if (!throttlePeriodOngoing) {
const SourceHandler::FsmResult& srcResult = srcHandler.stateMachine();
if (srcResult.packetsSent > 0) {
signals::CFDP_MSG_COUNTER.fetch_add(srcResult.packetsSent, std::memory_order_relaxed);
}
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.packetsSent > 0) {
signals::CFDP_MSG_COUNTER.fetch_add(srcResult.packetsSent, std::memory_order_relaxed);
}
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++;
}
}
if (shortDelay) {
TaskFactory::delayTask(config::CFDP_SHORT_DELAY_MS);
continue;
}
TaskFactory::delayTask(config::CFDP_REGULAR_DELAY_MS);
}
auto& fsmRes = destHandler.performStateMachine();
// TODO: Error handling?
while (fsmRes.callStatus == CallStatus::CALL_AGAIN) {
destHandler.performStateMachine();
// TODO: Error handling?
}
return status;
}
ReturnValue_t CfdpHandler::handleCfdpPacket(TmTcMessage& msg) {
ReturnValue_t CfdpHandler::handlePduPacket(TmTcMessage& msg) {
auto accessorPair = tcStore->getData(msg.getStorageId());
if (accessorPair.first != OK) {
return accessorPair.first;
@@ -102,16 +147,19 @@ ReturnValue_t CfdpHandler::handleCfdpPacket(TmTcMessage& msg) {
auto passToDestHandler = [&]() {
accessorPair.second.release();
PacketInfo info(type, msg.getStorageId(), directive);
result = destHandler.passPacket(info);
return destHandler.passPacket(info);
};
auto passToSourceHandler = [&]() {
accessorPair.second.release();
PacketInfo info(type, msg.getStorageId(), directive);
// Implement this function.
// result = srcHandler.passPacket(info);
};
if (directive == FileDirective::METADATA or directive == FileDirective::EOF_DIRECTIVE or
directive == FileDirective::PROMPT) {
// Section b) of 4.5.3: These PDUs should always be targeted towards the file receiver a.k.a.
// the destination handler
passToDestHandler();
return passToDestHandler();
} else if (directive == FileDirective::FINISH or directive == FileDirective::NAK or
directive == FileDirective::KEEP_ALIVE) {
// Section c) of 4.5.3: These PDUs should always be targeted towards the file sender a.k.a.
@@ -128,9 +176,78 @@ ReturnValue_t CfdpHandler::handleCfdpPacket(TmTcMessage& msg) {
if (ackedDirective == FileDirective::EOF_DIRECTIVE) {
passToSourceHandler();
} else if (ackedDirective == FileDirective::FINISH) {
passToDestHandler();
return passToDestHandler();
}
}
}
return result;
}
ReturnValue_t CfdpHandler::handleCfdpRequest(CommandMessage& msg) {
if (msg.getCommand() == CfdpMessage::PUT_REQUEST) {
sif::info << "Received CFDP put request" << std::endl;
if (srcHandler.getState() != CfdpState::IDLE) {
if (putRequestQueue.full()) {
// TODO: Trigger event and discard request. Queue is full, too many requests.
return FAILED;
}
putRequestQueue.push(CfdpMessage::getStoreId(&msg));
} else {
PutRequest putRequest;
auto accessorPair = ipcStore.getData(CfdpMessage::getStoreId(&msg));
const uint8_t* dataPtr = accessorPair.second.data();
size_t dataSize = accessorPair.second.size();
ReturnValue_t result =
putRequest.deSerialize(&dataPtr, &dataSize, SerializeIF::Endianness::MACHINE);
if (result != OK) {
return result;
}
RemoteEntityCfg* remoteCfg;
remoteCfgProvider.getRemoteCfg(putRequest.getDestId(), &remoteCfg);
if (remoteCfg == nullptr) {
sif::error << "CfdpHandler: No remote configuration found for destination ID "
<< putRequest.getDestId() << std::endl;
// TODO: Trigger event
return FAILED;
}
sif::info << "Starting file copy operation for source file "
<< putRequest.getSourceName().getString() << " and dest file "
<< putRequest.getDestName().getString() << std::endl;
return srcHandler.transactionStart(putRequest, *remoteCfg);
}
}
return OK;
}
ReturnValue_t CfdpHandler::handlePduPacketMessages() {
ReturnValue_t status;
ReturnValue_t result = OK;
TmTcMessage pduMsg;
for (status = pduQueue.receiveMessage(&pduMsg); status == returnvalue::OK;
status = pduQueue.receiveMessage(&pduMsg)) {
result = handlePduPacket(pduMsg);
if (result != OK) {
// TODO: Maybe add printout with context specific information?
status = result;
}
}
return status;
}
ReturnValue_t CfdpHandler::handleCfdpMessages() {
ReturnValue_t status;
ReturnValue_t result;
CommandMessage cfdpMsg;
for (status = cfdpRequestQueue.receiveMessage(&cfdpMsg); status == returnvalue::OK;
status = cfdpRequestQueue.receiveMessage(&cfdpMsg)) {
result = handleCfdpRequest(cfdpMsg);
if (result != OK) {
sif::warning << "Handling CFDP request failed with code 0x" << std::setw(4) << std::hex
<< result << std::dec << std::endl;
triggerEvent(cfdp::events::HANDLING_CFDP_REQUEST_FAILED, 0, result);
// TODO: Maybe add printout with context specific information?
status = result;
}
}
return status;
}

43
mission/cfdp/CfdpHandler.h
View File

@@ -1,6 +1,11 @@
#ifndef FSFW_EXAMPLE_HOSTED_CFDPHANDLER_H
#define FSFW_EXAMPLE_HOSTED_CFDPHANDLER_H
#include <etl/queue.h>
#include <fsfw/cfdp/handler/SourceHandler.h>
#include <fsfw/ipc/CommandMessage.h>
#include <fsfw/timemanager/Countdown.h>
#include <utility>
#include "fsfw/cfdp/handler/DestHandler.h"
@@ -8,22 +13,29 @@
#include "fsfw/tasks/ExecutableObjectIF.h"
#include "fsfw/tmtcservices/AcceptsTelecommandsIF.h"
#include "fsfw/tmtcservices/TmTcMessage.h"
#include "fsfw/util/SeqCountProvider.h"
struct FsfwHandlerParams {
FsfwHandlerParams(object_id_t objectId, HasFileSystemIF& vfs, AcceptsTelemetryIF& packetDest,
StorageManagerIF& tcStore, StorageManagerIF& tmStore, MessageQueueIF& msgQueue)
StorageManagerIF& tcStore, StorageManagerIF& tmStore,
StorageManagerIF& ipcStore, MessageQueueIF& tmtcQueue,
MessageQueueIF& cfdpQueue)
: objectId(objectId),
vfs(vfs),
packetDest(packetDest),
tcStore(tcStore),
tmStore(tmStore),
msgQueue(msgQueue) {}
ipcStore(ipcStore),
tmtcQueue(tmtcQueue),
cfdpQueue(cfdpQueue) {}
object_id_t objectId{};
HasFileSystemIF& vfs;
AcceptsTelemetryIF& packetDest;
StorageManagerIF& tcStore;
StorageManagerIF& tmStore;
MessageQueueIF& msgQueue;
StorageManagerIF& ipcStore;
MessageQueueIF& tmtcQueue;
MessageQueueIF& cfdpQueue;
};
struct CfdpHandlerCfg {
@@ -50,22 +62,39 @@ struct CfdpHandlerCfg {
class CfdpHandler : public SystemObject, public ExecutableObjectIF, public AcceptsTelecommandsIF {
public:
explicit CfdpHandler(const FsfwHandlerParams& fsfwParams, const CfdpHandlerCfg& cfdpCfg);
explicit CfdpHandler(const FsfwHandlerParams& fsfwParams, const CfdpHandlerCfg& cfdpCfg,
const std::atomic_bool& throttleSignal);
[[nodiscard]] const char* getName() const override;
[[nodiscard]] uint32_t getIdentifier() const override;
[[nodiscard]] MessageQueueId_t getRequestQueue() const override;
ReturnValue_t initialize() override;
ReturnValue_t performOperation(uint8_t operationCode) override;
[[noreturn]] ReturnValue_t performOperation(uint8_t operationCode) override;
private:
MessageQueueIF& msgQueue;
MessageQueueIF& pduQueue;
MessageQueueIF& cfdpRequestQueue;
bool throttlePeriodOngoing = false;
cfdp::LocalEntityCfg localCfg;
cfdp::RemoteConfigTableIF& remoteCfgProvider;
cfdp::FsfwParams fsfwParams;
SeqCountProviderU16 seqCntProvider;
cfdp::DestHandler destHandler;
cfdp::SourceHandler srcHandler;
etl::queue<store_address_t, 16> putRequestQueue;
StorageManagerIF& ipcStore;
StorageManagerIF* tcStore = nullptr;
StorageManagerIF* tmStore = nullptr;
ReturnValue_t handleCfdpPacket(TmTcMessage& msg);
const std::atomic_bool& throttleSignal;
ReturnValue_t handlePduPacketMessages();
ReturnValue_t handlePduPacket(TmTcMessage& msg);
ReturnValue_t handleCfdpRequest(CommandMessage& msg);
ReturnValue_t handleCfdpMessages();
};
#endif // FSFW_EXAMPLE_HOSTED_CFDPHANDLER_H

51
mission/cfdp/CfdpUser.cpp Normal file
View File

@@ -0,0 +1,51 @@
#include "CfdpUser.h"
#include <fsfw/ipc/QueueFactory.h>
using namespace returnvalue;
namespace cfdp {
EiveUserHandler::EiveUserHandler(HasFileSystemIF& vfs, StorageManagerIF& ipcStore,
MessageQueueId_t cfdpRequestId)
: cfdp::UserBase(vfs), userQueue(QueueFactory::instance()->createMessageQueue(10)) {
if (userQueue == nullptr) {
sif::error << "EiveUserHandler: Queue creation failed" << std::endl;
return;
}
userQueue->setDefaultDestination(cfdpRequestId);
reservedMsgParser = new ReservedMessageParser(ipcStore, *userQueue, cfdpRequestId);
}
EiveUserHandler::~EiveUserHandler() { QueueFactory::instance()->deleteMessageQueue(userQueue); }
void EiveUserHandler::transactionIndication(const cfdp::TransactionId& id) {}
void EiveUserHandler::eofSentIndication(const cfdp::TransactionId& id) {}
void EiveUserHandler::transactionFinishedIndication(const cfdp::TransactionFinishedParams& params) {
sif::info << "File transaction finished for transaction with " << params.id << std::endl;
}
void EiveUserHandler::metadataRecvdIndication(const cfdp::MetadataRecvdParams& params) {
sif::info << "Metadata received for transaction with " << params.id << std::endl;
if (params.numberOfMsgsToUser > 0 and params.msgsToUserArray != nullptr) {
ReturnValue_t result =
reservedMsgParser->parse(params.msgsToUserArray, params.numberOfMsgsToUser);
if (result != OK) {
sif::warning << "EiveUserHandler: Parsing reserved CFDP messages failed" << std::endl;
}
}
}
void EiveUserHandler::fileSegmentRecvdIndication(const cfdp::FileSegmentRecvdParams& params) {}
void EiveUserHandler::reportIndication(const cfdp::TransactionId& id,
cfdp::StatusReportIF& report) {}
void EiveUserHandler::suspendedIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code) {
}
void EiveUserHandler::resumedIndication(const cfdp::TransactionId& id, size_t progress) {}
void EiveUserHandler::faultIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code,
size_t progress) {}
void EiveUserHandler::abandonedIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code,
size_t progress) {}
void EiveUserHandler::eofRecvIndication(const cfdp::TransactionId& id) {
sif::info << "EOF PDU received for transaction with " << id << std::endl;
}
} // namespace cfdp

37
mission/cfdp/CfdpUser.h Normal file
View File

@@ -0,0 +1,37 @@
#ifndef MISSION_CFDP_CFDPUSER_H_
#define MISSION_CFDP_CFDPUSER_H_
#include <fsfw/cfdp/handler/ReservedMessageParser.h>
#include <fsfw/cfdp/handler/UserBase.h>
namespace cfdp {
class EiveUserHandler : public cfdp::UserBase {
public:
explicit EiveUserHandler(HasFileSystemIF& vfs, StorageManagerIF& ipcStore,
MessageQueueId_t cfdpRequestId);
virtual ~EiveUserHandler();
void transactionIndication(const cfdp::TransactionId& id) override;
void eofSentIndication(const cfdp::TransactionId& id) override;
void transactionFinishedIndication(const cfdp::TransactionFinishedParams& params) override;
void metadataRecvdIndication(const cfdp::MetadataRecvdParams& params) override;
void fileSegmentRecvdIndication(const cfdp::FileSegmentRecvdParams& params) override;
void reportIndication(const cfdp::TransactionId& id, cfdp::StatusReportIF& report) override;
void suspendedIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code);
void resumedIndication(const cfdp::TransactionId& id, size_t progress) override;
void faultIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code,
size_t progress) override;
void abandonedIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code,
size_t progress) override;
void eofRecvIndication(const cfdp::TransactionId& id) override;
private:
MessageQueueIF* userQueue;
ReservedMessageParser* reservedMsgParser;
};
} // namespace cfdp
#endif /* MISSION_CFDP_CFDPUSER_H_ */

57
mission/cfdp/Config.h
View File

@@ -1,57 +0,0 @@
#ifndef MISSION_CFDP_CONFIG_H_
#define MISSION_CFDP_CONFIG_H_
#include "fsfw/cfdp.h"
namespace cfdp {
class EiveUserHandler : public cfdp::UserBase {
public:
explicit EiveUserHandler(HasFileSystemIF& vfs) : cfdp::UserBase(vfs) {}
virtual ~EiveUserHandler() = default;
void transactionIndication(const cfdp::TransactionId& id) override {}
void eofSentIndication(const cfdp::TransactionId& id) override {}
void transactionFinishedIndication(const cfdp::TransactionFinishedParams& params) override {
sif::info << "File transaction finished for transaction with " << params.id << std::endl;
}
void metadataRecvdIndication(const cfdp::MetadataRecvdParams& params) override {
sif::info << "Metadata received for transaction with " << params.id << std::endl;
}
void fileSegmentRecvdIndication(const cfdp::FileSegmentRecvdParams& params) override {}
void reportIndication(const cfdp::TransactionId& id, cfdp::StatusReportIF& report) override {}
void suspendedIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code) override {}
void resumedIndication(const cfdp::TransactionId& id, size_t progress) override {}
void faultIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code,
size_t progress) override {}
void abandonedIndication(const cfdp::TransactionId& id, cfdp::ConditionCode code,
size_t progress) override {}
void eofRecvIndication(const cfdp::TransactionId& id) override {
sif::info << "EOF PDU received for transaction with " << id << std::endl;
}
};
class EiveFaultHandler : public cfdp::FaultHandlerBase {
public:
void noticeOfSuspensionCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Notice of suspension detected for transaction " << id
<< " with condition code: " << cfdp::getConditionCodeString(code) << std::endl;
}
void noticeOfCancellationCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Notice of suspension detected for transaction " << id
<< " with condition code: " << cfdp::getConditionCodeString(code) << std::endl;
}
void abandonCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Transaction " << id
<< " was abandoned, condition code : " << cfdp::getConditionCodeString(code)
<< std::endl;
}
void ignoreCb(cfdp::TransactionId& id, cfdp::ConditionCode code) override {
sif::warning << "Fault ignored for transaction " << id
<< ", condition code: " << cfdp::getConditionCodeString(code) << std::endl;
}
};
} // namespace cfdp
#endif /* MISSION_CFDP_CONFIG_H_ */

156
mission/com/LiveTmTask.cpp
View File

@@ -5,8 +5,15 @@
#include <fsfw/tasks/TaskFactory.h>
#include <fsfw/timemanager/Stopwatch.h>
#include "mission/sysDefs.h"
static constexpr bool DEBUG_TM_QUEUE_SPEED = false;
std::atomic_bool signals::CFDP_CHANNEL_THROTTLE_SIGNAL = false;
std::atomic_uint32_t signals::CFDP_MSG_COUNTER = 0;
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 +21,47 @@ 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;
uint32_t consecutiveRegularCounter = 0;
uint32_t consecutiveCfdpCounter = 0;
bool isCfdp = false;
while (true) {
// The funnel tasks are scheduled here directly as well.
ReturnValue_t result = channel.handleNextTm(!ptmeLocked);
if (result == DirectTmSinkIF::IS_BUSY) {
sif::error << "Lost live TM, PAPB busy" << std::endl;
isCfdp = false;
// TODO: Must read CFDP TM queue and regular TM queue and forward them. Handle regular queue
// first.
handledTm = false;
updateBusyFlag();
if (!channelIsBusy) {
result = handleRegularTmQueue();
if (result == MessageQueueIF::EMPTY) {
result = handleCfdpTmQueue();
isCfdp = true;
}
if (result == returnvalue::OK) {
handledTm = true;
if (DEBUG_TM_QUEUE_SPEED) {
if (isCfdp) {
consecutiveCfdpCounter++;
} else {
consecutiveRegularCounter++;
}
}
} else if (result != MessageQueueIF::EMPTY) {
sif::warning << "LiveTmTask: TM queue failure, returncode 0x" << std::hex << std::setw(4)
<< result << std::dec << std::endl;
}
}
if (result == MessageQueueIF::EMPTY) {
cfdpBackpressureHandling();
if (!handledTm) {
if (tmFunnelCd.hasTimedOut()) {
pusFunnel.performOperation(0);
cfdpFunnel.performOperation(0);
@@ -32,10 +69,19 @@ ReturnValue_t LiveTmTask::performOperation(uint8_t opCode) {
}
// Read command queue during idle times.
readCommandQueue();
if (DEBUG_TM_QUEUE_SPEED) {
if (consecutiveCfdpCounter > 0) {
sif::debug << "Consecutive CFDP TM handled: " << consecutiveCfdpCounter << std::endl;
}
if (consecutiveRegularCounter > 0) {
sif::debug << "Consecutive regular TM handled: " << consecutiveRegularCounter
<< std::endl;
}
consecutiveRegularCounter = 0;
consecutiveCfdpCounter = 0;
}
// 40 ms IDLE delay. Might tweak this in the future.
TaskFactory::delayTask(40);
} else {
packetCounter++;
}
}
}
@@ -94,9 +140,103 @@ void LiveTmTask::readCommandQueue(void) {
}
}
ReturnValue_t LiveTmTask::handleRegularTmQueue() {
return handleGenericTmQueue(*regularTmQueue, false);
}
ReturnValue_t LiveTmTask::handleCfdpTmQueue() { return handleGenericTmQueue(*cfdpTmQueue, true); }
ReturnValue_t LiveTmTask::handleGenericTmQueue(MessageQueueIF& queue, bool isCfdp) {
TmTcMessage message;
ReturnValue_t result = queue.receiveMessage(&message);
if (result == MessageQueueIF::EMPTY) {
return result;
}
if (isCfdp and signals::CFDP_MSG_COUNTER > 0) {
signals::CFDP_MSG_COUNTER--;
}
if (DEBUG_CFDP_TO_LIVE_TM_TASK and signals::CFDP_MSG_COUNTER > 0) {
sif::debug << "LiveTmTask: CFDP message counter: " << signals::CFDP_MSG_COUNTER << std::endl;
}
store_address_t storeId = message.getStorageId();
const uint8_t* data = nullptr;
size_t size = 0;
result = tmStore->getData(storeId, &data, &size);
if (result != returnvalue::OK) {
sif::warning << "VirtualChannel::performOperation: Failed to read data from TM store"
<< std::endl;
tmStore->deleteData(storeId);
return result;
}
if (!ptmeLocked) {
size_t writtenSize = 0;
result = channel.write(data, size, writtenSize);
if (result == DirectTmSinkIF::PARTIALLY_WRITTEN) {
result = channel.handleWriteCompletionSynchronously(writtenSize, 200);
if (result != returnvalue::OK) {
// TODO: Event? Might lead to dangerous spam though..
sif::warning << "LiveTmTask: Synchronous write of last segment failed with code 0x"
<< std::setw(4) << std::hex << result << std::dec << std::endl;
}
} else if (result != returnvalue::OK) {
sif::error << "LiveTmTask: Channel write failed with code 0x" << std::hex << std::setw(4)
<< result << std::dec << std::endl;
}
}
// Try delete in any case, ignore failures (which should not happen), it is more important to
// propagate write errors.
tmStore->deleteData(storeId);
return result;
}
void LiveTmTask::throttleCfdp() {
throttlePeriodOngoing = true;
signals::CFDP_CHANNEL_THROTTLE_SIGNAL = true;
if (DEBUG_CFDP_TO_LIVE_TM_TASK) {
sif::debug << "Throttling CFDP" << std::endl;
}
}
void LiveTmTask::releaseCfdp() {
throttlePeriodOngoing = false;
signals::CFDP_CHANNEL_THROTTLE_SIGNAL = false;
if (DEBUG_CFDP_TO_LIVE_TM_TASK) {
sif::debug << "Releasing CFDP" << std::endl;
}
}
void LiveTmTask::updateBusyFlag() {
// We cache this as a member, because the busy bit can toggle very quickly..
channelIsBusy = channel.isBusy();
}
void LiveTmTask::cfdpBackpressureHandling() {
if (channelIsBusy and !throttlePeriodOngoing) {
// 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.
if (signals::CFDP_MSG_COUNTER >= (config::LIVE_CHANNEL_CFDP_QUEUE_SIZE / 2)) {
throttleCfdp();
}
} else if (!channelIsBusy and throttlePeriodOngoing) {
// Half full/empty flow control: Release the CFDP is the queue is empty enough.
if (signals::CFDP_MSG_COUNTER <= (config::LIVE_CHANNEL_CFDP_QUEUE_SIZE / 4)) {
releaseCfdp();
}
}
}
ModeTreeChildIF& LiveTmTask::getModeTreeChildIF() { return *this; }
ReturnValue_t LiveTmTask::initialize() {
modeHelper.initialize();
tmStore = ObjectManager::instance()->get<StorageManagerIF>(objects::TM_STORE);
if (tmStore == nullptr) {
return ObjectManagerIF::CHILD_INIT_FAILED;
}
return returnvalue::OK;
}
MessageQueueId_t LiveTmTask::getNormalLiveQueueId() const { return regularTmQueue->getId(); }
MessageQueueId_t LiveTmTask::getCfdpLiveQueueId() const { return cfdpTmQueue->getId(); }

28
mission/com/LiveTmTask.h
View File

@@ -11,6 +11,12 @@
#include <mission/tmtc/CfdpTmFunnel.h>
#include <mission/tmtc/PusTmFunnel.h>
#include <cstdint>
#include "eive/definitions.h"
static constexpr bool DEBUG_CFDP_TO_LIVE_TM_TASK = false;
class LiveTmTask : public SystemObject,
public HasModesIF,
public ExecutableObjectIF,
@@ -18,35 +24,51 @@ 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;
StorageManagerIF* tmStore = nullptr;
ModeHelper modeHelper;
Mode_t mode = HasModesIF::MODE_OFF;
Countdown tmFunnelCd = Countdown(100);
PusTmFunnel& pusFunnel;
CfdpTmFunnel& cfdpFunnel;
VirtualChannelWithQueue& channel;
uint32_t packetCounter = 0;
VirtualChannel& channel;
const std::atomic_bool& ptmeLocked;
bool throttlePeriodOngoing = false;
bool channelIsBusy = false;
void readCommandQueue(void);
ReturnValue_t handleRegularTmQueue();
ReturnValue_t handleCfdpTmQueue();
ReturnValue_t handleGenericTmQueue(MessageQueueIF& queue, bool isCfdp);
MessageQueueId_t getCommandQueue() const override;
void getMode(Mode_t* mode, Submode_t* submode) override;
void cfdpBackpressureHandling();
ReturnValue_t checkModeCommand(Mode_t mode, Submode_t submode,
uint32_t* msToReachTheMode) override;
void startTransition(Mode_t mode, Submode_t submode) override;
void announceMode(bool recursive) override;
void throttleCfdp();
void releaseCfdp();
void updateBusyFlag();
object_id_t getObjectId() const override;
const HasHealthIF* getOptHealthIF() const override;

12
mission/com/TmStoreTaskBase.cpp
View File

@@ -138,8 +138,16 @@ ReturnValue_t TmStoreTaskBase::performDump(PersistentTmStoreWithTmQueue& store,
return result;
}
dumpedLen = tmReader.getFullPacketLen();
result = channel.write(tmReader.getFullData(), dumpedLen);
if (result == DirectTmSinkIF::IS_BUSY) {
size_t writtenSize = 0;
result = channel.write(tmReader.getFullData(), dumpedLen, writtenSize);
if (result == VirtualChannelIF::PARTIALLY_WRITTEN) {
result = channel.handleWriteCompletionSynchronously(writtenSize, 200);
if (result != returnvalue::OK) {
// TODO: Event? Might lead to dangerous spam though..
sif::warning << "PersistentTmStore: Synchronous write of last segment failed with code 0x"
<< std::setw(4) << std::hex << result << std::dec << std::endl;
}
} else if (result == DirectTmSinkIF::IS_BUSY) {
sif::warning << "PersistentTmStore: Unexpected VC channel busy" << std::endl;
} else if (result != returnvalue::OK) {
sif::warning << "PersistentTmStore: Unexpected VC channel write failure" << std::endl;

65
mission/com/VirtualChannel.cpp
View File

@@ -1,25 +1,78 @@
#include "VirtualChannel.h"
#include <fsfw/tasks/TaskFactory.h>
VirtualChannel::VirtualChannel(object_id_t objectId, uint8_t vcId, const char* vcName, PtmeIF& ptme,
const std::atomic_bool& txOn)
: SystemObject(objectId), ptme(ptme), vcId(vcId), vcName(vcName), txOn(txOn) {}
ReturnValue_t VirtualChannel::initialize() { return returnvalue::OK; }
ReturnValue_t VirtualChannel::sendNextTm(const uint8_t* data, size_t size) {
return write(data, size);
ReturnValue_t VirtualChannel::sendNextTm(const uint8_t* data, size_t size, size_t& writtenSize) {
return write(data, size, writtenSize);
}
ReturnValue_t VirtualChannel::write(const uint8_t* data, size_t size) {
return ptme.writeToVc(vcId, data, size);
ReturnValue_t VirtualChannel::write(const uint8_t* data, size_t size, size_t& writtenSize) {
if (!ptme.containsVc(vcId)) {
return CHANNEL_DOES_NOT_EXIST;
}
return ptme.getVirtChannel(vcId)->write(data, size, writtenSize);
}
uint8_t VirtualChannel::getVcid() const { return vcId; }
ReturnValue_t VirtualChannel::advanceWrite(size_t& writtenSize) {
if (!ptme.containsVc(vcId)) {
return CHANNEL_DOES_NOT_EXIST;
}
return ptme.getVirtChannel(vcId)->advanceWrite(writtenSize);
}
bool VirtualChannel::writeActive() const {
if (!ptme.containsVc(vcId)) {
return CHANNEL_DOES_NOT_EXIST;
}
return ptme.getVirtChannel(vcId)->writeActive();
}
const char* VirtualChannel::getName() const { return vcName.c_str(); }
bool VirtualChannel::isBusy() const { return ptme.isBusy(vcId); }
bool VirtualChannel::isBusy() const {
if (!ptme.containsVc(vcId)) {
return CHANNEL_DOES_NOT_EXIST;
}
return ptme.getVirtChannel(vcId)->isBusy();
}
void VirtualChannel::cancelTransfer() { ptme.cancelTransfer(vcId); }
void VirtualChannel::cancelTransfer() {
if (!ptme.containsVc(vcId)) {
return;
}
ptme.getVirtChannel(vcId)->cancelTransfer();
}
bool VirtualChannel::isTxOn() const { return txOn; }
ReturnValue_t VirtualChannel::handleWriteCompletionSynchronously(size_t& writtenSize,
unsigned maxCompletionTimeMs) {
unsigned delayMs = 0;
while (true) {
if (isBusy()) {
if (delayMs >= maxCompletionTimeMs) {
break;
}
TaskFactory::delayTask(10);
delayMs += 10;
continue;
}
ReturnValue_t result = advanceWrite(writtenSize);
if (result == returnvalue::OK) {
// Transfer complete
return result;
} else if (result != PARTIALLY_WRITTEN) {
// Some error where we can not or should not continue the transfer.
return result;
}
}
return returnvalue::FAILED;
}

12
mission/com/VirtualChannel.h
View File

@@ -15,6 +15,10 @@
*/
class VirtualChannel : public SystemObject, public VirtualChannelIF {
public:
static constexpr uint8_t CLASS_ID = CLASS_ID::VIRTUAL_CHANNEL;
static constexpr ReturnValue_t CHANNEL_DOES_NOT_EXIST = returnvalue::makeCode(CLASS_ID, 0);
/**
* @brief Constructor
*
@@ -25,9 +29,13 @@ class VirtualChannel : public SystemObject, public VirtualChannelIF {
const std::atomic_bool& linkStateProvider);
ReturnValue_t initialize() override;
ReturnValue_t sendNextTm(const uint8_t* data, size_t size);
ReturnValue_t sendNextTm(const uint8_t* data, size_t size, size_t& writtenSize);
bool isBusy() const override;
ReturnValue_t write(const uint8_t* data, size_t size) override;
ReturnValue_t write(const uint8_t* data, size_t size, size_t& writtenSize) override;
ReturnValue_t advanceWrite(size_t& writtenSize) override;
ReturnValue_t handleWriteCompletionSynchronously(size_t& writtenSize,
unsigned maxCompletionTimeMs);
bool writeActive() const override;
void cancelTransfer() override;
uint8_t getVcid() const;
bool isTxOn() const;

13
mission/com/VirtualChannelWithQueue.cpp
View File

@@ -36,8 +36,19 @@ ReturnValue_t VirtualChannelWithQueue::handleNextTm(bool performWriteOp) {
return result;
}
// TODO: Hnadle partial write handling
size_t writtenSize = 0;
if (performWriteOp) {
result = write(data, size);
result = write(data, size, writtenSize);
if (result == PARTIALLY_WRITTEN) {
result = handleWriteCompletionSynchronously(writtenSize, 200);
if (result != returnvalue::OK) {
// TODO: Event? Might lead to dangerous spam though..
sif::warning
<< "VirtualChannelWithQueue: Synchronous write of last segment failed with code 0x"
<< std::setw(4) << std::hex << result << std::dec << std::endl;
}
}
}
// Try delete in any case, ignore failures (which should not happen), it is more important to
// propagate write errors.

47
mission/genericFactory.cpp
View File

@@ -22,7 +22,9 @@
#include <fsfw/tcdistribution/PusDistributor.h>
#include <fsfw/timemanager/CdsShortTimeStamper.h>
#include <fsfw_hal/host/HostFilesystem.h>
#include <mission/cfdp/CfdpFaultHandler.h>
#include <mission/cfdp/CfdpHandler.h>
#include <mission/cfdp/CfdpUser.h>
#include <mission/controller/ThermalController.h>
#include <mission/genericFactory.h>
#include <mission/persistentTmStoreDefs.h>
@@ -44,7 +46,6 @@
#include "devices/gpioIds.h"
#include "eive/definitions.h"
#include "fsfw/pus/Service11TelecommandScheduling.h"
#include "mission/cfdp/Config.h"
#include "mission/system/acs/RwAssembly.h"
#include "mission/system/acs/acsModeTree.h"
#include "mission/system/tcs/tcsModeTree.h"
@@ -85,7 +86,6 @@ EntityId REMOTE_CFDP_ID(UnsignedByteField<uint16_t>(config::EIVE_GROUND_CFDP_ENT
RemoteEntityCfg GROUND_REMOTE_CFG(REMOTE_CFDP_ID);
OneRemoteConfigProvider REMOTE_CFG_PROVIDER(GROUND_REMOTE_CFG);
HostFilesystem HOST_FS;
EiveUserHandler USER_HANDLER(HOST_FS);
EiveFaultHandler EIVE_FAULT_HANDLER;
} // namespace cfdp
@@ -98,7 +98,8 @@ void ObjectFactory::produceGenericObjects(HealthTableIF** healthTable_, PusTmFun
CfdpTmFunnel** cfdpFunnel, SdCardMountedIF& sdcMan,
StorageManagerIF** ipcStore, StorageManagerIF** tmStore,
PersistentTmStores& stores,
uint32_t eventManagerQueueDepth, bool enableHkSets) {
uint32_t eventManagerQueueDepth, bool enableHkSets,
bool routeToPersistentStores) {
// Framework objects
new EventManager(objects::EVENT_MANAGER, eventManagerQueueDepth);
auto healthTable = new HealthTable(objects::HEALTH_TABLE);
@@ -111,44 +112,45 @@ void ObjectFactory::produceGenericObjects(HealthTableIF** healthTable_, PusTmFun
StorageManagerIF* tcStore;
{
PoolManager::LocalPoolConfig poolCfg = {{250, 16}, {250, 32}, {250, 64},
{150, 128}, {120, 1024}, {120, 2048}};
{150, 128}, {120, 1200}, {120, 2048}};
tcStore = new PoolManager(objects::TC_STORE, poolCfg);
}
{
PoolManager::LocalPoolConfig poolCfg = {{600, 32}, {400, 64}, {400, 128},
{300, 512}, {250, 1024}, {150, 2048}};
{400, 512}, {800, 1200}, {150, 2048}};
*tmStore = new PoolManager(objects::TM_STORE, poolCfg);
}
{
PoolManager::LocalPoolConfig poolCfg = {{300, 16}, {250, 32}, {150, 64}, {150, 128},
{100, 256}, {50, 512}, {50, 1024}, {10, 2048}};
{100, 256}, {50, 512}, {50, 1200}, {10, 2048}};
*ipcStore = new PoolManager(objects::IPC_STORE, poolCfg);
}
PoolManager::LocalPoolConfig poolCfg = {{300, 32}, {400, 64}, {250, 128},
{150, 512}, {150, 1024}, {150, 2048}};
{150, 512}, {400, 1200}, {150, 2048}};
auto* ramToFileStore = new PoolManager(objects::DOWNLINK_RAM_STORE, poolCfg);
#if OBSW_ADD_TCPIP_SERVERS == 1
#if OBSW_ADD_TMTC_UDP_SERVER == 1
auto udpBridge =
new UdpTmTcBridge(objects::UDP_TMTC_SERVER, objects::CCSDS_PACKET_DISTRIBUTOR, 120);
auto udpBridge = new UdpTmTcBridge(objects::UDP_TMTC_SERVER, objects::CCSDS_PACKET_DISTRIBUTOR,
config::UDP_MSG_QUEUE_DEPTH);
new UdpTcPollingTask(objects::UDP_TMTC_POLLING_TASK, objects::UDP_TMTC_SERVER);
sif::info << "Created UDP server for TMTC commanding with listener port "
<< udpBridge->getUdpPort() << std::endl;
udpBridge->setMaxNumberOfPacketsStored(config::MAX_STORED_CMDS_UDP);
udpBridge->setMaxNumberOfPacketsStored(config::UDP_MAX_STORED_CMDS);
#endif
#if OBSW_ADD_TMTC_TCP_SERVER == 1
auto tcpBridge =
new TcpTmTcBridge(objects::TCP_TMTC_SERVER, objects::CCSDS_PACKET_DISTRIBUTOR, 120);
auto tcpBridge = new TcpTmTcBridge(objects::TCP_TMTC_SERVER, objects::CCSDS_PACKET_DISTRIBUTOR,
config::TCP_MSG_QUEUE_DEPTH);
TcpTmTcServer::TcpConfig cfg(true, true);
auto tcpServer = new TcpTmTcServer(objects::TCP_TMTC_POLLING_TASK, objects::TCP_TMTC_SERVER, cfg);
// TCP is stream based. Use packet ID as start marker when parsing for space packets
tcpServer->setSpacePacketParsingOptions({common::PUS_PACKET_ID, common::CFDP_PACKET_ID});
sif::info << "Created TCP server for TMTC commanding with listener port "
<< tcpServer->getTcpPort() << std::endl;
tcpBridge->setMaxNumberOfPacketsStored(config::MAX_STORED_CMDS_TCP);
tcpBridge->setMaxNumberOfPacketsStored(config::TCP_MAX_STORED_CMDS);
tcpBridge->setNumberOfSentPacketsPerCycle(config::TCP_MAX_NUMBER_TMS_SENT_PER_CYCLE);
#endif /* OBSW_USE_TMTC_TCP_BRIDGE == 0 */
#endif /* OBSW_ADD_TCPIP_BRIDGE == 1 */
@@ -223,8 +225,12 @@ void ObjectFactory::produceGenericObjects(HealthTableIF** healthTable_, PusTmFun
**ipcStore, config::MAX_CFDP_FUNNEL_QUEUE_DEPTH, sdcMan,
config::CFDP_SEQUENCE_COUNT_FILE,
core::SAVE_CFDP_SEQUENCE_COUNT);
*cfdpFunnel = new CfdpTmFunnel(cfdpFunnelCfg, stores.cfdpStore->getReportReceptionQueue(0),
*ramToFileStore, config::EIVE_CFDP_APID);
std::optional<MessageQueueId_t> fileStoreDest{};
if (routeToPersistentStores) {
fileStoreDest = stores.cfdpStore->getReportReceptionQueue(0);
}
*cfdpFunnel =
new CfdpTmFunnel(cfdpFunnelCfg, fileStoreDest, *ramToFileStore, config::EIVE_CFDP_APID);
#if OBSW_ADD_TCPIP_SERVERS == 1
#if OBSW_ADD_TMTC_UDP_SERVER == 1
@@ -274,16 +280,19 @@ void ObjectFactory::produceGenericObjects(HealthTableIF** healthTable_, PusTmFun
CfdpDistribCfg distribCfg(objects::CFDP_DISTRIBUTOR, *tcStore, cfdpMsgQueue);
new CfdpDistributor(distribCfg);
auto* msgQueue = QueueFactory::instance()->createMessageQueue(32);
auto* tmtcQueue = QueueFactory::instance()->createMessageQueue(32);
auto* cfdpQueue = QueueFactory::instance()->createMessageQueue(16);
auto eiveUserHandler = new cfdp::EiveUserHandler(HOST_FS, **ipcStore, cfdpQueue->getId());
FsfwHandlerParams params(objects::CFDP_HANDLER, HOST_FS, **cfdpFunnel, *tcStore, **tmStore,
*msgQueue);
**ipcStore, *tmtcQueue, *cfdpQueue);
cfdp::IndicationCfg indicationCfg;
UnsignedByteField<uint16_t> apid(config::EIVE_LOCAL_CFDP_ENTITY_ID);
cfdp::EntityId localId(apid);
GROUND_REMOTE_CFG.defaultChecksum = cfdp::ChecksumType::CRC_32;
CfdpHandlerCfg cfdpCfg(localId, indicationCfg, USER_HANDLER, EIVE_FAULT_HANDLER, PACKET_LIST,
GROUND_REMOTE_CFG.maxFileSegmentLen = config::CFDP_MAX_FILE_SEGMENT_LEN;
CfdpHandlerCfg cfdpCfg(localId, indicationCfg, *eiveUserHandler, EIVE_FAULT_HANDLER, PACKET_LIST,
LOST_SEGMENTS, REMOTE_CFG_PROVIDER);
auto* cfdpHandler = new CfdpHandler(params, cfdpCfg);
auto* cfdpHandler = new CfdpHandler(params, cfdpCfg, signals::CFDP_CHANNEL_THROTTLE_SIGNAL);
// All CFDP packets arrive wrapped inside CCSDS space packets
CcsdsDistributorIF::DestInfo info("CFDP Destination", config::EIVE_CFDP_APID,
cfdpHandler->getRequestQueue(), true);

2
mission/genericFactory.h
View File

@@ -46,7 +46,7 @@ void produceGenericObjects(HealthTableIF** healthTable, PusTmFunnel** pusFunnel,
CfdpTmFunnel** cfdpFunnel, SdCardMountedIF& sdcMan,
StorageManagerIF** ipcStore, StorageManagerIF** tmStore,
PersistentTmStores& stores, uint32_t eventManagerQueueDepth,
bool enableHkSets);
bool enableHkSets, bool routeToPersistentStores);
void createGenericHeaterComponents(GpioIF& gpioIF, PowerSwitchIF& pwrSwitcher,
HeaterHandler*& heaterHandler);

14
mission/scheduling.cpp
View File

@@ -4,13 +4,13 @@
#include "fsfw/tasks/PeriodicTaskIF.h"
void scheduling::scheduleTmpTempSensors(PeriodicTaskIF* tmpTask) {
const std::array<object_id_t, 4> tmpIds = {objects::TMP1075_HANDLER_TCS_0,
objects::TMP1075_HANDLER_TCS_1,
objects::TMP1075_HANDLER_PLPCDU_0,
// damaged.
// objects::TMP1075_HANDLER_PLPCDU_1,
objects::TMP1075_HANDLER_IF_BOARD};
void scheduling::scheduleTmpTempSensors(PeriodicTaskIF* tmpTask, bool schedulePlPcdu1) {
std::vector<object_id_t> tmpIds = {objects::TMP1075_HANDLER_TCS_0, objects::TMP1075_HANDLER_TCS_1,
objects::TMP1075_HANDLER_PLPCDU_0,
objects::TMP1075_HANDLER_IF_BOARD};
if (schedulePlPcdu1) {
tmpIds.push_back(objects::TMP1075_HANDLER_PLPCDU_1);
}
for (const auto& tmpId : tmpIds) {
tmpTask->addComponent(tmpId, DeviceHandlerIF::PERFORM_OPERATION);
tmpTask->addComponent(tmpId, DeviceHandlerIF::SEND_WRITE);

2
mission/scheduling.h
View File

@@ -4,7 +4,7 @@
class PeriodicTaskIF;
namespace scheduling {
void scheduleTmpTempSensors(PeriodicTaskIF* tmpSensors);
void scheduleTmpTempSensors(PeriodicTaskIF* tmpSensors, bool schedulePlPcdu1);
void scheduleRtdSensors(PeriodicTaskIF* periodicTask);
} // namespace scheduling

8
mission/sysDefs.h
View File

@@ -9,7 +9,15 @@
#include <atomic>
#include <cstring>
#include "eive/eventSubsystemIds.h"
namespace signals {
extern std::atomic_bool CFDP_CHANNEL_THROTTLE_SIGNAL;
extern std::atomic_uint16_t I2C_FATAL_ERRORS;
extern std::atomic_uint32_t CFDP_MSG_COUNTER;
} // namespace signals
namespace satsystem {

4
mission/system/systemTree.cpp
View File

@@ -10,6 +10,7 @@
#include "eive/objects.h"
#include "mission/com/defs.h"
#include "mission/sysDefs.h"
#include "mission/system/acs/acsModeTree.h"
#include "mission/system/power/epsModeTree.h"
#include "mission/system/tcs/tcsModeTree.h"
@@ -54,7 +55,8 @@ void satsystem::init(bool commandPlPcdu1) {
EIVE_SYSTEM.setInitialMode(satsystem::Mode::BOOT, 0);
}
EiveSystem satsystem::EIVE_SYSTEM = EiveSystem(objects::EIVE_SYSTEM, 12, 24, I2C_FATAL_ERRORS);
EiveSystem satsystem::EIVE_SYSTEM =
EiveSystem(objects::EIVE_SYSTEM, 12, 24, signals::I2C_FATAL_ERRORS);
auto EIVE_SEQUENCE_BOOT = std::make_pair(satsystem::Mode::BOOT, FixedArrayList<ModeListEntry, 5>());
auto EIVE_TABLE_BOOT_TGT =

37
mission/tmtc/CfdpTmFunnel.cpp
View File

@@ -3,8 +3,10 @@
#include "fsfw/ipc/QueueFactory.h"
#include "fsfw/tmtcpacket/ccsds/SpacePacketCreator.h"
#include "fsfw/tmtcservices/TmTcMessage.h"
#include "mission/sysDefs.h"
CfdpTmFunnel::CfdpTmFunnel(TmFunnelBase::FunnelCfg cfg, MessageQueueId_t fileStoreDest,
CfdpTmFunnel::CfdpTmFunnel(TmFunnelBase::FunnelCfg cfg,
std::optional<MessageQueueId_t> fileStoreDest,
StorageManagerIF& ramToFileStore, uint16_t cfdpInCcsdsApid)
: TmFunnelBase(cfg),
fileStoreDest(fileStoreDest),
@@ -25,7 +27,9 @@ ReturnValue_t CfdpTmFunnel::performOperation(uint8_t) {
saveSequenceCount = false;
}
status = tmQueue->receiveMessage(&currentMessage);
uint32_t handledPackets = 0;
while (status == returnvalue::OK) {
handledPackets++;
status = handlePacket(currentMessage);
if (status != returnvalue::OK) {
sif::warning << "CfdpTmFunnel packet handling failed" << std::endl;
@@ -38,6 +42,11 @@ ReturnValue_t CfdpTmFunnel::performOperation(uint8_t) {
}
status = tmQueue->receiveMessage(&currentMessage);
}
if (handledPackets > 0) {
// Very useful for profiling and debugging
// sif::debug << "CfdpFunnel: Handled " << handledPackets << " packets in one cycle" <<
// std::endl;
}
if (status == MessageQueueIF::EMPTY) {
return returnvalue::OK;
@@ -57,7 +66,7 @@ ReturnValue_t CfdpTmFunnel::handlePacket(TmTcMessage& msg) {
auto spacePacketHeader =
SpacePacketCreator(ccsds::PacketType::TM, false, cfdpInCcsdsApid,
ccsds::SequenceFlags::UNSEGMENTED, sourceSequenceCount++, 0);
sourceSequenceCount = sourceSequenceCount & ccsds::LIMIT_SEQUENCE_COUNT;
sourceSequenceCount = sourceSequenceCount % ccsds::LIMIT_SEQUENCE_COUNT;
spacePacketHeader.setCcsdsLenFromTotalDataFieldLen(cfdpPacketLen);
uint8_t* newPacketData = nullptr;
store_address_t newStoreId{};
@@ -88,14 +97,22 @@ ReturnValue_t CfdpTmFunnel::handlePacket(TmTcMessage& msg) {
msg.setStorageId(newStoreId);
store_address_t origStoreId = newStoreId;
store_address_t storageId;
result = ramToFileStore.addData(&storageId, newPacketData, packetLen);
TmTcMessage fileMsg(storageId);
if (result != returnvalue::OK) {
sif::error << "PusLiveDemux::handlePacket: Store too full to create data copy" << std::endl;
} else {
tmQueue->sendMessage(fileStoreDest, &fileMsg);
if (fileStoreDest.has_value()) {
store_address_t storageId;
result = ramToFileStore.addData(&storageId, newPacketData, packetLen);
TmTcMessage fileMsg(storageId);
if (result == returnvalue::OK) {
tmQueue->sendMessage(fileStoreDest.value(), &fileMsg);
} else if (result == StorageManagerIF::DATA_STORAGE_FULL) {
sif::error << "CfdpTmFunnel::handlePacket: RAM to File Store too full to create data copy"
<< std::endl;
}
}
return demultiplexLivePackets(origStoreId, newPacketData, packetLen);
}
uint32_t CfdpTmFunnel::addLiveDestination(const char* name,
const AcceptsTelemetryIF& downlinkDestination,
uint8_t vcid) {
return TmFunnelBase::addLiveDestination(name, downlinkDestination, vcid);
}

6
mission/tmtc/CfdpTmFunnel.h
View File

@@ -13,16 +13,18 @@
class CfdpTmFunnel : public TmFunnelBase {
public:
CfdpTmFunnel(TmFunnelBase::FunnelCfg cfg, MessageQueueId_t fileStoreDest,
CfdpTmFunnel(TmFunnelBase::FunnelCfg cfg, std::optional<MessageQueueId_t> fileStoreDest,
StorageManagerIF& ramToFileStore, uint16_t cfdpInCcsdsApid);
[[nodiscard]] const char* getName() const override;
uint32_t addLiveDestination(const char* name, const AcceptsTelemetryIF& downlinkDestination,
uint8_t vcid = 0) override;
ReturnValue_t performOperation(uint8_t opCode);
ReturnValue_t initialize() override;
private:
ReturnValue_t handlePacket(TmTcMessage& msg);
MessageQueueId_t fileStoreDest;
std::optional<MessageQueueId_t> fileStoreDest;
StorageManagerIF& ramToFileStore;
uint16_t cfdpInCcsdsApid;
};

35
mission/tmtc/DirectTmSinkIF.h
View File

@@ -13,18 +13,45 @@ class DirectTmSinkIF {
static constexpr uint8_t CLASS_ID = CLASS_ID::TM_SINK;
static constexpr ReturnValue_t IS_BUSY = returnvalue::makeCode(CLASS_ID, 0);
static constexpr ReturnValue_t PARTIALLY_WRITTEN = returnvalue::makeCode(CLASS_ID, 1);
static constexpr ReturnValue_t NO_WRITE_ACTIVE = returnvalue::makeCode(CLASS_ID, 2);
/**
* @brief Implements the functionality to write to a TM sink directly
* @brief Implements the functionality to write to a TM sink directly.
*
* The write might not be completed immediately! If PARTIALLY_WRITTEN is returned, the user
* should poll the ready for packet status bit and call @advanceWrite continuously until
* the transfer is completed.
*
* @param data Pointer to buffer holding the data to write
* @param size Number of bytes to write
* @return returnvalue::OK on success, returnvalue::FAILED on failure, IS_BUSY
* if the TM sink is busy.
* @param writtenSize Size written during write call.
* @return returnvalue::OK on full write success, IS_BUSY if a previous write transfer has not
* been completed yet or the PAPB interface is not ready for a packet, PARTIALLY_WRITTEN
* if some bytes were written, but the transfer has not been completed yet.
*/
virtual ReturnValue_t write(const uint8_t* data, size_t size) = 0;
virtual ReturnValue_t write(const uint8_t* data, size_t size, size_t& writtenSize) = 0;
/**
* Advances a active file transfer.
* @param writtenSize
* @return returnvalue::OK if the packet write process is complete, PARTIALLY_WRITTEN if
* some bytes were written but the transfer is not complete yet.
* NO_WRITE_ACTIVE if this is called without a valid previous write call.
*/
virtual ReturnValue_t advanceWrite(size_t& writtenSize) = 0;
/**
* Is busy, so no write operation can not be started and write advancement
* is not possible.
* @return
*/
virtual bool isBusy() const = 0;
/**
* The PAPB interface is currently busy writing a packet and a new packet can not be written yet.
* @return
*/
virtual bool writeActive() const = 0;
};
#endif /* MISSION_TMTC_DIRECTTMSINKIF_H_ */

43
mission/tmtc/PusLiveDemux.cpp
View File

@@ -9,8 +9,13 @@ ReturnValue_t PusLiveDemux::demultiplexPackets(StorageManagerIF& tmStore,
store_address_t origStoreId, const uint8_t* tmData,
size_t tmSize) {
ReturnValue_t result = returnvalue::OK;
// sif::debug << "tm size: " << tmSize << " for " << destinations.size() << " destinations" <<
// std::endl;
for (unsigned int idx = 0; idx < destinations.size(); idx++) {
const auto& dest = destinations[idx];
if (dest.isFull) {
continue;
}
if ((destinations.size() > 1) and (idx < (destinations.size() - 1))) {
// Create copy of data to ensure each TM recipient has its own copy. That way, we don't need
// to bother with send order and where the data is deleted.
@@ -18,9 +23,20 @@ ReturnValue_t PusLiveDemux::demultiplexPackets(StorageManagerIF& tmStore,
result = tmStore.addData(&storeId, tmData, tmSize);
if (result == returnvalue::OK) {
message.setStorageId(storeId);
} else {
} else if (result == StorageManagerIF::DATA_STORAGE_FULL) {
#if FSFW_CPP_OSTREAM_ENABLED == 1
sif::error << "PusLiveDemux::handlePacket: Store too full to create data copy" << std::endl;
uint8_t fillCounts[16];
uint8_t written = 0;
tmStore.getFillCount(fillCounts, &written);
sif::error << "Fill counts: [";
for (uint8_t fillIdx = 0; fillIdx < written; fillIdx++) {
sif::error << fillCounts[fillIdx];
if (fillIdx < written - 1) {
sif::error << ", ";
}
}
sif::error << "]" << std::endl;
#endif
}
} else {
@@ -39,8 +55,25 @@ ReturnValue_t PusLiveDemux::demultiplexPackets(StorageManagerIF& tmStore,
return result;
}
void PusLiveDemux::addDestination(const char* name, const AcceptsTelemetryIF& downlinkDestination,
uint8_t vcid) {
auto queueId = downlinkDestination.getReportReceptionQueue(vcid);
destinations.emplace_back(name, queueId, vcid);
uint32_t PusLiveDemux::addDestination(const char* name,
const AcceptsTelemetryIF& downlinkDestination, uint8_t vcid) {
return addDestinationByRawId(name, downlinkDestination.getReportReceptionQueue(vcid), vcid);
}
void PusLiveDemux::setDestFull(uint32_t listIndex) {
if (destinations.size() > 0 and listIndex <= destinations.size() - 1) {
destinations[listIndex].isFull = true;
}
}
void PusLiveDemux::setDestAvailable(uint32_t listIndex) {
if (destinations.size() > 0 and listIndex <= destinations.size() - 1) {
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;
}

9
mission/tmtc/PusLiveDemux.h
View File

@@ -14,8 +14,12 @@ class PusLiveDemux {
ReturnValue_t demultiplexPackets(StorageManagerIF& tmStore, store_address_t origStoreId,
const uint8_t* tmData, size_t tmSize);
void addDestination(const char* name, const AcceptsTelemetryIF& downlinkDestination,
uint8_t vcid = 0);
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);
void setDestAvailable(uint32_t listIndex);
private:
struct Destination {
@@ -24,6 +28,7 @@ class PusLiveDemux {
const char* name;
MessageQueueId_t queueId;
bool isFull = false;
uint8_t virtualChannel = 0;
};

13
mission/tmtc/TmFunnelBase.cpp
View File

@@ -30,9 +30,10 @@ MessageQueueId_t TmFunnelBase::getReportReceptionQueue(uint8_t virtualChannel) c
return tmQueue->getId();
}
void TmFunnelBase::addLiveDestination(const char *name,
const AcceptsTelemetryIF &downlinkDestination, uint8_t vcid) {
liveDemux.addDestination(name, downlinkDestination, vcid);
uint32_t TmFunnelBase::addLiveDestination(const char *name,
const AcceptsTelemetryIF &downlinkDestination,
uint8_t vcid) {
return liveDemux.addDestination(name, downlinkDestination, vcid);
}
ReturnValue_t TmFunnelBase::initialize() {
@@ -69,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);
}

7
mission/tmtc/TmFunnelBase.h
View File

@@ -37,8 +37,11 @@ class TmFunnelBase : public AcceptsTelemetryIF, public SystemObject {
};
explicit TmFunnelBase(FunnelCfg cfg);
[[nodiscard]] MessageQueueId_t getReportReceptionQueue(uint8_t virtualChannel) const override;
void addLiveDestination(const char* name, const AcceptsTelemetryIF& downlinkDestination,
uint8_t vcid = 0);
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);
ReturnValue_t demultiplexLivePackets(store_address_t origStoreId, const uint8_t* tmData,
size_t tmSize);
ReturnValue_t initialize() override;