Embedded Systems and Real-Time Operation

What Makes a System Embedded?

An embedded system is a computer built into a larger device to perform a dedicated function. Users rarely see it as a computer. It runs the engine, the brakes, the thermostat, or the infusion pump. In a cyber-physical system, the embedded computer is the "cyber" half of the loop introduced in Introduction to Cyber-Physical Systems.

What separates embedded design from general-purpose programming is a set of constraints that are all active at the same time:

Computing platforms

The Hardware/Software Interface

Software reaches the physical world through peripherals: hardware blocks that the processor controls by reading and writing special addresses called memory-mapped registers. Writing a bit to a register can start a motor; reading one can return a sensor value.

Getting data in and out: polling, interrupts, and DMA

Common buses and networks

What "Real-Time" Really Means

A real-time system is one whose correctness depends on when results are produced as well as what they are. Real-time does not mean fast. A system that responds in 10 ms every time is more real-time than one that responds in 1 ms on average and 200 ms occasionally. The goal is predictability.

Timing vocabulary

Why execution time varies

The same code on the same input can take different amounts of time from run to run. Each item below is a source of timing variability that a WCET bound has to cover:

• Caches: a cache hit takes a few cycles; a miss to DRAM can take hundreds.
• Pipelines and branch prediction: mispredicted branches flush work already in progress.
• Interrupts and preemption: other code runs in the middle of yours.
• Multicore interference: cores share caches, memory controllers, and buses, so one core's workload slows another's.
• Operating system and runtime: page faults, garbage collection, and lock contention.
• Input-dependent paths: loops and branches whose length depends on the data.

WCET can be estimated by measurement (run many times and add a safety margin, which is never a proof) or bounded by static analysis of the binary against a model of the processor (sound, but pessimistic and only possible for analyzable hardware). Safety-critical systems often disable caches or pin tasks to cores just to make timing analyzable.

The Periodic Task Model

Real-time scheduling theory abstracts software into tasks. Each periodic task τ_i releases a new job every T_i time units, and each job needs up to C_i units of processor time (its WCET) before a relative deadline D_i. In the simplest and most common case, D_i = T_i: each job must finish before the next one is released.

Real-Time Scheduling

Schedulability tests

A schedulability test decides, before deployment, whether a task set will always meet its deadlines. The table below lists the classic tests for one processor with independent tasks and D = T.

Response-time analysis says that task i's worst case happens when every higher-priority task j releases at the same instant (the critical instant). Then i must wait for its own C_i plus every higher-priority job released while it is still waiting.

Example: Schedulability analysis

The code below applies all three tests to two task sets. Each task is (name, C, T) with D = T, and RM priorities are assigned by period.

import math

def utilization(tasks):
    return sum(c / t for _, c, t in tasks)

def liu_layland_bound(n):
    return n * (2 ** (1 / n) - 1)

def response_times(tasks):
    ordered = sorted(tasks, key=lambda task: task[2])
    results = []
    for i, (name, c, t) in enumerate(ordered):
        higher = ordered[:i]
        r = c + sum(hc for _, hc, _ in higher)
        while True:
            nxt = c + sum(math.ceil(r / ht) * hc for _, hc, ht in higher)
            if nxt == r or nxt > t:
                break
            r = nxt
        results.append((name, nxt, t, nxt <= t))
    return results

def analyze(label, tasks):
    u = utilization(tasks)
    bound = liu_layland_bound(len(tasks))
    print(f'{label}: U = {u:.3f}, RM bound = {bound:.3f}, EDF bound = 1.000')
    print(f'  RM utilization test: {"pass" if u <= bound else "inconclusive"}')
    print(f'  EDF utilization test: {"pass" if u <= 1 else "fail"}')
    for name, r, t, ok in response_times(tasks):
        print(f'  RM response time {name:<10} R = {r:>2}  deadline = {t:>2}  {"ok" if ok else "MISS"}')

analyze('Set A', [('imu', 1, 4), ('control', 2, 6), ('logger', 3, 12)])
analyze('Set B', [('control', 2, 5), ('vision', 4, 7)])

Output:

Set A: U = 0.833, RM bound = 0.780, EDF bound = 1.000
  RM utilization test: inconclusive
  EDF utilization test: pass
  RM response time imu        R =  1  deadline =  4  ok
  RM response time control    R =  3  deadline =  6  ok
  RM response time logger     R = 10  deadline = 12  ok
Set B: U = 0.971, RM bound = 0.828, EDF bound = 1.000
  RM utilization test: inconclusive
  EDF utilization test: pass
  RM response time control    R =  2  deadline =  5  ok
  RM response time vision     R =  8  deadline =  7  MISS

Example: Watching RM and EDF schedule the same tasks

This discrete-time simulator runs Set B for one hyperperiod (35 time units). Each character is one time unit: c for control, v for vision, . for idle. A job still unfinished at its deadline is counted as a miss and dropped.

from math import lcm

def simulate(tasks, policy):
    horizon = lcm(*(t for _, _, t in tasks))
    jobs, timeline, misses = [], [], []
    for now in range(horizon):
        for name, c, t in tasks:
            if now % t == 0:
                jobs.append({'name': name, 'left': c, 'deadline': now + t, 'period': t})
        for job in [j for j in jobs if j['deadline'] <= now]:
            misses.append((job['name'], job['deadline']))
            jobs.remove(job)
        if not jobs:
            timeline.append('.')
            continue
        key = (lambda j: j['period']) if policy == 'RM' else (lambda j: j['deadline'])
        job = min(jobs, key=key)
        job['left'] -= 1
        timeline.append(job['name'][0])
        if job['left'] == 0:
            jobs.remove(job)
    return ''.join(timeline), misses

tasks = [('control', 2, 5), ('vision', 4, 7)]
for policy in ('RM', 'EDF'):
    timeline, misses = simulate(tasks, policy)
    print(f'{policy:<3} {timeline}')
    print(f'    misses: {misses if misses else "none"}')

Output:

RM  ccvvvccvvvccv.vccvvvccvvvccvvvccvv.
    misses: [('vision', 7)]
EDF ccvvvvccvvvvccvccvvvccvvvvccvvvvcc.
    misses: none

Shared Resources and Priority Inversion

Tasks share data, buses, and devices, usually protected by a mutex. Locking breaks the simple rule that the highest-priority ready task runs. A high-priority task can be forced to wait for a low-priority task that holds a lock it needs. A short wait of that kind is bounded blocking and is unavoidable.

Unbounded priority inversion happens when medium-priority tasks, which do not need the lock at all, preempt the low-priority lock holder. The high-priority task then waits for all of them as well.

Example: Priority inversion with and without inheritance

Three jobs: L (low) locks a shared resource for 4 units, M (medium) needs 6 units of CPU but no lock, and H (high) needs the lock for 2 units. Each segment is ('cpu', n) or ('cs', n) for a critical section. Each character in the timeline is the job running in that time unit.

def run(jobs, inheritance, horizon=20):
    state = {n: {'seg': 0, 'left': j['segs'][0][1], 'done': None} for n, j in jobs.items()}
    owner, timeline = None, []
    for now in range(horizon):
        ready = [n for n, j in jobs.items() if j['release'] <= now and state[n]['done'] is None]
        def wants_lock(n):
            return jobs[n]['segs'][state[n]['seg']][0] == 'cs'
        def blocked(n):
            return wants_lock(n) and owner not in (None, n)
        def priority(n):
            p = jobs[n]['prio']
            if inheritance and n == owner:
                waiting = [jobs[w]['prio'] for w in ready if blocked(w)]
                p = max([p] + waiting)
            return p
        runnable = [n for n in ready if not blocked(n)]
        if not runnable:
            timeline.append('.')
            continue
        n = max(runnable, key=priority)
        if wants_lock(n):
            owner = n
        s = state[n]
        s['left'] -= 1
        timeline.append(n)
        if s['left'] == 0:
            if wants_lock(n):
                owner = None
            s['seg'] += 1
            if s['seg'] == len(jobs[n]['segs']):
                s['done'] = now + 1
            else:
                s['left'] = jobs[n]['segs'][s['seg']][1]
    return ''.join(timeline), {n: s['done'] - jobs[n]['release'] for n, s in state.items()}

jobs = {
    'L': {'prio': 1, 'release': 0, 'segs': [('cpu', 1), ('cs', 4), ('cpu', 1)]},
    'M': {'prio': 2, 'release': 2, 'segs': [('cpu', 6)]},
    'H': {'prio': 3, 'release': 3, 'segs': [('cpu', 1), ('cs', 2), ('cpu', 1)]},
}
for inheritance in (False, True):
    timeline, response = run(jobs, inheritance)
    label = 'with inheritance   ' if inheritance else 'without inheritance'
    print(f'{label} {timeline}  H response time = {response["H"]}')

Output:

without inheritance LLMHMMMMMLLLHHHL....  H response time = 12
with inheritance    LLMHLLLHHHMMMMML....  H response time = 7

Software Platforms

Where Python fits

Standard CPython on a desktop operating system is not a hard real-time platform. The OS scheduler, garbage collection, and the interpreter itself add unbounded delays. Python is still valuable in CPS work: for analysis and simulation (like the examples on this page), for supervisory layers above a real-time core, and on microcontrollers through MicroPython, which offers direct register, timer, and interrupt access for prototyping.

One technique that matters in any language is how a periodic loop waits. Sleeping for a fixed period after doing the work lets every cycle's execution time and wake-up delay accumulate into drift. Sleeping until an absolute release time keeps the long-run rate exact, so only per-cycle jitter remains.

Example: Drift versus jitter in a periodic loop

import statistics
import time

def naive(period, cycles):
    start = time.perf_counter()
    stamps = []
    for _ in range(cycles):
        stamps.append(time.perf_counter())
        sum(i * i for i in range(2000))
        time.sleep(period)
    return start, stamps

def absolute(period, cycles):
    start = time.perf_counter()
    stamps = []
    for k in range(cycles):
        stamps.append(time.perf_counter())
        sum(i * i for i in range(2000))
        delay = start + (k + 1) * period - time.perf_counter()
        if delay > 0:
            time.sleep(delay)
    return start, stamps

def report(label, period, start, stamps):
    ideal = [start + k * period for k in range(len(stamps))]
    lateness = [(s - i) * 1e3 for s, i in zip(stamps, ideal)]
    gaps = [(b - a) * 1e3 for a, b in zip(stamps, stamps[1:])]
    print(f'{label:<9} mean period {statistics.mean(gaps):6.3f} ms  '
          f'jitter (stdev) {statistics.stdev(gaps):5.3f} ms  '
          f'drift after {len(stamps)} cycles {lateness[-1]:7.2f} ms')

period, cycles = 0.005, 400
report('naive', period, *naive(period, cycles))
report('absolute', period, *absolute(period, cycles))

Sample output (your numbers will differ by machine and load):

naive     mean period  5.247 ms  jitter (stdev) 0.111 ms  drift after 400 cycles   98.39 ms
absolute  mean period  5.001 ms  jitter (stdev) 0.210 ms  drift after 400 cycles    0.23 ms

Beyond a Single Processor

A Practical Real-Time Design Checklist

1. Derive each deadline and period from the physics (plant dynamics, safety margins), not from convenience.
2. Bound the WCET of every task, and state how it was obtained (measurement with margin, or static analysis).
3. Choose a scheduling policy and run an exact schedulability test, including blocking time from shared resources and interrupt overhead.
4. Use mutexes with priority inheritance or ceiling protocols. Never use plain binary semaphores for mutual exclusion between priorities.
5. Keep ISRs short and defer work to tasks; protect every variable shared with an ISR.
6. Schedule periodic loops against absolute time, and measure the jitter on target hardware under worst-case load.
7. Add runtime monitoring: watchdogs, deadline-miss counters, and a defined safe behavior when a deadline is missed (see Safety in Cyber-Physical Systems).

The Series

Textbook Reading

Tutorials, Tools, and Interactive Learning

Further Reading