Cancel Multiple Instance Activity, and a cancelled fork strand ends i…

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

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

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