Lyx – Threads & Nebenläufigkeit

Lyx unterstützt Nebenläufigkeit über POSIX-Threads (pthreads), gekapselt in der Unit std.thread. Sie bietet sechs Primitive: Thread (Ausführungsfaden), Mutex (gegenseitiger Ausschluss), Cond (Bedingungsvariablen), Atomic (lock-freie Ganzzahl), TLSKey (Thread-Local Storage) und SharedMem (gemeinsamer Speicher via mmap).

Threads teilen sich den Adressraum des Prozesses. Alle globalen Variablen und Heap-Objekte sind von jedem Thread aus erreichbar. Das macht schnelle Kommunikation möglich, erfordert aber sorgfältige Synchronisation: Gleichzeitige Schreibzugriffe auf dieselbe Variable ohne Schutz führen zu Race Conditions — undefiniertem, plattformabhängigem Verhalten.

Diese Seite behandelt POSIX-Threads auf Linux und macOS. Für Bare-Metal-Targets (ARM Cortex-M, RISC-V ohne OS) und RTOS-Umgebungen (ESP32/FreeRTOS) siehe: Nebenläufigkeit auf Embedded-Targets & RTOS

import std.thread;
import std.io;

ThreadCreate startet einen neuen Ausführungsfaden. Die Thread-Funktion erhält ein int64-Argument (Pointer auf Daten oder Wert) und wird parallel zum aufrufenden Thread ausgeführt.

unit hello_thread;
import std.thread;
import std.io;

fn Worker(arg: int64): int64 {
    PrintLn(StrConcat("Hallo vom Thread!", IntToStr(arg)));
    return 0;
}

fn main(): int64 {
    var t := ThreadCreate(Worker as int64, 42);   // startet Worker(42)
    ThreadJoin(t);                                 // wartet bis Worker endet
    PrintLn("Thread beendet.");
    return 0;
}

Vollständige API-Referenz:

unit multi_thread;
import std.thread;
import std.io;

fn ComputeChunk(id: int64): int64 {
    var sum: int64 := 0;
    var start := id * 1000;
    for i := start to start + 999 do {
        sum := sum + i;
    }
    PrintLn(StrConcat(StrConcat("Thread ", IntToStr(id)), " fertig"));
    return sum;
}

fn main(): int64 {
    var t0 := ThreadCreate(ComputeChunk as int64, 0);
    var t1 := ThreadCreate(ComputeChunk as int64, 1);
    var t2 := ThreadCreate(ComputeChunk as int64, 2);
    var t3 := ThreadCreate(ComputeChunk as int64, 3);

    ThreadJoin(t0);
    ThreadJoin(t1);
    ThreadJoin(t2);
    ThreadJoin(t3);

    PrintLn("Alle Threads abgeschlossen.");
    return 0;
}

Wenn zwei Threads dieselbe Variable lesen-modifizieren-schreiben, ohne sich abzustimmen, können Änderungen verloren gehen:

// ✗ FALSCH: Race Condition
var counter: int64 := 0;

fn IncrementBad(arg: int64): int64 {
    var i: int64 := 0;
    while (i < 100000) {
        counter++;   // Lesen + Addieren + Schreiben — nicht atomar!
        i++;
    }
    return 0;
}

// Ergebnis: nicht deterministisch (weniger als 200000)

Lösung: Mutex für exklusiven Zugriff oder Atomic für lock-freie Inkremente.

Ein Mutex (Mutual Exclusion) stellt sicher, dass immer nur ein Thread einen kritischen Abschnitt betreten kann. Der Lyx-Mutex ist rekursiv-sicher: Derselbe Thread kann MutexLock mehrfach aufrufen, ohne sich selbst zu blockieren.

Vollständige API-Referenz:

unit mutex_counter;
import std.thread;
import std.io;

var counter: int64 := 0;
var mtx: Mutex;

fn Increment(arg: int64): int64 {
    var i: int64 := 0;
    while (i < 100000) {
        MutexLock(mtx);
        counter++;           // kritischer Abschnitt: nur ein Thread gleichzeitig
        MutexUnlock(mtx);
        i++;
    }
    return 0;
}

fn main(): int64 {
    mtx := MutexNew();

    var t0 := ThreadCreate(Increment as int64, 0);
    var t1 := ThreadCreate(Increment as int64, 0);

    ThreadJoin(t0);
    ThreadJoin(t1);

    PrintLn(IntToStr(counter));  // immer 200000
    return 0;
}

Mutex immer in einem klar abgegrenzten Block sperren/freigeben. Niemals mit return oder break aus einem gesperrten Abschnitt springen, ohne vorher MutexUnlock aufzurufen:

fn SafeUpdate(value: int64): bool {
    MutexLock(mtx);

    if (value < 0) {
        MutexUnlock(mtx);   // ← Pflicht vor return!
        return false;
    }

    counter := value;
    MutexUnlock(mtx);
    return true;
}

Cond (Condition Variable) ermöglicht es einem Thread zu warten, bis ein anderer Thread ein bestimmtes Ereignis signalisiert. Das ist das Kernmuster für Producer/Consumer-Queues und Task-Dispatcher.

Vollständige API-Referenz:

unit producer_consumer;
import std.thread;
import std.io;

con QUEUE_SIZE: int64 := 64;

var queue:     [64]int64;
var q_head:    int64 := 0;
var q_tail:    int64 := 0;
var q_count:   int64 := 0;
var mtx:       Mutex;
var not_empty: Cond;
var not_full:  Cond;

fn Enqueue(value: int64) {
    MutexLock(mtx);
    while (q_count == QUEUE_SIZE) limit(100000) {
        CondWait(not_full, mtx);     // warte bis Platz frei
    }
    queue[q_tail] := value;
    q_tail := (q_tail + 1) % QUEUE_SIZE;
    q_count++;
    CondSignal(not_empty);           // Consumer benachrichtigen
    MutexUnlock(mtx);
}

fn Dequeue(): int64 {
    MutexLock(mtx);
    while (q_count == 0) limit(100000) {
        CondWait(not_empty, mtx);    // warte bis Item vorhanden
    }
    var value := queue[q_head];
    q_head := (q_head + 1) % QUEUE_SIZE;
    q_count--;
    CondSignal(not_full);            // Producer benachrichtigen
    MutexUnlock(mtx);
    return value;
}

fn Producer(arg: int64): int64 {
    for i := 0 to 199 do {
        Enqueue(i);
    }
    Enqueue(-1);   // Sentinel: Consumer soll enden
    return 0;
}

fn Consumer(arg: int64): int64 {
    while (true) {
        var v := Dequeue();
        if (v == -1) { break; }
        Print(IntToStr(v));
    }
    return 0;
}

fn main(): int64 {
    mtx       := MutexNew();
    not_empty := CondNew();
    not_full  := CondNew();

    var prod := ThreadCreate(Producer as int64, 0);
    var cons := ThreadCreate(Consumer as int64, 0);

    ThreadJoin(prod);
    ThreadJoin(cons);
    return 0;
}

Atomic ist ein int64-Wert, der ohne Mutex gelesen und verändert werden kann. Das Lesen-Modifizieren-Schreiben geschieht als eine unteilbare Hardwareoperation (LOCK XADD auf x86_64, LDADD auf ARM64).

Atomic-Operationen sind deutlich schneller als Mutex-basierter Code — aber nur für einfache Zähler und Flags geeignet.

Vollständige API-Referenz:

unit atomic_counter;
import std.thread;
import std.io;

var hits: Atomic;

fn Worker(arg: int64): int64 {
    var i: int64 := 0;
    while (i < 100000) {
        AtomicAdd(hits, 1);   // kein Mutex nötig
        i++;
    }
    return 0;
}

fn main(): int64 {
    hits := AtomicNew(0);

    var t0 := ThreadCreate(Worker as int64, 0);
    var t1 := ThreadCreate(Worker as int64, 0);
    var t2 := ThreadCreate(Worker as int64, 0);
    var t3 := ThreadCreate(Worker as int64, 0);

    ThreadJoin(t0);  ThreadJoin(t1);
    ThreadJoin(t2);  ThreadJoin(t3);

    PrintLn(IntToStr(AtomicLoad(hits)));  // immer 400000
    return 0;
}

CAS ist der Baustein für lock-freie Datenstrukturen. Hier ein lock-freies Push auf eine Ganzzahl:

import std.thread;   // Atomic, CAS

fn TrySetFlag(flag: Atomic, expected: int64, new_val: int64): bool {
    return CAS(flag, expected, new_val) == 1;
}

fn main(): int64 {
    var flag := AtomicNew(0);

    // Exakt ein Thread kann von 0 → 1 wechseln:
    if (TrySetFlag(flag, 0, 1)) {
        PrintLn("Dieser Thread hat das Flag gesetzt.");
    } else {
        PrintLn("Anderer Thread war schneller.");
    }
    return 0;
}

TLS (Netzwerk-Guide) ermöglicht pro-Thread-Variablen: Jeder Thread hat seinen eigenen Wert für denselben Schlüssel. Typische Einsatzfälle: Thread-ID-Tracking, Error-Context, per-Thread-Buffer.

Vollständige API-Referenz:

unit tls_example;
import std.thread;
import std.io;

var thread_id_key: TLSKey;

fn Worker(id: int64): int64 {
    TLSSetValue(thread_id_key, id);   // speichere ID lokal für diesen Thread

    // … andere Funktionen können die ID ohne Parameter-Übergabe lesen:
    var my_id := TLSGetValue(thread_id_key);
    PrintLn(StrConcat("Thread-ID: ", IntToStr(my_id)));
    return 0;
}

fn main(): int64 {
    thread_id_key := TLSKeyCreate();

    var t0 := ThreadCreate(Worker as int64, 100);
    var t1 := ThreadCreate(Worker as int64, 200);
    var t2 := ThreadCreate(Worker as int64, 300);

    ThreadJoin(t0);  ThreadJoin(t1);  ThreadJoin(t2);
    return 0;
}

SharedMem erstellt einen anonymen Speicherbereich via mmap, der von mehreren Threads (und optional Prozessen) geteilt werden kann. Anders als Heap-Speicher (malloc) ist mmap-Speicher direkt adressiert und kann für IPC (Inter-Process Communication) genutzt werden.

Vollständige API-Referenz:

SharedMem ist ein Struct mit zwei Feldern: data (Basisadresse als int64) und size. Gelesen und geschrieben wird über peek64 (Rohspeicher)/poke64 — eine Zeiger-Syntax gibt es in Lyx nicht.

unit shared_mem;
import std.thread;
import std.io;

fn Writer(arg: int64): int64 {
    poke64(peek64(arg), 12345);   // arg zeigt auf eine Zelle mit der Zieladresse
    return 0;
}

fn main(): int64 {
    var mem: SharedMem := SharedMemCreate(64);   // 64 Byte anonymes Shared Memory

    // Adresse in eine Zelle legen, damit der Thread sie als int64 erhält
    var cell: int64 := alloc(8);
    poke64(cell, mem.data);

    var t := ThreadCreate(Writer as int64, cell);
    ThreadJoin(t);

    PrintLn(IntToStr(peek64(mem.data)));   // 12345

    SharedMemFree(mem);
    return 0;
}

Ein Thread-Pool hält eine feste Anzahl Worker-Threads bereit und verteilt Aufgaben über eine Mutex-geschützte Queue. Neue Tasks werden eingereiht, Worker holen sie und verarbeiten sie.

unit thread_pool;
import std.thread;
import std.io;
import std.math;

con POOL_SIZE:  int64 := 4;
con QUEUE_CAP:  int64 := 256;

// ── Aufgaben-Queue ────────────────────────────────────────────────────────────
var tasks:      [256]int64;
var t_head:     int64 := 0;
var t_tail:     int64 := 0;
var t_count:    int64 := 0;
var done:       int64 := 0;   // 1 = keine weiteren Tasks

var q_mtx:      Mutex;
var q_ready:    Cond;

fn PushTask(value: int64) {
    MutexLock(q_mtx);
    tasks[t_tail] := value;
    t_tail := (t_tail + 1) % QUEUE_CAP;
    t_count++;
    CondSignal(q_ready);
    MutexUnlock(q_mtx);
}

fn PopTask(): (int64, bool) {
    MutexLock(q_mtx);
    while (t_count == 0 & done == 0) limit(1000000) {
        CondWait(q_ready, q_mtx);
    }
    if (t_count == 0) {
        MutexUnlock(q_mtx);
        return (0, false);   // keine Tasks mehr
    }
    var val := tasks[t_head];
    t_head := (t_head + 1) % QUEUE_CAP;
    t_count--;
    MutexUnlock(q_mtx);
    return (val, true);
}

// ── Worker ────────────────────────────────────────────────────────────────────
var result_mtx: Mutex;
var total:      int64 := 0;

fn Worker(id: int64): int64 {
    while (true) {
        var task, ok := PopTask();
        if (!ok) { break; }

        // Aufgabe: Quadratzahl berechnen
        var sq := task * task;

        MutexLock(result_mtx);
        total := total + sq;
        MutexUnlock(result_mtx);
    }
    return 0;
}

// ── Hauptprogramm ─────────────────────────────────────────────────────────────
fn main(): int64 {
    q_mtx      := MutexNew();
    result_mtx := MutexNew();
    q_ready    := CondNew();

    // Worker-Threads starten
    var workers: [4]Thread;
    for i := 0 to POOL_SIZE - 1 do {
        workers[i] := ThreadCreate(Worker as int64, i);
    }

    // Aufgaben einreihen: 1² + 2² + ... + 100²
    for i := 1 to 100 do {
        PushTask(i);
    }

    // Fertig signalisieren
    MutexLock(q_mtx);
    done := 1;
    CondBroadcast(q_ready);   // alle Worker aufwecken
    MutexUnlock(q_mtx);

    // Warten
    for i := 0 to POOL_SIZE - 1 do {
        ThreadJoin(workers[i]);
    }

    PrintLn(IntToStr(total));  // Summe der Quadrate 1..100 = 338350
    return 0;
}

Faustregel:

• Einfacher Zähler / Flag → Atomic
• Komplexe Datenstruktur, mehrere Felder → Mutex
• Warten auf Ereignis → Cond + Mutex
• Per-Thread-Zustand → TLS
• Großer Puffer ohne Kopie → SharedMem

Threads und @flight_crit schließen sich nicht vollständig aus, aber die Kombination erfordert sorgfältige Architektur.

Threads werden einmalig in der Init-Phase gestartet. Der Regelzyklus kommuniziert ausschließlich über Atomics oder Mutex-geschützte feste Puffer — keine dynamische Thread-Erzeugung, kein unbeschränktes Warten.

@dal(B)
unit flight_ipc;
import std.thread;
import std.time;    // Sleep

// Shared State zwischen Sensor-Thread und Regler-Thread
var sensor_val:   Atomic;
var sensor_valid: Atomic;
var ipc_mtx:      Mutex;

@flight_crit
@stack_limit(1024)
fn SensorThread(arg: int64): int64 {
    while (true) {
        var raw := ReadHardwareSensor();
        AtomicAdd(sensor_valid, 0);      // Memory-Barrier
        var v := AtomicNew(raw);
        sensor_val := v;
        AtomicAdd(sensor_valid, 1);      // Signalisiere neuen Wert
        Sleep(10);
    }
    return 0;
}

@flight_crit
@stack_limit(512)
fn ControlThread(arg: int64): int64 {
    var last_seq: int64 := 0;
    while (true) {
        var seq := AtomicLoad(sensor_valid);
        if (seq != last_seq) {
            var val := AtomicLoad(sensor_val);   // atomar gelesen
            last_seq := seq;
            ComputeControl(val);
        }
        Sleep(5);
    }
    return 0;
}

fn main(): int64 {
    sensor_val   := AtomicNew(0);
    sensor_valid := AtomicNew(0);
    ipc_mtx      := MutexNew();

    // Init-Phase: Threads starten
    var t_sensor  := ThreadCreate(SensorThread as int64, 0);
    var t_control := ThreadCreate(ControlThread as int64, 0);

    // Warten (Endlosschleife im @dal(C)-Hauptmodul)
    ThreadJoin(t_sensor);
    ThreadJoin(t_control);
    return 0;
}

Weiterführende Seiten:

• Memory Management – Heap, Stack, SharedMem
• std.thread – vollständige API-Referenz
• Low-Level: Speicherzugriff & Pragmas
• ABI – Thread-Funktionen und Register-Konventionen
• DO-178C – Nebenläufigkeit in zertifizierter Software

Codebeispiele geprüft: gegen lyxc 1.2.5C übersetzt (Prüflauf 2026-09-08 über die gesamte Doku: 574 Vollprogramme, 0 echte Fehler; zusätzlich 5159 Aufrufe gegen die pub fn-Signaturen in aurum/std gehalten, 0 Abweichungen).