🪄 Die Fiber-Illusion: Was uns RxJS über echte asynchrone Pipelines in PHP gelehrt hat

vom 18. Juli 2026

Die meisten Entwickler lernen RxJS durch die flüssige Syntax kennen:

source.pipe(map(...), mergeMap(...), retry(...));

Unvergesslich – und der uninteressanteste Teil beim Entwurf einer PHP-Pipeline-Engine.

Wir nutzten RxJS als architektonisches Spiegelbild für Darkwood Flow. Teams, die Scraper, Importer und Medienpipelines entwickeln, fragen sich immer wieder, ob sie „reaktive“ APIs benötigen. Unsere Frage war präziser:

Was kann Flow von RxJS lernen, ohne selbst zu RxJS zu werden?

RxJS beschreibt Werte im Zeitverlauf anhand eines expliziten Lebenszyklus: Zusammensetzen mit pipe, Starten bei subscribe, Verfolgen aktiver Vorgänge, Warten auf deren Abschluss, Weiterleiten von Fehlern, Beenden bei unsubscribe. Scheduler sind für Timer relevant; sie ermöglichen jedoch nicht das gleichzeitige Abrufen von URLs. Der entscheidende Aspekt ist der Ausführungslebenszyklus.

Unter der Bedienoberfläche sieht die Maschine folgendermaßen aus:

subscription starts execution
→ values propagate through a chain
→ asynchronous inner work is tracked
→ completion waits for active work
→ errors terminate or restart the run
→ unsubscription triggers teardown

Flow steht bereits vor denselben Fragen, wenn es über die einfache Bühnenkomposition hinausgeht:

• Wann beginnt eine Pipeline?
• Wann ist es fertig?
• Was ist noch aktiv? Was passiert im Fehlerfall? Wer sperrt Ressourcen?
• Wie können wir einen Lauf stoppen?

In Flow wird jeder Wert in einem Befehlspaket namens Ip übertragen. Pakete werden von Strategien empfangen, von einem Treiber eingeplant und nach await() gesammelt. Wir unterstützten bereits Fibers, Amp, React, Swoole und ähnliche Technologien – und haben bereits viele Pakete durch den Prozess fetch → hash → save geleitet. Was uns fehlte, war die Transparenz bezüglich nativer, nicht-blockierender E/A: überlappende Wartezeiten an echten Sockets, ohne Amp oder React als feste Abhängigkeit zu benötigen und ohne so zu tun, als ob ein Fiber um einen blockierenden Lesevorgang herum ausreichen würde.

JavaScript stattet RxJS mit einer Ereignisschleife aus. Die PHP-Befehlszeile (CLI) hingegen nicht. Ein Fiber kann die kooperative PHP-Ausführung unterbrechen; er kann jedoch einen blockierenden Socket-Aufruf nicht selbstständig in überlappende E/A umwandeln. Flow passt sich mit expliziten Primitiven an:

PHP streams → non-blocking mode → stream_select() → Fibers → Driver → Flow

Fazit: Übernehmen Sie die Lifecycle-Struktur von RxJS. Behalten Sie die Terminologie von Flow bei (Jobs, Pakete, Treiber). Diese Studie war nie ein Exkurs in RxPHP – der interessante Kern sind die Prinzipien, nicht die Typen.

Um unsere Integrität zu wahren, haben wir eine realitätsnahe Pipeline betrieben – nicht nur künstliche Zeitschaltuhren:

URLs → fetch concurrently → checksum → save → per-URL result → summary

Reale Sockets legen blockierendes Verhalten offen. Parallelitätsbeschränkungen werden entweder eingehalten oder nicht. Timeouts und Aufräumarbeiten werden sichtbar. Eine fehlgeschlagene URL darf nicht fünf erfolgreiche Verbindungen zunichtemachen, wenn dies die Produktregel ist. Künstliche delay()-Demos beweisen zwar die Fiber-Planung, aber nicht, dass offene Sockets eine stream_select-Wartezeit teilen.

These: Flow sollte sich die Disziplin von RxJS in Bezug auf Ausführung, Parallelität, Abschluss und Abbau zunutze machen – und gleichzeitig eine PHP-native Engine bleiben, die auf Jobs, Paketen, Strategien und Treibern basiert.

Anders formuliert: Wir dachten, ein Fiber-basierter Treiber würde Flow bereits asynchrone Ausführung ermöglichen. Die Beschäftigung mit RxJS zwang uns jedoch, den Begriff „asynchron“ neu zu definieren. Der Rest dieses Artikels beschreibt diese Erkenntnis.

Unter den Operatoren: Wie RxJS tatsächlich funktioniert

Wir haben RxJS 8.0.0-alpha.14 (Kerntypen in @rxjs/observable) untersucht. Die Paketstruktur ist versionsspezifisch; die Lebenszyklusprinzipien sind jedoch umfassender als in dieser Alpha-Version.

from(urls).pipe(
  mergeMap(fetchImage, 4),
  map(hashBinary),
  retry(2),
  finalize(cleanup),
);

Komposition ist nicht Ausführung

Observable.pipe wandelt Operatoren in neue Observables um. Es werden noch keine Daten abgerufen. Operatoren in diesem Baum geben typischerweise new Observable(subscriber => { ... }) zurück; der ältere lift-Pfad ist ab Version 8 veraltet. Flow trennt bereits Komposition (FlowFactory / fn) von Ausführung (await()) – gleiche Aufteilung, andere Namen.

Das Abonnement startet den Lauf

subscribe kapselt den Konsumenten in einen Subscriber, verbindet den Produzenten und registriert den Abbau. Operatoren abonnieren Upstream; next / error / complete schieben Daten in Richtung Senke. Hilfsfunktionen wie operate({ destination, ... }) verknüpfen Kindelemente mit dem Abbaubaum des Ziels, sodass das Abmelden kaskadierend wirkt.

Ein Subscription speichert Finalisierer und führt sie idempotent bei unsubscribe aus. Terminalsignale stoppen die Benachrichtigungen und kündigen das Abonnement. Die Bereinigung ist Teil des Beendens eines Laufs. Kalte Quellen (einschließlich from(array)) isolieren Läufe: Jedes Subscription erhält seinen eigenen Producer.

mergeMap verfolgt aktive innere

mergeMap(project, concurrent) delegiert an eine gemeinsame Merge-Logik: einen Zähler für aktive innere Prozesse, einen Puffer für Überlauf und die Fertigstellung erst, wenn der äußere Prozess abgeschlossen ist, der Puffer leer ist und keine inneren Prozesse mehr vorhanden sind. Innere Eingaben werden über from(project(...)) verarbeitet.

Wir brauchten mergeMap in PHP nicht. Wir brauchten seine Garantie: begrenzte laufende Arbeit und Abschluss, der auf jede aktive Operation wartet.

erneut versuchen und abschließen

Im Fehlerfall meldet retry das aktuelle innere Abonnement ab und meldet sich für dieselbe Quelle erneut an. Bei einem kalten from(urls) wird die Liste dadurch neu gestartet – eine leistungsstarke, aber oft ungeeignete Methode für Batch-I/O. finalize(callback) protokolliert die Bereinigung des äußeren Abonnements (abgeschlossen, Fehler oder Abmeldung), nicht zwischen den Wiederholungsversuchen.

Was man stehlen darf und was nicht

Dies ist der Kern der Studie.

Analogien sind hilfreich – und dabei bleibt es auch:

• Subscription ≈ ein zukünftiger Flow-Ausführungshandle
• Planer ≈ Fahrer
• mergeMap(n) ≈ beschränkte parallele Phase (MaxIpStrategy + resultierende Jobs)

Wichtigste Erkenntnis: Kopieren Sie den Lebenszyklusvertrag. Kopieren Sie nicht das Typsystem.

Was Flow bereits hatte

Vor der Studie war Flow kein unbeschriebenes Blatt. Diese Ausgangslage machte die Fiber-Illusion so überzeugend.

Wir könnten es bereits:

• Stufen mit FlowFactory (Generator-Do-Notation) oder fn() zusammensetzen
• Werte in Ip-Paketen verschieben
• Jobs ausführen (JobInterface oder Closures)
• Zulassung mit MaxIpStrategy($n) begrenzen
• Zeitplan über Treiber (standardmäßig FiberDriver, außerdem Amp, React, Swoole, …)
• Barriere bei await() (void—Ergebnisse benötigen einen Collector-Job)
• Fehler werden über den optionalen errorJob weitergeleitet

build pipeline → push Ip(s) → await()
→ Driver admits packets, runs jobs, forwards results
→ when nothing remains queued or active, await returns

Durch das Hinzufügen eines Pakets wird dieses in die erste Stufe eingereiht. Innerhalb von await() fragt der Treiber die Strategie jeder Stufe ab, welche Pakete ausgeführt werden dürfen, startet Job-Fibers (oder Koroutinen) und leitet erfolgreiche Pakete weiter oder ruft errorJob auf. Abgeschlossene Pakete geben Parallelitätsplätze frei.

Mentales Modell: Pakete durchlaufen verschiedene Phasen, Strategien begrenzen die Anzahl der ausgeführten Pakete, der Treiber ist für das Warten verantwortlich.

Mit FiberDriver, mehreren gleichzeitig übertragenen Paketen, delay() und MaxIpStrategy(4) sah es nach asynchroner Ausführung aus. Amp-Timer-Demos überschnitten sich tatsächlich. Der blinde Fleck war die native blockierende Stream-E/A – und wir hatten uns in Bezug auf die Bedeutung von „asynchron“ in diesem Zusammenhang geirrt.

Die Faserillusion

Was wir glaubten

Vier URLs pushen → vier zulassen → vier Fibers starten → await(). Wenn Jobs E/A-Operationen durchgeführt haben, müssen sich die Wartezeiten überschneiden.

RxJS machte diese Annahme unbequem: Wenn mergeMap aktive innere Knoten verfolgt, was verfolgte dann Flow – Pakete, Fibers oder Sockets?

Was die Geschichte ans Licht brachte

file_get_contents($url); // still blocks the process

Innerhalb eines FiberDriver-Jobs wird dieser Aufruf nie bei Bereitschaft angehalten. Der Treiber wartet auf den Block. MaxIpStrategy(4) kann vier Pakete zulassen, während der Prozess noch am Socket serialisiert – oder den Scheduler einfrieren, während nach parallelen Prozessen gesucht wird.

Dieselbe Falle: fread wird blockiert. Fibers helfen nur, wenn der Job pusht. Ohne eine Readiness-API gibt es keinen Grund für einen Push.

Ein Fiber kann die PHP-Ausführung unterbrechen. Er kann jedoch keinen blockierenden Socket-Aufruf in einen nicht-blockierenden umwandeln.

RxJS erzwang diese Terminologie: Zugriffsbeschränkungen sind keine I/O-Überlappung. Flow musste diese Ehrlichkeit übernehmen.

Der benötigte Stapel

non-blocking streams
→ stream_select()
→ Fiber suspension on wait
→ readiness-based resume
→ Driver await loop
→ Flow stages

Dieser Stack wurde zum Mittelpunkt der Studie – und deshalb wurde StreamSelectDriver in darkwood/flow aufgenommen.

Fazit: „Async Driver“ ohne Yield-on-Readiness-Strategie ist nur ein Slogan. Mit stream_select wird es zur festen Regel.

Konkrete Ergebnisse: die Bildverarbeitung

Die Demo befindet sich in content/rxjs-flow und nutzt lokalen Flow über einen Composer-Pfad-Symlink. Eine kurzlebige Demo-Engine erprobte stream_select und wurde anschließend absichtlich entfernt. Nur eine wiederverwendbare Laufzeitumgebung.

php bin/console app:fetch-images --concurrency=4 --timeout=5

Die Parallelität wird in der Fetch-Phase über MaxIpStrategy($concurrency) gesteuert. HTTP 404 ist ein weicher Fehler pro Element bei DTOs – nicht bei errorJob –, sodass fünf erfolgreiche Abrufe neben einem Fehler auftreten können. Traces geben flow.started, item.started, completed, failed und flow.completed aus.

Repräsentativer Lauf (concurrency=2):

[flow.started] items=6 concurrency=2
[item.started] url=http://127.0.0.1:8765/image-a.png
[item.started] url=http://127.0.0.1:8765/image-b.png
[item.started] url=http://127.0.0.1:8765/image-c.png
[item.completed] url=…/image-a.png checksum=… path=…
…
[item.failed] url=…/fail.png error=HTTP 404
[flow.completed] ok=5 failed=1 elapsed_ms=711.23

Die Testumgebungen verwenden php -S (Single-Threaded) mit kurzen Routenverzögerungen, sodass die Poolübergabe sichtbar ist. Clientseitige Wartezeitbegrenzungen sind real; eine tatsächliche Beschleunigung erfordert einen Server mit mehreren Verbindungen. Der Fetch-Job vermittelt HTTP – Weiterleitungen, TLS-Sonderfälle, Chunked-Besonderheiten, Proxys, Komprimierung und partielle E/A werden nicht behandelt. Produktionsclients gehören weiterhin in Jobs; Treiber bleiben generisch.

Nächster Abnahmetest (erwartet, hier nicht als gemessen beansprucht):

5 × ~1s delayed endpoints on a multi-connection backend
concurrency=1 → ~5s
concurrency=4 → ~2s

StreamSelectDriver: Was wir ausgeliefert haben

StreamSelectDriver multiplexiert die native Stream-Bereitschaft auf Fibers:

$ready = $driver->waitReadable($stream, $timeoutSeconds);
$ready = $driver->waitWritable($stream, $timeoutSeconds);

Beide geben einen booleschen Wert zurück und müssen innerhalb eines Fiber-Jobs ausgeführt werden. Intern gilt für den Treiber Folgendes:

1. Aufzeichnungen Glasfaser, Stream, Modus, optionale Frist
2. Setzt aus
3. Erstellt stream_select-Sets aus ausstehenden Wartezeiten.
4. Setzt bereite Fibers mit true fort, zeitüberschreitende Wartezeiten mit false.
5. Setzt weiterhin einfache delay()-Anhalten beim normalen Schleifentakt fort.

Die gleiche DriverInterface wie bei anderen Treibern (async, defer, await, delay, tick), plus diese Warte-APIs, die Jobs je nach Typ aktivieren.

MaxIpStrategy FIFO-Fix: Bei maximaler Kapazität keine erneute Eingabe in die Warteschlange (die rotierte Reihenfolge). Gleichzeitige Traces machten den Fehler deutlich.

Warum nicht FiberDriver erweitern? Die standardmäßige Wiederaufnahme von Fiber-Prozessen ist nicht mit einer bereitschaftsgesteuerten Wiederaufnahme identisch. Ein dedizierter Treiber hält die Planungsregel explizit.

Driver     → schedule, wait, resume, deadlines
Job        → protocol + application work
IpStrategy → packet admission / concurrency
Collector  → terminal aggregate

$driver = new StreamSelectDriver();
$flow = (new FlowFactory($driver))->create(static function () use (...) {
    yield [$fetchJob, null, new MaxIpStrategy($concurrency)];
    yield $hashJob;
    yield $saveJob;
    yield $collectorFn;
});

Fazit: Jobs werden erledigt. Treiber warten. HTTP existiert nie im Treiber – oder jedes Protokoll erfindet ein Framework.

Was „concurrency=4“ tatsächlich einschränkt

Eine Begrenzung der Parallelverarbeitung ist nur dann sinnvoll, wenn die Laufzeitumgebung genau angeben kann, was sie begrenzt.

MaxIpStrategy(4) begrenzt die Paketverarbeitung in einem Schritt. Allein dadurch entstehen keine sich überschneidenden Netzwerkwartezeiten.

Bei blockierenden E/A-Vorgängen ist die Beschränkung rein kosmetischer Natur. Mit StreamSelectDriver und dem Erzeugen von Fetch-Jobs:

MaxIpStrategy(4)
+ non-blocking fetch
+ StreamSelectDriver
→ ≤ 4 fetch packets admitted
→ ≤ 4 fetch Fibers active
→ ≤ 4 sockets waiting on select

Das ist Flows Übersetzung von mergeMap(..., 4). Der Gültigkeitsbereich ist eine Stufe: Hashing/Speichern kann sich mit dem nächsten Abruf überschneiden – das ist eine Pipeline-Überlappung, keine fehlende Begrenzung.

Fehler, Abschluss, Bereinigung

Sammeln vs. schnelles Fail-Response. Batch-Importe benötigen weiche Fehler pro Element; kritische Veröffentlichungen erfordern möglicherweise errorJob oder einen Prozessabbruch. Flow unterstützt beides. Timeouts für die Warteschicht sind bereits vorhanden; vollständige Wiederholungs-/Timeout-Operatoren können warten, bis sich Muster in verschiedenen Anwendungen wiederholen.

Abschluss bedeutet: Quelle erschöpft, Warteschlangen leer, keine aktiven Jobs, keine ausstehenden Stream-Wartezeiten, Ressourcen geschlossen. Aktuell wird diese Barriere durch await() plus einen Collector realisiert. Ein zukünftiges $flow->start() / Execution (Abbruch, Ergebnis, Fehler) ist eine optionale Benutzerfreundlichkeit – für diesen Abschnitt nicht erforderlich.

Aufräumarbeiten bedeuten „endlich“ + Wartezeiten – RxJS-Teardown im PHP-Gewand, keine Hoffnung auf den Destruktor. Strukturierte Traces sind einem internen Event-Bus in puncto Beobachtbarkeit vom ersten Tag an überlegen.

Mapping RxJS → Flow

Beibehalten: Lazy Compose / Explicit Run, Multi-Item-Pipelines, begrenzte Parallelität, deterministische Vervollständigung, Bereinigung, Isolation pro Ausführung, Beobachtbarkeit.
Vereinfachen: Jobs + Pakete; Treiber; Sammler – kein öffentliches next/error/complete-Protokoll; Limits – keine Operatornamen.
Verschieben: öffentliche Ausführung, Abbruchtoken, erstklassige Wiederholungs-/Timeout-Operationen, switchMap / exhaustMap, Hot Streams.
Abgelehnt: Öffentliche APIs in RxJS-Form, Subjects, Promise/Operator-Zoos, HTTP-in-Driver, Dual-Engines.

Flow bleibt eine ausgereifte, pragmatische Pipeline-Engine mit einem echten asynchronen Executor – kein reaktives Framework im Gewand der PHP-Syntax.

Wohin Flow als Nächstes geht

Keep:   lifecycle discipline (start → bound concurrency → wait → tear down)
Shape:  Job + Ip + MaxIpStrategy + Driver + await + collector
Next:   multi-connection overlap benchmarks
        cancel tokens on StreamSelectDriver waits
        richer await ergonomics if collectors become painful

Darkwood Flow wird nicht zu RxJS für PHP. Es setzt vielmehr auf eine PHP-native Pipeline-Engine: begrenzte Parallelität, bereitkeitsbasierte E/A, deterministische Vervollständigung und explizite Bereinigung – mit Treibern, die ihre Wartezeiten nachweisen können.

Wenn ein Job auf E/A wartet, wartet die Laufzeitumgebung dann tatsächlich mit – oder tut sie nur so?

Ressourcen

• Demo-Repository: github.com/matyo91/rxjs-flow
• Präsentationsfolien: github.com/matyo91/slidewire (/slides/the-fiber-illusion)