An iteration's slot comes from its token, not from the loop's cursor

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' }
+]

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.

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' }

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.

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