A parallel scope completes whoever accounts for its last iteration

source/FluxBase-Core-Tests/FBCancellationTest.class.st

@@ -748,3 +748,91 @@ FBCancellationTest >> testAJoinReleasedFromAWithdrawalQueuesOnItsOwnExecutorNotO
 	self assert: (instance activations select: [ :each | each node == outerJoin ]) size equals: 2.
 	self assertConsistent: instance
 ]
+
+{ #category : 'helpers' }
+FBCancellationTest >> loopWaitingTwiceInto: aCollector [
+	"start -> fork -> wait (external) -> scope join -> after -> end over #( 10 20 );
+	after's argument is collected"
+	| start fork wait after flow instance |
+	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 |
+			aCollector add: act argument.
+			act value: act argument ]).
+	after addOutgoingNode: FBEndEvent new.
+	flow := FBFlow new start: start.
+	instance := flow newInstance.
+	instance executeWith: #( 10 20 ).
+	^ instance
+]
+
+{ #category : 'tests' }
+FBCancellationTest >> testCancellingTheLastWaitingIterationCompletesTheLoop [
+	"the first iteration was resumed and passed the scope join; the second, still
+	waiting, is taken back. The loop is over, and the node behind the scope join runs
+	with the first iteration's value and a nil slot for the second - it used to be left
+	unrun, with the instance reading #done"
+	| seen instance entries |
+	seen := OrderedCollection new.
+	instance := self loopWaitingTwiceInto: seen.
+	entries := instance openWaitingEntries copy.
+	self assert: (instance signal: entries first correlationId with: 1).
+	instance execute.
+	self assert: instance state equals: #waiting.
+	self assert: seen isEmpty.
+	self assertConsistent: instance.
+
+	instance cancelStrand: entries second token.
+
+	self assert: instance state equals: #running.
+	self assertConsistent: instance.
+	instance execute.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: { #( 1 nil ) }.
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBCancellationTest >> testCancellingEveryIterationCompletesTheLoopWithNothing [
+	"both iterations are taken back before either arrived: the loop completes with an
+	empty result and the flow goes on behind it - a loop is one activity of the flow,
+	unlike an AND-join, which releases nothing when nothing arrived"
+	| seen instance |
+	seen := OrderedCollection new.
+	instance := self loopWaitingTwiceInto: seen.
+	self assert: instance openWaitingEntries size equals: 2.
+
+	instance cancelStrand: instance openWaitingEntries first token.
+	self assert: instance state equals: #waiting.
+	self assert: seen isEmpty.
+	self assertConsistent: instance.
+	instance cancelStrand: instance openWaitingEntries first token.
+
+	self assert: instance state equals: #running.
+	self assertConsistent: instance.
+	instance execute.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: { #( nil nil ) }.
+	self assert: (self cancelledEventsOf: instance) size equals: 2.
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBCancellationTest >> testALoopInACancelledCaseDoesNotGoOnBehindItsJoin [
+	"Cancel Case while the loop waits: the loop's own scope fell with the root scope,
+	so completing it queues nothing - nothing runs behind the scope join, and the
+	instance reads #cancelled"
+	| seen instance |
+	seen := OrderedCollection new.
+	instance := self loopWaitingTwiceInto: seen.
+	self assert: instance state equals: #waiting.
+
+	instance cancel.
+
+	self assert: instance state equals: #cancelled.
+	self assert: instance hasEnded.
+	self assert: seen isEmpty.
+	self assertConsistent: instance
+]

source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st

@@ -291,3 +291,104 @@ FBParallelExecutorTest >> testTheLoopVariableFollowsTheIterationBeingStepped [
 	self assert: instance state equals: #done.
 	self assert: seen asArray equals: { 20 -> 'B'. 10 -> 'A' }
 ]
+
+{ #category : 'helpers' }
+FBParallelExecutorTest >> loopOver: aCollection bodyWhen: aBlock into: aCollector [
+	"start -> fork -> body (x2, when aBlock) -> scope join -> after -> end; after's
+	argument is collected"
+	| start fork body after flow instance |
+	start := FBStartEvent new.
+	fork := start addOutgoingNode: (FBParallelNode named: 'fork').
+	body := (FBPluggableTask named: 'body') block: [ :act | act value: act argument * 2 ].
+	fork --> (body when: aBlock).
+	after := (body addOutgoingNode: (FBParallelJoinNode named: 'scope join'))
+		addOutgoingNode: ((FBPluggableTask named: 'after') block: [ :act |
+			aCollector add: act argument.
+			act value: act argument ]).
+	after addOutgoingNode: FBEndEvent new.
+	flow := FBFlow new start: start.
+	instance := flow newInstance.
+	instance executeWith: aCollection.
+	^ instance
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testALastIterationFilteredByAConditionStillCompletesTheLoop [
+	"the last iteration's body is filtered by the condition on the fork's outgoing: the
+	loop is over all the same and the node behind the scope join runs, with a nil slot
+	for the iteration that had no body. It used to run only when the last iteration
+	itself passed the scope join - a filtered last iteration ended the flow silently"
+	| seen instance |
+	seen := OrderedCollection new.
+	instance := self loopOver: #( 1 2 3 ) bodyWhen: [ :x | x < 3 ] into: seen.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: { #( 2 4 nil ) }
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testALoopWhoseIterationsAreAllFilteredCompletesWithNothing [
+	"no iteration ran a body: the loop completes with an empty result and the flow goes
+	on behind it - the loop is one activity, not a join of independent strands"
+	| seen instance |
+	seen := OrderedCollection new.
+	instance := self loopOver: #( 1 2 3 ) bodyWhen: [ :x | false ] into: seen.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: { #( nil nil nil ) }
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testAnEmptyInputCompletesTheLoopAtOnce [
+	"nothing to iterate: the scope completes with an empty result and the flow goes on
+	behind the loop. It used to end there, silently, with the fork left #created"
+	| seen instance fork |
+	seen := OrderedCollection new.
+	instance := self loopOver: #( ) bodyWhen: [ :x | true ] into: seen.
+	fork := instance flow nodes detect: [ :each | each name = 'fork' ].
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: { #( ) }.
+	self assert: (instance activationOfNode: fork) state equals: #done.
+	self assert: instance isRootExecutor
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testTokensBehindTheScopeJoinBelongToTheForksStrand [
+	"behind the loop the strand is the fork's again: the token of the node after the
+	scope join lives in the fork's scope and descends from the fork's token - not from
+	the last iteration's, whose scope is over"
+	| seen instance fork after forkToken afterToken |
+	seen := OrderedCollection new.
+	instance := self loopOver: #( 1 2 ) bodyWhen: [ :x | true ] into: seen.
+	fork := instance flow nodes detect: [ :each | each name = 'fork' ].
+	after := instance flow nodes detect: [ :each | each name = 'after' ].
+	forkToken := (instance activationOfNode: fork) token.
+	afterToken := (instance activationOfNode: after) token.
+	self assert: afterToken scope identicalTo: forkToken scope.
+	self assert: (afterToken hasAncestor: forkToken).
+	self assert: (afterToken scope iterationOwnedBy: fork) isNil
+]
+
+{ #category : 'tests' }
+FBParallelExecutorTest >> testTheScopeJoinIsFoundPastAnInnerLoop [
+	"the outer loop's last iteration is filtered, so the outer scope completes without an
+	arrival and has to find its own join by itself - past the inner loop's join, which
+	closes the inner scope only. The first outer iteration ran the inner loop and arrived
+	with its result, the second left a nil slot"
+	| start outer inner body seen after flow instance |
+	seen := OrderedCollection new.
+	start := FBStartEvent new.
+	outer := start addOutgoingNode: (FBParallelNode named: 'outer').
+	inner := FBParallelNode named: 'inner'.
+	outer --> (inner when: [ :x | x size > 1 ]).
+	body := inner addOutgoingNode: ((FBPluggableTask named: 'body') block: [ :act | act value: act argument * 2 ]).
+	after := ((body addOutgoingNode: (FBParallelJoinNode named: 'inner join'))
+		addOutgoingNode: (FBParallelJoinNode named: 'outer join'))
+		addOutgoingNode: ((FBPluggableTask named: 'after') block: [ :act |
+			seen add: act argument.
+			act value: act argument ]).
+	after addOutgoingNode: FBEndEvent new.
+	flow := FBFlow new start: start.
+	instance := flow newInstance.
+	instance executeWith: #( #( 1 2 ) #( 3 ) ).
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: { { #( 2 4 ). nil } }
+]

source/FluxBase-Core/FBParallelExecutor.class.st

@@ -68,8 +68,86 @@ FBParallelExecutor >> continue [
         self queueScopeBody ].
     pending isEmpty ifFalse: [ ^ self ].
     suspended isEmpty ifFalse: [ ^ self ].
-    (self activationOf: node) done.
-    self stop
+    self completeScope
+]
+
+{ #category : 'services' }
+FBParallelExecutor >> completeScope [
+	"the loop is over - every iteration accounted for: arrived at the scope join,
+	filtered by a condition, or taken back (#atEnd) - and closes the way its scope join
+	closes it, whoever accounted for the last iteration: the join's own step
+	(FBParallelJoinNodeBehavior), a withdrawal (#settleAfterWithdrawal), a filtered last
+	iteration or an empty input (#continue). Before, only the join's own step closed the
+	scope; the other endings popped me and left the flow behind the loop unrun, while
+	the instance read #done.
+
+	I am popped, the fork is done, and the scope join's activation carries one value per
+	iteration - nil where none arrived - and goes on behind the join on a token of the
+	fork's strand: behind the loop the strand is the fork's again, in the fork's scope,
+	not in the last iteration's. The loop completes with nothing as well - an empty
+	input, every iteration filtered, every iteration taken back: the flow behind it goes
+	on with an empty result, since the loop is one activity of the flow and not a join of
+	independent strands (Norbert, 2026-09-10; an AND-join with nothing arrived releases
+	nothing, see FBJoinState>>canRelease). A loop whose own scope was cancelled queues
+	nothing here (FBBaseExecutor>>addPending:withToken:).
+
+	A body that ends the strand instead - every path through it reaching an end event -
+	has no join to go on from (#scopeJoinNode answers nil): the scope closes and that is
+	the end of the fork's strand, as it is when every branch of a split ends"
+
+	| fork join |
+	fork := self activationOf: node.
+	join := self scopeJoinActivation.
+	self popThenResume: [
+		fork done.
+		join ifNotNil: [
+			join values: join arguments.
+			join token: fork childToken.
+			join done.
+			join followAllOutgoing ] ]
+]
+
+{ #category : 'services' }
+FBParallelExecutor >> scopeJoinActivation [
+	"the activation of my scope join - the one the iterations arrived at, or a fresh one
+	when none did (#completeScope with nothing arrived); nil for a body without a join
+	to go on from"
+	| join |
+	join := self scopeJoinNode ifNil: [ ^ nil ].
+	^ self activations reverse
+		detect: [ :each | each node == join and: [ each executor == self ] ]
+		ifNone: [ self activations add: (self newActivation node: join; yourself) ]
+]
+
+{ #category : 'services' }
+FBParallelExecutor >> scopeJoinNode [
+	"the node that closes my scope: the FBParallelJoinNode of my fork - the first one
+	reachable from the body at my own nesting level, an inner loop's join being passed
+	with that loop - or, when the body closes the scope implicitly, the join synthesized
+	for it (#impliedContinuationNode, see FBNode>>effectiveOutgoingFor:) - or nil, when
+	every path through the body ends the strand at a flow terminal (an end event inside
+	the loop) and there is nothing behind the loop to go on to. Asked when the scope
+	completes without an arrival, which is the one moment no iteration has led the way"
+	| seen frontier openLeaf |
+	openLeaf := false.
+	seen := IdentitySet new.
+	frontier := OrderedCollection new.
+	node outgoing do: [ :each | frontier add: each target -> 0 ].
+	[ frontier isEmpty ] whileFalse: [
+		| pair candidate depth |
+		pair := frontier removeFirst.
+		candidate := pair key.
+		depth := pair value.
+		(seen includes: candidate) ifFalse: [
+			seen add: candidate.
+			(candidate isKindOf: FBParallelJoinNode) ifTrue: [
+				depth = 0 ifTrue: [ ^ candidate ].
+				depth := depth - 1 ].
+			(candidate isKindOf: FBParallelNode) ifTrue: [ depth := depth + 1 ].
+			candidate outgoing
+				ifEmpty: [ candidate isFlowTerminal ifFalse: [ openLeaf := true ] ]
+				ifNotEmpty: [ :outgoing | outgoing do: [ :each | frontier add: each target -> depth ] ] ] ].
+	^ openLeaf ifTrue: [ self impliedContinuationNode ] ifFalse: [ nil ]
 ]
 
 { #category : 'execution' }
@@ -162,9 +240,15 @@ FBParallelExecutor >> lastValueOf: aFBNode [
 ]
 
 { #category : 'accessing' }
-FBParallelExecutor >> list: aCollection [ 
+FBParallelExecutor >> list: aCollection [
+	"the input I iterate. An empty one has no iteration: the seed #node: queued falls
+	out of #pending again, and the cursor stands at 0 so that #atEnd holds - the fork's
+	behavior sends #continue right after pushing me, and the scope completes at once
+	with an empty result (#completeScope)"
 	list := aCollection.
-
+	aCollection isEmpty ifTrue: [
+		pending removeAll: pending copy.
+		index := 0 ]
 ]
 
 { #category : 'instance creation' }

source/FluxBase-Core/FBParallelJoinNodeBehavior.class.st

@@ -8,20 +8,12 @@ Class {
 
 { #category : 'execution' }
 FBParallelJoinNodeBehavior >> execute [
-	"neither the fork's own activation nor this join's own activation ever reached #done:
-	FBParallelExecutor>>continue's own '(self activationOf: node) done. self stop' tail is
-	only reached when nothing ever pops this executor first - but a join always pops it
-	directly via #popThenResume: once #atEnd, bypassing that tail entirely. Mirror
-	FBTaskBehavior>>execute's value/done/followAllOutgoing order for both activations here
-	instead - #done is itself a no-op (see FBParallelActivation>>done) until the executor is
-	truly at its last index, so this is safe to call unconditionally, exactly like a plain
-	task's activation already does on every iteration."
+	"one arrival per iteration: my argument slot already holds this iteration's value
+	(FBParallelActivation>>argument:, slotted by token). The scope closes once every
+	iteration is accounted for - here, when this arrival was the last thing outstanding;
+	from the executor itself when a withdrawal or a filtered iteration was
+	(FBParallelExecutor>>completeScope, the one way the scope closes)"
 	| executor |
 	executor := activation flowInstance executor.
-	executor atEnd ifTrue: [
-		executor popThenResume: [
-			(executor activationOf: executor node) done.
-			activation values: activation arguments.
-			activation done.
-			activation followAllOutgoing ] ]
+	executor atEnd ifTrue: [ executor completeScope ]
 ]

source/FluxBase-Core/FBParallelNodeBehavior.class.st

@@ -8,21 +8,25 @@ Class {
 
 { #category : 'execution' }
 FBParallelNodeBehavior >> execute [
-
+	| executor |
 	(activation argument isCollection and: [
 		 activation argument isSequenceable ]) ifFalse: [
 		FBError signal:
 			'Input for parallel execution has to be a non empty collection.' ].
-	activation argument size > 0 ifFalse: [ ^ self ].
 	"the input this node iterates. written before the outgoing guard below, since a node
 	with no loop body never gets an executor and its output still has to answer something
 	(FBFlowTest>>testParallelNodeAsLastNodeHasOutput). while a scope of its own is open
 	the node's output comes from that scope's #list and #index instead - this is only
 	what is left to answer from once it is closed, read back through
 	FBBaseActivation>>currentValue rather than #value"
-	activation value: activation argument first.
+	activation argument ifNotEmpty: [ :input | activation value: input first ].
 	activation node outgoing notEmpty ifFalse: [ ^ self ].
-	activation flowInstance pushExecutor: (FBParallelExecutor new
-			 node: activation node;
-			 list: activation argument)
+	executor := FBParallelExecutor new
+		node: activation node;
+		list: activation argument.
+	activation flowInstance pushExecutor: executor.
+	"an empty input has no iteration to step: the scope completes at once, with an empty
+	result, and the flow goes on behind the loop's join (FBParallelExecutor>>completeScope)
+	- it used to end here, silently, with the fork left #created"
+	activation argument ifEmpty: [ executor continue ]
 ]