F´ Flight Software - C/C++ Documentation
A framework for building embedded system applications to NASA flight quality standards.
ComQueue.cpp
Go to the documentation of this file.
1 // ======================================================================
2 // \title ComQueue.cpp
3 // \author vbai
4 // \brief cpp file for ComQueue component implementation class
5 // ======================================================================
6 
7 #include <Fw/Com/ComPacket.hpp>
8 #include <Fw/Types/Assert.hpp>
10 #include <type_traits>
11 #include "Fw/Types/BasicTypes.hpp"
12 
13 namespace Svc {
14 
15 // ----------------------------------------------------------------------
16 // Construction, initialization, and destruction
17 // ----------------------------------------------------------------------
18 
19 using FwUnsignedIndexType = std::make_unsigned<FwIndexType>::type;
20 
21 ComQueue ::QueueConfigurationTable ::QueueConfigurationTable() {
22  static_assert(static_cast<FwUnsignedIndexType>(std::numeric_limits<FwIndexType>::max()) >=
23  FW_NUM_ARRAY_ELEMENTS(this->entries),
24  "Number of entries must fit into FwIndexType");
25  for (FwIndexType i = 0; i < static_cast<FwIndexType>(FW_NUM_ARRAY_ELEMENTS(this->entries)); i++) {
26  this->entries[i].priority = 0;
27  this->entries[i].depth = 0;
28  this->entries[i].mode = Types::QUEUE_FIFO;
29  this->entries[i].overflowMode = Types::QUEUE_DROP_NEWEST;
30  }
31 }
32 
33 ComQueue ::ComQueue(const char* const compName)
34  : ComQueueComponentBase(compName),
35  m_state(WAITING),
36  m_buffer_state(OWNED),
37  m_allocationId(static_cast<FwEnumStoreType>(-1)),
38  m_allocator(nullptr),
39  m_allocation(nullptr) {
40  // Initialize throttles to "off"
41  for (FwIndexType i = 0; i < TOTAL_PORT_COUNT; i++) {
42  this->m_throttle[i] = false;
43  }
44 
45  static_assert(TOTAL_PORT_COUNT >= 1, "ComQueue must have more than one port");
46 }
47 
49 
51  // Deallocate memory ignoring error conditions
52  if ((this->m_allocator != nullptr) && (this->m_allocation != nullptr)) {
53  this->m_allocator->deallocate(this->m_allocationId, this->m_allocation);
54  }
55 }
56 
58  FwEnumStoreType allocationId,
59  Fw::MemAllocator& allocator) {
60  FwIndexType currentPriorityIndex = 0;
61  FwSizeType totalAllocation = 0;
62 
63  // Store/initialize allocator members
64  this->m_allocator = &allocator;
65  this->m_allocationId = allocationId;
66  this->m_allocation = nullptr;
67 
68  // Initializes the sorted queue metadata list in priority (sorted) order. This is accomplished by walking the
69  // priority values in priority order from 0 to TOTAL_PORT_COUNT. At each priory value, the supplied queue
70  // configuration table is walked and any entry matching the current priority values is used to add queue metadata to
71  // the prioritized list. This results in priority-sorted queue metadata objects that index back into the unsorted
72  // queue data structures.
73  //
74  // The total allocation size is tracked for passing to the allocation call and is a summation of
75  // (depth * message size) for each prioritized metadata object of (depth * message size)
76  for (FwIndexType currentPriority = 0; currentPriority < TOTAL_PORT_COUNT; currentPriority++) {
77  // Walk each queue configuration entry and add them into the prioritized metadata list when matching the current
78  // priority value
79  for (FwIndexType entryIndex = 0;
80  entryIndex < static_cast<FwIndexType>(FW_NUM_ARRAY_ELEMENTS(queueConfig.entries)); entryIndex++) {
81  // Check for valid configuration entry
82  FW_ASSERT(queueConfig.entries[entryIndex].priority < TOTAL_PORT_COUNT,
83  static_cast<FwAssertArgType>(queueConfig.entries[entryIndex].priority),
84  static_cast<FwAssertArgType>(TOTAL_PORT_COUNT), static_cast<FwAssertArgType>(entryIndex));
85  // A zero depth is the default-constructed value and divides by zero in the overflow check below
86  FW_ASSERT(queueConfig.entries[entryIndex].depth > 0, static_cast<FwAssertArgType>(entryIndex));
87 
88  if (currentPriority == queueConfig.entries[entryIndex].priority) {
89  // Set up the queue metadata object in order to track priority, depth, index into the queue list of the
90  // backing queue object, and message size. Both index and message size are calculated where priority and
91  // depth are copied from the configuration object.
92  QueueMetadata& entry = this->m_prioritizedList[currentPriorityIndex];
93  entry.priority = queueConfig.entries[entryIndex].priority;
94  entry.depth = queueConfig.entries[entryIndex].depth;
95  entry.mode = queueConfig.entries[entryIndex].mode;
96  entry.overflowMode = queueConfig.entries[entryIndex].overflowMode;
97  entry.index = entryIndex;
98  // Message size is determined by the type of object being stored, which in turn is determined by the
99  // index of the entry. Those lower than COM_PORT_COUNT are Fw::ComBuffers and those larger Fw::Buffer.
100  entry.msgSize = (entryIndex < COM_PORT_COUNT) ? static_cast<FwSizeType>(Fw::ComBuffer::SERIALIZED_SIZE)
101  : static_cast<FwSizeType>(Fw::Buffer::SERIALIZED_SIZE);
102  // Overflow checks
103  FW_ASSERT((std::numeric_limits<FwSizeType>::max() / entry.depth) >= entry.msgSize,
104  static_cast<FwAssertArgType>(entry.depth), static_cast<FwAssertArgType>(entry.msgSize));
105  FW_ASSERT(std::numeric_limits<FwSizeType>::max() - (entry.depth * entry.msgSize) >= totalAllocation);
106  totalAllocation += entry.depth * entry.msgSize;
107  currentPriorityIndex++;
108  }
109  }
110  }
111  // Allocate a single chunk of memory from the memory allocator. Memory recover is neither needed nor used.
112  bool recoverable = false;
113  FwSizeType actualAllocation = totalAllocation;
114  this->m_allocation = this->m_allocator->allocate(this->m_allocationId, actualAllocation, recoverable);
115  FW_ASSERT(this->m_allocation != nullptr);
116  FW_ASSERT(actualAllocation >= totalAllocation, static_cast<FwAssertArgType>(actualAllocation),
117  static_cast<FwAssertArgType>(totalAllocation));
118 
119  // Each of the backing queue objects must be supplied memory to store the queued messages. These data regions are
120  // sub-portions of the total allocated data. This memory is passed out by looping through each queue in prioritized
121  // order and passing out the memory to each queue's setup method.
122  FwSizeType allocationOffset = 0;
123  for (FwIndexType i = 0; i < TOTAL_PORT_COUNT; i++) {
124  // Get current queue's allocation size and safety check the values
125  FwSizeType allocationSize = this->m_prioritizedList[i].depth * this->m_prioritizedList[i].msgSize;
126  FW_ASSERT(this->m_prioritizedList[i].index < static_cast<FwIndexType>(FW_NUM_ARRAY_ELEMENTS(this->m_queues)),
127  static_cast<FwAssertArgType>(this->m_prioritizedList[i].index));
128  FW_ASSERT((allocationSize + allocationOffset) <= totalAllocation, static_cast<FwAssertArgType>(allocationSize),
129  static_cast<FwAssertArgType>(allocationOffset), static_cast<FwAssertArgType>(totalAllocation));
130 
131  // Setup queue's memory allocation, depth, and message size. Setup is skipped for a depth 0 queue
132  if (allocationSize > 0) {
133  this->m_queues[this->m_prioritizedList[i].index].setup(
134  reinterpret_cast<U8*>(this->m_allocation) + allocationOffset, allocationSize,
135  this->m_prioritizedList[i].depth, this->m_prioritizedList[i].msgSize, this->m_prioritizedList[i].mode,
136  this->m_prioritizedList[i].overflowMode);
137  }
138  allocationOffset += allocationSize;
139  }
140  // Safety check that all memory was used as expected
141  FW_ASSERT(allocationOffset == totalAllocation, static_cast<FwAssertArgType>(allocationOffset),
142  static_cast<FwAssertArgType>(totalAllocation));
143 }
144 
145 // ----------------------------------------------------------------------
146 // Handler implementations for commands
147 // ----------------------------------------------------------------------
148 
149 void ComQueue ::FLUSH_QUEUE_cmdHandler(FwOpcodeType opCode,
150  U32 cmdSeq,
151  const Svc::QueueType& queueType,
152  FwIndexType index) {
153  // Acquire the queue that we need to drain
154  FwIndexType queueIndex = this->getQueueNum(queueType, index);
155 
156  // Validate queue index
157  if (queueIndex < 0 || queueIndex >= TOTAL_PORT_COUNT) {
158  this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::VALIDATION_ERROR);
159  return;
160  }
161  FW_ASSERT(queueIndex >= 0 && queueIndex < TOTAL_PORT_COUNT, static_cast<FwAssertArgType>(queueIndex));
162 
163  this->drainQueue(queueIndex);
164  this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::OK);
165 }
166 
167 void ComQueue ::FLUSH_ALL_QUEUES_cmdHandler(FwOpcodeType opCode, U32 cmdSeq) {
168  for (FwIndexType i = 0; i < TOTAL_PORT_COUNT; i++) {
169  this->drainQueue(i);
170  }
171  this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::OK);
172 }
173 
174 void ComQueue::SET_QUEUE_PRIORITY_cmdHandler(FwOpcodeType opCode,
175  U32 cmdSeq,
176  const Svc::QueueType& queueType,
177  FwIndexType index,
178  FwIndexType newPriority) {
179  // Acquire the queue we are to reprioritize
180  FwIndexType queueIndex = this->getQueueNum(queueType, index);
181 
182  // Validate queue index
183  if (queueIndex < 0 || queueIndex >= TOTAL_PORT_COUNT) {
184  this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::VALIDATION_ERROR);
185  return;
186  }
187 
188  // Validate priority range
189  if (newPriority < 0 || newPriority >= TOTAL_PORT_COUNT) {
190  this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::VALIDATION_ERROR);
191  return;
192  }
193 
194  // Find our queue in the prioritized list & update the priority
195  for (FwIndexType prioIndex = 0; prioIndex < TOTAL_PORT_COUNT; prioIndex++) {
196  // Each entry must reference a valid queue index
197  FW_ASSERT(m_prioritizedList[prioIndex].index >= 0 && m_prioritizedList[prioIndex].index < TOTAL_PORT_COUNT,
198  static_cast<FwAssertArgType>(m_prioritizedList[prioIndex].index));
199  // If the port based index matches, then update
200  if (m_prioritizedList[prioIndex].index == queueIndex) {
201  m_prioritizedList[prioIndex].priority = newPriority;
202  break; // Since we shouldn't find more than one queue at this port index
203  }
204  }
205 
206  // Re-sort the prioritized list to maintain priority ordering
207  // Using simple bubble sort since TOTAL_PORT_COUNT is typically small
208  for (FwIndexType i = 0; i < TOTAL_PORT_COUNT - 1; i++) {
209  for (FwIndexType j = 0; (j < TOTAL_PORT_COUNT - i - 1) && (j < TOTAL_PORT_COUNT - 1); j++) {
210  if (m_prioritizedList[j].priority > m_prioritizedList[j + 1].priority) {
211  // Swap metadata
212  QueueMetadata temp = m_prioritizedList[j];
213  m_prioritizedList[j] = m_prioritizedList[j + 1];
214  m_prioritizedList[j + 1] = temp;
215  }
216  }
217  }
218 
219  // Emit event for successful priority change
220  this->log_ACTIVITY_HI_QueuePriorityChanged(queueType, index, newPriority);
221 
222  // Send command response
223  this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::OK);
224 }
225 
226 // ----------------------------------------------------------------------
227 // Handler implementations for user-defined typed input ports
228 // ----------------------------------------------------------------------
229 
230 void ComQueue::comPacketQueueIn_handler(const FwIndexType portNum, Fw::ComBuffer& data, U32 context) {
231  // Ensure that the port number of comPacketQueueIn is consistent with the expectation
232  FW_ASSERT(portNum >= 0 && portNum < COM_PORT_COUNT, static_cast<FwAssertArgType>(portNum));
233  (void)this->enqueue(portNum, data);
234 }
235 
236 void ComQueue::bufferQueueIn_handler(const FwIndexType portNum, Fw::Buffer& fwBuffer) {
237  FW_ASSERT(std::numeric_limits<FwIndexType>::max() - COM_PORT_COUNT > portNum);
238  const FwIndexType queueNum = static_cast<FwIndexType>(portNum + COM_PORT_COUNT);
239  // Ensure that the port number of bufferQueueIn is consistent with the expectation
240  FW_ASSERT(portNum >= 0 && portNum < BUFFER_PORT_COUNT, static_cast<FwAssertArgType>(portNum));
241  FW_ASSERT(queueNum < TOTAL_PORT_COUNT);
242  bool success = this->enqueue(queueNum, fwBuffer);
243  if (!success) {
244  this->bufferReturnOut_out(portNum, fwBuffer);
245  }
246 }
247 
248 void ComQueue::comStatusIn_handler(const FwIndexType portNum, Fw::Success& condition) {
249  switch (this->m_state) {
250  // On success, the queue should be processed. On failure, the component should still wait.
251  case WAITING:
252  if (condition.e == Fw::Success::SUCCESS) {
253  this->m_state = READY;
254  this->processQueue();
255  // A message may or may not be sent. Thus, READY or WAITING are acceptable final states.
256  FW_ASSERT((this->m_state == WAITING || this->m_state == READY),
257  static_cast<FwAssertArgType>(this->m_state));
258  } else {
259  this->m_state = WAITING;
260  }
261  break;
262  // Both READY and unknown states should not be possible at this point. To receive a status message we must be
263  // one of the WAITING or RETRY states.
264  default:
265  FW_ASSERT(false, static_cast<FwAssertArgType>(this->m_state));
266  break;
267  }
268 }
269 
270 void ComQueue::run_handler(const FwIndexType portNum, U32 context) {
271  // Downlink the high-water marks for the Fw::ComBuffer array types
272  ComQueueDepth comQueueDepth;
273  FW_ASSERT(comQueueDepth.SIZE <= COM_PORT_COUNT, static_cast<FwAssertArgType>(comQueueDepth.SIZE));
274  for (U32 i = 0; i < comQueueDepth.SIZE; i++) {
275  comQueueDepth[i] = static_cast<U32>(this->m_queues[i].get_high_water_mark());
276  this->m_queues[i].clear_high_water_mark();
277  }
278  this->tlmWrite_comQueueDepth(comQueueDepth);
279 
280  // Downlink the high-water marks for the Fw::Buffer array types
281  BuffQueueDepth buffQueueDepth;
282  FW_ASSERT((buffQueueDepth.SIZE + COM_PORT_COUNT) <= TOTAL_PORT_COUNT,
283  static_cast<FwAssertArgType>(buffQueueDepth.SIZE));
284  for (U32 i = 0; i < buffQueueDepth.SIZE; i++) {
285  buffQueueDepth[i] = static_cast<U32>(this->m_queues[i + COM_PORT_COUNT].get_high_water_mark());
286  this->m_queues[i + COM_PORT_COUNT].clear_high_water_mark();
287  }
288  this->tlmWrite_buffQueueDepth(buffQueueDepth);
289 }
290 
291 void ComQueue ::dataReturnIn_handler(FwIndexType portNum, Fw::Buffer& data, const ComCfg::FrameContext& context) {
292  static_assert(std::numeric_limits<FwIndexType>::is_signed, "FwIndexType must be signed");
293  // This handler runs on the returning caller's thread: take ownership atomically
294  const BufferState previousState = this->m_buffer_state.exchange(OWNED);
295  FW_ASSERT(previousState == UNOWNED, static_cast<FwAssertArgType>(previousState));
296  // For the buffer queues, the index of the queue is portNum offset by COM_PORT_COUNT since
297  // the first COM_PORT_COUNT queues are for ComBuffer. So we have for buffer queues:
298  // queueNum = portNum + COM_PORT_COUNT
299  // Since queueNum is used as APID, we can retrieve the original portNum like such:
300  FwIndexType bufferReturnPortNum = static_cast<FwIndexType>(context.get_comQueueIndex() - ComQueue::COM_PORT_COUNT);
301  // Failing this assert means that context.apid was modified since ComQueue set it, which should not happen
302  FW_ASSERT(bufferReturnPortNum < BUFFER_PORT_COUNT, static_cast<FwAssertArgType>(bufferReturnPortNum));
303  if (bufferReturnPortNum >= 0) {
304  // It is a coding error not to connect the associated bufferReturnOut port for each dataReturnIn port
305  FW_ASSERT(this->isConnected_bufferReturnOut_OutputPort(bufferReturnPortNum),
306  static_cast<FwAssertArgType>(bufferReturnPortNum));
307  // If this is a buffer port, return the buffer to the BufferDownlink
308  this->bufferReturnOut_out(bufferReturnPortNum, data);
309  }
310 }
311 
312 // ----------------------------------------------------------------------
313 // Hook implementations for typed async input ports
314 // ----------------------------------------------------------------------
315 
316 void ComQueue::bufferQueueIn_overflowHook(FwIndexType portNum, Fw::Buffer& fwBuffer) {
317  FW_ASSERT(portNum >= 0 && portNum < BUFFER_PORT_COUNT, static_cast<FwAssertArgType>(portNum));
318  this->bufferReturnOut_out(portNum, fwBuffer);
319 }
320 
321 // ----------------------------------------------------------------------
322 // Private helper methods
323 // ----------------------------------------------------------------------
324 
325 bool ComQueue::enqueue(const FwIndexType queueNum, const Fw::ComBuffer& data) {
326  // Enqueue the given message onto the matching queue. When no space is available then emit the queue overflow event,
327  // set the appropriate throttle, and move on. Will assert if passed a message for a depth 0 queue.
328  FW_ASSERT(queueNum >= 0 && queueNum < COM_PORT_COUNT, static_cast<FwAssertArgType>(queueNum));
329 
330  const Fw::SerializeStatus status = this->m_queues[queueNum].enqueue(data);
331  return this->handleEnqueueStatus(queueNum, QueueType::COM_QUEUE, queueNum, false, status);
332 }
333 
334 bool ComQueue::enqueue(const FwIndexType queueNum, const Fw::Buffer& data) {
335  // Enqueue the given message onto the matching queue. When no space is available then emit the queue overflow event,
336  // set the appropriate throttle, and move on. Will assert if passed a message for a depth 0 queue.
337  FW_ASSERT(queueNum >= COM_PORT_COUNT && queueNum < TOTAL_PORT_COUNT, static_cast<FwAssertArgType>(queueNum));
338  const FwIndexType portNum = static_cast<FwIndexType>(queueNum - COM_PORT_COUNT);
339 
340  // For buffer queues with DROP_OLDEST, check if the queue is full before enqueuing.
341  // If full, dequeue the oldest entry first so we can return buffer ownership before
342  // Queue::enqueue() silently discards it via rotate. This prevents buffer-pool leaks.
343  bool preEmptiveOverflow = false;
344  Types::Queue& queue = this->m_queues[queueNum];
345  for (FwIndexType i = 0; i < TOTAL_PORT_COUNT; i++) {
346  if (this->m_prioritizedList[i].index == queueNum &&
347  this->m_prioritizedList[i].overflowMode == Types::QUEUE_DROP_OLDEST &&
348  queue.getQueueSize() >= this->m_prioritizedList[i].depth) {
349  // Queue is full and will drop oldest; remove the front entry to return ownership.
350  // popFront() always removes from the front (oldest) regardless of queue mode,
351  // matching the rotate-based removal that Queue::enqueue() uses for DROP_OLDEST.
352  Fw::Buffer droppedBuffer;
353  Fw::SerializeStatus dequeueStatus = queue.popFront(droppedBuffer);
354  FW_ASSERT(dequeueStatus == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(dequeueStatus));
355  this->bufferReturnOut_out(portNum, droppedBuffer);
356  preEmptiveOverflow = true;
357  break;
358  }
359  }
360 
361  const Fw::SerializeStatus status = this->m_queues[queueNum].enqueue(data);
362  return this->handleEnqueueStatus(queueNum, QueueType::BUFFER_QUEUE, portNum, preEmptiveOverflow, status);
363 }
364 
365 bool ComQueue::handleEnqueueStatus(const FwIndexType queueNum,
366  QueueType queueType,
367  const FwIndexType portNum,
368  const bool preEmptiveOverflow,
369  const Fw::SerializeStatus status) {
370  if (preEmptiveOverflow || status == Fw::FW_SERIALIZE_NO_ROOM_LEFT ||
372  if (!this->m_throttle[queueNum]) {
373  this->log_WARNING_HI_QueueOverflow(queueType, portNum);
374  this->m_throttle[queueNum] = true;
375  }
376  }
377 
378  // When the component is already in READY state process the queue to send out the next available message immediately
379  if (this->m_state == READY) {
380  this->processQueue();
381  }
382 
383  // Check if the buffer was accepted or must be returned
384  return status != Fw::FW_SERIALIZE_NO_ROOM_LEFT;
385 }
386 
387 void ComQueue::sendComBuffer(Fw::ComBuffer& comBuffer, FwIndexType queueIndex) {
388  FW_ASSERT(this->m_state == READY);
389  Fw::Buffer outBuffer(comBuffer.getBuffAddr(), static_cast<Fw::Buffer::SizeType>(comBuffer.getSize()));
390 
391  // Context value is used to determine what to do when the buffer returns on the dataReturnIn port
392  ComCfg::FrameContext context;
393  FwPacketDescriptorType descriptor = 0;
394  Fw::SerializeStatus status = comBuffer.deserializeTo(descriptor);
395  FW_ASSERT(status == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(status));
396  context.set_apid(static_cast<ComCfg::Apid::T>(descriptor));
397  context.set_comQueueIndex(queueIndex);
398  const BufferState previousState = this->m_buffer_state.exchange(UNOWNED);
399  FW_ASSERT(previousState == OWNED, static_cast<FwAssertArgType>(previousState));
400  this->dataOut_out(0, outBuffer, context);
401  // Set state to WAITING for the status to come back
402  this->m_state = WAITING;
403 }
404 
405 void ComQueue::sendBuffer(Fw::Buffer& buffer, FwIndexType queueIndex) {
406  // Retry buffer expected to be cleared as we are either transferring ownership or have already deallocated it.
407  FW_ASSERT(this->m_state == READY);
408 
409  // Context value is used to determine what to do when the buffer returns on the dataReturnIn port
410  ComCfg::FrameContext context;
411  FwPacketDescriptorType descriptor;
412  Fw::SerializeStatus status = buffer.getDeserializer().deserializeTo(descriptor);
413  FW_ASSERT(status == Fw::FW_SERIALIZE_OK, static_cast<FwAssertArgType>(status));
414  context.set_apid(static_cast<ComCfg::Apid::T>(descriptor));
415  context.set_comQueueIndex(queueIndex);
416  const BufferState previousState = this->m_buffer_state.exchange(UNOWNED);
417  FW_ASSERT(previousState == OWNED, static_cast<FwAssertArgType>(previousState));
418  this->dataOut_out(0, buffer, context);
419  // Set state to WAITING for the status to come back
420  this->m_state = WAITING;
421 }
422 
423 void ComQueue::drainQueue(FwIndexType index) {
424  FW_ASSERT(index >= 0 && index < TOTAL_PORT_COUNT, static_cast<FwAssertArgType>(index));
425  Types::Queue& queue = this->m_queues[index];
426 
427  // Read all messages from the queue and discard them
429  const FwSizeType available = queue.getQueueSize();
430  for (FwSizeType i = 0; (i < available) && (status == Fw::FW_SERIALIZE_OK); i++) {
431  if (index < COM_PORT_COUNT) {
432  // Dequeueing deserializes the persisted Fw::ComBuffer from the queue's storage
433  Fw::ComBuffer comBuffer;
434  status = queue.dequeue(comBuffer);
435  } else {
436  // For buffer queues, if the buffer requires ownership return, return it via the bufferReturnOut port
437  // Dequeueing deserializes the persisted Fw::Buffer from the queue's storage
438  Fw::Buffer buffer;
439  status = queue.dequeue(buffer);
440  this->bufferReturnOut_out(static_cast<FwIndexType>(index - COM_PORT_COUNT), buffer);
441  }
442  }
443 }
444 
445 void ComQueue::processQueue() {
446  FwIndexType priorityIndex = 0;
447  FwIndexType sendPriority = 0;
448  // Check that we are in the appropriate state
449  FW_ASSERT(this->m_state == READY);
450 
451  // Walk all the queues in priority order. Send the first message that is available in priority order. No balancing
452  // is done within this loop.
453  for (priorityIndex = 0; priorityIndex < TOTAL_PORT_COUNT; priorityIndex++) {
454  QueueMetadata& entry = this->m_prioritizedList[priorityIndex];
455  Types::Queue& queue = this->m_queues[entry.index];
456 
457  // Continue onto next prioritized queue if there is no items in the current queue
458  if (queue.getQueueSize() == 0) {
459  continue;
460  }
461 
462  // Send out the message based on the type
463  if (entry.index < COM_PORT_COUNT) {
464  // Dequeue deserializes the persisted Fw::ComBuffer from the queue's storage
465  FW_ASSERT(this->m_buffer_state.load() == OWNED);
466  auto dequeue_status = queue.dequeue(this->m_dequeued_com_buffer);
468  static_cast<FwAssertArgType>(dequeue_status));
469  this->sendComBuffer(this->m_dequeued_com_buffer, entry.index);
470  } else {
471  Fw::Buffer buffer;
472  auto dequeue_status = queue.dequeue(buffer);
474  static_cast<FwAssertArgType>(dequeue_status));
475  this->sendBuffer(buffer, entry.index);
476  }
477 
478  // Update the throttle and the index that was just sent
479  this->m_throttle[entry.index] = false;
480 
481  // Priority used in the next loop
482  sendPriority = entry.priority;
483  break;
484  }
485 
486  // Starting on the priority entry after the one dispatched and continuing through the end of the set of entries that
487  // share the same priority, rotate those entries such that the currently dispatched queue is last and the rest are
488  // shifted up by one. This effectively round-robins the queues of the same priority.
489  for (priorityIndex++;
490  priorityIndex < TOTAL_PORT_COUNT && (this->m_prioritizedList[priorityIndex].priority == sendPriority);
491  priorityIndex++) {
492  // Swap the previous entry with this one.
493  QueueMetadata temp = this->m_prioritizedList[priorityIndex];
494  this->m_prioritizedList[priorityIndex] = this->m_prioritizedList[priorityIndex - 1];
495  this->m_prioritizedList[priorityIndex - 1] = temp;
496  }
497 }
498 
499 FwIndexType ComQueue::getQueueNum(Svc::QueueType queueType, FwIndexType portNum) {
500  // Acquire the queue that we need to drain
501  return static_cast<FwIndexType>(portNum + ((queueType == QueueType::COM_QUEUE) ? 0 : COM_PORT_COUNT));
502 }
503 } // end namespace Svc
Serialization/Deserialization operation was successful.
void cmdResponse_out(FwOpcodeType opCode, U32 cmdSeq, Fw::CmdResponse response)
Emit command response.
U16 FwPacketDescriptorType
The width of packet descriptors when they are serialized by the framework.
virtual void * allocate(const FwEnumStoreType identifier, FwSizeType &size, bool &recoverable, FwSizeType alignment=alignof(std::max_align_t))=0
void dataOut_out(FwIndexType portNum, Fw::Buffer &data, const ComCfg::FrameContext &context) const
Invoke output port dataOut.
FwIdType FwOpcodeType
The type of a command opcode.
std::make_unsigned< FwIndexType >::type FwUnsignedIndexType
Definition: ComQueue.cpp:19
Representing success.
PlatformSizeType FwSizeType
void tlmWrite_buffQueueDepth(const Svc::BuffQueueDepth &arg, Fw::Time _tlmTime=Fw::Time()) const
void configure(const QueueConfigurationTable &queueConfig, FwEnumStoreType allocationId, Fw::MemAllocator &allocator)
Definition: ComQueue.cpp:57
I32 FwEnumStoreType
Serialization succeeded, but deleted old data.
QueueConfigurationEntry entries[TOTAL_PORT_COUNT]
Definition: ComQueue.hpp:76
configuration table for each queue
Definition: ComQueue.hpp:75
static const FwIndexType TOTAL_PORT_COUNT
Total count of input buffer ports and thus total queues.
Definition: ComQueue.hpp:42
ComQueue(const char *const compName)
Definition: ComQueue.cpp:33
Serializable::SizeType getSize() const override
Get current buffer size.
SerializeStatus deserializeTo(U8 &val, Endianness mode=Endianness::BIG) override
Deserialize an 8-bit unsigned integer value.
void setup(U8 *const storage, const FwSizeType storage_size, const FwSizeType depth, const FwSizeType message_size, const QueueMode mode=QUEUE_FIFO, const QueueOverflowMode overflow_mode=QUEUE_DROP_NEWEST)
setup the queue object to setup storage
Definition: Queue.cpp:17
No room left in the buffer to serialize data.
void set_apid(ComCfg::Apid::T apid)
Set member apid.
void log_ACTIVITY_HI_QueuePriorityChanged(const Svc::QueueType &queueType, FwIndexType indexType, FwIndexType newPriority) const
void clear_high_water_mark()
Definition: Queue.cpp:198
void cleanup()
Definition: ComQueue.cpp:50
static const FwIndexType BUFFER_PORT_COUNT
Definition: ComQueue.hpp:37
static const FwIndexType COM_PORT_COUNT
< Count of Fw::Com input ports and thus Fw::Com queues
Definition: ComQueue.hpp:34
An enumeration of queue data types.
FwIndexType get_comQueueIndex() const
Get member comQueueIndex.
SerializeStatus
forward declaration for string
ExternalSerializeBufferWithMemberCopy getDeserializer()
Definition: Buffer.cpp:165
void log_WARNING_HI_QueueOverflow(const Svc::QueueType &queueType, FwIndexType index) const
FwIndexType priority
Priority of the queue [0, TOTAL_PORT_COUNT)
Definition: ComQueue.hpp:61
Fw::SerializeStatus enqueue(const U8 *const message, const FwSizeType size)
pushes a fixed-size message onto the queue
Definition: Queue.cpp:36
U8 * getBuffAddr()
Get buffer address for data filling (non-const version)
FwSizeType depth
Depth of the queue [0, infinity)
Definition: ComQueue.hpp:60
void bufferReturnOut_out(FwIndexType portNum, Fw::Buffer &fwBuffer) const
Invoke output port bufferReturnOut.
Command successfully executed.
Memory Allocation base class.
void set_comQueueIndex(FwIndexType comQueueIndex)
Set member comQueueIndex.
Types::QueueOverflowMode overflowMode
Overflow handling mode (DROP_NEWEST or DROP_OLDEST)
Definition: ComQueue.hpp:63
First-In-First-Out: dequeue from front.
Definition: Queue.hpp:26
enum T e
The raw enum value.
FwSizeType get_high_water_mark() const
Definition: Queue.cpp:193
PlatformIndexType FwIndexType
Drop the newest (incoming) message on overflow.
Definition: Queue.hpp:34
FwSizeType SizeType
The size type for a buffer - for backwards compatibility.
Definition: Buffer.hpp:67
Types::QueueMode mode
Queue mode (FIFO or LIFO)
Definition: ComQueue.hpp:62
C++ header for working with basic fprime types.
#define FW_NUM_ARRAY_ELEMENTS(a)
number of elements in an array
Definition: BasicTypes.h:94
Type used to pass context info between components during framing/deframing.
bool isConnected_bufferReturnOut_OutputPort(FwIndexType portNum) const
void tlmWrite_comQueueDepth(const Svc::ComQueueDepth &arg, Fw::Time _tlmTime=Fw::Time()) const
Command failed validation.
RateGroupDivider component implementation.
virtual void deallocate(const FwEnumStoreType identifier, void *ptr)=0
Drop the oldest (front) message on overflow.
Definition: Queue.hpp:35
Fw::SerializeStatus popFront(U8 *const message, const FwSizeType size)
removes and returns the oldest (front) message regardless of queue mode
Definition: Queue.cpp:162
FwSizeType getQueueSize() const
Definition: Queue.cpp:202
Fw::SerializeStatus dequeue(U8 *const message, const FwSizeType size)
pops a fixed-size message off the queue
Definition: Queue.cpp:98
#define FW_ASSERT(...)
Definition: Assert.hpp:14
Success/Failure.
Size of Fw::Buffer when serialized.
Definition: Buffer.hpp:70
Auto-generated base for ComQueue component.