diff --git a/source/FluxBase-Core-Tests/FBCancellationTest.class.st b/source/FluxBase-Core-Tests/FBCancellationTest.class.st index 380a1f2..730f647 100644 --- a/source/FluxBase-Core-Tests/FBCancellationTest.class.st +++ b/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 +] diff --git a/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st b/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st index 1f96deb..31a089b 100644 --- a/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st +++ b/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 } } +] diff --git a/source/FluxBase-Core/FBParallelExecutor.class.st b/source/FluxBase-Core/FBParallelExecutor.class.st index 3fd7363..d679931 100644 --- a/source/FluxBase-Core/FBParallelExecutor.class.st +++ b/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' } diff --git a/source/FluxBase-Core/FBParallelJoinNodeBehavior.class.st b/source/FluxBase-Core/FBParallelJoinNodeBehavior.class.st index d8d655b..aa68c27 100644 --- a/source/FluxBase-Core/FBParallelJoinNodeBehavior.class.st +++ b/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 ] ] diff --git a/source/FluxBase-Core/FBParallelNodeBehavior.class.st b/source/FluxBase-Core/FBParallelNodeBehavior.class.st index d3c019b..8ee28b3 100644 --- a/source/FluxBase-Core/FBParallelNodeBehavior.class.st +++ b/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 ] ]