Skip to content

ComQueue queue ordering, drop mode, & command to reprioritize #4674

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Willmac16 wants to merge 26 commits into
base: devel
Choose a base branch
from
Open

ComQueue queue ordering, drop mode, & command to reprioritize #4674

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
7fea326
Add compile-time FIFO/LIFO and drop oldest/newest configuration to Co…
claude Dec 27, 2025
a14624d
Add comprehensive unit tests for Queue and CircularBuffer
claude Dec 27, 2025
cc8de5e
Add build-ut/ to .gitignore
claude Dec 27, 2025
093dfd4
Add comprehensive unit tests for ComQueue queue/overflow modes
claude Dec 27, 2025
249d58b
Add runtime queue priority configuration command
claude Dec 27, 2025
3cc8d9c
Use Queue class enum types instead of ComQueue enums
claude Dec 27, 2025
4d8ca15
Rearrange cladue additions
Willmac16 Jan 30, 2026
55f8caf
Port Utils/Types Cmake to new register_fprime_* format
Willmac16 Jan 30, 2026
8f9974b
Add DISCARDED_EXISTING to Fw::SerializeStatus
Willmac16 Jan 30, 2026
01f99e0
Queue now returns FW_SERIALIZE_DISCARDED_EXISTING if in DROP_OLDEST &…
Willmac16 Jan 30, 2026
5c0530e
Redo ComQueue Reprioritize Indexing
Willmac16 Jan 30, 2026
df2368f
Remove "the the"
Willmac16 Jan 30, 2026
52222c9
Revert build-ut gitignore
Willmac16 Jan 30, 2026
56d8f4b
Put the command handlers back where they were
Willmac16 Jan 30, 2026
fc2e329
Check & event upon the new queue return code
Willmac16 Feb 2, 2026
fc3cdd3
Update docs & rename getUnifiedQueueIndex to getQueueNum to match var
Willmac16 Feb 2, 2026
3093d14
More docs updates
Willmac16 Feb 2, 2026
9430e48
Fix tweak to getQueueNum ternary
Willmac16 Feb 2, 2026
07628d6
Correct Types & Alignment
Willmac16 Feb 3, 2026
e3405b0
Type corrections in the cpp
Willmac16 Feb 3, 2026
ca8093a
Validate commanded priority is non-negative & switch UT priority to
Willmac16 Feb 3, 2026
02cf048
Don't comflate priority index & queue index + avoid static casting
Willmac16 Feb 3, 2026
b8aaeb5
Test for negative command args & ensure queues are sorted post
Willmac16 Feb 3, 2026
6fccfc9
Spelling
Willmac16 Feb 3, 2026
b8aaef9
Add TOTAL_PORT_COUNT static assert
Willmac16 Feb 5, 2026
00f04d6
formatting
LeStarch Feb 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 9 additions & 8 deletions Fw/Types/Serializable.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,15 @@ namespace Fw {

class StringBase; //!< forward declaration for string
typedef enum {
FW_SERIALIZE_OK, //!< Serialization/Deserialization operation was successful
FW_SERIALIZE_FORMAT_ERROR, //!< Data was the wrong format (e.g. wrong packet type)
FW_SERIALIZE_NO_ROOM_LEFT, //!< No room left in the buffer to serialize data
FW_DESERIALIZE_BUFFER_EMPTY, //!< Deserialization buffer was empty when trying to read more data
FW_DESERIALIZE_FORMAT_ERROR, //!< Deserialization data had incorrect values (unexpected data types)
FW_DESERIALIZE_SIZE_MISMATCH, //!< Data was left in the buffer, but not enough to deserialize
FW_DESERIALIZE_TYPE_MISMATCH, //!< Deserialized type ID didn't match
FW_DESERIALIZE_IMMUTABLE, //!< Attempted to deserialize into an immutable buffer
FW_SERIALIZE_OK, //!< Serialization/Deserialization operation was successful
FW_SERIALIZE_FORMAT_ERROR, //!< Data was the wrong format (e.g. wrong packet type)
FW_SERIALIZE_NO_ROOM_LEFT, //!< No room left in the buffer to serialize data
FW_DESERIALIZE_BUFFER_EMPTY, //!< Deserialization buffer was empty when trying to read more data
FW_DESERIALIZE_FORMAT_ERROR, //!< Deserialization data had incorrect values (unexpected data types)
FW_DESERIALIZE_SIZE_MISMATCH, //!< Data was left in the buffer, but not enough to deserialize
FW_DESERIALIZE_TYPE_MISMATCH, //!< Deserialized type ID didn't match
FW_DESERIALIZE_IMMUTABLE, //!< Attempted to deserialize into an immutable buffer
FW_SERIALIZE_DISCARDED_EXISTING, //!< Serialization succeeded, but deleted old data
} SerializeStatus;

class SerialBufferBase; //!< forward declaration
Expand Down
84 changes: 75 additions & 9 deletions Svc/ComQueue/ComQueue.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
for (FwIndexType i = 0; i < static_cast<FwIndexType>(FW_NUM_ARRAY_ELEMENTS(this->entries)); i++) {
this->entries[i].priority = 0;
this->entries[i].depth = 0;
this->entries[i].mode = Types::QUEUE_FIFO;
this->entries[i].overflowMode = Types::QUEUE_DROP_NEWEST;
}
}

Expand All @@ -39,6 +41,8 @@
for (FwIndexType i = 0; i < TOTAL_PORT_COUNT; i++) {
this->m_throttle[i] = false;
}

static_assert(TOTAL_PORT_COUNT >= 1, "ComQueue must have more than one port");
}

ComQueue ::~ComQueue() {}
Expand Down Expand Up @@ -86,6 +90,8 @@
QueueMetadata& entry = this->m_prioritizedList[currentPriorityIndex];
entry.priority = queueConfig.entries[entryIndex].priority;
entry.depth = queueConfig.entries[entryIndex].depth;
entry.mode = queueConfig.entries[entryIndex].mode;
entry.overflowMode = queueConfig.entries[entryIndex].overflowMode;
entry.index = entryIndex;
// Message size is determined by the type of object being stored, which in turn is determined by the
// index of the entry. Those lower than COM_PORT_COUNT are Fw::ComBuffers and those larger Fw::Buffer.
Expand Down Expand Up @@ -119,7 +125,8 @@
if (allocationSize > 0) {
this->m_queues[this->m_prioritizedList[i].index].setup(
reinterpret_cast<U8*>(this->m_allocation) + allocationOffset, allocationSize,
this->m_prioritizedList[i].depth, this->m_prioritizedList[i].msgSize);
this->m_prioritizedList[i].depth, this->m_prioritizedList[i].msgSize, this->m_prioritizedList[i].mode,
this->m_prioritizedList[i].overflowMode);
}
allocationOffset += allocationSize;
}
Expand All @@ -132,10 +139,16 @@
// Handler implementations for commands
// ----------------------------------------------------------------------

void ComQueue ::FLUSH_QUEUE_cmdHandler(FwOpcodeType opCode, U32 cmdSeq, Svc::QueueType queueType, FwIndexType index) {
// Acquire the queue that we need to drain
FwIndexType queueIndex =
(queueType == QueueType::COM_QUEUE) ? index : static_cast<FwIndexType>(index + COM_PORT_COUNT);
FwIndexType queueIndex = this->getQueueNum(queueType, index);

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter index has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter queueType has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter index has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter queueType has not been checked.

// Validate queue index
if (queueIndex < 0 || queueIndex >= TOTAL_PORT_COUNT) {
this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::VALIDATION_ERROR);

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter opCode has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter opCode has not been checked.
return;

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter opCode has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter opCode has not been checked.
}

this->drainQueue(queueIndex);
this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::OK);
}
Expand All @@ -145,8 +158,57 @@
this->drainQueue(i);
}
this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::OK);
}

void ComQueue::SET_QUEUE_PRIORITY_cmdHandler(FwOpcodeType opCode,

Check notice

Code scanning / CodeQL

Long function without assertion

All functions of more than 10 lines should have at least one assertion.

Check notice

Code scanning / CodeQL

Long function without assertion Note

All functions of more than 10 lines should have at least one assertion.
U32 cmdSeq,
Svc::QueueType queueType,
FwIndexType index,
FwIndexType newPriority) {
// Acquire the queue we are to reprioritize
FwIndexType queueIndex = this->getQueueNum(queueType, index);

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter index has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter queueType has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter index has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter queueType has not been checked.

// Validate queue index
if (queueIndex < 0 || queueIndex >= TOTAL_PORT_COUNT) {
this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::VALIDATION_ERROR);

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter opCode has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter opCode has not been checked.
return;
}

// Validate priority range
if (newPriority < 0 || newPriority >= TOTAL_PORT_COUNT) {
this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::VALIDATION_ERROR);

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter opCode has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter cmdSeq has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter opCode has not been checked.
return;
}

// Find our queue in the prioritized list & update the priority
for (FwIndexType prioIndex = 0; prioIndex < TOTAL_PORT_COUNT; prioIndex++) {
// If the port based index matches, then update
if (m_prioritizedList[prioIndex].index == queueIndex) {
m_prioritizedList[prioIndex].priority = newPriority;
break; // Since we shouldn't find more than one queue at this port index
}
}

Copy link
Collaborator

@LeStarch LeStarch Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should assert (static_assert) TOTAL_PORT_COUNT >= 1

// Re-sort the prioritized list to maintain priority ordering
// Using simple bubble sort since TOTAL_PORT_COUNT is typically small
for (FwIndexType i = 0; i < TOTAL_PORT_COUNT - 1; i++) {
for (FwIndexType j = 0; j < TOTAL_PORT_COUNT - i - 1; j++) {
if (m_prioritizedList[j].priority > m_prioritizedList[j + 1].priority) {
// Swap metadata
QueueMetadata temp = m_prioritizedList[j];
m_prioritizedList[j] = m_prioritizedList[j + 1];
m_prioritizedList[j + 1] = temp;
}
}
Comment on lines 195 to 202

Check warning

Code scanning / CodeQL

Unbounded loop

This loop does not have a fixed bound.
Comment on lines +195 to +202

Check warning

Code scanning / CodeQL

Unbounded loop Warning

This loop does not have a fixed bound.
}

// Emit event for successful priority change
this->log_ACTIVITY_HI_QueuePriorityChanged(queueType, queueIndex, newPriority);

// Send command response
this->cmdResponse_out(opCode, cmdSeq, Fw::CmdResponse::OK);
}

// ----------------------------------------------------------------------
// Handler implementations for user-defined typed input ports
// ----------------------------------------------------------------------
Expand Down Expand Up @@ -251,24 +313,23 @@
static_cast<FwAssertArgType>(queueType), static_cast<FwAssertArgType>(queueNum));
const FwIndexType portNum =
static_cast<FwIndexType>(queueNum - ((queueType == QueueType::COM_QUEUE) ? 0 : COM_PORT_COUNT));
bool rvStatus = true;
FW_ASSERT(expectedSize == size, static_cast<FwAssertArgType>(size), static_cast<FwAssertArgType>(expectedSize));
FW_ASSERT(portNum >= 0, static_cast<FwAssertArgType>(portNum));
Fw::SerializeStatus status = this->m_queues[queueNum].enqueue(data, size);
if (status == Fw::FW_SERIALIZE_NO_ROOM_LEFT) {
if (status == Fw::FW_SERIALIZE_NO_ROOM_LEFT || status == Fw::FW_SERIALIZE_DISCARDED_EXISTING) {
if (!this->m_throttle[queueNum]) {
this->log_WARNING_HI_QueueOverflow(queueType, static_cast<U32>(portNum));
this->log_WARNING_HI_QueueOverflow(queueType, portNum);
this->m_throttle[queueNum] = true;
}

rvStatus = false;
}

// When the component is already in READY state process the queue to send out the next available message immediately
if (this->m_state == READY) {
this->processQueue();
}

return rvStatus;
// Check if the buffer was accepted or must be returned
return status != Fw::FW_SERIALIZE_NO_ROOM_LEFT;
}

void ComQueue::sendComBuffer(Fw::ComBuffer& comBuffer, FwIndexType queueIndex) {
Expand Down Expand Up @@ -385,4 +446,9 @@
this->m_prioritizedList[priorityIndex - 1] = temp;
}
}

FwIndexType ComQueue::getQueueNum(Svc::QueueType queueType, FwIndexType portNum) {
// Acquire the queue that we need to drain
return static_cast<FwIndexType>(portNum + ((queueType == QueueType::COM_QUEUE) ? 0 : COM_PORT_COUNT));

Check warning

Code scanning / CodeQL

Unchecked function argument

This use of parameter portNum has not been checked.

Check warning

Code scanning / CodeQL

Unchecked function argument Warning

This use of parameter portNum has not been checked.
}
} // end namespace Svc
48 changes: 11 additions & 37 deletions Svc/ComQueue/ComQueue.fpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,30 +36,10 @@ module Svc {
@ Port for scheduling telemetry output
async input port run: Svc.Sched drop

@ Flush a specific queue. This will discard all queued data in the specified queue removing it from eventual
@ downlink. Buffers requiring ownership return will be returned via the bufferReturnOut port.
async command FLUSH_QUEUE(queueType: QueueType @< The Queue data type
indexType: FwIndexType @< The index of the queue (within the supplied type) to flush
)
@ Flush all queues. This will discard all queued data removing it from eventual downlink. Buffers requiring
@ ownership return will be returned via the bufferReturnOut port.
async command FLUSH_ALL_QUEUES()
# ----------------------------------------------------------------------
# Special ports
# ----------------------------------------------------------------------

@ Port for emitting events
event port Log

@ Port for emitting text events
text event port LogText

@ Port for getting the time
time get port Time

@ Port for emitting telemetry
telemetry port Tlm

@ Command receive port
command recv port CmdDisp

Expand All @@ -69,26 +49,20 @@ module Svc {
@ Command response port
command resp port CmdStatus

# ----------------------------------------------------------------------
# Events
# ----------------------------------------------------------------------
@ Port for emitting events
event port Log

@ Queue overflow event
event QueueOverflow(
queueType: QueueType @< The Queue data type
index: U32 @< index of overflowed queue
) \
severity warning high \
format "The {} queue at index {} overflowed"
@ Port for emitting text events
text event port LogText

# ----------------------------------------------------------------------
# Telemetry
# ----------------------------------------------------------------------
@ Port for getting the time
time get port Time

@ Depth of queues of Fw::ComBuffer type
telemetry comQueueDepth: ComQueueDepth id 0
@ Port for emitting telemetry
telemetry port Tlm

@ Depth of queues of Fw::Buffer type
telemetry buffQueueDepth: BuffQueueDepth id 1
include "ComQueueCommands.fppi"
include "ComQueueEvents.fppi"
include "ComQueueTelemetry.fppi"
}
}
36 changes: 30 additions & 6 deletions Svc/ComQueue/ComQueue.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@ namespace Svc {
// ----------------------------------------------------------------------

class ComQueue final : public ComQueueComponentBase {
// Added to enable easy testing of set queue priority command
friend class ComQueueTester;

//! State of the currently transmitted buffer
enum BufferState { OWNED, UNOWNED };

Expand All @@ -47,10 +50,16 @@ class ComQueue final : public ComQueueComponentBase {
* Priority is an integer between 0 (inclusive) and TOTAL_PORT_COUNT (exclusive). Queues with lower priority values
* will be serviced first. Priorities may be repeated and queues sharing priorities will be serviced in a balanced
* manner.
*
* Queue mode determines whether messages are dequeued in FIFO or LIFO order.
*
* Overflow mode determines whether the newest or oldest message is dropped when the queue is full.
*/
struct QueueConfigurationEntry {
FwSizeType depth; //!< Depth of the queue [0, infinity)
FwIndexType priority; //!< Priority of the queue [0, TOTAL_PORT_COUNT)
FwSizeType depth; //!< Depth of the queue [0, infinity)
FwIndexType priority; //!< Priority of the queue [0, TOTAL_PORT_COUNT)
Types::QueueMode mode; //!< Queue mode (FIFO or LIFO)
Types::QueueOverflowMode overflowMode; //!< Overflow handling mode (DROP_NEWEST or DROP_OLDEST)
};

/**
Expand Down Expand Up @@ -81,10 +90,12 @@ class ComQueue final : public ComQueueComponentBase {
* method. Index and message size are calculated by the configuration call.
*/
struct QueueMetadata {
FwSizeType depth; //!< Depth of the queue in messages
FwIndexType priority; //!< Priority of the queue
FwIndexType index; //!< Index of this queue in the prioritized list
FwSizeType msgSize; //!< Message size of messages in this queue
FwSizeType depth; //!< Depth of the queue in messages
FwIndexType priority; //!< Priority of the queue
Types::QueueMode mode; //!< Queue mode (FIFO or LIFO)
Types::QueueOverflowMode overflowMode; //!< Overflow handling mode
FwIndexType index; //!< Index of this queue in m_queues
FwSizeType msgSize; //!< Message size of messages in this queue
};

/**
Expand Down Expand Up @@ -145,6 +156,16 @@ class ComQueue final : public ComQueueComponentBase {
U32 cmdSeq //!< The command sequence number
) override;

//! Handler for SET_QUEUE_PRIORITY command
//!
void SET_QUEUE_PRIORITY_cmdHandler(
FwOpcodeType opCode, //!< The opcode
U32 cmdSeq, //!< The command sequence number
Svc::QueueType queueType, //!< The Queue data type
FwIndexType indexType, //!< The index of the queue (within the supplied type) to modify
FwIndexType newPriority //!< New priority value for the queue
) override;

private:
// ----------------------------------------------------------------------
// Handler implementations for user-defined typed input ports
Expand Down Expand Up @@ -222,6 +243,9 @@ class ComQueue final : public ComQueueComponentBase {
//!
void processQueue();

//! Convert Queue Type & Index into single queueIndex
FwIndexType getQueueNum(Svc::QueueType queueType, FwIndexType portNum);

private:
// ----------------------------------------------------------------------
// Member variables
Expand Down
17 changes: 17 additions & 0 deletions Svc/ComQueue/ComQueueCommands.fppi
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
@ Flush a specific queue. This will discard all queued data in the specified queue removing it from eventual
@ downlink. Buffers requiring ownership return will be returned via the bufferReturnOut port.
async command FLUSH_QUEUE(queueType: QueueType @< The Queue data type
indexType: FwIndexType @< The index of the queue (within the supplied type) to flush
)

@ Flush all queues. This will discard all queued data removing it from eventual downlink. Buffers requiring
@ ownership return will be returned via the bufferReturnOut port.
async command FLUSH_ALL_QUEUES()

@ Set the priority of a specific queue at runtime
async command SET_QUEUE_PRIORITY(
queueType: QueueType @< The Queue data type
indexType: FwIndexType @< The index of the queue (within the supplied type) to modify
newPriority: FwIndexType @< New priority value for the queue
)

16 changes: 16 additions & 0 deletions Svc/ComQueue/ComQueueEvents.fppi
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
@ Queue overflow event
event QueueOverflow(
queueType: QueueType @< The Queue data type
index: FwIndexType @< index of overflowed queue
) \
severity warning high \
format "The {} queue at index {} overflowed"

@ Queue priority changed event
event QueuePriorityChanged(
queueType: QueueType @< The Queue data type
indexType: FwIndexType @< The index of the queue (within the supplied type) that was modified
newPriority: FwIndexType @< New priority value
) \
severity activity high \
format "{} {} priority changed to {}"
5 changes: 5 additions & 0 deletions Svc/ComQueue/ComQueueTelemetry.fppi
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
@ Depth of queues of Fw::ComBuffer type
telemetry comQueueDepth: ComQueueDepth id 0

@ Depth of queues of Fw::Buffer type
telemetry buffQueueDepth: BuffQueueDepth id 1
Loading
Loading