Shared series in Pipes

ASTPlayground/MainForm.pas

@@ -168,6 +168,8 @@ uses
 
 procedure TForm1.FormDestroy(Sender: TObject);
 begin
+    Timer1.Enabled := false;
+    FCurrExec.SetVoid;
     FSim := nil;
 end;
 

Src/Data/Myc.Data.Stream.pas

@@ -78,18 +78,49 @@ type
 
     { TPipeSource acts as a selective accumulator for a single field of a stream. }
     TPipeSource = class(TInterfacedObject, ISeries)
-    public
         type
-            TSignalProc = reference to procedure(const Signal: TStreamSignal);
+            TSharedSourceKey = record
+                SourceID: Pointer;
+                FieldID: Integer;
+            end;
+
+            TSharedSource = class(TObject)
+            strict private
+                FField: IKeyword;
+                FStream: IStream;
+                FTag: TSubscriptionTag;
+                FObservers: TMycNotifyList<TPipeSource>;
+                FFieldIndex: Integer;
+                FMaxLookback: Int64;
+                FData: IWriteableSeries;
+
+                procedure HandleSignal(const Signal: TStreamSignal);
+
+            private
+                function Attach(const Observer: TPipeSource): TSubscriptionTag;
+                function Detach(Tag: TSubscriptionTag): Boolean;
+
+            public
+                constructor Create(const AStream: IStream; const AField: IKeyword);
+                destructor Destroy; override;
+                property Field: IKeyword read FField;
+                property Stream: IStream read FStream;
+                property Data: IWriteableSeries read FData;
+            end;
 
     strict private
-        FSource: IStream;
+        class var
-        FTag: TSubscriptionTag;
+            FCacheLock: TSpinLock;
+            FCache: TObjectDictionary<TSharedSourceKey, TSharedSource>;
+
+        class constructor CreateClass;
+        class destructor DestroyClass;
+    private
+        FSource: TSharedSource;
         FOnSignal: TSignalProc;
-        FFieldIndex: Integer;
-        FData: IWriteableSeries;
         FLastSeenCycle: Int64;
         FMaxLookback: Int64;
+        FTag: TSubscriptionTag;
 
         { ISeries implementation }
         function GetCount: Int64;
@@ -105,6 +136,7 @@ type
         destructor Destroy; override;
 
         property LastSeenCycle: Int64 read FLastSeenCycle;
+        property MaxLookback: Int64 read FMaxLookback;
     end;
 
     TPipeConfig = TArray<TArray<TScalarRecordField>>;
@@ -254,35 +286,150 @@ begin
     Emit(FLastCycleID, @RowData[0]);
 end;
 
+{ TSharedSource }
+
+constructor TPipeSource.TSharedSource.Create(const AStream: IStream; const AField: IKeyword);
+begin
+    inherited Create;
+    FStream := AStream;
+    FField := AField;
+
+    FMaxLookback := -1;
+
+    FFieldIndex := FStream.Def.IndexOf(FField);
+    if (FFieldIndex < 0) then
+        raise EArgumentException.CreateFmt('Field %s not found in source stream.', [FField.Name]);
+
+    FData := TScalarSeries.Create(FStream.Def[FFieldIndex]);
+
+    FTag := FStream.Subscribe(HandleSignal);
+end;
+
+destructor TPipeSource.TSharedSource.Destroy;
+begin
+    FStream.Unsubscribe(FTag);
+    inherited;
+end;
+
+function TPipeSource.TSharedSource.Attach(const Observer: TPipeSource): TSubscriptionTag;
+begin
+    FObservers.Lock;
+    try
+        FMaxLookback := -1;
+        Result := FObservers.Advise(Observer);
+    finally
+        FObservers.Release;
+    end;
+end;
+
+function TPipeSource.TSharedSource.Detach(Tag: TSubscriptionTag): Boolean;
+begin
+    FObservers.Lock;
+    try
+        FObservers.Unadvise(Tag);
+        FMaxLookback := -1;
+
+        // Hint to caller, if this was the last observer
+        Result := FObservers.First = nil;
+    finally
+        FObservers.Release;
+    end;
+end;
+
+procedure TPipeSource.TSharedSource.HandleSignal(const Signal: TStreamSignal);
+begin
+    if (Signal.Kind = skData) then
+    begin
+        FObservers.Lock;
+        try
+            if FMaxLookback < 0 then
+            begin
+                FMaxLookback := 0;
+                var Item := FObservers.First;
+                while Item <> nil do
+                begin
+                    var lb := Item.Receiver.MaxLookback;
+                    if lb > FMaxLookback then
+                        FMaxLookback := lb;
+                    Item := Item.Next;
+                end;
+            end;
+
+            FData.Add(Signal.Fields[FFieldIndex], FMaxLookback);
+
+            FObservers.Notify(
+                function(const Obs: TPipeSource): Boolean
+                begin
+                    Obs.HandleSignal(Signal);
+                    Result := True;
+                end
+            );
+        finally
+            FObservers.Release;
+        end;
+    end;
+end;
+
 { TPipeSource }
 
 constructor TPipeSource.Create(const ASource: IStream; const AField: IKeyword; AMaxLookback: Int64; const AOnSignal: TSignalProc);
+var
+    Key: TSharedSourceKey;
 begin
     inherited Create;
-    FSource := ASource;
     FOnSignal := AOnSignal;
     FMaxLookback := AMaxLookback;
     FLastSeenCycle := -1;
 
-    FFieldIndex := ASource.Def.IndexOf(AField);
+    Key.SourceID := Pointer(ASource);
-    if (FFieldIndex < 0) then
+    Key.FieldID := AField.Idx;
-        raise EArgumentException.CreateFmt('Field %s not found in source stream.', [AField.Name]);
 
-    FData := TScalarSeries.Create(ASource.Def.Items[FFieldIndex]);
+    FCacheLock.Enter;
-
+    try
-    FTag := FSource.Subscribe(procedure(const Signal: TStreamSignal) begin HandleSignal(Signal); end);
+        if not FCache.TryGetValue(Key, FSource) then
+        begin
+            FSource := TSharedSource.Create(ASource, AField);
+            FCache.Add(Key, FSource);
+        end;
+        FTag := FSource.Attach(Self);
+    finally
+        FCacheLock.Exit;
+    end;
 end;
 
 destructor TPipeSource.Destroy;
 begin
     Detach;
+
+    FCacheLock.Enter;
+    try
+        if FSource.Detach(FTag) then
+        begin
+            var Key: TSharedSourceKey;
+            Key.SourceID := Pointer(FSource.Stream);
+            Key.FieldID := FSource.Field.Idx;
+            FCache.Remove(Key);
+        end;
+    finally
+        FCacheLock.Exit;
+    end;
+
     inherited;
 end;
 
+class constructor TPipeSource.CreateClass;
+begin
+    FCache := TObjectDictionary<TSharedSourceKey, TSharedSource>.Create([doOwnsValues]);
+end;
+
+class destructor TPipeSource.DestroyClass;
+begin
+    Assert(FCache.Count = 0, 'Not all pipes have been properly released.');
+    FCache.Free;
+end;
+
 procedure TPipeSource.Detach;
 begin
-    FSource.Unsubscribe(FTag);
-    FTag := nil;
     FOnSignal := nil;
 end;
 
@@ -291,7 +438,6 @@ begin
     if (Signal.Kind = skData) then
     begin
         FLastSeenCycle := Signal.CycleID;
-        FData.Add(Signal.Fields[FFieldIndex], FMaxLookback);
         if Assigned(FOnSignal) then
             FOnSignal(Signal);
     end;
@@ -299,17 +445,21 @@ end;
 
 function TPipeSource.GetCount: Int64;
 begin
-    Result := FData.Count;
+    Result := FSource.Data.Count;
 end;
 
 function TPipeSource.GetItems(Idx: Integer): TScalar;
 begin
-    Result := FData.Items[Idx];
+    // Test the max lookback of THIS source, because the shared one might have a larger max lookback
+    if Idx >= FMaxLookback then
+        raise EArgumentException.Create('Item index is greater than max lookback');
+
+    Result := FSource.Data[Idx];
 end;
 
 function TPipeSource.GetTotalCount: Int64;
 begin
-    Result := FData.TotalCount;
+    Result := FSource.Data.TotalCount;
 end;
 
 { TPipeStream }

Src/Data/Myc.Data.Value.pas

@@ -98,6 +98,8 @@ type
 
         class function Map(const SourceSeries: ISeries; const MapperFunc: TFunc): ISeries; static;
 
+        procedure SetVoid;
+
         // --- Existing Accessors ---
         function AsScalar: TScalar; inline;
         function AsMethod: TFunc; inline;
@@ -377,6 +379,12 @@ begin
     Result := ITuple<TDataValue>(FInterface);
 end;
 
+procedure TDataValue.SetVoid;
+begin
+    FInterface := nil;
+    FKind := vkVoid;
+end;
+
 function TDataValue.ToString: String;
 var
     sb: TStringBuilder;