// ====================================================================== // \title TransactionRx.cpp // \brief cpp file for CFDP RX Transaction state machine // // This file is a port of RX transaction state machine operations from the following files // from the NASA Core Flight System (cFS) CFDP (CF) Application, version 3.0.0, // adapted for use within the F-Prime (F') framework: // - cf_cfdp_r.c (receive-file transaction state handling routines) // - cf_cfdp_dispatch.c (RX state machine dispatch functions) // // This file contains various state handling routines for // transactions which are receiving a file, as well as dispatch // functions for RX state machines and top-level transaction dispatch. // // ====================================================================== // // NASA Docket No. GSC-18,447-1 // // Copyright (c) 2019 United States Government as represented by the // Administrator of the National Aeronautics and Space Administration. // All Rights Reserved. // // Licensed under the Apache License, Version 2.0 (the "License"); you may // not use this file except in compliance with the License. You may obtain // a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // // ====================================================================== #include #include #include #include #include #include #include #include #include #include namespace Svc { namespace Ccsds { namespace Cfdp { // ====================================================================== // Construction and Destruction // ====================================================================== Transaction::Transaction(Channel* channel, U8 channelId, Engine* engine, CfdpManager* manager) : m_state(TxnState::TXN_STATE_UNDEF), m_txn_class(Cfdp::Class::CLASS_1), m_history(nullptr), m_chunks(nullptr), m_inactivity_timer(), m_ack_timer(), m_fsize(0), m_foffs(0), m_fd(), m_crc(), m_keep(Cfdp::Keep::KEEP), m_chan_num(channelId), // Initialize from parameter m_priority(0), m_initType(TransactionInitType::INIT_BY_COMMAND), m_cl_node{}, m_pb(nullptr), m_state_data{}, m_flags{}, m_cfdpManager(manager), // Initialize from parameter m_chan(channel), // Initialize from parameter m_engine(engine) // Initialize from parameter { // Fully zero the union storage memset(&this->m_state_data, 0, sizeof(this->m_state_data)); memset(&this->m_flags, 0, sizeof(this->m_flags)); } Transaction::~Transaction() {} void Transaction::reset() { // Reset transaction state to default values this->m_state = TxnState::TXN_STATE_UNDEF; this->m_txn_class = Cfdp::Class::CLASS_1; this->m_fsize = 0; this->m_foffs = 0; this->m_keep = Cfdp::Keep::KEEP; this->m_priority = 0; this->m_initType = TransactionInitType::INIT_BY_COMMAND; this->m_crc = CFDP::Checksum(0); this->m_pb = nullptr; // Fully zero the union storage memset(&this->m_state_data, 0, sizeof(this->m_state_data)); memset(&this->m_flags, 0, sizeof(this->m_flags)); // Close the file if it is open if (this->m_fd.isOpen()) { this->m_fd.close(); } // Disable timers to ensure clean state for next transaction // This prevents stale timers from a previous transaction firing in a new context this->m_inactivity_timer.disableTimer(); this->m_ack_timer.disableTimer(); // The following state information is PRESERVED across reset (NOT modified): // - this->m_cfdpManager // Channel binding // - this->m_chan // Channel binding // - this->m_engine // Channel binding // - this->m_chan_num // Channel binding // - this->m_history // Assigned when transaction is activated // - this->m_chunks // Assigned when transaction is activated // - this->m_cl_node // Managed by queue operations in freeTransaction() } // ====================================================================== // RX State Machine - Public Methods // ====================================================================== void Transaction::r1Recv(const Fw::Buffer& buffer) { static const FileDirectiveDispatchTable r1_fdir_handlers = {{ nullptr, /* CFDP_FileDirective_INVALID_MIN */ nullptr, /* 1 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 2 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 3 is unused in the CFDP_FileDirective_t enum */ &Transaction::r1SubstateRecvEof, /* CFDP_FileDirective_EOF */ nullptr, /* CFDP_FileDirective_FIN */ nullptr, /* CFDP_FileDirective_ACK */ nullptr, /* CFDP_FileDirective_METADATA */ nullptr, /* CFDP_FileDirective_NAK */ nullptr, /* CFDP_FileDirective_PROMPT */ nullptr, /* 10 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 11 is unused in the CFDP_FileDirective_t enum */ nullptr, /* CFDP_FileDirective_KEEP_ALIVE */ }}; static const RSubstateDispatchTable substate_fns = {{ &r1_fdir_handlers, /* RxSubState::RX_SUB_STATE_FILEDATA */ &r1_fdir_handlers, /* RxSubState::RX_SUB_STATE_EOF */ &r1_fdir_handlers, /* RxSubState::RX_SUB_STATE_CLOSEOUT_SYNC */ }}; this->rDispatchRecv(buffer, &substate_fns, &Transaction::r1SubstateRecvFileData); } void Transaction::r2Recv(const Fw::Buffer& buffer) { static const FileDirectiveDispatchTable r2_fdir_handlers_normal = {{ nullptr, /* CFDP_FileDirective_INVALID_MIN */ nullptr, /* 1 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 2 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 3 is unused in the CFDP_FileDirective_t enum */ &Transaction::r2SubstateRecvEof, /* CFDP_FileDirective_EOF */ nullptr, /* CFDP_FileDirective_FIN */ nullptr, /* CFDP_FileDirective_ACK */ &Transaction::r2RecvMd, /* CFDP_FileDirective_METADATA */ nullptr, /* CFDP_FileDirective_NAK */ nullptr, /* CFDP_FileDirective_PROMPT */ nullptr, /* 10 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 11 is unused in the CFDP_FileDirective_t enum */ nullptr, /* CFDP_FileDirective_KEEP_ALIVE */ }}; static const FileDirectiveDispatchTable r2_fdir_handlers_finack = {{ nullptr, /* CFDP_FileDirective_INVALID_MIN */ nullptr, /* 1 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 2 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 3 is unused in the CFDP_FileDirective_t enum */ &Transaction::r2SubstateRecvEof, /* CFDP_FileDirective_EOF */ nullptr, /* CFDP_FileDirective_FIN */ &Transaction::r2RecvFinAck, /* CFDP_FileDirective_ACK */ nullptr, /* CFDP_FileDirective_METADATA */ nullptr, /* CFDP_FileDirective_NAK */ nullptr, /* CFDP_FileDirective_PROMPT */ nullptr, /* 10 is unused in the CFDP_FileDirective_t enum */ nullptr, /* 11 is unused in the CFDP_FileDirective_t enum */ nullptr, /* CFDP_FileDirective_KEEP_ALIVE */ }}; static const RSubstateDispatchTable substate_fns = {{ &r2_fdir_handlers_normal, /* RxSubState::RX_SUB_STATE_FILEDATA */ &r2_fdir_handlers_normal, /* RxSubState::RX_SUB_STATE_EOF */ &r2_fdir_handlers_finack, /* RxSubState::RX_SUB_STATE_CLOSEOUT_SYNC */ }}; this->rDispatchRecv(buffer, &substate_fns, &Transaction::r2SubstateRecvFileData); } void Transaction::rAckTimerTick() { U8 ack_limit = 0; /* note: the ack timer is only ever armed on class 2 */ if (this->m_state != TxnState::TXN_STATE_R2 || !this->m_flags.com.ack_timer_armed) { /* nothing to do */ return; } if (this->m_ack_timer.getStatus() == Timer::Status::RUNNING) { this->m_ack_timer.run(); } else { /* ACK timer expired, so check for completion */ if (!this->m_flags.rx.complete) { this->r2Complete(true); } else if (this->m_state_data.receive.sub_state == RxSubState::RX_SUB_STATE_CLOSEOUT_SYNC) { /* Increment acknak counter */ ++this->m_state_data.receive.r2.acknak_count; /* Check limit and handle if needed */ ack_limit = this->m_cfdpManager->getAckLimitParam(this->m_chan_num); if (this->m_state_data.receive.r2.acknak_count >= ack_limit) { this->m_cfdpManager->log_WARNING_LO_RxAckLimitReached(this->getClass(), this->m_history->src_eid, this->m_history->seq_num); this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_ACK_LIMIT_NO_FIN); this->m_cfdpManager->incrementFaultAckLimit(this->m_chan_num); /* give up on this */ this->m_engine->finishTransaction(this, true); this->m_flags.com.ack_timer_armed = false; } else { this->m_flags.rx.send_fin = true; } } /* re-arm the timer if it is still pending */ if (this->m_flags.com.ack_timer_armed) { /* whether sending FIN or waiting for more filedata, need ACK timer armed */ this->m_engine->armAckTimer(this); } } } void Transaction::rTick(I32* cont /* unused */) { /* Steven is not real happy with this function. There should be a better way to separate out * the logic by state so that it isn't a bunch of if statements for different flags */ Status::T sret; bool pending_send; if (!this->m_flags.com.inactivity_fired) { if (this->m_inactivity_timer.getStatus() == Timer::Status::RUNNING) { this->m_inactivity_timer.run(); // Check if timer just expired naturally (after run()) if (this->m_inactivity_timer.getStatus() == Timer::Status::EXPIRED) { this->m_flags.com.inactivity_fired = true; /* HOLD state is the normal path to recycle transaction objects, not an error */ /* Canceled transactions timing out is also normal */ /* inactivity is abnormal in any other state */ if (this->m_state != TxnState::TXN_STATE_HOLD && !this->m_flags.com.canceled) { this->rSendInactivityEvent(); /* in class 2 this also triggers sending an early FIN response */ if (this->m_state == TxnState::TXN_STATE_R2) { this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_INACTIVITY_DETECTED); } } } } } pending_send = true; /* maybe; tbd */ /* rx maintenance: possibly process send_eof_ack, send_nak or send_fin */ if (this->m_flags.rx.send_eof_ack) { sret = this->m_engine->sendAck(this, AckTxnStatus::ACK_TXN_STATUS_ACTIVE, FileDirective::FILE_DIRECTIVE_END_OF_FILE, static_cast(this->m_state_data.receive.r2.eof_cc), this->m_history->peer_eid, this->m_history->seq_num); /* if SUCCESS, move on. If NO_BUF_AVAIL, retry later. If ERROR, stop retrying. */ if (sret == Cfdp::Status::SUCCESS) { this->m_flags.rx.send_eof_ack = false; } else if (sret == Cfdp::Status::ERROR) { /* Serialization failed - error already logged in serializeAndSendPdu */ /* Clear flag to avoid infinite retry loop */ this->m_flags.rx.send_eof_ack = false; pending_send = false; } /* else NO_BUF_AVAIL: leave flag set to retry next tick */ } else if (this->m_flags.rx.send_nak) { if (!this->rSubstateSendNak()) { this->m_flags.rx.send_nak = false; /* will re-enter on error */ } } else if (this->m_flags.rx.send_fin) { if (!this->r2SubstateSendFin()) { this->m_flags.rx.send_fin = false; /* will re-enter on error */ } } else { /* no pending responses to the sender */ pending_send = false; } /* if the inactivity timer ran out, then there is no sense * pending for responses for anything. Send out anything * that we need to send (i.e. the FIN) just in case the sender * is still listening to us but do not expect any future ACKs. * * Bound the deferral like the TX path: after a small retry budget, recycle regardless. */ bool retries_exhausted = false; if (this->m_flags.com.inactivity_fired && pending_send) { if (this->m_flags.com.post_inactivity_send_retries >= this->m_cfdpManager->getPostInactivitySendRetriesParam()) { retries_exhausted = true; } else { this->m_flags.com.post_inactivity_send_retries++; } } if (this->m_flags.com.inactivity_fired && (!pending_send || retries_exhausted)) { /* the transaction is now recyclable - this means we will * no longer have a record of this transaction seq. If the sender * wakes up or if the network delivers severely delayed PDUs at * some future point, then they will be seen as spurious. They * will no longer be associable with this transaction at all */ this->m_chan->recycleTransaction(this); /* NOTE: this must be the last thing in here. Do not use txn after this */ } else { /* transaction still valid so process the ACK timer, if relevant */ this->rAckTimerTick(); } } void Transaction::rCancel() { /* for cancel, only need to send FIN if R2 */ if ((this->m_state == TxnState::TXN_STATE_R2) && (this->m_state_data.receive.sub_state < RxSubState::RX_SUB_STATE_CLOSEOUT_SYNC)) { this->m_flags.rx.send_fin = true; } else { this->r1Reset(); /* if R1, just call it quits */ } } void Transaction::rInit() { Os::File::Status status; Fw::String tmpDir; Fw::String dst; if (this->m_state == TxnState::TXN_STATE_R2) { if (!this->m_flags.rx.md_recv) { tmpDir = this->m_cfdpManager->getTmpDirParam(this->m_chan_num); /* we need to make a temp file and then do a NAK for md PDU */ /* the transaction already has a history, and that has a buffer that we can use to * hold the temp filename which is defined by the sequence number and the source entity ID */ // Create destination filepath with format: /:.tmp dst.format("%s/%" CFDP_PRI_ENTITY_ID ":%" CFDP_PRI_TRANSACTION_SEQ ".tmp", tmpDir.toChar(), this->m_history->src_eid, this->m_history->seq_num); this->m_history->fnames.dst_filename = dst; this->m_cfdpManager->log_ACTIVITY_LO_RxTempFileCreated(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, this->m_history->fnames.dst_filename); } this->m_engine->armAckTimer(this); } status = this->m_fd.open(this->m_history->fnames.dst_filename.toChar(), Os::File::OPEN_CREATE, Os::File::OVERWRITE); if (status != Os::File::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxFileCreateFailed(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, this->m_history->fnames.dst_filename, status); this->m_cfdpManager->incrementFaultFileOpen(this->m_chan_num); if (this->m_state == TxnState::TXN_STATE_R2) { this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_FILESTORE_REJECTION); } else { this->r1Reset(); } } else { this->m_state_data.receive.sub_state = RxSubState::RX_SUB_STATE_FILEDATA; } } void Transaction::r2SetFinTxnStatus(TxnStatus txn_stat) { this->m_engine->setTxnStatus(this, txn_stat); this->m_flags.rx.send_fin = true; } void Transaction::r1Reset() { this->m_engine->finishTransaction(this, true); } void Transaction::r2Reset() { if ((this->m_state_data.receive.sub_state == RxSubState::RX_SUB_STATE_CLOSEOUT_SYNC) || (static_cast(this->m_state_data.receive.r2.eof_cc) != static_cast(ConditionCode::CONDITION_CODE_NO_ERROR)) || TxnStatusIsError(this->m_history->txn_stat) || this->m_flags.com.canceled) { this->r1Reset(); /* it's done */ } else { /* not waiting for FIN ACK, so trigger send FIN */ this->m_flags.rx.send_fin = true; } } Status::T Transaction::rCheckCrc(U32 expected_crc) { Status::T ret = Cfdp::Status::SUCCESS; U32 crc_result; // The F' version does not have an equivalent finalize call as it // - Never stores a partial word internally // - Never needs to "flush" anything // - Always accounts for padding at update time crc_result = this->m_crc.getValue(); if (crc_result != expected_crc) { this->m_cfdpManager->log_WARNING_LO_RxCrcMismatch(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, expected_crc, crc_result); this->m_cfdpManager->incrementFaultCrcMismatch(this->m_chan_num); ret = Cfdp::Status::ERROR; } return ret; } void Transaction::r2Complete(I32 ok_to_send_nak) { U32 ret; bool send_nak = false; bool send_fin = false; U8 nack_limit = 0; /* checking if r2 is complete. Check NAK list, and send NAK if appropriate */ /* if all data is present, then there will be no gaps in the chunk */ if (!TxnStatusIsError(this->m_history->txn_stat)) { /* first, check if md is received. If not, send specialized NAK */ if (!this->m_flags.rx.md_recv) { send_nak = true; } else { /* only look for 1 gap, since the goal here is just to know that there are gaps */ ret = this->m_chunks->chunks.computeGaps(1, this->m_fsize, 0, nullptr, nullptr); if (ret) { /* there is at least 1 gap, so send a NAK */ send_nak = true; } else if (this->m_flags.rx.eof_recv) { /* the EOF was received, and there are no NAKs -- process completion in send FIN state */ send_fin = true; } } if (send_nak && ok_to_send_nak) { /* Increment the acknak counter */ ++this->m_state_data.receive.r2.acknak_count; /* Check limit and handle if needed */ nack_limit = this->m_cfdpManager->getNackLimitParam(this->m_chan_num); if (this->m_state_data.receive.r2.acknak_count >= nack_limit) { this->m_cfdpManager->log_WARNING_LO_RxNakLimitReached(this->getClass(), this->m_history->src_eid, this->m_history->seq_num); send_fin = true; this->m_cfdpManager->incrementFaultNakLimit(this->m_chan_num); /* don't use CFDP_R2_SetFinTxnStatus because many places in this function set send_fin */ this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_NAK_LIMIT_REACHED); this->m_state_data.receive.r2.acknak_count = 0; /* reset for fin/ack */ } else { this->m_flags.rx.send_nak = true; } } if (send_fin) { this->m_flags.rx.complete = true; /* latch completeness, since send_fin is cleared later */ /* the transaction is now considered complete, but this will not overwrite an * error status code if there was one set */ this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_NO_ERROR); } /* always go to RxSubState::RX_SUB_STATE_FILEDATA, and let tick change state */ this->m_state_data.receive.sub_state = RxSubState::RX_SUB_STATE_FILEDATA; } } // ====================================================================== // RX State Machine - Private Helper Methods // ====================================================================== Status::T Transaction::rProcessFd(const Fw::Buffer& buffer) { Status::T ret = Cfdp::Status::SUCCESS; /* this function is only entered for data PDUs */ // Deserialize FileData PDU from buffer FileDataPdu fd; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = fd.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { this->m_cfdpManager->log_WARNING_LO_FailFileDataPduDeserialization(this->getChannelId(), static_cast(deserStatus)); ret = Cfdp::Status::ERROR; } /* * NOTE: The decode routine should have left a direct pointer to the data and actual data length * within the PDU. The length has already been verified, too. Should not need to make any * adjustments here, just write it. */ FileSize offset = fd.getOffset(); U16 dataSize = fd.getDataSize(); const U8* dataPtr = fd.getData(); // Reject file data past the declared file size (subtraction form avoids offset + dataSize overflow). if ((ret == Cfdp::Status::SUCCESS) && this->m_flags.rx.md_recv) { if ((offset > this->m_fsize) || ((this->m_fsize - offset) < dataSize)) { this->m_cfdpManager->log_WARNING_LO_RxFileDataOutOfBounds( this->getClass(), this->m_history->src_eid, this->m_history->seq_num, offset, dataSize, this->m_fsize); this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); this->m_cfdpManager->incrementFaultFileSizeMismatch(this->m_chan_num); ret = Cfdp::Status::ERROR; } } // Seek to file offset if needed if (ret == Cfdp::Status::SUCCESS) { if (this->m_state_data.receive.cached_pos != offset) { Os::File::Status status = this->m_fd.seek(offset, Os::File::SeekType::ABSOLUTE); if (status != Os::File::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxSeekFailed(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, offset, status); this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); this->m_cfdpManager->incrementFaultFileSeek(this->m_chan_num); ret = Cfdp::Status::ERROR; } } } // Write file data if (ret == Cfdp::Status::SUCCESS) { FwSizeType write_size = dataSize; Os::File::Status status = this->m_fd.write(dataPtr, write_size, Os::File::WaitType::WAIT); if (status != Os::File::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxWriteFailed(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, dataSize, static_cast(write_size)); this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_FILESTORE_REJECTION); this->m_cfdpManager->incrementFaultFileWrite(this->m_chan_num); ret = Cfdp::Status::ERROR; } else { this->m_state_data.receive.cached_pos = static_cast(dataSize) + offset; this->m_cfdpManager->addRecvFileDataBytes(this->m_chan_num, dataSize); } } return ret; } Status::T Transaction::rSubstateRecvEof(const Fw::Buffer& buffer) { Status::T ret = Cfdp::Status::SUCCESS; // Deserialize EOF PDU from buffer EofPdu eof; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = eof.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { this->m_cfdpManager->log_WARNING_LO_FailEofPduDeserialization(this->getChannelId(), static_cast(deserStatus)); ret = Cfdp::Status::REC_PDU_BAD_EOF_ERROR; } if (ret == Cfdp::Status::SUCCESS) { // NOTE: Engine::recvEof() currently always returns SUCCESS, so the failure branch is // omitted here. Once recvEof() performs real EOF-level validation (see the TLV // "Future enhancement" TODO in Engine::recvEof) and can return an error status, add an // else branch that emits log_WARNING_LO_RxInvalidEofPdu, increments recvErrors, and sets // Cfdp::Status::REC_PDU_BAD_EOF_ERROR for the failure case. if (!this->m_engine->recvEof(this, eof)) { /* this function is only entered for PDUs identified as EOF type */ ConditionCode cc = eof.getConditionCode(); /* Only check size if MD received and EOF doesn't have a non-zero condition code (e.g., don't check size for * canceled transactions) */ if (this->m_flags.rx.md_recv && (cc == ConditionCode::CONDITION_CODE_NO_ERROR) && (eof.getFileSize() != this->m_fsize)) { this->m_cfdpManager->log_WARNING_LO_RxFileSizeMismatch(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, this->m_fsize, eof.getFileSize()); this->m_cfdpManager->incrementFaultFileSizeMismatch(this->m_chan_num); ret = Cfdp::Status::REC_PDU_FSIZE_MISMATCH_ERROR; } /* Log condition code if non-zero (cancel or error) - applies to both Class 1 and Class 2 */ if (cc != ConditionCode::CONDITION_CODE_NO_ERROR) { /* Set transaction status from condition code to prevent completion event */ this->m_engine->setTxnStatus(this, static_cast(static_cast(cc))); if (cc == ConditionCode::CONDITION_CODE_CANCEL_REQUEST_RECEIVED) { /* Increment receive EOF cancellation counter (normal operation) */ this->m_cfdpManager->incrementRecvEofCanceled(this->m_chan_num); this->m_cfdpManager->log_ACTIVITY_HI_RxEofCancelReceived(this->getClass(), this->m_history->src_eid, this->m_history->seq_num); } else { /* Increment RX EOF error counter (protocol error) */ this->m_cfdpManager->incrementFaultRxEofError(this->m_chan_num); this->m_cfdpManager->log_WARNING_LO_RxEofWithError(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, static_cast(cc)); } } } } return ret; } void Transaction::r1SubstateRecvEof(const Fw::Buffer& buffer) { // Deserialize EOF PDU from buffer EofPdu eof; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = eof.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { // Bad EOF, reset transaction this->m_cfdpManager->log_WARNING_LO_FailEofPduDeserialization(this->getChannelId(), static_cast(deserStatus)); this->r1Reset(); return; } Status::T ret = this->rSubstateRecvEof(buffer); U32 crc = eof.getChecksum(); ConditionCode cc = eof.getConditionCode(); if (ret == Cfdp::Status::SUCCESS) { /* Only check CRC if no error condition code */ if (cc == ConditionCode::CONDITION_CODE_NO_ERROR) { /* Verify CRC */ if (this->rCheckCrc(crc) == Cfdp::Status::SUCCESS) { /* successfully processed the file */ this->m_keep = Cfdp::Keep::KEEP; /* save the file */ } /* if file failed to process, there's nothing to do. CFDP_R_CheckCrc() generates an event on failure */ } } /* after exit, always reset since we are done */ /* reset even if the EOF failed -- class 1, so it won't come again! */ this->r1Reset(); } void Transaction::r2SubstateRecvEof(const Fw::Buffer& buffer) { Status::T ret; if (!this->m_flags.rx.eof_recv) { // Deserialize EOF PDU from buffer EofPdu eof; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = eof.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { // Bad EOF, return to FILEDATA substate this->m_cfdpManager->log_WARNING_LO_FailEofPduDeserialization(this->getChannelId(), static_cast(deserStatus)); this->m_state_data.receive.sub_state = RxSubState::RX_SUB_STATE_FILEDATA; return; } ret = this->rSubstateRecvEof(buffer); /* did receiving EOF succeed? */ if (ret == Cfdp::Status::SUCCESS) { this->m_flags.rx.eof_recv = true; /* need to remember the EOF CRC for later */ this->m_state_data.receive.r2.eof_crc = eof.getChecksum(); this->m_state_data.receive.r2.eof_size = eof.getFileSize(); /* always ACK the EOF, even if we're not done */ this->m_state_data.receive.r2.eof_cc = static_cast(eof.getConditionCode()); this->m_flags.rx.send_eof_ack = true; /* defer sending ACK to tick handling */ /* only check for complete if EOF with no errors */ if (static_cast(this->m_state_data.receive.r2.eof_cc) == static_cast(ConditionCode::CONDITION_CODE_NO_ERROR)) { this->r2Complete(true); /* CFDP_R2_Complete() will change state */ } else { /* All CFDP CC values directly correspond to a Transaction Status of the same numeric value */ this->m_engine->setTxnStatus( this, static_cast(static_cast(this->m_state_data.receive.r2.eof_cc))); this->r2Reset(); } } else { /* bad EOF sent? */ if (ret == Cfdp::Status::REC_PDU_FSIZE_MISMATCH_ERROR) { this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); } else { /* can't do anything with this bad EOF, so return to FILEDATA */ this->m_state_data.receive.sub_state = RxSubState::RX_SUB_STATE_FILEDATA; } } } } void Transaction::r1SubstateRecvFileData(const Fw::Buffer& buffer) { Status::T ret; // Deserialize FileData PDU from buffer FileDataPdu fd; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = fd.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { // Bad file data PDU, reset transaction this->m_cfdpManager->log_WARNING_LO_FailFileDataPduDeserialization(this->getChannelId(), static_cast(deserStatus)); this->r1Reset(); return; } /* got file data PDU? */ ret = this->m_engine->recvFd(this, fd); if (ret == Cfdp::Status::SUCCESS) { ret = this->rProcessFd(buffer); } if (ret == Cfdp::Status::SUCCESS) { /* class 1 digests CRC */ this->m_crc.update(fd.getData(), fd.getOffset(), static_cast(fd.getDataSize())); } else { /* Reset transaction on failure */ this->r1Reset(); } } void Transaction::r2SubstateRecvFileData(const Fw::Buffer& buffer) { Status::T ret; // If CRC calculation has started (file reopened in READ mode), ignore late FileData PDUs. // This can happen if retransmitted FileData arrives after EOF was received and CRC began. if (this->m_state_data.receive.r2.rx_crc_calc_bytes > 0) { // Silently ignore - file is complete and we're calculating CRC // No EVR needed - late retransmissions are expected in CFDP Class 2 return; } // Deserialize FileData PDU from buffer FileDataPdu fd; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = fd.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { // Bad file data PDU, reset transaction this->m_cfdpManager->log_WARNING_LO_FailFileDataPduDeserialization(this->getChannelId(), static_cast(deserStatus)); this->r2Reset(); return; } /* got file data PDU? */ ret = this->m_engine->recvFd(this, fd); if (ret == Cfdp::Status::SUCCESS) { ret = this->rProcessFd(buffer); } if (ret == Cfdp::Status::SUCCESS) { /* class 2 does CRC at FIN, but track gaps */ this->m_chunks->chunks.add(fd.getOffset(), static_cast(fd.getDataSize())); if (this->m_flags.rx.fd_nak_sent) { this->r2Complete(false); /* once nak-retransmit received, start checking for completion at each fd */ } if (!this->m_flags.rx.complete) { this->m_engine->armAckTimer(this); /* re-arm ACK timer, since we got data */ } this->m_state_data.receive.r2.acknak_count = 0; } else { /* Reset transaction on failure */ this->r2Reset(); } } void Transaction::r2GapCompute(const Chunk* chunk, NakPdu& nak) { FW_ASSERT(chunk->size > 0, static_cast(chunk->size)); // Calculate segment offsets relative to scope start FileSize offsetStart = chunk->offset - nak.getScopeStart(); FileSize offsetEnd = offsetStart + chunk->size; // Add segment to NAK PDU (returns false if array is full) nak.addSegment(offsetStart, offsetEnd); } void Transaction::r2GapComputeWrapper(const Chunk* chunk, void* opaque) { struct GapComputeContext { Transaction* txn; NakPdu* nak; }; GapComputeContext* ctx = static_cast(opaque); ctx->txn->r2GapCompute(chunk, *ctx->nak); } Status::T Transaction::rSubstateSendNak() { Status::T status = Cfdp::Status::SUCCESS; // Create and initialize NAK PDU NakPdu nakPdu; Cfdp::PduDirection direction = PduDirection::DIRECTION_TOWARD_SENDER; if (this->m_flags.rx.md_recv) { // We have metadata, so send NAK with file data gaps nakPdu.initialize(direction, this->getClass(), // transmission mode this->m_history->peer_eid, // source EID (receiver) this->m_history->seq_num, // transaction sequence number this->m_cfdpManager->getLocalEidParam(), // destination EID (sender) 0, // scope start 0 // scope end ); // Compute gaps and add segments to NAK PDU U32 chunkCount = this->m_chunks->chunks.getCount(); U32 maxChunks = this->m_chunks->chunks.getMaxChunks(); U32 gapLimit = (chunkCount < maxChunks) ? maxChunks : (maxChunks - 1); // For each gap found, add it as a segment to the NAK PDU via callback struct GapComputeContext { Transaction* txn; NakPdu* nak; } gapCtx = {this, &nakPdu}; U32 gapCount = this->m_chunks->chunks.computeGaps(static_cast(gapLimit), this->m_fsize, 0, &Transaction::r2GapComputeWrapper, &gapCtx); if (!gapCount) { // No gaps left, file reception is complete this->m_flags.rx.complete = true; status = Cfdp::Status::SUCCESS; } else { // Gaps are present, send the NAK PDU status = this->m_engine->sendNak(this, nakPdu); if (status == Cfdp::Status::SUCCESS) { this->m_flags.rx.fd_nak_sent = true; this->m_cfdpManager->addSentNakSegmentRequests(this->m_chan_num, gapCount); } } } else { // Need to send NAK to request metadata PDU again // Special case: scope start/end and segment[0] all zeros requests metadata nakPdu.initialize(direction, this->getClass(), // transmission mode this->m_history->peer_eid, // source EID (receiver) this->m_history->seq_num, // transaction sequence number this->m_cfdpManager->getLocalEidParam(), // destination EID (sender) 0, // scope start (special value) 0 // scope end (special value) ); // Add special segment [0,0] to request metadata nakPdu.addSegment(0, 0); status = this->m_engine->sendNak(this, nakPdu); } return status; } Status::T Transaction::r2CalcCrcChunk() { U8 buf[R2CrcChunkSize]; FileSize count_bytes; FileSize want_offs_size; FwSizeType read_size; Os::File::Status fileStatus; Status::T ret = Cfdp::Status::SUCCESS; FileSize rx_crc_calc_bytes_per_cycle = 0; memset(buf, 0, sizeof(buf)); count_bytes = 0; // Open file for CRC calculation if needed if (ret == Cfdp::Status::SUCCESS) { if (this->m_state_data.receive.r2.rx_crc_calc_bytes == 0) { this->m_crc = CFDP::Checksum(0); // For Class 2 RX, the file was opened in WRITE mode for receiving FileData PDUs. // Now we need to READ it for CRC calculation. Close and reopen in READ mode. if (this->m_fd.isOpen()) { this->m_fd.close(); } fileStatus = this->m_fd.open(this->m_history->fnames.dst_filename.toChar(), Os::File::OPEN_READ); if (fileStatus != Os::File::OP_OK) { this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); ret = Cfdp::Status::ERROR; } else { // Reset cached position since we just reopened the file this->m_state_data.receive.cached_pos = 0; } } } // Process file in chunks if (ret == Cfdp::Status::SUCCESS) { rx_crc_calc_bytes_per_cycle = this->m_cfdpManager->getRxCrcCalcBytesPerCycleParam(); while ((ret == Cfdp::Status::SUCCESS) && (count_bytes < rx_crc_calc_bytes_per_cycle) && (this->m_state_data.receive.r2.rx_crc_calc_bytes < this->m_fsize)) { want_offs_size = this->m_state_data.receive.r2.rx_crc_calc_bytes + static_cast(sizeof(buf)); if (want_offs_size > this->m_fsize) { read_size = this->m_fsize - this->m_state_data.receive.r2.rx_crc_calc_bytes; } else { read_size = sizeof(buf); } if (this->m_state_data.receive.cached_pos != this->m_state_data.receive.r2.rx_crc_calc_bytes) { fileStatus = this->m_fd.seek(this->m_state_data.receive.r2.rx_crc_calc_bytes, Os::File::SeekType::ABSOLUTE); if (fileStatus != Os::File::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxSeekCrcFailed( this->getClass(), this->m_history->src_eid, this->m_history->seq_num, this->m_state_data.receive.r2.rx_crc_calc_bytes, fileStatus); // this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); this->m_cfdpManager->incrementFaultFileSeek(this->m_chan_num); ret = Cfdp::Status::ERROR; } } if (ret == Cfdp::Status::SUCCESS) { FwSizeType expected_read_size = read_size; fileStatus = this->m_fd.read(buf, read_size, Os::File::WaitType::WAIT); if (fileStatus != Os::File::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxReadCrcFailed( this->getClass(), this->m_history->src_eid, this->m_history->seq_num, static_cast(expected_read_size), static_cast(read_size)); this->m_engine->setTxnStatus(this, TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); this->m_cfdpManager->incrementFaultFileRead(this->m_chan_num); ret = Cfdp::Status::ERROR; } else { this->m_crc.update(buf, this->m_state_data.receive.r2.rx_crc_calc_bytes, static_cast(read_size)); this->m_state_data.receive.r2.rx_crc_calc_bytes += static_cast(read_size); this->m_state_data.receive.cached_pos = this->m_state_data.receive.r2.rx_crc_calc_bytes; count_bytes += static_cast(read_size); // Reset inactivity timer to indicate transaction is actively processing this->m_engine->armInactTimer(this); } } } } // Check final CRC if all bytes processed if (ret == Cfdp::Status::SUCCESS) { if (this->m_state_data.receive.r2.rx_crc_calc_bytes == this->m_fsize) { /* all bytes calculated, so now check */ if (this->rCheckCrc(this->m_state_data.receive.r2.eof_crc) == Cfdp::Status::SUCCESS) { /* CRC matched! We are happy */ this->m_keep = Cfdp::Keep::KEEP; /* save the file */ /* set FIN PDU status */ this->m_state_data.receive.r2.dc = FinDeliveryCode::FIN_DELIVERY_CODE_COMPLETE; this->m_state_data.receive.r2.fs = FinFileStatus::FIN_FILE_STATUS_RETAINED; } else { this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_FILE_CHECKSUM_FAILURE); } this->m_flags.com.crc_calc = true; } else { // Not all bytes processed yet, return ERROR to signal need to continue ret = Cfdp::Status::ERROR; } } return ret; } Status::T Transaction::r2SubstateSendFin() { Status::T sret; Status::T ret = Cfdp::Status::SUCCESS; if (!TxnStatusIsError(this->m_history->txn_stat) && !this->m_flags.com.crc_calc) { /* no error, and haven't checked CRC -- so start checking it */ if (this->r2CalcCrcChunk()) { ret = Cfdp::Status::ERROR; /* signal to caller to re-enter next tick */ } } if (ret != Cfdp::Status::ERROR) { sret = this->m_engine->sendFin(this, this->m_state_data.receive.r2.dc, this->m_state_data.receive.r2.fs, static_cast(TxnStatusToConditionCode(this->m_history->txn_stat))); /* Serialization error already logged in serializeAndSendPdu if ERROR returned */ this->m_state_data.receive.sub_state = RxSubState::RX_SUB_STATE_CLOSEOUT_SYNC; /* whether or not FIN send successful, ok to transition state */ if (sret != Cfdp::Status::SUCCESS) { ret = Cfdp::Status::ERROR; } } /* if no message, then try again next time */ return ret; } void Transaction::r2RecvFinAck(const Fw::Buffer& buffer) { // Deserialize ACK PDU from buffer AckPdu ack; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = ack.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { // Bad ACK PDU this->m_cfdpManager->log_WARNING_LO_FailAckPduDeserialization(this->getChannelId(), static_cast(deserStatus)); this->m_cfdpManager->incrementRecvErrors(this->m_chan_num); return; } // ACK PDU has been validated during deserialization // Got fin-ack, so time to close the state this->r2Reset(); } void Transaction::r2RecvMd(const Fw::Buffer& buffer) { Fw::String fname; Os::File::Status fileStatus; Os::FileSystem::Status fileSysStatus; bool success = true; /* it isn't an error to get another MD PDU, right? */ if (!this->m_flags.rx.md_recv) { /* NOTE: this->m_flags.rx.md_recv always 1 in R1, so this is R2 only */ /* parse the md PDU. this will overwrite the transaction's history, which contains our filename. so let's * save the filename in a local buffer so it can be used with moveFile upon successful parsing of * the md PDU */ fname = this->m_history->fnames.dst_filename; // Deserialize Metadata PDU from buffer MetadataPdu md; // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Fw::SerializeStatus deserStatus = md.deserializeFrom(sb); if (deserStatus != Fw::FW_SERIALIZE_OK) { // Bad metadata PDU this->m_cfdpManager->log_WARNING_LO_FailMetadataPduDeserialization(this->getChannelId(), static_cast(deserStatus)); return; } // PDU validation already done during deserialization this->m_engine->recvMd(this, md); /* successfully obtained md PDU */ if (this->m_flags.rx.eof_recv) { /* EOF was received, so check that md and EOF sizes match */ if (this->m_state_data.receive.r2.eof_size != this->m_fsize) { this->m_cfdpManager->log_WARNING_LO_RxEofMdSizeMismatch(this->getClass(), this->m_history->src_eid, this->m_history->seq_num, this->m_fsize, this->m_state_data.receive.r2.eof_size); this->m_cfdpManager->incrementFaultFileSizeMismatch(this->m_chan_num); this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_FILE_SIZE_ERROR); success = false; } } if (success) { /* close and rename file */ this->m_fd.close(); fileSysStatus = Os::FileSystem::moveFile(fname.toChar(), this->m_history->fnames.dst_filename.toChar()); if (fileSysStatus != Os::FileSystem::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxFileRenameFailed( this->getClass(), this->m_history->src_eid, this->m_history->seq_num, fname, this->m_history->fnames.dst_filename, fileSysStatus); this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_FILESTORE_REJECTION); this->m_cfdpManager->incrementFaultFileRename(this->m_chan_num); success = false; } else { // File was successfully renamed, open for writing fileStatus = this->m_fd.open(this->m_history->fnames.dst_filename.toChar(), Os::File::OPEN_WRITE); if (fileStatus != Os::File::OP_OK) { this->m_cfdpManager->log_WARNING_LO_RxFileReopenFailed( this->getClass(), this->m_history->src_eid, this->m_history->seq_num, this->m_history->fnames.dst_filename, fileStatus); this->r2SetFinTxnStatus(TxnStatus::TXN_STATUS_FILESTORE_REJECTION); this->m_cfdpManager->incrementFaultFileOpen(this->m_chan_num); success = false; } } if (success) { this->m_state_data.receive.cached_pos = 0; /* reset psn due to open */ this->m_flags.rx.md_recv = true; this->m_state_data.receive.r2.acknak_count = 0; /* in case part of NAK */ this->r2Complete(true); /* check for completion now that md is received */ } } } } void Transaction::rSendInactivityEvent() { this->m_cfdpManager->log_WARNING_LO_RxInactivityTimeout(this->getClass(), this->m_history->src_eid, this->m_history->seq_num); this->m_cfdpManager->incrementFaultInactivityTimer(this->m_chan_num); } // ====================================================================== // Dispatch Methods // ====================================================================== void Transaction::rDispatchRecv(const Fw::Buffer& buffer, const RSubstateDispatchTable* dispatch, StateRecvFunc fd_fn) { StateRecvFunc selected_handler; FW_ASSERT(this->m_state_data.receive.sub_state < RxSubState::RX_SUB_STATE_NUM_STATES, static_cast(this->m_state_data.receive.sub_state), static_cast(RxSubState::RX_SUB_STATE_NUM_STATES)); selected_handler = nullptr; // Peek at PDU type from buffer Cfdp::PduTypeEnum::T pduType = Cfdp::peekPduType(buffer); // Special handling for file data PDU if (pduType == Cfdp::PduTypeEnum::FILE_DATA) { /* For file data PDU, use the provided fd_fn */ if (!TxnStatusIsError(this->m_history->txn_stat)) { selected_handler = fd_fn; } } else { // Not a file-data PDU - parse as a directive PDU to get the directive code. // const_cast: Fw::SerialBuffer requires non-const U8* even for deserialization (read-only) Fw::SerialBuffer sb(const_cast(buffer.getData()), buffer.getSize()); sb.setBuffLen(buffer.getSize()); Cfdp::PduHeader header; if (header.fromSerialBuffer(sb) == Fw::FW_SERIALIZE_OK) { // Read directive code (first byte after header for directive PDUs) U8 directiveCodeByte; if (sb.deserializeTo(directiveCodeByte) == Fw::FW_SERIALIZE_OK) { FileDirective directiveCode = static_cast(directiveCodeByte); if (directiveCode < FileDirective::FILE_DIRECTIVE_INVALID_MAX) { /* The CFDP_R_SubstateDispatchTable_t is only used with file directive PDU */ if (dispatch->state[static_cast(this->m_state_data.receive.sub_state)] != nullptr) { selected_handler = dispatch->state[static_cast(this->m_state_data.receive.sub_state)] ->fdirective[static_cast(directiveCode)]; } } else { this->m_cfdpManager->incrementRecvSpurious(this->m_chan_num); this->m_cfdpManager->log_WARNING_LO_RxInvalidDirectiveCode( this->getClass(), this->m_history->src_eid, this->m_history->seq_num, directiveCodeByte, static_cast(this->m_state_data.receive.sub_state)); } } } } /* * NOTE: if no handler is selected, this will drop packets on the floor here. */ if (selected_handler != nullptr) { (this->*selected_handler)(buffer); } else { this->m_cfdpManager->incrementRecvDropped(this->m_chan_num); } } } // namespace Cfdp } // namespace Ccsds } // namespace Svc