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

TDataSeries<T>

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-05 14:36:05 +02:00
parent 6bed68748d
commit f8c3ffceb8
14 changed files with 1051 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Core.Futures.pas
+16 -16
View File
@@ -59,22 +59,22 @@ begin
// Subscribe the job execution to AGate. // Subscribe the job execution to AGate.
// The job will run when AGate notifies the subscriber returned by Run. // The job will run when AGate notifies the subscriber returned by Run.
FInit := FInit :=
ATaskManager.CreateTask( ATaskManager.CreateTask(
AGate, AGate,
procedure procedure
begin begin
try try
try try
Self.FResult := AProc(); Self.FResult := AProc();
except except
Self.FResult := Default(T); // Set result to Default(T) on error Self.FResult := Default(T); // Set result to Default(T) on error
raise; // Re-raise for TaskFactory to handle raise; // Re-raise for TaskFactory to handle
end; end;
finally finally
Self.FDone.Notify; // Signal that this future is done (successfully or with error) Self.FDone.Notify; // Signal that this future is done (successfully or with error)
end; end;
end end
); );
end; end;
destructor TMycGateFuncFuture<T>.Destroy; destructor TMycGateFuncFuture<T>.Destroy;
Src/Myc.Core.Notifier.pas
+1 -1
View File
@@ -43,7 +43,7 @@ type
procedure Release; inline; // Releases the previously acquired exclusive lock. procedure Release; inline; // Releases the previously acquired exclusive lock.
function IsLocked: Boolean; inline; // Checks if the list is currently locked by any thread. function IsLocked: Boolean; inline; // Checks if the list is currently locked by any thread.
procedure Notify( procedure Notify(
Func: TPredicate<T> Func: TPredicate<T>
); // Iterates through registered receivers and invokes the predicate; removes receiver if predicate returns false. ); // Iterates through registered receivers and invokes the predicate; removes receiver if predicate returns false.
end; end;
Src/Myc.Core.Tasks.pas
+11 -11
View File
@@ -211,17 +211,17 @@ begin
capturedProc := Proc; // Capture Proc for the anonymous method capturedProc := Proc; // Capture Proc for the anonymous method
CreateAnonymousThread( CreateAnonymousThread(
'Thread', 'Thread',
procedure procedure
begin begin
try try
capturedProc(); // Execute the provided procedure capturedProc(); // Execute the provided procedure
finally finally
capturedProc := nil; // Clear the captured proc capturedProc := nil; // Clear the captured proc
res.Notify; // Signal completion via the latch's Notify method res.Notify; // Signal completion via the latch's Notify method
end; end;
end) end)
.Start; .Start;
// Return the state interface of the latch // Return the state interface of the latch
exit(res.State); exit(res.State);
Src/Myc.Test.Core.Atomic.pas
+3 -3
View File
@@ -279,9 +279,9 @@ begin
entryData := AllocEntryData(123); entryData := AllocEntryData(123);
// This should not raise an assertion error from TSList.Push. // This should not raise an assertion error from TSList.Push.
Assert.WillNotRaise( Assert.WillNotRaise(
procedure begin FList.Push(@entryData.Entry); end, procedure begin FList.Push(@entryData.Entry); end,
nil, nil,
'Pushing an aligned entry should not raise an exception/assertion.' 'Pushing an aligned entry should not raise an exception/assertion.'
); );
Assert.AreEqual(1, FList.QueryDepth, 'QueryDepth should be 1 after pushing an aligned entry.'); Assert.AreEqual(1, FList.QueryDepth, 'QueryDepth should be 1 after pushing an aligned entry.');
end; end;
Src/Myc.Test.Core.Lazy.pas
+62 -62
View File
@@ -187,14 +187,14 @@ begin
procExecuted := False; procExecuted := False;
expectedValue := 50; expectedValue := 50;
funcLazy := funcLazy :=
TMycFuncLazy<Integer>.Create( TMycFuncLazy<Integer>.Create(
sourceDirty.State, sourceDirty.State,
function: Integer function: Integer
begin begin
procExecuted := True; procExecuted := True;
Result := expectedValue; Result := expectedValue;
end end
); );
Assert.IsNotNull(funcLazy, 'TMycFuncLazy<Integer> instance should not be nil'); Assert.IsNotNull(funcLazy, 'TMycFuncLazy<Integer> instance should not be nil');
Assert.IsTrue(funcLazy.GetChanged.IsSet, 'Changed.IsSet should be true before the first Pop by design'); Assert.IsTrue(funcLazy.GetChanged.IsSet, 'Changed.IsSet should be true before the first Pop by design');
value := 0; value := 0;
@@ -223,9 +223,9 @@ begin
result := funcLazy.Pop(value); result := funcLazy.Pop(value);
Assert.IsTrue(result, 'First Pop should return true by design, even if FProc is nil'); Assert.IsTrue(result, 'First Pop should return true by design, even if FProc is nil');
Assert.AreEqual( Assert.AreEqual(
preCallValue, preCallValue,
value, value,
'Value should be unchanged as FProc was nil and current Pop implementation does not assign Default(T)' 'Value should be unchanged as FProc was nil and current Pop implementation does not assign Default(T)'
); );
Assert.IsFalse(funcLazy.GetChanged.IsSet, 'Changed state should be false after Pop'); Assert.IsFalse(funcLazy.GetChanged.IsSet, 'Changed state should be false after Pop');
end; end;
@@ -260,14 +260,14 @@ begin
initialProcValue := 44; initialProcValue := 44;
procExecutedCount := 0; procExecutedCount := 0;
funcLazy := funcLazy :=
TMycFuncLazy<Integer>.Create( TMycFuncLazy<Integer>.Create(
sourceDirty.State, sourceDirty.State,
function: Integer function: Integer
begin begin
Inc(procExecutedCount); Inc(procExecutedCount);
Result := initialProcValue; Result := initialProcValue;
end end
); );
Helper_ConsumeInitialPop(funcLazy, initialProcValue); Helper_ConsumeInitialPop(funcLazy, initialProcValue);
Assert.AreEqual(1, procExecutedCount, 'Proc should have executed once for initial pop'); Assert.AreEqual(1, procExecutedCount, 'Proc should have executed once for initial pop');
result := funcLazy.Pop(value); result := funcLazy.Pop(value);
@@ -305,17 +305,17 @@ begin
sourceDirty.Reset; sourceDirty.Reset;
procCallCount := 0; procCallCount := 0;
funcLazy := funcLazy :=
TMycFuncLazy<Integer>.Create( TMycFuncLazy<Integer>.Create(
sourceDirty.State, sourceDirty.State,
function: Integer function: Integer
begin begin
Inc(procCallCount); Inc(procCallCount);
if procCallCount = 1 then if procCallCount = 1 then
Result := 30 Result := 30
else else
Result := 300 + procCallCount; Result := 300 + procCallCount;
end end
); );
currentExpectedValue := 30; currentExpectedValue := 30;
Helper_ConsumeInitialPop(funcLazy, currentExpectedValue); Helper_ConsumeInitialPop(funcLazy, currentExpectedValue);
Assert.AreEqual(1, procCallCount, 'Proc executed for initial Pop'); Assert.AreEqual(1, procCallCount, 'Proc executed for initial Pop');
@@ -342,14 +342,14 @@ begin
sourceDirty.Reset; sourceDirty.Reset;
procCallCount := 0; procCallCount := 0;
funcLazy := funcLazy :=
TMycFuncLazy<Integer>.Create( TMycFuncLazy<Integer>.Create(
sourceDirty.State, sourceDirty.State,
function: Integer function: Integer
begin begin
Inc(procCallCount); Inc(procCallCount);
Result := 40; Result := 40;
end end
); );
resultPop1 := funcLazy.Pop(value); resultPop1 := funcLazy.Pop(value);
Assert.IsTrue(resultPop1, 'First Pop should return true by design'); Assert.IsTrue(resultPop1, 'First Pop should return true by design');
Assert.AreEqual(40, value, 'Value from first Pop'); Assert.AreEqual(40, value, 'Value from first Pop');
@@ -378,19 +378,19 @@ begin
firstSourceChangeValue := 52; firstSourceChangeValue := 52;
funcLazy := funcLazy :=
TMycFuncLazy<Integer>.Create( TMycFuncLazy<Integer>.Create(
sourceDirty.State, sourceDirty.State,
function: Integer function: Integer
begin begin
Inc(procCallCount); Inc(procCallCount);
if procCallCount = 1 then if procCallCount = 1 then
Result := initialValue // For initial pop Result := initialValue // For initial pop
else if procCallCount = 2 then else if procCallCount = 2 then
Result := firstSourceChangeValue // For pop after first effective source change Result := firstSourceChangeValue // For pop after first effective source change
else else
Result := 999; // Should not be reached in this specific test logic Result := 999; // Should not be reached in this specific test logic
end end
); );
// 1. Initial Pop (consumes "by design" changed state) // 1. Initial Pop (consumes "by design" changed state)
Assert.IsTrue(funcLazy.GetChanged.IsSet, 'Changed state should be true before first Pop (by design)'); Assert.IsTrue(funcLazy.GetChanged.IsSet, 'Changed state should be true before first Pop (by design)');
@@ -436,9 +436,9 @@ begin
Assert.IsTrue(funcLazyObj.Pop(tempVal), 'Initial Pop should succeed'); Assert.IsTrue(funcLazyObj.Pop(tempVal), 'Initial Pop should succeed');
funcLazyObj.Destroy; funcLazyObj.Destroy;
Assert.WillNotRaise( Assert.WillNotRaise(
procedure begin sourceDirty.Notify; end, procedure begin sourceDirty.Notify; end,
nil, nil,
'Destroy test assumes TMycSubscription.Unsubscribe works. Verified by no crash on source notify post-destroy.' 'Destroy test assumes TMycSubscription.Unsubscribe works. Verified by no crash on source notify post-destroy.'
); );
sourceDirty := nil; sourceDirty := nil;
end; end;
@@ -457,14 +457,14 @@ begin
expectedValue := 70; expectedValue := 70;
procExecuted := False; procExecuted := False;
funcLazy := funcLazy :=
TMycFuncLazy<Integer>.Create( TMycFuncLazy<Integer>.Create(
sourceDirty.State, sourceDirty.State,
function: Integer function: Integer
begin begin
procExecuted := True; procExecuted := True;
Result := expectedValue; Result := expectedValue;
end end
); );
Assert.IsNotNull(funcLazy, 'TMycFuncLazy<Integer> instance should not be nil'); Assert.IsNotNull(funcLazy, 'TMycFuncLazy<Integer> instance should not be nil');
Assert.IsTrue(funcLazy.GetChanged.IsSet, 'funcLazy.GetChanged.IsSet should be true after creation (by design)'); Assert.IsTrue(funcLazy.GetChanged.IsSet, 'funcLazy.GetChanged.IsSet should be true after creation (by design)');
value := 0; value := 0;
Src/Myc.Test.Lazy.pas
+54 -54
View File
@@ -146,14 +146,14 @@ begin
procExecuted := False; procExecuted := False;
// FChangingSignal is reset in Setup // FChangingSignal is reset in Setup
lazyIntf := lazyIntf :=
TLazy<Integer>.Construct( TLazy<Integer>.Construct(
FChangingSignal.State, FChangingSignal.State,
function: Integer function: Integer
begin begin
procExecuted := True; procExecuted := True;
Result := 10; Result := 10;
end end
); );
Assert.IsNotNull(lazyIntf, 'TLazy.Construct should return a valid interface'); Assert.IsNotNull(lazyIntf, 'TLazy.Construct should return a valid interface');
lazyRec := lazyIntf; // Implicit conversion lazyRec := lazyIntf; // Implicit conversion
@@ -171,14 +171,14 @@ begin
procExecuted := False; procExecuted := False;
expectedValue := 20; expectedValue := 20;
lazyIntf := lazyIntf :=
TLazy<Integer>.Construct( TLazy<Integer>.Construct(
FChangingSignal.State, FChangingSignal.State,
function: Integer function: Integer
begin begin
procExecuted := True; procExecuted := True;
Result := expectedValue; Result := expectedValue;
end end
); );
lazyRec := lazyIntf; lazyRec := lazyIntf;
ConsumeInitialPop(lazyRec, expectedValue, 'TestConstruct_FirstPop'); ConsumeInitialPop(lazyRec, expectedValue, 'TestConstruct_FirstPop');
@@ -236,14 +236,14 @@ var
begin begin
procCallCount := 0; procCallCount := 0;
lazyIntf := lazyIntf :=
TLazy<Integer>.Construct( TLazy<Integer>.Construct(
FChangingSignal.State, FChangingSignal.State,
function: Integer function: Integer
begin begin
Inc(procCallCount); Inc(procCallCount);
Result := 100 + procCallCount; // Value changes per call Result := 100 + procCallCount; // Value changes per call
end end
); );
lazyRec := lazyIntf; lazyRec := lazyIntf;
ConsumeInitialPop(lazyRec, 101, 'TestConstruct_StateInteraction_PopResetsChangedAfterSignal (Initial)'); // procCallCount = 1 ConsumeInitialPop(lazyRec, 101, 'TestConstruct_StateInteraction_PopResetsChangedAfterSignal (Initial)'); // procCallCount = 1
@@ -267,14 +267,14 @@ var
begin begin
procCallCount := 0; procCallCount := 0;
lazyIntf := lazyIntf :=
TLazy<Integer>.Construct( TLazy<Integer>.Construct(
FChangingSignal.State, FChangingSignal.State,
function: Integer function: Integer
begin begin
Inc(procCallCount); Inc(procCallCount);
Result := 200 + procCallCount; Result := 200 + procCallCount;
end end
); );
lazyRec := lazyIntf; lazyRec := lazyIntf;
// 1. Initial Pop // 1. Initial Pop
@@ -322,12 +322,12 @@ begin
// This should trigger its destructor, which should unsubscribe from localChangingSignal. // This should trigger its destructor, which should unsubscribe from localChangingSignal.
Assert.WillNotRaise( Assert.WillNotRaise(
procedure procedure
begin begin
localChangingSignal.Notify; // Notify the source AFTER the lazy object is supposed to be gone. localChangingSignal.Notify; // Notify the source AFTER the lazy object is supposed to be gone.
end, end,
nil, // Default: any exception is a failure nil, // Default: any exception is a failure
'Notifying source after lazy object is freed should not crash, indicating unsubscription.' 'Notifying source after lazy object is freed should not crash, indicating unsubscription.'
); );
localChangingSignal := nil; // Clean up the local signal itself. localChangingSignal := nil; // Clean up the local signal itself.
@@ -368,14 +368,14 @@ var
begin begin
procCallCount := 0; procCallCount := 0;
originalLazyIntf := originalLazyIntf :=
TLazy<Integer>.Construct( TLazy<Integer>.Construct(
FChangingSignal.State, FChangingSignal.State,
function: Integer function: Integer
begin begin
Inc(procCallCount); Inc(procCallCount);
Result := 70 + procCallCount; Result := 70 + procCallCount;
end end
); );
wrappedLazyRec := TLazy<Integer>.Create(originalLazyIntf); wrappedLazyRec := TLazy<Integer>.Create(originalLazyIntf);
// First Pop via wrapper (initial pop) // First Pop via wrapper (initial pop)
@@ -450,14 +450,14 @@ var
begin begin
procExecuted := False; procExecuted := False;
lazyIntf := lazyIntf :=
TLazy<Integer>.Construct( TLazy<Integer>.Construct(
FChangingSignal.State, FChangingSignal.State,
function: Integer function: Integer
begin begin
procExecuted := True; procExecuted := True;
Result := 100; Result := 100;
end end
); );
lazyRec := lazyIntf; lazyRec := lazyIntf;
ConsumeInitialPop(lazyRec, 100, 'TestPop_AfterInitialAndNoSignal (Initial)'); ConsumeInitialPop(lazyRec, 100, 'TestPop_AfterInitialAndNoSignal (Initial)');
Src/Myc.Trade.DataSeries.pas
+542
View File
@@ -0,0 +1,542 @@
unit Myc.Trade.DataSeries;
(* ------------------------------------------------------------------------------
This unit provides functions to load tick data (TTickData) from
binary files. The data originates from a C# application
and has the structure OADateTime (Double), Ask (Single), and Bid (Single).
The files follow the naming convention:
For .tab files: Symbol_Year_MM.tab OR Symbol_Year_MM.tab_zip
For .tab-live files: Symbol_Year_MM.tab-live (NEVER compressed)
(MM represents the two-digit month, e.g., 01 for January, 12 for December)
If both compressed (.tab_zip) and uncompressed (.tab) versions of a .tab
file exist, the compressed version is preferred. Files ending with _zip
are expected to be ZIP archives containing the actual data file
(e.g., Symbol_Year_MM.tab).
Loading occurs in two phases:
1. All .tab files (regular or _zip) are processed chronologically by
month and year. Unique ticks are loaded, and the timestamp of the
latest tick across all .tab files is determined (overallLastTabTickTime).
Assumes timestamps are unique across all .tab files.
2. All .tab-live files (which are never compressed), for months/years
corresponding to found .tab files, are processed. Ticks from a
.tab-live file are only integrated if their timestamp is strictly later
than overallLastTabTickTime. Assumes such ticks are globally unique.
If a .tab-live file results in no new data being added under these rules,
it is deleted after processing.
The loading function automatically detects subsequent monthly/yearly files.
Main Function:
LoadTickDataSeries - Loads a series of tick data files.
------------------------------------------------------------------------------ *)
interface
uses
System.SysUtils,
System.Classes,
System.Generics.Collections,
System.IOUtils,
Myc.Futures,
Myc.Signals;
type
TDataSeries<T: record> = class
type
TDataPoint = packed record
OADateTime: Double;
Data: T;
end;
protected
class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
public
class function LoadDataFile(var LoadGate: TLatch; const FileName: string): TFuture<TArray<TDataPoint>>; static;
class function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>; static;
end;
TAskBidItem = packed record
Ask: Single;
Bid: Single;
end;
TAskBid = class(TDataSeries<TAskBidItem>)
type
TTick = TDataSeries<TAskBidItem>.TDataPoint;
end;
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
// Finds the full filename of the oldest (=first) file for each symbol in the path.
function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
implementation
uses
System.Zip,
Myc.TaskManager;
// TryParseFileName parses filenames like "Symbol_Year_MM.Extension" or "Symbol_Year_MM.Extension_zip"
// to extract Path, Symbol, Year, and Month.
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
var
fileNameNoPath: string;
nameForParsing: string;
parts: TArray<string>;
baseName: string;
begin
Result := False; // Default to failure
PathValue := ''; // Initialize all out parameters
SymbolValue := '';
YearValue := 0;
MonthValue := 0;
PathValue := TPath.GetDirectoryName(FileName);
fileNameNoPath := TPath.GetFileName(FileName);
nameForParsing := fileNameNoPath;
// Handle cases like "Symbol_Year_MM.Extension_zip" by removing the trailing "_zip" before extension removal
if nameForParsing.EndsWith('_zip', True) then
begin
// Get the part before "_zip"
baseName := nameForParsing.Substring(0, nameForParsing.Length - 4);
// Get the extension of the original name (e.g. ".csv_zip")
// and check if baseName still contains an extension that needs to be removed by GetFileNameWithoutExtension
if TPath.GetExtension(baseName) <> '' then
begin
nameForParsing := baseName;
end
else
begin
// If baseName (e.g. "EURUSD_2023_01") has no extension, it is the correct parsing base
// However, TPath.GetFileNameWithoutExtension would also work correctly here,
// but this handles a potential case where a file might be named "Symbol_Year_MM_zip" (no dot before zip)
end;
end;
// Remove the actual extension (e.g., ".csv", ".dat") from "Symbol_Year_MM.Extension" or the modified "Symbol_Year_MM.Extension" part of "_zip"
nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing); // e.g., "EURUSD_2023_01"
parts := nameForParsing.Split(['_']);
if Length(parts) < 3 then // Need at least Symbol_Year_Month
exit;
// Month is the last part.
if not TryStrToInt(parts[High(parts)], MonthValue) then
begin
exit;
end;
if (MonthValue < 1) or (MonthValue > 12) then // Validate month range
begin
MonthValue := 0; // Set to invalid if out of range
exit;
end;
// Year is the second to last part.
if not TryStrToInt(parts[High(parts) - 1], YearValue) then
begin
MonthValue := 0; // Also invalidate month if year parsing fails
exit;
end;
// Ensure YearValue is plausible, e.g. not 0 or negative, adjust as needed for your data's year range
if YearValue <= 0 then
begin
YearValue := 0;
MonthValue := 0;
exit;
end;
// Symbol consists of all parts before Year and Month.
SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2));
if SymbolValue = '' then // Ensure symbol is not empty
begin
// Reset all out parameters on failure before exit
PathValue := '';
YearValue := 0;
MonthValue := 0;
exit;
end;
Result := True;
end;
// Finds the oldest file for each symbol in the given directory.
// The file extension is ignored for the purpose of identifying the symbol, year, and month.
function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
type
TTrackedFileInfo = record
FullFileName: string;
Year: Integer;
Month: Integer;
end;
var
oldestFilesPerSymbol: TDictionary<string, TTrackedFileInfo>;
fileNames: TArray<string>;
currentFile: string;
parsedPath: string;
parsedSymbol: string;
parsedYear: Integer;
parsedMonth: Integer;
trackedInfo: TTrackedFileInfo;
begin
oldestFilesPerSymbol := TDictionary<string, TTrackedFileInfo>.Create; // Dictionary to track the oldest file per symbol
try
if not TDirectory.Exists(DirectoryPath) then
begin
// Or raise an exception, depending on desired error handling
exit;
end;
fileNames := TDirectory.GetFiles(DirectoryPath); // Get all files in the directory
for currentFile in fileNames do
begin
// Attempt to parse the file name
if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
begin
// Check if this symbol is already tracked or if this file is older
if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then
begin
// If current file's year is less, or year is same and month is less, it's older
if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then
begin
trackedInfo.FullFileName := currentFile; // Store full path to the file
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo); // Update the entry for this symbol
end;
end
else
begin
// First time we see this symbol, so it's the oldest so far
trackedInfo.FullFileName := currentFile; // Store full path to the file
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo); // Add new entry for this symbol
end;
end;
end;
// Populate the result array with the full names of the oldest files
SetLength(Result, oldestFilesPerSymbol.Count);
var n := 0;
for trackedInfo in oldestFilesPerSymbol.Values do
begin
Result[n] := trackedInfo.FullFileName;
inc(n);
end;
finally
oldestFilesPerSymbol.Free; // Free the dictionary
end;
end;
// Asynchronously reads tick data from a single specified file.
// Returns a TFuture that will provide an array of TAskBidSeries.
class function TDataSeries<T>.LoadDataFile(var LoadGate: TLatch; const FileName: string): TFuture<TArray<TDataPoint>>;
var
parsedPath, parsedSymbol: string;
parsedYear, parsedMonth: Integer;
begin
Result := TFuture<TArray<TDataPoint>>.Null;
// Attempt to parse year and month from the filename
if not TryParseFileName(FileName, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
exit;
if FileName.EndsWith('_zip', True) then
begin
var zipFileBytes := TFile.ReadAllBytes(FileName); // Read all bytes of the .zip file
if Length(zipFileBytes) = 0 then
exit;
var capFileName := FileName;
Result :=
TFuture<TBytes>
.Construct(
TLatch.Enqueue(LoadGate),
function: TBytes
begin
Result := TFile.ReadAllBytes(capFileName); // Read all bytes of the .zip file
end)
.Chain<TArray<TDataPoint>>(
function(bytes: TBytes): TArray<TDataPoint>
begin
var zipMemoryStream := TBytesStream.Create(zipFileBytes); // Create a memory stream to hold the zip file content
try
zipMemoryStream.Position := 0;
Result := ReadCompressedData(zipMemoryStream);
finally
zipMemoryStream.Free;
end;
end);
end
else if TFile.Exists(FileName) then
begin
var capFileName := FileName;
Result :=
TFuture<TArray<TDataPoint>>.Construct(
function: TArray<TDataPoint>
begin
if TFile.Exists(capFileName) then
begin
var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
try
try
Result := ReadUncompressedData(stream);
except
on E: EReadError do
begin
raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]);
end;
end;
finally
stream.Free;
end;
end;
end
);
end;
end;
class function TDataSeries<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>;
// Local type declaration for storing file information for phase 1 processing
type
TPhase1FileEntry = record
Path: string;
Year: Integer;
Month: Integer;
end;
var
initialYear, initialMonth: Integer;
currentTabYear, currentTabMonth: Integer;
pathName, symbolName: string;
tabFileEntryRecord: TPhase1FileEntry; // Variable to construct record instances
loadedState: TState;
loadedFiles: TArray<TFuture<TArray<TDataPoint>>>;
liveData: TFuture<TArray<TDataPoint>>;
begin // Start of LoadDataSeries
if not TryParseFileName(FileName, pathName, symbolName, initialYear, initialMonth) then
begin
raise EArgumentException.CreateFmt(
'Invalid initial file name format: %s. Expected Symbol_Year_MM.Extension or Symbol_Year_MM.Extension_zip',
[FileName]);
end;
currentTabYear := initialYear;
currentTabMonth := initialMonth;
var tabBaseNameFormat: string;
var compressedTabPath, uncompressedTabPath: string;
var actualTabFileToLoad: string;
var maxTabYearFound := -1;
var maxTabMonthFound := -1;
var filesToLoadPhase1 := TList<TPhase1FileEntry>.Create; // Create list with the named record type
try
while True do
begin
tabBaseNameFormat := Format('%s_%.4d_%.2d.tab', [symbolName, currentTabYear, currentTabMonth]);
compressedTabPath := TPath.Combine(pathName, tabBaseNameFormat + '_zip');
uncompressedTabPath := TPath.Combine(pathName, tabBaseNameFormat);
actualTabFileToLoad := '';
if TFile.Exists(compressedTabPath) then
begin
actualTabFileToLoad := compressedTabPath;
end
else if TFile.Exists(uncompressedTabPath) then
begin
actualTabFileToLoad := uncompressedTabPath;
end;
if actualTabFileToLoad = '' then
break; // No more sequential .tab files
// Construct the record and add it to the list
tabFileEntryRecord.Path := actualTabFileToLoad;
tabFileEntryRecord.Year := currentTabYear;
tabFileEntryRecord.Month := currentTabMonth;
filesToLoadPhase1.Add(tabFileEntryRecord);
// Advance to the next month
Inc(currentTabMonth);
if currentTabMonth > 12 then
begin
currentTabMonth := 1;
Inc(currentTabYear);
end;
end;
// Now load and process identified .tab files
var loadedFileList := TList<TFuture<TArray<TDataPoint>>>.Create;
var loadStates := TList<TState>.Create;
try
var LoadGate: TLatch;
for var tabFileEntry: TPhase1FileEntry in filesToLoadPhase1 do // Iterate with the correct type
begin
var data := LoadDataFile(LoadGate, tabFileEntry.Path);
loadedFileList.Add(TFuture<TArray<TDataPoint>>.Create(data));
loadStates.Add(data.Done);
if tabFileEntry.Year > maxTabYearFound then
begin
maxTabYearFound := tabFileEntry.Year;
maxTabMonthFound := tabFileEntry.Month;
end
else if (tabFileEntry.Year = maxTabYearFound) and (tabFileEntry.Month > maxTabMonthFound) then
begin
maxTabMonthFound := tabFileEntry.Month;
end;
end;
var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, maxTabYearFound, maxTabMonthFound]);
liveData := LoadDataFile(LoadGate, TPath.Combine(pathName, liveFileBaseName));
loadStates.Add(liveData.Done);
loadedFileList.Add(liveData);
loadedState := TState.All(loadStates.ToArray);
loadedFiles := loadedFileList.ToArray;
finally
loadStates.Free;
loadedFileList.Free;
end;
finally
filesToLoadPhase1.Free;
end;
Result :=
TFuture<TArray<TDataPoint>>.Construct(
loadedState,
function: TArray<TDataPoint>
begin
var tickList := TList<TDataPoint>.Create;
try
var totalTicks := Length(liveData.Result);
var overallLastTabTickTime := 0.0;
for var P in loadedFiles do
Inc(totalTicks, Length(P.Result));
tickList.Capacity := totalTicks;
for var P in loadedFiles do
begin
if Length(P.Result) > 0 then
begin
for var iterTickData in P.Result do
if iterTickData.OADateTime > overallLastTabTickTime then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.OADateTime;
end;
end;
end;
Result := tickList.ToArray;
finally
tickList.Free;
end;
end
);
end;
class function TDataSeries<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>;
var
decompressionStream: TStream; // Holds uncompressed data of a single zip entry
localHeader: TZipHeader;
entryIndex: Integer;
i: Integer;
zipFileInstance: TZipFile;
begin
// Initialize Result record
SetLength(Result, 0);
decompressionStream := nil;
zipFileInstance := nil;
try
InputStream.Position := 0; // Reset stream position to the beginning for TZipFile
// Open TZipFile from this memory stream
zipFileInstance := TZipFile.Create;
zipFileInstance.Open(InputStream, TZipMode.zmRead); // Open the zip from the memory stream
if zipFileInstance.FileCount = 0 then
begin
// If TZipFile opens an empty or invalid stream successfully but finds no files
exit; // FileSuccessfullyLoaded remains False
end;
entryIndex := -1;
for i := 0 to zipFileInstance.FileCount - 1 do
begin
if zipFileInstance.FileNames[i].EndsWith('.tab', True) then
begin
entryIndex := i;
break;
end;
end;
if entryIndex = -1 then // If specific name not found, default to the first entry
begin
if zipFileInstance.FileCount > 0 then
entryIndex := 0
else
exit; // Should not happen if FileCount > 0 check passed
end;
// Decompress the specific entry from the in-memory zip file.
// TZipFile.Read creates and populates decompressionStream with uncompressed data.
zipFileInstance.Read(entryIndex, decompressionStream, localHeader);
if not Assigned(decompressionStream) then
exit; // Exit if no valid stream could be prepared
Result := ReadUncompressedData(decompressionStream);
finally
// Ensure all potentially created objects are freed
if Assigned(decompressionStream) then // Holds uncompressed data from a zip entry
decompressionStream.Free;
if Assigned(zipFileInstance) then
zipFileInstance.Free;
end;
end;
class function TDataSeries<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>;
var
fileSize: Int64;
recordCount: Integer;
bytesRead: Integer;
begin
// Initialize Result record
SetLength(Result, 0);
InputStream.Position := 0;
fileSize := InputStream.Size;
if fileSize = 0 then
exit;
if (fileSize mod SizeOf(TDataPoint)) <> 0 then
raise EReadError.CreateFmt('File corrupted. Size %d is not a multiple of record size %d.', [fileSize, SizeOf(TDataPoint)]);
recordCount := fileSize div SizeOf(TDataPoint);
if recordCount > 0 then
begin
SetLength(Result, recordCount);
bytesRead := InputStream.Read(Result[0], fileSize);
if bytesRead <> fileSize then
raise EReadError.CreateFmt('Read error. Expected to read %d bytes, but actually read %d bytes.', [fileSize, bytesRead]);
end;
end;
end.
Test/TestCoreFutures.pas
+83 -83
View File
@@ -99,15 +99,15 @@ begin
LInitStateAsState := TState.Null; LInitStateAsState := TState.Null;
LFuture := LFuture :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
FTaskFactory, FTaskFactory,
LInitStateAsState, LInitStateAsState,
function: Integer function: Integer
begin begin
Inc(Self.FProcExecutionCount); Inc(Self.FProcExecutionCount);
Result := CExpectedResult; Result := CExpectedResult;
end end
); );
FTaskFactory.WaitFor(LFuture.Done); // Accesses IMycFuture.GetDone [cite: 110, 234, 352] FTaskFactory.WaitFor(LFuture.Done); // Accesses IMycFuture.GetDone [cite: 110, 234, 352]
@@ -129,15 +129,15 @@ begin
LInitLatch := TLatch.Construct(1); // Create an init state that is not yet set [cite: 77, 212, 330] LInitLatch := TLatch.Construct(1); // Create an init state that is not yet set [cite: 77, 212, 330]
LFuture := LFuture :=
TMycGateFuncFuture<string>.Create( TMycGateFuncFuture<string>.Create(
FTaskFactory, FTaskFactory,
LInitLatch.State, LInitLatch.State,
function: string function: string
begin begin
Inc(Self.FProcExecutionCount); Inc(Self.FProcExecutionCount);
Result := CExpectedResult; Result := CExpectedResult;
end end
); );
Assert.AreEqual(0, FProcExecutionCount, 'AProc should not have executed yet.'); Assert.AreEqual(0, FProcExecutionCount, 'AProc should not have executed yet.');
Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be done yet.'); Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be done yet.');
@@ -168,15 +168,15 @@ begin
LInitStateAsState := TState.Null; // Immediate execution LInitStateAsState := TState.Null; // Immediate execution
LFuture := LFuture :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
LLocalTaskFactory, LLocalTaskFactory,
LInitStateAsState, LInitStateAsState,
function: Integer function: Integer
begin begin
Inc(Self.FProcExecutionCount); Inc(Self.FProcExecutionCount);
raise Exception.Create(CExceptionMessage); raise Exception.Create(CExceptionMessage);
end end
); );
try try
LLocalTaskFactory.WaitFor(LFuture.Done); LLocalTaskFactory.WaitFor(LFuture.Done);
@@ -225,15 +225,15 @@ begin
LInitLatch := TLatch.Construct(1); LInitLatch := TLatch.Construct(1);
LFuture := LFuture :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
FTaskFactory, FTaskFactory,
LInitLatch.State, LInitLatch.State,
function: Integer function: Integer
begin begin
Inc(Self.FProcExecutionCount); Inc(Self.FProcExecutionCount);
Result := 123; Result := 123;
end end
); );
Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be marked as done initially.'); Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be marked as done initially.');
@@ -264,44 +264,44 @@ begin
// Create MainFuture, AInitState is the GateLatch's state. // Create MainFuture, AInitState is the GateLatch's state.
LMainFuture := LMainFuture :=
TMycGateFuncFuture<string>.Create( TMycGateFuncFuture<string>.Create(
FTaskFactory, FTaskFactory,
LGateLatch.State, LGateLatch.State,
function: string function: string
begin begin
Inc(Self.FProcExecutionCount, 10); // Indicate MainFuture's proc ran Inc(Self.FProcExecutionCount, 10); // Indicate MainFuture's proc ran
Result := CMainFutureResult; Result := CMainFutureResult;
end end
); );
// Setup Prerequisite Futures // Setup Prerequisite Futures
// PrerequisiteFuture1 // PrerequisiteFuture1
LInitStateP1 := TLatch.Construct(1); // Controllable init state for PF1 LInitStateP1 := TLatch.Construct(1); // Controllable init state for PF1
LPrerequisiteFuture1 := LPrerequisiteFuture1 :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
FTaskFactory, FTaskFactory,
LInitStateP1.State, LInitStateP1.State,
function: Integer function: Integer
begin begin
Inc(Self.FProcExecutionCount, 1); // PF1 ran Inc(Self.FProcExecutionCount, 1); // PF1 ran
Result := 1; Result := 1;
end end
); );
LSub1 := TLatchNotifierSubscriber.Create(LGateLatch); LSub1 := TLatchNotifierSubscriber.Create(LGateLatch);
Subscriptions[1] := LPrerequisiteFuture1.Done.Subscribe(LSub1); // Subscribe to PF1's completion [cite: 56, 188, 306] Subscriptions[1] := LPrerequisiteFuture1.Done.Subscribe(LSub1); // Subscribe to PF1's completion [cite: 56, 188, 306]
// PrerequisiteFuture2 // PrerequisiteFuture2
LInitStateP2 := TLatch.Construct(1); // Controllable init state for PF2 LInitStateP2 := TLatch.Construct(1); // Controllable init state for PF2
LPrerequisiteFuture2 := LPrerequisiteFuture2 :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
FTaskFactory, FTaskFactory,
LInitStateP2.State, LInitStateP2.State,
function: Integer function: Integer
begin begin
Inc(Self.FProcExecutionCount, 1); // PF2 ran Inc(Self.FProcExecutionCount, 1); // PF2 ran
Result := 2; Result := 2;
end end
); );
LSub2 := TLatchNotifierSubscriber.Create(LGateLatch); LSub2 := TLatchNotifierSubscriber.Create(LGateLatch);
Subscriptions[2] := LPrerequisiteFuture2.Done.Subscribe(LSub2); // Subscribe to PF2's completion Subscriptions[2] := LPrerequisiteFuture2.Done.Subscribe(LSub2); // Subscribe to PF2's completion
@@ -347,29 +347,29 @@ begin
// Future A // Future A
LFutureA := LFutureA :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
FTaskFactory, FTaskFactory,
LTriggerLatch.State, LTriggerLatch.State,
function: Integer function: Integer
begin begin
LFlagFutureARan := True; LFlagFutureARan := True;
System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment
Result := 100; Result := 100;
end end
); );
// Future B // Future B
LFutureB := LFutureB :=
TMycGateFuncFuture<Integer>.Create( TMycGateFuncFuture<Integer>.Create(
FTaskFactory, FTaskFactory,
LTriggerLatch.State, LTriggerLatch.State,
function: Integer function: Integer
begin begin
LFlagFutureBRan := True; LFlagFutureBRan := True;
System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment
Result := 200; Result := 200;
end end
); );
Assert.IsFalse(LFutureA.Done.IsSet, 'FutureA should not be done yet.'); Assert.IsFalse(LFutureA.Done.IsSet, 'FutureA should not be done yet.');
Assert.IsFalse(LFutureB.Done.IsSet, 'FutureB should not be done yet.'); Assert.IsFalse(LFutureB.Done.IsSet, 'FutureB should not be done yet.');
Test/TestFutures.pas
+158 -158
View File
@@ -75,13 +75,13 @@ var
begin begin
// Test construction with a simple function that returns an Integer // Test construction with a simple function that returns an Integer
fut := fut :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
function: Integer function: Integer
begin begin
TThread.Sleep(20); // Simulate some background work TThread.Sleep(20); // Simulate some background work
Result := 42; Result := 42;
end end
); );
Assert.IsNotNull(fut.Done, 'Future.Done property should not be nil after construction.'); // Static string [cite: 148] Assert.IsNotNull(fut.Done, 'Future.Done property should not be nil after construction.'); // Static string [cite: 148]
fut.WaitFor(); // Wait for the future to complete its execution [cite: 148] fut.WaitFor(); // Wait for the future to complete its execution [cite: 148]
@@ -99,14 +99,14 @@ var
begin begin
// Test construction with a nil gate, which should execute the task immediately // Test construction with a nil gate, which should execute the task immediately
fut := fut :=
TFuture<Integer>.Construct( TFuture<Integer>.Construct(
nil, // Explicitly providing a nil gate [cite: 147] nil, // Explicitly providing a nil gate [cite: 147]
function: Integer function: Integer
begin begin
TThread.Sleep(20); // Simulate work TThread.Sleep(20); // Simulate work
Result := 43; Result := 43;
end end
); );
Assert.IsNotNull(fut.Done, 'Future.Done should not be nil when constructed with a nil gate.'); // Static string [cite: 148] Assert.IsNotNull(fut.Done, 'Future.Done should not be nil when constructed with a nil gate.'); // Static string [cite: 148]
fut.WaitFor(); // [cite: 148] fut.WaitFor(); // [cite: 148]
@@ -128,13 +128,13 @@ begin
// Construct a future with a gate that is already set // Construct a future with a gate that is already set
fut := fut :=
TFuture<Integer>.Construct( TFuture<Integer>.Construct(
presetGate, // [cite: 147] presetGate, // [cite: 147]
function: Integer function: Integer
begin begin
Result := 44; // This should execute quickly Result := 44; // This should execute quickly
end end
); );
Assert.IsNotNull(fut.Done, 'Future.Done should not be nil for preset gate construct.'); // Static string [cite: 148] Assert.IsNotNull(fut.Done, 'Future.Done should not be nil for preset gate construct.'); // Static string [cite: 148]
fut.WaitFor(); // [cite: 148] fut.WaitFor(); // [cite: 148]
@@ -157,10 +157,10 @@ begin
// Construct a future with a gate that is not yet set // Construct a future with a gate that is not yet set
fut := fut :=
TFuture<Integer>.Construct( TFuture<Integer>.Construct(
delayedGate.State, // Get the IMycState interface from the latch [cite: 147, 59] delayedGate.State, // Get the IMycState interface from the latch [cite: 147, 59]
function: Integer begin Result := 45; end function: Integer begin Result := 45; end
); );
Assert.IsNotNull(fut.Done, 'Future.Done should not be nil for delayed gate construct.'); // Static string [cite: 148] Assert.IsNotNull(fut.Done, 'Future.Done should not be nil for delayed gate construct.'); // Static string [cite: 148]
@@ -189,23 +189,23 @@ var
begin begin
// Create an initial future // Create an initial future
fut1 := fut1 :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
function: Integer function: Integer
begin begin
TThread.Sleep(20); // Simulate work TThread.Sleep(20); // Simulate work
Result := 100; Result := 100;
end end
); );
// Chain a second future that depends on the result of the first // Chain a second future that depends on the result of the first
fut2 := fut2 :=
fut1.Chain<string>( // [cite: 147] fut1.Chain<string>( // [cite: 147]
function(Input: Integer): string // This function receives the result of fut1 function(Input: Integer): string // This function receives the result of fut1
begin begin
TThread.Sleep(20); // Simulate further work TThread.Sleep(20); // Simulate further work
Result := 'Value: ' + Input.ToString; // Use ToString for converting Integer to String Result := 'Value: ' + Input.ToString; // Use ToString for converting Integer to String
end end
); );
Assert.IsNotNull(fut2.Done, 'Chained Future.Done should not be nil.'); // Static string [cite: 148] Assert.IsNotNull(fut2.Done, 'Chained Future.Done should not be nil.'); // Static string [cite: 148]
fut2.WaitFor(); // Wait for the chained future to complete [cite: 148] fut2.WaitFor(); // Wait for the chained future to complete [cite: 148]
@@ -229,16 +229,16 @@ begin
// First future depends on the delayedGate // First future depends on the delayedGate
fut1 := fut1 :=
TFuture<Integer>.Construct( TFuture<Integer>.Construct(
delayedGate.State, // [cite: 147, 59] delayedGate.State, // [cite: 147, 59]
function: Integer begin Result := 200; end function: Integer begin Result := 200; end
); );
// Second future is chained to the first // Second future is chained to the first
fut2 := fut2 :=
fut1.Chain<string>( // [cite: 147] fut1.Chain<string>( // [cite: 147]
function(Input: Integer): string begin Result := 'ChainVal: ' + Input.ToString; end function(Input: Integer): string begin Result := 'ChainVal: ' + Input.ToString; end
); );
Assert.IsNotNull(fut2.Done, 'Chained (with gate) Future.Done should not be nil.'); // Static string [cite: 148] Assert.IsNotNull(fut2.Done, 'Chained (with gate) Future.Done should not be nil.'); // Static string [cite: 148]
@@ -268,33 +268,33 @@ var
begin begin
// Initial future A // Initial future A
futA := futA :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
function: Integer function: Integer
begin begin
TThread.Sleep(10); TThread.Sleep(10);
Result := 10; Result := 10;
end end
); );
// Future B, chained from A // Future B, chained from A
futB := futB :=
futA.Chain<Real>( // [cite: 147] futA.Chain<Real>( // [cite: 147]
function(InputA: Integer): Real function(InputA: Integer): Real
begin begin
TThread.Sleep(10); TThread.Sleep(10);
Result := InputA * 2.5; // Calculation: 10 * 2.5 = 25.0 Result := InputA * 2.5; // Calculation: 10 * 2.5 = 25.0
end end
); );
// Future C, chained from B // Future C, chained from B
futC := futC :=
futB.Chain<string>( // [cite: 147] futB.Chain<string>( // [cite: 147]
function(InputB: Real): string function(InputB: Real): string
begin begin
TThread.Sleep(10); TThread.Sleep(10);
Result := 'Final: ' + FloatToStr(InputB); // Convert Real to String Result := 'Final: ' + FloatToStr(InputB); // Convert Real to String
end end
); );
Assert.IsNotNull(futC.Done, 'Multi-chained Future.Done should not be nil.'); // Static string [cite: 148] Assert.IsNotNull(futC.Done, 'Multi-chained Future.Done should not be nil.'); // Static string [cite: 148]
futC.WaitFor(); // Wait for the last future in the chain [cite: 148] futC.WaitFor(); // Wait for the last future in the chain [cite: 148]
@@ -318,13 +318,13 @@ var
begin begin
// Parametric test for Future<string> // Parametric test for Future<string>
fut := fut :=
TFuture<string>.Construct( // [cite: 146] TFuture<string>.Construct( // [cite: 146]
function: string function: string
begin begin
TThread.Sleep(10); TThread.Sleep(10);
Result := ParamValue; // Use the parameter in the future's function Result := ParamValue; // Use the parameter in the future's function
end end
); );
Assert.IsNotNull(fut.Done, 'Parametric Future.Done should not be nil.'); // Static string [cite: 148] Assert.IsNotNull(fut.Done, 'Parametric Future.Done should not be nil.'); // Static string [cite: 148]
fut.WaitFor(); // [cite: 148] fut.WaitFor(); // [cite: 148]
@@ -354,22 +354,22 @@ begin
begin begin
// Capture loop variable for use in anonymous method // Capture loop variable for use in anonymous method
Futures[i] := Futures[i] :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
( (
function(captureIndex: Integer): TFunc<Integer> function(captureIndex: Integer): TFunc<Integer>
begin begin
Result := Result :=
function: Integer function: Integer
var var
delay: Integer; delay: Integer;
begin begin
delay := 10 + Random(40); // Random delay between 10ms and 49ms delay := 10 + Random(40); // Random delay between 10ms and 49ms
TThread.Sleep(delay); TThread.Sleep(delay);
Result := captureIndex; // Return the captured index Result := captureIndex; // Return the captured index
end; end;
end end
)(i) )(i)
); );
doneStates[i] := Futures[i].Done; // [cite: 148] doneStates[i] := Futures[i].Done; // [cite: 148]
end; end;
@@ -378,13 +378,13 @@ begin
// Create a master future that waits on the combined State.All state // Create a master future that waits on the combined State.All state
masterFuture := masterFuture :=
TFuture<Boolean>.Construct( TFuture<Boolean>.Construct(
combinedStateAll, // [cite: 147] combinedStateAll, // [cite: 147]
function: Boolean function: Boolean
begin begin
Result := True; // This function executes when combinedStateAll is set Result := True; // This function executes when combinedStateAll is set
end end
); );
masterFuture.WaitFor(); // Wait for all futures to complete via the master future [cite: 148] masterFuture.WaitFor(); // Wait for all futures to complete via the master future [cite: 148]
Assert.IsTrue(combinedStateAll.IsSet, 'Combined State.All should be set after masterFuture.WaitFor().'); // Static string [cite: 55] Assert.IsTrue(combinedStateAll.IsSet, 'Combined State.All should be set after masterFuture.WaitFor().'); // Static string [cite: 55]
@@ -424,25 +424,25 @@ begin
begin begin
// Capture loop variable // Capture loop variable
Futures[i] := Futures[i] :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
( (
function(captureIndex: Integer): TFunc<Integer> function(captureIndex: Integer): TFunc<Integer>
begin begin
Result := Result :=
function: Integer function: Integer
var var
delay: Integer; delay: Integer;
begin begin
if captureIndex = QuickFutureIndex then if captureIndex = QuickFutureIndex then
delay := 5 // Short delay for the 'quick' future delay := 5 // Short delay for the 'quick' future
else else
delay := 50 + Random(100); // Longer random delay (50-149ms) for others delay := 50 + Random(100); // Longer random delay (50-149ms) for others
TThread.Sleep(delay); TThread.Sleep(delay);
Result := captureIndex; Result := captureIndex;
end; end;
end end
)(i) )(i)
); );
doneStates[i] := Futures[i].Done; // [cite: 148] doneStates[i] := Futures[i].Done; // [cite: 148]
end; end;
@@ -451,13 +451,13 @@ begin
// Create a master future that waits on the combined State.Any state // Create a master future that waits on the combined State.Any state
masterFuture := masterFuture :=
TFuture<Boolean>.Construct( TFuture<Boolean>.Construct(
combinedStateAny, // [cite: 147] combinedStateAny, // [cite: 147]
function: Boolean function: Boolean
begin begin
Result := True; // This function executes when combinedStateAny is set Result := True; // This function executes when combinedStateAny is set
end end
); );
masterFuture.WaitFor(); // Wait for at least one future to complete [cite: 148] masterFuture.WaitFor(); // Wait for at least one future to complete [cite: 148]
Assert.IsTrue(combinedStateAny.IsSet, 'Combined State.Any should be set after masterFuture.WaitFor().'); // Static string [cite: 55] Assert.IsTrue(combinedStateAny.IsSet, 'Combined State.Any should be set after masterFuture.WaitFor().'); // Static string [cite: 55]
@@ -492,20 +492,20 @@ var
begin begin
// Create an outer future that, when resolved, produces another (inner) future. // Create an outer future that, when resolved, produces another (inner) future.
outerFuture := outerFuture :=
TFuture<TFuture<Integer>>.Construct( // [cite: 146] TFuture<TFuture<Integer>>.Construct( // [cite: 146]
function: TFuture<Integer> // This lambda returns a Future<Integer> function: TFuture<Integer> // This lambda returns a Future<Integer>
begin begin
TThread.Sleep(10); // Simulate work for the outer future to produce the inner one TThread.Sleep(10); // Simulate work for the outer future to produce the inner one
Result := Result :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
function: Integer function: Integer
begin begin
TThread.Sleep(10); // Simulate work for the inner future TThread.Sleep(10); // Simulate work for the inner future
Result := 123; // The final value Result := 123; // The final value
end end
);
end
); );
end
);
Assert.IsNotNull(outerFuture.Done, 'Outer future''s Done state should not be nil.'); // Static string [cite: 148] Assert.IsNotNull(outerFuture.Done, 'Outer future''s Done state should not be nil.'); // Static string [cite: 148]
outerFuture.WaitFor(); // Wait for the outer future to complete and yield the inner future [cite: 148] outerFuture.WaitFor(); // Wait for the outer future to complete and yield the inner future [cite: 148]
@@ -531,31 +531,31 @@ var
begin begin
// Create an initial future // Create an initial future
initialFuture := initialFuture :=
TFuture<Integer>.Construct( // [cite: 146] TFuture<Integer>.Construct( // [cite: 146]
function: Integer function: Integer
begin begin
TThread.Sleep(10); TThread.Sleep(10);
Result := 77; Result := 77;
end end
); );
// Chain it with a function that itself returns a new Future<string> // Chain it with a function that itself returns a new Future<string>
outerChainedFuture := outerChainedFuture :=
initialFuture.Chain<TFuture<string>>( // [cite: 147] initialFuture.Chain<TFuture<string>>( // [cite: 147]
function(Input: Integer): TFuture<string> // This lambda returns a Future<string> function(Input: Integer): TFuture<string> // This lambda returns a Future<string>
begin begin
TThread.Sleep(10); // Simulate work in the chain function TThread.Sleep(10); // Simulate work in the chain function
// Input is the result of initialFuture (77) // Input is the result of initialFuture (77)
Result := Result :=
TFuture<string>.Construct( // [cite: 146] TFuture<string>.Construct( // [cite: 146]
function: string function: string
begin begin
TThread.Sleep(10); // Simulate work for the inner-most future TThread.Sleep(10); // Simulate work for the inner-most future
Result := 'Value: ' + Input.ToString; // Input is captured (77) Result := 'Value: ' + Input.ToString; // Input is captured (77)
end end
);
end
); );
end
);
Assert.IsNotNull(outerChainedFuture.Done, 'Outer chained future''s Done state should not be nil.'); // Static string [cite: 148] Assert.IsNotNull(outerChainedFuture.Done, 'Outer chained future''s Done state should not be nil.'); // Static string [cite: 148]
outerChainedFuture.WaitFor(); // Wait for initialFuture to complete AND the chain function to execute [cite: 148] outerChainedFuture.WaitFor(); // Wait for initialFuture to complete AND the chain function to execute [cite: 148]
Test/TestNotifier.pas
+17 -17
View File
@@ -106,11 +106,11 @@ var
begin begin
predicateCallCount := 0; predicateCallCount := 0;
dummyPredicate := dummyPredicate :=
function(Item: IMyTestInterface): Boolean function(Item: IMyTestInterface): Boolean
begin begin
Inc(predicateCallCount); Inc(predicateCallCount);
Result := True; Result := True;
end; end;
FNotifier.Lock; FNotifier.Lock;
Assert.IsTrue(FNotifier.IsLocked, 'Notifier should be locked.'); Assert.IsTrue(FNotifier.IsLocked, 'Notifier should be locked.');
@@ -209,12 +209,12 @@ begin
itemsProcessedCount := 0; itemsProcessedCount := 0;
FNotifier.Notify( FNotifier.Notify(
function(Item: IMyTestInterface): Boolean function(Item: IMyTestInterface): Boolean
begin begin
Inc(itemsProcessedCount); Inc(itemsProcessedCount);
TMyTestReceiver(Item).Foo; TMyTestReceiver(Item).Foo;
Result := Item.GetValue <> 20; // Remove item with value 20 Result := Item.GetValue <> 20; // Remove item with value 20
end end
); );
Assert.AreEqual(3, itemsProcessedCount, 'Notify predicate called for all 3 items.'); Assert.AreEqual(3, itemsProcessedCount, 'Notify predicate called for all 3 items.');
@@ -229,12 +229,12 @@ begin
itemsProcessedCount := 0; itemsProcessedCount := 0;
FNotifier.Notify( FNotifier.Notify(
function(Item: IMyTestInterface): Boolean function(Item: IMyTestInterface): Boolean
begin begin
Inc(itemsProcessedCount); Inc(itemsProcessedCount);
TMyTestReceiver(Item).Foo; TMyTestReceiver(Item).Foo;
Result := True; // Keep remaining Result := True; // Keep remaining
end end
); );
Assert.AreEqual(2, itemsProcessedCount, 'Notify predicate called for 2 remaining items.'); Assert.AreEqual(2, itemsProcessedCount, 'Notify predicate called for 2 remaining items.');
Assert.AreEqual(1, rcv1.CallCount, 'Receiver1 called again.'); Assert.AreEqual(1, rcv1.CallCount, 'Receiver1 called again.');
Test/TestNotifier_ChaosStress.pas
+27 -32
View File
@@ -213,11 +213,11 @@ begin
// if not FOwnerFixture.FNotifier.IsFinalized then // Don't notify if finalized // if not FOwnerFixture.FNotifier.IsFinalized then // Don't notify if finalized
begin begin
FOwnerFixture.FNotifier.Notify( FOwnerFixture.FNotifier.Notify(
function(Item: IMyStressTestInterface): Boolean function(Item: IMyStressTestInterface): Boolean
begin begin
Item.Foo(FThreadID); // Pass ThreadID as notification type for context Item.Foo(FThreadID); // Pass ThreadID as notification type for context
Result := True; // Keep item Result := True; // Keep item
end end
); );
end; end;
finally finally
@@ -359,11 +359,11 @@ begin
// if not FNotifier.IsFinalized then // if not FNotifier.IsFinalized then
begin begin
FNotifier.Notify( FNotifier.Notify(
function(Item: IMyStressTestInterface): Boolean function(Item: IMyStressTestInterface): Boolean
begin begin
actualLiveReceiversInNotifier.Add(Item); actualLiveReceiversInNotifier.Add(Item);
Result := True; Result := True;
end end
); );
end; end;
finally finally
@@ -373,12 +373,12 @@ begin
// 2. Compare overall counts // 2. Compare overall counts
Assert.AreEqual( Assert.AreEqual(
totalExpectedLiveByThreadsAtEnd, totalExpectedLiveByThreadsAtEnd,
actualLiveInNotifierAtEnd, actualLiveInNotifierAtEnd,
Format( Format(
'Mismatch in live item count at end. Threads expected %d, Notifier has %d.', 'Mismatch in live item count at end. Threads expected %d, Notifier has %d.',
[totalExpectedLiveByThreadsAtEnd, actualLiveInNotifierAtEnd] [totalExpectedLiveByThreadsAtEnd, actualLiveInNotifierAtEnd]
) )
); );
// 3. Detailed check: Iterate all receivers ever created. // 3. Detailed check: Iterate all receivers ever created.
@@ -400,27 +400,22 @@ begin
end; end;
Assert.AreEqual( Assert.AreEqual(
receiver.ExpectedToBeAdvised, receiver.ExpectedToBeAdvised,
found, found,
Format( Format(
'Receiver ID %d: Thread expected it to be advised=%d, but its presence in Notifier is %d. NotificationCount=%d', 'Receiver ID %d: Thread expected it to be advised=%d, but its presence in Notifier is %d. NotificationCount=%d',
[ [receiver.GetInstanceID, Integer(receiver.ExpectedToBeAdvised), Integer(found), receiver.GetNotificationCount]
receiver.GetInstanceID, )
Integer(receiver.ExpectedToBeAdvised),
Integer(found),
receiver.GetNotificationCount
]
)
); );
// Further checks on NotificationCount could be added if specific notification patterns were expected. // Further checks on NotificationCount could be added if specific notification patterns were expected.
// For this chaos test, ensuring count is non-negative and consistent with advised state is a good start. // For this chaos test, ensuring count is non-negative and consistent with advised state is a good start.
Assert.IsTrue( Assert.IsTrue(
receiver.GetNotificationCount >= 0, receiver.GetNotificationCount >= 0,
Format( Format(
'Receiver ID %d has non-positive notification count: %d', 'Receiver ID %d has non-positive notification count: %d',
[receiver.GetInstanceID, receiver.GetNotificationCount] [receiver.GetInstanceID, receiver.GetNotificationCount]
) )
); );
if found and (receiver.GetNotificationCount = 0) then if found and (receiver.GetNotificationCount = 0) then
begin begin
Test/TestNotifier_Threading.pas
+20 -20
View File
@@ -256,23 +256,23 @@ begin
FNotifier.Lock; FNotifier.Lock;
try try
FNotifier.Notify( FNotifier.Notify(
function(Item: IMyTestInterface): Boolean function(Item: IMyTestInterface): Boolean
begin begin
Inc(currentNotifyCount); Inc(currentNotifyCount);
Result := True; // Keep item in the Notifier Result := True; // Keep item in the Notifier
end end
); );
finally finally
FNotifier.Release; FNotifier.Release;
end; end;
Assert.AreEqual( Assert.AreEqual(
totalAdvisedExpected, totalAdvisedExpected,
currentNotifyCount, currentNotifyCount,
'The number of items in the Notifier after all Advise operations (' 'The number of items in the Notifier after all Advise operations ('
+ currentNotifyCount.ToString + currentNotifyCount.ToString
+ ') does not match the expected number (' + ') does not match the expected number ('
+ totalAdvisedExpected.ToString + totalAdvisedExpected.ToString
+ ').' + ').'
); );
// 2. Execute UnadviseAll // 2. Execute UnadviseAll
@@ -288,19 +288,19 @@ begin
FNotifier.Lock; FNotifier.Lock;
try try
FNotifier.Notify( FNotifier.Notify(
function(Item: IMyTestInterface): Boolean function(Item: IMyTestInterface): Boolean
begin begin
Inc(currentNotifyCount); Inc(currentNotifyCount);
Result := True; Result := True;
end end
); );
finally finally
FNotifier.Release; FNotifier.Release;
end; end;
Assert.AreEqual( Assert.AreEqual(
0, 0,
currentNotifyCount, currentNotifyCount,
'The Notifier should be empty after UnadviseAll, but still contains ' + currentNotifyCount.ToString + ' items.' 'The Notifier should be empty after UnadviseAll, but still contains ' + currentNotifyCount.ToString + ' items.'
); );
end; end;
Test/TestSignals_Latch.pas
+11 -14
View File
@@ -129,9 +129,9 @@ begin
end; end;
procedure TTestMycLatch.TestNotify_ReturnValueAndState_AfterOneNotify( procedure TTestMycLatch.TestNotify_ReturnValueAndState_AfterOneNotify(
InitialCount: Integer; InitialCount: Integer;
ExpectedReturn: Boolean; ExpectedReturn: Boolean;
ExpectedIsSet: Boolean ExpectedIsSet: Boolean
); );
var var
latch: IMycLatch; latch: IMycLatch;
@@ -141,14 +141,14 @@ begin
returnedValueFromNotify := latch.Notify; // This is IMycLatch (as IMycSubscriber).Notify returnedValueFromNotify := latch.Notify; // This is IMycLatch (as IMycSubscriber).Notify
Assert.AreEqual( Assert.AreEqual(
ExpectedReturn, ExpectedReturn,
returnedValueFromNotify, returnedValueFromNotify,
'Unexpected return value from Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.' 'Unexpected return value from Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.'
); );
Assert.AreEqual( Assert.AreEqual(
ExpectedIsSet, ExpectedIsSet,
latch.State.IsSet, latch.State.IsSet,
'Unexpected IsSet state after Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.' 'Unexpected IsSet state after Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.'
); );
end; end;
@@ -504,11 +504,8 @@ begin
Assert.AreEqual(1, mockSubInitial.NotifyCount, 'Initial subscriber (unadvised) not notified again.'); Assert.AreEqual(1, mockSubInitial.NotifyCount, 'Initial subscriber (unadvised) not notified again.');
// mockSubNew should also not be notified again, as it was only notified immediately // mockSubNew should also not be notified again, as it was only notified immediately
// and not persistently added to FSubscribers for subsequent Latch.Notify calls. // and not persistently added to FSubscribers for subsequent Latch.Notify calls.
Assert.AreEqual( Assert
1, .AreEqual(1, mockSubNew.NotifyCount, 'New subscriber (notified once on subscribe) not notified by subsequent Latch.Notify calls.');
mockSubNew.NotifyCount,
'New subscriber (notified once on subscribe) not notified by subsequent Latch.Notify calls.'
);
end; end;
initialization initialization
Test/TestTasks.pas
+46 -46
View File
@@ -92,13 +92,13 @@ begin
jobCompletedLatch := TLatch.Construct(1); // Changed from Signals.CreateLatch jobCompletedLatch := TLatch.Construct(1); // Changed from Signals.CreateLatch
FFactory FFactory
.Run( .Run(
procedure procedure
begin begin
jobExecuted := True; jobExecuted := True;
jobCompletedLatch.Notify; jobCompletedLatch.Notify;
end) end)
.Notify; .Notify;
FFactory.WaitFor(jobCompletedLatch.State); FFactory.WaitFor(jobCompletedLatch.State);
Assert.IsTrue(jobExecuted, 'Immediate job should have executed'); Assert.IsTrue(jobExecuted, 'Immediate job should have executed');
@@ -114,13 +114,13 @@ begin
jobCompletedLatch := TLatch.Construct(1); // Changed from Signals.CreateLatch jobCompletedLatch := TLatch.Construct(1); // Changed from Signals.CreateLatch
subscriber := subscriber :=
FFactory.Run( FFactory.Run(
procedure procedure
begin begin
jobExecuted := True; jobExecuted := True;
jobCompletedLatch.Notify; jobCompletedLatch.Notify;
end end
); );
Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job'); Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job');
// Use TMycLatch.Null for comparison // Use TMycLatch.Null for comparison
@@ -162,12 +162,12 @@ begin
Assert.IsFalse(FFactory.InWorkerThread, 'Test method itself should not be in a factory worker thread'); Assert.IsFalse(FFactory.InWorkerThread, 'Test method itself should not be in a factory worker thread');
FFactory.EnqueueJob( FFactory.EnqueueJob(
procedure procedure
begin begin
inMainThreadInJob := FFactory.InMainThread; inMainThreadInJob := FFactory.InMainThread;
inWorkerThreadInJob := FFactory.InWorkerThread; inWorkerThreadInJob := FFactory.InWorkerThread;
jobDoneLatch.Notify; jobDoneLatch.Notify;
end end
); );
FFactory.WaitFor(jobDoneLatch.State); FFactory.WaitFor(jobDoneLatch.State);
@@ -183,20 +183,20 @@ begin
jobDoneLatch := TLatch.Construct(1); jobDoneLatch := TLatch.Construct(1);
FFactory.EnqueueJob( FFactory.EnqueueJob(
procedure procedure
begin begin
try try
raise TMycTaskFactory.ETaskException.Create('Test Exception From Job'); raise TMycTaskFactory.ETaskException.Create('Test Exception From Job');
finally finally
jobDoneLatch.Notify; jobDoneLatch.Notify;
end; end;
end end
); );
Assert.WillRaise( Assert.WillRaise(
procedure begin FFactory.WaitFor(jobDoneLatch.State); end, procedure begin FFactory.WaitFor(jobDoneLatch.State); end,
TMycTaskFactory.ETaskException, TMycTaskFactory.ETaskException,
'WaitFor should re-raise the exception from the job' 'WaitFor should re-raise the exception from the job'
); );
var noExceptionRaised: Boolean := True; var noExceptionRaised: Boolean := True;
@@ -213,9 +213,9 @@ begin
FFactory.Teardown; FFactory.Teardown;
Assert.WillRaise( Assert.WillRaise(
procedure begin FFactory.EnqueueJob(procedure begin end); end, procedure begin FFactory.EnqueueJob(procedure begin end); end,
TMycTaskFactory.ETaskException, TMycTaskFactory.ETaskException,
'Running a job on a torn-down factory should raise ETaskException' 'Running a job on a torn-down factory should raise ETaskException'
); );
end; end;
@@ -230,18 +230,18 @@ begin
dummyStateToWaitFor := TLatch.Construct(1).State; dummyStateToWaitFor := TLatch.Construct(1).State;
FFactory.EnqueueJob( FFactory.EnqueueJob(
procedure procedure
begin begin
try try
FFactory.WaitFor(dummyStateToWaitFor); FFactory.WaitFor(dummyStateToWaitFor);
except except
on E: TMycTaskFactory.ETaskException do on E: TMycTaskFactory.ETaskException do
begin begin
exceptionCaughtInJob := True; exceptionCaughtInJob := True;
end; end;
end; end;
jobDoneLatch.Notify; jobDoneLatch.Notify;
end end
); );
FFactory.WaitFor(jobDoneLatch.State); FFactory.WaitFor(jobDoneLatch.State);