Streams refactoring
This commit is contained in:
@@ -616,7 +616,7 @@ begin
|
|||||||
begin
|
begin
|
||||||
SetLength(args, Length(S));
|
SetLength(args, Length(S));
|
||||||
for k := 0 to High(S) do
|
for k := 0 to High(S) do
|
||||||
args[k] := S[k].Items[0];
|
args[k] := S[k];
|
||||||
|
|
||||||
res := lambdaFunc(args);
|
res := lambdaFunc(args);
|
||||||
|
|
||||||
|
|||||||
@@ -294,15 +294,7 @@ procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
|
|||||||
var
|
var
|
||||||
resultVal: TArray<TScalar.TValue>;
|
resultVal: TArray<TScalar.TValue>;
|
||||||
begin
|
begin
|
||||||
// We reuse the FObservers lock from the base class to protect state
|
// We assume thread safety is handled by the caller
|
||||||
// But since FObservers is private, we access it via method or need to make it protected.
|
|
||||||
// Making it protected or using a separate lock is cleaner.
|
|
||||||
// For now, let's assume we rely on the fact that OnSignal is usually serialized per thread
|
|
||||||
// or add a lock.
|
|
||||||
// -> Ideally, TCustomDataStream should expose Lock/Unlock or we add a Lock here.
|
|
||||||
|
|
||||||
// Simplification: We assume thread safety is handled by the caller or we add a dedicated lock.
|
|
||||||
// For this example, let's just do the logic:
|
|
||||||
|
|
||||||
for var src in FSources do
|
for var src in FSources do
|
||||||
if src.LastSeenCycle < CurrentCycle then
|
if src.LastSeenCycle < CurrentCycle then
|
||||||
|
|||||||
Reference in New Issue
Block a user