F´ Flight Software - C/C++ Documentation
A framework for building embedded system applications to NASA flight quality standards.
Channel.cpp
Go to the documentation of this file.
1 // ======================================================================
2 // \title Channel.cpp
3 // \brief CFDP Channel operations implementation
4 //
5 // This file is a port of channel-specific functions from the following files
6 // from the NASA Core Flight System (cFS) CFDP (CF) Application, version 3.0.0,
7 // adapted for use within the F-Prime (F') framework:
8 // - cf_cfdp.c (channel processing functions)
9 // - cf_utils.c (channel transaction and resource management)
10 //
11 // ======================================================================
12 //
13 // NASA Docket No. GSC-18,447-1
14 //
15 // Copyright (c) 2019 United States Government as represented by the
16 // Administrator of the National Aeronautics and Space Administration.
17 // All Rights Reserved.
18 //
19 // Licensed under the Apache License, Version 2.0 (the "License");
20 // you may not use this file except in compliance with the License.
21 // You may obtain a copy of the License at
22 //
23 // http://www.apache.org/licenses/LICENSE-2.0
24 //
25 // Unless required by applicable law or agreed to in writing, software
26 // distributed under the License is distributed on an "AS IS" BASIS,
27 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
28 // See the License for the specific language governing permissions and
29 // limitations under the License.
30 //
31 // ======================================================================
32 
33 #include <string.h>
34 #include <new>
35 
36 #include <Fw/FPrimeBasicTypes.hpp>
37 
42 
43 namespace Svc {
44 namespace Ccsds {
45 namespace Cfdp {
46 
47 // ----------------------------------------------------------------------
48 // Construction
49 // ----------------------------------------------------------------------
50 
52  U8 channelId,
53  CfdpManager* cfdpManager,
54  Fw::MemAllocator& allocator,
55  FwEnumStoreType memId)
56  : m_engine(engine),
57  m_numCmdTx(0),
58  m_currentTxn(nullptr),
59  m_cfdpManager(cfdpManager),
60  m_tickType(0),
61  m_channelId(channelId),
62  m_flowState(Cfdp::Flow::NOT_FROZEN),
63  m_outgoingCounter(0),
64  m_transactions(nullptr),
65  m_histories(nullptr),
66  m_chunks(nullptr),
67  m_chunkMem(nullptr) {
68  FW_ASSERT(engine != nullptr);
69  FW_ASSERT(cfdpManager != nullptr);
70 
71  // Initialize queue pointers
72  for (U32 i = 0; i < QueueId::NUM; i++) {
73  m_qs[i] = nullptr;
74  }
75 
76  // Initialize command/history lists
77  for (U32 i = 0; i < static_cast<U32>(Direction::DIRECTION_NUM); i++) {
78  m_cs[i] = nullptr;
79  }
80 
81  // Initialize poll directory playback state
82  for (U32 i = 0; i < MaxPollingDirPerChan; i++) {
83  m_polldir[i].pb.busy = false;
84  m_polldir[i].pb.diropen = false;
85  m_polldir[i].pb.counted = false;
86  m_polldir[i].pb.num_ts = 0;
87  m_polldir[i].pb.pending_file = "";
88  }
89 
90  // Initialize playback structures
91  for (U32 i = 0; i < MaxCommandedPlaybackDirectoriesPerChan; i++) {
92  m_playback[i].busy = false;
93  m_playback[i].diropen = false;
94  m_playback[i].counted = false;
95  m_playback[i].num_ts = 0;
96  m_playback[i].pending_file = "";
97  }
98 
99  // Allocate and initialize per-channel resources
100  U32 j, k;
101  History* history;
102  Transaction* txn;
103  CfdpChunkWrapper* cw;
104  CListNode** list_head;
105  U32 chunk_mem_offset = 0;
106  U32 total_chunks_needed;
107 
108  // Initialize chunk configuration for this channel
109  const U32 rxChunksPerChannel[] = CFDP_CHANNEL_NUM_RX_CHUNKS_PER_TRANSACTION;
110  const U32 txChunksPerChannel[] = CFDP_CHANNEL_NUM_TX_CHUNKS_PER_TRANSACTION;
111  m_dirMaxChunks[static_cast<U32>(Direction::DIRECTION_RX)] = rxChunksPerChannel[m_channelId];
112  m_dirMaxChunks[static_cast<U32>(Direction::DIRECTION_TX)] = txChunksPerChannel[m_channelId];
113 
114  // Calculate total chunks needed for this channel
115  total_chunks_needed = 0;
116  for (k = 0; k < static_cast<U32>(Direction::DIRECTION_NUM); ++k) {
117  total_chunks_needed += m_dirMaxChunks[k] * CFDP_NUM_TRANSACTIONS_PER_CHANNEL;
118  }
119 
120  // Allocate arrays using the provided allocator
121  FwSizeType transactionsSize = CFDP_NUM_TRANSACTIONS_PER_CHANNEL * sizeof(Transaction);
122  m_transactions = static_cast<Transaction*>(allocator.allocate(memId, transactionsSize));
123  FW_ASSERT(m_transactions != nullptr);
124 
125  FwSizeType chunksSize =
127  m_chunks = static_cast<CfdpChunkWrapper*>(allocator.allocate(memId, chunksSize));
128  FW_ASSERT(m_chunks != nullptr);
129 
130  FwSizeType historiesSize = NumHistoriesPerChannel * sizeof(History);
131  m_histories = static_cast<History*>(allocator.allocate(memId, historiesSize));
132  FW_ASSERT(m_histories != nullptr);
133 
134  FwSizeType chunkMemSize = total_chunks_needed * sizeof(Chunk);
135  m_chunkMem = static_cast<Chunk*>(allocator.allocate(memId, chunkMemSize));
136  FW_ASSERT(m_chunkMem != nullptr);
137 
138  // Initialize transactions using placement new with parameterized constructor
139  cw = m_chunks;
140  for (j = 0; j < CFDP_NUM_TRANSACTIONS_PER_CHANNEL; ++j) {
141  // Construct transaction in-place with parameterized constructor
142  txn = new (&m_transactions[j]) Transaction(this, m_channelId, m_engine, m_cfdpManager);
143 
144  // Put transaction on free list
145  this->freeTransaction(txn);
146 
147  // Initialize chunk wrappers for this transaction (TX and RX)
148  for (k = 0; k < static_cast<U32>(Direction::DIRECTION_NUM); ++k, ++cw) {
149  list_head = this->getChunkListHead(static_cast<U8>(k));
150 
151  // Use placement new to construct CfdpChunkWrapper with the new class-based interface
152  new (cw) CfdpChunkWrapper(static_cast<ChunkIdx>(m_dirMaxChunks[k]), &m_chunkMem[chunk_mem_offset]);
153  chunk_mem_offset += m_dirMaxChunks[k];
155  CfdpCListInsertBack(list_head, &cw->cl_node);
156  }
157  }
158 
159  // Initialize histories using placement new (History contains Fw::String which needs proper construction)
160  for (j = 0; j < NumHistoriesPerChannel; ++j) {
161  history = new (&m_histories[j]) History(); // Use placement new with default constructor
162  CfdpCListInitNode(&history->cl_node);
163  this->insertBackInQueue(QueueId::HIST_FREE, &history->cl_node);
164  }
165 }
166 
168  // Cleanup should have been called before destruction
169  // This is enforced by Engine::~Engine()
170 }
171 
173  // Call destructors and deallocate all internal arrays
174  if (m_transactions != nullptr) {
175  // Manually call destructors since we used placement new
176  for (U32 j = 0; j < CFDP_NUM_TRANSACTIONS_PER_CHANNEL; ++j) {
177  m_transactions[j].~Transaction();
178  }
179  allocator.deallocate(memId, m_transactions);
180  m_transactions = nullptr;
181  }
182 
183  if (m_chunks != nullptr) {
184  // Manually call destructors since we used placement new
185  for (U32 j = 0; j < (CFDP_NUM_TRANSACTIONS_PER_CHANNEL * static_cast<U32>(Direction::DIRECTION_NUM)); ++j) {
186  m_chunks[j].~CfdpChunkWrapper();
187  }
188  allocator.deallocate(memId, m_chunks);
189  m_chunks = nullptr;
190  }
191 
192  if (m_histories != nullptr) {
193  // Call destructors on History objects
194  for (U32 j = 0; j < NumHistoriesPerChannel; ++j) {
195  m_histories[j].~History();
196  }
197  allocator.deallocate(memId, m_histories);
198  m_histories = nullptr;
199  }
200 
201  if (m_chunkMem != nullptr) {
202  allocator.deallocate(memId, m_chunkMem);
203  m_chunkMem = nullptr;
204  }
205 }
206 
207 // ----------------------------------------------------------------------
208 // Channel Processing
209 // ----------------------------------------------------------------------
210 
212  Transaction* txn;
213  CycleTxArgs args;
214 
215  if (m_cfdpManager->getDequeueEnabledParam(m_channelId)) {
216  args.chan = this;
217  args.ran_one = 0;
218 
219  // loop through as long as there are pending transactions, and a message buffer to send their PDUs on
220 
221  // NOTE: tick processing is higher priority than sending new filedata PDUs, so only send however many
222  // PDUs that can be sent once we get to here
223  if (!this->m_currentTxn) { // don't enter if currentTxn is set, since we need to pick up where we left off on
224  // tick processing next scheduler cycle
225 
226  // Process pending transactions until queue is empty or something runs
227  while (true) {
228  // Context for static wrapper: pass both Channel* and CycleTxArgs*
229  struct CycleTxContext {
230  Channel* channel;
231  CycleTxArgs* args;
232  } cycleTxCtx = {this, &args};
233 
234  // Attempt to run something on TXA
236 
237  // Keep going until QueueId::PEND is empty or something is run
238  if (args.ran_one || m_qs[QueueId::PEND] == nullptr) {
239  break;
240  }
241 
242  txn = container_of_cpp(m_qs[QueueId::PEND], &Transaction::m_cl_node);
243 
244  // Class 2 transactions need a chunklist for NAK processing, get one now.
245  // Class 1 transactions don't need chunks since they don't support NAKs.
246  if (txn->getClass() == Cfdp::Class::CLASS_2) {
247  if (txn->m_chunks == nullptr) {
248  txn->m_chunks = this->findUnusedChunks(Direction::DIRECTION_TX);
249  }
250  if (txn->m_chunks == nullptr) {
251  // Chunklist unavailable - EVR already emitted by Engine
252  // Leave transaction pending until a chunklist is available.
253  break;
254  }
255  }
256 
257  m_engine->armInactTimer(txn);
258  this->moveTransaction(txn, QueueId::TXA);
259  }
260  }
261 
262  // in case the loop exited due to no message buffers, clear it and start from the top next time
263  this->m_currentTxn = nullptr;
264  }
265 }
266 
268  bool reset = true;
269 
270  void (Transaction::* fns[static_cast<U8>(CfdpTickType::CFDP_TICK_TYPE_NUM_TYPES)])(I32*) = {
273 
274  FW_ASSERT(m_tickType < static_cast<U8>(CfdpTickType::CFDP_TICK_TYPE_NUM_TYPES), m_tickType);
275 
276  for (; m_tickType < static_cast<U8>(CfdpTickType::CFDP_TICK_TYPE_NUM_TYPES); ++m_tickType) {
277  TickArgs args = {this, fns[m_tickType], 0, 0};
278 
279  // Safety bound: retry loop should not exceed the number of transactions in the queue
280  // Each retry processes one transaction that may request continuation
281  constexpr U32 maxRetries = MaxSimultaneousRx + MaxCommandedPlaybackFilesPerChan +
284 
285  for (U32 retry = 0; retry < maxRetries; ++retry) {
286  args.cont = 0;
287 
288  // Context for static wrapper: pass both Channel* and TickArgs*
289  struct TickContext {
290  Channel* channel;
291  TickArgs* args;
292  } tickCtx = {this, &args};
293 
294  CfdpCListTraverse(m_qs[qs[m_tickType]], &Channel::doTickWrapper, &tickCtx);
295 
296  if (args.early_exit) {
297  // early exit means we ran out of available outgoing messages this scheduler cycle.
298  // If current tick type is NAK response, then reset tick type. It would be
299  // bad to let NAK response starve out RX or TXW ticks on the next cycle.
300  //
301  // If RX ticks use up all available messages, then we pick up where we left
302  // off on the next cycle. (This causes some RX tick counts to be missed,
303  // but that's ok. Precise timing isn't required.)
304  //
305  // This scheme allows the following priority for use of outgoing messages:
306  //
307  // RX state messages
308  // TXW state messages
309  // NAK response (could be many)
310  //
311  // New file data on TXA
312  if (m_tickType != static_cast<U8>(CfdpTickType::CFDP_TICK_TYPE_TXW_NAK)) {
313  reset = false;
314  }
315 
316  break;
317  }
318 
319  if (!args.cont) {
320  break; // No continuation requested, exit retry loop
321  }
322  }
323 
324  if (!reset) {
325  break;
326  }
327  }
328 
329  if (reset) {
330  m_tickType = static_cast<U8>(CfdpTickType::CFDP_TICK_TYPE_RX); // reset tick type
331  }
332 }
333 
335  U32 i;
336  U8 playback_count = 0;
337 
338  for (i = 0; i < MaxCommandedPlaybackDirectoriesPerChan; ++i) {
339  this->processPlaybackDirectory(&m_playback[i]);
340  // Count active playback operations
341  if (m_playback[i].busy) {
342  playback_count++;
343  }
344  }
345 
346  // Update playback counter telemetry
347  Cfdp::ChannelTelemetry& tlm = m_engine->getChannelTelemetryRef(m_channelId);
348  tlm.set_playbackCounter(playback_count);
349 }
350 
352  CfdpPollDir* pd;
353  U32 i;
354  U8 poll_count = 0;
355  Status::T status;
356 
357  for (i = 0; i < MaxPollingDirPerChan; ++i) {
358  pd = &m_polldir[i];
359 
360  if (pd->enabled) {
361  poll_count++;
362 
363  if ((pd->pb.busy == false) && (pd->pb.num_ts == 0)) {
364  if ((pd->intervalTimer.getStatus() != Timer::Status::RUNNING) && (pd->intervalSec > 0)) {
365  // timer was not set, so set it now
367  } else if (pd->intervalTimer.getStatus() == Timer::Status::EXPIRED) {
368  // the timer has expired
369  status = m_engine->playbackDirInitiate(&pd->pb, pd->srcDir, pd->dstDir, pd->cfdpClass,
370  Cfdp::Keep::DELETE, m_channelId, pd->priority, pd->destEid);
371  if (status != Cfdp::Status::SUCCESS) {
372  // error occurred in playback directory, so reset the timer
373  // an event is sent when initiating playback directory so there is no reason to
374  // to have another here
376  }
377  } else {
378  pd->intervalTimer.run();
379  }
380  } else {
381  // playback is active, so step it
382  this->processPlaybackDirectory(&pd->pb);
383  }
384  }
385  }
386 
387  // Update poll counter telemetry
388  Cfdp::ChannelTelemetry& tlm = m_engine->getChannelTelemetryRef(m_channelId);
389  tlm.set_pollCounter(poll_count);
390 }
391 
392 // ----------------------------------------------------------------------
393 // Transaction Management
394 // ----------------------------------------------------------------------
395 
397  CListNode* node;
398  Transaction* txn;
399  QueueId::T q_index; // initialized below in if
400 
401  if (m_qs[QueueId::FREE]) {
402  node = m_qs[QueueId::FREE];
403  txn = container_of_cpp(node, &Transaction::m_cl_node);
404 
405  this->removeFromQueue(QueueId::FREE, &txn->m_cl_node);
406 
407  // now that a transaction is acquired, must also acquire a history slot to go along with it
408  if (m_qs[QueueId::HIST_FREE]) {
409  q_index = QueueId::HIST_FREE;
410  } else {
411  // no free history, so take the oldest one from the channel's history queue
412  FW_ASSERT(m_qs[QueueId::HIST]);
413  q_index = QueueId::HIST;
414  }
415 
416  txn->m_history = container_of_cpp(m_qs[q_index], &History::cl_node);
417 
418  this->removeFromQueue(q_index, &txn->m_history->cl_node);
419 
420  // Reset all history fields to initial state (matches constructor zero-init)
421  // This is necessary when recycling from HIST queue to clear stale data
422  txn->m_history->txn_stat = TxnStatus::TXN_STATUS_UNDEFINED; // Critical: prevents error status inheritance
423  txn->m_history->src_eid = 0;
424  txn->m_history->peer_eid = 0;
425  txn->m_history->seq_num = 0;
426  txn->m_history->fnames.src_filename = "";
427  txn->m_history->fnames.dst_filename = "";
428  // Note: cl_node is managed by queue operations (already handled by removeFromQueue)
429  // Note: dir is explicitly set below (already handled)
430 
431  // Indicate that this was freshly pulled from the free list
432  // notably this state is distinguishable from items still on the free list
433  txn->m_state = TxnState::TXN_STATE_INIT;
434 
435  // Clear the FREE tag now that this transaction has been taken off the FREE
436  // list. freeTransaction() marks q_index == QueueId::FREE for anything sitting
437  // on the free list; leaving that tag set on an acquired-but-not-yet-enqueued
438  // transaction would break the invariant relied on by
439  // Engine::finishTransaction()'s double-free guard (a live txn must never look
440  // FREE). The caller (startRxTransaction / txFileInitiate) will assign the real
441  // queue via insertSortPrio()/direct assignment; until then PEND (0) is the
442  // neutral, not-on-FREE-list default that matches reset()'s zeroed m_flags.
443  txn->m_flags.com.q_index = QueueId::PEND;
444 
445  txn->m_history->dir = direction;
446  txn->m_chan = this; // Set channel pointer
447 
448  // Re-initialize the linked list node to clear stale pointers from FREE list
449  CfdpCListInitNode(&txn->m_cl_node);
450  } else {
451  txn = nullptr;
452  }
453 
454  return txn;
455 }
456 
458  // need to find transaction by sequence number. It will either be the active transaction (front of Q_PEND),
459  // or on Q_TX or Q_RX. Once a transaction moves to history, then it's done.
460  //
461  // Let's put QueueId::RX up front, because most RX packets will be file data PDUs
462  CfdpTraverseTransSeqArg ctx = {transaction_sequence_number, src_eid, nullptr};
463  CListNode* ptrs[] = {m_qs[QueueId::RX], m_qs[QueueId::PEND], m_qs[QueueId::TXA], m_qs[QueueId::TXW]};
464  Transaction* ret = nullptr;
465 
466  for (CListNode* head : ptrs) {
468  if (ctx.txn) {
469  ret = ctx.txn;
470  break;
471  }
472  }
473 
474  return ret;
475 }
476 
478  I32 counter = 0;
479 
480  // Context for static wrapper
481  struct TraverseAllContext {
483  void* userContext;
484  I32* counter;
485  } ctx = {fn, context, &counter};
486 
487  for (I32 queueidx = QueueId::PEND; queueidx <= QueueId::RX; ++queueidx) {
489  }
490 
491  return counter;
492 }
493 
495  this->removeFromQueue(QueueId::HIST, &history->cl_node);
496  this->insertBackInQueue(QueueId::HIST_FREE, &history->cl_node);
497 }
498 
499 // ----------------------------------------------------------------------
500 // Transaction Queue Management
501 // ----------------------------------------------------------------------
502 
504  FW_ASSERT(txn);
505  CfdpCListRemove(&m_qs[txn->m_flags.com.q_index], &txn->m_cl_node);
506 
507  // Update queue depth telemetry
508  Cfdp::ChannelTelemetry& tlm = m_engine->getChannelTelemetryRef(m_channelId);
509  switch (txn->m_flags.com.q_index) {
510  case Cfdp::QueueId::FREE:
511 
512  tlm.set_queueFree(static_cast<U16>(tlm.get_queueFree() - 1));
513  break;
514  case Cfdp::QueueId::TXA:
515 
516  tlm.set_queueTxActive(static_cast<U16>(tlm.get_queueTxActive() - 1));
517  break;
518  case Cfdp::QueueId::TXW:
519 
520  tlm.set_queueTxWaiting(static_cast<U16>(tlm.get_queueTxWaiting() - 1));
521  break;
522  case Cfdp::QueueId::RX:
523 
524  tlm.set_queueRx(static_cast<U16>(tlm.get_queueRx() - 1));
525  break;
526  case Cfdp::QueueId::HIST:
527 
528  tlm.set_queueHistory(static_cast<U16>(tlm.get_queueHistory() - 1));
529  break;
530  case Cfdp::QueueId::PEND:
532  // PEND and HIST_FREE queues are not tracked in telemetry
533  break;
534  default:
535  FW_ASSERT(0, txn->m_flags.com.q_index);
536  }
537 }
538 
540  FW_ASSERT(txn);
541  Cfdp::ChannelTelemetry& tlm = m_engine->getChannelTelemetryRef(m_channelId);
542 
543  // Decrement old queue
544  CfdpCListRemove(&m_qs[txn->m_flags.com.q_index], &txn->m_cl_node);
545  switch (txn->m_flags.com.q_index) {
546  case Cfdp::QueueId::FREE:
547 
548  tlm.set_queueFree(static_cast<U16>(tlm.get_queueFree() - 1));
549  break;
550  case Cfdp::QueueId::TXA:
551 
552  tlm.set_queueTxActive(static_cast<U16>(tlm.get_queueTxActive() - 1));
553  break;
554  case Cfdp::QueueId::TXW:
555 
556  tlm.set_queueTxWaiting(static_cast<U16>(tlm.get_queueTxWaiting() - 1));
557  break;
558  case Cfdp::QueueId::RX:
559 
560  tlm.set_queueRx(static_cast<U16>(tlm.get_queueRx() - 1));
561  break;
562  case Cfdp::QueueId::HIST:
563 
564  tlm.set_queueHistory(static_cast<U16>(tlm.get_queueHistory() - 1));
565  break;
566  case Cfdp::QueueId::PEND:
568  // PEND and HIST_FREE queues are not tracked in telemetry
569  break;
570  default:
571  FW_ASSERT(0, txn->m_flags.com.q_index);
572  }
573 
574  // Increment new queue
575  CfdpCListInsertBack(&m_qs[queue], &txn->m_cl_node);
576  txn->m_flags.com.q_index = queue;
577  switch (queue) {
578  case Cfdp::QueueId::FREE:
579  tlm.set_queueFree(static_cast<U16>(tlm.get_queueFree() + 1));
580  break;
581  case Cfdp::QueueId::TXA:
582  tlm.set_queueTxActive(static_cast<U16>(tlm.get_queueTxActive() + 1));
583  break;
584  case Cfdp::QueueId::TXW:
585  tlm.set_queueTxWaiting(static_cast<U16>(tlm.get_queueTxWaiting() + 1));
586  break;
587  case Cfdp::QueueId::RX:
588  tlm.set_queueRx(static_cast<U16>(tlm.get_queueRx() + 1));
589  break;
590  case Cfdp::QueueId::HIST:
591  tlm.set_queueHistory(static_cast<U16>(tlm.get_queueHistory() + 1));
592  break;
593  case Cfdp::QueueId::PEND:
595  // PEND and HIST_FREE queues are not tracked in telemetry
596  break;
597  default:
598  FW_ASSERT(0, queue);
599  }
600 }
601 
603  // Reset transaction to default state (preserves channel context)
604  txn->reset();
605 
606  // Initialize the linked list node for the FREE queue
607  CfdpCListInitNode(&txn->m_cl_node);
608  this->insertBackInQueue(QueueId::FREE, &txn->m_cl_node);
609 
610  // Mark the transaction as residing on the FREE list. insertBackInQueue() only
611  // performs the list insertion (unlike insertSortPrio(), which also updates
612  // q_index), and txn->reset() zeroes m_flags so q_index would otherwise be left
613  // at 0 (== QueueId::PEND). Without this, a freed transaction is never tagged
614  // FREE, and Engine::finishTransaction()'s double-free guard
615  // (q_index == QueueId::FREE) can never fire. Setting it here upholds the
616  // invariant: "a transaction on the FREE list has q_index == FREE".
617  txn->m_flags.com.q_index = QueueId::FREE;
618 }
619 
621  CListNode** chunklist_head;
622  QueueId::T hist_destq;
623 
624  // File should have been closed by the state machine, but if
625  // it still hanging open at this point, close it now so its not leaked.
626  // This is not normal/expected so log it if this happens.
627  if (true == txn->m_fd.isOpen()) {
628  this->m_cfdpManager->log_WARNING_LO_DanglingFileHandleClosed(txn->getChannelId(), txn->m_history->seq_num);
629  txn->m_fd.close();
630  }
631 
632  this->dequeueTransaction(txn); // this makes it "float" (not in any queue)
633 
634  // this should always be
635  if (txn->m_history != nullptr) {
636  if (txn->m_chunks != nullptr) {
637  chunklist_head = this->getChunkListHead(static_cast<U8>(txn->m_history->dir));
638  if (chunklist_head != nullptr) {
639  // Reset chunk list to clear stale data from previous transaction
640  txn->m_chunks->chunks.reset();
641  CfdpCListInsertBack(chunklist_head, &txn->m_chunks->cl_node);
642  txn->m_chunks = nullptr;
643  }
644  }
645 
646  if (txn->m_flags.com.keep_history) {
647  // move transaction history to history queue
648  hist_destq = QueueId::HIST;
649  } else {
650  hist_destq = QueueId::HIST_FREE;
651  }
652  this->insertBackInQueue(hist_destq, &txn->m_history->cl_node);
653  txn->m_history = nullptr;
654  }
655 
656  // this wipes it and puts it back onto the list to be found by
657  // Channel::findUnusedTransaction(). Need to preserve the chan_num
658  // and keep it associated with this channel, though.
659  this->freeTransaction(txn);
660 }
661 
663  bool insert_back = false;
664 
665  FW_ASSERT(txn);
666 
667  // look for proper position on PEND queue for this transaction.
668  // This is a simple priority sort.
669 
670  if (!m_qs[queue]) {
671  // list is empty, so just insert
672  insert_back = true;
673  } else {
674  CfdpTraversePriorityArg arg = {nullptr, txn->getPriority()};
676  if (arg.txn) {
677  this->insertAfterInQueue(queue, &arg.txn->m_cl_node, &txn->m_cl_node);
678  } else {
679  insert_back = true;
680  }
681  }
682 
683  if (insert_back) {
684  this->insertBackInQueue(queue, &txn->m_cl_node);
685  }
686  txn->m_flags.com.q_index = queue;
687 }
688 
689 // ----------------------------------------------------------------------
690 // Channel State Management
691 // ----------------------------------------------------------------------
692 
694  FW_ASSERT(m_numCmdTx); // sanity check
695  --m_numCmdTx;
696 }
697 
699  // Done with this TX transaction
700  if (this->m_currentTxn == txn) {
701  this->m_currentTxn = nullptr;
702  }
703 }
704 
706  this->m_currentTxn = txn;
707 }
708 
709 // ----------------------------------------------------------------------
710 // Resource Management
711 // ----------------------------------------------------------------------
712 
714  CListNode** result;
715 
716  if (direction < static_cast<U32>(Direction::DIRECTION_NUM)) {
717  result = &m_cs[direction];
718  } else {
719  result = nullptr;
720  }
721 
722  return result;
723 }
724 
726  CfdpChunkWrapper* ret = nullptr;
727  CListNode* node;
728  CListNode** chunklist_head;
729 
730  chunklist_head = this->getChunkListHead(static_cast<U8>(dir));
731 
732  // this should never be null
733  FW_ASSERT(chunklist_head);
734 
735  if (*chunklist_head != nullptr) {
736  node = CfdpCListPop(chunklist_head);
737  if (node != nullptr) {
739  }
740  }
741 
742  return ret;
743 }
744 
745 // ----------------------------------------------------------------------
746 // Private helper methods
747 // ----------------------------------------------------------------------
748 
749 void Channel::processPlaybackDirectory(Playback* pb) {
750  Transaction* txn;
752  Os::Directory::Status status;
753 
754  // either there's no transaction (first one) or the last one was finished, so check for a new one
755 
756  while (pb->diropen && (pb->num_ts < NumTransactionsPerPlayback)) {
757  if (pb->pending_file.length() == 0) {
758  status = pb->dir.read(path);
759  if (status == Os::Directory::NO_MORE_FILES) {
760  // Directory playback complete - success reported via TxFileTransferCompleted EVR
761  pb->dir.close();
762  pb->diropen = false;
763  break;
764  }
765  if (status != Os::Directory::OP_OK) {
766  // Directory read error - emit EVR and close playback
768  static_cast<I32>(status));
769  pb->dir.close();
770  pb->diropen = false;
771  break;
772  }
773 
774  pb->pending_file = path;
775  } else {
777  if (txn == nullptr) {
778  // while not expected this can certainly happen, because
779  // rx transactions consume in these as well.
780  // should not need to do anything special, will come back next tick
781  break;
782  }
783 
784  // Append file name to source/destination folders
785  txn->m_history->fnames.src_filename = pb->fnames.src_filename;
786  txn->m_history->fnames.src_filename += "/";
787  txn->m_history->fnames.src_filename += pb->pending_file;
788 
789  txn->m_history->fnames.dst_filename = pb->fnames.dst_filename;
790  txn->m_history->fnames.dst_filename += "/";
791  txn->m_history->fnames.dst_filename += pb->pending_file;
792 
793  m_engine->txFileInitiate(txn, pb->cfdp_class, pb->keep, m_channelId, pb->priority, pb->dest_id);
794 
795  txn->m_pb = pb;
796  ++pb->num_ts;
797 
798  pb->pending_file = ""; // continue reading dir
799  }
800  }
801 
802  if (!pb->diropen && !pb->num_ts) {
803  // the directory has been exhausted, and there are no more active transactions
804  // for this playback -- so mark it as not busy
805  pb->busy = false;
806  }
807 }
808 
809 void Channel::updatePollPbCounted(Playback* pb, I32 up, U8* counter) {
810  if (pb->counted != up) {
811  // only handle on state change
812  pb->counted = !!up; // !! ensure 0 or 1, should be optimized out
813 
814  if (up) {
815  ++*counter;
816  } else {
817  FW_ASSERT(*counter); // sanity check it isn't zero
818  --*counter;
819  }
820  }
821 }
822 
824  CycleTxArgs* args = static_cast<CycleTxArgs*>(context);
825  Transaction* txn = container_of_cpp(node, &Transaction::m_cl_node);
826  CListTraverseStatus ret = CLIST_TRAVERSE_EXIT; // default option is exit traversal
827 
828  if (txn->m_flags.com.suspended) {
829  ret = CLIST_TRAVERSE_CONTINUE; // suspended, so move on to next
830  } else {
831  FW_ASSERT(txn->m_flags.com.q_index == QueueId::TXA); // huh?
832 
833  // if no more messages, then chan->m_currentTxn will be set.
834  // If the transaction sent the last filedata PDU and EOF, it will move itself
835  // off the active queue. Run until either of these occur.
836  while (!this->m_currentTxn && txn->m_flags.com.q_index == QueueId::TXA) {
837  m_engine->dispatchTx(txn);
838  }
839 
840  args->ran_one = 1;
841  }
842 
843  return ret;
844 }
845 
847  CListTraverseStatus ret =
848  CLIST_TRAVERSE_CONTINUE; // CLIST_TRAVERSE_CONTINUE means don't tick one, keep looking for currentTxn
849  TickArgs* args = static_cast<TickArgs*>(context);
850  Transaction* txn = container_of_cpp(node, &Transaction::m_cl_node);
851  if (!this->m_currentTxn || (this->m_currentTxn == txn)) {
852  // found where we left off, so clear that and move on
853  this->m_currentTxn = nullptr;
854  if (!txn->m_flags.com.suspended) {
855  (txn->*args->fn)(&args->cont);
856  }
857 
858  // if this->m_currentTxn was set to not-nullptr above, then exit early
859  // NOTE: if channel is frozen, then tick processing won't have been entered.
860  // so there is no need to check it here
861  if (this->m_currentTxn) {
862  ret = CLIST_TRAVERSE_EXIT;
863  args->early_exit = true;
864  }
865  }
866 
867  return ret; // don't tick one, keep looking for currentTxn
868 }
869 
872  return &m_transactions[index];
873 }
874 
877  return &m_histories[index];
878 }
879 
880 // ----------------------------------------------------------------------
881 // Static callback wrapper implementations
882 // ----------------------------------------------------------------------
883 
885  struct CycleTxContext {
886  Channel* channel;
887  CycleTxArgs* args;
888  };
889  CycleTxContext* ctx = static_cast<CycleTxContext*>(context);
890  return ctx->channel->cycleTxFirstActive(node, ctx->args);
891 }
892 
894  struct TickContext {
895  Channel* channel;
896  TickArgs* args;
897  };
898  TickContext* ctx = static_cast<TickContext*>(context);
899  return ctx->channel->doTick(node, ctx->args);
900 }
901 
903  struct TraverseAllContext {
905  void* userContext;
906  I32* counter;
907  };
908  TraverseAllContext* ctx = static_cast<TraverseAllContext*>(context);
909  Transaction* txn = container_of_cpp(node, &Transaction::m_cl_node);
910  ctx->fn(txn, ctx->userContext);
911  ++(*ctx->counter);
913 }
914 
915 } // namespace Cfdp
916 } // namespace Ccsds
917 } // namespace Svc
CListNode cl_node
for connection to a CList
Definition: Types.hpp:246
U8 getPriority() const
Get transaction priority.
CfdpTxnFilenames fnames
file names associated with this history entry
Definition: Types.hpp:245
CFDP Protocol Engine.
Definition: Engine.hpp:89
void moveTransaction(Transaction *txn, QueueId::T queue)
Move a transaction from one queue to another.
Definition: Channel.cpp:539
Fw::String dstDir
path to destination dir
Definition: Types.hpp:309
CFDP Channel class.
Definition: Channel.hpp:56
virtual void * allocate(const FwEnumStoreType identifier, FwSizeType &size, bool &recoverable, FwSizeType alignment=alignof(std::max_align_t))=0
Structure for use with the Channel::doTick() function.
Definition: Engine.hpp:70
~Channel()
Destruct a Channel.
Definition: Channel.cpp:167
Argument structure for use with CfdpCListTraverseR()
Definition: Utils.hpp:73
Transaction * txn
output transaction pointer
Definition: Utils.hpp:54
History * getHistory(U32 index)
Get a history by index (for testing)
Definition: Channel.cpp:875
bool early_exit
early exit result
Definition: Engine.hpp:73
CFDP Transaction state machine class.
PlatformSizeType FwSizeType
void removeFromQueue(QueueId::T queueidx, CListNode *node)
Remove a node from a channel queue.
Definition: Channel.hpp:520
I32 FwEnumStoreType
CListTraverseStatus
Traverse status for circular list operations.
Definition: Clist.hpp:45
void(Transaction::* fn)(I32 *)
member function pointer
Definition: Engine.hpp:72
U16 num_ts
number of transactions
Definition: Types.hpp:282
void(*)(Transaction *txn, void *context) CfdpTraverseAllTransactionsFunc
Callback function type for use with Channel::traverseAllTransactions()
Definition: Types.hpp:421
void setCurrentTxn(const Transaction *txn)
Set current transaction.
Definition: Channel.cpp:705
State assigned to a newly allocated transaction object.
U32 EntityId
Entity id size.
first one on this list is active
void set_queueTxActive(U16 queueTxActive)
Set member queueTxActive.
void sTickNak(I32 *cont)
Perform NAK response for TX transactions.
I32 cont
if 1, then re-traverse the list
Definition: Engine.hpp:74
I32 traverseAllTransactions(CfdpTraverseAllTransactionsFunc fn, void *context)
Traverses all transactions on all active queues and performs an operation on them.
Definition: Channel.cpp:477
U8 priority
priority to use when placing transactions on the pending queue
Definition: Types.hpp:304
Wrapper around a CfdpChunkList object.
Definition: Types.hpp:260
void reset()
Reset the chunk list to empty state.
Definition: Chunk.cpp:58
Fw::StringTemplate< MaxFilePathSize > pending_file
Definition: Types.hpp:285
CfdpChunkList chunks
Chunk list for gap tracking.
Definition: Types.hpp:261
File will be deleted after the CFDP transaction.
Definition: KeepEnumAc.hpp:38
T
The raw enum type.
CFDP class 2 - Reliable transfer (Acknowledged)
Definition: ClassEnumAc.hpp:44
Channel * chan
channel object
Definition: Engine.hpp:63
static CListTraverseStatus prioritySearchCallback(CListNode *node, void *context)
Static callback for priority search.
Definition: Utils.cpp:89
Continue traversing the list.
Definition: Clist.hpp:46
void rTick(I32 *cont)
Perform tick (time-based) processing for R transactions.
void set_pollCounter(U8 pollCounter)
Set member pollCounter.
void processPollingDirectories()
Process all polling directories for this channel.
Definition: Channel.cpp:351
U32 intervalSec
number of seconds to wait before trying a new directory
Definition: Types.hpp:302
CFDP Playback entry.
Definition: Types.hpp:278
I32 ran_one
should be set to 1 if a transaction was cycled
Definition: Engine.hpp:64
CListTraverseStatus cycleTxFirstActive(CListNode *node, void *context)
Traverse callback for cycling the first active transaction.
Definition: Channel.cpp:823
void set_queueTxWaiting(U16 queueTxWaiting)
Set member queueTxWaiting.
U16 get_queueTxActive() const
Get member queueTxActive.
U32 TransactionSeq
transaction sequence number size
U8 getChannelId() const
Get channel ID.
Fw::Enabled getDequeueEnabledParam(U8 channelIndex)
bool keep_history
whether history should be preserved during recycle
Definition: Types.hpp:365
Status getStatus(void)
Get the status of a CFDP timer.
Definition: Timer.cpp:37
void set_queueRx(U16 queueRx)
Set member queueRx.
static constexpr U32 CFDP_NUM_TRANSACTIONS_PER_CHANNEL
Maximum possible number of transactions that may exist on a single CFDP channel.
Definition: Types.hpp:42
void cycleTx()
Cycle the TX side of this channel.
Definition: Channel.cpp:211
#define CFDP_CHANNEL_NUM_TX_CHUNKS_PER_TRANSACTION
TX chunks per transaction (per channel)
Definition: CfdpCfg.hpp:58
U16 get_queueTxWaiting() const
Get member queueTxWaiting.
static CListTraverseStatus cycleTxFirstActiveWrapper(CListNode *node, void *context)
Static wrapper for cycleTxFirstActive callback.
Definition: Channel.cpp:884
void close() override
Close directory.
Definition: Directory.cpp:73
Argument structure for use with CList_Traverse()
Definition: Utils.hpp:51
void insertAfterInQueue(QueueId::T queueidx, CListNode *start, CListNode *after)
Insert a node after another in a channel queue.
Definition: Channel.hpp:524
Structure for use with the Channel::cycleTx() function.
Definition: Engine.hpp:62
Status read(char *fileNameBuffer, FwSizeType buffSize) override
Get next filename from directory stream.
Definition: Directory.cpp:54
void armInactTimer(Transaction *txn)
Arm the inactivity timer for a transaction.
Definition: Engine.cpp:109
void reset()
Reset transaction to default state.
Channel(Engine *engine, U8 channelId, CfdpManager *cfdpManager, Fw::MemAllocator &allocator, FwEnumStoreType memId)
Construct a Channel.
Definition: Channel.cpp:51
void clearCurrentIfMatch(Transaction *txn)
Check if current transaction matches and clear if so.
Definition: Channel.cpp:698
void tickTransactions()
Tick all transactions on this channel.
Definition: Channel.cpp:267
U16 get_queueFree() const
Get member queueFree.
Direction dir
direction of this history entry
Definition: Types.hpp:247
void freeTransaction(Transaction *txn)
Frees and resets a transaction and returns it for later use.
Definition: Channel.cpp:602
void dequeueTransaction(Transaction *txn)
Free a transaction from the queue it&#39;s on.
Definition: Channel.cpp:503
Playback pb
State of the current playback requests.
Definition: Types.hpp:299
void dispatchTx(Transaction *txn)
Dispatch TX state machine for a transaction.
Definition: Engine.cpp:169
void setTimer(U32 timerDuration)
Initialize a CFDP timer and start its execution.
Definition: Timer.cpp:27
void txFileInitiate(Transaction *txn, Class::T cfdp_class, Keep::T keep, U8 chan, U8 priority, EntityId dest_id)
Initiate a file transfer transaction.
Definition: Engine.cpp:727
Stop traversing the list.
Definition: Clist.hpp:47
void close() override
close the file, if not opened then do nothing
Definition: File.cpp:90
Directory stream has no more files.
Definition: Directory.hpp:27
static CListTraverseStatus findBySequenceNumberCallback(CListNode *node, void *context)
Static callback for finding transaction by sequence number.
Definition: Utils.cpp:74
void CfdpCListTraverse(CListNode *start, CListFunc fn, void *context)
Traverse the entire list, calling the given function on all nodes.
Definition: Clist.cpp:139
Status::T playbackDirInitiate(Playback *pb, const Fw::String &src_filename, const Fw::String &dst_filename, Class::T cfdp_class, Keep::T keep, U8 chan, U8 priority, EntityId dest_id)
Initiate playback of a directory.
Definition: Engine.cpp:821
Fw::String srcDir
path to source dir
Definition: Types.hpp:308
CListNode * CfdpCListPop(CListNode **head)
Remove the first node from a list and return it.
Definition: Clist.cpp:89
void decrementCmdTxCounter()
Decrement the command TX counter for this channel.
Definition: Channel.cpp:693
void CfdpCListInsertBack(CListNode **head, CListNode *node)
Insert the given node into the back of a list.
Definition: Clist.cpp:69
Transaction * txn
OUT: holds value of transaction with which to call CfdpCListInsertAfter on.
Definition: Utils.hpp:74
static CListTraverseStatus doTickWrapper(CListNode *node, void *context)
Static wrapper for doTick callback.
Definition: Channel.cpp:893
uint8_t U8
8-bit unsigned integer
Definition: BasicTypes.h:54
void processPlaybackDirectories()
Process all playback directories for this channel.
Definition: Channel.cpp:334
CFDP operation has been successful.
void insertSortPrio(Transaction *txn, QueueId::T queue)
Insert a transaction into a priority sorted transaction queue.
Definition: Channel.cpp:662
void sTick(I32 *cont)
Perform tick (time-based) processing for S transactions.
void set_playbackCounter(U8 playbackCounter)
Set member playbackCounter.
CfdpTxnFilenames fnames
Definition: Types.hpp:281
Cfdp::ChannelTelemetry & getChannelTelemetryRef(U8 channelId)
Get reference to channel telemetry for Channel class.
Definition: Engine.cpp:1182
Transaction * findUnusedTransaction(Direction direction)
Find an unused transaction on this channel.
Definition: Channel.cpp:396
Class::T getClass() const
Get transaction class (CLASS_1 or CLASS_2)
TransactionSeq seq_num
transaction identifier, stays constant for entire transfer
Definition: Types.hpp:251
void cleanup(Fw::MemAllocator &allocator, FwEnumStoreType memId)
Clean up dynamically allocated resources.
Definition: Channel.cpp:172
Memory Allocation base class.
void log_WARNING_LO_PlaybackDirReadFailed(const Fw::StringBase &directory, I32 status) const
Log event PlaybackDirReadFailed.
Transaction * getTransaction(U32 index)
Get a transaction by index (for testing)
Definition: Channel.cpp:870
void insertBackInQueue(QueueId::T queueidx, CListNode *node)
Insert a node at the back of a channel queue.
Definition: Channel.hpp:528
Class::T cfdpClass
the CFDP class to send
Definition: Types.hpp:305
constexpr Container * container_of_cpp(Member *member_ptr, Member Container::*member)
Obtains a pointer to the parent structure.
Definition: Clist.hpp:79
void CfdpCListRemove(CListNode **head, CListNode *node)
Remove the given node from the list.
Definition: Clist.cpp:102
Operation was successful.
Definition: Directory.hpp:22
void run(void)
Runs a one second increment of the CFDP timers.
Definition: Timer.cpp:41
Fw::Enabled enabled
Enabled flag.
Definition: Types.hpp:311
CListTraverseStatus doTick(CListNode *node, void *context)
Traverse callback for ticking a transaction.
Definition: Channel.cpp:846
Circular linked list node structure.
Definition: Clist.hpp:66
Transaction * findTransactionBySequenceNumber(TransactionSeq transaction_sequence_number, EntityId src_eid)
Finds an active transaction by sequence number.
Definition: Channel.cpp:457
Timer intervalTimer
Timer object used to poll the directory.
Definition: Types.hpp:300
Structure for the telemetry array of CFDP channels.
RateGroupDivider component implementation.
virtual SizeType length() const
Get the length of the string.
CfdpFlagsCommon com
applies to all transactions
Definition: Types.hpp:402
TxnStatus txn_stat
final status of operation
Definition: Types.hpp:248
void resetHistory(History *history)
Returns a history structure back to its unused state.
Definition: Channel.cpp:494
EntityId src_eid
the source eid of the transaction
Definition: Types.hpp:249
void log_WARNING_LO_DanglingFileHandleClosed(U8 channelId, U32 transactionSeq) const
Log event DanglingFileHandleClosed.
virtual void deallocate(const FwEnumStoreType identifier, void *ptr)=0
#define CFDP_CHANNEL_NUM_RX_CHUNKS_PER_TRANSACTION
RX chunks per transaction (per channel)
Definition: CfdpCfg.hpp:37
Directory poll entry.
Definition: Types.hpp:298
EntityId destEid
destination entity id
Definition: Types.hpp:306
U8 q_index
Q index this is in.
Definition: Types.hpp:359
void CfdpCListTraverseR(CListNode *end, CListFunc fn, void *context)
Reverse list traversal, starting from end, calling given function on all nodes.
Definition: Clist.cpp:173
void set_queueHistory(U16 queueHistory)
Set member queueHistory.
CListNode ** getChunkListHead(U8 direction)
Gets the head of the chunk list for this channel + direction.
Definition: Channel.cpp:713
U16 get_queueHistory() const
Get member queueHistory.
CfdpChunkWrapper * findUnusedChunks(Direction dir)
Find unused chunks for this channel.
Definition: Channel.cpp:725
T
The raw enum type.
void set_queueFree(U16 queueFree)
Set member queueFree.
void recycleTransaction(Transaction *txn)
Recover resources associated with a transaction.
Definition: Channel.cpp:620
Direction
Direction identifier.
Definition: Types.hpp:154
#define FW_ASSERT(...)
Definition: Assert.hpp:14
bool isOpen() const
determine if the file is open
Definition: File.cpp:98
CListNode cl_node
Circular list node for pooling.
Definition: Types.hpp:262
static CListTraverseStatus traverseAllTransactionsWrapper(CListNode *node, void *context)
Static wrapper for traverseAllTransactions callback.
Definition: Channel.cpp:902
Pairs an offset with a size to identify a specific piece of a file.
Definition: Chunk.hpp:49
EntityId peer_eid
peer_eid is always the "other guy", same src_eid for RX
Definition: Types.hpp:250
void CfdpCListInitNode(CListNode *node)
Initialize a clist node.
Definition: Clist.cpp:43
CFDP History entry.
Definition: Types.hpp:244