From 2c56f6e75055105e30c0085d38b390672d00c72e Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Sun, 15 Jun 2025 17:29:56 +0200 Subject: [PATCH] Data file handling --- Src/Myc.Core.FileCache.pas | 180 +++++++++ Src/Myc.Core.Notifier.pas | 2 +- Src/Myc.Core.Signals.pas | 8 +- Src/Myc.Signals.FMX.pas | 63 +++ Src/Myc.Signals.pas | 30 +- Src/Myc.Test.Trade.DataStream.pas | 5 +- Src/Myc.Trade.DataStream.pas | 647 ++++++++++++++---------------- 7 files changed, 578 insertions(+), 357 deletions(-) create mode 100644 Src/Myc.Core.FileCache.pas create mode 100644 Src/Myc.Signals.FMX.pas diff --git a/Src/Myc.Core.FileCache.pas b/Src/Myc.Core.FileCache.pas new file mode 100644 index 0000000..9298a47 --- /dev/null +++ b/Src/Myc.Core.FileCache.pas @@ -0,0 +1,180 @@ +unit Myc.Core.FileCache; + +interface + +uses + System.SysUtils, + System.IOUtils, + System.Generics.Collections, + System.Generics.Defaults, + System.SyncObjs, + Myc.Futures; + +type + // Manages a cache for time-series data files loaded asynchronously. + // The cache is activated by assigning a function to the IsMemoryLow property. + // It evicts least-recently-used items when memory is low. + TDataFileCache = class + type + TLoaderFunc = reference to function(const Filename: String): TFuture; + private + type + // Holds metadata for a cached file. + TCachedFile = class + public + Name: String; + Age: TDateTime; + LastUsed: TDateTime; + Data: TFuture; + end; + private + FCachedFiles: TDictionary; + FLoader: TLoaderFunc; + FLock: TCriticalSection; + public + constructor Create(const ALoader: TLoaderFunc); + destructor Destroy; override; + + // Removes all items from the cache, waiting for any pending operations to complete. + procedure Clear; + + // Retrieves a data future from the cache or uses the provided loader function to create and cache it. + // Caching is only active if the IsMemoryLow property is assigned. + function GetOrAdd(const AFileName: string): TFuture; + end; + +var + IsMemoryLow: TFunc; + +implementation + +uses + {$IFDEF MSWINDOWS} Winapi.Windows, {$endif} + System.Math; + +{ TDataFileCache } + +constructor TDataFileCache.Create(const ALoader: TLoaderFunc); +begin + inherited Create; + FLoader := ALoader; + FLock := TCriticalSection.Create; + FCachedFiles := TObjectDictionary.Create([doOwnsValues]); +end; + +destructor TDataFileCache.Destroy; +begin + Clear; + FCachedFiles.Free; + FLock.Free; + inherited; +end; + +procedure TDataFileCache.Clear; +begin + FLock.Enter; + try + // Wait for any pending load operations to finish before clearing. + for var cachedFile in FCachedFiles.Values do + cachedFile.Data.WaitFor; + FCachedFiles.Clear; + finally + FLock.Leave; + end; +end; + +function TDataFileCache.GetOrAdd(const AFileName: string): TFuture; +var + fileList: TArray; + i: Integer; + cachedFile: TCachedFile; + currentFileAge: TDateTime; +begin + // If no memory check function is assigned, caching is disabled. + // Simply execute the loader function and return the result. + if not Assigned(IsMemoryLow) then + begin + Result := FLoader(AFilename); + exit; + end; + + FLock.Enter; + try + // Caching is active. First, try to free memory if needed. + fileList := FCachedFiles.Values.ToArray; + TArray.Sort( + fileList, + TComparer.Construct( + function(const Left, Right: TCachedFile): Integer + begin + // Sort by LastUsed timestamp, oldest first. + Result := Sign(Left.LastUsed - Right.LastUsed); + end + ) + ); + + for i := 0 to High(fileList) do + begin + if not IsMemoryLow() then + break; // Stop evicting if memory pressure is relieved. + + if fileList[i].Name <> AFileName then + FCachedFiles.Remove(fileList[i].Name); + end; + + // Check if a valid entry exists in the cache. + if FCachedFiles.TryGetValue(AFileName, cachedFile) then + begin + if FileAge(AFileName, currentFileAge, true) and (currentFileAge = cachedFile.Age) then + begin + // Cache hit and file is unchanged. Update usage timestamp and return cached future. + cachedFile.LastUsed := Now; + exit(cachedFile.Data); + end + else + begin + // File is stale or no longer exists. Remove from cache. + FCachedFiles.Remove(AFileName); + end; + end; + + // Cache miss or stale entry. Execute the loader function. + Result := FLoader(AFilename); + + // Create a new cache entry for the loaded future. + cachedFile := TCachedFile.Create; + cachedFile.Name := AFileName; + if FileAge(AFileName, cachedFile.Age, true) then + begin + cachedFile.Data := Result; + cachedFile.LastUsed := Now; + FCachedFiles.Add(AFileName, cachedFile); + end + else + begin + // If we cannot get the file's age, we cannot cache it reliably. + cachedFile.Free; + end; + finally + FLock.Leave; + end; +end; + +initialization + {$IFDEF MSWINDOWS} + IsMemoryLow := + function: Boolean + var + memStatusEx: TMemoryStatusEx; + begin + Result := false; + memStatusEx.dwLength := SizeOf(TMemoryStatusEx); // Initialize the structure size + if GlobalMemoryStatusEx(memStatusEx) then + begin + // memStatusEx.dwMemoryLoad indicates the approximate percentage of physical memory that is in use. + Result := memStatusEx.dwMemoryLoad >= 80; + end; + end; + {$endif} + +end. diff --git a/Src/Myc.Core.Notifier.pas b/Src/Myc.Core.Notifier.pas index 9b6408a..d4e83fc 100644 --- a/Src/Myc.Core.Notifier.pas +++ b/Src/Myc.Core.Notifier.pas @@ -59,7 +59,7 @@ procedure TMycNotifyList.Destroy; begin // Because refcounting is thread-safe, this will always be entered once after all references // to Self are dropped. No locking needed! - Assert(not IsLocked); + // Assert(not IsLocked); FFirst := FFirst and not 3; UnadviseAll; end; diff --git a/Src/Myc.Core.Signals.pas b/Src/Myc.Core.Signals.pas index 2ce13ea..480b9d6 100644 --- a/Src/Myc.Core.Signals.pas +++ b/Src/Myc.Core.Signals.pas @@ -25,7 +25,7 @@ type end; // A state that acts as an event. It's state ist always set and it will always Trigger it's subscribers, if it gets notified by itself. - TMycEvent = class(TInterfacedObject, TSignal.ISignal, TEvent.IEvent) + TMycEvent = class(TInterfacedObject, TSignal.ISubscriber, TSignal.ISignal, TEvent.IEvent) strict private FSubscribers: TMycNotifyList; protected @@ -49,7 +49,7 @@ type // It is initialized with a count. Calls to its Notify method (as an TSignal.ISubscriber) // decrement this count. When the count reaches zero, the latch transitions to the "set" // state (IsSet becomes true), notifies its current subscribers once, and then remains set. - TMycLatch = class(TInterfacedObject, TState.IState, TLatch.ILatch) + TMycLatch = class(TInterfacedObject, TSignal.ISubscriber, TSignal.ISignal, TState.IState, TLatch.ILatch) strict private FSubscribers: TMycNotifyList; // List of subscribers waiting for this latch to be set. private @@ -74,7 +74,7 @@ type function Notify: Boolean; end; - TMycNullLatch = class(TInterfacedObject, TLatch.ILatch) + TMycNullLatch = class(TInterfacedObject, TSignal.ISubscriber, TLatch.ILatch) private function GetState: TState; function Notify: Boolean; @@ -83,7 +83,7 @@ type // TMycFlag implements a resettable "dirty flag". // It can be explicitly set to dirty (via Notify) or reset to clean (via Reset). // Subscribers are notified of relevant state changes. - TMycFlag = class(TInterfacedObject, TState.IState, TFlag.IFlag) + TMycFlag = class(TInterfacedObject, TSignal.ISubscriber, TSignal.ISignal, TState.IState, TFlag.IFlag) strict private FSubscribers: TMycNotifyList; // List of subscribers interested in state changes of this dirty flag. private diff --git a/Src/Myc.Signals.FMX.pas b/Src/Myc.Signals.FMX.pas new file mode 100644 index 0000000..3cf79d8 --- /dev/null +++ b/Src/Myc.Signals.FMX.pas @@ -0,0 +1,63 @@ +unit Myc.Signals.FMX; + +interface + +uses + System.Classes, + FMX.Controls, + Myc.Signals; + +type + TFMXSignalLink = class(TComponent, TSignal.ISubscriber) + private + [volatile] + FInvalidated: Integer; + procedure InvalidateControl; + + protected + function Notify: Boolean; + + public + constructor Create(AOwner: TControl); reintroduce; + destructor Destroy; override; + function Reset: Boolean; + end; + +implementation + +uses + System.SyncObjs; + +{ TFMXSignalLink } + +constructor TFMXSignalLink.Create(AOwner: TControl); +begin + inherited Create(AOwner); + FInvalidated := 0; +end; + +destructor TFMXSignalLink.Destroy; +begin + TThread.RemoveQueuedEvents(InvalidateControl); + inherited; +end; + +procedure TFMXSignalLink.InvalidateControl; +begin + with TControl(Owner) do + InvalidateRect(LocalRect); +end; + +function TFMXSignalLink.Notify: Boolean; +begin + if TInterlocked.Exchange(FInvalidated, 1) = 0 then + TThread.Queue(nil, InvalidateControl); + exit( true ); +end; + +function TFMXSignalLink.Reset: Boolean; +begin + Result := TInterlocked.Exchange(FInvalidated, 0) > 0; +end; + +end. diff --git a/Src/Myc.Signals.pas b/Src/Myc.Signals.pas index 2fb0bd1..8303edf 100644 --- a/Src/Myc.Signals.pas +++ b/Src/Myc.Signals.pas @@ -3,7 +3,8 @@ unit Myc.Signals; interface uses - System.SysUtils; + System.SysUtils, + System.SyncObjs; type TSignal = record @@ -136,6 +137,8 @@ type strict private class var FNull: ILatch; + FQueueGate: TLatch; + FQueueLock: TSpinLock; class constructor ClassCreate; private @@ -150,7 +153,8 @@ type class function CreateLatch(Count: Integer): TLatch; static; - class function Enqueue(var Gate: TLatch; Count: Integer = 1): TState; static; + class function Enqueue(var Gate: TLatch): TState; overload; static; + class function Enqueue: TState; overload; static; class property Null: ILatch read FNull; @@ -390,6 +394,7 @@ class constructor TLatch.ClassCreate; begin // Create a singleton null latch instance that is initially (and always) set. FNull := TMycNullLatch.Create; + FQueueLock := TSpinLock.Create(false); end; constructor TLatch.Create(const ALatch: TLatch.ILatch); @@ -407,12 +412,23 @@ begin Result := FNull; end; -class function TLatch.Enqueue(var Gate: TLatch; Count: Integer = 1): TState; +class function TLatch.Enqueue(var Gate: TLatch): TState; begin - var gateState := Gate.State; - Gate := TMycLatch.Create(Count); - gateState.Subscribe(Gate); - exit(Gate.State); + var Latch: TLatch.ILatch := TMycLatch.Create(1); + + FQueueLock.Enter; + try + Gate.State.Subscribe(Latch); + Gate := Latch; + Result := Latch.State; + finally + FQueueLock.Exit; + end; +end; + +class function TLatch.Enqueue: TState; +begin + Result := Enqueue(FQueueGate); end; function TLatch.GetState: TState; diff --git a/Src/Myc.Test.Trade.DataStream.pas b/Src/Myc.Test.Trade.DataStream.pas index f9348e2..2c67b9c 100644 --- a/Src/Myc.Test.Trade.DataStream.pas +++ b/Src/Myc.Test.Trade.DataStream.pas @@ -59,9 +59,6 @@ end; procedure TTest_TABFileServer_Equivalence.TearDownFixture; begin FServer := nil; - - // Resets the global IsMemoryLow function pointer, if it was set for testing. - IsMemoryLow := nil; end; procedure TTest_TABFileServer_Equivalence.Setup; @@ -90,7 +87,7 @@ var expectedData: TArray>; begin // 1. Expected data is loaded directly using LoadDataSeries. - filename := TAuraTABFileServer.FindFirstDataFile(C_TEST_PATH, C_TEST_SYMBOL); + filename := (FServer as TAuraDataServer).FindFirstDataFile(C_TEST_SYMBOL); Assert.IsNotEmpty(filename, 'Test data file could not be found.'); expectedData := FServer.LoadDataSeries(filename).WaitFor; diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas index 3602f84..5791c17 100644 --- a/Src/Myc.Trade.DataStream.pas +++ b/Src/Myc.Trade.DataStream.pas @@ -26,10 +26,10 @@ uses Myc.Futures, Myc.Signals, Myc.Lazy, - Myc.Trade.DataPoint; + Myc.Trade.DataPoint, + Myc.Core.FileCache; type - // Represents a generic data stream capable of providing sequential data chunks. // IsHistory: // - true, if this stream is a history stream. Once HasData becomes false, it reached it's end and will not provide more data. @@ -39,13 +39,16 @@ type ['{A6E246A2-E84E-49AB-A63E-333E561E488C}'] function GetHasData: TSignal; function GetChunk(var Data: array of TDataPoint): Integer; + function GetSymbol: String; function IsHistory: Boolean; property HasData: TSignal read GetHasData; + property Symbol: String read GetSymbol; end; // Abstract base class for IDataStream implementations. TDataStream = class(TInterfacedObject, IDataStream) protected + function GetSymbol: String; virtual; abstract; function GetHasData: TSignal; virtual; abstract; function GetChunk(var Data: array of TDataPoint): Integer; virtual; abstract; function IsHistory: Boolean; virtual; abstract; @@ -79,31 +82,31 @@ type IAuraDataServer = interface(IDataServer) ['{481E27DE-DC58-49AF-B8A8-B6316980C423}'] function LoadDataFile(const FileName: string): TFuture>>; - function LoadDataSeries(const FileName: string): TFuture>>; - function EnumerateAssetFiles: TArray; + function EnumerateAssets: TArray; end; // Aura files // Generic server for loading and managing sequential time-series data from files. TAuraDataServer = class(TInterfacedObject, IDataServer, IAuraDataServer) + public type + TDataFile = record + Extension: String; + Path, Symbol: String; + Year, Month: Integer; + function GetBaseFileName: string; + function GetFullFileName: string; + function GetIsValid: Boolean; + property IsValid: Boolean read GetIsValid; + end; + private - type - TCachedFile = class - public - Name: String; - Age: TDateTime; - LastUsed: TDateTime; - Data: TFuture>>; - end; - private - // The cache is per-instance. - FCachedFiles: TDictionary; + FCachedFiles: TDataFileCache>>; FPath: String; + function DoLoad(const FileName: string): TFuture>>; function GetPath: String; strict private - // The load gate is shared across all instances of TAuraDataServer. class var FLoadGate: TLatch; @@ -115,65 +118,58 @@ type constructor Create(const APath: String); destructor Destroy; override; - class function TryParseFileName( - const FileName: string; - out PathValue, SymbolValue: string; - out YearValue, MonthValue: Integer - ): Boolean; virtual; abstract; + class function TryParseFileName(const FileName: string; out DataFile: TDataFile): Boolean; virtual; abstract; - class function FindFirstDataFile(const Path, Symbol: string): string; - class function FindNextDataFile(const FileName: string): string; - class function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; + function FindFirstDataFile(const Symbol: string): TDataFile; + class function FindNextDataFile(const CurrentFile: TDataFile): TDataFile; + class function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; - // IDataServer implementation function CreateStream(const Symbol: String): IDataStream; virtual; abstract; procedure ClearCache; - function EnumerateAssetFiles: TArray; - + function EnumerateAssets: TArray; function LoadDataFile(const FileName: string): TFuture>>; - function LoadDataSeries(const FileName: string): TFuture>>; - property Path: String read GetPath; end; // Implements a data stream that reads from Aura-specific historical data files. - TAuraFileStream = class(TDataStream) + TAuraFileStream = class(TDataStream, IDataStream) + private + type + // Changed: Renamed "File" to "FileInfo" for clarity. + TDataStream = record + FileInfo: TAuraDataServer.TDataFile; + Data: TFuture>>; + end; private FDataServer: TAuraDataServer; + FSymbol: String; FHasData: TEvent; - FCurrentFileName: string; - FCurrentData: TFuture>>; - FNextFileName: string; - FNextData: TFuture>>; + FCurrent: TFuture; + FNext: TFuture; FCurrPosInFile: Int64; FLastTimeStamp: TDateTime; + procedure PreloadNext; protected + function GetSymbol: String; override; function GetHasData: TSignal; override; function GetChunk(var Data: array of TDataPoint): Integer; override; function IsHistory: Boolean; override; public - constructor Create(ADataServer: TAuraDataServer; const AFilename: String); + constructor Create(ADataServer: TAuraDataServer; const ASymbol: String); destructor Destroy; override; procedure AfterConstruction; override; + procedure BeforeDestruction; override; end; // Aura tick data file Ask-Bid TAuraTABFileServer = class(TAuraDataServer) public - // IDataServer implementation function CreateStream(const Symbol: String): IDataStream; override; - - class function TryParseFileName( - const FileName: string; - out PathValue, SymbolValue: string; - out YearValue, MonthValue: Integer - ): Boolean; override; + function LoadDataSeries(const FileName: string): TFuture>>; + class function TryParseFileName(const FileName: string; out DataFile: TAuraDataServer.TDataFile): Boolean; override; end; -var - IsMemoryLow: TFunc; - implementation uses @@ -182,12 +178,33 @@ uses System.StrUtils, Myc.TaskManager; +{ TAuraDataServer.TDataFile } + +function TAuraDataServer.TDataFile.GetBaseFileName: string; +begin + Result := Format('%s_%.4d_%.2d', [Symbol, Year, Month]); +end; + +function TAuraDataServer.TDataFile.GetFullFileName: string; +begin + Result := TPath.Combine(Path, GetBaseFileName + Extension); +end; + +function TAuraDataServer.TDataFile.GetIsValid: Boolean; +begin + Result := (Symbol <> '') and (Year > 0); +end; + +{ TAuraDataServer } + constructor TAuraDataServer.Create(const APath: String); begin inherited Create; - // Initialize the per-instance cache. - FCachedFiles := TObjectDictionary.Create([doOwnsValues]); FPath := APath; + + FCachedFiles := + TDataFileCache>> + .Create(function(const Filename: String): TFuture>> begin Result := DoLoad(Filename); end); end; destructor TAuraDataServer.Destroy; @@ -199,78 +216,59 @@ end; procedure TAuraDataServer.ClearCache; begin - for var cF in FCachedFiles.Values do - cF.Data.WaitFor; - FCachedFiles.Clear; end; -class function TAuraDataServer.FindFirstDataFile(const Path, Symbol: string): string; +function TAuraDataServer.FindFirstDataFile(const Symbol: string): TDataFile; var - oldestFiles: TArray; - fileName: string; - pathValue, symbolValue: string; - yearValue, monthValue: Integer; + oldestFiles: TArray; + fileInfo: TDataFile; begin - oldestFiles := FindOldestFilesPerSymbol(Path); - for fileName in oldestFiles do + oldestFiles := FindOldestFilesPerSymbol(FPath); + for fileInfo in oldestFiles do begin - if TryParseFileName(fileName, pathValue, symbolValue, yearValue, monthValue) then - begin - if SameText(symbolValue, Symbol) then - exit(fileName); - end; + if SameText(fileInfo.Symbol, Symbol) then + exit(fileInfo); end; - Result := ''; + Result := Default(TDataFile); end; -class function TAuraDataServer.FindNextDataFile(const FileName: string): string; +class function TAuraDataServer.FindNextDataFile(const CurrentFile: TDataFile): TDataFile; var - pathValue, symbolValue: string; - yearValue, monthValue: Integer; - nextBaseName, compressedPath, uncompressedPath: string; + dataFileToProbe: TDataFile; begin - if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then - exit(''); + Result := Default(TDataFile); - Inc(monthValue); - if monthValue > 12 then + if not CurrentFile.IsValid then + exit; + + // Changed: Simplified logic, no reparsing needed. + dataFileToProbe := CurrentFile; + Inc(dataFileToProbe.Month); + if (dataFileToProbe.Month > 12) then begin - monthValue := 1; - Inc(yearValue); + dataFileToProbe.Month := 1; + Inc(dataFileToProbe.Year); end; - nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]); + dataFileToProbe.Extension := '.tab_zip'; + if TFile.Exists(dataFileToProbe.GetFullFileName) then + exit(dataFileToProbe); - compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip'); - if TFile.Exists(compressedPath) then - exit(compressedPath); - - uncompressedPath := TPath.Combine(pathValue, nextBaseName); - if TFile.Exists(uncompressedPath) then - exit(uncompressedPath); - - Result := ''; + dataFileToProbe.Extension := '.tab'; + if TFile.Exists(dataFileToProbe.GetFullFileName) then + exit(dataFileToProbe); end; -class function TAuraDataServer.FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; -type - TTrackedFileInfo = record - FullFileName: string; - Year: Integer; - Month: Integer; - end; +class function TAuraDataServer.FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; var - oldestFilesPerSymbol: TDictionary; + oldestFilesPerSymbol: TDictionary; fileNames: TArray; currentFile: string; - parsedPath: string; - parsedSymbol: string; - parsedYear: Integer; - parsedMonth: Integer; - trackedInfo: TTrackedFileInfo; + dataFile: TDataFile; + trackedInfo: TDataFile; begin - oldestFilesPerSymbol := TDictionary.Create; + oldestFilesPerSymbol := TDictionary.Create; try if not TDirectory.Exists(DirectoryPath) then exit(nil); @@ -278,43 +276,37 @@ begin fileNames := TDirectory.GetFiles(DirectoryPath); for currentFile in fileNames do begin - if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then + if TryParseFileName(currentFile, dataFile) then begin - if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then + if oldestFilesPerSymbol.TryGetValue(dataFile.Symbol, trackedInfo) then begin - if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then + if (dataFile.Year < trackedInfo.Year) or ((dataFile.Year = trackedInfo.Year) and (dataFile.Month < trackedInfo.Month)) then begin - trackedInfo.FullFileName := currentFile; - trackedInfo.Year := parsedYear; - trackedInfo.Month := parsedMonth; - oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo); + oldestFilesPerSymbol.AddOrSetValue(dataFile.Symbol, dataFile); end; end else begin - trackedInfo.FullFileName := currentFile; - trackedInfo.Year := parsedYear; - trackedInfo.Month := parsedMonth; - oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo); + oldestFilesPerSymbol.Add(dataFile.Symbol, dataFile); end; end; end; - SetLength(Result, oldestFilesPerSymbol.Count); - var n := 0; - for trackedInfo in oldestFilesPerSymbol.Values do - begin - Result[n] := trackedInfo.FullFileName; - inc(n); - end; + Result := oldestFilesPerSymbol.Values.ToArray; finally oldestFilesPerSymbol.Free; end; end; -function TAuraDataServer.EnumerateAssetFiles: TArray; +function TAuraDataServer.EnumerateAssets: TArray; +var + dataFiles: TArray; + i: Integer; begin - Result := FindOldestFilesPerSymbol(FPath); + dataFiles := FindOldestFilesPerSymbol(FPath); + SetLength(Result, Length(dataFiles)); + for i := 0 to High(dataFiles) do + Result[i] := dataFiles[i].Symbol; end; function TAuraDataServer.GetPath: String; @@ -322,59 +314,20 @@ begin Result := FPath; end; -function TAuraDataServer.LoadDataFile(const FileName: string): TFuture>>; -var - parsedPath, parsedSymbol: string; - parsedYear, parsedMonth: Integer; +function TAuraDataServer.DoLoad(const FileName: string): TFuture>>; begin Result := TFuture>>.Null; - if not TryParseFileName(FileName, parsedPath, parsedSymbol, parsedYear, parsedMonth) then + if not TFile.Exists(FileName) then exit; - if Assigned(IsMemoryLow) then - begin - if IsMemoryLow() then - begin - var fL := FCachedFiles.Values.ToArray; - TArray.Sort( - fL, - TComparer - .Construct(function(const Left, Right: TCachedFile): Integer begin Result := Sign(Left.LastUsed - Right.LastUsed); end) - ); - for var i := 0 to High(fL) do - begin - if not IsMemoryLow() then - break; - if fL[i].Name <> FileName then - FCachedFiles.Remove(fL[i].Name); - end; - end; - - var cachedFile: TCachedFile; - if FCachedFiles.TryGetValue(Filename, cachedFile) then - begin - var age: TDateTime; - if FileAge(Filename, age, true) then - begin - if age = cachedFile.Age then - begin - cachedFile.LastUsed := Now; - exit(cachedFile.Data); - end - else - FCachedFiles.Remove(FileName); - end; - end; - end; - var capFileName := FileName; if FileName.EndsWith('_zip', True) then begin Result := TFuture .Construct( - TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate + TLatch.Enqueue(FLoadGate), function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end) .Chain>>( function(bytes: TBytes): TArray> @@ -388,11 +341,11 @@ begin end; end); end - else if TFile.Exists(FileName) then + else begin Result := TFuture>>.Construct( - TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate + TLatch.Enqueue(FLoadGate), function: TArray> begin if TFile.Exists(capFileName) then @@ -412,108 +365,11 @@ begin end ); end; - - if Assigned(IsMemoryLow) then - begin - var cachedFile := TCachedFile.Create; - cachedFile.Name := Filename; - if FileAge(Filename, cachedFile.Age, true) then - begin - cachedFile.Data := Result; - cachedFile.LastUsed := Now; - FCachedFiles.Add(Filename, cachedFile); - end; - end; end; -function TAuraDataServer.LoadDataSeries(const FileName: string): TFuture>>; -var - loadedState: TState; - loadedFiles: TArray>>>; - liveData: TFuture>>; - tabFiles: TList; - currentFile: string; - liveFilePath: string; - pathName, symbolName: string; - yearValue, monthValue: Integer; +function TAuraDataServer.LoadDataFile(const FileName: string): TFuture>>; begin - tabFiles := TList.Create; - try - currentFile := FileName; - if TFile.Exists(currentFile) then - while currentFile <> '' do - begin - tabFiles.Add(currentFile); - currentFile := FindNextDataFile(currentFile); - end; - - liveFilePath := ''; - if tabFiles.Count > 0 then - begin - var lastTabFile := tabFiles.Last; - if TryParseFileName(lastTabFile, pathName, symbolName, yearValue, monthValue) then - begin - var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, yearValue, monthValue]); - var potentialLivePath := TPath.Combine(pathName, liveFileBaseName); - if TFile.Exists(potentialLivePath) then - liveFilePath := potentialLivePath; - end; - end; - - var loadedFileList := TList>>>.Create; - var loadStates := TList.Create; - try - for currentFile in tabFiles do - begin - var data := LoadDataFile(currentFile); - loadedFileList.Add(data); - loadStates.Add(data.Done); - end; - - liveData := TFuture>>.Null; - if liveFilePath <> '' then - begin - liveData := LoadDataFile(liveFilePath); - loadedFileList.Add(liveData); - loadStates.Add(liveData.Done); - end; - - loadedState := TState.All(loadStates.ToArray); - loadedFiles := loadedFileList.ToArray; - finally - loadStates.Free; - loadedFileList.Free; - end; - finally - tabFiles.Free; - end; - - Result := - TFuture>>.Construct( - loadedState, - function: TArray> - begin - var tickList := TList>.Create; - try - var overallLastTabTickTime: TDateTime := 0; - var cnt := 0; - for var P in loadedFiles do - inc(cnt, Length(P.Value)); - tickList.Capacity := cnt; - - for var P in loadedFiles do - for var iterTickData in P.Value do - if overallLastTabTickTime < iterTickData.Time then - begin - tickList.Add(iterTickData); - overallLastTabTickTime := iterTickData.Time; - end; - Result := tickList.ToArray; - finally - tickList.Free; - end; - end - ); + Result := FCachedFiles.GetOrAdd(Filename); end; class function TAuraDataServer.ReadCompressedData(const InputStream: TStream): TArray>; @@ -583,14 +439,13 @@ end; { TAuraFileStream } -constructor TAuraFileStream.Create(ADataServer: TAuraDataServer; const AFilename: String); +constructor TAuraFileStream.Create(ADataServer: TAuraDataServer; const ASymbol: String); begin inherited Create; Assert(Assigned(ADataServer)); FDataServer := ADataServer; - FCurrentFileName := AFilename; - // Create an event - FHasData := TEvent.CreateEvent; // interface helper benutzen! + FSymbol := ASymbol; + FHasData := TEvent.CreateEvent; end; destructor TAuraFileStream.Destroy; @@ -599,56 +454,69 @@ begin end; procedure TAuraFileStream.AfterConstruction; +var + firstFile: TFuture.TDataFile>; begin inherited; + + firstFile := TFuture.TDataFile>.Construct(function: TAuraDataServer.TDataFile begin Result := FDataServer.FindFirstDataFile(FSymbol); end); + + FCurrent := + firstFile.Chain( + function(const FileInfo: TAuraDataServer.TDataFile): TDataStream + begin + if FileInfo.IsValid then + begin + // Changed: Use renamed field "FileInfo". + Result.FileInfo := FileInfo; + Result.Data := FDataServer.LoadDataFile(FileInfo.GetFullFileName); + Result.Data.Done.Subscribe(FHasData); + end; + end + ); + FLastTimeStamp := 0; FCurrPosInFile := 0; - FCurrentData := FDataServer.LoadDataFile(FCurrentFileName); - // Forward all future done events, because this means there is new data. - FCurrentData.Done.Subscribe(FHasData); - FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName); - if FNextFileName <> '' then - begin - FNextData := FDataServer.LoadDataFile(FNextFileName); - FNextData.Done.Subscribe(FHasData); - end; + PreloadNext; +end; + +procedure TAuraFileStream.BeforeDestruction; +begin + FCurrent.WaitFor; + FCurrent.Value.Data.WaitFor; + FNext.WaitFor; + FNext.Value.Data.WaitFor; + inherited; end; function TAuraFileStream.GetChunk(var Data: array of TDataPoint): Integer; var item: TDataPoint; - currData: TArray>; +label + loop; begin Result := 0; - // This is an asynchronous operation! We don't wait for data. It's totally valid to result nothing, if there is nothing. - if not FCurrentData.Done.IsSet then + loop: + if not FCurrent.Done.IsSet then + exit; + if not FCurrent.Value.Data.Done.IsSet then exit; - if FCurrPosInFile >= Length(FCurrentData.Value) then + // Changed: Use renamed field "FileInfo". + if not FCurrent.Value.FileInfo.IsValid then + exit; + + if FCurrPosInFile >= Length(FCurrent.Value.Data.Value) then begin - FCurrentData := FNextData; - FCurrentFileName := FNextFileName; FCurrPosInFile := 0; - FNextData := TFuture>>.Null; - - if FCurrentFileName = '' then - exit; - - FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName); - if FNextFileName <> '' then - begin - FNextData := FDataServer.LoadDataFile(FNextFileName); - FNextData.Done.Subscribe(FHasData); - end; - - if not FCurrentData.Done.IsSet then - exit; + FCurrent := FNext; + PreloadNext; + goto loop; end; - currData := FCurrentData.Value; - + var currData := FCurrent.Value.Data.Value; var maxLen := Length(Data); while (Result < maxLen) and (FCurrPosInFile < Length(currData)) do begin @@ -662,8 +530,6 @@ begin Inc(FCurrPosInFile); end; - // If there is data left in the current file, signal new data. If it is finished, we do nothing, because - // the next signal will come from the next future being done. if FCurrPosInFile < Length(currData) then begin FHasData.Notify; @@ -675,92 +541,191 @@ begin Result := FHasData.Signal; end; +function TAuraFileStream.GetSymbol: String; +begin + Result := FSymbol; +end; + function TAuraFileStream.IsHistory: Boolean; begin Result := True; end; +procedure TAuraFileStream.PreloadNext; +begin + FNext := + FCurrent.Chain( + function(const Prev: TDataStream): TDataStream + var + nextFileInfo: TAuraDataServer.TDataFile; + begin + // Changed: Use renamed field "FileInfo". + if Prev.FileInfo.IsValid then + begin + nextFileInfo := TAuraDataServer.FindNextDataFile(Prev.FileInfo); + if nextFileInfo.IsValid then + begin + Result.FileInfo := nextFileInfo; + Result.Data := FDataServer.LoadDataFile(nextFileInfo.GetFullFileName); + Result.Data.Done.Subscribe(FHasData); + end; + end; + end + ); +end; + { TAuraTABFileServer } function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream; -var - fileName: string; begin - // FindFirstDataFile uses the server's path and the given symbol to locate the initial data file. - fileName := FindFirstDataFile(FPath, Symbol); - - // Exit if no file is found or if it is not a .tab file, preventing stream creation for invalid sources. - if (fileName = '') or (not TPath.GetExtension(fileName).StartsWith('.tab')) then - exit(nil); - - // Creates a new stream instance, passing itself as the data server and the found filename. - Result := TAuraFileStream.Create(Self, fileName); + Result := TAuraFileStream.Create(Self, Symbol); end; -class function TAuraTABFileServer.TryParseFileName( - const FileName: string; - out PathValue, SymbolValue: string; - out YearValue, MonthValue: Integer -): Boolean; +function TAuraTABFileServer.LoadDataSeries(const FileName: string): TFuture>>; var - fileNameNoPath: string; - nameForParsing: string; + loadedState: TState; + loadedFiles: TArray>>>; + liveData: TFuture>>; + tabFiles: TList; + liveFilePath: string; + currentFileInfo: TDataFile; +begin + tabFiles := TList.Create; + try + if TryParseFileName(FileName, currentFileInfo) then + begin + while currentFileInfo.IsValid do + begin + tabFiles.Add(currentFileInfo.GetFullFileName); + currentFileInfo := FindNextDataFile(currentFileInfo); + end; + end; + + liveFilePath := ''; + if tabFiles.Count > 0 then + begin + var lastTabFile := tabFiles.Last; + if TryParseFileName(lastTabFile, currentFileInfo) then + begin + var liveFileBaseName := Format('%s_%d_%02d.tab-live', [currentFileInfo.Symbol, currentFileInfo.Year, currentFileInfo.Month]); + var potentialLivePath := TPath.Combine(currentFileInfo.Path, liveFileBaseName); + if TFile.Exists(potentialLivePath) then + liveFilePath := potentialLivePath; + end; + end; + + var loadedFileList := TList>>>.Create; + var loadStates := TList.Create; + try + for var fileStr in tabFiles do + begin + var data := LoadDataFile(fileStr); + loadedFileList.Add(data); + loadStates.Add(data.Done); + end; + + liveData := TFuture>>.Null; + if liveFilePath <> '' then + begin + liveData := LoadDataFile(liveFilePath); + loadedFileList.Add(liveData); + loadStates.Add(liveData.Done); + end; + + loadedState := TState.All(loadStates.ToArray); + loadedFiles := loadedFileList.ToArray; + finally + loadStates.Free; + loadedFileList.Free; + end; + finally + tabFiles.Free; + end; + + Result := + TFuture>>.Construct( + loadedState, + function: TArray> + begin + var tickList := TList>.Create; + try + var overallLastTabTickTime: TDateTime := 0; + var cnt := 0; + for var P in loadedFiles do + inc(cnt, Length(P.Value)); + tickList.Capacity := cnt; + + for var P in loadedFiles do + for var iterTickData in P.Value do + if overallLastTabTickTime < iterTickData.Time then + begin + tickList.Add(iterTickData); + overallLastTabTickTime := iterTickData.Time; + end; + Result := tickList.ToArray; + finally + tickList.Free; + end; + end + ); +end; + +class function TAuraTABFileServer.TryParseFileName(const FileName: string; out DataFile: TAuraDataServer.TDataFile): Boolean; +var + fileNameNoPath, nameForParsing, baseName, ext: string; parts: TArray; - baseName: string; begin Result := False; - PathValue := ''; - SymbolValue := ''; - YearValue := 0; - MonthValue := 0; - PathValue := TPath.GetDirectoryName(FileName); + DataFile := Default(TDataFile); + + DataFile.Path := TPath.GetDirectoryName(FileName); fileNameNoPath := TPath.GetFileName(FileName); nameForParsing := fileNameNoPath; + ext := ''; if nameForParsing.EndsWith('_zip', True) then begin - baseName := nameForParsing.Substring(0, nameForParsing.Length - 4); - if TPath.GetExtension(baseName) <> '' then - begin - nameForParsing := baseName; - end; + ext := '_zip'; + nameForParsing := nameForParsing.Substring(0, nameForParsing.Length - 4); end; - nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing); - parts := nameForParsing.Split(['_']); + DataFile.Extension := TPath.GetExtension(nameForParsing) + ext; + baseName := TPath.GetFileNameWithoutExtension(nameForParsing); + + parts := baseName.Split(['_']); if Length(parts) < 3 then exit; - if not TryStrToInt(parts[High(parts)], MonthValue) then + if not TryStrToInt(parts[High(parts)], DataFile.Month) then exit; - if (MonthValue < 1) or (MonthValue > 12) then + if (DataFile.Month < 1) or (DataFile.Month > 12) then begin - MonthValue := 0; + DataFile.Month := 0; exit; end; - if not TryStrToInt(parts[High(parts) - 1], YearValue) then + if not TryStrToInt(parts[High(parts) - 1], DataFile.Year) then begin - MonthValue := 0; + DataFile.Month := 0; exit; end; - if YearValue <= 0 then + if DataFile.Year <= 0 then begin - YearValue := 0; - MonthValue := 0; + DataFile.Year := 0; + DataFile.Month := 0; exit; end; - SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2)); + DataFile.Symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2)); - if SymbolValue = '' then + if DataFile.Symbol = '' then begin - PathValue := ''; - YearValue := 0; - MonthValue := 0; + DataFile.Path := ''; + DataFile.Year := 0; + DataFile.Month := 0; exit; end;