From d0ed83b6c6ef124594e804ce989883e95c544728 Mon Sep 17 00:00:00 2001 From: Norbert Hartl Date: Thu, 10 Sep 2026 12:52:46 +0200 Subject: [PATCH] Cancel Multiple Instance Activity, and a cancelled fork strand ends its loop FBFlowInstance>>cancelIterationsOf: cancels every iteration of a running parallel node that is still running or waiting - their scopes fall, sub flow frames included - and lets no further one start. The loop then completes with what arrived, nil in the slots taken back, and the flow goes on behind it (WCP-26: control passes on). A node that is not looping answers empty and leaves no trace. Cancelling the fork's own strand is the other thing (WCP-19 on the loop): the loop starts no further iteration and nothing goes on behind its join. Until now it took the iterations and then started the next one. For that, cancelStrand: marks the named token itself when its activation is still standing - a fork whose loop is open, a sub flow node whose child runs - and records an FBNodeActivationCancelled for it; a token of an activation already done stays as it is. Co-Authored-By: Claude Fable 5.1 --- .../FBCancellationTest.class.st | 144 ++++++++++++++++++ source/FluxBase-Core/FBFlowInstance.class.st | 32 +++- .../FluxBase-Core/FBParallelExecutor.class.st | 45 +++++- 3 files changed, 212 insertions(+), 9 deletions(-) diff --git a/source/FluxBase-Core-Tests/FBCancellationTest.class.st b/source/FluxBase-Core-Tests/FBCancellationTest.class.st index 730f647..2c3edf4 100644 --- a/source/FluxBase-Core-Tests/FBCancellationTest.class.st +++ b/source/FluxBase-Core-Tests/FBCancellationTest.class.st @@ -836,3 +836,147 @@ FBCancellationTest >> testALoopInACancelledCaseDoesNotGoOnBehindItsJoin [ self assert: seen isEmpty. self assertConsistent: instance ] + +{ #category : 'tests' } +FBCancellationTest >> testCancellingTheRemainingIterationsCompletesTheLoopWithWhatArrived [ + "Cancel Multiple Instance Activity: one iteration has arrived, the other still + waits. Cancelling the remaining iterations takes the wait, and the loop completes + with the first iteration's value and a nil slot - the flow goes on behind it" + | seen instance entries fork taken | + seen := OrderedCollection new. + instance := self loopWaitingTwiceInto: seen. + fork := instance flow nodes detect: [ :each | each name = 'fork' ]. + entries := instance openWaitingEntries copy. + self assert: (instance signal: entries first correlationId with: 1). + instance execute. + self assert: instance state equals: #waiting. + self assertConsistent: instance. + + taken := instance cancelIterationsOf: fork. + + self assert: taken asArray equals: { entries second token }. + self assert: instance openWaitingEntries isEmpty. + 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 assert: (instance events count: [ :each | each isKindOf: FBScopeCancelled ]) equals: 1. + self assertConsistent: instance +] + +{ #category : 'tests' } +FBCancellationTest >> testCancellingTheIterationsOfANodeThatIsNotLoopingIsSilent [ + "before the loop opened and after it closed there is nothing to cancel: empty + answer, no event, the instance untouched" + | seen instance fork before | + seen := OrderedCollection new. + instance := (self loopWaitingTwiceInto: seen) flow newInstance. + fork := instance flow nodes detect: [ :each | each name = 'fork' ]. + self assert: (instance cancelIterationsOf: fork) isEmpty. + self assert: instance state equals: #created. + + instance executeWith: #( 10 20 ). + instance openWaitingEntries copy do: [ :each | instance signal: each correlationId with: 0 ]. + instance execute. + self assert: instance state equals: #done. + before := instance events size. + self assert: (instance cancelIterationsOf: fork) isEmpty. + self assert: instance events size equals: before. + self assertConsistent: instance +] + +{ #category : 'tests' } +FBCancellationTest >> testCancellingTheRemainingIterationsInsideASplitBranchLetsTheJoinGoOn [ + "the loop runs in one branch of an AND-split, its sibling has arrived at the join. + Cancelling the loop's iterations completes the loop, its branch arrives with the + loop's result, and the join releases with both" + | start split fork wait b join after flow instance | + start := FBStartEvent new. + split := start addOutgoingNode: (FBAndSplit named: 'split'). + "the sibling first, and the loop one node later: the sibling has to arrive at the + join before the loop's frame takes the top - the root queue is stepped in order" + b := split addOutgoingNode: ((FBPluggableTask named: 'b') block: [ :act | act value: 3 ]). + join := FBAndJoin named: 'join'. + b --> join. + fork := (split addOutgoingNode: ((FBPluggableTask named: 'pre') block: [ :act | act value: act argument ])) + addOutgoingNode: (FBParallelNode named: 'fork'). + wait := fork addOutgoingNode: (FBExternalWait new condition: [ :each | false ]; yourself). + (wait addOutgoingNode: (FBParallelJoinNode named: 'scope join')) --> join. + after := 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 ). + self assert: instance state equals: #waiting. + self assert: (instance activations count: [ :each | each node == join ]) equals: 1. + self assertConsistent: instance. + + instance cancelIterationsOf: fork. + + self assertConsistent: instance. + instance execute. + self assert: instance state equals: #done. + self assert: (instance activations count: [ :each | each node == after ]) equals: 1. + self assert: (instance activationOfNode: after) argument keys asSet equals: { b. wait outgoing first target } asSet. + self assert: ((instance activationOfNode: after) argument at: wait outgoing first target) equals: #( nil nil ). + self assertConsistent: instance +] + +{ #category : 'tests' } +FBCancellationTest >> testCancellingTheForksStrandTakesTheLoopAndNothingRunsBehindIt [ + "Cancel Activity on the loop itself: the fork's strand and everything descending + from it fall, no further iteration starts, and nothing goes on behind the scope + join - the strand is gone. Used to take the iterations and then start the next one" + | seen instance fork forkToken | + seen := OrderedCollection new. + instance := self loopWaitingTwiceInto: seen. + fork := instance flow nodes detect: [ :each | each name = 'fork' ]. + forkToken := (instance activationOfNode: fork) token. + self assert: instance openWaitingEntries size equals: 2. + self assertConsistent: instance. + + instance cancelStrand: forkToken. + + self assert: forkToken isCancelled. + self assert: instance openWaitingEntries isEmpty. + self assert: instance isRootExecutor. + self assert: instance state equals: #done. + self assert: instance hasEnded. + self assert: seen isEmpty. + self assertConsistent: instance +] + +{ #category : 'tests' } +FBCancellationTest >> testCancellingTheRemainingIterationsTakesAWaitingSubFlowWithThem [ + "an iteration calls a sub flow that waits, so its frame is on top and the loop + cannot even start the next iteration. Cancelling the iterations takes the child's + wait, the frame pops, and the loop completes with nothing" + | start fork subNode after flow instance seen | + seen := OrderedCollection new. + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + subNode := fork addOutgoingNode: FBSubFlowNode new. + subNode subFlow: self flowWaitingFromOutside. + after := (subNode addOutgoingNode: (FBParallelJoinNode named: 'scope 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: #( 10 20 ). + self assert: instance state equals: #waiting. + self assert: (instance executor isKindOf: FBSubFlowExecutor). + self assert: instance openWaitingEntries size equals: 1. + self assertConsistent: instance. + + instance cancelIterationsOf: fork. + + self assert: instance openWaitingEntries isEmpty. + self assertConsistent: instance. + instance execute. + self assert: instance state equals: #done. + self assert: seen asArray equals: { #( nil nil ) }. + self assertConsistent: instance +] diff --git a/source/FluxBase-Core/FBFlowInstance.class.st b/source/FluxBase-Core/FBFlowInstance.class.st index 26883a8..eb4146f 100644 --- a/source/FluxBase-Core/FBFlowInstance.class.st +++ b/source/FluxBase-Core/FBFlowInstance.class.st @@ -966,13 +966,25 @@ FBFlowInstance >> cancelStrand: aToken [ Answers the tokens taken - empty, and without a trace, when nothing of the strand was left: cancelling twice is cancelling once. + aToken itself may stand on no queue at all: the token of an activation that is still + running - a parallel node whose loop is open, a sub flow node whose child runs. + That activation is part of the strand and reads #cancelled from now on; its token is + marked before the withdrawal, since the settling after it asks (a loop whose fork's + strand fell starts nothing more and goes on with nothing, + FBParallelExecutor>>isStrandCancelled). An activation already done is history, its + token stays as it is - the strand from there on is what falls. + This is Cancel Activity (WCP-19), and the selection an abbrechender Join makes over - its siblings. Joins do not yet know about cancelled branches - a join a cancelled - branch was heading to keeps waiting for it; that is the next slice" + its siblings. A join a cancelled branch was heading to knows (FBJoinState>>canRelease)" - | taken | + | taken standing | + standing := aToken isCancelled not and: [ + aToken activation notNil and: [ aToken activation isDone not ] ]. + standing ifTrue: [ aToken cancelled ]. taken := self withdrawStrandsSatisfying: [ :each | each == aToken or: [ each hasAncestor: aToken ] ]. - taken ifEmpty: [ ^ taken ]. + (standing and: [ (taken includes: aToken) not ]) ifTrue: [ + self addEvent: (FBNodeActivationCancelled new activation: aToken activation) ]. + (taken isEmpty and: [ standing not ]) ifTrue: [ ^ taken ]. self addEvent: (FBStrandCancelled new token: aToken). self endIfOver. ^ taken @@ -1002,6 +1014,18 @@ FBFlowInstance >> cancelScope: aScope [ ^ taken ] +{ #category : 'cancelling' } +FBFlowInstance >> cancelIterationsOf: aParallelNode [ + "Cancel Multiple Instance Activity (WCP-26) on aParallelNode's running loop: the + iterations still running or waiting are cancelled, no further one starts, and the + loop completes with what arrived - the flow goes on behind it + (FBParallelExecutor>>cancelRemainingIterations). Answers the tokens taken. A node + that is not looping right now - before its first iteration or after its last, or + looping in a sub flow of mine rather than in me - answers empty and leaves no trace" + executor ifNil: [ ^ #() ]. + ^ (self parallelExecutorFrom: aParallelNode ifAbsent: [ ^ #() ]) cancelRemainingIterations +] + { #category : 'cancelling' } FBFlowInstance >> cancel [ "Cancel Case (WCP-20): cancel my root scope, and with it everything that runs or diff --git a/source/FluxBase-Core/FBParallelExecutor.class.st b/source/FluxBase-Core/FBParallelExecutor.class.st index d679931..ea51f61 100644 --- a/source/FluxBase-Core/FBParallelExecutor.class.st +++ b/source/FluxBase-Core/FBParallelExecutor.class.st @@ -57,10 +57,11 @@ FBParallelExecutor >> buildImpliedContinuationNode [ { #category : 'services' } FBParallelExecutor >> continue [ self closeFinishedIterations. - self isInCancelledScope ifTrue: [ - "the scope my loop runs in was cancelled: no further iteration starts, and what - was queued for the current one falls. Nothing follows the fork either - the - scope join never runs, and a token of a cancelled scope queues nothing" + (self isInCancelledScope or: [ self isStrandCancelled ]) ifTrue: [ + "the scope my loop runs in was cancelled, or the fork's own strand was: no + further iteration starts, and what was queued for the current one falls. Nothing + follows the fork either - #completeScope goes on behind the join only for a + living strand, and a token of a cancelled scope queues nothing" pending removeAll: pending copy. index := list size ]. (pending isEmpty and: [ index < list size ]) ifTrue: [ @@ -100,13 +101,47 @@ FBParallelExecutor >> completeScope [ join := self scopeJoinActivation. self popThenResume: [ fork done. - join ifNotNil: [ + (join notNil and: [ self isStrandCancelled not ]) ifTrue: [ join values: join arguments. join token: fork childToken. join done. join followAllOutgoing ] ] ] +{ #category : 'cancelling' } +FBParallelExecutor >> cancelRemainingIterations [ + "Cancel Multiple Instance Activity (WCP-26): every iteration still running or + waiting is cancelled and no further one starts; then the scope completes with what + arrived - nil in the slots of the iterations taken back - and the flow goes on + behind the loop (#completeScope, reached through the settling that follows each + withdrawal). The fork's own strand stays and carries the result on; a cancelled fork + strand is the other thing (FBFlowInstance>>cancelStrand: on the fork's token): it + takes the loop with it and nothing goes on behind. + + Every iteration still open has its scope (#openIterationScopes), and cancelling the + scope takes its strands, its waits and its addresses, sub flow frames included + (FBFlowInstance>>cancelScope:). The seed of an iteration queued but not yet stepped + carries no token and lies in no scope yet; it falls here. Answers the tokens taken" + + | taken | + index := list size. + pending removeAll: pending copy. + taken := OrderedCollection new. + (openIterationScopes ifNil: [ #() ]) copy do: [ :each | + taken addAll: (flowInstance cancelScope: each) ]. + "with nothing open there was no withdrawal to settle after - the scope has to + complete all the same" + taken ifEmpty: [ self settleAfterWithdrawal ]. + ^ taken +] + +{ #category : 'testing' } +FBParallelExecutor >> isStrandCancelled [ + "was the fork's own strand taken back (FBFlowInstance>>cancelStrand:)? Then the loop + is over without a continuation: no further iteration, nothing behind the join" + ^ (self activationOf: node) token isCancelled +] + { #category : 'services' } FBParallelExecutor >> scopeJoinActivation [ "the activation of my scope join - the one the iterations arrived at, or a fresh one