Generic OSAL Implementations

This package provides generic implementations of various OSAL modules. These are implemented using generic data structures, other OSAL types, and C++ code. These specifically avoid calls down into any underlying operating system.

Available implementations:

1. Os::PriorityQueue - Heap-based priority queue implementation using Os::Mutex for thread safety
2. Os::LocklessPriorityQueue - ISR-safe, lockless priority queue with at-initialization memory allocation
3. Os::PriorityMemQueue - ISR-safe, heap-based priority queue with per-priority memory pools and configuration

Os::PriorityQueue

Os::PriorityQueue is an in-memory implementation of Os::Queue. It allows projects that desire in-memory queue support to use Os::Queues. Os::PriorityQueue allocates memory for its underlying data structures through the registered Fw::MemAllocator and deallocates it through the same allocator. These actions are taken during create and object destruction. This implies that Os::PriorityQueue should be instantiated and initialized during system initialization.

For memory protection, Os::PriorityQueue delegates to Os::Mutex and Os::ConditionVariable.

[!WARNING] This Queue implementation is insufficient to be used for sending messages in ISR context due to the use of Os::Mutex as mentioned in above.

Priority Ordering: Larger priority numbers have higher priority than lower priority numbers (i.e., priority 0 is the lowest priority, there is no upper bound on priority value).

Requirements

The requirements for Os::Generic::PriorityQueue are as follows:

[!NOTE] Os::PriorityQueue is simpler than Os::PriorityMemQueue but lacks ISR safety and per-priority configuration capabilities.

Os::PriorityQueue Key Algorithms

Os::PriorityQueue stores messages in a set of dynamically allocated unordered parallel arrays. These arrays store: message data, and message data size respectively. There is also an index-free list that stores the indices that are available for storage in the fixed size arrays.

In order to prioritize messages, a Types::MaxHeap data structure is used.

When a message is received from a calling sender, find_index returns a free index from the free list. The data is copied into the message data array, and the size into the size array using this free index via store_data. The index is then inserted into the max heap structure for prioritization. When the queue is full and the BLOCKING option was supplied, the sender will block on the m_full condition variable until notified of a dequeue.

When a message is dequeued, the highest priority index is removed from the max heap. The data is copied out from the data array, and the size from the size array using that index via load_data and sent to the calling receiver. The index is then returned to the free list via return_index to indicate that it may be reused. When the queue is empty and the BLOCKING option was supplied, the receiver will block on the m_empty condition variable until notified of a enqueue.

If the queue is empty and data was received, the m_empty condition variable is notified to unblock waiting receivers. If the queue is full and data was dequeued, the m_full condition variable is notified to unblock waiting receivers.

Types::MaxHeap Data Structure

The Types::MaxHeap data structure is used to prioritize a list of indices using the given priority. This heap uses a dynamically allocated maximum-length array to back a binary tree storage structure. The first element is the root of the tree, left children are calculated using 2x + 1 and right children using 2x + 2. A node's parent is at (x - 1)/2.

When an index is pushed into this structure, it starts at the first unused element (first free child of some node) and iteratively swaps with its parent as long as the new nodes priority is larger than the parent. This ensures that the higher-priority elements are closer to the root of the tree. Specifically, the root is the highest priority element.

When an index is pulled from this structure, the root is removed as it is the highest priority data structure. The last leaf of the tree is elevated to the root and heapify is then called to restore the max-heap invariant of the data structure.

heapify starts at the newly ill-ordered root. It iteratively swaps this node with the highest-priority child until this node is the largest of the three (parent, left child, and right child) or until this node is swapped into a leaf position without children. The max-heap invariant is now restored.

Os::LocklessPriorityQueue

Os::LocklessPriorityQueue is an ISR-safe, lockless implementation of Os::Queue that provides strict-priority delivery without requiring any operating-system lock. It is intended for flight-software contexts where a producer or consumer may run in interrupt context and therefore cannot block on an OS-level mutex or condition variable.

All memory is allocated exactly once during create through the registered Fw::MemAllocator. No allocation occurs during send, receive, getMessagesAvailable, or getMessageHighWaterMark. The non-blocking variants of send and receive use only lock-free atomic operations and bounded memcpy, making them safe to invoke from ISR context.

[!NOTE] The blocking variants (BlockingType::BLOCKING) spin-wait and must not be invoked from ISR context.

Os::LocklessPriorityQueue Key Algorithms

The queue stores messages in a fixed pool of pre-allocated slots. Each slot is governed by a four-state atomic state machine (FREE -> WRITING -> READY -> READING -> FREE) with an embedded ABA tag that prevents the ABA problem across concurrent producers and consumers.

Producers scan the slot array for a FREE slot, claim it via compare-exchange, populate the message data, and publish with a release store transitioning to READY. Consumers scan for the highest-priority READY slot (with FIFO tiebreak by sequence number), claim it via compare-exchange, copy the message out, and release back to FREE.

All non-blocking control paths are bounded by depth * MAX_RETRY_PASSES. The detailed algorithm, memory-ordering rationale, and requirements traceability are documented in sdd-lockless-queue.md.

Os::PriorityMemQueue

Os::PriorityMemQueue is an ISR-safe and SMP-safe, priority-based memory queue implementation for F´ using lock-free atomic circular buffers (AtomicQueue). Each priority level has its own dedicated AtomicQueue, providing O(1) enqueue and dequeue without mutexes or interrupt disable.

The key components are:

1. AtomicQueue - One lock-free MPMC circular buffer per priority level for ISR-safe + SMP-safe message storage
2. Atomic Priority Mask - Lock-free bitmask tracking which priorities are enabled
3. Counting Semaphore - Notification mechanism for blocking receive operations
4. PriorityMemQueue - Main queue implementation that implements Os::QueueInterface

Requirements

The requirements for Os::Generic::PriorityMemQueue are as follows:

[!WARNING] Send/receive operations from ISR context must not use blocking behavior

[!NOTE] Os::PriorityMemQueue provides ISR safety and per-priority configuration at the cost of increased complexity and memory usage compared to Os::PriorityQueue.

AtomicQueue-Based Architecture

Overview

PriorityMemQueue uses AtomicQueue — a lock-free MPMC (multi-producer, multi-consumer) circular buffer — for message storage at each priority level. This provides:

• ISR Safety: Enqueue and dequeue use only atomic CAS operations; no interrupt disable or mutex
• SMP Safety: All position tracking uses std::atomic with acquire/release ordering on slot sequence numbers
• Generic: No OS-specific APIs; works on any platform providing std::atomic

Per-Priority AtomicQueues

Each priority level has its own dedicated AtomicQueue, allocated via the F´ memory allocator:

struct PriorityMemQueueHandle {
    I8 m_priorityMap[32];                  // Priority→index mapping (-1 = unused)
    Types::AtomicQueue* m_atomicQueues;    // Array sized to configured priorities
    FwSizeType m_numActivePriorities;      // Number of configured priorities
    std::atomic<U32> m_priorityMask;       // Bit mask of enabled priorities
    Os::CountingSemaphore* m_notEmptySem;  // Semaphore for blocking receive
    std::atomic<U32>* m_highWaterMarks;    // Per-priority peak depth
};

Send Flow

1. Look up priority in sparse map: index = m_priorityMap[priority]
2. Resolve to AtomicQueue at mapped index (or fallback to default priority 0)
3. Enqueue via lock-free CAS: atomicQueue->enqueue(buffer, size)
4. Update per-priority high water mark
5. Post semaphore to wake receiver: m_notEmptySem->post()

Receive Flow

1. Load enabled priority mask with memory_order_acquire
2. Scan from highest to lowest priority:
   • Check if priority is enabled in bitmask (skips disabled priorities)
   • Look up priority in sparse map: index = m_priorityMap[priority]
   • If priority not configured (index < 0), skip to next
   • Get AtomicQueue at mapped index
   • Call getSize() as a cheap pre-filter (2 relaxed loads)
   • If getSize() > 0, attempt dequeue() via lock-free CAS
   • On success, return message; on failure (MPSC contention), continue to next priority
3. If no message found:
   • BLOCKING: wait() on semaphore, then re-scan
   • NONBLOCKING: return EMPTY

Liveness is guaranteed by the semaphore — a spurious getSize() miss simply causes a re-scan after the next semaphore post.

Sparse Priority Optimization

When using non-consecutive priorities (e.g., {0, 15, 31}), allocating arrays sized to maxPriority + 1 wastes significant memory on unused entries.

To mitigate that, the priority queue uses a sparse priority map:

• m_priorityMap[32]: Maps priority value → array index (-1 for unconfigured)
• m_atomicQueues: Sized to numActivePriorities (not maxPriority + 1)
• m_highWaterMarks: Sized to numActivePriorities

Memory Savings Example (3 priorities: {0, 15, 31}):

• Before: 32 × 84 bytes = 2,688 bytes per queue
• After: 3 × 84 bytes + 32 bytes (map) = 284 bytes per queue
• Savings: 89% reduction (2,404 bytes saved)

For deployments with many queues using sparse priorities, this approach significantly reduces memory footprint and reduces the complexity of understanding how adding new queue priorities will impact memory use (each priority adds 84 bytes + message memory).

Performance Impact: Negligible (<6 cycles for write operations, <2% overhead for worst case receive)

PriorityMemQueue Overview

PriorityMemQueue is the main class that implements Os::QueueInterface. It provides a multi-priority message queue with configurable ISR-safe operations and flexible per-priority configuration.

Priority Ordering: Larger priority numbers have higher priority than lower priority numbers (i.e., priority 0 is the lowest priority, priority 31 is the highest).

Key Features:

• Multi-Priority Support: Up to 32 independent priority levels (0-31)
• Sparse Priority Allocation: Memory allocated only for configured priorities, reducing waste for non-consecutive priority values
• Per-Priority Memory Pools: Each configured priority has dedicated buffer allocation
• ISR-Safe Operations: Non-blocking send/receive can be called from interrupt context (platform-dependent)
• Flexible Configuration: Static configuration allows per-component, per-priority sizing
• Priority Management: Runtime enable/disable of individual priority levels
• Blocking Support: Optional blocking behavior for send and receive operations using counting semaphores
• High Water Mark Tracking: Monitors peak queue usage for system analysis

Synchronization

PriorityMemQueue uses a counting semaphore for blocking receive operations:

• m_notEmpty: Semaphore count represents number of messages available across all priorities
  • Initialized to 0 (queue starts empty)
  • post() called after successful enqueue to signal message availability
  • wait() blocks receiver when count reaches 0 (queue empty)
  • After wait() returns, receiver dequeues from highest priority with messages

A semaphore is used (as opposed to a condition variable & mutex) because:

• Semaphores are ISR safe (true for most RTOSs, including VxWorks, FreeRTOS, RTEMS, Greenhills Integrity, etc.)
• Simplifies synchronization
  • No need for separate mutex acquisition before checking queue state
  • No double-check pattern required to prevent lost notifications
  • Semaphore count intrinsically tracks message availability

Configuration

The static configuration system allows per-component, per-priority queue sizing. This enables fine-grained memory allocation tailored to each component's messaging patterns.

If a configuration is not provided for a given queue instance ID, then create() will use the message max size and depth arguments for priority Os::Queue::DEFAULT_PRIORITY (0).

Using priority_buffer_analyzer.py for Message Size Calculation

F´ provides priority_buffer_analyzer.py to automatically calculate the maximum message sizes for each priority level based on the topology's port connections. This eliminates manual calculation and ensures correct buffer sizing.

Script Location: ${FPRIME_FRAMEWORK_PATH}/cmake/autocoder/scripts/priority_buffer_analyzer.py

Integration Steps:

1. Add CMake hook in Deployment CMakeLists.txt:

   cmakeCopy

   # Add hook for port priority analyzer
   set(PRIORITY_BUFFER_HEADER_DIR "${CMAKE_BINARY_DIR}/${FPRIME_CURRENT_MODULE}/Top")
   set(PRIORITY_BUFFER_HEADER "${PRIORITY_BUFFER_HEADER_DIR}/PriorityBufferSizesAc.hpp")
   
   add_custom_command(
       OUTPUT "${PRIORITY_BUFFER_HEADER}"
       COMMAND ${CMAKE_COMMAND} -E make_directory "${PRIORITY_BUFFER_HEADER_DIR}"
       COMMAND ${PYTHON}
           "${FPRIME_FRAMEWORK_PATH}/cmake/autocoder/scripts/priority_buffer_analyzer.py"
           --build-dir "${CMAKE_BINARY_DIR}"
           --topology-path "${PRIORITY_BUFFER_HEADER_DIR}"
           --output "${PRIORITY_BUFFER_HEADER}"
       COMMENT "Generating PriorityBufferSizesAc.hpp for ${FPRIME_CURRENT_MODULE} deployment"
       VERBATIM
   )
   
   add_custom_target(priority_buffer_header DEPENDS "${PRIORITY_BUFFER_HEADER}")
   
   # Ensure priority buffer header is generated before building deployment
   add_dependencies(${FPRIME_CURRENT_MODULE} priority_buffer_header)
   

2. Include generated header in Main.cpp:

   cppCopy

   #include <Deployment/Top/PriorityBufferSizesAc.hpp>
   

3. Use generated constants for configuration:

   cppCopy

   // The script analyzes topology and generates constants per component/priority
   Os::Generic::PriorityMemQueue::QueuePriorityConfig cmdDispPriorityConfigs[] = {
       {Os::Generic::Queue::DEFAULT_PRIORITY, 
        PriorityBufferConfig::F_Prime_Svc_CmdDispatcher::PRIORITY_0, 10},
       {2, PriorityBufferConfig::F_Prime_Svc_CmdDispatcher::PRIORITY_2, 20},
       {3, PriorityBufferConfig::F_Prime_Svc_CmdDispatcher::PRIORITY_3, 20},
       {10, PriorityBufferConfig::F_Prime_Svc_CmdDispatcher::PRIORITY_10, 4},
   };
   
   Os::Generic::PriorityMemQueue::QueueConfig queueConfigs[] = {
       {Deployment::InstanceIds::CdhCore_cmdDisp, 
        FW_NUM_ARRAY_ELEMENTS(cmdDispPriorityConfigs), 
        &cmdDispPriorityConfigs[0]},
   };
   
   Os::Generic::PriorityMemQueue::configure(queueConfigs, 
                                            FW_NUM_ARRAY_ELEMENTS(queueConfigs), 
                                            false, 
                                            allocatorId);
   

Generated Header Structure:

The script generates PriorityBufferSizesAc.hpp with constants for each component and priority level:

namespace PriorityBufferConfig {
    constexpr FwSizeType DATA_OFFSET = sizeof(FwEnumStoreType) + sizeof(FwIndexType);
    
    namespace F_Prime_Svc_CmdDispatcher {
        // Maximum size for priority 0 messages (Cmd port)
        static constexpr FwSizeType PRIORITY_0 = 
            Fw::CmdPortBuffer::CAPACITY + DATA_OFFSET;
        
        // Maximum size for priority 2 messages (CmdResponse port)
        static constexpr FwSizeType PRIORITY_2 = 
            Fw::CmdResponsePortBuffer::CAPACITY + DATA_OFFSET;
        // ... etc
    }
}

How It Works:

1. Script parses topology dictionaries to identify all port connections by priority
2. For each component and priority, finds maximum serialized size across all port types
3. Adds DATA_OFFSET (priority + port ID overhead) to each size
4. Generates namespaced constants for use in configuration

Benefits:

• Eliminates manual message size calculations
• Automatically accounts for port buffer overhead (DATA_OFFSET)
• Updates automatically when topology or port types change
• Reduces risk of undersized buffers causing runtime assertions

ISR Safety

The AtomicQueue-based implementation provides ISR safety through:

1. Lock-free enqueue/dequeue: Uses atomic CAS on slot sequence numbers; no mutex, no interrupt disable
2. Atomic priority mask: std::atomic<U32> with acquire/release ordering — no spinlock needed
3. Non-blocking path: With NONBLOCKING, send and receive never call any blocking OS primitive

ISR Usage Pattern:

// From ISR context - use NONBLOCKING
queue.send(data, size, priority, QueueInterface::BlockingType::NONBLOCKING);
queue.receive(dest, capacity, QueueInterface::BlockingType::NONBLOCKING, size, pri);

[!WARNING] Blocking operations (BlockingType::BLOCKING) use a counting semaphore (post()/wait()) and must NOT be called from ISR context. Semaphore blocking from ISR context is undefined behavior on most RTOSs.

[!NOTE] ISR safety of post() (called by send()) depends on the platform's counting semaphore implementation. Verify Os_CountingSemaphore ISR safety for your target platform before using send() from ISR context.