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

Refactoring

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-02 12:27:52 +02:00
parent 86e2729a52
commit 762e7f83e2
8 changed files with 294 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Core.Futures.pas
+80
View File
@@ -0,0 +1,80 @@
unit Myc.Core.Futures;
interface
uses
System.SysUtils,
Myc.Signals, Myc.TaskManager, Myc.Futures;
type
// Abstract base class for IMycFuture<T> implementations.
TMycFuture<T> = class abstract( TInterfacedObject, IMycFuture<T> )
protected
function GetResult: T; virtual; abstract;
function GetDone: IMycState; virtual; abstract;
end;
// Concrete future implementation triggered by an initial state, executing a TFunc<T>.
TMycInitStateFuncFuture<T> = class( TMycFuture<T> )
private
FInit: TMycSubscription;
FDone: IMycLatch;
FResult: T;
protected
function GetResult: T; override;
function GetDone: IMycState; override;
public
// Creates a future that executes AProc after AGate is set, using ATaskFactory.
constructor Create( const ATaskManager: IMycTaskManager; const AGate: IMycState; AProc: TFunc<T> );
destructor Destroy; override;
end;
implementation
{ TMycInitStateFuncFuture<T> }
constructor TMycInitStateFuncFuture<T>.Create( const ATaskManager: IMycTaskManager; const AGate: IMycState; AProc: TFunc<T> );
begin
inherited Create;
FDone := TMycLatch.CreateLatch( 1 );
// Subscribe the job execution to AGate.
// The job will run when AGate notifies the subscriber returned by Run.
FInit := ATaskManager.CreateTask(
AGate,
procedure
begin
try
try
Self.FResult := AProc( );
except
Self.FResult := Default ( T ); // Set result to Default(T) on error
raise; // Re-raise for TaskFactory to handle
end;
finally
Self.FDone.Notify; // Signal that this future is done (successfully or with error)
end;
end );
end;
destructor TMycInitStateFuncFuture<T>.Destroy;
begin
Assert( FDone.State.IsSet );
FInit.Unsubscribe;
inherited Destroy;
end;
function TMycInitStateFuncFuture<T>.GetDone: IMycState;
begin
Result := FDone.State;
end;
function TMycInitStateFuncFuture<T>.GetResult: T;
begin
Assert( FDone.State.IsSet, 'Result is not yet available.' );
Result := FResult;
end;
end.
Src/Myc.Core.Tasks.pas
+80 -47
View File
@@ -4,55 +4,55 @@ interface
uses
System.SysUtils, System.Classes, System.SyncObjs,
Myc.Core.Atomic, Myc.Signals; // Myc.Signals now requires direct use of TMycLatch static members
Myc.Core.Atomic, Myc.TaskManager, Myc.Signals; // Myc.Signals now requires direct use of TMycLatch static members
type
IMycTaskFactory = interface
{$region 'property access'}
// Retrieves the number of worker threads.
function GetThreadCount: Integer;
{$endregion}
// Creates and starts a new, independent thread executing Proc.
// Returns IMycState to await thread completion.
// Exceptions within Proc are caught; the first one is stored
// and can be re-raised in the main thread via WaitFor or Teardown.
function CreateThread(const Proc: TProc): IMycState;
// Returns True if called from the main application thread.
function InMainThread: Boolean;
// Returns True if called from one of the factory's worker threads.
function InWorkerThread: Boolean;
// Schedules Job for execution by a worker thread.
// If StartCount > 0, Job is deferred until Notify is called StartCount times on the returned IMycSubscriber.
// If StartCount <= 0, Job is enqueued immediately.
// Returns IMycSubscriber for deferred jobs, or TMycLatch.Null for immediate jobs or if Job is nil.
// Exceptions during Job execution are caught; the first one is stored
// and can be re-raised in the main thread via WaitFor or Teardown.
// Raises ETaskException if the factory is already terminated.
function Run(Job: TProc): IMycSubscriber;
procedure EnqueueJob(const Job: TProc);
// Shuts down the task factory: terminates worker threads and cleans up resources.
// Before shutdown, any first stored exception from a worker thread (from Run or CreateThread tasks managed by this factory)
// will be re-raised in the calling (main) thread.
procedure Teardown;
// Waits for the operation associated with State to complete.
// Must not be called from a worker thread of this factory.
// After waiting, or if the state is already set, any first stored exception
// from any worker thread of this factory will be re-raised in the calling (main) thread.
// Raises ETaskException if called from within a worker thread.
procedure WaitFor(State: IMycState);
// Gets the number of worker threads.
property ThreadCount: Integer read GetThreadCount;
end;
IMycTaskFactory = interface( IMycTaskManager )
{$region 'property access'}
// Retrieves the number of worker threads.
function GetThreadCount: Integer;
{$endregion}
TMycTaskFactory = class(TInterfacedObject, IMycTaskFactory)
// Creates and starts a new, independent thread executing Proc.
// Returns IMycState to await thread completion.
// Exceptions within Proc are caught; the first one is stored
// and can be re-raised in the main thread via WaitFor or Teardown.
function CreateThread(const Proc: TProc): IMycState;
// Returns True if called from the main application thread.
function InMainThread: Boolean;
// Returns True if called from one of the factory's worker threads.
function InWorkerThread: Boolean;
// Schedules Job for execution by a worker thread.
// If StartCount > 0, Job is deferred until Notify is called StartCount times on the returned IMycSubscriber.
// If StartCount <= 0, Job is enqueued immediately.
// Returns IMycSubscriber for deferred jobs, or TMycLatch.Null for immediate jobs or if Job is nil.
// Exceptions during Job execution are caught; the first one is stored
// and can be re-raised in the main thread via WaitFor or Teardown.
// Raises ETaskException if the factory is already terminated.
function Run(Job: TProc): IMycSubscriber;
procedure EnqueueJob(const Job: TProc);
// Shuts down the task factory: terminates worker threads and cleans up resources.
// Before shutdown, any first stored exception from a worker thread (from Run or CreateThread tasks managed by this factory)
// will be re-raised in the calling (main) thread.
procedure Teardown;
// Waits for the operation associated with State to complete.
// Must not be called from a worker thread of this factory.
// After waiting, or if the state is already set, any first stored exception
// from any worker thread of this factory will be re-raised in the calling (main) thread.
// Raises ETaskException if called from within a worker thread.
procedure WaitFor(State: IMycState);
// Gets the number of worker threads.
property ThreadCount: Integer read GetThreadCount;
end;
TMycTaskFactory = class(TInterfacedObject, IMycTaskManager, IMycTaskFactory)
type
ETaskException = class(Exception) end;
private
@@ -67,6 +67,8 @@ type
FWorkStack: TMycAtomicStack<TProc>; // Stack of pending jobs
FWorkThreads: TArray<TThread>; // Array of worker threads
procedure WorkerThread;
class var
FTaskManagerLock: Integer;
protected
procedure ExecuteJob;
@@ -81,10 +83,14 @@ type
procedure HandleException;
procedure EnqueueJob(const Job: TProc);
function Run(Job: TProc): IMycSubscriber;
function CreateTask(const Gate: IMycState; const Proc: TProc): TMycSubscription;
procedure WaitFor(State: IMycState);
procedure Teardown;
function InMainThread: Boolean;
function InWorkerThread: Boolean;
class procedure AquireTaskManager;
class procedure ReleaseTaskManager;
end;
implementation
@@ -175,6 +181,20 @@ begin
Result.NameThreadForDebugging(DbgName); // Set thread name for debugging
end;
function TMycTaskFactory.CreateTask(const Gate: IMycState; const Proc: TProc):
TMycSubscription;
begin
if not Assigned(Gate) or Gate.IsSet then
begin
EnqueueJob( Proc );
end
else
begin
// The job will run when Gate notifies the subscriber returned by Run.
Result := Gate.Subscribe( Run( Proc ) );
end;
end;
function TMycTaskFactory.CreateThread(const Proc: TProc): IMycState;
var
res: IMycLatch;
@@ -261,6 +281,20 @@ begin
exit(false); // Not found in worker thread list
end;
class procedure TMycTaskFactory.AquireTaskManager;
begin
if not Assigned(TaskManager) then
TaskManager := TMycTaskFactory.Create;
inc( FTaskManagerLock );
end;
class procedure TMycTaskFactory.ReleaseTaskManager;
begin
dec( FTaskManagerLock );
if FTaskManagerLock = 0 then
TaskManager := nil;
end;
function TMycTaskFactory.Run(Job: TProc): IMycSubscriber;
begin
if FTerminated <> 0 then
@@ -361,7 +395,6 @@ end;
function TMycPendingJob.Notify: Boolean;
var
newCount: Integer;
P: TProc;
begin
PPointer(@P)^ := TInterlocked.Exchange( PPointer(@FJob)^, nil );
Src/Myc.Futures.pas
+36 -69
View File
@@ -4,7 +4,7 @@ interface
uses
System.SysUtils,
Myc.Signals, Myc.Core.Tasks;
Myc.Signals, Myc.TaskManager;
type
// Represents the eventual result of an asynchronous operation.
@@ -19,91 +19,58 @@ type
property Done: IMycState read GetDone;
end;
// Abstract base class for IMycFuture<T> implementations.
TMycFuture<T> = class abstract(TInterfacedObject, IMycFuture<T>)
protected
function GetResult: T; virtual; abstract;
function GetDone: IMycState; virtual; abstract;
end;
Future<T> = record
strict private
FFuture: IMycFuture<T>;
class constructor CreateClass;
class destructor DestroyClass;
constructor Create(const AFuture: IMycFuture<T>);
public
class function Construct(const Proc: TFunc<T>): Future<T>; overload; static;
class function Construct(const Gate: IMycState; const Proc: TFunc<T>): Future<T>; overload; static;
// Concrete future implementation triggered by an initial state, executing a TFunc<T>.
TMycInitStateFuncFuture<T> = class(TMycFuture<T>)
private
FInit: TMycSubscription;
FDone: IMycLatch;
FResult: T;
protected
function GetResult: T; override;
function GetDone: IMycState; override;
public
// Creates a future that executes AProc after AInitState is set, using ATaskFactory.
constructor Create(const ATaskFactory: IMycTaskFactory; const AInitState:
IMycState; AProc: TFunc<T>);
// Cleans up resources, primarily by unsubscribing from the initial state.
destructor Destroy; override;
function Chain<S>(const Proc: TFunc<T, S>): Future<S>;
end;
implementation
{ TMycInitStateFuncFuture<T> }
uses
Myc.Core.Futures, Myc.Core.Tasks;
constructor TMycInitStateFuncFuture<T>.Create(const ATaskFactory:
IMycTaskFactory; const AInitState: IMycState; AProc: TFunc<T>);
constructor Future<T>.Create(const AFuture: IMycFuture<T>);
begin
inherited Create;
Assert(Assigned(AInitState), 'AInitState must be assigned.');
FDone := TMycLatch.CreateLatch(1);
// Subscribe the job execution to AInitState.
// The job will run when AInitState notifies the subscriber returned by Run.
// AStartCount for Run is 1, meaning it waits for one Notify from AInitState's subscription.
FInit := AInitState.Subscribe(
ATaskFactory.Run(
procedure
begin
try
try
Self.FResult := AProc();
except
Self.FResult := Default(T); // Set result to Default(T) on error
raise; // Re-raise for TaskFactory to handle
end;
finally
Self.FDone.Notify; // Signal that this future is done (successfully or with error)
end;
end
)
);
FFuture := AFuture;
end;
destructor TMycInitStateFuncFuture<T>.Destroy;
class constructor Future<T>.CreateClass;
begin
Assert( FInit.State.IsSet );
// Explicitly unsubscribe from the initial state to ensure cleanup.
FInit.Unsubscribe;
inherited Destroy;
TMycTaskFactory.AquireTaskManager;
end;
function TMycInitStateFuncFuture<T>.GetDone: IMycState;
class destructor Future<T>.DestroyClass;
begin
Result := FDone.State;
TMycTaskFactory.ReleaseTaskManager;
end;
function TMycInitStateFuncFuture<T>.GetResult: T;
function Future<T>.Chain<S>(const Proc: TFunc<T, S>): Future<S>;
begin
if not FDone.State.IsSet then
begin
// Design decision: Caller must ensure future is done.
raise Exception.Create('Result is not yet available for the future.');
end;
var Cap := FFuture;
// Returns FResult. If AProc failed, FResult is Default(T) and the exception
// was raised to be handled by the TaskFactory.
Result := FResult;
Result := Future<S>.Construct( FFuture.Done,
function: S
begin
Result := Proc( Cap.Result );
end );
end;
class function Future<T>.Construct(const Proc: TFunc<T>): Future<T>;
begin
Result.Create( TMycInitStateFuncFuture<T>.Create( TaskManager, nil, Proc ) );
end;
class function Future<T>.Construct(const Gate: IMycState; const Proc: TFunc<T>): Future<T>;
begin
Result.Create( TMycInitStateFuncFuture<T>.Create( TaskManager, Gate, Proc ) );
end;
end.
Src/Myc.Signals.pas
+50 -48
View File
@@ -21,10 +21,11 @@ type
FState: IMycState;
FTag: TMycNotifyList<IMycSubscriber>.TTag; // Tag identifying this subscription within the notifier list.
function GetState: IMycState;
constructor Create(const AState: IMycState; ATag: TMycNotifyList<IMycSubscriber>.TTag);
public
constructor Create(const AState: IMycState; ATag: TMycNotifyList<IMycSubscriber>.TTag);
// Unsubscribes from the associated signal.
procedure Unsubscribe;
class operator Initialize(out Dest: TMycSubscription);
property State: IMycState read GetState;
end;
@@ -154,44 +155,44 @@ uses
System.SyncObjs; // For TInterlocked
type
// TMycNullState implements a "null object" pattern for IMycState.
// This state is always considered "set".
TMycNullState = class(TInterfacedObject, IMycState)
protected
// IMycState
function GetIsSet: Boolean;
function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription;
procedure Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
public
constructor Create;
end;
{ TMycNullState }
constructor TMycNullState.Create;
begin
inherited Create;
end;
function TMycNullState.GetIsSet: Boolean;
begin
Result := true; // Null state is always set.
end;
function TMycNullState.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription;
begin
// Since the state is always set, notify the subscriber immediately.
if Assigned(Subscriber) then
begin
Subscriber.Notify;
end;
// No ongoing subscription is necessary or meaningful for a null state.
Result.Create(Self, nil); // Return an empty subscription.
end;
procedure TMycNullState.Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
begin
// Unsubscribing from a null state has no effect.
// TMycNullState implements a "null object" pattern for IMycState.
// This state is always considered "set".
TMycNullState = class(TInterfacedObject, IMycState)
protected
// IMycState
function GetIsSet: Boolean;
function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription;
procedure Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
public
constructor Create;
end;
{ TMycNullState }
constructor TMycNullState.Create;
begin
inherited Create;
end;
function TMycNullState.GetIsSet: Boolean;
begin
Result := true; // Null state is always set.
end;
function TMycNullState.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription;
begin
// Since the state is always set, notify the subscriber immediately.
if Assigned(Subscriber) then
begin
Subscriber.Notify;
end;
// No ongoing subscription is necessary or meaningful for a null state.
Result.Create(Self, nil); // Return an empty subscription.
end;
procedure TMycNullState.Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
begin
// Unsubscribing from a null state has no effect.
end;
{ TMycSubscription }
@@ -206,17 +207,20 @@ end;
procedure TMycSubscription.Unsubscribe;
begin
if Assigned(FState) then
begin
FState.Unsubscribe(FTag);
FTag := nil;
FState := nil;
end;
FState.Unsubscribe(FTag);
FState := TMycState.Null;
FTag := nil;
end;
function TMycSubscription.GetState: IMycState;
begin
Result := FState;
Result := FState
end;
class operator TMycSubscription.Initialize(out Dest: TMycSubscription);
begin
Dest.FState := TMycState.Null;
Dest.FTag := nil;
end;
{ TMycLatch }
@@ -301,7 +305,6 @@ function TMycLatch.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription
var
alreadySet: Boolean;
begin
Result.Create(TMycState.Null, nil); // Default to an empty subscription.
if not Assigned(Subscriber) then
exit;
@@ -409,7 +412,6 @@ end;
function TMycDirty.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription;
begin
Result.Create(TMycState.Null, nil); // Default to an empty subscription
if not Assigned(Subscriber) then
exit;
Src/Myc.TaskManager.pas
+19
View File
@@ -0,0 +1,19 @@
unit Myc.TaskManager;
interface
uses
System.SysUtils,
Myc.Signals;
type
IMycTaskManager = interface
function CreateTask( const Gate: IMycState; const Proc: TProc ): TMycSubscription;
end;
var
TaskManager: IMycTaskManager;
implementation
end.