Future<T> helper record
This commit is contained in:
Michael Schimmel
@@ -0,0 +1,362 @@
|
||||
unit Myc.Core.Signals;
|
||||
|
||||
interface
|
||||
|
||||
uses
|
||||
System.SysUtils,
|
||||
Myc.Core.Notifier,
|
||||
Myc.Signals;
|
||||
|
||||
type
|
||||
TMycState = class abstract( TInterfacedObject, IMycState )
|
||||
strict private
|
||||
class var
|
||||
FNull: IMycState;
|
||||
class constructor ClassCreate;
|
||||
protected
|
||||
function GetIsSet: Boolean; virtual; abstract;
|
||||
public
|
||||
function Subscribe(Subscriber: IMycSubscriber): TMycSubscription; virtual; abstract;
|
||||
procedure Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag); virtual; abstract;
|
||||
property IsSet: Boolean read GetIsSet;
|
||||
// Provides access to the singleton null latch instance (always set).
|
||||
class property Null: IMycState read FNull;
|
||||
end;
|
||||
|
||||
// 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(Subscriber: IMycSubscriber): TMycSubscription;
|
||||
procedure Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
|
||||
public
|
||||
constructor Create;
|
||||
end;
|
||||
|
||||
// TMycLatch implements a countdown latch.
|
||||
// It is initialized with a count. Calls to its Notify method (as an IMycSubscriber)
|
||||
// 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(TMycState, IMycLatch)
|
||||
strict private
|
||||
FSubscribers: TMycNotifyList<IMycSubscriber>; // List of subscribers waiting for this latch to be set.
|
||||
class var
|
||||
FNull: IMycLatch; // Singleton instance of a latch that is always set.
|
||||
class constructor ClassCreate; // Class constructor to initialize FNull.
|
||||
private
|
||||
[volatile] FCount: Integer; // The internal countdown value for the latch.
|
||||
function GetState: IMycState; // Implementation for IMycLatch.GetState and IMycFlag.GetState.
|
||||
protected
|
||||
function GetIsSet: Boolean; override; final; // Implementation for IMycState.GetIsSet.
|
||||
public
|
||||
// Creates the latch with an initial count.
|
||||
// If ACount is 0 or less, the latch is effectively pre-set (delegates to FNull via CreateLatch factory).
|
||||
constructor Create(ACount: Integer);
|
||||
destructor Destroy; override;
|
||||
|
||||
// Factory method: creates a new countdown latch or returns the Null latch if Count <= 0.
|
||||
class function CreateLatch(Count: Integer): IMycLatch; static;
|
||||
|
||||
// IMycSignal implementation
|
||||
function Subscribe(Subscriber: IMycSubscriber): TMycSubscription; override;
|
||||
procedure Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag); override;
|
||||
|
||||
// IMycSubscriber implementation for TMycLatch itself.
|
||||
// Decrements the internal count. If the count reaches zero,
|
||||
// registered subscribers are notified, and the latch becomes permanently set.
|
||||
function Notify: Boolean;
|
||||
|
||||
// Provides access to the singleton null latch instance (always set).
|
||||
class property Null: IMycLatch read FNull;
|
||||
end;
|
||||
|
||||
// TMycDirty 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.
|
||||
TMycDirty = class(TMycState, IMycDirty)
|
||||
strict private
|
||||
FSubscribers: TMycNotifyList<IMycSubscriber>; // List of subscribers interested in state changes of this dirty flag.
|
||||
private
|
||||
[volatile] FFlag: Boolean; // Internal state: true if dirty/set, false if clean/reset.
|
||||
function GetState: IMycState; // Implementation for IMycFlag.GetState.
|
||||
protected
|
||||
function GetIsSet: Boolean; override; final; // Implementation for IMycState.GetIsSet (true if dirty).
|
||||
public
|
||||
// Creates a new dirty flag, initially set to dirty (true).
|
||||
constructor Create;
|
||||
destructor Destroy; override;
|
||||
|
||||
// Factory method to create a new TMycDirty instance.
|
||||
class function CreateDirty: IMycDirty; static;
|
||||
|
||||
// IMycSignal implementation
|
||||
function Subscribe(Subscriber: IMycSubscriber): TMycSubscription; override;
|
||||
procedure Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag); override;
|
||||
|
||||
// IMycSubscriber implementation for TMycDirty.
|
||||
// Sets this flag to dirty (true). Notifies subscribers if it transitioned from clean to dirty.
|
||||
function Notify: Boolean;
|
||||
|
||||
// IMycDirty implementation. Resets the flag to clean (false).
|
||||
function Reset: Boolean;
|
||||
end;
|
||||
|
||||
implementation
|
||||
|
||||
uses
|
||||
System.SyncObjs; // For TInterlocked
|
||||
|
||||
class constructor TMycState.ClassCreate;
|
||||
begin
|
||||
// Create a singleton null latch instance that is initially (and always) set.
|
||||
FNull := TMycNullState.Create(); // Calls the instance constructor
|
||||
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(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;
|
||||
|
||||
{ TMycLatch }
|
||||
|
||||
constructor TMycLatch.Create(ACount: Integer);
|
||||
begin
|
||||
inherited Create;
|
||||
Assert( ACount >= 0 );
|
||||
FCount := ACount;
|
||||
FSubscribers.Create;
|
||||
end;
|
||||
|
||||
class constructor TMycLatch.ClassCreate;
|
||||
begin
|
||||
// Create a singleton null latch instance that is initially (and always) set.
|
||||
FNull := TMycLatch.Create(0); // Calls the instance constructor
|
||||
end;
|
||||
|
||||
destructor TMycLatch.Destroy;
|
||||
begin
|
||||
FSubscribers.Destroy;
|
||||
inherited;
|
||||
end;
|
||||
|
||||
class function TMycLatch.CreateLatch(Count: Integer): IMycLatch;
|
||||
begin
|
||||
// Factory method: returns a new latch or the shared Null latch if Count <= 0.
|
||||
if Count > 0 then
|
||||
Result := TMycLatch.Create(Count) // Instance constructor
|
||||
else
|
||||
Result := FNull;
|
||||
end;
|
||||
|
||||
function TMycLatch.Notify: Boolean;
|
||||
var
|
||||
currentCountAfterDecrement: Integer;
|
||||
shouldNotifySubscribers: Boolean;
|
||||
begin
|
||||
FSubscribers.Lock;
|
||||
try
|
||||
currentCountAfterDecrement := TInterlocked.Decrement(FCount);
|
||||
|
||||
if currentCountAfterDecrement < -MaxInt div 2 then
|
||||
TInterlocked.Increment(FCount); // Defend against extreme underflow.
|
||||
|
||||
shouldNotifySubscribers := (currentCountAfterDecrement = 0); // Notify only on transition to zero.
|
||||
|
||||
if shouldNotifySubscribers then
|
||||
begin
|
||||
FSubscribers.Notify(
|
||||
function(Subscriber: IMycSubscriber): Boolean
|
||||
begin
|
||||
Result := Subscriber.Notify;
|
||||
end);
|
||||
end;
|
||||
|
||||
// Returns true if the latch has not yet been set by this Notify call (count > 0).
|
||||
Result := currentCountAfterDecrement > 0;
|
||||
|
||||
// If the latch is now set (or was already set), unadvise all subscribers.
|
||||
// This ensures one-shot notification for the current set of subscribers.
|
||||
if currentCountAfterDecrement <= 0 then
|
||||
FSubscribers.UnadviseAll;
|
||||
finally
|
||||
FSubscribers.Release;
|
||||
end;
|
||||
end;
|
||||
|
||||
function TMycLatch.GetIsSet: Boolean;
|
||||
begin
|
||||
// The latch is set if its count has reached zero or less.
|
||||
Result := FCount <= 0;
|
||||
end;
|
||||
|
||||
function TMycLatch.GetState: IMycState;
|
||||
begin
|
||||
// TMycLatch itself implements IMycState.
|
||||
exit(Self);
|
||||
end;
|
||||
|
||||
function TMycLatch.Subscribe(Subscriber: IMycSubscriber): TMycSubscription;
|
||||
var
|
||||
alreadySet: Boolean;
|
||||
begin
|
||||
if not Assigned(Subscriber) then
|
||||
exit;
|
||||
|
||||
FSubscribers.Lock;
|
||||
try
|
||||
alreadySet := (FCount <= 0); // Check if latch is already set.
|
||||
|
||||
if alreadySet then
|
||||
begin
|
||||
// If already set, notify immediately; no persistent subscription needed.
|
||||
Subscriber.Notify;
|
||||
end
|
||||
else
|
||||
begin
|
||||
// If not yet set, advise the subscriber.
|
||||
Result.Create(Self, FSubscribers.Advise(Subscriber));
|
||||
end;
|
||||
finally
|
||||
FSubscribers.Release;
|
||||
end;
|
||||
end;
|
||||
|
||||
procedure TMycLatch.Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
|
||||
begin
|
||||
if Tag = nil then
|
||||
exit;
|
||||
|
||||
FSubscribers.Lock;
|
||||
try
|
||||
FSubscribers.Unadvise(Tag);
|
||||
finally
|
||||
FSubscribers.Release;
|
||||
end;
|
||||
end;
|
||||
|
||||
{ TMycDirty }
|
||||
|
||||
constructor TMycDirty.Create;
|
||||
begin
|
||||
inherited Create;
|
||||
FFlag := true; // A new dirty flag is initially considered dirty.
|
||||
FSubscribers.Create;
|
||||
end;
|
||||
|
||||
class function TMycDirty.CreateDirty: IMycDirty;
|
||||
begin
|
||||
// Factory method to create a new TMycDirty instance.
|
||||
Result := TMycDirty.Create;
|
||||
end;
|
||||
|
||||
destructor TMycDirty.Destroy;
|
||||
begin
|
||||
FSubscribers.Destroy; // Clean up the subscriber list.
|
||||
inherited;
|
||||
end;
|
||||
|
||||
function TMycDirty.Reset: Boolean;
|
||||
begin
|
||||
// Sets the flag to false (clean) and returns true if it was previously true (dirty).
|
||||
Result := TInterlocked.Exchange(FFlag, false);
|
||||
end;
|
||||
|
||||
function TMycDirty.GetIsSet: Boolean;
|
||||
begin
|
||||
// IsSet is true if the flag is dirty (FFlag = true).
|
||||
Result := FFlag;
|
||||
end;
|
||||
|
||||
function TMycDirty.GetState: IMycState;
|
||||
begin
|
||||
// TMycDirty itself implements IMycState.
|
||||
exit(Self);
|
||||
end;
|
||||
|
||||
function TMycDirty.Notify: Boolean;
|
||||
var
|
||||
wasPreviouslyClean: Boolean;
|
||||
begin
|
||||
FSubscribers.Lock;
|
||||
try
|
||||
// Set the flag to true (dirty) and check if it was previously false (clean).
|
||||
wasPreviouslyClean := not TInterlocked.Exchange(FFlag, true);
|
||||
|
||||
if wasPreviouslyClean then // Only notify subscribers if state changed from clean to dirty.
|
||||
begin
|
||||
FSubscribers.Notify(
|
||||
function(Subscriber: IMycSubscriber): Boolean
|
||||
begin
|
||||
Result := Subscriber.Notify;
|
||||
end);
|
||||
end;
|
||||
// The return value of this Notify (as an IMycSubscriber) indicates if this
|
||||
// TMycDirty instance itself would want more notifications if it were subscribed to something.
|
||||
// Here, it reflects whether a state change to dirty occurred due to this call.
|
||||
Result := wasPreviouslyClean;
|
||||
finally
|
||||
FSubscribers.Release;
|
||||
end;
|
||||
end;
|
||||
|
||||
function TMycDirty.Subscribe(Subscriber: IMycSubscriber): TMycSubscription;
|
||||
begin
|
||||
if not Assigned(Subscriber) then
|
||||
exit;
|
||||
|
||||
// For TMycDirty, we always add the subscriber and then check if we need to notify immediately.
|
||||
// Ref counting for the subscriber interface is assumed to be handled by TMycNotifyList.Advise/Unadvise.
|
||||
// Or, if explicit management is needed like in TMycLatch, it should be added.
|
||||
// For consistency with TMycLatch, let's add AddRef/Release here too.
|
||||
FSubscribers.Lock;
|
||||
try
|
||||
// Add subscriber regardless of current state.
|
||||
Result.Create(Self, FSubscribers.Advise(Subscriber));
|
||||
if FFlag then // If currently dirty, notify the new subscriber immediately.
|
||||
begin
|
||||
Subscriber.Notify;
|
||||
end;
|
||||
finally
|
||||
FSubscribers.Release;
|
||||
end;
|
||||
end;
|
||||
|
||||
procedure TMycDirty.Unsubscribe(Tag: TMycNotifyList<IMycSubscriber>.TTag);
|
||||
begin
|
||||
if Tag = nil then
|
||||
exit;
|
||||
|
||||
FSubscribers.Lock;
|
||||
try
|
||||
FSubscribers.Unadvise(Tag);
|
||||
finally
|
||||
FSubscribers.Release;
|
||||
end;
|
||||
end;
|
||||
|
||||
end.
|
||||
@@ -41,13 +41,6 @@ type
|
||||
// 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;
|
||||
@@ -89,6 +82,7 @@ type
|
||||
function InMainThread: Boolean;
|
||||
function InWorkerThread: Boolean;
|
||||
|
||||
// Initialize global TaskManager
|
||||
class procedure AquireTaskManager;
|
||||
class procedure ReleaseTaskManager;
|
||||
end;
|
||||
|
||||
+48
-10
@@ -22,14 +22,26 @@ type
|
||||
Future<T> = record
|
||||
strict private
|
||||
FFuture: IMycFuture<T>;
|
||||
function GetDone: IMycState; inline;
|
||||
function GetResult: T; inline;
|
||||
|
||||
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;
|
||||
|
||||
function Chain<S>(const Proc: TFunc<T, S>): Future<S>;
|
||||
public
|
||||
constructor Create(const AFuture: IMycFuture<T>);
|
||||
class operator Implicit(const A: IMycFuture<T>): Future<T>; overload;
|
||||
class operator Implicit(const A: Future<T>): IMycFuture<T>; overload;
|
||||
|
||||
class function Construct(const Proc: TFunc<T>): IMycFuture<T>; overload; static;
|
||||
class function Construct(const Gate: IMycState; const Proc: TFunc<T>): IMycFuture<T>; overload; static;
|
||||
|
||||
function Chain<S>(const Proc: TFunc<T, S>): IMycFuture<S>;
|
||||
|
||||
function WaitFor: T;
|
||||
|
||||
property Done: IMycState read GetDone;
|
||||
property Result: T read GetResult;
|
||||
end;
|
||||
|
||||
implementation
|
||||
@@ -52,7 +64,7 @@ begin
|
||||
TMycTaskFactory.ReleaseTaskManager;
|
||||
end;
|
||||
|
||||
function Future<T>.Chain<S>(const Proc: TFunc<T, S>): Future<S>;
|
||||
function Future<T>.Chain<S>(const Proc: TFunc<T, S>): IMycFuture<S>;
|
||||
begin
|
||||
var Cap := FFuture;
|
||||
|
||||
@@ -63,14 +75,40 @@ begin
|
||||
end );
|
||||
end;
|
||||
|
||||
class function Future<T>.Construct(const Proc: TFunc<T>): Future<T>;
|
||||
class function Future<T>.Construct(const Proc: TFunc<T>): IMycFuture<T>;
|
||||
begin
|
||||
Result.Create( TMycInitStateFuncFuture<T>.Create( TaskManager, nil, Proc ) );
|
||||
Result := TMycInitStateFuncFuture<T>.Create( TaskManager, nil, Proc );
|
||||
end;
|
||||
|
||||
class function Future<T>.Construct(const Gate: IMycState; const Proc: TFunc<T>): Future<T>;
|
||||
class function Future<T>.Construct(const Gate: IMycState; const Proc: TFunc<T>): IMycFuture<T>;
|
||||
begin
|
||||
Result.Create( TMycInitStateFuncFuture<T>.Create( TaskManager, Gate, Proc ) );
|
||||
Result := TMycInitStateFuncFuture<T>.Create( TaskManager, Gate, Proc );
|
||||
end;
|
||||
|
||||
function Future<T>.GetDone: IMycState;
|
||||
begin
|
||||
Result := FFuture.Done;
|
||||
end;
|
||||
|
||||
function Future<T>.GetResult: T;
|
||||
begin
|
||||
Result := FFuture.Result;
|
||||
end;
|
||||
|
||||
function Future<T>.WaitFor: T;
|
||||
begin
|
||||
TaskManager.WaitFor( FFuture.Done );
|
||||
Result := FFuture.Result;
|
||||
end;
|
||||
|
||||
class operator Future<T>.Implicit(const A: IMycFuture<T>): Future<T>;
|
||||
begin
|
||||
Result.Create( A );
|
||||
end;
|
||||
|
||||
class operator Future<T>.Implicit(const A: Future<T>): IMycFuture<T>;
|
||||
begin
|
||||
Result := A.FFuture;
|
||||
end;
|
||||
|
||||
end.
|
||||
|
||||
@@ -9,6 +9,13 @@ uses
|
||||
type
|
||||
IMycTaskManager = interface
|
||||
function CreateTask( const Gate: IMycState; const Proc: TProc ): TMycSubscription;
|
||||
|
||||
// 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);
|
||||
end;
|
||||
|
||||
var
|
||||
|
||||
Reference in New Issue
Block a user