From 8960a5683ea799d0d7539d1395fe8978c8a6f701 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Thu, 18 Dec 2025 13:09:47 +0100 Subject: [PATCH] Adding Pipes --- Src/Data/Myc.Data.Scalar.pas | 12 +-- Src/Data/Myc.Data.Stream.Pipes.pas | 122 +++++++++++++++-------------- 2 files changed, 68 insertions(+), 66 deletions(-) diff --git a/Src/Data/Myc.Data.Scalar.pas b/Src/Data/Myc.Data.Scalar.pas index 74299cd..509cff2 100644 --- a/Src/Data/Myc.Data.Scalar.pas +++ b/Src/Data/Myc.Data.Scalar.pas @@ -200,7 +200,7 @@ type function GetTotalCount: Int64; {$endregion} procedure Add(const Item: IKeywordMapping; Lookback: Int64 = -1); overload; - procedure Add(const Items: TArray; Lookback: Int64 = -1); overload; + procedure Add(const Items: array of TScalar.TValue; Lookback: Int64 = -1); overload; property RecordCount: Int64 read GetRecordCount; property TotalCount: Int64 read GetTotalCount; end; @@ -235,13 +235,12 @@ type function GetDef: IScalarRecordDefinition; inline; function GetTotalCount: Int64; inline; function GetItemRef(Idx: Integer): TChunkArray.PT; inline; - function GetMapping: IKeywordMapping; public constructor Create(const ADef: IScalarRecordDefinition); destructor Destroy; override; procedure Add(const Item: IKeywordMapping; Lookback: Int64 = -1); overload; - procedure Add(const Items: TArray; Lookback: Int64 = -1); overload; + procedure Add(const Items: array of TScalar.TValue; Lookback: Int64 = -1); overload; property RecordCount: Int64 read GetRecordCount; property Def: IScalarRecordDefinition read GetDef; @@ -829,7 +828,7 @@ begin inc(FTotalCount); end; -procedure TScalarRecordSeries.Add(const Items: TArray; Lookback: Int64 = -1); +procedure TScalarRecordSeries.Add(const Items: array of TScalar.TValue; Lookback: Int64 = -1); begin Assert(Length(Items) = FDef.Count, 'Array does not fit series definition'); @@ -864,11 +863,6 @@ begin Result := FArray.ItemRef[(FArray.Count - len) - (Idx * len)]; end; -function TScalarRecordSeries.GetMapping: IKeywordMapping; -begin - Result := Self; -end; - function TScalarRecordSeries.GetTotalCount: Int64; begin Result := FTotalCount; diff --git a/Src/Data/Myc.Data.Stream.Pipes.pas b/Src/Data/Myc.Data.Stream.Pipes.pas index b3d80ed..20fbd52 100644 --- a/Src/Data/Myc.Data.Stream.Pipes.pas +++ b/Src/Data/Myc.Data.Stream.Pipes.pas @@ -77,11 +77,12 @@ type // Acts as IStream (Producer) for downstream consumers and manages internal inputs. TPipeStream = class(TInterfacedObject, IStream) type - TProc = reference to procedure(const Series: TArray); + TEmitProc = reference to procedure(const Data: array of TScalar.TValue); + TProc = reference to procedure(const Sources: TArray; const Emit: TEmitProc); private - FConfig: TPipeConfig; FSources: TArray; FObservers: TMycNotifyList; + FSourceSeries: TArray; FSeries: IWriteableScalarRecordSeries; // Barrier State @@ -90,17 +91,21 @@ type // Execution Logic // The compiled lambda function representing (fn [inputs...] ...) FLambda: TProc; - FLambdaArgs: TArray; procedure CheckBarrierAndFire(CurrentCycle: Int64); function GetSeries: IScalarRecordSeries; public - constructor Create(const AConfig: TPipeConfig; const ASources: TArray; const ALambda: TProc); + constructor Create( + const AConfig: TPipeConfig; + const ADef: IScalarRecordDefinition; + const ASources: TArray; + const ALambda: TProc + ); destructor Destroy; override; // Injected into lambda scope as "emit" - procedure Emit(const Value: IScalarRecord); + procedure Emit(const Data: array of TScalar.TValue); // IStream Implementation function Subscribe(const Observer: IStreamObserver): TSubscriptionTag; @@ -138,20 +143,25 @@ end; { TPipeStream } -constructor TPipeStream.Create(const AConfig: TPipeConfig; const ASources: TArray; const ALambda: TProc); -var - fields: TArray; +constructor TPipeStream.Create( + const AConfig: TPipeConfig; + const ADef: IScalarRecordDefinition; + const ASources: TArray; + const ALambda: TProc +); begin inherited Create; - FConfig := AConfig; FLambda := ALambda; FLastFiredCycleID := -1; Assert(Length(AConfig) = Length(ASources)); - SetLength(FSources, Length(ASources)); + FSeries := TScalarRecordSeries.Create(ADef); // Construct record series definition from source config + SetLength(FSources, Length(ASources)); + SetLength(FSourceSeries, Length(ASources)); + var n := 0; for var i := 0 to High(AConfig) do begin @@ -160,22 +170,14 @@ begin inc(n, Length(AConfig[i])); end; - SetLength(FLambdaArgs, n); - + SetLength(FSourceSeries, n); n := 0; for var i := 0 to High(AConfig) do - begin for var j := 0 to High(AConfig[i]) do begin - fields[n] := AConfig[i][j]; - - FLambdaArgs[n] := ASources[i].Series[j].Value; - + FSourceSeries[n] := ASources[i].Series.Fields[AConfig[i][j].Key]; inc(n); end; - end; - - FSeries := TScalarRecordSeries.Create(TScalarRecord.TRegistry.Intern(fields)); end; destructor TPipeStream.Destroy; @@ -210,52 +212,58 @@ begin end; procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64); -begin - // Barrier Check - for var src in FSources do - begin - if src.LastSeenCycle < CurrentCycle then - exit; - end; - - // --- Barrier Open --- - - if FLastFiredCycleID >= CurrentCycle then - exit; - - FLastFiredCycleID := CurrentCycle; - - // Execute Lambda - if Assigned(FLambda) then - begin - try - FLambda(FLambdaArgs); - except - // Error handling strategy: Propagate/Crash for now. - raise; - end; - end; -end; - -procedure TPipeStream.Emit(const Value: IScalarRecord); begin FObservers.Lock; try - FSeries.Add(Value); - var signal := TStreamSignal.Create(skData, FLastFiredCycleID); + // Barrier Check + for var src in FSources do + begin + if src.LastSeenCycle < CurrentCycle then + exit; + end; - FObservers.Notify( - function(const Obs: IStreamObserver): Boolean - begin - Obs.OnSignal(signal); - Result := True; - end - ); + // --- Barrier Open --- + + if FLastFiredCycleID >= CurrentCycle then + exit; + + FLastFiredCycleID := CurrentCycle; + + // Execute Lambda + if Assigned(FLambda) then + begin + try + FLambda(FSourceSeries, Emit); + except + // Error handling strategy: Propagate/Crash for now. + raise; + end; + end; finally FObservers.Release; end; end; +procedure TPipeStream.Emit(const Data: array of TScalar.TValue); +begin + Assert(Length(Data) = FSeries.Def.Count); + + // Emit has to be called from the Lambda, which is called in CheckBarrierAndFire! + Assert(FObservers.IsLocked); + + FSeries.Add(Data); + + var signal := TStreamSignal.Create(skData, FLastFiredCycleID); + + FObservers.Notify( + function(const Obs: IStreamObserver): Boolean + begin + Obs.OnSignal(signal); + Result := True; + end + ); +end; + function TPipeStream.GetSeries: IScalarRecordSeries; begin Result := FSeries;