From ae6f0afc4e6243d70fb00b85a77dd4bf062b5e55 Mon Sep 17 00:00:00 2001 From: Norbert Hartl Date: Thu, 10 Sep 2026 12:24:57 +0200 Subject: [PATCH] An iteration's slot comes from its token, not from the loop's cursor Inside a parallel scope one activation serves every iteration of a node, with one slot per iteration. The slot was the executor's #index, which points at the newest iteration queued - while an earlier iteration may still be waiting and go on later. Two waits resumed in reverse order wrote both values into the second slot, and the node behind the scope join saw #(nil 'A'). Every iteration scope now knows its #index (FBScope), and a parallel activation reads its slot off its token's iteration scope (FBParallelActivation>>iterationIndex, FBParallelExecutor>>iterationOf:). The executor keeps a #currentIteration for what runs within a step or a resume - the loop variable and a predecessor's value are read for that iteration (#lastValueOf:, #outputFor:). A node resumed after a condition wait gets the parked token put back on its activation, so the slot the value goes into is the resumed iteration's. Tokens without a scope, persisted before iterations had one, fall back to the cursor as before. Co-Authored-By: Claude Fable 5.1 --- .../FBParallelExecutorTest.class.st | 80 +++++++++++++++++++ source/FluxBase-Core/FBNode.class.st | 8 +- .../FBParallelActivation.class.st | 29 ++++--- .../FluxBase-Core/FBParallelExecutor.class.st | 49 ++++++++++-- source/FluxBase-Core/FBScope.class.st | 26 +++++- 5 files changed, 174 insertions(+), 18 deletions(-) diff --git a/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st b/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st index e947ece..1f96deb 100644 --- a/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st +++ b/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st @@ -211,3 +211,83 @@ FBParallelExecutorTest >> testEachIterationRunsInItsOwnScope [ self assert: (instance activations detect: [ :each | each node == fork ]) token scope identicalTo: instance scope ] + +{ #category : 'tests' } +FBParallelExecutorTest >> testAnIterationScopeKnowsItsIndex [ + | start fork body flow instance scopes | + scopes := OrderedCollection new. + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + body := (FBPluggableTask named: 'body') block: [ :act | + scopes add: act token scope. + act value: act argument ]. + fork --> body. + (body addOutgoingNode: (FBParallelJoinNode named: 'scope join')) + addOutgoingNode: FBEndEvent new. + flow := FBFlow new start: start. + instance := flow newInstance. + instance executeWith: #( 10 20 30 ). + self assert: (scopes collect: [ :each | each index ]) asArray equals: #( 1 2 3 ). + self assert: (scopes collect: [ :each | each iterationOwnedBy: fork ]) asArray equals: scopes asArray. + self assert: (instance scope iterationOwnedBy: fork) isNil +] + +{ #category : 'tests' } +FBParallelExecutorTest >> testALateResumedIterationWritesIntoItsOwnSlot [ + "two iterations wait; the second is resumed first, then the first. Each value has to + land in the slot of its own iteration. The slot used to be the executor's cursor, + which had moved on to the second iteration by the time the first went on - the + first iteration's value overwrote the second's, and the node behind the scope join + saw #(nil 'A')" + | start fork wait after flow instance entries | + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + wait := fork addOutgoingNode: (FBExternalWait new condition: [ :each | false ]; yourself). + after := (wait addOutgoingNode: (FBParallelJoinNode named: 'scope join')) + addOutgoingNode: ((FBPluggableTask named: 'after') block: [ :act | act value: act argument ]). + after addOutgoingNode: FBEndEvent new. + flow := FBFlow new start: start. + instance := flow newInstance. + instance executeWith: #( 10 20 ). + entries := instance openWaitingEntries copy. + self assert: entries size equals: 2. + + self assert: (instance signal: entries second correlationId with: 'B'). + instance execute. + self assert: instance state equals: #waiting. + self assert: (instance signal: entries first correlationId with: 'A'). + instance execute. + + self assert: instance state equals: #done. + self assert: (instance activationOfNode: after) argument equals: #( 'A' 'B' ) +] + +{ #category : 'tests' } +FBParallelExecutorTest >> testTheLoopVariableFollowsTheIterationBeingStepped [ + "the fork's #output is the loop variable scripts read. While the first iteration, + resumed late, runs its next node, it has to read the first element - not the one + the cursor stands on, which is the second iteration's by then. Same for a + predecessor's value: the probe's argument is its own iteration's wait result" + | start fork wait probe seen flow instance entries | + seen := OrderedCollection new. + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + wait := fork addOutgoingNode: (FBExternalWait new condition: [ :each | false ]; yourself). + probe := wait addOutgoingNode: ((FBPluggableTask named: 'probe') block: [ :act | + seen add: (act instance activationOfNode: fork) output -> act argument. + act value: act argument ]). + (probe addOutgoingNode: (FBParallelJoinNode named: 'scope join')) + addOutgoingNode: FBEndEvent new. + flow := FBFlow new start: start. + instance := flow newInstance. + instance executeWith: #( 10 20 ). + entries := instance openWaitingEntries copy. + + self assert: (instance signal: entries second correlationId with: 'B'). + instance execute. + self assert: (instance signal: entries first correlationId with: 'A'). + instance execute. + + self assert: instance state equals: #done. + self assert: seen asArray equals: { 20 -> 'B'. 10 -> 'A' } +] diff --git a/source/FluxBase-Core/FBNode.class.st b/source/FluxBase-Core/FBNode.class.st index bac5ece..fbd4495 100644 --- a/source/FluxBase-Core/FBNode.class.st +++ b/source/FluxBase-Core/FBNode.class.st @@ -418,8 +418,11 @@ FBNode >> resultTransition: aSymbol [ { #category : 'executing' } FBNode >> resumeIn: anExecutor withToken: aToken value: aValue [ "I was activated and then parked inside my own execution - a condition that did not - hold (FBWaitForConditionBehavior). My activation is still standing and still carries - its own token, so aToken has nothing to restore here. The impulse is my result rather + hold (FBWaitForConditionBehavior). My activation is still standing; aToken is the + token it parked with, and it is put back on it: inside a parallel scope one + activation serves every iteration of a node, and a later iteration may have run it + since and left its own token there - the slot the resumed value goes into is read + off the token (FBParallelActivation>>iterationIndex). The impulse is my result rather than my input: it becomes my activation's #value and my outgoing transitions are followed, the way the wait would have gone on had the condition held. @@ -435,6 +438,7 @@ FBNode >> resumeIn: anExecutor withToken: aToken value: aValue [ | activation | activation := anExecutor activationOf: self. + aToken ifNotNil: [ activation token: aToken ]. activation resumed. activation value: aValue. activation done. diff --git a/source/FluxBase-Core/FBParallelActivation.class.st b/source/FluxBase-Core/FBParallelActivation.class.st index 8719fa1..a38ae66 100644 --- a/source/FluxBase-Core/FBParallelActivation.class.st +++ b/source/FluxBase-Core/FBParallelActivation.class.st @@ -13,12 +13,12 @@ Class { { #category : 'accessing' } FBParallelActivation >> argument [ - ^ arguments at: executor index + ^ arguments at: self iterationIndex ] { #category : 'accessing' } FBParallelActivation >> argument: anObject [ - arguments at: executor index put: anObject + arguments at: self iterationIndex put: anObject ] { #category : 'accessing' } @@ -37,11 +37,10 @@ FBParallelActivation >> done [ { #category : 'accessing' } FBParallelActivation >> currentValue [ - "the counterpart of #value:, which writes values at: executor index while #value - answers the whole per-iteration array - executor being this activation's *creating* - executor, i.e. the enclosing scope, exactly as for #argument" + "the counterpart of #value:, which writes one slot of the per-iteration array while + #value answers the whole array - the slot of my token's iteration, see #iterationIndex" - ^ values at: executor index + ^ values at: self iterationIndex ] { #category : 'initialization' } @@ -55,14 +54,14 @@ FBParallelActivation >> initializeValues: anInteger [ { #category : 'as yet unclassified' } FBParallelActivation >> metadataAt: key ifAbsent: aBlock [ - ^ (self metadataAtIndex: executor index) at: key + ^ (self metadataAtIndex: self iterationIndex) at: key ifAbsent: aBlock ] { #category : 'as yet unclassified' } FBParallelActivation >> metadataAt: key put: anObject [ - (self metadataAtIndex: executor index) at: key + (self metadataAtIndex: self iterationIndex) at: key put: anObject ] @@ -117,8 +116,18 @@ FBParallelActivation >> value [ ] { #category : 'accessing' } -FBParallelActivation >> value: anObject [ - values at: executor index put: anObject +FBParallelActivation >> value: anObject [ + values at: self iterationIndex put: anObject +] + +{ #category : 'accessing' } +FBParallelActivation >> iterationIndex [ + "my slot in the per-iteration arrays: the iteration my token belongs to, asked of + my creating executor - the parallel scope I run in (FBParallelExecutor>>iterationOf:). + I am one activation for all iterations of my node, so the slot cannot be mine to + keep; and it is not the executor's cursor either: that points at the newest + iteration, while I may be running for an earlier one that waited" + ^ executor iterationOf: self token ] { #category : 'accessing' } diff --git a/source/FluxBase-Core/FBParallelExecutor.class.st b/source/FluxBase-Core/FBParallelExecutor.class.st index 9670c38..3fd7363 100644 --- a/source/FluxBase-Core/FBParallelExecutor.class.st +++ b/source/FluxBase-Core/FBParallelExecutor.class.st @@ -8,7 +8,8 @@ Class { 'map', 'iterationScope', 'iterationScopeIndex', - 'openIterationScopes' + 'openIterationScopes', + 'current' ], #category : 'FluxBase-Core-Model', #package : 'FluxBase-Core', @@ -107,6 +108,39 @@ FBParallelExecutor >> isParallel [ ^ true ] +{ #category : 'accessing' } +FBParallelExecutor >> iterationOf: aToken [ + "the slot of the iteration aToken belongs to, read off the iteration scope its scope + lies in (FBScope>>iterationOwnedBy:). This is what an activation that runs late has + to write into and read from: a wait resumed after the loop went on to the next + iteration, whose step comes when #index already points further. A token without a + scope - persisted before iterations had one - or with none of mine falls back to + the cursor, which is right whenever iterations do not overlap" + aToken ifNil: [ ^ index ]. + ^ (aToken scope ifNotNil: [ :scope | scope iterationOwnedBy: node ]) + ifNil: [ index ] + ifNotNil: [ :iteration | iteration index ] +] + +{ #category : 'accessing' } +FBParallelExecutor >> currentIteration [ + "the iteration whose step or resume is running - what the loop variable and a + predecessor's value are read for (#lastValueOf:, #outputFor:). #index is not that: + it is the newest iteration queued, and the loop moves on to the next one while an + earlier one still waits. Set by #step from the work item's token and by + #resume:withToken:value: from the token going on; the cursor before any step" + ^ current ifNil: [ index ] +] + +{ #category : 'executing' } +FBParallelExecutor >> resume: aSuspension withToken: aToken value: aValue [ + "a resume is a step of the iteration whose token goes on: what runs now reads the + loop off that iteration, see #currentIteration. A resume that names no token - the + oldest parked one goes on - reads off the cursor, as before" + current := self iterationOf: aToken. + ^ super resume: aSuspension withToken: aToken value: aValue +] + { #category : 'accessing' } FBParallelExecutor >> lastValueOf: aFBNode [ "resolve aFBNode's value for the iteration this executor is currently on. @@ -120,10 +154,10 @@ FBParallelExecutor >> lastValueOf: aFBNode [ (see FBParallelJoinNodeBehavior, and the FluxBase 'nested parallel node' ticket this fixes)" | activation | - aFBNode == node ifTrue: [ ^ list at: index ]. + aFBNode == node ifTrue: [ ^ list at: self currentIteration ]. activation := self activationOf: aFBNode. ^ (activation executor == self) - ifTrue: [ activation value at: index ] + ifTrue: [ activation value at: self currentIteration ] ifFalse: [ activation value ] ] @@ -182,9 +216,9 @@ FBParallelExecutor >> outputFor: anActivation [ self node == anActivation node ifTrue: [ (self activationOf: node) isDone ifTrue: [ ^ list first ]. - ^ list at: index ]. + ^ list at: self currentIteration ]. (self activationOf: node) isDone ifTrue: [ ^ anActivation value first ]. - ^ anActivation value at: self index + ^ anActivation value at: (self iterationOf: anActivation token) ] { #category : 'private' } @@ -240,6 +274,10 @@ FBParallelExecutor >> step [ | workItem transition lastValue | workItem := pending removeFirst. transition := workItem transition. + "the iteration this step belongs to, read off the work item's token: a body seed + (no token yet, see #scopeToken) is the newest iteration's, everything else carries + the token of the iteration that queued it" + current := workItem token ifNil: [ index ] ifNotNil: [ :token | self iterationOf: token ]. lastValue := transition source lastValueIn: self. (self shouldFollowTransition: transition with: lastValue) ifTrue: [ @@ -265,6 +303,7 @@ FBParallelExecutor >> currentIterationScope [ iterationScopeIndex = index ifTrue: [ ^ iterationScope ]. iterationScope := ((self activationOf: node) token scope ifNil: [ flowInstance ensureScope ]) childOwnedBy: node. + iterationScope index: index. iterationScopeIndex := index. openIterationScopes ifNil: [ openIterationScopes := OrderedCollection new ]. openIterationScopes add: iterationScope. diff --git a/source/FluxBase-Core/FBScope.class.st b/source/FluxBase-Core/FBScope.class.st index c533108..ec612ad 100644 --- a/source/FluxBase-Core/FBScope.class.st +++ b/source/FluxBase-Core/FBScope.class.st @@ -29,6 +29,9 @@ the executors' pending and suspended bookkeeping and the waiting registry ownerNode the node that opened me, nil for a root state #active while I am alive, #cancelled once I was cancelled (#cancelled); #isCancelled asks my outer scopes as well + index for the scope of one iteration of a parallel node: which iteration - + the slot in the per-iteration arrays of the activations inside that + scope (FBParallelExecutor>>iterationOf:). Nil for every other scope #depth is the nesting depth, derived by walking #outer. #callDepth counts only the scopes a sub flow call opened, which is what the recursion guard has to compare: an @@ -41,7 +44,8 @@ Class { 'id', 'outer', 'ownerNode', - 'state' + 'state', + 'index' ], #category : 'FluxBase-Core-Model', #package : 'FluxBase-Core', @@ -77,6 +81,26 @@ FBScope >> depth [ ^ outer ifNil: [ 0 ] ifNotNil: [ :aScope | aScope depth + 1 ] ] +{ #category : 'accessing' } +FBScope >> index [ + ^ index +] + +{ #category : 'accessing' } +FBScope >> index: anInteger [ + index := anInteger +] + +{ #category : 'accessing' } +FBScope >> iterationOwnedBy: aNode [ + "the iteration of aNode - a parallel node - that I am or lie in: myself if aNode + opened me as one of its iterations, else what my outer scope answers, nil at the + root. A token inside a parallel scope reads its iteration off this, however deeply + nested in sub flow calls or inner loops it stands (FBParallelExecutor>>iterationOf:)" + (ownerNode == aNode and: [ index notNil ]) ifTrue: [ ^ self ]. + ^ outer ifNotNil: [ :each | each iterationOwnedBy: aNode ] +] + { #category : 'as yet unclassified' } FBScope >> id [ ^ id