Real-Time Operating Systems

1The superloop and its limit

int main(void)
{
    init();
    for (;;) {
        read_sensors();       /*  2 ms */
        update_control();     /*  1 ms */
        log_to_flash();       /* 50 ms — the problem */
        handle_ui();          /*  1 ms */
    }
}

Everything runs in one sequence at one priority. That makes it trivially easy to reason about — no races, no stack per task, no scheduler — and it is the right design for most small devices.

The limit is visible above. read_sensors runs once every 54 ms, not every 2, because the flash write is in front of it. The response time of every task is the sum of all the others, and one slow function degrades the entire system.

Two ways out without an RTOS. Break the long operation into a state machine that yields each iteration — week 47's technique, on a microcontroller. Or move it into an interrupt, which only works when the work is short.

When neither is enough, a scheduler is the answer.

2Tasks

An RTOS gives each task its own stack and switches between them. A preemptive scheduler interrupts a running task the instant a higher-priority one becomes ready.

void sensor_task(void *arg)
{
    for (;;) {                                  /* a task never returns */
        read_sensors();
        vTaskDelay(pdMS_TO_TICKS(10));          /* yields the processor */
    }
}

xTaskCreate(sensor_task, "sensor", 256, NULL, 3, NULL);
/*                        name    stack  arg  priority  handle */
vTaskStartScheduler();                          /* never returns */

The examples here use FreeRTOS because it is the most widely deployed and runs under QEMU; Zephyr, ThreadX and RT-Thread differ in spelling, not in concept.

The distinction that matters: a blocked task costs nothing. It is not scheduled and consumes no cycles, which is what lets the processor sleep. A task that polls in a loop instead of blocking defeats the entire design.

3Scheduling

"Real-time" means predictable, not fast. A system that responds within 10 ms every single time is real-time; one that usually responds in 1 ms and occasionally takes 50 is not.

Assign priorities by deadline, not by importance. A task that must respond in 1 ms outranks one that must respond in 100 ms, even if the second is doing something more valuable — that is rate-monotonic scheduling, and it is provably optimal for fixed priorities.

Starvation is the failure mode: a high-priority task that never blocks prevents every lower one from running at all. Every task must block on something — a delay, a queue, a semaphore.

4Communication

Sharing a global between tasks is week 41's data race with no thread sanitizer available. Use the primitives instead.

Queues

QueueHandle_t q = xQueueCreate(10, sizeof(Reading));

xQueueSend(q, &reading, portMAX_DELAY);        /* blocks if full  */
xQueueReceive(q, &reading, portMAX_DELAY);     /* blocks if empty */

xQueueSendFromISR(q, &reading, &woken);        /* the ISR variant */

A queue copies the data, so ownership never crosses — the sender may reuse its buffer immediately. It also provides the blocking that lets the receiver sleep. This is week 41's producer-consumer, with the mutex and condition variable inside the primitive.

The FromISR suffix is not optional: the ordinary functions may block, and blocking inside an interrupt handler deadlocks the system.

Semaphores and mutexes

A mutex is not merely a binary semaphore with a different name. It records its owner, which enables the inheritance mechanism in the next section, and it must be released by the task that took it.

5Priority inversion

The classic real-time failure, and the one that famously reset the Mars Pathfinder rover repeatedly in 1997.

Low  priority: takes mutex M, starts working
Mid  priority: becomes ready, preempts Low  ← Low never finishes
High priority: becomes ready, needs M, blocks ← waits for Low
                                               which waits for Mid

The high-priority task is blocked behind a medium-priority task it does not interact with, for an unbounded time. The priorities have effectively inverted.

Priority inheritance fixes it: while a high-priority task waits on a mutex, the owner temporarily inherits that priority, so it cannot be preempted by the medium task and finishes promptly. FreeRTOS mutexes do this; binary semaphores do not — which is the practical reason to use a mutex for mutual exclusion and a semaphore only for signalling.

The remaining rules are week 41's, unchanged: consistent lock ordering to prevent deadlock, and critical sections as short as possible.

6Worked example: the same device, two ways

A device that samples a sensor every 10 ms, updates a control loop, writes a slow log, and responds to input. First as a superloop, then with a scheduler — the same work, measured.

Superloop

/* superloop.c */
#include <stdint.h>
#include "board.h"          /* uart_puts, put_int, tick_count from week 51 */

#define LOG_COST_TICKS 50   /* the slow operation */

static uint32_t sensor_runs, control_runs, log_runs, ui_runs;
static uint32_t sensor_worst_gap, last_sensor_tick;

static void busy_wait(uint32_t ticks)
{
    uint32_t start = tick_count;
    while (tick_count - start < ticks) { }
}

static void read_sensors(void)
{
    uint32_t now = tick_count;
    uint32_t gap = now - last_sensor_tick;
    if (gap > sensor_worst_gap) sensor_worst_gap = gap;
    last_sensor_tick = now;
    sensor_runs++;
    busy_wait(2);
}

static void update_control(void) { control_runs++; busy_wait(1); }
static void write_log(void)      { log_runs++;     busy_wait(LOG_COST_TICKS); }
static void handle_ui(void)      { ui_runs++;      busy_wait(1); }

int main(void)
{
    board_init();
    uart_puts("\n=== superloop ===\n");

    uint32_t started = tick_count;
    while (tick_count - started < 2000) {     /* 2 seconds */
        read_sensors();
        update_control();
        write_log();
        handle_ui();
    }

    uart_puts("in 2000 ticks:\n");
    uart_puts("  sensor runs      : "); put_int((int32_t)sensor_runs);
    uart_puts("\n  control runs     : "); put_int((int32_t)control_runs);
    uart_puts("\n  log runs         : "); put_int((int32_t)log_runs);
    uart_puts("\n  WORST sensor gap : "); put_int((int32_t)sensor_worst_gap);
    uart_puts(" ticks  (wanted 10)\n");
    uart_puts("\n  the log's 50 ticks are in front of every sensor read\n");

    for (;;) { }
}

in 2000 ticks:
  sensor runs      : 37
  control runs     : 37
  log runs         : 37
  WORST sensor gap : 54 ticks  (wanted 10)

Thirty-seven samples where two hundred were wanted, and a worst-case gap of 54 ticks — the sum of every other task. No amount of tuning fixes this; the structure causes it.

With FreeRTOS

/* rtos.c */
#include "FreeRTOS.h"
#include "task.h"
#include "queue.h"
#include "semphr.h"
#include "board.h"

#define PRIO_SENSOR  4      /* 10 ms deadline: highest */
#define PRIO_CONTROL 3
#define PRIO_UI      2
#define PRIO_LOG     1      /* no deadline: lowest     */

typedef struct { uint32_t tick; int32_t value; } Reading;

static QueueHandle_t     readings;      /* sensor  → control */
static QueueHandle_t     log_queue;     /* control → log     */
static SemaphoreHandle_t stats_mutex;   /* a MUTEX: inheritance */

static uint32_t sensor_runs, control_runs, log_runs;
static uint32_t sensor_worst_gap, last_sensor_tick;

static void busy_wait(uint32_t ticks)
{
    TickType_t start = xTaskGetTickCount();
    while (xTaskGetTickCount() - start < ticks) { }   /* real work */
}

/* --- highest priority: a hard 10 ms period --- */
static void sensor_task(void *arg)
{
    (void)arg;
    TickType_t last_wake = xTaskGetTickCount();

    for (;;) {
        uint32_t now = (uint32_t)xTaskGetTickCount();
        uint32_t gap = now - last_sensor_tick;

        if (xSemaphoreTake(stats_mutex, pdMS_TO_TICKS(5)) == pdTRUE) {
            if (last_sensor_tick != 0 && gap > sensor_worst_gap) {
                sensor_worst_gap = gap;
            }
            sensor_runs++;
            xSemaphoreGive(stats_mutex);
        }
        last_sensor_tick = now;

        Reading r = { .tick = now, .value = (int32_t)(now % 100) };
        xQueueSend(readings, &r, 0);          /* never block the sensor */

        busy_wait(2);
        vTaskDelayUntil(&last_wake, pdMS_TO_TICKS(10));   /* exact period */
    }
}

/* --- blocks on the queue: costs nothing while waiting --- */
static void control_task(void *arg)
{
    (void)arg;
    Reading r;
    for (;;) {
        if (xQueueReceive(readings, &r, portMAX_DELAY) == pdTRUE) {
            if (xSemaphoreTake(stats_mutex, portMAX_DELAY) == pdTRUE) {
                control_runs++;
                xSemaphoreGive(stats_mutex);
            }
            busy_wait(1);
            xQueueSend(log_queue, &r, 0);     /* drop if the log is behind */
        }
    }
}

/* --- lowest priority: preempted constantly, and that is correct --- */
static void log_task(void *arg)
{
    (void)arg;
    Reading r;
    for (;;) {
        if (xQueueReceive(log_queue, &r, portMAX_DELAY) == pdTRUE) {
            if (xSemaphoreTake(stats_mutex, portMAX_DELAY) == pdTRUE) {
                log_runs++;
                xSemaphoreGive(stats_mutex);
            }
            busy_wait(50);                    /* the slow operation */
        }
    }
}

static void report_task(void *arg)
{
    (void)arg;
    vTaskDelay(pdMS_TO_TICKS(2000));

    uart_puts("\n=== FreeRTOS, same work ===\n");
    uart_puts("in 2000 ticks:\n");
    uart_puts("  sensor runs      : "); put_int((int32_t)sensor_runs);
    uart_puts("\n  control runs     : "); put_int((int32_t)control_runs);
    uart_puts("\n  log runs         : "); put_int((int32_t)log_runs);
    uart_puts("\n  WORST sensor gap : "); put_int((int32_t)sensor_worst_gap);
    uart_puts(" ticks  (wanted 10)\n");

    uart_puts("\nstack high-water marks (words unused):\n");
    uart_puts("  sensor : ");
    put_int((int32_t)uxTaskGetStackHighWaterMark(NULL));
    uart_puts("\n");
    uart_puts("  free heap: ");
    put_int((int32_t)xPortGetFreeHeapSize());
    uart_puts(" bytes\n");

    vTaskSuspend(NULL);
}

int main(void)
{
    board_init();

    readings    = xQueueCreate(8, sizeof(Reading));
    log_queue   = xQueueCreate(8, sizeof(Reading));
    stats_mutex = xSemaphoreCreateMutex();      /* MUTEX, not a semaphore */

    xTaskCreate(sensor_task,  "sensor",  256, NULL, PRIO_SENSOR,  NULL);
    xTaskCreate(control_task, "control", 256, NULL, PRIO_CONTROL, NULL);
    xTaskCreate(log_task,     "log",     256, NULL, PRIO_LOG,     NULL);
    xTaskCreate(report_task,  "report",  256, NULL, PRIO_UI,      NULL);

    vTaskStartScheduler();                       /* never returns */
    for (;;) { }
}

=== FreeRTOS, same work ===
in 2000 ticks:
  sensor runs      : 199
  control runs     : 199
  log runs         : 38
  WORST sensor gap : 10 ticks  (wanted 10)

What the numbers say

The sensor now meets its deadline every time, and the log does the same amount of work — it simply does it in the gaps. That is the whole value proposition, and it costs about ten kilobytes.

Build and run under QEMU

git clone --depth 1 https://github.com/FreeRTOS/FreeRTOS-Kernel
arm-none-eabi-gcc -mcpu=cortex-m3 -mthumb -ffreestanding -O2 -g \
    -I. -IFreeRTOS-Kernel/include \
    -IFreeRTOS-Kernel/portable/GCC/ARM_CM3 \
    -T firmware.ld -o rtos.elf \
    startup.c rtos.c board.c \
    FreeRTOS-Kernel/tasks.c FreeRTOS-Kernel/queue.c \
    FreeRTOS-Kernel/list.c  FreeRTOS-Kernel/timers.c \
    FreeRTOS-Kernel/portable/GCC/ARM_CM3/port.c \
    FreeRTOS-Kernel/portable/MemMang/heap_4.c

qemu-system-arm -M lm3s6965evb -nographic -kernel rtos.elf

A FreeRTOSConfig.h is required; the kernel repository ships an example for this exact machine.

Cause priority inversion

Replace the mutex with a binary semaphore and add a medium-priority task that never blocks:

stats_mutex = xSemaphoreCreateBinary();          /* NO inheritance */
xSemaphoreGive(stats_mutex);

static void greedy_task(void *arg)
{
    (void)arg;
    for (;;) { busy_wait(30); vTaskDelay(1); }    /* mid priority */
}
xTaskCreate(greedy_task, "greedy", 256, NULL, PRIO_CONTROL, NULL);

The sensor task now blocks waiting for the low-priority log task to release the semaphore, while the medium-priority greedy task prevents the log task from running. The worst sensor gap jumps from 10 ticks to 40 or more.

Change one line back — xSemaphoreCreateMutex() — and the gap returns to 10. Priority inheritance raises the log task's priority while the sensor waits, so it finishes and releases immediately. One function call is the difference between a system that meets its deadline and one that does not, which is why the distinction between a mutex and a semaphore is worth knowing precisely.

Measure the stacks

uxTaskGetStackHighWaterMark(handle)   /* words still unused, ever */

Run the worst-case workload and check each task. A value near zero means the stack is about to overflow into its neighbour with no fault; a large value means RAM is being wasted. This is week 52's painting, provided by the kernel.

Enable configCHECK_FOR_STACK_OVERFLOW 2 in the configuration and provide vApplicationStackOverflowHook — the kernel then checks the pattern on every switch and calls you rather than corrupting memory silently.

7Choosing

Start with a superloop. Most embedded products ship with one, and it is the only design with no concurrency bugs available to it. Move to an RTOS when you can name a deadline the superloop demonstrably misses — as this week's measurement did — and not before.

8Common mistakes

10Where this leads

Week 55 is about finding out what went wrong on a device you cannot attach a debugger to: on-chip debugging, tracing on a constrained target, hardware abstraction layers that let firmware be tested on a host, and the simulator-in-the-loop setup this whole block has been using.