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