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

Notify list refactoring

Browse Source
This commit is contained in:
Michael Schimmel
2025-07-15 09:46:16 +02:00
parent ce0cba720a
commit 1a07468ad8
4 changed files with 325 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Core.Notifier.pas
+99 -59
View File
@@ -19,8 +19,15 @@ type
PItem = ^TItem;
// Internal structure for storing a receiver and list linkage.
TItem = record
Next, Prev: PItem; // Pointers to the next and previous items in the doubly linked list.
private
FNext: PItem;
FPrev: PItem;
function GetNext: PItem; inline;
public
Receiver: T; // The registered interface instance (the event sink).
property Next: PItem read GetNext;
property Prev: PItem read FPrev;
end;
// Notify function. If this results false, it will be removed from the notify list
@@ -31,6 +38,9 @@ type
// Bit 0 of the address stores the lock state (0 = locked, 1 = unlocked).
[volatile]
FList: PItem;
class var
FReverseOnNotify: Boolean;
class constructor CreateClass;
// Allocates memory for a new TItem.
class function AllocItem: PItem; static; inline;
@@ -62,10 +72,18 @@ type
// by setting its interface reference to nil. The list item itself is not freed here.
procedure Notify(const Func: TNotifyProc);
property First: PItem read GetFirst;
// If this is true, the list is reversed after each Notify, hoping for a better distrubution of events.
class property ReverseOnNotify: Boolean read FReverseOnNotify write FReverseOnNotify;
end;
implementation
class constructor TMycNotifyList<T>.CreateClass;
begin
FReverseOnNotify := true;
end;
class operator TMycNotifyList<T>.Initialize(out Dest: TMycNotifyList<T>);
begin
NativeUInt(Dest.FList) := 1;
@@ -88,10 +106,10 @@ begin
Item := AllocItem;
Item.Receiver := Receiver;
Item.Prev := nil;
Item.Next := FList;
if Item.Next <> nil then
Item.Next.Prev := Item;
Item.FPrev := nil;
Item.FNext := FList;
if Item.FNext <> nil then
Item.FNext.FPrev := Item;
FList := Item;
@@ -133,108 +151,123 @@ var
begin
Assert(IsLocked);
var rev: PItem := nil;
var tmp: PItem := nil;
// Invariant: Append all detached items at end of the list. We can't simply free them, because the subscription is owned by the client.
if FReverseOnNotify then
begin
// Method: stack assigned items and rebuild the list from the stack (reversing the order)
var stack: PItem := nil;
var released: PItem := nil;
while (FList <> nil) and Assigned(FList.Receiver) do
begin
Item := FList;
FList := Item.Next;
FList := Item.FNext;
if Func(Item.Receiver) then
begin
Item.Next := rev;
rev := Item;
// item stays valid
Item.FNext := stack;
stack := Item;
end
else
begin
// release the receiver
Item.Receiver := nil;
Item.Next := tmp;
tmp := Item;
Item.FNext := released;
released := Item;
end;
end;
while tmp <> nil do
// The list now only contains old released items. Push the newly released ones.
while released <> nil do
begin
Item := tmp;
tmp := Item.Next;
Item := released;
released := Item.FNext;
Item.Prev := nil;
Item.Next := FList;
Item.FPrev := nil;
Item.FNext := FList;
FList := Item;
if Item.Next <> nil then
Item.Next.Prev := Item;
if Item.FNext <> nil then
Item.FNext.FPrev := Item;
end;
while rev <> nil do
// Now push the valid items, so that they are at the beginning of the list.
while stack <> nil do
begin
Item := rev;
rev := Item.Next;
Item := stack;
stack := Item.FNext;
Item.Prev := nil;
Item.Next := FList;
Item.FPrev := nil;
Item.FNext := FList;
FList := Item;
if Item.Next <> nil then
Item.Next.Prev := Item;
if Item.FNext <> nil then
Item.FNext.FPrev := Item;
end;
{
tmp := nil;
Last := nil;
end
else
begin
// Method: filter released items and add then to the end of the list
var released: PItem := nil;
var lastValid: PItem := nil;
Item := FList;
while (Item <> nil) and Assigned(Item.Receiver) do
begin
if not Func(Item.Receiver) then
begin
// Receiver wants no more notifications, detach it
// item is now invalid
if Item = FList then
FList := Item.Next;
FList := Item.FNext;
if Item.Prev <> nil then
Item.Prev.Next := Item.Next;
if Item.Next <> nil then
Item.Next.Prev := Item.Prev;
if Item.FPrev <> nil then
Item.FPrev.FNext := Item.FNext;
if Item.FNext <> nil then
Item.FNext.FPrev := Item.FPrev;
// release the receiver
Item.Receiver := nil;
// and save the list item in a tmp list for later use
var nxt := Item.Next;
Item.Next := tmp;
tmp := Item;
var nxt := Item.FNext;
Item.FNext := released;
released := Item;
Item := nxt;
end
else
begin
Last := Item;
Item := Last.Next;
// save a pointer to the last valid item
lastValid := Item;
Item := lastValid.FNext;
end;
end;
// Append all detached items at end of the list. We can't simply free them, because the subscription is owned by the client.
while tmp <> nil do
// insert all new released items after the last valid item
while released <> nil do
begin
Item := tmp;
tmp := Item.Next;
Item := released;
released := Item.FNext;
if Last = nil then
if lastValid = nil then
begin
Item.Prev := nil;
Item.Next := FList;
Item.FPrev := nil;
Item.FNext := FList;
FList := Item;
end
else
begin
Item.Prev := Last;
Item.Next := Last.Next;
if Item.Prev <> nil then
Item.Prev.Next := Item;
Item.FPrev := lastValid;
Item.FNext := lastValid.Next;
if Item.FPrev <> nil then
Item.FPrev.FNext := Item;
end;
if Item.FNext <> nil then
Item.FNext.FPrev := Item;
end;
if Item.Next <> nil then
Item.Next.Prev := Item;
end;
}
end;
procedure TMycNotifyList<T>.Release;
@@ -270,12 +303,12 @@ begin
Item.Receiver := nil;
if Item = FList then
FList := Item.Next;
FList := Item.FNext;
if Item.Prev <> nil then
Item.Prev.Next := Item.Next;
if Item.Next <> nil then
Item.Next.Prev := Item.Prev;
if Item.FPrev <> nil then
Item.FPrev.FNext := Item.FNext;
if Item.FNext <> nil then
Item.FNext.FPrev := Item.FPrev;
FreeItem(Item);
end;
@@ -287,4 +320,11 @@ begin
Unadvise(TTag(FList));
end;
function TMycNotifyList<T>.TItem.GetNext: PItem;
begin
if not Assigned(Receiver) then
exit(nil);
Result := FNext;
end;
end.
Src/Myc.Fmx.Chart.Series.pas
+1
View File
@@ -11,6 +11,7 @@ uses
Myc.Trade.Types,
Myc.Trade.DataArray,
Myc.Trade.DataPoint,
Myc.Trade.DataPoint.Impl,
Myc.Fmx.Chart;
type
Src/Myc.Trade.DataPoint.Impl.pas
+182 -21
View File
@@ -3,12 +3,57 @@ unit Myc.Trade.DataPoint.Impl;
interface
uses
System.SysUtils,
Myc.Signals,
Myc.Trade.Types,
Myc.Trade.DataPoint;
Myc.Trade.DataPoint,
Myc.Core.Notifier;
type
// Null object implementation for IMycConverter
// Abstract base class for data consumers.
TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
protected
function ProcessData(const Value: T): TState; virtual; abstract;
end;
// Concrete data provider that manages a list of processors (listeners).
TMycDataProvider<T> = class abstract(TContainedObject, TDataProvider<T>.IDataProvider)
private
FListeners: TMycNotifyList<IMycProcessor<T>>;
public
constructor Create(const Controller: IInterface);
destructor Destroy; override;
// Notifies all linked processors.
function Broadcast(const Value: T): TState;
// Link a Processor
function Link(const Processor: IMycProcessor<T>): TDataProvider<T>.TTag;
// Unlink a linked strategy
procedure Unlink(Tag: TDataProvider<T>.TTag);
end;
// Null object implementation for IDataProvider.
TNullDataProvider<T> = class(TInterfacedObject, TDataProvider<T>.IDataProvider)
public
function Link(const Receiver: IMycProcessor<T>): TDataProvider<T>.TTag;
procedure Unlink(Tag: TDataProvider<T>.TTag);
end;
// Abstract base class for components that process data of type S and provide data of type T.
TMycConverter<S, T> = class abstract(TMycProcessor<S>, TConverter<S, T>.IConverter)
private
FSender: TMycDataProvider<T>;
function GetSender: TDataProvider<T>.IDataProvider;
protected
function ProcessData(const Value: S): TState; override; abstract;
// Broadcasts the given data to all linked processors.
function Broadcast(const Value: T): TState;
public
constructor Create;
destructor Destroy; override;
property Sender: TDataProvider<T>.IDataProvider read GetSender;
end;
// Null object implementation for IConverter.
TNullConverter<S, T> = class(TInterfacedObject, TConverter<S, T>.IConverter)
private
function GetSender: TDataProvider<T>.IDataProvider;
@@ -16,22 +61,7 @@ type
function ProcessData(const Value: S): TState;
end;
TMycConverter<S, T> = class abstract(TMycProcessor<S>, TConverter<S, T>.IConverter)
private
FSender: TMycDataProvider<T>;
function GetSender: TDataProvider<T>.IDataProvider;
protected
function ProcessData(const Value: S): TState; override; abstract;
// Broadcasts the given data to all linked processors.
function Broadcast(const Value: T): TState;
public
constructor Create;
destructor Destroy; override;
property Sender: TDataProvider<T>.IDataProvider read GetSender;
end;
// A generic converter that uses a function reference for the conversion logic.
TMycGenericConverter<S, T> = class(TMycConverter<S, T>)
private
FFunc: TConstFunc<S, T>;
@@ -41,12 +71,14 @@ type
constructor Create(const AFunc: TConstFunc<S, T>);
end;
// A converter specialized for calculating indicators.
TMycIndicator<S, T> = class(TMycConverter<S, T>)
protected
function ProcessData(const Value: S): TState; override; final;
function Calculate(const Value: S): T; virtual; abstract;
end;
// A converter that counts incoming data points and outputs the current count.
TMycDataCounter<T> = class(TMycConverter<T, Int64>)
private
FCount: Int64;
@@ -56,27 +88,124 @@ type
constructor Create;
end;
// A converter that takes an array and broadcasts each element individually.
TMycTicker<T> = class(TMycConverter<TArray<T>, T>)
public
function ProcessData(const Values: TArray<T>): TState; override;
end;
// A converter that reads a specific field from a record using RTTI.
TMycRecordFieldReader<S, T> = class(TMycConverter<S, T>)
private
FOffset: Integer;
public
constructor Create(const AFieldName: String);
function ProcessData(const Values: S): TState; override;
end;
// A generic processor implementation that uses a function reference.
TMycGenericProcessor<T> = class(TMycProcessor<T>)
type
TProc = reference to function(const Value: T): TState;
private
FProc: TProc;
protected
function ProcessData(const Value: T): TState; override; final;
public
constructor Create(const AProc: TProc);
end;
// A processor implementation that is owned by a controller.
TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
type
TProc = function(const Value: T): TState of object;
private
FProc: TProc;
function ProcessData(const Value: T): TState;
public
constructor Create(const Controller: IInterface; const AProc: TProc);
end;
implementation
uses
System.TypInfo,
System.SysUtils,
System.RTTI;
{ TMycDataProvider<T> }
constructor TMycDataProvider<T>.Create(const Controller: IInterface);
begin
inherited Create(Controller);
end;
destructor TMycDataProvider<T>.Destroy;
begin
FListeners.Finalize;
inherited Destroy;
end;
function TMycDataProvider<T>.Broadcast(const Value: T): TState;
begin
FListeners.Lock;
try
var item := FListeners.First;
if item = nil then
exit(TState.Null);
var i := 1;
while item.Next <> nil do
begin
inc(i);
item := item.Next;
end;
var done := TLatch.CreateLatch(i);
while item <> nil do
begin
item.Receiver.ProcessData(Value).Signal.Subscribe(done);
item := item.Prev;
end;
Result := done.State;
finally
FListeners.Release;
end;
end;
function TMycDataProvider<T>.Link(const Processor: IMycProcessor<T>): TDataProvider<T>.TTag;
begin
// Add the Processor to the notification list
FListeners.Lock;
try
Result := FListeners.Advise(Processor);
finally
FListeners.Release;
end;
end;
procedure TMycDataProvider<T>.Unlink(Tag: TDataProvider<T>.TTag);
begin
FListeners.Lock;
try
FListeners.Unadvise(Tag);
finally
FListeners.Release;
end;
end;
{ TNullDataProvider<T> }
function TNullDataProvider<T>.Link(const Receiver: IMycProcessor<T>): TDataProvider<T>.TTag;
begin
Result := nil;
end;
procedure TNullDataProvider<T>.Unlink(Tag: TDataProvider<T>.TTag);
begin
// Do nothing in the null implementation.
end;
{ TMycConverter<S, T> }
constructor TMycConverter<S, T>.Create;
@@ -101,7 +230,7 @@ begin
Result := FSender;
end;
{ TNullConverter }
{ TNullConverter<S, T> }
function TNullConverter<S, T>.GetSender: TDataProvider<T>.IDataProvider;
begin
@@ -133,6 +262,8 @@ begin
Result := Broadcast(Calculate(Value));
end;
{ TMycDataCounter<T> }
constructor TMycDataCounter<T>.Create;
begin
inherited Create;
@@ -145,6 +276,8 @@ begin
inc(FCount);
end;
{ TMycTicker<T> }
function TMycTicker<T>.ProcessData(const Values: TArray<T>): TState;
begin
var done := TLatch.CreateLatch(Length(Values));
@@ -156,6 +289,8 @@ begin
Result := done.State;
end;
{ TMycRecordFieldReader<S, T> }
constructor TMycRecordFieldReader<S, T>.Create(const AFieldName: String);
begin
inherited Create;
@@ -193,4 +328,30 @@ begin
Result := Broadcast(PT(fieldPtr)^);
end;
{ TMycGenericProcessor<T> }
constructor TMycGenericProcessor<T>.Create(const AProc: TProc);
begin
inherited Create;
FProc := AProc;
end;
function TMycGenericProcessor<T>.ProcessData(const Value: T): TState;
begin
Result := FProc(Value);
end;
{ TMycContainedProcessor<T> }
constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TProc);
begin
inherited Create(Controller);
FProc := AProc;
end;
function TMycContainedProcessor<T>.ProcessData(const Value: T): TState;
begin
Result := FProc(Value);
end;
end.
Src/Myc.Trade.DataPoint.pas
+1 -159
View File
@@ -4,8 +4,7 @@ interface
uses
Myc.Signals,
Myc.Trade.Types,
Myc.Core.Notifier;
Myc.Trade.Types;
type
// Represents a time-stamped data point in a series.
@@ -19,11 +18,6 @@ type
function ProcessData(const Value: T): TState;
end;
TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
protected
function ProcessData(const Value: T): TState; virtual; abstract;
end;
TDataProvider<T> = record
type
TTag = Pointer;
@@ -56,48 +50,6 @@ type
class property Null: IDataProvider read FNull;
end;
TMycDataProvider<T> = class abstract(TContainedObject, TDataProvider<T>.IDataProvider)
private
FListeners: TMycNotifyList<IMycProcessor<T>>;
public
constructor Create(const Controller: IInterface);
destructor Destroy; override;
// Notifies all linked processors.
function Broadcast(const Value: T): TState;
procedure Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
// Link a Processor
function Link(const Processor: IMycProcessor<T>): TDataProvider<T>.TTag;
// Unlink a linked strategy
procedure Unlink(Tag: TDataProvider<T>.TTag);
end;
TMycGenericProcessor<T> = class(TMycProcessor<T>)
type
TProc = reference to function(const Value: T): TState;
private
FProc: TProc;
protected
function ProcessData(const Value: T): TState; override; final;
public
constructor Create(const AProc: TProc);
end;
TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
type
TProc = function(const Value: T): TState of object;
private
FProc: TProc;
function ProcessData(const Value: T): TState;
public
constructor Create(const Controller: IInterface; const AProc: TProc);
end;
TNullDataProvider<T> = class(TInterfacedObject, TDataProvider<T>.IDataProvider)
public
function Link(const Receiver: IMycProcessor<T>): TDataProvider<T>.TTag;
procedure Unlink(Tag: TDataProvider<T>.TTag);
end;
// Interface helper for IMycConverter<S,T> providing the null object pattern.
TConverter<S, T> = record
type
@@ -150,18 +102,6 @@ implementation
uses
Myc.Trade.DataPoint.Impl;
{ TNullDataProvider }
function TNullDataProvider<T>.Link(const Receiver: IMycProcessor<T>): TDataProvider<T>.TTag;
begin
Result := nil;
end;
procedure TNullDataProvider<T>.Unlink(Tag: TDataProvider<T>.TTag);
begin
// Do nothing in the null implementation.
end;
{ TDataPoint<T> }
constructor TDataPoint<T>.Create(ATime: TDateTime; const AData: T);
@@ -216,104 +156,6 @@ begin
FDataProvider.Unlink(Tag);
end;
{ TMycGenericProcessor<T> }
constructor TMycGenericProcessor<T>.Create(const AProc: TProc);
begin
inherited Create;
FProc := AProc;
end;
function TMycGenericProcessor<T>.ProcessData(const Value: T): TState;
begin
Result := FProc(Value);
end;
{ TMycContainedProcessor<T> }
constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TProc);
begin
inherited Create(Controller);
FProc := AProc;
end;
function TMycContainedProcessor<T>.ProcessData(const Value: T): TState;
begin
Result := FProc(Value);
end;
{ TMycDataProvider<T> }
constructor TMycDataProvider<T>.Create(const Controller: IInterface);
begin
inherited Create(Controller);
end;
destructor TMycDataProvider<T>.Destroy;
begin
FListeners.Finalize;
inherited Destroy;
end;
function TMycDataProvider<T>.Broadcast(const Value: T): TState;
begin
FListeners.Lock;
try
var item := FListeners.First;
if item = nil then
exit(TState.Null);
var i := 1;
while item.Next <> nil do
begin
inc(i);
item := item.Next;
end;
var done := TLatch.CreateLatch(i);
while item <> nil do
begin
item.Receiver.ProcessData(Value).Signal.Subscribe(done);
item := item.Prev;
end;
Result := done.State;
finally
FListeners.Release;
end;
end;
procedure TMycDataProvider<T>.Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
begin
FListeners.Lock;
try
FListeners.Notify(Func);
finally
FListeners.Release;
end;
end;
function TMycDataProvider<T>.Link(const Processor: IMycProcessor<T>): TDataProvider<T>.TTag;
begin
// Add the Processor to the notification list
FListeners.Lock;
try
Result := FListeners.Advise(Processor);
finally
FListeners.Release;
end;
end;
procedure TMycDataProvider<T>.Unlink(Tag: TDataProvider<T>.TTag);
begin
FListeners.Lock;
try
FListeners.Unadvise(Tag);
finally
FListeners.Release;
end;
end;
{ TConverter<S, T> }
class constructor TConverter<S, T>.CreateClass;