MycLib/Src/Data/Myc.Data.Stream.pas

unit Myc.Data.Stream;

interface

uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.SyncObjs,
    Myc.Data.Scalar,
    Myc.Data.Keyword,
    Myc.Core.Notifier;

type
    TSignalKind = (skData, skHeartbeat);

    TStreamSignal = record
    private
        FCycleID: Int64;
        FKind: TSignalKind;
    public
        constructor Create(AKind: TSignalKind; ACycleID: Int64);
        function ToString: string;
        property CycleID: Int64 read FCycleID;
        property Kind: TSignalKind read FKind;
    end;

    IStreamObserver = interface
        procedure OnSignal(const Signal: TStreamSignal);
    end;

    TSubscriptionTag = Pointer;

    IStream = interface
        {$region 'private'}
        function GetSeries: IScalarRecordSeries;
        {$endregion}

        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
        procedure Unsubscribe(Tag: TSubscriptionTag);

        // Returns the Series of the records this stream produces. Streams are scalar record series. Always!
        property Series: IScalarRecordSeries read GetSeries;
    end;

    // =========================================================================
    // 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

uses
    Winapi.Windows;

constructor TStreamSignal.Create(AKind: TSignalKind; ACycleID: Int64);
begin
    FKind := AKind;
    FCycleID := ACycleID;
end;

function TStreamSignal.ToString: string;
begin
    if Kind = skData then
        Result := Format('Signal(Data, #%d)', [CycleID])
    else
        Result := Format('Signal(Heartbeat, #%d)', [CycleID]);
end;

{ 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
            var m := ASources[i].Series.IndexOf(AConfig[i][j].Key);
            if m >= 0 then
            begin
                FSourceSeries[n] := ASources[i].Series[m];
                inc(n);
            end;
        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 assume thread safety is handled by the caller

    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
    TMycNotifyList<IStreamObserver>.ReverseOnNotify := False;

end.