diff --git a/Doc/Pipe Optimization.md b/Doc/Pipe Optimization.md new file mode 100644 index 0000000..c100383 --- /dev/null +++ b/Doc/Pipe Optimization.md @@ -0,0 +1,66 @@ + +--- + +## Stream & Pipe Optimization + +### 1. Global Series Deduplication (The "Registry" Pattern) + +Die aktuelle Architektur erzeugt für jeden Pipe-Eingang einen eigenen Akkumulator. Bei komplexen Graphen mit identischen Datenquellen führt dies zu redundantem Speicherverbrauch. + +* **Konzept:** Einführung einer **Global Series Registry**, die als Mediator zwischen `IStream` und `TPipeSource` fungiert. +* **Mechanismus:** Anstatt eine `TScalarSeries` privat zu instanziieren, fordert die `TPipeSource` eine Serie über einen Composite-Key `(IStream, IKeyword)` an. +* **Vorteil:** Wenn zehn Pipes den `Close`-Preis desselben Tickers beobachten, existiert im gesamten Arbeitsspeicher nur eine einzige Instanz der Daten-Historie. +* **Lookback-Management:** Die Registry überwacht die Anforderungen aller Konsumenten und stellt sicher, dass die geteilte Serie immer den **maximalen Lookback** aller beteiligten Pipes vorhält. + +--- + +### 2. Local State Optimization (Ring-Buffer Series) + +Für Berechnungen innerhalb der Lambda-Funktion (z.B. kurzfristige Delta-Vergleiche) sind die auf Chunks basierenden `TScalarSeries` aufgrund ihrer Verwaltungs-Overheads ineffizient. + +* **Konzept:** Implementierung spezialisierter **Local Ring-Buffer**, die für ultrakurze Lookbacks (z.B. ) optimiert sind. +* **Mechanismus:** Ein statisch allozierter Speicherblock mit `truncate-on-add`-Logik. Im Gegensatz zu Chunks gibt es hier keine Heap-Fragmentation durch dynamisches Wachsen. +* **Scope:** Diese Strukturen sind "Lambda-Local". Sie werden nicht über die Registry geteilt, sondern dienen als privater, blitzschneller "Scratchpad-Speicher" für zustandsbehaftete Berechnungen. + +--- + +### 3. Signal Transmission & Pressure Relief + +Das ständige Allokieren von `TArray` in der `Emit`-Methode erzeugt bei hohen Tick-Raten signifikanten Druck auf den Memory-Manager und das Reference-Counting. + +* **Konzept:** **Allocation Pooling** oder **Buffer Slicing** für Signal-Payloads. +* **Mechanismus:** Einführung eines `TValueBufferPool`. Streams leihen sich ein Array für den `Emit`-Vorgang aus. Sobald der letzte Observer das Signal verarbeitet hat, kehrt das Array in den Pool zurück. +* **Alternative:** Nutzung von `Record`-basierten Memory-Slices, um Kopier-Operationen beim Erzeugen des Signals gänzlich zu eliminieren (**Zero-Copy**). + +--- + +### 4. Runtime Adaptability (Dynamic Lookback) + +Aktuell ist der Lookback einer Serie bei der Initialisierung festgeschrieben. In einem dynamischen System müssen Pipes jedoch zur Laufzeit hinzugeschaltet werden können. + +* **Konzept:** **Elastic Series Buffers**. +* **Mechanismus:** Die `TScalarSeries` erhält die Fähigkeit, ihre Chunk-Kapazität dynamisch nach oben zu korrigieren, ohne die Integrität der bestehenden Indizes zu verletzen. +* **Motivation:** Dies ermöglicht "Hot-Swapping" von Logik-Komponenten. Eine neue Analyse-Pipe kann sich in einen laufenden Stream einklinken und eine Erweiterung der Historie anfordern, ohne dass das System neu gestartet werden muss. + +--- + +### 5. Computational Economy (Lazy/Dirty Logic) + +In tiefen Graphen werden oft Pipes getriggert, deren Eingangsdaten sich zwar im Cycle geändert haben, deren für die Berechnung relevante Werte aber identisch geblieben sind. + +* **Konzept:** **Change-Detection & Lazy Evaluation**. +* **Mechanismus:** Jedes Signal erhält eine Versionierung. Pipes führen ihre Lambda nur aus, wenn sich die Versionen der Eingangs-Serien tatsächlich geändert haben oder wenn ein "Dirty"-Flag gesetzt ist. +* **Ziel:** Massive Einsparung von CPU-Zyklen in Szenarien, in denen viele Pipes auf hochfrequenten Streams hängen, aber nur selten Schwellenwerte überschritten werden. + +--- + +### Zusammenfassung der Architektur-Ziele + +| Fokus | Methode | Resultat | +| --- | --- | --- | +| **Memory** | Global Registry | Eliminierung redundanter Daten-Historien. | +| **Latency** | Ring-Buffers | Beschleunigung lokaler Lambda-Berechnungen. | +| **Throughput** | Buffer Pooling | Minimierung von GC-Pausen und Allokations-Overhead. | +| **Agility** | Dynamic Lookback | Unterstützung von On-the-fly Rekonfiguration. | + +--- \ No newline at end of file diff --git a/Src/AST/Myc.Ast.Evaluator.pas b/Src/AST/Myc.Ast.Evaluator.pas index aead007..1ff9b71 100644 --- a/Src/AST/Myc.Ast.Evaluator.pas +++ b/Src/AST/Myc.Ast.Evaluator.pas @@ -636,7 +636,7 @@ begin end; //TODO Lookback not evaluated in script! - Result := TPipeStream.Create(config, outputDef, sources, pipeAdapter, 1000); + Result := TPipeStream.Create(config, outputDef, sources, pipeAdapter, 1000, 1); end; end. diff --git a/Src/Data/Myc.Data.Stream.pas b/Src/Data/Myc.Data.Stream.pas index e72438a..9f93249 100644 --- a/Src/Data/Myc.Data.Stream.pas +++ b/Src/Data/Myc.Data.Stream.pas @@ -17,16 +17,17 @@ type // TStreamSignal represents the data packet traveling through the graph. // It carries the CycleID for synchronization and the actual data payload. TStreamSignal = record - strict private + private FCycleID: Int64; - FKind: TSignalKind; - FData: TArray; + FData: TScalar.PValue; + function GetFields(Idx: Integer): TScalar.TValue; inline; + function GetKind: TSignalKind; inline; public - constructor Create(AKind: TSignalKind; ACycleID: Int64; const AData: TArray = nil); + constructor Create(ACycleID: Int64; const AData: TScalar.PValue); function ToString: string; + property Fields[Idx: Integer]: TScalar.TValue read GetFields; property CycleID: Int64 read FCycleID; - property Kind: TSignalKind read FKind; - property Data: TArray read FData; + property Kind: TSignalKind read GetKind; end; TSignalProc = reference to procedure(const Signal: TStreamSignal); @@ -55,7 +56,7 @@ type function GetDef: IScalarRecordDefinition; protected { Broadcasts data to all subscribers. Stateless. } - procedure Emit(const Value: array of TScalar.TValue; ACycleID: Int64); + procedure Emit(CycleID: Int64; Values: TScalar.PValue); public constructor Create(const ADef: IScalarRecordDefinition); destructor Destroy; override; @@ -83,6 +84,7 @@ type public type TSignalProc = reference to procedure(const Signal: TStreamSignal); + strict private FSource: IStream; FTag: TSubscriptionTag; @@ -90,21 +92,21 @@ type FFieldIndex: Integer; FData: IWriteableSeries; FLastSeenCycle: Int64; - FLookback: Int64; + FMaxLookback: Int64; { ISeries implementation } function GetCount: Int64; function GetItems(Idx: Integer): TScalar; function GetTotalCount: Int64; - private procedure HandleSignal(const Signal: TStreamSignal); - protected + + private procedure Detach; + public - constructor Create(const ASource: IStream; const AField: IKeyword; ALookback: Int64; const AOnSignal: TSignalProc); + constructor Create(const ASource: IStream; const AField: IKeyword; AMaxLookback: Int64; const AOnSignal: TSignalProc); destructor Destroy; override; - function IsReady(RequiredLookback: Int64): Boolean; property LastSeenCycle: Int64 read FLastSeenCycle; end; @@ -114,11 +116,13 @@ type public type TPipeLambda = reference to function(const Sources: array of ISeries; out Results: array of TScalar.TValue): Boolean; + strict private FSources: TArray; FResults: TArray; FLastFiredCycleID: Int64; - FLookback: Int64; + FMaxLookback: Int64; + FRequiredLookback: Int64; FLambda: TPipeLambda; private @@ -130,27 +134,45 @@ type const ADef: IScalarRecordDefinition; const ASources: TArray; const ALambda: TPipeLambda; - ALookback: Int64 + AMaxLookback, ARequiredLookback: Int64 ); destructor Destroy; override; - property Lookback: Int64 read FLookback; + property MaxLookback: Int64 read FMaxLookback; + property RequiredLookback: Int64 read FRequiredLookback; end; implementation { TStreamSignal } -constructor TStreamSignal.Create(AKind: TSignalKind; ACycleID: Int64; const AData: TArray); +constructor TStreamSignal.Create(ACycleID: Int64; const AData: TScalar.PValue); begin - FKind := AKind; FCycleID := ACycleID; FData := AData; end; +function TStreamSignal.GetFields(Idx: Integer): TScalar.TValue; +var + P: TScalar.PValue; +begin + Assert(FData <> nil, 'Heartbeat does not contain data'); + P := FData; + inc(P, Idx); + Result := P^; +end; + +function TStreamSignal.GetKind: TSignalKind; +begin + if FData <> nil then + Result := skData + else + Result := skHeartbeat; +end; + function TStreamSignal.ToString: string; begin if Kind = skData then - Result := Format('Signal(Data, #%d, Fields: %d)', [CycleID, Length(Data)]) + Result := Format('Signal(Data, #%d)', [CycleID]) else Result := Format('Signal(Heartbeat, #%d)', [CycleID]); end; @@ -197,19 +219,11 @@ begin end; end; -procedure TCustomDataStream.Emit(const Value: array of TScalar.TValue; ACycleID: Int64); +procedure TCustomDataStream.Emit(CycleID: Int64; Values: TScalar.PValue); var LSignal: TStreamSignal; - LData: TArray; - I: Integer; begin - // Convert open array to dynamic array for signal payload - SetLength(LData, Length(Value)); - for I := 0 to High(Value) do - LData[I] := Value[I]; - - LSignal := TStreamSignal.Create(skData, ACycleID, LData); - + LSignal.Create(CycleID, Values); FObservers.Lock; try FObservers.Notify( @@ -220,6 +234,7 @@ begin end ); finally + LSignal.FData := nil; FObservers.Release; end; end; @@ -238,17 +253,17 @@ begin raise EArgumentException.Create('Push: Data field count mismatch.'); Inc(FLastCycleID); - Emit(RowData, FLastCycleID); + Emit(FLastCycleID, @RowData[0]); end; { TPipeSource } -constructor TPipeSource.Create(const ASource: IStream; const AField: IKeyword; ALookback: Int64; const AOnSignal: TSignalProc); +constructor TPipeSource.Create(const ASource: IStream; const AField: IKeyword; AMaxLookback: Int64; const AOnSignal: TSignalProc); begin inherited Create; FSource := ASource; FOnSignal := AOnSignal; - FLookback := ALookback; + FMaxLookback := AMaxLookback; FLastSeenCycle := -1; FFieldIndex := ASource.Def.IndexOf(AField); @@ -278,17 +293,12 @@ begin if (Signal.Kind = skData) then begin FLastSeenCycle := Signal.CycleID; - FData.Add(Signal.Data[FFieldIndex], FLookback); + FData.Add(Signal.Fields[FFieldIndex], FMaxLookback); if Assigned(FOnSignal) then FOnSignal(Signal); end; end; -function TPipeSource.IsReady(RequiredLookback: Int64): Boolean; -begin - Result := (FLastSeenCycle >= 0) and (FData.Count >= RequiredLookback); -end; - function TPipeSource.GetCount: Int64; begin Result := FData.Count; @@ -311,14 +321,14 @@ constructor TPipeStream.Create( const ADef: IScalarRecordDefinition; const ASources: TArray; const ALambda: TPipeLambda; - ALookback: Int64 + AMaxLookback, ARequiredLookback: Int64 ); var i, j, n: Integer; begin inherited Create(ADef); FLambda := ALambda; - FLookback := ALookback; + FMaxLookback := AMaxLookback; FLastFiredCycleID := -1; // Flatten sources from config @@ -327,7 +337,7 @@ begin inc(n, Length(AConfig[i])); SetLength(FSources, n); - SetLength(FResults, n); + SetLength(FResults, ADef.Count); n := 0; for i := 0 to High(ASources) do @@ -339,12 +349,13 @@ begin TPipeSource.Create( ASources[i], AConfig[i][j].Key, - FLookback, + FMaxLookback, procedure(const Signal: TStreamSignal) begin CheckBarrierAndFire(Signal.CycleID); end ); inc(n); end; end; + FRequiredLookback := ARequiredLookback; end; destructor TPipeStream.Destroy; @@ -356,12 +367,13 @@ end; procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64); begin - const requiredLookback = 10; + if FLastFiredCycleID >= CurrentCycle then + exit; - // 1. Check if all sources reached this cycle AND have enough history (Warm-up) + // Check if all sources reached this cycle AND have enough history (Warm-up) for var src in FSources do begin - if src.Count < requiredLookback then + if src.Count < FRequiredLookback then exit; var LSource := src as TPipeSource; @@ -369,18 +381,13 @@ begin exit; end; - if FLastFiredCycleID >= CurrentCycle then - Exit; - FLastFiredCycleID := CurrentCycle; if Assigned(FLambda) then begin // Execute transformation on synchronized window if FLambda(FSources, FResults) then - begin - Emit(FResults, FLastFiredCycleID); - end; + Emit(FLastFiredCycleID, @FResults[0]); end; end;