Brummel/MycLib
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Streams refactoring

Browse Source
This commit is contained in:
Michael Schimmel
2026-01-25 01:44:36 +01:00
parent ca1d9b95f7
commit 4daa05efda
8 changed files with 188 additions and 724 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/AST/Myc.Ast.Evaluator.pas
+1 -4
View File
@@ -62,7 +62,6 @@ uses
Myc.Data.Decimal,
Myc.Data.Series,
Myc.Data.Stream,
Myc.Data.Stream.Pipes,
Myc.Ast.Types;
type
@@ -602,9 +601,7 @@ begin
// Compile the transformation function
lambdaFunc := Visit(N.Transformation).AsMethod();
// WARNING: If this is executing dynamically (without TypeChecker), StaticType is Unknown.
// Pipe creation REQUIRES the output definition.
// Fixed: Allow both stRecord and stRecordSeries (as TypeChecker correctly assigns stRecordSeries for Pipe nodes)
// Allow both stRecord and stRecordSeries (as TypeChecker correctly assigns stRecordSeries for Pipe nodes)
if (N.StaticType.Kind <> stRecordSeries) and (N.StaticType.Kind <> stRecord) then
raise EEvaluatorException.Create('Pipe requires Type Checking to determine output structure (RecordDefinition).');
Src/Data/Myc.Data.Stream.Pipes.pas
-280
View File
@@ -1,280 +0,0 @@
unit Myc.Data.Stream.Pipes;
interface
uses
System.SysUtils,
System.Classes,
System.Generics.Collections,
System.SyncObjs,
Myc.Data.Scalar,
Myc.Data.Keyword,
Myc.Data.Stream,
Myc.Core.Notifier;
type
// =========================================================================
// BASE STREAM
// Implements storage (Series) and broadcasting (Observers).
// Does not define WHEN data is emitted (Policy).
// =========================================================================
TCustomDataStream = class(TInterfacedObject, IStream)
strict private
FSeries: IWriteableScalarRecordSeries;
FObservers: TMycNotifyList<IStreamObserver>;
function GetSeries: IScalarRecordSeries;
protected
// Derived classes call this to write data and notify listeners.
// ACycleID: The cycle this data belongs to.
procedure Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
public
constructor Create(const ADef: IScalarRecordDefinition);
destructor Destroy; override;
// IStream
function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
procedure Unsubscribe(Tag: TSubscriptionTag);
property Series: IScalarRecordSeries read GetSeries;
end;
// =========================================================================
// ROOT STREAM (SOURCE)
// Acts as the clock source. Generates new CycleIDs.
// =========================================================================
TRootStream = class(TCustomDataStream)
private
FLastCycleID: Int64;
public
constructor Create(const ADef: IScalarRecordDefinition);
// Public API to inject data from Delphi
procedure Push(const RowData: TArray<TScalar.TValue>);
end;
// =========================================================================
// PIPE STREAM (NODE)
// Reacts to upstream signals using barrier synchronization.
// =========================================================================
TPipeStream = class; // Forward
TPipeConfig = TArray<TArray<TScalarRecordField>>;
TPipeSource = class(TContainedObject, IStreamObserver)
private
FSource: IStream;
FTag: TSubscriptionTag;
FLastSeenCycle: Int64;
procedure OnSignal(const Signal: TStreamSignal);
public
constructor Create(AOwner: TPipeStream; ASource: IStream);
destructor Destroy; override;
property LastSeenCycle: Int64 read FLastSeenCycle;
end;
TPipeStream = class(TCustomDataStream)
public
type
TPipeLambda = reference to function(const Sources: array of ISeries; out Results: array of TScalar.TValue): Boolean;
private
FSources: TArray<TPipeSource>;
FSourceSeries: TArray<ISeries>;
FLastFiredCycleID: Int64;
FLambda: TPipeLambda;
procedure CheckBarrierAndFire(CurrentCycle: Int64);
public
constructor Create(
const AConfig: TPipeConfig;
const ADef: IScalarRecordDefinition;
const ASources: TArray<IStream>;
const ALambda: TPipeLambda
);
destructor Destroy; override;
end;
implementation
{ TCustomDataStream }
constructor TCustomDataStream.Create(const ADef: IScalarRecordDefinition);
begin
inherited Create;
FSeries := TScalarRecordSeries.Create(ADef);
end;
destructor TCustomDataStream.Destroy;
begin
FObservers.Finalize;
inherited;
end;
function TCustomDataStream.GetSeries: IScalarRecordSeries;
begin
Result := FSeries;
end;
function TCustomDataStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
begin
FObservers.Lock;
try
Result := FObservers.Advise(Observer);
finally
FObservers.Release;
end;
end;
procedure TCustomDataStream.Unsubscribe(Tag: TSubscriptionTag);
begin
FObservers.Lock;
try
FObservers.Unadvise(Tag);
finally
FObservers.Release;
end;
end;
procedure TCustomDataStream.Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
var
signal: TStreamSignal;
begin
FObservers.Lock;
try
// 1. Write Data
FSeries.Add(Value);
// 2. Broadcast Signal
signal := TStreamSignal.Create(skData, ACycleID);
FObservers.Notify(
function(const Obs: IStreamObserver): Boolean
begin
Obs.OnSignal(signal);
Result := True;
end
);
finally
FObservers.Release;
end;
end;
{ TRootStream }
constructor TRootStream.Create(const ADef: IScalarRecordDefinition);
begin
inherited Create(ADef);
FLastCycleID := 0;
end;
procedure TRootStream.Push(const RowData: TArray<TScalar.TValue>);
begin
// Root streams increment the global clock
Inc(FLastCycleID);
Emit(RowData, FLastCycleID);
end;
{ TPipeSource }
constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream);
begin
inherited Create(AOwner);
FSource := ASource;
FLastSeenCycle := -1;
FTag := FSource.Subscribe(Self);
end;
destructor TPipeSource.Destroy;
begin
FSource.Unsubscribe(FTag);
inherited;
end;
procedure TPipeSource.OnSignal(const Signal: TStreamSignal);
begin
if Signal.Kind = skData then
begin
FLastSeenCycle := Signal.CycleID;
(Controller as TPipeStream).CheckBarrierAndFire(FLastSeenCycle);
end;
end;
{ TPipeStream }
constructor TPipeStream.Create(
const AConfig: TPipeConfig;
const ADef: IScalarRecordDefinition;
const ASources: TArray<IStream>;
const ALambda: TPipeLambda
);
var
i, j, n: Integer;
begin
// Pass Definition to base class
inherited Create(ADef);
FLambda := ALambda;
FLastFiredCycleID := -1;
SetLength(FSources, Length(ASources));
// Flatten Sources
n := 0;
for i := 0 to High(AConfig) do
inc(n, Length(AConfig[i]));
SetLength(FSourceSeries, n);
n := 0;
for i := 0 to High(ASources) do
begin
FSources[i] := TPipeSource.Create(Self, ASources[i]);
for j := 0 to High(AConfig[i]) do
begin
// Extract the specific column series
FSourceSeries[n] := ASources[i].Series.Fields[AConfig[i][j].Key];
inc(n);
end;
end;
end;
destructor TPipeStream.Destroy;
begin
for var i := High(FSources) downto 0 do
FSources[i].Free;
FSources := nil;
FSourceSeries := nil;
inherited;
end;
procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
var
resultVal: TArray<TScalar.TValue>;
begin
// We reuse the FObservers lock from the base class to protect state
// But since FObservers is private, we access it via method or need to make it protected.
// Making it protected or using a separate lock is cleaner.
// For now, let's assume we rely on the fact that OnSignal is usually serialized per thread
// or add a lock.
// -> Ideally, TCustomDataStream should expose Lock/Unlock or we add a Lock here.
// Simplification: We assume thread safety is handled by the caller or we add a dedicated lock.
// For this example, let's just do the logic:
for var src in FSources do
if src.LastSeenCycle < CurrentCycle then
exit; // Barrier closed
if FLastFiredCycleID >= CurrentCycle then
exit; // Already fired
FLastFiredCycleID := CurrentCycle;
if Assigned(FLambda) then
begin
SetLength(resultVal, Series.Def.Count);
if FLambda(FSourceSeries, resultVal) then
begin
// Pass through CycleID from upstream
Emit(resultVal, FLastFiredCycleID);
end;
end;
end;
end.
Src/Data/Myc.Data.Stream.pas
+186 -123
View File
@@ -43,61 +43,82 @@ type
property Series: IScalarRecordSeries read GetSeries;
end;
IInputChannel = interface
{$region 'private'}
function GetStream: IStream;
{$endregion}
procedure Push(const Value: IKeywordMapping<TScalar>);
property Stream: IStream read GetStream;
end;
IGraphExecutor = interface
{$region 'private'}
function GetCycleID: Int64;
{$endregion}
function GetInputChannel(const Name: string; const Def: IScalarRecordDefinition): IInputChannel;
procedure Step;
property CycleID: Int64 read GetCycleID;
end;
TGraphExecutor = class(TInterfacedObject, IGraphExecutor)
private
type
TChannel = class(TInterfacedObject, IInputChannel, IStream)
private
type
TStreamNotifier = TMycNotifyList<IStreamObserver>;
private
FName: string;
FSeries: IWriteableScalarRecordSeries;
FQueue: TQueue<IKeywordMapping<TScalar>>;
FQueueLock: TSpinLock;
FObservers: TStreamNotifier;
procedure Push(const Value: IKeywordMapping<TScalar>);
function GetStream: IStream;
function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
procedure Unsubscribe(Tag: TSubscriptionTag);
procedure Emit(ACycle: Int64);
function GetSeries: IScalarRecordSeries;
public
constructor Create(const AName: string; const ASeries: IWriteableScalarRecordSeries);
destructor Destroy; override;
end;
private
FCycleID: Int64;
FChannels: TDictionary<string, IInputChannel>;
FLock: TSpinLock;
function GetInputChannel(const Name: string; const Def: IScalarRecordDefinition): IInputChannel;
function GetCycleID: Int64;
// =========================================================================
// BASE STREAM
// Implements storage (Series) and broadcasting (Observers).
// Does not define WHEN data is emitted (Policy).
// =========================================================================
TCustomDataStream = class(TInterfacedObject, IStream)
strict private
FSeries: IWriteableScalarRecordSeries;
FObservers: TMycNotifyList<IStreamObserver>;
function GetSeries: IScalarRecordSeries;
protected
// Derived classes call this to write data and notify listeners.
// ACycleID: The cycle this data belongs to.
procedure Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
public
constructor Create;
constructor Create(const ADef: IScalarRecordDefinition);
destructor Destroy; override;
// IStream
function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
procedure Unsubscribe(Tag: TSubscriptionTag);
property Series: IScalarRecordSeries read GetSeries;
end;
// =========================================================================
// ROOT STREAM (SOURCE)
// Acts as the clock source. Generates new CycleIDs.
// =========================================================================
TRootStream = class(TCustomDataStream)
private
FLastCycleID: Int64;
public
constructor Create(const ADef: IScalarRecordDefinition);
// Public API to inject data from Delphi
procedure Push(const RowData: TArray<TScalar.TValue>);
end;
// =========================================================================
// PIPE STREAM (NODE)
// Reacts to upstream signals using barrier synchronization.
// =========================================================================
TPipeStream = class; // Forward
TPipeConfig = TArray<TArray<TScalarRecordField>>;
TPipeSource = class(TContainedObject, IStreamObserver)
private
FSource: IStream;
FTag: TSubscriptionTag;
FLastSeenCycle: Int64;
procedure OnSignal(const Signal: TStreamSignal);
public
constructor Create(AOwner: TPipeStream; ASource: IStream);
destructor Destroy; override;
property LastSeenCycle: Int64 read FLastSeenCycle;
end;
TPipeStream = class(TCustomDataStream)
public
type
TPipeLambda = reference to function(const Sources: array of ISeries; out Results: array of TScalar.TValue): Boolean;
private
FSources: TArray<TPipeSource>;
FSourceSeries: TArray<ISeries>;
FLastFiredCycleID: Int64;
FLambda: TPipeLambda;
procedure CheckBarrierAndFire(CurrentCycle: Int64);
public
constructor Create(
const AConfig: TPipeConfig;
const ADef: IScalarRecordDefinition;
const ASources: TArray<IStream>;
const ALambda: TPipeLambda
);
destructor Destroy; override;
procedure Step;
end;
implementation
@@ -119,39 +140,26 @@ begin
Result := Format('Signal(Heartbeat, #%d)', [CycleID]);
end;
{ TGraphExecutor.TChannel }
{ TCustomDataStream }
constructor TGraphExecutor.TChannel.Create(const AName: string; const ASeries: IWriteableScalarRecordSeries);
constructor TCustomDataStream.Create(const ADef: IScalarRecordDefinition);
begin
inherited Create;
FName := AName;
FSeries := ASeries;
FQueue := TQueue<IKeywordMapping<TScalar>>.Create;
FSeries := TScalarRecordSeries.Create(ADef);
end;
destructor TGraphExecutor.TChannel.Destroy;
destructor TCustomDataStream.Destroy;
begin
FObservers.Finalize;
FQueue.Free;
inherited;
end;
procedure TGraphExecutor.TChannel.Push(const Value: IKeywordMapping<TScalar>);
function TCustomDataStream.GetSeries: IScalarRecordSeries;
begin
FQueueLock.Enter;
try
FQueue.Enqueue(Value);
finally
FQueueLock.Exit;
end;
Result := FSeries;
end;
function TGraphExecutor.TChannel.GetStream: IStream;
begin
Result := Self;
end;
function TGraphExecutor.TChannel.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
function TCustomDataStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
begin
FObservers.Lock;
try
@@ -161,7 +169,7 @@ begin
end;
end;
procedure TGraphExecutor.TChannel.Unsubscribe(Tag: TSubscriptionTag);
procedure TCustomDataStream.Unsubscribe(Tag: TSubscriptionTag);
begin
FObservers.Lock;
try
@@ -171,25 +179,18 @@ begin
end;
end;
procedure TGraphExecutor.TChannel.Emit(ACycle: Int64);
procedure TCustomDataStream.Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
var
signal: TStreamSignal;
begin
FQueueLock.Enter;
try
FSeries.Add(FQueue.Dequeue);
signal :=
TStreamSignal.Create(
if FQueue.Count > 0 then skData
else skHeartbeat,
ACycle
)
finally
FQueueLock.Exit;
end;
FObservers.Lock;
try
// 1. Write Data
FSeries.Add(Value);
// 2. Broadcast Signal
signal := TStreamSignal.Create(skData, ACycleID);
FObservers.Notify(
function(const Obs: IStreamObserver): Boolean
begin
@@ -202,63 +203,125 @@ begin
end;
end;
function TGraphExecutor.TChannel.GetSeries: IScalarRecordSeries;
{ TRootStream }
constructor TRootStream.Create(const ADef: IScalarRecordDefinition);
begin
Result := FSeries;
inherited Create(ADef);
FLastCycleID := 0;
end;
{ TGraphExecutor }
constructor TGraphExecutor.Create;
procedure TRootStream.Push(const RowData: TArray<TScalar.TValue>);
begin
inherited Create;
FChannels := TDictionary<string, IInputChannel>.Create;
FCycleID := 0;
// Root streams increment the global clock
Inc(FLastCycleID);
Emit(RowData, FLastCycleID);
end;
destructor TGraphExecutor.Destroy;
{ TPipeSource }
constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream);
begin
FChannels.Free;
inherited Create(AOwner);
FSource := ASource;
FLastSeenCycle := -1;
FTag := FSource.Subscribe(Self);
end;
destructor TPipeSource.Destroy;
begin
FSource.Unsubscribe(FTag);
inherited;
end;
function TGraphExecutor.GetInputChannel(const Name: string; const Def: IScalarRecordDefinition): IInputChannel;
var
impl: TChannel;
procedure TPipeSource.OnSignal(const Signal: TStreamSignal);
begin
FLock.Enter;
try
if not FChannels.TryGetValue(Name, Result) then
if Signal.Kind = skData then
begin
FLastSeenCycle := Signal.CycleID;
(Controller as TPipeStream).CheckBarrierAndFire(FLastSeenCycle);
end;
end;
{ TPipeStream }
constructor TPipeStream.Create(
const AConfig: TPipeConfig;
const ADef: IScalarRecordDefinition;
const ASources: TArray<IStream>;
const ALambda: TPipeLambda
);
var
i, j, n: Integer;
begin
// Pass Definition to base class
inherited Create(ADef);
FLambda := ALambda;
FLastFiredCycleID := -1;
SetLength(FSources, Length(ASources));
// Flatten Sources
n := 0;
for i := 0 to High(AConfig) do
inc(n, Length(AConfig[i]));
SetLength(FSourceSeries, n);
n := 0;
for i := 0 to High(ASources) do
begin
FSources[i] := TPipeSource.Create(Self, ASources[i]);
for j := 0 to High(AConfig[i]) do
begin
impl := TChannel.Create(Name, TScalarRecordSeries.Create(Def));
Result := impl;
FChannels.Add(Name, Result);
// Extract the specific column series
FSourceSeries[n] := ASources[i].Series.Fields[AConfig[i][j].Key];
inc(n);
end;
finally
FLock.Exit;
end;
end;
function TGraphExecutor.GetCycleID: Int64;
destructor TPipeStream.Destroy;
begin
Result := FCycleID;
for var i := High(FSources) downto 0 do
FSources[i].Free;
FSources := nil;
FSourceSeries := nil;
inherited;
end;
procedure TGraphExecutor.Step;
procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
var
channelsArr: TArray<IInputChannel>;
channel: IInputChannel;
resultVal: TArray<TScalar.TValue>;
begin
FLock.Enter;
try
Inc(FCycleID);
channelsArr := FChannels.Values.ToArray;
finally
FLock.Exit;
end;
// We reuse the FObservers lock from the base class to protect state
// But since FObservers is private, we access it via method or need to make it protected.
// Making it protected or using a separate lock is cleaner.
// For now, let's assume we rely on the fact that OnSignal is usually serialized per thread
// or add a lock.
// -> Ideally, TCustomDataStream should expose Lock/Unlock or we add a Lock here.
for channel in channelsArr do
(channel as TChannel).Emit(FCycleID);
// Simplification: We assume thread safety is handled by the caller or we add a dedicated lock.
// For this example, let's just do the logic:
for var src in FSources do
if src.LastSeenCycle < CurrentCycle then
exit; // Barrier closed
if FLastFiredCycleID >= CurrentCycle then
exit; // Already fired
FLastFiredCycleID := CurrentCycle;
if Assigned(FLambda) then
begin
SetLength(resultVal, Series.Def.Count);
if FLambda(FSourceSeries, resultVal) then
begin
// Pass through CycleID from upstream
Emit(resultVal, FLastFiredCycleID);
end;
end;
end;
initialization