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

Data pipeline refactoring

Browse Source
This commit is contained in:
Michael Schimmel
2025-12-14 12:06:29 +01:00
parent f88fe9f5ef
commit f0567a32a1
2 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Data/Myc.Data.Pipeline.pas
+302
View File
@@ -0,0 +1,302 @@
unit Myc.Data.Pipeline;
interface
{$M+}
uses
Myc.Signals,
Myc.Mutable,
Myc.Data.Series;
type
TTag = Pointer;
// A generic interface for components that consume data of type T.
IConsumer<T> = interface
function Consume(const Value: T): TState;
end;
// A producer generates data and distributes it to linked consumers.
IProducer<T> = interface
function Link(const Consumer: IConsumer<T>): TTag;
procedure Unlink(Tag: TTag);
end;
// A converter is a producer that internally uses a consumer to transform data.
IConverter<S, T> = interface(IProducer<T>)
{$region 'private'}
function GetConsumer: IConsumer<S>;
{$endregion}
property Consumer: IConsumer<S> read GetConsumer;
end;
TConvertFunc<S, T> = reference to function(const Value: S): T;
TBroadcastFunc<T> = reference to function(const Value: T): TState;
TAggregateFunc<S, T> = reference to function(const Value: S; const Broadcast: TBroadcastFunc<T>): TState;
// Interface helper for IProducer providing the null object pattern nad subscriptions for linked consumers.
TProducer<T> = record
public
type
TSubscription = record
private
FOwner: IProducer<T>;
FTag: TTag;
public
procedure Unlink;
class operator Implicit(A: TSubscription): TProducer<T>; overload;
property Owner: IProducer<T> read FOwner;
end;
private
FProducer: IProducer<T>;
class function GetNull: IProducer<T>; static; inline;
public
constructor Create(const AProducer: IProducer<T>);
// Managed record operators
class operator Initialize(out Dest: TProducer<T>);
class operator Implicit(const A: IProducer<T>): TProducer<T>; overload;
class operator Implicit(const A: TProducer<T>): IProducer<T>; overload;
function CreateLink(const Consumer: IConsumer<T>): TSubscription; inline;
function CreateEndpoint(Lookback: Int64; out Series: TLazy<TSeries<T>>): TSubscription;
function CreateSequence(Count: Integer; out Seq: TArray<IProducer<T>>): TSubscription; overload; experimental;
// Chain consumers
function Chain(const Next: IConsumer<T>): IConsumer<T>; overload; inline;
function Chain<R>(const Next: IConverter<T, R>): TProducer<R>; overload; inline;
function Chain<R>(const Func: TConvertFunc<T, R>): TProducer<R>; overload; inline;
// Extracts the field of a record by it's name (using RTTI).
function Field<R>(const FieldName: String): TProducer<R>; inline;
function MakeParallel: TProducer<T>; inline;
// Provides access to the null object instance.
class property Null: IProducer<T> read GetNull;
end;
// Interface helper for IConverter<S,T> providing the null object pattern.
TConverter<S, T> = record
private
FConverter: IConverter<S, T>;
function GetConsumer: IConsumer<S>; inline;
function GetProducer: TProducer<T>; inline;
class function GetNull: IConverter<S, T>; static; inline;
public
constructor Create(const AConverter: IConverter<S, T>);
// Managed record operators
class operator Initialize(out Dest: TConverter<S, T>);
class operator Implicit(const A: IConverter<S, T>): TConverter<S, T>; overload;
class operator Implicit(const A: TConverter<S, T>): IConverter<S, T>; overload;
class function Construct(const Consumer: IConsumer<S>; const Producer: TProducer<T>): TConverter<S, T>; static;
class function CreateConverter(const Func: TConvertFunc<S, T>): TConverter<S, T>; static;
class function CreateAggregation(const Func: TAggregateFunc<S, T>): TConverter<S, T>; static;
property Consumer: IConsumer<S> read GetConsumer;
property Producer: TProducer<T> read GetProducer;
class property Null: IConverter<S, T> read GetNull;
end;
// Tools for creating specific converter instances.
TConverter = record
class function CreateCounter<T>: TConverter<T, Int64>; static;
class function CreateTicker<T>: TConverter<TArray<T>, T>; static;
class function CreateIdentity<T>: TConverter<T, T>; static;
class function CreateEndpoint<T>(Lookback: Int64; out Series: TLazy<TSeries<T>>): IConsumer<T>; static;
type
TJoinMode = (jmAll, jmAny);
class function Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; static;
end;
implementation
uses
Myc.Data.Pipeline.Impl;
{ TProducer<T> }
procedure TProducer<T>.TSubscription.Unlink;
begin
if FTag <> nil then
begin
FOwner.Unlink(FTag);
FTag := nil;
end;
end;
class operator TProducer<T>.TSubscription.Implicit(A: TSubscription): TProducer<T>;
begin
Result := A.FOwner;
end;
constructor TProducer<T>.Create(const AProducer: IProducer<T>);
begin
FProducer := AProducer;
if not Assigned(FProducer) then
FProducer := Null;
end;
function TProducer<T>.Chain(const Next: IConsumer<T>): IConsumer<T>;
begin
FProducer.Link(Next);
Result := Next;
end;
function TProducer<T>.Chain<R>(const Next: IConverter<T, R>): TProducer<R>;
begin
FProducer.Link(Next.Consumer);
Result := Next;
end;
function TProducer<T>.Chain<R>(const Func: TConvertFunc<T, R>): TProducer<R>;
begin
Result := Chain<R>(TMycGenericConverter<T, R>.Create(Func));
end;
function TProducer<T>.CreateEndpoint(Lookback: Int64; out Series: TLazy<TSeries<T>>): TSubscription;
begin
Result := CreateLink(TMycDataEndpoint<T>.CreateDataEndpoint(Lookback, Series));
end;
function TProducer<T>.CreateSequence(Count: Integer; out Seq: TArray<IProducer<T>>): TSubscription;
begin
Result := CreateLink(TMycSequence<T>.CreateSequence(Count, Seq));
end;
function TProducer<T>.Field<R>(const FieldName: String): TProducer<R>;
begin
Result := Chain<R>(TMycRecordFieldReader<T, R>.Create(FieldName));
end;
class function TProducer<T>.GetNull: IProducer<T>;
begin
Result := TMycProducer<T>.Null;
end;
class operator TProducer<T>.Initialize(out Dest: TProducer<T>);
begin
Dest.FProducer := Null;
end;
class operator TProducer<T>.Implicit(const A: IProducer<T>): TProducer<T>;
begin
Result.Create(A);
end;
class operator TProducer<T>.Implicit(const A: TProducer<T>): IProducer<T>;
begin
Result := A.FProducer;
end;
function TProducer<T>.CreateLink(const Consumer: IConsumer<T>): TSubscription;
begin
Result.FOwner := FProducer;
Result.FTag := FProducer.Link(Consumer);
end;
function TProducer<T>.MakeParallel: TProducer<T>;
begin
Result := Chain<T>(TMycParallelConverter<T>.Create as IConverter<T, T>);
end;
constructor TConverter<S, T>.Create(const AConverter: IConverter<S, T>);
begin
FConverter := AConverter;
if not Assigned(FConverter) then
FConverter := Null;
end;
class function TConverter<S, T>.Construct(const Consumer: IConsumer<S>; const Producer: TProducer<T>): TConverter<S, T>;
begin
Result := TMycComposedConverter<S, T>.Create(Consumer, Producer);
end;
class function TConverter<S, T>.CreateAggregation(const Func: TAggregateFunc<S, T>): TConverter<S, T>;
begin
Result := TMycGenericAggregator<S, T>.Create(Func);
end;
class function TConverter<S, T>.CreateConverter(const Func: TConvertFunc<S, T>): TConverter<S, T>;
begin
Result := TMycGenericConverter<S, T>.Create(Func);
end;
function TConverter<S, T>.GetConsumer: IConsumer<S>;
begin
Result := FConverter.Consumer;
end;
function TConverter<S, T>.GetProducer: TProducer<T>;
begin
Result := FConverter;
end;
class function TConverter<S, T>.GetNull: IConverter<S, T>;
begin
Result := TMycConverter<S, T>.Null;
end;
class operator TConverter<S, T>.Initialize(out Dest: TConverter<S, T>);
begin
Dest.FConverter := Null;
end;
class operator TConverter<S, T>.Implicit(const A: IConverter<S, T>): TConverter<S, T>;
begin
Result.Create(A);
end;
class operator TConverter<S, T>.Implicit(const A: TConverter<S, T>): IConverter<S, T>;
begin
Result := A.FConverter;
end;
{ TConverter }
class function TConverter.CreateCounter<T>: TConverter<T, Int64>;
begin
Result := TMycDataCounter<T>.Create;
end;
class function TConverter.CreateEndpoint<T>(Lookback: Int64; out Series: TLazy<TSeries<T>>): IConsumer<T>;
begin
Result := TMycDataEndpoint<T>.CreateDataEndpoint(Lookback, Series);
end;
class function TConverter.CreateIdentity<T>: TConverter<T, T>;
begin
Result := TMycIdentityConverter<T>.Create;
end;
class function TConverter.CreateTicker<T>: TConverter<TArray<T>, T>;
begin
Result := TMycTicker<T>.Create;
end;
class function TConverter.Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>;
var
joiner: IDataJoin<T>;
begin
case Mode of
jmAll: joiner := TMycDataJoinAll<T>.Create(Length(Producers));
jmAny: joiner := TMycDataJoinAny<T>.Create(Length(Producers));
else
Assert(false);
end;
Result := joiner;
for var i := 0 to High(Producers) do
Producers[i].Chain(joiner.Consumers[i]);
end;
end.
Src/Data/Myc.Data.Series.pas
+410
View File
@@ -0,0 +1,410 @@
unit Myc.Data.Series;
interface
type
TChunkArray<T> = record
type
PT = ^T;
private
const
ChunkSize = 1024;
type
TChunk = array of T;
private
FChunks: TArray<TChunk>;
FOffset: Integer;
FCount: Integer;
function GetItems(Idx: Integer): T; inline;
function GetItemRef(Idx: Integer): PT; inline;
public
constructor Create(const AChunks: TArray<TChunk>; ACount: Integer; AOffset: Integer);
class operator Initialize(out Dest: TChunkArray<T>);
// Add a single item without creating a temporary array.
procedure Add(const Data: T; MaxCount: Integer); overload;
// Add items, but ensure that the result has MaxCount items.
procedure Add(const Data: array of T; MaxCount: Integer); overload;
class function CreateFromArray(const AData: TArray<T>; MaxCount: Integer): TChunkArray<T>; static;
// Creates a deep copy of the array, optionally truncating it to the last MaxCount items.
function Copy(MaxCount: Integer = -1): TChunkArray<T>;
property Count: Integer read FCount;
property Items[Idx: Integer]: T read GetItems; default;
property ItemRef[Idx: Integer]: PT read GetItemRef;
end;
// A series where the last added item has index 0.
TSeries<T> = record
private
FArray: TChunkArray<T>;
FTotalCount: Int64;
function GetCount: Integer;
function GetItems(Idx: Integer): T;
public
constructor Create(const AArray: TChunkArray<T>; ATotalCount: Int64);
class operator Initialize(out Dest: TSeries<T>);
// Creates a deep copy of the series, optionally truncating it to the last Lookback items.
function Copy(Lookback: Integer = -1): TSeries<T>;
procedure Add(const Data: T; Lookback: Int64 = -1); overload;
procedure Add(const Data: array of T; Lookback: Integer); overload;
class function CreateFromArray(const AData: TArray<T>; Lookback: Integer): TSeries<T>; static;
property Count: Integer read GetCount;
property TotalCount: Int64 read FTotalCount;
property Items[Idx: Integer]: T read GetItems; default;
end;
implementation
uses
System.Generics.Collections,
System.Math;
{ TChunkArray<T> }
constructor TChunkArray<T>.Create(const AChunks: TArray<TChunk>; ACount, AOffset: Integer);
begin
FChunks := AChunks;
FCount := ACount;
FOffset := AOffset;
Assert(FOffset < ChunkSize);
end;
procedure TChunkArray<T>.Add(const Data: T; MaxCount: Integer);
var
newCount, totalNewOffset: Integer;
oldLogicalEnd, newLogicalEnd, oldStartChunk, newStartChunk, oldEndChunk, newEndChunk, chunksToRemove, chunksToAdd: Integer;
begin
if MaxCount < 0 then
MaxCount := MaxInt;
if MaxCount <= 0 then
exit;
// 1. Calculate the new logical state (numToAdd is always 1)
newCount := Min(FCount + 1, MaxCount);
totalNewOffset := FOffset + FCount + 1 - newCount;
// 2. Determine if the underlying FChunks array needs restructuring
oldLogicalEnd := FOffset + FCount - 1;
newLogicalEnd := totalNewOffset + newCount - 1;
oldStartChunk := FOffset div ChunkSize;
newStartChunk := totalNewOffset div ChunkSize;
oldEndChunk := -1;
if FCount > 0 then
oldEndChunk := oldLogicalEnd div ChunkSize;
newEndChunk := -1;
if newCount > 0 then
newEndChunk := newLogicalEnd div ChunkSize;
chunksToRemove := newStartChunk - oldStartChunk;
chunksToAdd := newEndChunk - oldEndChunk;
// 3. Perform the add operation
if (chunksToRemove > 0) or (chunksToAdd > 0) then
begin
// --- Rebuild Path: Chunks must be added or removed ---
var newChunks: TArray<TChunk>;
var oldChunkCount := Length(FChunks);
var newChunkCount := oldChunkCount - chunksToRemove + chunksToAdd;
SetLength(newChunks, newChunkCount);
var numChunksToKeep := oldChunkCount - chunksToRemove;
if numChunksToKeep > 0 then
TArray.Copy<TChunk>(FChunks, newChunks, chunksToRemove, 0, numChunksToKeep);
for var i := numChunksToKeep to newChunkCount - 1 do
SetLength(newChunks[i], ChunkSize);
FChunks := newChunks;
// Copy the new data item into the rebuilt chunks
var writeStartLogicalIdx := FOffset + FCount;
var physicalWriteStartIdx := writeStartLogicalIdx - (chunksToRemove * ChunkSize);
var chunkIdx := physicalWriteStartIdx div ChunkSize;
var idxInChunk := physicalWriteStartIdx mod ChunkSize;
Assert(chunkIdx < Length(FChunks), 'Chunk index out of bounds during rebuild');
FChunks[chunkIdx][idxInChunk] := Data;
end
else
begin
// --- Fast Path: Window slides within existing chunk allocation ---
var writeStartLogicalIdx := FOffset + FCount;
var chunkIdx := writeStartLogicalIdx div ChunkSize;
var idxInChunk := writeStartLogicalIdx mod ChunkSize;
Assert(chunkIdx < Length(FChunks), 'Chunk index out of bounds on fast path');
if Length(FChunks[chunkIdx]) = 0 then // Safeguard
SetLength(FChunks[chunkIdx], ChunkSize);
FChunks[chunkIdx][idxInChunk] := Data;
end;
// 4. Finalize the new state
FOffset := totalNewOffset mod ChunkSize;
FCount := newCount;
end;
class function TChunkArray<T>.CreateFromArray(const AData: TArray<T>; MaxCount: Integer): TChunkArray<T>;
begin
Result.Add(AData, MaxCount);
end;
procedure TChunkArray<T>.Add(const Data: array of T; MaxCount: Integer);
var
dataLen, dataReadOffset, numToAdd, newCount, totalNewOffset: Integer;
oldLogicalEnd, newLogicalEnd, oldStartChunk, newStartChunk, oldEndChunk, newEndChunk, chunksToRemove, chunksToAdd: Integer;
begin
if MaxCount < 0 then
MaxCount := MaxInt;
dataLen := Length(Data);
if (dataLen = 0) or (MaxCount <= 0) then
exit;
// 1. Trim input data if it's larger than MaxCount
dataReadOffset := 0;
if dataLen > MaxCount then
dataReadOffset := dataLen - MaxCount;
numToAdd := dataLen - dataReadOffset;
// 2. Calculate the new logical state (total offset and count)
newCount := Min(FCount + numToAdd, MaxCount);
// Total logical offset of the new window's start
totalNewOffset := FOffset + FCount + numToAdd - newCount;
// 3. Determine if the underlying FChunks array needs restructuring
oldLogicalEnd := FOffset + FCount - 1;
newLogicalEnd := totalNewOffset + newCount - 1;
oldStartChunk := FOffset div ChunkSize;
newStartChunk := totalNewOffset div ChunkSize;
oldEndChunk := -1;
if FCount > 0 then
oldEndChunk := oldLogicalEnd div ChunkSize;
newEndChunk := -1;
if newCount > 0 then
newEndChunk := newLogicalEnd div ChunkSize;
chunksToRemove := newStartChunk - oldStartChunk;
chunksToAdd := newEndChunk - oldEndChunk;
// 4. Perform the add operation
if (chunksToRemove > 0) or (chunksToAdd > 0) then
begin
// --- Rebuild Path: Chunks must be added or removed ---
var newChunks: TArray<TChunk>;
var oldChunkCount := Length(FChunks);
var newChunkCount := oldChunkCount - chunksToRemove + chunksToAdd;
SetLength(newChunks, newChunkCount);
// Copy references to the chunks that are kept
var numChunksToKeep := oldChunkCount - chunksToRemove;
if numChunksToKeep > 0 then
TArray.Copy<TChunk>(FChunks, newChunks, chunksToRemove, 0, numChunksToKeep);
// Allocate new chunks at the end
for var i := numChunksToKeep to newChunkCount - 1 do
SetLength(newChunks[i], ChunkSize);
FChunks := newChunks;
// Copy the new data into the rebuilt chunks
var writeStartLogicalIdx := FOffset + FCount;
var physicalWriteStartIdx := writeStartLogicalIdx - (chunksToRemove * ChunkSize);
var remainingToAdd := numToAdd;
var currentDataOffset := dataReadOffset;
while remainingToAdd > 0 do
begin
var chunkIdx := physicalWriteStartIdx div ChunkSize;
var idxInChunk := physicalWriteStartIdx mod ChunkSize;
var spaceInChunk := ChunkSize - idxInChunk;
var countToCopy := Min(remainingToAdd, spaceInChunk);
Assert(chunkIdx < Length(FChunks), 'Chunk index out of bounds during rebuild');
TArray.Copy<T>(Data, FChunks[chunkIdx], currentDataOffset, idxInChunk, countToCopy);
inc(physicalWriteStartIdx, countToCopy);
inc(currentDataOffset, countToCopy);
dec(remainingToAdd, countToCopy);
end;
end
else
begin
// --- Fast Path: Window slides within existing chunk allocation ---
var writeStartLogicalIdx := FOffset + FCount;
var remainingToAdd := numToAdd;
var currentDataOffset := dataReadOffset;
while remainingToAdd > 0 do
begin
var chunkIdx := writeStartLogicalIdx div ChunkSize;
var idxInChunk := writeStartLogicalIdx mod ChunkSize;
var spaceInChunk := ChunkSize - idxInChunk;
var countToCopy := Min(remainingToAdd, spaceInChunk);
Assert(chunkIdx < Length(FChunks), 'Chunk index out of bounds on fast path');
if Length(FChunks[chunkIdx]) = 0 then // Should not happen on fast path, but as a safeguard
SetLength(FChunks[chunkIdx], ChunkSize);
TArray.Copy<T>(Data, FChunks[chunkIdx], currentDataOffset, idxInChunk, countToCopy);
inc(writeStartLogicalIdx, countToCopy);
inc(currentDataOffset, countToCopy);
dec(remainingToAdd, countToCopy);
end;
end;
// 5. Finalize the new state
FOffset := totalNewOffset mod ChunkSize;
FCount := newCount;
end;
function TChunkArray<T>.GetItems(Idx: Integer): T;
begin
Result := GetItemRef(Idx)^;
end;
function TChunkArray<T>.Copy(MaxCount: Integer): TChunkArray<T>;
var
effectiveCount: Integer;
logicalStartIndex: Integer;
physicalStartIndex, physicalEndIndex: Integer;
startChunkIdx, endChunkIdx: Integer;
newOffset: Integer;
newChunkCount: Integer;
i: Integer;
begin
// 1. Determine the number of items to copy.
effectiveCount := FCount;
if (MaxCount >= 0) and (MaxCount < FCount) then
effectiveCount := MaxCount;
if effectiveCount = 0 then
begin
// Return an empty array.
Result := Default(TChunkArray<T>);
exit;
end;
// 2. Calculate physical location of the data to be copied.
logicalStartIndex := FCount - effectiveCount;
physicalStartIndex := FOffset + logicalStartIndex;
physicalEndIndex := physicalStartIndex + effectiveCount - 1;
startChunkIdx := physicalStartIndex div ChunkSize;
endChunkIdx := physicalEndIndex div ChunkSize;
newOffset := physicalStartIndex mod ChunkSize;
// 3. Create the new chunk array for the result.
newChunkCount := endChunkIdx - startChunkIdx + 1;
SetLength(Result.FChunks, newChunkCount);
// 4. Copy chunks. Boundary chunks are deep-copied, internal chunks are shared.
if newChunkCount = 1 then
begin
// All data is in a single chunk, which is both a start and end boundary.
// It must always be a deep copy.
Result.FChunks[0] := System.Copy(FChunks[startChunkIdx]);
end
else
begin
// First chunk (start boundary) must be a deep copy.
Result.FChunks[0] := System.Copy(FChunks[startChunkIdx]);
// Middle, immutable chunks can be shared by reference.
// This loop only runs if there are 3 or more chunks in the copy.
for i := 1 to newChunkCount - 2 do
begin
Result.FChunks[i] := FChunks[startChunkIdx + i];
end;
// Last chunk (end boundary) must be a deep copy.
Result.FChunks[newChunkCount - 1] := System.Copy(FChunks[endChunkIdx]);
end;
// 5. Finalize the new TChunkArray state.
Result.FOffset := newOffset;
Result.FCount := effectiveCount;
end;
function TChunkArray<T>.GetItemRef(Idx: Integer): PT;
begin
Assert((Idx >= 0) and (Idx < FCount));
var physicalIdx := Idx + FOffset;
Result := @FChunks[physicalIdx div ChunkSize][physicalIdx mod ChunkSize];
end;
class operator TChunkArray<T>.Initialize(out Dest: TChunkArray<T>);
begin
Dest.FOffset := 0;
Dest.FCount := 0;
end;
{ TSeries<T> }
constructor TSeries<T>.Create(const AArray: TChunkArray<T>; ATotalCount: Int64);
begin
FArray := AArray;
FTotalCount := ATotalCount;
end;
procedure TSeries<T>.Add(const Data: T; Lookback: Int64);
begin
FArray.Add(Data, Lookback);
inc(FTotalCount);
end;
procedure TSeries<T>.Add(const Data: array of T; Lookback: Integer);
begin
FArray.Add(Data, Lookback);
inc(FTotalCount, Length(Data));
end;
function TSeries<T>.Copy(Lookback: Integer): TSeries<T>;
begin
// Create a copy of the underlying chunk array, optionally truncating it.
Result.FArray := FArray.Copy(Lookback);
// The new series' total count is the number of items it actually contains.
Result.FTotalCount := Result.FArray.Count;
end;
class function TSeries<T>.CreateFromArray(const AData: TArray<T>; Lookback: Integer): TSeries<T>;
begin
Result.FArray.Add(AData, Lookback);
Result.FTotalCount := Length(AData);
end;
function TSeries<T>.GetCount: Integer;
begin
Result := FArray.Count;
end;
function TSeries<T>.GetItems(Idx: Integer): T;
begin
// Access is reversed: series index 0 is the last element in the underlying array
Assert((Idx >= 0) and (Idx < FArray.Count), 'Index out of bounds');
Result := FArray.Items[FArray.Count - 1 - Idx];
end;
class operator TSeries<T>.Initialize(out Dest: TSeries<T>);
begin
Dest.FTotalCount := 0;
end;
end.