====== 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).
> **Globale Arrays und Structs tragen seit lyxc 1.0.17C** (#1256, #1299). Auf Modulebene deklarierte Aggregate verhalten sich jetzt wie lokale: ''%%var q: [8]int64;%%'' nimmt einen schreibenden Indexzugriff an, ein Initialisierer (''%%:= [1,2,3,4,5,6,7,8]%%'') bleibt erhalten, und ein Feldzugriff auf einen globalen Struct (''s.x := 3'') wirkt.
>
> Bis 1.0.16x stürzte der schreibende Zugriff ab, der Initialisierer wurde verworfen, und Struct-Felder blieben still auf ''0''.
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: [[lyx_-_programmiersprache:guides:rtos-embedded-concurrency|Nebenläufigkeit auf Embedded-Targets & RTOS]]
import std.thread;
import std.io;
===== 1. Threads erstellen & beenden =====
''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;
}
==== Thread-API ====
Vollständige API-Referenz:
^ Funktion ^ Beschreibung ^
| ''ThreadCreate(func: int64, arg: int64): Thread'' | Erstellt und startet neuen Thread; ''func'' ist Funktionszeiger (als int64) |
| ''ThreadJoin(t: Thread): void'' | Wartet auf Beendigung des Threads |
| ''ThreadSelf(): int64'' | Handle des aktuellen Threads (für Logging, TLS-Lookup) |
| ''ThreadIsRunning(t: Thread): int64'' | ungleich 0 wenn der Thread noch läuft |
> **''ThreadJoin'' liefert keinen Wert.** Die Rückgabe der Thread-Funktion lässt sich darüber nicht abholen — wer sie braucht, schreibt sie im Thread in eine gemeinsame Zelle (''Atomic'' oder ''alloc''+''poke64'') und liest sie nach dem Join.
> Eine Zuweisung wie ''var r: int64 %%:=%% ThreadJoin(t);'' übersetzt trotzdem und liefert still ''0'' — die Typprüfung greift bei einer ''void''-Rückgabe nicht.
>
> **''ThreadExit'' gibt es nicht.** Ein Thread endet mit ''return'' aus seiner Funktion.
==== Mehrere Threads starten ====
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;
}
===== 2. Race Conditions – das grundlegende Problem =====
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.
===== 3. Mutex – Gegenseitiger Ausschluss =====
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.
==== API ====
Vollständige API-Referenz:
^ Funktion ^ Beschreibung ^
| ''MutexNew(): Mutex'' | Erstellt und initialisiert neuen Mutex |
| ''MutexLock(m: Mutex): void'' | Sperrt den Mutex (blockiert, bis er frei ist) |
| ''MutexUnlock(m: Mutex): void'' | Gibt den Mutex frei |
==== Geschützter Zähler ====
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-Scope-Muster ====
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;
}
===== 4. Bedingungsvariablen (Cond) – Producer/Consumer =====
''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.
==== API ====
Vollständige API-Referenz:
^ Funktion ^ Beschreibung ^
| ''CondNew(): Cond'' | Erstellt neue Bedingungsvariable |
| ''CondWait(c: Cond, m: Mutex): void'' | Gibt Mutex atomar frei und wartet auf Signal; nach Rückkehr ist Mutex wieder gesperrt |
| ''CondSignal(c: Cond): void'' | Weckt **einen** wartenden Thread auf |
| ''CondBroadcast(c: Cond): void'' | Weckt **alle** wartenden Threads auf |
==== Producer/Consumer mit Ringpuffer ====
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;
}
> **CondWait-Invariante:**
> ''CondWait'' muss immer in einer ''while''-Schleife (nicht ''if'') aufgerufen werden. Spurious Wakeups — Aufwachen ohne echtes Signal — sind im POSIX-Standard erlaubt. Die Schleife prüft die Bedingung erneut und schläft weiter, falls sie noch nicht erfüllt ist.
===== 5. Atomare Operationen =====
''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.
==== API ====
Vollständige API-Referenz:
^ Funktion ^ Beschreibung ^
| ''AtomicNew(initial: int64): Atomic'' | Erstellt atomaren Wert mit Startwert |
| ''AtomicAdd(a: Atomic, delta: int64): int64'' | Addiert ''delta'' atomar; gibt **neuen** Wert zurück |
| ''CAS(a: Atomic, oldVal: int64, newVal: int64): int64'' | Compare-And-Swap: schreibt ''newVal'' nur wenn aktueller Wert ''oldVal'' ist; gibt 1 (Erfolg) oder 0 (Mismatch) zurück |
==== Lock-freier Zähler ====
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;
}
==== Compare-And-Swap (CAS) – Lock-freier Stack ====
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;
}
===== 6. Thread-Local Storage (TLS) =====
TLS ([[lyx_-_programmiersprache:guides:netzwerk|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.
==== API ====
Vollständige API-Referenz:
^ Funktion ^ Beschreibung ^
| ''TLSKeyCreate(): TLSKey'' | Erstellt neuen TLS-Schlüssel (global registriert) |
| ''TLSSetValue(key: TLSKey, value: int64): int64'' | Setzt Thread-lokalen Wert für diesen Schlüssel |
| ''TLSGetValue(key: TLSKey): int64'' | Liest Thread-lokalen Wert; gibt 0 zurück wenn nicht gesetzt |
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;
}
===== 7. Shared Memory (SharedMem) =====
''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.
==== API ====
Vollständige API-Referenz:
^ Funktion ^ Beschreibung ^
| ''SharedMemCreate(size: int64): SharedMem'' | Erstellt anonymes Shared-Memory-Segment (''mmap'') |
| ''SharedMemFree(mem: SharedMem): int64'' | Gibt Shared Memory frei (''munmap'') |
''SharedMem'' ist ein Struct mit zwei Feldern: ''data'' (Basisadresse als ''int64'') und ''size''. Gelesen und geschrieben wird über ''peek64'' ([[lyx_-_programmiersprache:sprache:rohspeicher|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;
}
===== 8. Vollständiges Beispiel: Thread-Pool =====
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;
}
===== 9. Synchronisations-Primitive im Vergleich =====
^ Primitiv ^ Einsatz ^ Overhead ^ Blockiert ^
| ''Mutex'' | Exklusiver Zugriff auf beliebige Daten | Mittel (Syscall bei Konflikt) | Ja |
| ''Cond'' | Warten auf Zustandsänderung (Producer/Consumer) | Niedrig (mit Mutex kombiniert) | Ja |
| ''Atomic'' | Einfache Zähler, Flags, Sequenznummern | Sehr niedrig (Hardwareinstruktion) | Nein |
| ''TLSKey'' | Per-Thread-Kontext (Error-State, Logger) | Sehr niedrig | Nein |
| ''SharedMem'' | Großer gemeinsamer Puffer, IPC | Niedrig (mmap) | Nein |
**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**
===== 10. Nebenläufigkeit in Safety-Code =====
Threads und @flight_crit schließen sich nicht vollständig aus, aber die Kombination erfordert sorgfältige Architektur.
^ Aspekt ^ @flight_crit ^ @dal(A) ^ @dal(B/C) ^
| ''ThreadCreate'' in Init-Phase | ✅ Erlaubt | ✅ Erlaubt | ✅ Erlaubt |
| ''ThreadCreate'' im Regelzyklus | ❌ Verboten | ❌ Verboten | ⚠️ Warnung |
| ''MutexLock'' / ''MutexUnlock'' | ✅ Erlaubt | ✅ Mit Nachweis | ✅ Erlaubt |
| ''CondWait'' ohne Timeout | ❌ Verboten | ❌ Verboten | ⚠️ Warnung |
| ''CondWait'' mit ''limit'' | ✅ Erlaubt | ✅ Mit Nachweis | ✅ Erlaubt |
| ''AtomicAdd'', ''CAS'' | ✅ Erlaubt | ✅ Erlaubt | ✅ Erlaubt |
| ''SharedMemCreate'' in Init | ✅ Erlaubt | ✅ Erlaubt | ✅ Erlaubt |
==== Empfohlenes Muster für @dal(B) ====
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;
}
===== 11. Häufige Fehler =====
^ Fehler ^ Ursache ^ Lösung ^
| Deadlock | Thread A hält Mutex 1, wartet auf Mutex 2; Thread B umgekehrt | Mutexe immer in derselben Reihenfolge sperren |
| Race Condition | Shared Variable ohne Mutex/Atomic | Atomic für primitive Werte; Mutex für Structs/Arrays |
| Vergessenes MutexUnlock | Früher ''return'' aus gesperrtem Block | Immer vor ''return'' entsperren |
| Spurious Wakeup ignoriert | ''if'' statt ''while'' bei CondWait | Immer ''while (Bedingung) limit(N)'' um CondWait |
| Thread-Leak | ThreadJoin vergessen | Jeden gestarteten Thread joinen oder mit Sentinel beenden |
| Stack-Overflow im Thread | Standardstack-Größe zu klein (POSIX: 8 MB) | @stack_limit für Thread-Funktionen (wird bei jeder Übersetzung geprüft) |
**Weiterführende Seiten:**
* [[lyx_-_programmiersprache:sprache:memory-management|Memory Management – Heap, Stack, SharedMem]]
* [[lyx_-_programmiersprache:units:thread|std.thread – vollständige API-Referenz]]
* [[lyx_-_programmiersprache:sprache:pointer-inlining|Low-Level: Speicherzugriff & Pragmas]]
* [[lyx_-_programmiersprache:sprache:abi-calling-conventions|ABI – Thread-Funktionen und Register-Konventionen]]
* [[lyx_-_programmiersprache:guides:do-178c|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).