RK0: The Real-Time Kernel '0'

1. THE KERNEL AT A GLANCE

Design Approach

Real-time system software, and more broadly cyber-physical software, is highly domain-specific and hardware-dependent. However, much of the industry continues to pursue commonality primarily at the hardware–software interface. These interfaces can express syntax and some semantics, which is useful, but they remain dangerously limited. On their own, they do not capture the coordination rules necessary for predictable real-time software.

RK0 adopts a different approach. The deeper commonality in cyber-physical systems lies not in the peripheral register map, board description, or driver abstraction, but in the concurrency model. Computation progresses in response to urgency, precedence, exclusion, availability, notification, and state-transfer conditions. Regardless of the domain or hardware, the application layer repeatedly faces coordination problems that are neither infinite in variety nor unknown. In this sense, RTOS services can be more expressive than generic mechanisms alone.

Many real-time kernels inherit a general-purpose habit of overgeneralisation: they provide overloaded primitives whose meaning is mostly derived from usage. This can appear neutral, but often becomes displacement. Meaning is not removed; it is pushed into the application, usually without a clear framework, where it tends to degrade into ad hoc protocols, hidden assumptions, and fragile side-effects.

RK0 does not reject generic services; it provides them where generality is beneficial. Additionally, RK0 offers services whose semantics directly encode common real-time coordination patterns. The objective is not to prevent composition, but to eliminate the need for applications to reconstruct complex real-time semantics from weak primitives.

Several concrete examples illustrate this approach:

RK0 provides a Message Queue that is sufficiently generic to support both blocking and non-blocking use, as well as specific operations such as send to front, peek, and overwrite, which are standard for message queues. Each queue enforces a fixed message size of 1, 2, 4, or 8 words, passed by copy. Word-sized alignment promotes type safety, low overheads, and predictable execution cost, while passing by copy enhances data integrity.
1. No API is provided for arbitrary variable-size messages. If an application requires variable-size payloads, which is common but remains strictly an application concern, a recommended pattern is to use a memory partition in conjunction with a 1-word message queue that carries pointers to pool blocks. This approach ensures that the kernel primitive remains bounded and predictable, while allowing the application to optimise for its specific requirements.
The Most-Recent Message protocol addresses a recurring pattern: 1:N communication in which consumers require the latest relevant state rather than a backlog of samples. Last-message semantics is necessary for responsive control structures, including cascaded or hierarchical control loops. Keeping data integrity on a 1:N fully-asynchronous (e.g., using double/triple buffering) is not trivial and is a common need.
Mutexes illustrate another design guideline. RK0 does not support recursive mutex locking; such attempts are treated as faults. Mutex service focuses on fully transitive priority inheritance over nested locks: a legitimate case for resource dependencies.

Nevertheless, the critical factor is not only recognising common concurrency needs as a kernel concern, but also handle their worst-case scenarios.

Service Design Criteria:

When to provide a new service, when to leave it to the application and when a service can be overloaded?

The general criterion can be stated as:

Avoid new abstractions for intrinsic properties; avoid hiding application policy, but introduce abstractions where the concurrency relation itself is operationally distinct.

Examples of this criterion applied: we have different sleep() primitives but the property (sleep time) is inherent to each Task; non-opaque treatment of monitors, because application policy is needed to design these mechanisms efficiently impacting execution progress differently.

2. Real-time and communication model

2.1. Tasks and Scheduling Policy

A Task is the concurrency unit in RK0. It follows the Thread model.

A static task assumes the states INITIALISED, READY, RUNNING, WAITING — the last is split into different pseudo-states. A dynamic task (explained later) adds a TERMINATED state.

Aside from design details, functionally the scheduler is a priority-based preemptive — quite standard.

A major difference is that it deliberately has no built-in time-slice.

The scheduler is the only unit allowed to switch a task from READY to RUNNING. Combined with no built-in time-slicing, we claim:

Execution progress is expressed in the application code.

Knowing the that the READY tasks are within a table of FIFO Queues, and each row is related to a priority, we define an invariant:

The highest priority ready task is at the head of highest priority non-empty ready queue.

All tasks run under the same policy:

A task must switch to the READY state before being eligible for scheduling.
Only the Scheduler can switch a task from READY to RUNNING.
A task will switch from RUNNING to READY if yielding or if being preempted by a higher priority task. Otherwise it can only go to a WAITING state, and eventually switch back to READY.
When a task is preempted by a higher priority task, it switches from RUNNING to READY and is placed back on the head position of its Ready Queue. This means that it will be resumed as soon as it is the highest priority ready task again.
On the contrary, if a task yields, it tells the scheduler it has completed its cycle. Then, it will be enqueued on the ready queue tail — the last queue position.
WAITING means the Task is suspended until a condition is satisfied. Once the condition is true, it transitions to the READY state, enqueued on the tail of its ready queue.
So, tasks with the same priority cooperate by either yielding or waiting.
If a task is dispatched and never yields or waits, the scheduler will correctly keep it running, while there is no higher priority READY task. It is not incidental starvation: no reason to progress differently was expressed in the application. This idea is not strange: it is sequential logic.
Finally, Tasks with the same priority are initially placed on the Ready Queue associated with that priority in the order they are created.

PS: tasks are allowed to be created as non-preemptible, for exceptional cases the application might need.

2.2. Events, Signals and Messages

A system has state variables that determine its behaviour. A change in a state variable is caused by an event. The periodic hardware interrupt (SysTick) that increments the kernel runtime count is an example.

The notion of execution progress on a digital computer arises from observable changes in state, whatever those states represent. Therefore, given two observation logical instants, if the observed state differs, at least one event must have occurred in (real) time. Note that if there is no difference, we can’t state that no event has happened.

In this sense, an event is a logical construct derived from observed reality. Time runs on a continuum; the computer samples reality with varying granularity. Computation, therefore, always lags. Aware of that, a real-time system’s goal is to react to external stimuli so that a result is delivered to the environment while it is still useful.

On a real-time kernel, execution progress follows the urgency of tasks and precedence conditions. We design concurrent units (Tasks) and use kernel services to coordinate their execution, ensuring they are ordered and that they produce a time-bounded final response.

You might want to read this document along with the Current public Services API and the Service Map

2.3. Suitable Applications

RK0 targets applications with the following characteristics:

They are designed to handle particular devices in which real-time responsiveness is imperative.
Applications and middleware may be implemented alongside appropriate drivers.
Drivers may even include the application itself.
Untested programs are not loaded: After the software has been tested, it can be assumed reliable.

3. Architecture

The layered architecture can be split — roughly — into two: a top and a bottom layer. On the top, the Executive manages the resources needed by the application.

On the bottom, the Low-level Scheduler works as a software extension of the CPU.

Together, they implement the Task abstraction. This primitive is the Concurrency Unit and follows the Thread model. A Task is a Thread

In systems design jargon, the Executive enforces policy (what should happen). The Low-level Scheduler provides the mechanism (how it gets done). The services are the primitives that gradually translate policy decisions into concrete actions executed by the Scheduler.

RK0’s goal is determinism on low-end devices. Its multitasking engine does not split user space from kernel space. Tasks execute in privileged mode and use a dedicated process stack pointer, distinct from the system stack. The rationale:

Application tasks are not unknown entities at run time.
Implementing system calls as traps increases complexity in critical control paths, degrading determinism.
Relying on the ARMv6/7-M MPU decreases memory usage efficiency and introduces latency on control paths. It does not fit RK0’s deterministic execution model.

3.1. Scheduler Design Internals

A notable scheduler characteristic is its constant-time complexity (O(1), for 'choose-next' operation) with low latency.

This was achieved by carefully composing the data structures and algorithms.

RK0 can handle context-switching with an extended frame when a float-point co-processor is available. This must be informed when compiling by defining the symbol __FPU_PRESENT=1.

3.1.1. Task Control Block

Every primitive is associated to a data structure we refer to as its Control Block. A Task Control Block is a record for stack, resources, and time management. The table below partially represents a Task Control Block (as this document is live, this might not reflect the exact fields of the current version).

Task Control Block
Task name
Task ID
Status
Assigned Priority
Effective Priority
Saved Stack Pointer
Stack Address
Stack Size
Last wake-time
Next wake-time
Time-out Flag
Preemption Flag
Owned Resources List
Waiting Resources List
Event Register Control Block
Timeout List Node
TCB List Node

3.1.2. Task Queues

The backbone of the queues where tasks will wait for their turn to run is a circular doubly linked list: removing any item from a double list takes O(1) (provided we don’t need to search the item). As the kernel knows each task’s address, adding and removing is always O(1). Singly linked lists can’t achieve O(1) for removal.

3.1.3. Ready Queue Table

Another design choice to achieve O(1) is the global ready queue, which is a table of FIFO queues—each queue dedicated to a priority—and not a single ordered queue. So, enqueuing a ready task is always O(1). Given the sorting needed, the time complexity would be O(n) if tasks were placed on a single ready queue.

3.1.4. Waiting Queues

The scheduler does not have a unique waiting queue. Every kernel object that can block a task has an associated waiting queue. Because these queues are a scheduler component, they follow a priority discipline: the highest priority task is dequeued first, always.

When an event capable of switching tasks from WAITING to READY happens, one or more tasks (depending on the mechanism) are then placed on the ready list, unique to their priority. Now, they are waiting to be picked by the scheduler—that is the definition of READY.

3.1.5. The choose-next algorithm

As the ready queue table is indexed by priority - the index 0 points to the queue of ready tasks with priority 0, and so forth, and there are 32 possible priorities - a 32-bit integer can represent the state of the ready queue table. It is a BITMAP:

The BITMAP update happens whenever:

(1a) A task is readied, update: BITMAP |= (1U << task->priority);
(1b) An empty READY QUEUE becomes non-empty, update: BITMAP |= (1U << queueIndex)
(2): Every Time READY QUEUE becomes empty, update: BITMAP &= ~(1U << queueIndex);
EXAMPLE:

  Ready Queue Index :     (6)5 4 3 2 1 0
          Not empty :      1 1 1 0 0 1 0
                           ------------->
                 (LOW)  Effective Priority  (HIGH)
In this case, the scenario is a system with 7 priority task levels. Queues with priorities 6, 5, 4, and 1 are not empty.

Having the Ready Queue Table bitmap, we find the highest priority non-empty task list as follows:

(1) Isolate the rightmost '1':

RBITMAP = BITMAP & -BITMAP. (- is the bitwise operator for two's complement: ~BITMAP + 1) `

In this case:

                           [31]       [0]  :  Bit Position
                             0...1110010   :  BITMAP
                             1...0001110   : -BITMAP
                            =============
                             0...0000010   :  RBITMAP
                                     [1]

The rationale here is that, for a number N, its 2’s complement -N, flips all bits - except the rightmost '1' (by adding '1') . Then, N & -N results in a word with all 0-bits except for the less significant '1'.

(2) Extract the rightmost '1' position:

For ARMv7M, we benefit from the CLZ instruction to count the leading zeroes. As they are the number of zeroes on the left of the rightmost bit, '1', this value is subtracted from 31 to find the Ready Queue index.

unsigned __getReadyPrio(unsigned readyQBitmap)
{
    unsigned ret;
    __ASM volatile (
        "clz    %0, %1     \n"
        "neg    %0, %0     \n"
        "add    %0, %0, #31\n"
        : "=&r" (ret)
        : "r" (readyQBitmap)
        :
    );
    return (ret);
}

This instruction would return #30, and #31 - #30 = #01 in the example above.

For ARMv6M there is no suitable hardware instruction. The algorithm is written in C and counts the trailing zeroes, thus, the index number. Although it might vary depending on your compiler settings, it takes ~11 cycles (note it is still O(1)):

/*
  De Brujin's multiply+LUT
  (Hacker's Delight book)
*/

/* table is on a ram section  for efficiency */
 const static unsigned readyPrioTbl[32] =
{
 0, 1, 28, 2, 29, 14, 24, 3, 30, 22, 20, 15, 25, 17, 4, 8,
 31, 27, 13, 23, 21, 19, 16, 7, 26, 12, 18, 6, 11, 5, 10, 9
};

RK_FORCE_INLINE static inline
unsigned __getReadyPrio(unsigned readyQBitmap)
{
    unsigned mult = readyQBitmap * 0x077CB531U;

    /* Shift right the top 5 bits
     */
    unsigned idx = (mult >> 27);

    /* LUT */
    unsigned ret = (unsigned)readyPrioTbl[idx];
    return (ret);
}

For the example above, mult = 0x2 * 0x077CB531 = 0x0EF96A62. The 5 leftmost bits (the index) are 00001 → table[1] = 1.

During a context switch, the procedures to find the highest priority non-empty ready queue table index are as follows:

static inline RK_PRIO kCalcNextTaskPrio_(VOID)
{
    if (readyQBitMask == 0U)
    {
        return (idleTaskPrio);
    }
    readyQRightMask = readyQBitMask & -readyQBitMask;
    RK_PRIO prioVal = (RK_PRIO) (__getReadyPrio(readyQRightMask));
    return (prioVal);
}

/* O(1) complexity */
VOID kSwtch(VOID)
{
    nextTaskPrio = kCalcNextTaskPrio_();

    RK_TCB* nextRunPtr = NULL;

    kTCBQDeq(&readyQueue[nextTaskPrio], &nextRunPtr);

    runPtr = nextRunPtr;

}

4. Timers

4.1. Busy-wait delay

A busy-wait delay kBusyDelay(t) or kDelay(t) means 'consume t ticks of time RUNNING'.

So, while a busy-delay is active if the task is preempted, the elapsed time to resume is not taken into account to finish the busy-wait operation.

It is useful to simulate a workload or to just delay two calls within a task without blocking.

Context switching is probably the most significant overhead on a kernel. The time spent on the System Tick handler contributes to much of this overhead.

Design Choice:

Timers are kept on a single list; only the head element needs to be updated using a delta-queue approach.

Benefits:

Keep the overhead of updating timers as minimal as possible with the delta queue;

4.2. Timeout operations

Timeout Node
Timeout Type
Absolute Interval (Ticks)
Relative Interval (Ticks)
Waiting Queue Address
Next Timeout Node
Previous Timeout Node

Every task is prone to events triggered by timers described in this section. Every Task Control Block has a node to a timeout list. This list is doubly linked treated a delta-sequence.

A set T_set= {(T1,8), (T2,6), (T3,10)} will be started at a relative time 0 as a sequence T_seq= <(T2,6), (T1,2), (T3,2)>.

Thus, for every system tick, only the head element on the list needs to be decreased — yielding O(1) on decreasing, that happens on the Hardware interrupt for the System Tick.

The ordering for the delta-queue is not O(1), it is O(n). Note that a decrease happens on every SysTick interrupt; and the (re)ordering happens when adding a new node to the list.

4.2.1. Blocking Time-out

These are internal timers associated with kernel calls that are blocking. Thus, establishing an upper-bound waiting time might benefit them. When the time for unblocking is up, the kernel call returns, indicating a timeout. This value is passed as a number of ticks.

When blocking is associated with a kernel object (other than the Task Control Block), the timeout node will store the object waiting for queue’s address, so it can be removed if time expires.

A kernel call is made non-blocking, that is try semantics, by assigning the value RK_NO_WAIT, the function returns immediately if unsuccessful. The value RK_WAIT_FOREVER suspends a task indefinitely until the condition is satisfied. Timeout values above RK_MAX_PERIOD (0x7FFFFFFF) are invalid.

In practice, we often block either using RK_WAIT_FOREVER or do not block (try semantics, RK_NO_WAIT).

Use a bounded timeout when you are establishing an upper bound that if not met leads to refining the design or an alternative handle.

4.2.2. Sleep Delay

The sleepdelay() (aliased as sleep()) puts a task to sleep for the exact number of t ticks on every call — no matter when the last call has happened.

Example:

VOID Task1(VOID* args)
{
    RK_UNUSEARGS
    UINT count = 0;
    while (1)
    {

        logPost("Task1: sleep");
        kSleep(300);
        /* wake here */
        count += 1U;
        if (count >= 5)
        {
            kDelay(25); /* spin */
            count=0;
            /* every 5 activations there will be a drift */
        }
    }
}

Output:

0 ms :: Task1: sleep
300 ms :: Task1: sleep  <-- +300
600 ms :: Task1: sleep  <-- +300
900 ms :: Task1: sleep  <-- +300
1200 ms :: Task1: sleep <-- +300
1525 ms :: Task1: sleep <-- +325
1825 ms :: Task1: sleep <-- +300
2125 ms :: Task1: sleep <-- +300
2425 ms :: Task1: sleep
2725 ms :: Task1: sleep
3050 ms :: Task1: sleep
3350 ms :: Task1: sleep
3650 ms :: Task1: sleep
3950 ms :: Task1: sleep
4250 ms :: Task1: sleep
4575 ms :: Task1: sleep

4.2.3. Compensated Sleep Delays

These are suspensions that recompute the time considering the drift between calls. They are typically used to create periodic tasks with explicit periods. The general pattern is:

VOID Task(VOID *args)
{
    <initialisation>

    while(1)
    {
        <periodic code>; /* this has an execution time */
        scheduleNext(PERIOD); /* cycle is finished: compute delay to keep PERIOD */
    }
}

The RMS algorithm considers all tasks have a common phase grid, that is when they are made READY, or are eligible to be scheduled. This means that from the very first activation, the time elapsed is already take into account to when the next release should happen.

Nevertheless, it is common for kernels to provide sleep primitives that take into account a local anchor time normally set on the <initialisation> code — and will work as waitUntil(&anchor, PERIOD). RK0 provides both, and they suit different cases, as exposed below.

4.2.3.1. Sleep and Release (phase-locked)

sleeprelease(P) is used to delay a task so it is released on periodic rate. The sleep time is recalculated by the kernel on every call.

If the task wakes late by N ticks with 0 < N < P, the kernel compensates by scheduling the next wake earlier (shortening the next sleep) so that over two periods the phase is preserved:

Say a task is expected to return from its k_eth sleep at T_k+1+ = T+_k+ + P `. If the task is resumed at `+T_k+1+ = T+_k+ + P + N+, upon detecting this drift, the kernel sets: (T_k+2+ = T+_k+1+ + P - N)` for `+N < P.

This can be rewritten as:

(T_k+2+ = T+_k+ + P + N + P - N) ←→ (T+_k+2+ - T+_k+ = 2P)+

Example:

VOID Task1(VOID* args)
{
    RK_UNUSEARGS
    UINT count = 0;
    while (1)
    {

       logPost("Task1 released.");

        count += 1U;
        if (count >= 5)
        {
            kDelay(25); /* spin */
            count=0;
        }

        kSleepRelease(300); /*P=300 ticks; tick=1ms*/


    }
}

Output:

.
.

/* R is release time */

1200 ms :: Task1: sleep periodic (R==4P)        (n)
1525 ms :: Task1: sleep periodic (5P<R<6P)       |
1800 ms :: Task1: sleep periodic (6P)            |
2100 ms :: Task1: sleep periodic (7P)            |
2400 ms :: Task1: sleep periodic (8P)            |
2700 ms :: Task1: sleep periodic (9P)            |
3025 ms :: Task1: sleep periodic (10P<R<11P)     |
3300 ms :: Task1: sleep periodic (11P)           |
3600 ms :: Task1: sleep periodic (12P)           |
3900 ms :: Task1: sleep periodic (13P)           |
4200 ms :: Task1: sleep periodic (14P)           |
4525 ms :: Task1: sleep periodic (15P<R<16P)     |
4800 ms :: Task1: sleep periodic (16P)          (m)  m-n=12
.                                              -----
.                                           Phase=3600=12xP
.

This mechanism is phase-locked. When the lateness is greater or equal to P it skips one or more releases to stay locked to the phase grid. In some sense, the period value can be seen as a deadline — if not met, the scheduler rejects to run on that activation.

sleeprelease() makes easier to perform worst-response time analysis on periodic tasks.

A set of periodic tasks must have priorities assigned properly (highest request rate, highest priority — the lower the period, the higher the priority). For sleeprelease() this is mandatory given the common phase grid.

4.2.3.2. Sleep Until (local scope reference)

sleepuntil(anchor, period) is somehow similar to sleeprelease(), but differs in two important aspects:

The reference used to calculate how long to suspend to keep its rate is local to each task. It means the time before the first time the task is dispatched is dismissed.
A late release longer than 1 period will return and run immediately. It prioritises execution count within a time-window - not the phase across releases.

The snippet belows clearly demonstrates how each mechanism handles lateness that are longer than 1 period:

/* Every 3rd call both tasks will add up a delay longer than the task's period */

VOID HTask(VOID* args) /* higher priority: Period is 300 ticks */
{
    RK_UNUSEARGS
    UINT count = 0;
    while (1)
    {

        logPost("Higher: begin\r\n");
        /* wake here */
        count += 1U;
        kDelay(5);
        if (count >= 3)
        {
            kSleep(400); /* suspend */
            count=0;
        }
        logPost("Higher: end\r\n");
        kSleepRelease(300);

    }
}


VOID LTask(VOID *args) /* lower priority: Period is 400 ticks */
{
    RK_UNUSEARGS
    RK_TICK anchor = kTickGet();
    UINT count=0;
    while (1)
    {

        logPost("Lower: begin\r\n");
        /* wake here */
        count += 1U;
        kDelay(5);
        if (count >= 3)
        {
            kSleep(500);
            count=0;
        }
        logPost("Lower: end\r\n");
        kSleepUntil(&anchor, 400);
    }
}

Output:

       0 ms :: Higher: begin
       5 ms :: Higher: end
       5 ms :: Lower: begin
      10 ms :: Lower: end
     300 ms :: Higher: begin
     305 ms :: Higher: end
     405 ms :: Lower: begin
     410 ms :: Lower: end

    /* H 3rd run, expected next at 900ms */
     600 ms :: Higher: begin


   /* L 3rd run, expected next at 1205 ms */
     805 ms :: Lower: begin

    /* H Drift: 1005ms - 600ms = 405 ms > 300ms */
    1005 ms :: Higher end

    /* H released again @ next multiple of 300. */
    1200 ms :: Higher: begin
    1205 ms :: Higher: end

   /* L Drift: 1310ms - 805ms = 505 ms > 400ms */
    1310 ms :: Lower: end
    /* it runs again  immediately */
    1310 ms :: Lower: begin

One normally does not write a code with periodic tasks expecting they will not keep their rate. But on the field a transient overload might cause it to happen. If it does, you choose the policy that is best-fit for your task: preserve phase (skip) or preserve execution count.

Importantly, an ISR shall never block. Any blocking call from an ISR is invalid and triggers fault handling when error checking is enabled.

4.3. Callout Timers (Application Timers)

Timer Control Block
Option: Reload/One-Shot
Phase (Initial Delay)
Callout Function Pointer
Callout Argument
Timeout Node

These are Application Timers that will issue a callback when expiring.

Optionally, there is an initial phase delay, besides the option to be periodic or run once.

It should be clear Callout timers are for minimal urgent operations that need high time precision and in practice they run at priority that could be considered -1. They are not to be used as a substitute for periodic tasks.

Right usage: a soft keep-alive.
Wrong usage: a callback with several branches, with modulo operations to create a time-triggered pattern.

5. System Tick

A dedicated peripheral that generates an interrupt after a defined period provides the kernel time reference. For ARMv6/7M, this peripheral is the built-in SysTick, a 24-bit counter timer.

The 'housekeeping' accounts for global timer tracking and any tick-dependent condition that might change a task status. The handler performs some housekeeping on every tick. If a task whose execution progress was depending on time switches to READY the routine returns to call the scheduler. If an application timer is due, it signals the system task that performs the installed callback and additional logic.

Although many examples here are set @ 1ms tick, 10ms is a realistic case for low-end MCUs.

6. System Tasks

System Tasks perform housekeeping and other kernel maintenance outside interrupt handlers.

Currently there are the Idle Task the PostProcSysTask.

The PostProcSysTask executes Application Timer callbacks and work deferred from ISRs, such as broadcast-style flushes on Sleep Queues.

The Idle Task runs whenever there is no other ready task to be dispatched. The IdleTask is dispatched when the Ready bitmap is 0x00000000.

7. Memory Allocator

Memory Allocator Control Block
Associated Block Pool
Number of Blocks
Block Size
Number of Free Blocks
Free Block List

The standard C library malloc() leads to fragmentation and (also, because of that) is highly indeterministic. Unless we use it once - to allocate memory before starting up, it doesn’t fit. But often, we need to 'multiplex' memory amongst tasks over time, that is, to dynamically allocate and deallocate.

To avoid fragmentation, we use fixed-size memory blocks. So every RK_MEM_PARTITION kernel object controls allocation and deallocation of homogeneous blocks in memory, that can either be of any type. For instance, data structures for a request-response communication, or stack buffers for Tasks.

A simple approach would be a static table marking each block as free or taken. With this pattern, you will need to 'search' for the next available block, if any - the time for searching changes - bounding this search to a maximum number of blocks, or O(n). To optimise, an approach is to keep track of what is free using a dynamic table—a linked list of addresses. Now we have O(1).

We use "meta-data" to initialise the linked list. Every address holds the "next" address value. All addresses are within the range of a pool of fixed-size blocks. This approach limits the minimal size of a block to the size of a memory address—32 bits for our supported architecture.

Yet, this is the cheapest way to store meta-data. If not stored on the empty address itself, an extra 32-bit variable would be needed for each block, so it could have a size of less than 32 bits.

Allocating memory at runtime is a major source of latency (1), indeterministic (2) behaviour, and footprint overhead (3).

Design choice: the allocator’s design achieves low-cost, deterministic, fragmentation-free memory management by using fixed-size word-aligned block sizes (1)(2) and embedding metadata within the memory blocks themselves (3).

Benefits: Run-time memory allocation benefits have no real-time drawbacks.

The kernel will always round up the block size to the next multiple of 4. Say the user creates a memory pool, assigning blocks to be 6-byte wide; they will turn into 8-byte blocks.

7.1. How it works

When a routine calls alloc(), the address to be returned is the one a "free list" is pointing to, say addr1. Before returning addr1 to the caller, we update the free list to point to the value stored within addr1 - say addr8 at that moment.

When a routine calls free(addr1), we overwrite whatever has been written in addr1 with the value-free list point to (if no more alloc() were issued, it would still be addr8), and addr1 becomes the free list head again.

Allocating and deallocating fixed-size blocks using this structure and storing meta-data this way is as deterministic (O(1)) and economical as we can get for dynamic memory allocation.

8. Dynamic Kernel Objects

8.1. Dynamic Tasks

Dynamic Tasks are tasks that can be created and terminated after the scheduler has started and they rely on the dynamic allocation of stack buffers.

For real-time sanity, tasks created after the scheduler starts are already bad. If they are dynamic and unknown, it is a liability. This should be clear.

That said, the support for dynamic tasks as a service was made available so some 3rd party middlewares, that create tasks 'on the fly, can be integrated with less effort.'

In RK0 we call Static Tasks are those initialised (created) before the scheduler is started and will never be terminated.

Dynamic Tasks are created by other tasks. They can be terminated by other tasks and also, terminate themselves. The TERMINATED state is assigned.

This is just a mark on the TCB. The memory used by a terminated task can be recycled and assigned to a new task.

We handle each of them as follows:

Static/startup tasks:
- Objects declared with RK_DECLARE_TASK(…).
- Initialised with kTaskInit(…).
- Require an explicit stack buffer pointer.
Dynamic/runtime tasks:
- Objects declared with RK_DECLARE_DYNAMIC_TASK(…).
- Backed by one or more user-defined stack partitions (fixed block size), which can be declared using RK_DECLARE_DYNAMIC_STACK_POOL(…).
- Created with kTaskSpawn(…) receiving an RK_DYNAMIC_TASK_ATTR data structure.
- Its stack size depends on the Memory Pool assigned to its attributes — as each memory pool has homogeneous objects.
- kTaskTerminate(taskHandle) is used by a task to terminate another.
- kTaskTerminateSelf() a handled by the system post-processing task.

/* static tasks and dynamic tasks altogether */
/* in kconfig.h the total number of supported tasks is
set to 4 in this example */

#define STACK1SIZ 128U
#define STACK2SIZ 128U
#define DYNSTACKSIZ 256U

#define N_DYN_TASKS 2U

RK_DECLARE_TASK(task1Handle, Task1, stack1,
                STACK1SIZ)
RK_DECLARE_TASK(task2Handle, Task2, stack2,
                STACK2SIZ)

RK_DECLARE_DYNAMIC_TASK(task3Handle, Task3)
RK_DECLARE_DYNAMIC_TASK(task4Handle, Task4)

RK_DECLARE_DYNAMIC_STACK_POOL(dynamicTaskMem, stackPool,
                              N_DYN_TASKS, DYNSTACKSIZ)


static RK_DYNAMIC_TASK_ATTR task3Attr =
    {
        .taskFunc = Task3,
        .argsPtr = RK_NO_ARGS,
        .taskName = "Task3",
        .priority = 1U,
        .preempt = RK_PREEMPT,
        .stackMemPtr = &dynamicTaskMem
    };

static RK_DYNAMIC_TASK_ATTR task4Attr =
    {
        .taskFunc = Task4,
        .argsPtr = RK_NO_ARGS,
        .taskName = "Task4",
        .priority = 1U,
        .preempt = RK_PREEMPT,
        .stackMemPtr = &dynamicTaskMem
    };


VOID kApplicationInit(VOID)
{
    /* Initialise dynamic task memory partition */
    RK_ERR err = kMemPartitionInit(&dynamicTaskMem,
                                   stackPool,
                                   sizeof(stackPool[0]),
                                   N_DYN_TASKS);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /*initialise static tasks */
    err = kTaskInit(&task1Handle,
                           Task1,
                           RK_NO_ARGS,
                           "Task1",
                           stack1,
                           STACK1SIZ,
                           1U,
                           RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* note task2 has lower priority than 1, 3 and 4 */
    err = kTaskInit(&task2Handle,
                           Task2,
                           RK_NO_ARGS,
                           "Task2",
                           stack2,
                            STACK2SIZ,
                           2U,
                           RK_PREEMPT);

    K_ASSERT(err == RK_ERR_SUCCESS);
}

VOID Task1(VOID* args)
{
    RK_UNUSEARGS
    ULONG count = 0UL;
    while (1)
    {

        printf("COUNT1: %lu [TASK1] running\r\n", count);
        count++;

        /* Task1 spawns Task3 after 10 iterations */
        if (count == 10)
        {
            printf("COUNT1: %lu !!!! [TASK1] spawning Task3\r\n", count);
            RK_ERR err = kTaskSpawn(&task3Attr, &task3Handle);
            K_ASSERT(err == RK_ERR_SUCCESS);
        }

        kSleep(10U);
    }
}


VOID Task2(VOID* args)
{
    RK_UNUSEARGS
    ULONG count = 0UL;
    /* Task2 spawns Task4 right off q*/
    printf("COUNT2: !!!! %lu [TASK2] spawning Task4\r\n", count);
    RK_ERR err = kTaskSpawn(&task4Attr, &task4Handle);
    K_ASSERT(err == RK_ERR_SUCCESS);

    while (1)
    {

        printf("COUNT2: %lu [TASK2] running\r\n", count);
        count++;
        kSleep(10U);
    }
}

VOID Task3(VOID* args)
{
    RK_UNUSEARGS
    ULONG count = 0UL;
    while (1)
    {
        /* Task3 never terminates */
        printf("COUNT3: %lu [TASK3] running\r\n", count);
        count++;
        kSleep(10U);
    }
}

VOID Task4(VOID* args)
{
    RK_UNUSEARGS
    ULONG count = 0UL;
    while (1)
    {

        count++;
        printf("COUNT4: %lu [TASK4] running\r\n", count);
        if (count == 20)
        {
            /* Task4 terminates itself after 20 iterations */
            printf("COUNT4: %lu !!!! [TASK4] terminating self\r\n", count);
            RK_ERR err = kTaskTerminateSelf();
            K_ASSERT(err == RK_ERR_SUCCESS);
        }
        kSleep(10U);
    }
}

COUNT1: 0 [TASK1] running
COUNT2: !!!! 0 [TASK2] spawning Task4
COUNT4: 1 [TASK4] running
.
/* task3 is spawned later */
COUNT1: 10 !!!! [TASK1] spawning Task3
COUNT3: 0 [TASK3] running
COUNT2: 9 [TASK2] running
COUNT3: 1 [TASK3] running
COUNT1: 10 [TASK1] running
.
COUNT3: 9 [TASK3] running
COUNT1: 18 [TASK1] running
COUNT4: 19 [TASK4] running
COUNT2: 18 [TASK2] running
COUNT4: 20 [TASK4] running
COUNT4: 20 !!!! [TASK4] terminating self
.
/* from now on, only 1, 2, 3 keep going */
COUNT1: 19 [TASK1] running
COUNT3: 10 [TASK3] running
COUNT2: 19 [TASK2] running
.

There are several aspects to take into account when creating and mainly destroying tasks: destroying a task that holds a resource is particularly harmful. The kernel can only refuse to do so.
sleeprelease(P) is not to be used on dynamic tasks, as they do not share a common phase grid.

8.2. Dynamic Kernel Service Objects

When dynamic tasks are used, it is common to also need kernel objects whose lifetime is not static, such as a Mutex, Semaphore, Sleep Queue, Message Queue, Timer, or MRM. RK0 provides a kernel-owned dynamic object API for those non-task objects.

This API is enabled by RK_CONF_DYNAMIC_OBJECTS in kconfig.h. When enabled, each object type has a configured maximum and a kernel-owned memory partition backing that type:

RK_CONF_DYNAMIC_SEMAPHORES_MAX
RK_CONF_DYNAMIC_MUTEXES_MAX
RK_CONF_DYNAMIC_SLEEP_QUEUES_MAX
RK_CONF_DYNAMIC_MESG_QUEUES_MAX
RK_CONF_DYNAMIC_TIMERS_MAX
RK_CONF_DYNAMIC_MRMS_MAX

Memory Partitions themselves are not dynamically created by this layer. They are allocator infrastructure. Message Queues, Mailboxes, and MRMs also still require application-provided data buffers; the dynamic API allocates only the kernel control block.

Before creating any dynamic object, initialise all dynamic object partitions once from application initialisation context:

RK_INIT_OBJ_PARTITIONS

RK_INIT_OBJ_PARTITIONS is a RK0 statement-style macro. It is called without parentheses or trailing ';'

VOID kApplicationInit(VOID)
{
    RK_INIT_OBJ_PARTITIONS

    /* create tasks and dynamic objects after this point */
}

If RK_CONF_DYNAMIC_OBJECTS is OFF, the macro is a no-op and the dynamic create/destroy APIs are not exposed. The underlying kObjPartitionsInit() function returns RK_ERR for code that needs to inspect the initialisation status directly.

Dynamic objects use handle typedefs. In RK0 a _HANDLE type is a pointer:

/* Dynamic objects types */
RK_SEMAPHORE_HANDLE;
RK_MUTEX_HANDLE;
RK_SLEEP_QUEUE_HANDLE;
RK_MESG_QUEUE_HANDLE;
RK_MBOX_HANDLE;
RK_TIMER_HANDLE;
RK_MRM_HANDLE;

A create or destroy API receives a pointer to the handle. For example, RK_MUTEX_HANDLE * is still a pointer to the object pointer, but it makes ownership clearer and lets destroy set the caller handle to NULL.

RK_ERR kSemaphoreCreate(RK_SEMAPHORE_HANDLE *semaHandlePtr,
                        UINT initValue,
                        UINT maxValue);
RK_ERR kSemaphoreDestroy(RK_SEMAPHORE_HANDLE *semaHandlePtr);

RK_ERR kMutexCreate(RK_MUTEX_HANDLE *mutexHandlePtr, UINT protocol);
RK_ERR kMutexDestroy(RK_MUTEX_HANDLE *mutexHandlePtr);

RK_ERR kSleepQueueCreate(RK_SLEEP_QUEUE_HANDLE *sleepqHandlePtr);
RK_ERR kSleepQueueDestroy(RK_SLEEP_QUEUE_HANDLE *sleepqHandlePtr);

RK_ERR kMesgQueueCreate(RK_MESG_QUEUE_HANDLE *queueHandlePtr,
                        VOID *bufPtr,
                        ULONG mesgWords,
                        ULONG nMesg);
RK_ERR kMesgQueueDestroy(RK_MESG_QUEUE_HANDLE *queueHandlePtr);

#define kMboxCreate  kMesgQueueCreate
#define kMboxDestroy kMesgQueueDestroy

RK_ERR kTimerCreate(RK_TIMER_HANDLE *timerHandlePtr,
                    RK_TICK phase,
                    RK_TICK countTicks,
                    RK_TIMER_CALLOUT funPtr,
                    VOID *argsPtr,
                    RK_OPTION reload);
RK_ERR kTimerDestroy(RK_TIMER_HANDLE *timerHandlePtr);

RK_ERR kMRMCreate(RK_MRM_HANDLE *mrmHandlePtr,
                  RK_MRM_BUF *mrmPoolPtr,
                  VOID *mesgPoolPtr,
                  ULONG nBufs,
                  ULONG dataSizeWords);
RK_ERR kMRMDestroy(RK_MRM_HANDLE *mrmHandlePtr);

The destroy APIs are deliberately strict:

The object must have been allocated by the matching dynamic object partition. Do not pass a static object to a dynamic destroy call.
Create and destroy calls must not be made from ISR context.
On successful destroy, the object is removed from trace tracking, its storage is cleared, the block is returned to the dynamic partition, and the caller handle is set to NULL.
A Semaphore cannot be destroyed while tasks are waiting on it; a Sleep Queue cannot be destroyed while tasks are sleeping on it.
A Mutex cannot be destroyed while it is locked, owned, or has waiters.
A Message Queue or Mailbox cannot be destroyed while it has an owner, queued messages, blocked senders, blocked receivers, or active broadcast receivers.
A Timer destroy cancels the timer before returning the control block.
An MRM cannot be destroyed while a published buffer has users or while any non-current buffer remains outstanding.

Example using a dynamic Mutex:

VOID WorkerTask(VOID *args)
{
    RK_UNUSEARGS

    RK_ERR err;
    RK_MUTEX_HANDLE mutexHandle = NULL;

    err = kMutexCreate(&mutexHandle, RK_PRIO_INHERITANCE);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kMutexLock(mutexHandle, RK_WAIT_FOREVER);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* protected work */

    err = kMutexUnlock(mutexHandle);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kMutexDestroy(&mutexHandle);
    K_ASSERT(err == RK_ERR_SUCCESS);
    K_ASSERT(mutexHandle == NULL);
}

9. Inter-Task Communication: Signals and Messages

Inter-Task Communication (ITC) refers to the mechanisms that enable tasks to coordinate/cooperate/synchronise by means of sending or receiving information that falls into two logical categories: Signals Tokens or Messages.

Pure Signal/Event:
- An operation that conveys no data/token. It only has effect if the target task/object is on a WAITING state; otherwise it is lost.
Event/Signal Tokens:
- A Token is registered. It can accumulate up to a limit (as will be explained in Event Flags and Semaphores).
Messages:
- A Message is a means of coordinating and exchanging information altogether. Unlike Signals, each message may convey a different information on the same inter-task communication object.

There is no operation that tests for a signal token availability and doesn’t consume it.

On the other hand, a message can be read but not consumed.

9.1. Semaphores

Semaphore Control Block
Counter (Unsigned Integer)
Maximum Value
Waiting Queue

Semaphores are public kernel objects for signalling and waiting on countable events. Any task can wait or signal a semaphore.

A semaphore S is a nonnegative integer variable, apart from the operations it is subjected to. S is initialised to a nonnegative value. The two operations, called P and V, are defined as follows:

P(S): if S > 0 then S := S-1, else the process is suspended until S > 0.

V(S): if there are processes waiting, then one of them is resumed; else S := S+1.

(Dijkstra, 1968)

V() in RK0 semaphores maps to post() and P() to pend().

9.1.1. Counting Semaphore and Binary Semaphores

The typical use case for semaphores is as a "credit tracker": use pend() to consume a credit and post() to return a credit (for example, free slots in a queue). These are Counting Semaphores.

A semaphore that is intended never to saturate can use a maximum value of UINT32_MAX.

A Binary Semaphore is a Counting Semaphore with maximum value 1: the state is either available or unavailable. They are often used for task-to-task or ISR-to-task synchronisation. Binary Semaphores initialised to 1 can act as Mutual Exclusion (Mutex) Semaphores . In RK0 a Binary Semaphore is not a Mutex Lock

9.1.2. Semaphores in RK0

To initialise a semaphore in RK0, provide two values: initial count and maximum count. When the counter is at maximum, post() does not increment it and returns RK_ERR_SEMA_FULL.

This return code is not negative, so it is not an handled as an error, but normally for a counting semaphore if you established an upper bound and is signalling more times than that, it means credits are not being consumed.

query() inspects current state: non-negative means current count; negative means number of tasks waiting.

The operation for flushing a semaphore (waking all pending tasks) was deprecated on V0.16.0. Now only Sleep Queues have wake/flush().

9.1.2.1. Usage Example: Producer-consumer general solution

A general solution for a producer-consumer relationship, considers a buffer with K items, K>=1 slots, uses Semaphores both as credit-trackers and mutual exclusion mechanisms.

When items are inserted and removed from a memory region, whose capacity is bounded to K items, the following invariant holds:

0 < (Number of Inserted Items) – (Number of Extracted Items) < K.

/* a ring buffer of items */
#define BUFSIZ (K)
static ITEM_t buf[BUFSIZ]={0};
static UINT getIdx = 0U;
static UINT putIdx = 0U;
/* getIdx==putIdx==0 could either mean FULL or EMPTY for a regular
circular buffer with wrap-around. When using semaphores they define the state.
*/

RK_SEMAPHORE  itemSema; /* counting semaphore for number of items in the buffer */
RK_SEMAPHORE  slotSema; /* counting semaphore for number of free slots in the buffer */
RK_SEMAPHORE  mutexSema; /* binary semaphore for mutual exclusion (it is not a LOCK, there is no ownership notion) */


VOID kApplicationInit(VOID)
{

  /*buffer is initialised empty */
    kSemaphoreInit
    (   &itemSema,
        0,   /* no item  */
        K    /*max items */
    );

    kSemaphoreInit
    (   &slotSema,
        K, /* K free slots */
        K  /* max slots */
    );

    /* and free */
    kSemaphoreInit
    (   &mutexSema,
        1, /* free to access */
        1  /* 1 max task allowed */
    );


VOID PutItem(ITEM_t const * const insertItemPtr)
{
    RK_ERR err = -1;

    /* wait for room */
    err = kSemaphorePend(&slotSema, RK_WAIT_FOREVER);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* wait for availability */
    err = kSemaphorePend(&mutexSema,  RK_WAIT_FOREVER);
    K_ASSERT(err == RK_ERR_SUCCESS);
    buf[putIdx] = *insertItemPtr;
    putIdx += 1U;
    putIdx %= BUFSIZ;

    /* signal availability */
    err = kSemaphorePost(&mutexSema);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* signal item */
    err = kSemaphorePost(&itemSema);
    K_ASSERT(err == RK_ERR_SUCCESS);
}


 VOID GetItem(ITEM_t * const extractItemPtr)
{
    RK_ERR err = -1;
    /* wait for an item */
    err = kSemaphorePend(&itemSema, RK_WAIT_FOREVER);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* wait for availability */
    err = kSemaphorePend(&mutexSema,  RK_WAIT_FOREVER);
    K_ASSERT(err == RK_ERR_SUCCESS);

    *extractItemPtr = buf[getIdx];
    getIdx += 1U;
    getIdx %= BUFSIZ;

    /* signal availability */
    err = kSemaphorePost(&mutexSema);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* signal room */
    err = kSemaphorePost(&slotSema);
    K_ASSERT(err == RK_ERR_SUCCESS);
}

The solution above has Put() and Get() as blocking methods.

If the producer and the consumer run at different rates, eventually, they will synchronise to the lowest rate.

The numbers below are from a run with a buffer of 32 items (integers being incremented are the produced data).

The producer is twice faster than the consumer. Initially at every 2 insertions there is a single remove.

Put 59 <-
Put 60 <-
------
Got 30 ->
------
Put 61 <-
Put 62 <-
------
Got 31 ->
------
Put 63 <-
Put 64 <-
--------
Got 32  | ->
Put 65  . <-
       <x>[Full Queue, Producer blocks]
Got 33  | ->  [Consumer unblocks producer...]
Put 66  . <-
       <x>[Full Queue]
Got 34  | ->  [Consumer unblocks producer...]
Put 67  . <-
       <x>[Full Queue]

When two tasks at different rates insert and remove from the same buffer, and both operations are blocking, eventually they will run at the pace of the lowest task.

9.1.2.2. Usage Example: Bi-lateral synchronisation

This example is important to grasp the difference between Signal Tokens, Messages and Pure Events.

We need to create a dependency between Task1 and Task2: neither task may continue beyond the synchronisation point until both operations have completed. How this could be done with semaphores?

/*PSEUDOCODE*/

/* Binary Semaphores*/
SEMA work1Sema;
SEMA work2Sema;

/* Both binary semaphores are initialised to 0. */
AppInit():
BinarySemaInit(work1Sema, 0U);
BinarySemaInit(work2Sema, 0U);

/* Tasks synchronising so both proceed after work1 AND work2 completes */
Task1():
        while(TRUE)
        {

           work1();
           /* a V/post/signal on semaphore does not block */
           while(V(work1Sema) == RK_ERR_SEMA_FULL)
           {
                /* task2 hasnt consumed the token yet */
                kSleep(1);
           }  /* signal/post on work1sema */
           P(work2Sema);  /* wait/pend on work2sema */


Task2():
        while(TRUE)
        {
           work2();
           while(V(work2Sema) == RK_ERR_SEMA_FULL)
           {
                kSleep(1);
           }  /* signal/post on work1sema */
           P(work2Sema);

        }

What would be the case if V(work1/2Sema) blocked waiting until the Token was consumed? That would be Message semantics.

9.2. Sleep Queue

Sleep Queue Control Block
Task Waiting Queue

Sleep Queues are the crudest synchronisation primitive in RK0, because they are stateless (there is no token).

Sleep Queues are not standalone Condition Variable as we know it (e.g, from Pthreads) .

Unlike Semaphores, kSleepQueueSleep() unconditionally switches the caller to a SLEEPING state until the queue is signalled (or broadcasted). Calling a sleep() with RK_NO_WAIT is meaningless — because it has no try semantics.

Sleep Queue names usually reflect the condition associated with the monitor wait loop, or the action tasks will execute once signalled.

A signal() wakes the higher priority task. Different from Semaphores, Sleep Queues support waking several tasks at once via wake(n) — a broadcast. A wake(n) wakes at most n tasks if any. This provides some control over the the always questionable overhead of broadcast signals. If n=0 it will flush.

If broadcasting from an ISR, the operation is deferred for the PostProcessingTask, to keep the ISR short.

A broadcast harms overall responsiveness. Avoid flushing, mainly from interrupts

A query() returns the number of sleeping tasks.

Another particular operation for Sleep Queues is kSleepQueueUnready(): it moves a READY task to a Sleep Queue. This is done to prevent a task from being scheduled. Task states other than READY are not affected by this operation. Please note that using this operation is literally poking the scheduler, so it is not something one should expect to use frequently. kSleepQueueReady() reverts it.

9.3. Mutex Lock

Mutex Control Block
Locked State (Boolean)
Owner
Protocol Flag (`RK_PRIO_NONE` / `RK_PRIO_INHERITANCE`)
Waiting Queue
Mutex Node (list node within the owner TCB)

Some regions are critical and must not be executed by more than one task at once. Acquiring (lock()) a mutex before entering and releasing (unlock()) after leaving makes the region mutually exclusive.

A Mutex Lock is a binary semaphore with ownership: once a task locks a mutex only that task can unlock it.

If a task tries to acquire a locked mutex, it switches to BLOCKED until the owner unlocks it. When released, the highest-priority waiter is dequeued first. Unlike semaphores, unlocking by a non-owner is invalid and rejected.

Mutexes are only for mutual exclusion; they are not signalling primitives.

RK0 mutexes are non-recursive. Re-entrant locking of the same mutex returns RK_ERR_MUTEX_REC_LOCK and is considered a fault.

9.3.1. Priority Inversion and Priority Inheritance Protocol (PIP)

Let TH, TM, and TL be three tasks with priority high (H), medium (M) and low (L), respectively. Say TH is dispatched and blocks on a mutex that 'TL' has acquired (i.e.: "TL is blocking TH").

If 'TM' does not need the resource, it will run and preempt 'TL'. And, by transition, 'TH'.

From now on, 'TH' has an unbounded waiting time because any task with priority higher than 'L' that does not need the resource indirectly prevents it from being unblocked — awful.

The PIP avoids this unbounded waiting. It is characterised by an invariant, simply put:

PIP Invariant: At any instant a Task assumes the highest priority among the tasks it is blocking.

If employed on the situation described above, task TM cannot preempt TL, whose effective priority would have been raised to 'H'.

It is straightforward to reason about this when you consider the scenario of a single mutex.

When locks nest, the protocol also needs to be:

Transitive: if T1 blocks T2 and T2 blocks T3, the highest priority (T3) must propagate back to T1 through T2.

This is the hard part of a correct implementation: updates must preserve the invariant across changing wait chains and multiple mutexes.

This blog shows an even more intricate case of priority inversion handling.

RK0 Blog: Transitive Priority Inheritance on RK0 mutexes

Below, a case in which locks nest:

/* Task1 has the Highest nominal priority */
/* Task2 has the Medium nominal priority */
/* Task3 has Lowest nominal priority */

/* Note Task3 starts as 1 and 2 are delayed */

RK_DECLARE_TASK(task1Handle, Task1, stack1, STACKSIZE)
RK_DECLARE_TASK(task2Handle, Task2, stack2, STACKSIZE)
RK_DECLARE_TASK(task3Handle, Task3, stack3, STACKSIZE)


RK_MUTEX mutexA;
RK_MUTEX mutexB;

VOID kApplicationInit(VOID)
{
    K_ASSERT(!kTaskInit(&task1Handle, Task1, RK_NO_ARGS, "Task1", stack1, \
        STACKSIZE, 1, RK_PREEMPT));
    K_ASSERT(!kTaskInit(&task2Handle, Task2, RK_NO_ARGS, "Task2", stack2, \
        STACKSIZE, 2, RK_PREEMPT));
    K_ASSERT(!kTaskInit(&task3Handle, Task3, RK_NO_ARGS, "Task3", stack3, \
        STACKSIZE, 3, RK_PREEMPT));

/* mutexes initialised with priority inheritance enabled */
    kMutexInit(&mutexA, RK_PRIO_INHERITANCE);
    kMutexInit(&mutexB, RK_PRIO_INHERITANCE);
}


VOID Task3(VOID *args)
{
    RK_UNUSEARGS
    while (1)
    {
        printf("@ %lums: [TL] Attempting to LOCK 'A' | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kMutexLock(&mutexA, RK_WAIT_FOREVER);

        printf("@ %lums: [TL] LOCKED 'A' (in CS) | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kDelay(60); /* <-- important */

        printf("@%lums: [TL] About to UNLOCK 'A' | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kMutexUnlock(&mutexA);

        printf("--->");
        printf("@%lums: [TL] Exit CS | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kSleep(4);
    }
}

VOID Task2(VOID *args)
{
    RK_UNUSEARGS
    while (1)
    {
        kSleep(5);

        printf("@%lums: [TM] Attempting to LOCK 'B' | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);
        kMutexLock(&mutexB, RK_WAIT_FOREVER);

        printf("@%lums: [TM] LOCKED 'B', now trying to LOCK 'A' | Eff: %d | Nom: %d\r\n",
               kTickGet(), runPtr->priority, runPtr->prioNominal);
        kMutexLock(&mutexA, RK_WAIT_FOREVER);

        printf("@%lums: [TM] LOCKED 'A' (in CS) | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);
        kMutexUnlock(&mutexA);

        printf("@%lums: [TM] UNLOCKING 'B' | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kMutexUnlock(&mutexB);

        printf("--->");

        printf("@%lums: [TM] Exit CS | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);
    }
}

VOID Task1(VOID *args)
{
    RK_UNUSEARGS
    while (1)
    {
        kSleep(2);

        printf("@%lums: [TH] Attempting to LOCK 'B'| Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kMutexLock(&mutexB, RK_WAIT_FOREVER);

        printf("@%lums: [TH] LOCKED 'B' (in CS)  | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);

        kMutexUnlock(&mutexB);

        printf("--->");

        printf("@%lums: [TH] Exit CS | Eff: %d | Nom: %d\r\n", kTickGet(),
               runPtr->priority, runPtr->prioNominal);
    }
}

Result and comments:

>>>> TL locks 'A'. Higher priority tasks are sleeping. <<<<

@ 14720ms: [TL] Attempting to LOCK 'A' | Eff: 3 | Nom: 3
@ 14720ms: [TL] LOCKED 'A' (in CS) | Eff: 3 | Nom: 3

@14721ms: [TM] Attempting to LOCK 'B' | Eff: 2 | Nom: 2

>>>> TM acquires 'B' and is blocked by TL on 'A'. TL inherits TM's  priority. <<<<

@14721ms: [TM] LOCKED 'B', now trying to LOCK 'A' | Eff: 2 | Nom: 2

>>>> TH will blocked by TM on 'B': <<<<

@14722ms: [TH] Attempting to LOCK 'B'| Eff: 1 | Nom: 1

>>>> TM inherits TH's priority. TL inherits TH's priority via TM. <<<<

@14780ms: [TL] About to UNLOCK 'A' | Eff: 1 | Nom: 3

>>>> Upon unlocking 'A', TL is preempted by TM. It means TL's priority has been restored, as it is no longer blocking a higher priority task. <<<<


>>>> Now TM acquires 'A' <<<<

@14780ms: [TM] LOCKED 'A' (in CS) | Eff: 1 | Nom: 2

>>>> After releasing 'A', but before releasing 'B', TM's priority is still '1', as it is blocking TH while holding 'B'. <<<<

@14780ms: [TM] UNLOCKING 'B' | Eff: 1 | Nom: 2

>>>> Upon unlocking 'B' TM is preempted by TH. (TM's priority has been restored.) <<<<

@14780ms: [TH] LOCKED 'B' (in CS)  | Eff: 1 | Nom: 1

>>> RESULT: even though priority inversion was enforced, tasks leave the nested lock ordered by their nominal priority. <<<

--->@14780ms: [TH] Exit CS | Eff: 1 | Nom: 1
--->@14780ms: [TM] Exit CS | Eff: 2 | Nom: 2
--->@14780ms: [TL] Exit CS | Eff: 3 | Nom: 3

Importantly, the worst-case time is bounded by the time the lowest priority task holds a lock (60 ms in the example: 14720ms → 14780ms).

As for each priority update we check each waiting queue for each mutex a task owns, the time-complexity is linear O(owner*mutex). But, typically no task ever holds more than a few mutexes. Yet, one should not be encouraged to nest locks if not needed.

9.3.2. Choosing Between Mutex Protocols

RK0 chooses the protocol per mutex:

kMutexInit(&plainLock, RK_PRIO_NONE);
kMutexInit(&sharedLock, RK_PRIO_INHERITANCE);

Protocol	Priority source	Choose it when
Plain Mutex (`RK_PRIO_NONE`)	None.	The region is short, inversion is known to be harmless, or the caller wants ownership-only mutual exclusion.
Priority Inheritance (`RK_PRIO_INHERITANCE`)	Tasks actually blocked on the mutex, including transitive owner chains.	The resource may be held by a lower-priority task while a higher-priority task waits for it.

Passing any other numeric protocol value to kMutexInit() returns RK_ERR_INVALID_PARAM.

9.3.3. Locks vs Binary Semaphores

In RK0 terminology a Lock has ownership property. Once a task enters a region, only that task can unlock the Mutex object. A Mutex Semaphore is a counting semaphore that counts up to 1, and that is why it provides mutual exclusion, but far from optimal.

If there is no ownership any task can signal that semaphore (thus the region is now free); and with no ownership, priority inversion cannot be handled.

Anyway, binary semaphores are considerably faster than mutexes and can handle mutual exclusion on very simple scenarios - say, two tasks with the same priority accessing a shared buffer.

9.4. Task Event Register

Within Task Control Block
Event Register Value (`RK_EVENT_FLAG` type)
Required Events (`RK_EVENT_FLAG` type)
Satisfy Condition (options: `RK_EVENT_ALL` or `RK_EVENT_ANY`)

Each Task Control Block stores a 32-bit event register (an ULONG, typedefed as RK_EVENT_FLAG). A bit set within a task’s event register means another task or ISR has signalled an occurrence. The meaning is application-defined.

As they are not a public kernel object, only the task itself wait for a combination of binary tokens on its Event Register. Any other task can set.

A get/pend/wait operation for a Signal Token always consumes it. There is no service in RK0 that tests for the presence of a Signal and does not consume.

A query will return the state of an object, that is different.

If you need to check for data and not consume it, that is Message semantics.

Operations:
- A set(receiverTask, inputFlags) operation is always a OR of inputFlags over the current value stored on the Event Register. Thus, it is only able to set new tokens, not to clear.
- A get(requiredFlags, ALL/ANY, storePtr, timeout) will check if ALL or ANY of the requiredFlags are set. If so, the required flags are cleared and the task returns successful. If not, the task either blocks or returns immediately with a positive return code value. In the case conditions are satisfied, if storePtr is not NULL the values on the event register are copyied to the indicated address before being cleared so the task can inspect which flags were set — specially useful when using ANY.
- An eventClear(taskHandle, mask) will clear the bits marked as 1 in mask.
- An eventQuery(taskHandle, storeAddr) will inspect the current status of the event register on a task. In both operations if taskHandle is NULL, the API considers the caller as the target task handle.
- For convenience there are macros encoding the bit position as a 32-bit number. e.g., RK_EVENT_1 equals 0x00000001, …, RK_EVENT_6 equals 0x00000020; RK_EVENT_32 equals 0x80000000 and RK_ALL_EVENTS equals 0xFFFFFFFF.

9.5. Disabling Task Switching (Scheduler Lock)

Scheduler 'lock' here means that the caller task will not be preempted until calling 'unlock'. Think of it as 'no other task will preempt this region'.

Thus there is a potential priority inversion, but it is bounded. If a higher-priority task is readied while the scheduler is locked, the context switch happens immediately after unlocking.

It does not mean that interrupts are disabled, so ISRs can still preempt the running task, and even higher priority tasks will switch to READY if conditions are satisfied.

This mechanism is not suitable if you are protecting data from priorities altogether.

Often, we need a task to perform operations without being preempted. A mutex serialises access to a code region but does not prevent a task from being preempted while operating on data. Depending on the case, this can lead to inconsistent data state.

An aggressive way is to disable interrupts globally. For kernel services often it is the only way to keep data integrity. On the higher level it is feasible for very short operations and/or when you need to protect data from interrupts altogether.

A less aggressive approach is to make the task non-preemptible with kSchLock() before entering the critical region and kSchUnlock() when leaving. This way, interrupts are still being sensed, and even higher-priority tasks might switch to a ready state, but the running thread will not be preempted.

Note that for locking/unlocking the scheduler the global interrupts will be disabled for the time to increment/decrement a counter, therefore, if your atomic operation is as short as that (3 to 4 cycles), disabling/enabling global interrupts is a better alternative.

To add to the discussion, when two threads need to access the same data to 'read-modify-write', a lock-free mechanism is the LDREX/STREX operations of ARMv7M (or more generally C11 atomics). They do not avoid preemptions, and particularly in ARMv7m, if the data is touched by an ISR before the store-exclusive concludes, the ownership is lost. Typically used for multi-core spin-locking.

9.6. Monitors and the Condition Variable Model

Task Events and semaphores work by atomically updating state and testing predicates that control execution flow (for example, pend on a semaphore with count 0 blocks the caller).

A critical region guarded by a lock is either free or taken. What if we need to wait on a richer condition? We express this condition on a shared variable and check for it within a critical region.

If the condition is not satisfied, we need to block until it is, but we need to release the lock before sleeping, otherwise, the task that could change the condition and wake us up would be blocked by the lock we are holding.

To prevent that, the sleeping task releases the lock and goes to sleep atomically (from the task preemption perspective) — kSchLock()/kSchUnlock() are particularly suitable here.

If we create a data structure with state variables, a Mutex lock, the Sleep Queues associated with each condition, and a set of operations acting over this structure, we have an ADT. This ADT is called a Monitor.

9.6.1. Monitor Invariants

A Monitor needs to respect two invariants:

a single task can be active within a monitor;
only the active task within a monitor can check or change its state.

Given the above invariants, how do we keep a single active task within a monitor if the active task is the one waking other tasks?

This comes down to the Signalling Discipline.

9.6.2. Signalling Discipline

At any moment a single task can be active within a monitor. When the sleeping task is signalled, there are 3 common disciplines to follow: signal-and-leave (Hansen), signal-and-wait (Hoare) or signal-and-continue (Mesa).

Arguably, the most common is signal-and-continue — rather than leaving or suspending itself the active task might continue within the monitor. That is possible if the active task holds a lock the waking task needs to acquire to enter. Upon leaving, the active task must release the lock.

The major implication is that by the time the woken task enters the monitor, the condition it was waiting for might no longer be true. It sounds odd because a Monitor is about encapsulating a conditional critical region so no outsiders change its state. But, either a flush, a bad design — or a preemption anomaly — can violate that somehow.

Mesa Monitor has a typical test-loop pattern:

 --- snippet ---
 while (condition is FALSE)
 {
    /*unlock-wait sequence:*/

     ATOMIC_BEGIN /* scheduler lock, for instance */

     unlock(mutex);   /*the atomic unlock-sleep we referred earlier */
     sleep(condition)

     ATOMIC_END / *scheduler unlock */

     lock(mutex);
     /* when waking, the while clause is tested again */
 }
 --- snippet ---

9.6.3. Condition Variable Model

The Condition Variable Model allows a task to wait within a monitor construct and, when active, operate using signal(), wait() and broadcast() while respecting the monitor invariant.

Sleep Queues are like the seminal Condition Variable as introduced by Hoare:

Note that a condition "variable" is neither true nor false; indeed, it does not have any stored value accessible to the program. In practice, a condition variable will be represented by an (initially empty) queue of processes which are currently waiting on the condition; but this queue is invisible both to waiters and signallers. This design of the condition variable has been deliberately kept as primitive and rudimentary as possible (…)

(Monitors: An Operating System Structuring Concept, Hoare, 1974)

RK0 does not have a POSIX-like Condition Variable primitive. Sleep Queues are supposed to be combined with mutexes to create Monitor-like mechanisms.

There are helpers that follow the Mesa semantics (the same used in Pthreads):

kCondVarInit(&sleepQueue, &mutexLock)
kCondVarWait(&sleepQueue, &mutexLock, timeout)
kCondVarSignal(&sleepQueue)
kCondVarBroadcast(&sleepQueue)

kCondVarWait() is the monitor wait helper. The predicate is the surrounding test loop; the Sleep Queue only provides the suspension queue. When using it, a Mesa testing-loop reduces to:

  while(!condition)
  {
     kCondVarWait(&sleepQueue, &monitorLock, timeout);
  }

Besides providing the atomic unlock-sleep-relock sequence, this helper gives timeout to kSleepQueueSleep(). The timeout bounds the condition wait; the final mutex reacquisition uses RK_WAIT_FOREVER so the monitor invariant is restored before returning.

If you need a monitor policy different from Mesa, you can build it from the same primitives.

kCondVar* are task-context APIs and cannot be called from ISRs.

9.6.3.1. Usage Example: Synchronisation Barrier

A given number of tasks must reach a point in the program before all can proceed, so every task calls a barrWait(&barrier) to synchronise at the barrier, waiting until the number of required tasks is met.

When a task enters the barrier and increases the counter so it meets the required number, it broadcast a signal to all tasks sleeping within the monitor.

A new round starts.

(Interestingly, a barrier solves the need for mutual coincidence, the very opposite of mutual exclusion.)

In application.c in the repo, you can find a more realistic implementation that assigns time outs and implements this pattern using both shared state and message passing paradigms.

/* Synchronisation Barrier */

typedef struct
{
    RK_MUTEX lock;
    RK_SLEEP_QUEUE allSynch;
    UINT count; /* number of tasks in the barrier */
    UINT round; /* increased every time all tasks synch */
    UINT nRequired; /* number of tasks required */
} Barrier_t;

VOID BarrierInit(Barrier_t *const barPtr, UINT nRequired)
{
    kMutexInit(&barPtr->lock, RK_PRIO_INHERITANCE);
    kSleepQueueInit(&barPtr->allSynch);
    barPtr->count = 0;
    barPtr->round = 0;
    barPtr->nRequired = nRequired;

}

VOID BarrierWait(Barrier_t *const barPtr)
{
    UINT myRound = 0;
    kMutexLock(&barPtr->lock, RK_WAIT_FOREVER);

    /* save round number */
    myRound = barPtr->round;
    /* increase count on this round */
    barPtr->count++;

    if (barPtr->count == barPtr->nRequired)
    {
        /* reset counter, inc round, broadcast to sleeping tasks */
        barPtr->round++;
        barPtr->count = 0;
         kCondVarBroadcast(&barPtr->allSynch);
    }
    else
    {
        /* sequence: a proper wake signal might happen after inc round */
        while ((UINT)(barPtr->round - myRound) == 0U)
        {
            RK_ERR err = kCondVarWait(&barPtr->allSynch, &barPtr->lock,
                                      RK_WAIT_FOREVER);
            K_ASSERT(err==RK_ERR_SUCCESS);
        }
    }
    kMutexUnlock(&barPtr->lock);
}


#define N_REQUIRED 3

Barrier_t syncBarrier;

VOID kApplicationInit(VOID)
{

    K_ASSERT(!kTaskInit(&task1Handle, Task1, RK_NO_ARGS, "Task1", stack1, STACKSIZE, 2, RK_PREEMPT));
    K_ASSERT(!kTaskInit(&task2Handle, Task2, RK_NO_ARGS, "Task2", stack2, STACKSIZE, 3, RK_PREEMPT));
    K_ASSERT(!kTaskInit(&task3Handle, Task3, RK_NO_ARGS, "Task3", stack3, STACKSIZE, 1, RK_PREEMPT));
    BarrierInit(&syncBarrier, N_REQUIRED);
}
VOID Task1(VOID* args)
{
    RK_UNUSEARGS
    while (1)
    {
        kPuts("Task 1 is waiting at the barrier...\r\n");
        BarrierWait(&syncBarrier);
        kPuts("Task 1 passed the barrier!\r\n");
        kSleep(8);
    }
}

VOID Task2(VOID* args)
{
    RK_UNUSEARGS
    while (1)
    {
        kPuts("Task 2 is waiting at the barrier...\r\n");
        BarrierWait(&syncBarrier);
        kPuts("Task 2 passed the barrier!\r\n");
        kSleep(5);
    }
}

VOID Task3(VOID* args)
{
    RK_UNUSEARGS
    while (1)
    {
        kPuts("Task 3 is waiting at the barrier...\r\n");
        BarrierWait(&syncBarrier);
        kPuts("Task 3 passed the barrier!\r\n");
        kSleep(3);
    }
}

Note the sequence tasks run before entering the monitor and the sequence they leave. They leave ordered by priority when the flush happens because the mutexes enforces that queue discipline. The priority of Task2 is lower than Task1, although its request rate is higher (5 vs 8 ticks delay), so it leaves first.

9.6.3.2. Usage Example: Readers Writers Lock

Several readers and writers share a piece of memory. Readers can concurrently access the memory to read; a single writer is allowed (otherwise, data would be corrupted).

When a writer finishes, it checks for any readers waiting. If there is, the writer flushes the readers waiting queue. If not, it wakes a single writer, if any. When the last reader finishes, it signals a writer.

Every read or write operation begins with an acquire and finishes with a release.

PS: This RWLock implementation has a reader-preference policy, as when a writer finishes, it flushes sleeping readers. When the last reader finishes, it will signal writer waiting queue.

/* RW-Lock */

/* a single writer is allowed if there are no readers */
/* several readers are allowed if there is no writer*/
typedef struct
{
    RK_MUTEX     lock;
    RK_SLEEP_QUEUE writersGo;
    RK_SLEEP_QUEUE readersGo;
    INT          rwCount; /* number of active readers if > 0 */
                          /* active writer if -1             */

}RwLock_t;

VOID RwLockInit(RwLock_t *const rwLockPtr)
{

    kMutexInit(&rwLockPtr->lock, RK_PRIO_INHERITANCE);
    kSleepQueueInit(&rwLockPtr->writersGo);
    kSleepQueueInit(&rwLockPtr->readersGo);
    rwLockPtr->rwCount = 0;
}

/* A writer can acquire if  rwCount = 0 */
/* An active writer is indicated by rwCount = -1; */
VOID RwLockAcquireWrite(RwLock_t *const rwLockPtr)
{
    kMutexLock(&rwLockPtr->lock, RK_WAIT_FOREVER);
    /* if different than 0, there are either writers or readers */
    /* sleep to be signalled */
    while (rwLockPtr->rwCount != 0)
    {
         kCondVarWait(&rwLockPtr->writersGo, &rwLockPtr->lock, RK_WAIT_FOREVER);
        /* mutex is locked when waking up*/
    }
    /* woke here, set an active writer */
    rwLockPtr->rwCount = -1;
    kMutexUnlock(&rwLockPtr->lock);
}

/* a writer releases, waking up all waiting readers, if any */
/* if there are no readers, a writer can get in */
VOID RwLockReleaseWrite(RwLock_t *const rwLockPtr)
{
    kMutexLock(&rwLockPtr->lock, RK_WAIT_FOREVER);

    rwLockPtr->rwCount = 0; /* indicate no writers*/

    /* if there are waiting readers, flush */
    ULONG nWaitingReaders=0;
    kSleepQueueQuery(&rwLockPtr->readersGo, &nWaitingReaders);
    if (nWaitingReaders > 0)
    {
        /* condVarBroadcast is just an alias for an event flush */
         kCondVarBroadcast(&rwLockPtr->readersGo);
    }
    else
    {
        /* wake up a single writer if any */
         kCondVarSignal(&rwLockPtr->writersGo);
    }
    kMutexUnlock(&rwLockPtr->lock);
}

/* a reader can acquire if there are no writers */
VOID RwLockAcquireRead(RwLock_t *const rwLockPtr)
{
    kMutexLock(&rwLockPtr->lock, RK_WAIT_FOREVER);
    /* if there is an active writer, sleep */
    while (rwLockPtr->rwCount < 0)
    {
         kCondVarWait(&rwLockPtr->readersGo, &rwLockPtr->lock, RK_WAIT_FOREVER);
        /* mutex is locked when waking up*/
    }
    /* increase rwCount, so its > 0, indicating readers */
    rwLockPtr->rwCount ++;
    kMutexUnlock(&rwLockPtr->lock);
}

/* a reader releases and wakes a single writer */
/* if it is the last reader */
VOID RwLockReleaseRead(RwLock_t *const rwLockPtr)
{
    kMutexLock(&rwLockPtr->lock, RK_WAIT_FOREVER);
    rwLockPtr->rwCount --;
    if (rwLockPtr->rwCount == 0)
    {
         kCondVarSignal(&rwLockPtr->writersGo);
    }
    kMutexUnlock(&rwLockPtr->lock);
}

In the image below, 4 tasks — a fast writer (Task 1), a slow writer (Task 4) and two readers (Task3 is faster than Task2) — reading from and writing to a shared UINT variable:

9.7. Context-switching cost, blocking and priority assignment

The detailed exposition of Monitor-like constructions might lead the reader to understand we consider it a 'silver-bullet'. Not at all. Indeed, Monitor constructs and the Mesa semantics particularly can impose heavy context switching activity and they are a pattern optimised for the general-case. For general-purpose operating systems this is gold. We care about responsiveness, then we need to look at this pattern with a different perspective. Golden rule:

Avoid triggering context-switches for no useful work.

A task signalling a higher priority task’s Sleep Queue is saying — 'the condition you may need to run was probably satisfied'. Once the task is ready, the scheduler will dispatch it. It cannot see beyond that, and it does not re-test the monitor predicate.

If a lower priority task readies a higher priority task while holding a lock the signalled task needs, it is explictly causing a priority inversion.

Mutexes in RK0 will bound priority inversion, because kernel mechanisms are supposed handle the worst cases — the application could not avoid. Whether this will become problem or not, depends on the case. As it is expensive, Priority inversions are not to be injected on the code — this is an anti-pattern.

With that in mind, do not take the (functional, but generic) patterns here as optimal recipes without considering the priority of the tasks that are using monitor-like schemes, and your application demands. The run-time model is already simple to aid on reasoning.

Finally, this discussion applies to any blocking inter-task comunication, including message-passing mechanisms exposed on the next-section. ---

9.8. Buffered Message Passing: Queues/Mailboxes

Message Queue Control Block
Owner Task
Buffer Address
Message Size
Number of Mesages
Write Position
Read Position
Notify callback
Waiting Receivers
Waiting Senders

Message Queues (RK_MESG_QUEUE) are public message-passing kernel objects that successful is defined by being able to deposit a message on a buffer.

Each message queue has a backing storage holding N messages of a fixed-size S. We say that C=NxS[Words] is the queue capacity.

Message Queues transmit by copy. Sending and receiving are either blocking or non-blocking.

Each message queue will preserve discrete message history up to queue capacity when the producer uses blocking semantics.

If the producer occasionally outruns the consumer, a message queue amortises bursts or consumer lateness without missing data. A faster consumer, on the other hand, will eventually block on an empty queue; it does not drop messages. Drops occur when the producer outruns the consumer and uses non-blocking sends.

This was demonstrated using semaphores on the producer-consumer problem.

In practice, buffering gives time-data correlation only within queue capacity. Over long runs, effective throughput is bounded by the slowest stage, so either the producer blocks on a full queue — what we call backpressure — or the system accepts that some data will be missed.

Rule of thumb: long-term throughput equals the slowest stage; buffers only absorb short-term mismatch and jitter.

9.8.1. Size of a Message

Each declared queue has a fixed message-size at initialisation, and can assume, 1, 2, 4 or 8 WORDs (4, 8, 16, 32 BYTEs). This constraint is intentional. Word-aligned copies are faster, predictable and safer for type casting.

(A word-aligned single copy will take ~5 cycles in Cortex-M3/4/7, and ~6 cycles on Cortex-M0/M0+.)

9.8.2. Mailbox

A Mailbox is a Message Queue with depth 1. It is not a different kernel object: RK_MBOX is an alias for the same control block used by RK_MESG_QUEUE, initialised with one message slot.

A Mailbox is still buffered communication. A successful send means the message was copied into the single slot, not that a receiver has accepted it. An application can emulate a CSP-style synchronous handoff with a Mailbox plus a second acknowledgement wait, but the kernel then sees two separate dependencies rather than one unbuffered handoff. Use that pattern only when the application accepts the resulting timeout, cancellation, and priority semantics.

The usual queue operations still apply, but the single-slot capacity gives the mailbox two useful behaviours:

kMesgQueuePostOvw() can overwrite the current message. This operation is only valid for single-message queues; on queues with depth greater than one it returns `RK_ERR_MESGQ_NOT_A_MBOX.
kMboxBroadcast() with kMboxBroadcastRecv() can deliver one copied message to every task that is already blocked as a broadcast receiver.

9.8.2.1. Overwrite mailbox

An overwrite mailbox is useful when the receiver only needs the newest pending value and old values may be discarded. A producer can post a new value without waiting for the receiver to consume the previous one:

static RK_MBOX sensorMbox;
RK_DECLARE_MESG_QUEUE_BUF(sensorMboxBuf, SensorSample, 1U)

VOID kApplicationInit(VOID)
{
    RK_ERR err = kMboxInit(&sensorMbox, sensorMboxBuf,
                           RK_MESGQ_MESG_SIZE(SensorSample));
    K_ASSERT(err == RK_ERR_SUCCESS);
}

VOID ProducerTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        SensorSample sample = ReadSensor();
        RK_ERR err = kMesgQueuePostOvw(&sensorMbox, &sample);
        K_ASSERT(err == RK_ERR_SUCCESS);
        kSleepRelease(RK_MS_TO_TICKS(100));
    }
}

This is still a single-receiver mailbox pattern. If multiple tasks read from the same mailbox with normal receive operations, they compete for the one stored message.

9.8.2.2. Broadcast mailbox

A broadcast mailbox uses the same single slot differently. Receivers call kMboxBroadcastRecv() and block. A sender then calls kMboxBroadcast(). If no broadcast receiver is blocked, no message is deposited and the call returns RK_ERR_BUFFER_EMPTY.

When there are blocked broadcast receivers, the mailbox stores one copy of the message and wakes those receivers. Each targeted receiver receives the same message. The last receiver drains the mailbox slot, so the next broadcast can be accepted.

/*declare RK_MBOX stateMbox, backed by a storage stateMboxBuf
which size is a 1 * sizeof(StateFrame)*/
RK_DECLARE_MBOX(stateMbox, stateMboxBuf, StateFrame)

VOID BroadcasterTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        UINT nReceivers = 0U;
        StateFrame frame = BuildStateFrame();

        RK_ERR err = kMboxBroadcast(&stateMbox, &frame, &nReceivers);
        if (err == RK_ERR_SUCCESS)
        {
            logPost("broadcast to %u receivers", nReceivers);
        }
        kSleepRelease(RK_MS_TO_TICKS(200));
    }
}

VOID ReceiverTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        StateFrame frame;
        RK_ERR err = kMboxBroadcastRecv(&stateMbox, &frame,
                                        RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);
        ProcessFrame(&frame);
    }
}

Mailbox broadcast targets currently blocked broadcast receivers. It is not a history buffer and it does not remember a broadcast for receivers that arrive later. Use a Message Queue when history matters, or MRM when the intended model is many readers getting the latest state with explicit buffer lifetime management.

Note that a 1:N communication with last-message, and non-blocking send/recv has a dedicated service (the MRM) since it is a straight fit for servo-control loops.

9.8.3. Send to Front (Jam)

A normal send() deposits the message on queue’s tail. A jam() deposits the message on the queue’s head. For mailboxes a jam() is meaningless.

9.8.4. Notify callback

A callback can be registered for to be trigerred when a queue sends a message successfuly. This is a means of notification. This callback must be short, non-blocking, and normally will be a signalling a semaphore or a setting an event on a task.

9.8.5. Usage Examples

9.8.5.1. Mail Queue pattern

A mail queue is a very useful pattern for message-passing.

It is done by combining a Memory Partition and 1-word-size message-queue, with a `N>1`.

Sender allocates, receiver frees the partition. This keep integrity and low overhead as the copy from sender to queue and queue to receiver is reduced to 1-word.

Below, snippets of the Application Logger facility that uses this pattern.

/* Application Logger pattern */

/* standard log structure */
struct log
{
    RK_TICK t; /* timestamp */
    CHAR s[LOGLEN]; /*formatted string */
    UINT    level; /* level 0=message, 1=fault */
} K_ALIGN(4);

typedef struct log Log_t;

/* logger mem allocator + mem pool */
static RK_MEM_PARTITION qMem;
static Log_t logBufPool[LOGPOOLSIZ] K_ALIGN(4);

/* backing buffer for the logger queue */
/* (messages are 1-word-size, number equals the pool) */
RK_DECLARE_MESG_QUEUE_BUF(logQBuf, VOID *, LOGPOOLSIZ)

/* logger mail queue */
static RK_MESG_QUEUE logQ;

/* a sender will allocate a buffer write the log
message and enqueue on the mail queue, not-blocking
if the queue is full it returns the buffer immediately
if the memory pool is empty it drops the operation */

```c
/* excerpt of logPost(...) */
VOID logPost(/*formatted string */)
    ---snip---
    Log_t *logPtr = (Log_t*)kMemPartitionAlloc(&qMem);
    RK_BARRIER
    if (logPtr) /* available buffer */
    {

       < fill the buffer >

        /* use task name (port owner) */
        if (kMesgSend(logTaskHandle, &p, RK_NO_WAIT) != RK_ERR_SUCCESS)
        {
            /* queue is full, deallocate buf */
            RK_ERR err = kMemPartitionFree(&qMem, &p);
            K_ASSERT(err==RK_ERR_SUCCESS);
        }

    }
    ---snip---

/* excerpt of the logger task */

static VOID LoggerTask(VOID *args)
{
    RK_UNUSEARGS
    while (1)
    {

        VOID *recvPtr = NULL;

        /* drain the queue: keep receiving while successful */
        while (kMesgRecv(&recvPtr, RK_WAIT_FOREVER) == RK_ERR_SUCCESS)
        {
            < print buffer contents >

            /* deallocate */
            RK_ERR err = kMemPartitionFree(&qMem, recvPtr);
            K_ASSERT(err == RK_ERR_SUCCESS);

        }
    }
}

This is one of the many ways of using the Mail Queue pattern.

The entire implementation can be seen at app\logger.c.

9.8.5.2. Queue Select using Notify Callback

A task is receiving from many queues and need to know which one has been able to complete.

The notifyCbk(queue*) is executed every time a send is successful. In this case it is using an Event Signal to a task. The Signals Flag indicate the queue number - as a contract - of which queue has completed sends. Note that as sends may coalesce, while a flag caps at 1, the consumer will drain each queue until it is empty, or it is preempted. There are many options here; a bi-lateral synchronisation could be employed; a counting semaphore could be used so queues are read after a threshold value, etc.

/* Many-to-1 queue channels */

/* Consumer Select queue based on its event
flags, that a succesfull send triggers  */

#define LOG_PRIORITY 4 /* keep logger as lowest-priority user task */
#define STACKSIZE 256

#define NQUEUES 3 /* number of queues */
#define QSIZ 8 /* depth of each queue */

#define Q0_FLAG   RK_EVENT_1 // (1<<0)
#define Q1_FLAG   RK_EVENT_2 // (1<<1)
#define Q2_FLAG   RK_EVENT_3 // (1<<2)
#define QFLAGS   (ULONG)(Q0_FLAG | Q1_FLAG | Q2_FLAG)

typedef struct
{
    RK_TASK_HANDLE producer;
    UINT payload;
} MESG_t;

/* Succesful Send callbacks */
/* each callback follows this pattern */
static inline
VOID sendNotify0(RK_MESG_QUEUE *qPtr)
{
    (VOID)qPtr;
    kTaskEventSet(consumerHandle, Q0_FLAG);
    /* Q1 flag for queue1 and so forth */
}

/* each callback in installed using kMesgQueueInstallSendCbk
on kApplicationInit() */

/* helper to send */
static inline
VOID enqueueSample(RK_MESG_QUEUE *qPtr UINT payload)
{
    MESG_t mesg = {
        .payload = payload,
        .producer = RK_RUNNING_HANDLE,
    };
    RK_ERR err = kMesgQueueSend(qPtr, &mesg, RK_WAIT_FOREVER);
    K_ASSERT(err == RK_ERR_SUCCESS);
}


VOID Prod0Task(VOID *args)
{
    RK_UNUSEARGS
    UINT seq = 0U;

    while (1)
    {
        enqueueSample(&queues[0], seq++);
        kSleepRelease(25); /* fast producer */
    }
}

VOID Prod1Task(VOID *args)
{
    UINT seq = 0U;
    RK_UNUSEARGS

    while (1)
    {
        enqueueSample(&queues[1], seq++);

        /* every fourth sample, also tickle the third queue */
        if ((seq & 0x3U) == 0U)
        {
            enqueueSample(&queues[2], seq);
        }

        kSleepRelease(60);
    }
}

/* Consumer listens on all queues, selecting those on its signal flags. */
VOID ConsumerTask(VOID *args)
{
    RK_UNUSEARGS

    MESG_t recv = {0};
    ULONG flags = 0UL;

    while (1)
    {
        flags = 0UL;
        kTaskEventGet(QFLAGS, RK_EVENT_FLAGS_ANY, &flags,
                           RK_WAIT_FOREVER);

        for (UINT i = 0; i < NQUEUES; ++i)
        {
            if (flags & (1UL << i))
            {
                while (kMesgQueueRecv(&queues[i], (VOID*)&recv, RK_NO_WAIT) ==
                       RK_ERR_SUCCESS)
                {
                    logPost("Q%u <- sender=%s payload=%u", i,
                            RK_TASK_NAME(recv.producer), recv.payload);
                }
            }
        }
    }
}

       0 ms :: Q1 <- sender=Prod1 payload=0
     250 ms :: Q0 <- sender=Prod0 payload=1
     500 ms :: Q0 <- sender=Prod0 payload=2
     600 ms :: Q1 <- sender=Prod1 payload=1
     750 ms :: Q0 <- sender=Prod0 payload=3
    1000 ms :: Q0 <- sender=Prod0 payload=4
    1200 ms :: Q1 <- sender=Prod1 payload=2
    1250 ms :: Q0 <- sender=Prod0 payload=5
    1500 ms :: Q0 <- sender=Prod0 payload=6
    1750 ms :: Q0 <- sender=Prod0 payload=7
    1800 ms :: Q1 <- sender=Prod1 payload=3
    1800 ms :: Q2 <- sender=Prod1 payload=4
    2000 ms :: Q0 <- sender=Prod0 payload=8
    2250 ms :: Q0 <- sender=Prod0 payload=9
    2400 ms :: Q1 <- sender=Prod1 payload=4
    2500 ms :: Q0 <- sender=Prod0 payload=10
    2750 ms :: Q0 <- sender=Prod0 payload=11
    3000 ms :: Q0 <- sender=Prod0 payload=12
    3000 ms :: Q1 <- sender=Prod1 payload=5
    3250 ms :: Q0 <- sender=Prod0 payload=13
    3500 ms :: Q0 <- sender=Prod0 payload=14
    3600 ms :: Q1 <- sender=Prod1 payload=6
    3750 ms :: Q0 <- sender=Prod0 payload=15
    4000 ms :: Q0 <- sender=Prod0 payload=16
    4200 ms :: Q1 <- sender=Prod1 payload=7
    4200 ms :: Q2 <- sender=Prod1 payload=8
    4250 ms :: Q0 <- sender=Prod0 payload=17
    4500 ms :: Q0 <- sender=Prod0 payload=18
    4750 ms :: Q0 <- sender=Prod0 payload=19
    4800 ms :: Q1 <- sender=Prod1 payload=8
    5000 ms :: Q0 <- sender=Prod0 payload=20
    5250 ms :: Q0 <- sender=Prod0 payload=21

9.9. Task-to-task Message Passing

Task-to-task Message Passing is direct communication addressed to a task handle. The receiver endpoint is task-backed: endpoint state is carried by the receiver TCB, and callers target the task that owns the endpoint.

RK0 exposes three task-to-task contracts:

Synchronous (Rendezvous): one sender offers one bounded payload and blocks until the receiver copies it.
Invocation (Extended Rendezvous): a caller offers one request and remains blocked while the server accepts it, processes it, and replies.
Asynchronous: a sender transfers ownership of a pool-backed RK_MESG object to a receiver endpoint without blocking.

A task may be initialized to handle Synchronous/Invocation messages or Asynchronous Direct Messages, but not both. This keeps one clear receive policy per task endpoint.

9.9.1. Synchronous (Rendezvous)

Receiver TCB metadata
Maximum payload size
Pending sender queue
Pending message address
Pending message size
Waiting receiver storage address

Synchronous Message is unbuffered rendezvous. The sender is not satisfied by depositing data into kernel storage; it blocks until the receiver copies the offered payload into receiver-owned storage.

A successful send means:

The receiving task has copied the payload into its receive buffer.

It is a named operation: the message is sent to a task handle, not to a queue object. There is no standalone rendezvous object and no kernel payload buffer. The receiver task’s TCB holds the endpoint state, and blocked sender TCBs carry the pending source pointer and actual message size while they are waiting.

A receiving task becomes a Synchronous Message endpoint by calling kSynchMesgInit(receiverTask, maxMesgBytes). maxMesgBytes must be nonzero and a multiple of RK_WORD_SIZE (32-bit). It is the largest payload the receiver endpoint accepts. Each sender then calls kSynchMesgSend(receiverTask, mesgPtr, lengthBytes, timeout), where lengthBytes is the actual byte count for that handoff.

The sender’s source storage must remain valid until kSynchMesgSend() returns, because the kernel copies directly from that storage during the handoff. The receiver does not need to know the actual message size before it receives; kSynchMesgRecv(recvPtr, mesgBytesPtr, timeout) stores the copied byte count through mesgBytesPtr on success. mesgBytesPtr may be NULL if the receiver does not need the actual count. The destination buffer must still be large enough for the endpoint maximum configured at initialisation.

If the sender times out before 'meeting' the receiver, the receiver will never know about that message. A recv operation with RK_NO_WAIT is successful only if there is a message to receive. A send with RK_NO_WAIT is successful only if the receiver is waiting for a message.

When a high-priority sender blocks on a low-priority receiver, the receiver’s effective priority is raised to the highest waiting sender priority until the handoff completes or the waiting set changes.

Sender timeouts are part of the data-integrity contract. If a sender times out, the pending message is invalidated for the receiver. A later receive cannot consume that stale pointer. From the receiver’s perspective, a timed-out sender is no longer offering a valid message.

It also follows the single-authority rule:

a sender that owns any mutex is rejected;
the receiver is rejected if it owns any mutex when calling kSynchMesgRecv().

Those cases return RK_ERR_TASK_INVALID_ST. This prevents mixing direct mutex-protected access with a task-owned unbuffered handoff for the same resource.

9.9.1.1. Usage example

The example shows the main contract: the producer cannot outrun the controller. No new sample is handed off until the controller receives the previous one.

typedef struct
{
    UINT seq;
    ULONG value;
} Sample_t;

RK_DECLARE_TASK(producerHandle, ProducerTask, producerStack, STACKSIZE)
RK_DECLARE_TASK(controllerHandle, ControllerTask, controllerStack, STACKSIZE)

static Sample_t sample;

VOID kApplicationInit(VOID)
{
    RK_ERR err = kTaskInit(&controllerHandle, ControllerTask, RK_NO_ARGS,
                             "Ctrl", controllerStack, STACKSIZE, 2,
                             RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kTaskInit(&producerHandle, ProducerTask, RK_NO_ARGS,
                      "Prod", producerStack, STACKSIZE, 2, RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kSynchMesgInit(controllerHandle, sizeof(Sample_t));
    K_ASSERT(err == RK_ERR_SUCCESS);
}

VOID ProducerTask(VOID *args)
{
    RK_UNUSEARGS

    sample.seq = 0U;
    while (1)
    {
        sample.seq++;
        sample.value = 1000UL + sample.seq;

        /*
         * Send means send-and-wait-until-copied. The source storage remains
         * valid until kSynchMesgSend() returns.
         */
        RK_ERR err = kSynchMesgSend(controllerHandle, &sample,
                                     sizeof(sample), RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);

        kSleepPeriodic(RK_MS_TO_TICKS(100));
    }
}

VOID ControllerTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        Sample_t received = {0U, 0UL};
        ULONG mesgBytes = 0UL;
        RK_ERR err = kSynchMesgRecv(&received, &mesgBytes, RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);
        K_ASSERT(mesgBytes == sizeof(received));

        applyControl_(received.value);
    }
}

9.9.2. Invocation (Extended Rendezvous)

Invocation uses the same task-backed Synchronous Message endpoint, but extends the rendezvous with a server-side accept and a reply copied back to the caller. It models a direct client-server operation:

caller: kSynchMesgCall()
            send request
            block waiting for accept/reply
                                  server: kSynchMesgAccept()
                                          copy request
                                          process request
caller: reply copied        <---  server: kSynchMesgReply()
caller returns

A one-way synchronous send is just the first stage of that relation. The difference is completion:

kSynchMesgSend() completes when the receiver has copied the request.
kSynchMesgCall() completes when the server has replied, or when the caller timeout expires.

Invocation metadata
Server task endpoint state
Pending caller queue
Active caller
Request source address and size
Reply destination address and maximum size
Call state

The client fills an RK_SYNCH_ATTR with request and reply buffer metadata, then calls kSynchMesgCall(serverTask, attrPtr, timeout) or the alias kSynchMesgInvoke(). The request pointer must remain valid until the server accepts the call. The reply buffer belongs to the caller and must remain valid until kSynchMesgCall() returns.

The server calls kSynchMesgAccept(callPtr, recvPtr, reqBytesPtr, timeout). On success, the request has been copied into recvPtr, callPtr contains the active call metadata, and the caller remains blocked. The server then calls kSynchMesgReply(callPtr, replyPtr, replyBytes) to complete the extended rendezvous.

If a bounded invocation times out before the server accepts it, the caller is removed from the server’s pending caller list and returns RK_ERR_TIMEOUT. If it times out after accept, the active request is abandoned from the caller’s perspective. The server may still call kSynchMesgReply() to close the rendezvous, but no reply is copied to the caller.

While servicing an accepted invocation, the server runs at the effective priority required by the active caller relation. That priority contribution ends when the server replies or completes an abandoned call.

9.9.2.1. Usage example

typedef struct
{
    UINT opcode;
    ULONG value;
} Request_t;

typedef struct
{
    RK_ERR status;
    ULONG result;
} Reply_t;

RK_DECLARE_TASK(clientHandle, ClientTask, clientStack, STACKSIZE)
RK_DECLARE_TASK(serverHandle, ServerTask, serverStack, STACKSIZE)

VOID kApplicationInit(VOID)
{
    RK_ERR err = kTaskInit(&serverHandle, ServerTask, RK_NO_ARGS,
                           "Srv", serverStack, STACKSIZE, 2, RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kTaskInit(&clientHandle, ClientTask, RK_NO_ARGS,
                    "Cli", clientStack, STACKSIZE, 3, RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kSynchMesgInit(serverHandle, sizeof(Request_t));
    K_ASSERT(err == RK_ERR_SUCCESS);
}

VOID ClientTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        Request_t req = {1U, 42UL};
        Reply_t reply = {RK_ERR_SUCCESS, 0UL};
        ULONG replyBytes = 0UL;
        RK_SYNCH_ATTR attr = {&req, sizeof(req), &reply,
                              sizeof(reply), &replyBytes};

        RK_ERR err = kSynchMesgCall(serverHandle, &attr, RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);
        K_ASSERT(replyBytes == sizeof(reply));

        kSleepPeriodic(RK_MS_TO_TICKS(100));
    }
}

VOID ServerTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        RK_SYNCH_CALL_DATA call = {0};
        Request_t req = {0U, 0UL};
        ULONG reqBytes = 0UL;

        RK_ERR err = kSynchMesgAccept(&call, &req, &reqBytes,
                                      RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);
        K_ASSERT(reqBytes == sizeof(req));

        Reply_t reply = {RK_ERR_SUCCESS, req.value + 1UL};
        err = kSynchMesgReply(&call, &reply, sizeof(reply));
        K_ASSERT(err == RK_ERR_SUCCESS);
    }
}

9.9.3. Asynchronous Named Message-Passing

Asynchronous Named message passing pass message pointers from a pool of RK_MESG objects. The receiver endpoint is task-backed: the receiver TCB holds the pending-message list and the current receive wait state. The message itself is an RK_MESG header plus a fixed-size payload block allocated from an application-provided pool. Typically sender allocates a message buffer, receiver frees a message buffer.

Note messages are _enqueued — so a Task has queue with a capacity of C = N x WORDs.

If the receiver is not already waiting for a matching sender, the message object is linked into that receiver’s endpoint queue.

A successful asynchronous send means:

The RK_MESG object is no longer owned by the sender; it is owned by the receiver side, either already delivered to a waiting receiver or queued on the receiver endpoint.

kMesgSend() has no timeout argument and it does not block for endpoint queue space. The bounded resource is the message pool: if no message block is available, kMesgAlloc(poolPtr, mesgPtrPtr, timeout) is where the sender can use RK_NO_WAIT, a bounded timeout, or RK_WAIT_FOREVER.

No application payload bytes are copied by kMesgSend() or kMesgWait(). Only the RK_MESG pointer changes ownership. After a successful kMesgSend(), the sender must not read, write, or free that message unless a later application-level protocol returns it.

If kMesgSend() returns an error, ownership has not been transferred by the public contract. The sender is still responsible for the allocated message and must either retry according to its protocol or return the block to its pool with kMesgFree(). This rule prevents pool leaks on invalid destination, uninitialized endpoint, invalid message state, or other send failures.

Asynchronous Direct Message metadata
Message pool pointer
Sender task handle and ID
Receiver task handle and ID
Payload capacity
Message state
Receiver endpoint queue
Receiver wait sender filter

The application declares storage with RK_DECLARE_MESG_POOL(), or provides equivalent aligned storage, then initializes the pool with kMesgPoolInit(poolPtr, storagePtr, payloadBytes, nMesg, ceilingPrio). Each block has one RK_MESG header followed by payloadBytes bytes of application payload. kMesgAlloc(poolPtr, mesgPtrPtr, timeout) writes an owned message block to *mesgPtrPtr and returns RK_ERR_SUCCESS when a block is available. RK_NO_WAIT returns RK_ERR_BUFFER_EMPTY if the pool is empty. A bounded wait returns RK_ERR_TIMEOUT if no block returns before the timeout expires. When a message is freed while a task is blocked in kMesgAlloc() on the same pool, the freed block is handed directly to one waiting allocator. The payload address is obtained with kMesgPayload(), kMesgPayloadConst(), or the typed RK_MESG_PAYLOAD(mesgPtr, Type) helper.

To bound priority inversion on a full message pool, the application can enable a pool priority ceiling through the ceilingPrio argument. While a task owns at least one message from that pool, its effective priority is raised to at least the configured ceiling until ownership moves to another task or the message is freed. Choose the ceiling as the highest-priority task that may wait for that pool. Pass RK_MESG_PRIO_CEILING_NONE to disable the ceiling. A timeout on kMesgAlloc() limits the caller’s wait duration; the ceiling limits priority inversion while lower-priority tasks hold scarce message blocks.

A receiver becomes an asynchronous endpoint by calling kMesgEndpointInit(receiverTask). A task may own either a Synchronous/Invocation endpoint or an Asynchronous Direct Message endpoint, but not both. This is enforced at endpoint initialization; the second policy returns RK_ERR_HAS_OWNER.

The receiver obtains one message pointer with kMesgWait(fromTaskHandle, mesgPtrPtr, timeout). The sender filter can be RK_ANY_TASK or one specific task handle. kMesgWait() first scans messages already queued on the endpoint. Therefore, if a matching message was posted before the wait, the receiver does not block. With a specific sender filter, messages from other senders remain queued for later waits.

When no matching message is queued:

RK_NO_WAIT returns RK_ERR_BUFFER_EMPTY;
a bounded timeout returns RK_ERR_TIMEOUT if no matching message arrives;
RK_WAIT_FOREVER blocks until a matching sender posts.

If the receiver is already blocked in kMesgWait() and a matching message is sent, the message is delivered directly to the receiver’s destination pointer and that task is readied. Since send is nonblocking, Asynchronous Direct Message does not create a sender-to-receiver priority-inheritance relation. Normal scheduling still applies after a send readies a higher-priority receiver.

After receive, sender metadata is available from the message header:

kMesgGetSenderHandle(mesgPtr) returns the sending task handle recorded at kMesgSend();
kMesgGetSenderID(mesgPtr, senderIDPtr) writes the sender task ID and returns an error if no valid sender metadata is present.

The receiver returns a received message to its original pool with kMesgFree(). Once kMesgFree() succeeds, the pointer is invalid for the application and must not be reused.

Asynchronous Direct Message is compiled only when RK_CONF_ASYNCH_MESG == ON and RK_CONF_MESG_QUEUE == ON. The latter is required because the endpoint uses RK0 queue/list infrastructure internally; it does not make the endpoint a public RK_MESG_QUEUE.

9.9.3.1. Usage example

#define STACKSIZE 256U
#define ASYNC_POOL_DEPTH 4U

typedef struct
{
    UINT src;
    UINT seq;
    ULONG value;
} AsyncPayload_t;

RK_DECLARE_TASK(rxHandle, RxTask, rxStack, STACKSIZE)
RK_DECLARE_TASK(txHandle, TxTask, txStack, STACKSIZE)
RK_DECLARE_MESG_POOL(asyncPool, asyncPoolBuf,
                     AsyncPayload_t, ASYNC_POOL_DEPTH)

VOID kApplicationInit(VOID)
{
    RK_ERR err = kTaskInit(&rxHandle, RxTask, RK_NO_ARGS,
                           "Rx", rxStack, STACKSIZE, 2, RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    err = kTaskInit(&txHandle, TxTask, RK_NO_ARGS,
                    "Tx", txStack, STACKSIZE, 3, RK_PREEMPT);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /*
     * Pool setup: ASYNC_POOL_DEPTH fixed-size RK_MESG blocks, each carrying one
     * AsyncPayload_t payload. Tx may block waiting for asyncPool, so any task
     * that owns an asyncPool message runs at least at Tx priority until it
     * frees or transfers the buffer.
     */
    err = kMesgPoolInit(&asyncPool, asyncPoolBuf,
                        sizeof(AsyncPayload_t), ASYNC_POOL_DEPTH, 3);
    K_ASSERT(err == RK_ERR_SUCCESS);

    /* Endpoint setup: Rx can now receive direct messages addressed to rxHandle. */
    err = kMesgEndpointInit(rxHandle);
    K_ASSERT(err == RK_ERR_SUCCESS);
}

VOID TxTask(VOID *args)
{
    RK_UNUSEARGS

    UINT seq = 0U;
    while (1)
    {
        /*
         * Allocation: Tx waits here for pool availability. Send remains
         * nonblocking with respect to endpoint queue space.
         */
        RK_MESG *mesgPtr = NULL;
        RK_ERR err = kMesgAlloc(&asyncPool, &mesgPtr, RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);
        K_ASSERT(mesgPtr != NULL);

        /* Fill the typed payload behind the RK_MESG header. */
        AsyncPayload_t *payloadPtr =
            RK_MESG_PAYLOAD(mesgPtr, AsyncPayload_t);
        payloadPtr->src = 1U;
        payloadPtr->seq = ++seq;
        payloadPtr->value = 1000UL + seq;

        err = kMesgSend(rxHandle, mesgPtr);
        if (err != RK_ERR_SUCCESS)
        {
            /*
             * Send failure did not transfer ownership, so Tx must not leak the
             * block. Do not touch mesgPtr after a successful send.
             */
            RK_ERR freeErr = kMesgFree(mesgPtr);
            K_ASSERT(freeErr == RK_ERR_SUCCESS);
        }
        K_ASSERT(err == RK_ERR_SUCCESS);

        kSleepPeriodic(RK_MS_TO_TICKS(100));
    }
}

VOID RxTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        RK_MESG *mesgPtr = NULL;

        /*
         * Receive: ANY accepts the first queued or future message from any sender.
         * On success, Rx owns mesgPtr and must eventually free it.
         */
        RK_ERR err = kMesgWait(RK_ANY_TASK, &mesgPtr, RK_WAIT_FOREVER);
        K_ASSERT(err == RK_ERR_SUCCESS);

        K_ASSERT(kMesgGetSenderHandle(mesgPtr) == txHandle);
        RK_PID senderID = 0U;
        err = kMesgGetSenderID(mesgPtr, &senderID);
        K_ASSERT(err == RK_ERR_SUCCESS);
        K_ASSERT(senderID == kTaskGetPID(txHandle));

        AsyncPayload_t *payloadPtr =
            RK_MESG_PAYLOAD(mesgPtr, AsyncPayload_t);
        consume_(payloadPtr->value);

        /* Return the received message to its originating pool. */
        err = kMesgFree(mesgPtr);
        K_ASSERT(err == RK_ERR_SUCCESS);
    }
}

9.10. Most-Recent Message Protocol (MRM)

MRM Control Block
MRM Buffer Allocator
Data Buffer Allocator
Current MRM Buffer Address
Data Size (Message Size)

MRM Buffer
Data Buffer Address
Readers Count

Data Buffer
Application-dependent

There is little practical difference between a message that does not arrive and one with no valid (stale) data. But when wrong (or stale) data is processed - e.g., to define a set point on a loop - a system can fail badly.

Design Choice: provide a broadcast asynchronous message-passing scheme that guarantees data freshness and integrity for all readers.

Benefits: The system has a mechanism to meet strict deadlines that cannot be predicted on design time.

Control loops reacting to unpredictable time events—like a robot scanning an environment or a drive-by-wire system—require a different message-passing approach. Readers cannot "look at the past" and cannot block. The most recent data must be delivered non-blocking and have guaranteed integrity.

As owner-bound queues, the MRM is a high-level mechanism. It was chosen to be provided as a kernel service, given its distinctive nature and suitability for RK0 applications.

9.10.1. Functional Description

An MRM works as a 1-to-many asynchronous Mailbox - that enables several readers to get the most recent deposited message with no integrity issues. Whenever a reader reads an MRM buffer, it will find the most recent data transmitted. It can also be seen as an extension of the Double Buffer pattern for a 1:N communication.

The core idea of the MRM protocol is that readers can only access the buffer that is classified as the 'most recent buffer'. After a writer publish() a message, that will be the only message readers can get() — any former message being processed by a reader was grabbed before a new publish() - and, from now on, can only be unget(), eventually returning to the pool.

To clarify further, the communication steps are listed:

A producer first reserves an MRM Buffer - the reserved MRM Buffer is not available for reading until it is published.
A message buffer is allocated and filled, and its address is within an MRM Buffer. The producer publishes the message. From now on, it is the most recent message. Any former published buffer is no longer visible to new readers
A reader starts by getting an MRM Buffer. A get() operation delivers a copy of the message to the reader’s scope. Importantly, this operation increases the number of readers associated to that MRM Buffer.

Before ending its cycle, the task releases (unget()) the buffer; on releasing, the kernel checks if the caller task is the last reader and if the buffer being released is not the current MRM Buffer.

If the above conditions are met, the unget() operation will return the MRM buffer to the pool. If there are more readers, OR if it is the current buffer, it remains available.

When the reserve operation detects that the most recent buffer still has readers, a new buffer is allocated to be written and published. If it has no readers, it is reused.

This way, the worst case is a sequence of publish() with no unget() at all — this would lead to the writer finding no buffer to reserve. This is prevented by making: N Buffers = N tasks + 1.

9.10.1.1. MRM Control Block Configuration

What might lead to some confusion when initialising an MRM Control Block is the need for two different pools:

One pool will be the storage for the MRM Buffers, which is the data structure for the mechanism.
Another pool is for the actual payload. The messages.

Both pools must have the same number of elements: the number of tasks communicating + 1.

The size of the data buffers is application-dependent - and is passed as a number of words. The minimal message size is 32-bit.
If using data structures, keep it aligned to 4 to take advantage of the performance of aligned memory.

9.10.1.2. Usage Example: Immediate state transfer (Car Speed)

Consider a modern car - speed changes is an event of interest for many modules. Let us consider three modules and how they should react when speed varies:

Cruiser Control: For the Cruiser Control, a speed increase might signify the driver wants manual control back, and it will likely turn off.
Windshield Wipers: If they are on, a speed change can affect the electric motor’s adjustments to the air resistance.
Radio: Speed changes reflect the aerodynamic noise - the radio volume might need adjustment.

As the variations are unpredictable, we need a mechanism to deliver the last speed in order of importance for all these modules. From highest to lowest priority we elencate Cruise, Wipers, and Radio. (Criteria: safety → comfort).

To emulate this scenario, we can write an application with a higher priority task that sleeps and wakes up at pseudo-random times to produce random values that represent the (unpredictable) speed changes.

The snippet below has 4 periodic tasks. Tasks are periodic using the kSleepRelease() primitive.

The producer publishes new data at a random interval, preempting whatever task is running at the moment.

Despite the randomness of the updates, the actuators should keep their rate while reading the most recent state, with no integrity issues.

typedef struct
{
    UINT speed;
    ULONG timeStamp;
} Mesg_t;

#define STACKSIZE 256
#define N_MRM (5)                          /* Number of MRMs N Tasks + 1 */
#define MRM_MESG_SIZE (sizeof(Mesg_t) / 4) /* In WORDS */
RK_MRM MRMCtl;                             /* MRM control block */
RK_MRM_BUF buf[N_MRM];                     /* MRM pool */
Mesg_t data[N_MRM];                        /* message data pool */

RK_DECLARE_TASK(speedSensorHandle, SpeedSensorTask, stack1, STACKSIZE)
RK_DECLARE_TASK(cruiserHandle, CruiserTask, stack2, STACKSIZE)
RK_DECLARE_TASK(wiperHandle, WiperTask, stack3, STACKSIZE)
RK_DECLARE_TASK(radioHandle, RadioTask, stack4, STACKSIZE)
volatile UINT seq = 0;
VOID kApplicationInit(VOID)
{

    kTaskInit(&speedSensorHandle, SpeedSensorTask, RK_NO_ARGS, "SpeedTsk",
                stack1, STACKSIZE, 1, RK_PREEMPT);

    kTaskInit(&cruiserHandle, CruiserTask, RK_NO_ARGS, "CruiserTsk", stack2,
                STACKSIZE, 2, RK_PREEMPT);

    kTaskInit(&wiperHandle, WiperTask, RK_NO_ARGS, "WiperTsk", stack3,
                STACKSIZE, 3, RK_PREEMPT);

    kTaskInit(&radioHandle, RadioTask, RK_NO_ARGS, "RadioTsk", stack4,
                STACKSIZE, 4, RK_PREEMPT);

    kMRMInit(&MRMCtl, buf, data, N_MRM, MRM_MESG_SIZE);

    logInit(5);
}

VOID SpeedSensorTask(VOID *args)
{
    RK_UNUSEARGS

    Mesg_t sendMesg = {0};
    while (1)
    {
        RK_TICK sleepTicks = ((RK_TICK)rand() % 18) + 1;
        kSleep(sleepTicks);
        RK_TICK currTick = kTickGetMs();
        UINT speedValue = (UINT)(rand() % 170) + 1;
        sendMesg.speed = speedValue;
        sendMesg.timeStamp = currTick;
        /* grab a buffer */
        RK_MRM_BUF *bufPtr = kMRMReserve(&MRMCtl);
        if (bufPtr != NULL)
        {
            K_ASSERT(!kMRMPublish(&MRMCtl, bufPtr, &sendMesg));
            printf("!!!!! @%lums SPEED UPDATE: %u mph\r\n", kTickGetMs(), speedValue);
            seq += 1;
        }
        else
        { /* cannot fail */
            logError("MRM protocol could not find a free buffer\r\n");
        }
        /* publish  */
    }
}

VOID CruiserTask(VOID *args)
{
    RK_UNUSEARGS
    Mesg_t recvMesg = {0};
    while (1)
    {
        RK_MRM_BUF *readBufPtr = kMRMGet(&MRMCtl, &recvMesg);
        if (readBufPtr)
        {
            logPost("CRUISER: (%u mph, %lu ms)", recvMesg.speed, recvMesg.timeStamp);

            kMRMUnget(&MRMCtl, readBufPtr);
        }
        kSleepRelease(4);
    }
}

VOID WiperTask(VOID *args)
{
    RK_UNUSEARGS
    Mesg_t recvMesg = {0};

    while (1)
    {

        RK_MRM_BUF *readBufPtr = kMRMGet(&MRMCtl, &recvMesg);
        if (readBufPtr)
        {
            logPost("WIPERS: (%u mph, %lu ms)", recvMesg.speed, recvMesg.timeStamp);

            kMRMUnget(&MRMCtl, readBufPtr);
        }
        kSleepRelease(7);
    }
}
VOID RadioTask(VOID *args)
{
    RK_UNUSEARGS
    Mesg_t recvMesg = {0};
    while (1)

    {

        RK_MRM_BUF *readBufPtr = kMRMGet(&MRMCtl, &recvMesg);

        if (readBufPtr)
        {
            logPost("RADIO: (%u mph, %lu ms) ", recvMesg.speed, recvMesg.timeStamp);
            kMRMUnget(&MRMCtl, readBufPtr);
        }
        kSleepRelease(11);
    }
}

Thus, different situations can happen:

All tasks read the updated pair (speed, time)
Not all tasks receive the updated pair because another update happens in between.
No tasks receive an update - because another happens too soon.
No update happens between in the period of a given task. It receives the same value. No problems.

All these cases are on the log:

Logs show: (last speed record, record time)

  !!!@120ms SPEED UPDATE: 164 mph
     120 ms :: CRUISER: (164 mph, 120 ms)
     140 ms :: WIPERS: (164 mph, 120 ms)
 !!! @150ms SPEED UPDATE: 80 mph
     160 ms :: CRUISER: (80 mph, 150 ms)
     200 ms :: CRUISER: (80 mph, 150 ms)
     210 ms :: WIPERS: (80 mph, 150 ms)
     220 ms :: RADIO: (80 mph, 150 ms)
     240 ms :: CRUISER: (80 mph, 150 ms)
  !!!@280ms SPEED UPDATE: 49 mph
     280 ms :: CRUISER: (49 mph, 280 ms)
     280 ms :: WIPERS: (49 mph, 280 ms)
     320 ms :: CRUISER: (49 mph, 280 ms)
     330 ms :: RADIO: (49 mph, 280 ms)
     350 ms :: WIPERS: (49 mph, 280 ms)
     360 ms :: CRUISER: (49 mph, 280 ms)
     400 ms :: CRUISER: (49 mph, 280 ms)
     420 ms :: WIPERS: (49 mph, 280 ms)
     440 ms :: CRUISER: (49 mph, 280 ms)
     440 ms :: RADIO: (49 mph, 280 ms)
  !!!@450ms SPEED UPDATE: 87 mph
     480 ms :: CRUISER: (87 mph, 450 ms)
     490 ms :: WIPERS: (87 mph, 450 ms)
     520 ms :: CRUISER: (87 mph, 450 ms)
 !!! @540ms SPEED UPDATE: 110 mph
     550 ms :: RADIO: (110 mph, 540 ms)
     560 ms :: CRUISER: (110 mph, 540 ms)
     560 ms :: WIPERS: (110 mph, 540 ms)
     600 ms :: CRUISER: (110 mph, 540 ms)
     630 ms :: WIPERS: (110 mph, 540 ms)
 !!! @640ms SPEED UPDATE: 22 mph
     640 ms :: CRUISER: (22 mph, 640 ms)
     660 ms :: RADIO: (22 mph, 640 ms)

The highlight is that controllers can keep their pace, while receiving fresh data - you can see it on the timestamp on the image.

Again, they might receive the same data more than once or miss samples; what is important is that they are not lagging and consuming stale data.

9.10.1.3. Usage Example: Cascaded Robot Servo Loop

This code illustrates a robot control system capable of exploring unknown objects by integrating visual and tactile information. In order to do so, the robot has to apply forces on the object surface and follow its contour by means of visual feedback.

The system is designed as 2 servo loops — which inputs are the current image frames (what the robot is seeing) and the current torque (the current force on its arm).

ForceTask: A sensory acquisition process periodically reads the force/torque sensor. This task runs @20ms, and if late the robot might become unstable by applying inadequate force/torque on the environment.
VisionTask: A visual process periodically reads the image memory filled by the camera frame grabber and computes the next exploring direction based on a user-defined strategy. A missed deadline for this task could cause the robot to move on a wrong direction or to stomp on the object object surface.
ControlTask: Based on computed path and required force, a robot control process computes the Cartesian set points for the controller. This information either moves the robot direction tangential to the object surface, or apply forces normal to the surface.
DisplayTask: A display task is used for telemetry. This is the less critical one in the sense that if late, quality of result degrades but nothing is damaged.

(Adapted from: Giorgio Buttazzo, Hard Real-Time Computing Systems (Chapter 11))

/* --- EXCERPT --- */

typedef struct
{
    LONG normalForceMN; /* normal contact force [mN] */
    LONG torqueMNm;     /* wrist torque around x [mNm] */
    UINT contact;       /* binary contact flag */
    RK_TICK tsMs;       /* sample timestamp [ms] */
} FORCE_MRM_MESG_t;

typedef struct
{
    LONG txMilli;   /* tangential x component in milli-units (1000 -> 1.0) */
    LONG tyMilli;   /* tangential y component in milli-units (1000 -> 1.0) */
    UINT strategy;  /* selected exploration strategy/phase */
    RK_TICK tsMs;   /* sample timestamp [ms] */
} PATH_MRM_MESG_t;

typedef struct
{
    LONG vxMmps; /* commanded tangential velocity x [mm/s] */
    LONG vyMmps; /* commanded tangential velocity y [mm/s] */
    LONG vzMmps; /* commanded normal velocity from force loop [mm/s] */
    LONG xMm;    /* integrated x position [mm] */
    LONG yMm;    /* integrated y position [mm] */
    LONG zMm;    /* integrated z position [mm] */
} ROBOT_CMD_MESG_t;

/* Two shared MRM channels: latest force and latest path guidance. */
static RK_MRM mrmForce;
static RK_MRM mrmPath;


/* priority assigned higher for the shortest period */
#define PERIOD_FORCE_MS 20U /* highest critical */

#define PERIOD_CONTROL_MS 28U /* middle critical */

#define PERIOD_VISION_MS 80U /* lowest critical */

#define PERIOD_DISPLAY_MS 100U /* soft real-time task */

VOID ForceTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        /*  acquire simulated force sample and publish latest  */
        RK_TICK nowMs = kTickGetMs();
        FORCE_MRM_MESG_t mesg = simulateForceSensor_(nowMs);
        RK_ERR err = publishLatest_(&mrmForce, &mesg);
        K_ASSERT(err == RK_ERR_SUCCESS);

        logPost("[FORCE ] Fn=%ldmN T=%ldmNm contact=%u", mesg.normalForceMN,
                mesg.torqueMNm, mesg.contact);

        sleepHardTask_(periodForceTicks, "FORCE");
    }
}

VOID VisionTask(VOID *args)
{
    RK_UNUSEARGS

    while (1)
    {
        /*  compute next tangential direction and publish latest one */
        RK_TICK nowMs = kTickGetMs();
        PATH_MRM_MESG_t mesg = simulatePathFromVision_(nowMs);
        RK_ERR err = publishLatest_(&mrmPath, &mesg);
        K_ASSERT(err == RK_ERR_SUCCESS);

        logPost("[VISION] path=(%ld,%ld) strategy=%u", mesg.txMilli,
                mesg.tyMilli, mesg.strategy);

        sleepHardTask_(periodVisionTicks, "VISION");
    }
}

VOID ControlTask(VOID *args)
{
    RK_UNUSEARGS
    ROBOT_CMD_MESG_t cmd = {0};

    while (1)
    {
        FORCE_MRM_MESG_t forceMsg;
        PATH_MRM_MESG_t pathMsg;

        /*  read latest force/path snapshots  */
        RK_MRM_BUF *forceBufPtr = kMRMGet(&mrmForce, &forceMsg);
        RK_MRM_BUF *pathBufPtr = kMRMGet(&mrmPath, &pathMsg);
        while (forceBufPtr == NULL || pathBufPtr == NULL)
        {
            /*   wait until initial samples are available. */
            forceBufPtr = kMRMGet(&mrmForce, &forceMsg);
            pathBufPtr = kMRMGet(&mrmPath, &pathMsg);

            kSleep(1);
            continue;
        }
        kMRMUnget(&mrmForce, forceBufPtr);
        kMRMUnget(&mrmPath, pathBufPtr);

        /* normal-axis loop: force error -> bounded z velocity command. */
        LONG forceErr = FORCE_DES_MN - forceMsg.normalForceMN;
        LONG vzMmps = clampLong_(forceErr / 20L, -60L, 60L);

        /* tangential motion from vision direction vector in milli-units. */
        LONG vxCmd = (pathMsg.txMilli * TANGENTIAL_VEL_MMPS) / 1000L;
        LONG vyCmd = (pathMsg.tyMilli * TANGENTIAL_VEL_MMPS) / 1000L;

        if (forceMsg.contact == 0U)
        {
            /* No contact: stop tangential exploration and approach surface. */
            vxCmd = 0L;
            vyCmd = 0L;
            vzMmps = 30L;
        }

        cmd.vxMmps = vxCmd;
        cmd.vyMmps = vyCmd;
        cmd.vzMmps = vzMmps;

        /*  commanded velocities to a simple discrete position estimate. */
        cmd.xMm += (vxCmd * (LONG)periodControlEffMs) / 1000L;
        cmd.yMm += (vyCmd * (LONG)periodControlEffMs) / 1000L;
        cmd.zMm += (vzMmps * (LONG)periodControlEffMs) / 1000L;

        /* sim sensor closed-loop : higher positive vzMmps raises normal force. */
        gSimNormalForceMN = clampLong_(
            gSimNormalForceMN + (vzMmps * (LONG)periodControlEffMs) / 2L,
            0L,
            12000L);

        logPost("[CONTROL] v=(%ld,%ld) vz=%ld Fn=%ld err=%ld pos=(%ld,%ld,%ld)",
                cmd.vxMmps, cmd.vyMmps, cmd.vzMmps, forceMsg.normalForceMN,
                forceErr, cmd.xMm, cmd.yMm, cmd.zMm);

        sleepHardTask_(periodControlTicks, "CONTROL");
    }
}

VOID DisplayTask(VOID *args)
{
    RK_UNUSEARGS

    RK_TICK anchor = kTickGet();
    while (1)
    {
        FORCE_MRM_MESG_t forceMsg;
        PATH_MRM_MESG_t pathMsg;

        /*  less critical task: consume latest state for telemetry/visualization only. */
        RK_MRM_BUF *forceBufPtr = kMRMGet(&mrmForce, &forceMsg);
        RK_MRM_BUF *pathBufPtr = kMRMGet(&mrmPath, &pathMsg);


        kMRMUnget(&mrmForce, forceBufPtr);
        kMRMUnget(&mrmPath, pathBufPtr);

        logPost("[DISPLAY] Fn=%ld path=(%ld,%ld) ts=(%lu,%lu)",
                forceMsg.normalForceMN, pathMsg.txMilli, pathMsg.tyMilli,
                (ULONG)forceMsg.tsMs, (ULONG)pathMsg.tsMs);

        kSleepUntil(&anchor, periodDisplayTicks);
    }
}

      0 ms :: [FORCE ] Fn=4300mN T=-350mNm contact=1
       0 ms :: [VISION] path=(1000,0) strategy=0
       0 ms :: [DISPLAY] Fn=4300 path=(1000,0) ts=(0,0)
       2 ms :: [CONTROL] v=(80,0) vz=35 Fn=4300 err=700 pos=(2,0,0)
      20 ms :: [FORCE ] Fn=4790mN T=-350mNm contact=1
      28 ms :: [CONTROL] v=(80,0) vz=10 Fn=4790 err=210 pos=(4,0,0)
      40 ms :: [FORCE ] Fn=5000mN T=-315mNm contact=1
      56 ms :: [CONTROL] v=(80,0) vz=0 Fn=5000 err=0 pos=(6,0,0)
      60 ms :: [FORCE ] Fn=5000mN T=-315mNm contact=1
      80 ms :: [FORCE ] Fn=5070mN T=-280mNm contact=1
      80 ms :: [VISION] path=(1000,0) strategy=0
      84 ms :: [CONTROL] v=(80,0) vz=-3 Fn=5070 err=-70 pos=(8,0,0)
     100 ms :: [FORCE ] Fn=5028mN T=-280mNm contact=1
     100 ms :: [DISPLAY] Fn=5028 path=(1000,0) ts=(100,80)
     112 ms :: [CONTROL] v=(80,0) vz=-1 Fn=5028 err=-28 pos=(10,0,0)
     120 ms :: [FORCE ] Fn=5084mN T=-245mNm contact=1
     140 ms :: [FORCE ] Fn=5084mN T=-245mNm contact=1
     140 ms :: [CONTROL] v=(80,0) vz=-4 Fn=5084 err=-84 pos=(12,0,0)
     160 ms :: [FORCE ] Fn=5098mN T=-210mNm contact=1
     160 ms :: [VISION] path=(1000,0) strategy=0
     168 ms :: [CONTROL] v=(80,0) vz=-4 Fn=5098 err=-98 pos=(14,0,0)
     180 ms :: [FORCE ] Fn=5042mN T=-210mNm contact=1
     196 ms :: [CONTROL] v=(80,0) vz=-2 Fn=5042 err=-42 pos=(16,0,0)
     200 ms :: [FORCE ] Fn=5084mN T=-175mNm contact=1
     200 ms :: [DISPLAY] Fn=5084 path=(1000,0) ts=(200,160)
     220 ms :: [FORCE ] Fn=5084mN T=-175mNm contact=1
     224 ms :: [CONTROL] v=(80,0) vz=-4 Fn=5084 err=-84 pos=(18,0,0)
     240 ms :: [FORCE ] Fn=5098mN T=-140mNm contact=1
     240 ms :: [VISION] path=(1000,0) strategy=0
     252 ms :: [CONTROL] v=(80,0) vz=-4 Fn=5098 err=-98 pos=(20,0,0)
     260 ms :: [FORCE ] Fn=5042mN T=-140mNm contact=1
     280 ms :: [FORCE ] Fn=5112mN T=-105mNm contact=1
     280 ms :: [CONTROL] v=(80,0) vz=-5 Fn=5112 err=-112 pos=(22,0,0)
     300 ms :: [FORCE ] Fn=5042mN T=-105mNm contact=1
     300 ms :: [DISPLAY] Fn=5042 path=(1000,0) ts=(300,240)
     308 ms :: [CONTROL] v=(80,0) vz=-2 Fn=5042 err=-42 pos=(24,0,0)

    .
    .
    .

   30800 ms :: [VISION] path=(0,1000) strategy=1
   30800 ms :: [DISPLAY] Fn=5112 path=(0,1000) ts=(30800,30800)
   30820 ms :: [FORCE ] Fn=5042mN T=140mNm contact=1
   30828 ms :: [CONTROL] v=(0,80) vz=-2 Fn=5042 err=-42 pos=(1116,1088,36)
   30840 ms :: [FORCE ] Fn=5084mN T=175mNm contact=1
   30856 ms :: [CONTROL] v=(0,80) vz=-4 Fn=5084 err=-84 pos=(1116,1090,36)
   30860 ms :: [FORCE ] Fn=5028mN T=175mNm contact=1
   30880 ms :: [FORCE ] Fn=5098mN T=210mNm contact=1
   30880 ms :: [VISION] path=(0,1000) strategy=1
   30884 ms :: [CONTROL] v=(0,80) vz=-4 Fn=5098 err=-98 pos=(1116,1092,36)
   30900 ms :: [FORCE ] Fn=5042mN T=210mNm contact=1
   30900 ms :: [DISPLAY] Fn=5042 path=(0,1000) ts=(30900,30880)
   30912 ms :: [CONTROL] v=(0,80) vz=-2 Fn=5042 err=-42 pos=(1116,1094,36)
   30920 ms :: [FORCE ] Fn=5084mN T=245mNm contact=1
   30940 ms :: [FORCE ] Fn=5084mN T=245mNm contact=1
   30940 ms :: [CONTROL] v=(0,80) vz=-4 Fn=5084 err=-84 pos=(1116,1096,36)
   30960 ms :: [FORCE ] Fn=5098mN T=280mNm contact=1

Interpreting the output:

From 30800 ms:

path=(0,1000) means direction is pure +Y, so controller sets v=(0,80) mm/s.
Control runs every 28 ms, so position update is: dy = (80 * 28) / 1000 = 2 mm (integer)
That is why y: 1088 → 1090 → 1092.
x stays 1116 because vx=0.
z command is small and negative (vz=-2, -4) since force is slightly above 5000 mN, but: dz = (vz * 28)/1000 truncates to 0. vz remains 36.
at 30900, ts = (30900, 30880) says the robot is using the Fn informed (at inst 30900) and the last path/strategy, which was @ 30880 ms.

While the simulation model used to exercise the control is naive, the log shows the cascaded server-control loop is answering to environment changes on a timely and ordered manner.

10. Error Handling

10.1. Fail fast

While tracing and error handling are yet to be largely improved (and that is when the 1.0.0 version will be released), currently RK0 employs a policy of failing fast in debug mode.

When Error Checking is enabled, every kernel call will be 'defensive', checking for correctness of parameters and invariants, null dereferences, etc.

In these cases is more useful to allow the first error to halt the execution by calling an Error Handler function to observe the program state.

A trace structure records the address of the running TCB, its current stack pointer, the link register (that is, the PC at kErrHandler was called), and a time stamp.

This record is on a .noinit RAM section, so it is visible if CPU resets. A fault code is stored in a global faultID and on the trace structure. Developers can hook in custom behaviour.

If the kernel is configured to not halt on a fault, but Error Checking is enabled, functions will return negative values in case of an error.

On the other hand, when Error Checking is disabled or NDEBUG is defined nothing is checked, reducing code size and improving performance.

(Some deeper internal calls have assertion. For those, only NDEBUG defined ensures they are disabled.)

10.2. Stack Overflow

Stack overflow is detected (not prevented) using a "stack painting" with a sentinel word. Stack Overflow detection is enabled by defining the assembler preprocessor __KDEF_STACKOVFLW when compiling.

As a matter of fact, sizing your stack is something you must do diligently when programming a system. I would say a mechanism for stack overflow detection is on the bottom of the list of 'must-have' features.

One can take advantage of the static task model - it is possible to predict offline the deepest call within any task. The compiler flag -fstack-usage creates .su files indicating the depth of every function within a module. This is an example of compilation-unit output:

core/src/ksema.c:34:8:kSemaphoreInit 88  static
core/src/ksema.c:74:8:kSemaphorePend    96  static
core/src/ksema.c:189:8:kSemaphorePost   88  static
core/src/ksema.c:306:5:kSemaphoreQuery  40  static
core/src/kmutex.c:128:8:kMutexInit  16  static
core/src/kmutex.c:165:8:kMutexLock  120 static
core/src/kmutex.c:325:8:kMutexUnlock    96  static
core/src/kmutex.c:425:6:kMutexQuery 56  static

These are the worst cases. Now, you identify the depth of the longest chain of calls for a task using these services and add a generous safety margin — 30%. The cap depends on your budget.

Importantly, you also have to size the System Stack. This initial size is defined in linker.ld by the symbol Min_Stack_Size. In this case, account for the depth of main(), kApplicationInit(), and all interrupt handlers; again, inspect the longest call chain depth. Assume interrupts always add to the worst static depth, and account for nested interrupts.

10.3. Deadlocks

Most deadlock avoidance patterns are unsuitable to our domain. We shall be disciplined. Here there is a golden rule.

Ordered Locking:

The golden rule for locking is acquiring resources in an unidirectional order throughout the entire application:

acquire(A);
acquire(B);
acquire(C);
   .
   .
   .
release(C);
release(B);
release(A);

This breaks circular waiting.

For instance:

TaskA:
   wait(R1);
   wait(R2);
    /* critical section */
   signal(R2);
   signal(R1);

TaskB:
   wait(R1);
   wait(R2);
    /* critical section */
   signal(R2);
   signal(R1);

But, if:

Task1:
    wait(R1);
    wait(R2);
    .
    .

TaskB:
    wait(R2);
    wait(R1);
    .
    .

There are some possible outcomes:

Deadlock:
- TaskA runs: acquires R1
- TaskB runs: acquires R2
- TaskA runs: tries to acquire R2 — blocked
- TaskB runs: tries to acquire R1 — blocked
No deadlock:
- TaskA runs: acquires R1
- TaskA runs: acquires R2 (nobody is holding R2)
- TaskA releases both; TaskB runs and acquires both (in either order)

Overall, there is no deadlock if tasks do not overlap in critical sections. That is why systems run for years without deadlocks and eventually: ploft.

Importantly, the PIP protocol in RK0 can handle reverse unlocking w.r.t. priority inheritance; but the protocol cannot prevent a circular wait.

The application might apply a design pattern that ensures A resource may only be locked if its resourceID is greater than the largest resourceID of any locked resource. It is not that easy because some resource IDs numbering will be better than others.