Deferred choice: the first branch whose wait goes on wins

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

@@ -185,27 +185,30 @@ FBCancellationTest >> testCancellingAParkedTransitionTakesItsAddressAndLeavesThe
 { #category : 'tests' }
 FBCancellationTest >> testCancellingIntoASubFlowTakesTheChildsStrands [
 	"the child's strands descend from the caller's token (FBFlowInstance>>descendFrom:),
-	so cancelling the calling strand reaches a wait inside the sub flow. The child is
-	then over without an end, and until a cancelled scope carries that as its state,
-	the frame fails its caller the way an errored child does - the instance stops in
-	#error rather than hanging on a frame nothing will pop"
-	| child parent instance caller wait |
+	so cancelling the calling strand reaches a wait inside the sub flow. The call's own
+	token falls with the strand - the sub flow node stands, unqueued, and is part of it -
+	so the frame pops quietly, the call reads #cancelled, and with nothing left the
+	instance is over"
+	| child parent instance caller wait waitToken |
 	child := self flowWaitingFromOutside.
 	parent := self flowCalling: child.
 	instance := parent newInstance.
 	instance executeWith: 1.
 	self assert: instance state equals: #waiting.
 	caller := instance activations detect: [ :each | each node isKindOf: FBSubFlowNode ].
 	wait := instance suspended first.
+	waitToken := instance openWaitingEntries first token.
 	self assertConsistent: instance.
 
-	self assert: (instance cancelStrand: caller token) size equals: 1.
+	self assert: (instance cancelStrand: caller token) asSet equals: { waitToken. caller token } asSet.
 
 	self assert: instance suspended isEmpty.
 	self assert: instance subFlowInstances first suspended isEmpty.
 	self assert: (instance subFlowInstances first activations detect: [ :each | each node == wait ]) state equals: #cancelled.
-	self assert: instance state equals: #error.
-	self assert: instance atEnd.
+	self assert: caller state equals: #cancelled.
+	self assert: instance isRootExecutor.
+	self assert: instance state equals: #done.
+	self assert: instance hasEnded.
 	self assert: (self cancelledEventsOf: instance) size equals: 1.
 	self assertConsistent: instance
 ]

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

@@ -0,0 +1,221 @@
+"
+A deferred choice: the first branch whose wait goes on wins, the others fall - wherever
+they stand. Built on FBCancellationTest for its helpers and its state invariant.
+"
+Class {
+	#name : 'FBDeferredChoiceTest',
+	#superclass : 'FBCancellationTest',
+	#category : 'FluxBase-Core-Tests',
+	#package : 'FluxBase-Core-Tests'
+}
+
+{ #category : 'helpers' }
+FBDeferredChoiceTest >> choiceBetween: aNode and: anotherNode into: aCollector [
+	"start -> choice -> aNode -> after -> end
+	                \-> anotherNode -/        ; after's argument is collected"
+	| start choice after |
+	start := FBStartEvent new.
+	choice := start addOutgoingNode: (FBDeferredChoice named: 'choice').
+	choice --> aNode.
+	choice --> anotherNode.
+	after := (FBPluggableTask named: 'after') block: [ :act |
+		aCollector add: act argument.
+		act value: act argument ].
+	aNode --> after.
+	anotherNode --> after.
+	after addOutgoingNode: FBEndEvent new.
+	^ FBFlow new start: start
+]
+
+{ #category : 'helpers' }
+FBDeferredChoiceTest >> externalWaitNamed: aString [
+	^ FBExternalWait new condition: [ :each | false ]; name: aString; yourself
+]
+
+{ #category : 'helpers' }
+FBDeferredChoiceTest >> decisionsOf: anInstance [
+	^ anInstance events select: [ :each | each isKindOf: FBDeferredChoiceDecided ]
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testTheFirstBranchResumedWinsAndTheOtherFalls [
+	| seen a b instance entryA entryB |
+	seen := OrderedCollection new.
+	a := self externalWaitNamed: 'a'.
+	b := self externalWaitNamed: 'b'.
+	instance := (self choiceBetween: a and: b into: seen) newInstance.
+	instance executeWith: 1.
+	self assert: instance state equals: #waiting.
+	self assert: instance openWaitingEntries size equals: 2.
+	entryA := instance openWaitingEntries detect: [ :each | each token node == a ].
+	entryB := instance openWaitingEntries detect: [ :each | each token node == b ].
+	self assertConsistent: instance.
+
+	self assert: (instance signal: entryB correlationId with: 'B').
+
+	"b went on, a fell: its wait, its address, its activation reads cancelled"
+	self assert: instance openWaitingEntries isEmpty.
+	self assert: entryA token isCancelled.
+	self assert: (instance activationOfNode: a) state equals: #cancelled.
+	self assert: (instance activationOfNode: b) state equals: #done.
+	self assert: (self decisionsOf: instance) size equals: 1.
+	self assert: (self decisionsOf: instance) first choiceName equals: 'choice'.
+	self assert: (self decisionsOf: instance) first winnerTokenId equals: entryB token id.
+	self assertConsistent: instance.
+	instance execute.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: #( 'B' ).
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testASecondImpulseOnTheLoserIsSilent [
+	| seen a b instance entryA entryB before |
+	seen := OrderedCollection new.
+	a := self externalWaitNamed: 'a'.
+	b := self externalWaitNamed: 'b'.
+	instance := (self choiceBetween: a and: b into: seen) newInstance.
+	instance executeWith: 1.
+	entryA := instance openWaitingEntries detect: [ :each | each token node == a ].
+	entryB := instance openWaitingEntries detect: [ :each | each token node == b ].
+	self assert: (instance signal: entryA correlationId with: 'A').
+	instance execute.
+	before := instance events size.
+
+	self deny: (instance signal: entryB correlationId with: 'late').
+
+	self assert: instance events size equals: before.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: #( 'A' ).
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testAParkedTriggerBranchLoses [
+	"the losing branch leads into a trigger target: its transition was parked, never
+	activated. It falls with its address, and the trigger never runs"
+	| seen a trigger instance entryA entryTrigger |
+	seen := OrderedCollection new.
+	a := self externalWaitNamed: 'a'.
+	trigger := FBTriggerTask named: 'trigger'.
+	instance := (self choiceBetween: a and: trigger into: seen) newInstance.
+	instance executeWith: 1.
+	self assert: instance openWaitingEntries size equals: 2.
+	entryA := instance openWaitingEntries detect: [ :each | each token node == a ].
+	entryTrigger := instance openWaitingEntries detect: [ :each | each waitingNode == trigger ].
+	self assertConsistent: instance.
+
+	self assert: (instance signal: entryA correlationId with: 'A').
+
+	self assert: instance openWaitingEntries isEmpty.
+	self assert: instance suspended isEmpty.
+	self assert: entryTrigger token isCancelled.
+	self deny: (instance activations anySatisfy: [ :each | each node == trigger ]).
+	instance execute.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: #( 'A' ).
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testALoserInsideASubFlowFallsWithItsFrame [
+	"the losing branch called a sub flow that waits, so its frame is on top. Deciding
+	the choice takes the child's wait and the call itself: the frame pops quietly, the
+	call reads #cancelled, and the winner goes on in the root"
+	| seen a call instance entryA |
+	seen := OrderedCollection new.
+	a := self externalWaitNamed: 'a'.
+	call := FBSubFlowNode new.
+	call subFlow: self flowWaitingFromOutside.
+	instance := (self choiceBetween: a and: call into: seen) newInstance.
+	instance executeWith: 1.
+	self assert: instance state equals: #waiting.
+	self assert: (instance executor isKindOf: FBSubFlowExecutor).
+	self assert: instance openWaitingEntries size equals: 2.
+	entryA := instance openWaitingEntries detect: [ :each | each token node == a ].
+	self assertConsistent: instance.
+
+	self assert: (instance signal: entryA correlationId with: 'A').
+
+	self assert: instance openWaitingEntries isEmpty.
+	self assert: (instance activationOfNode: call) state equals: #cancelled.
+	self assertConsistent: instance.
+	instance execute.
+	self assert: instance isRootExecutor.
+	self assert: instance state equals: #done.
+	self assert: seen asArray equals: #( 'A' ).
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testAChoiceInsideAnIterationDecidesThatIterationOnly [
+	"each iteration of a parallel scope makes its own choice: deciding the second
+	iteration's leaves the first iteration's two waits standing"
+	| start fork choice a b after flow instance fork2 entriesOf seen |
+	seen := OrderedCollection new.
+	start := FBStartEvent new.
+	fork := start addOutgoingNode: (FBParallelNode named: 'fork').
+	choice := fork addOutgoingNode: (FBDeferredChoice named: 'choice').
+	a := choice addOutgoingNode: (self externalWaitNamed: 'a').
+	b := choice addOutgoingNode: (self externalWaitNamed: 'b').
+	after := (FBPluggableTask named: 'after') block: [ :act | seen add: act argument. act value: act argument ].
+	a --> after.
+	b --> after.
+	(after addOutgoingNode: (FBParallelJoinNode named: 'scope join')) addOutgoingNode: FBEndEvent new.
+	flow := FBFlow new start: start.
+	instance := flow newInstance.
+	instance executeWith: #( 10 20 ).
+	fork2 := fork.
+	entriesOf := [ :index | instance openWaitingEntries select: [ :each | (each token scope iterationOwnedBy: fork2) index = index ] ].
+	self assert: instance openWaitingEntries size equals: 4.
+	self assertConsistent: instance.
+
+	self assert: (instance signal: ((entriesOf value: 2) detect: [ :each | each token node == b ]) correlationId with: 'B2').
+
+	self assert: (entriesOf value: 2) isEmpty.
+	self assert: (entriesOf value: 1) size equals: 2.
+	self assert: (self decisionsOf: instance) size equals: 1.
+	self assertConsistent: instance.
+	instance execute.
+	self assert: (instance signal: ((entriesOf value: 1) detect: [ :each | each token node == a ]) correlationId with: 'A1').
+	instance execute.
+	self assert: instance state equals: #done.
+	self assert: (self decisionsOf: instance) size equals: 2.
+	self assert: seen asSet equals: #( 'A1' 'B2' ) asSet.
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testAConditionThatHoldsAtOnceDecidesTheChoice [
+	"a condition wait whose condition already holds is a wait that went on: it decides
+	the choice as it is entered, and the other branch falls before anything waited"
+	| seen a b instance |
+	seen := OrderedCollection new.
+	a := FBPluggableWait new condition: [ :each | true ]; name: 'a'; yourself.
+	b := self externalWaitNamed: 'b'.
+	instance := (self choiceBetween: a and: b into: seen) newInstance.
+	instance executeWith: 1.
+	self assert: instance state equals: #done.
+	self assert: instance openWaitingEntries isEmpty.
+	self assert: (self decisionsOf: instance) size equals: 1.
+	"once - a condition that holds passes no value on, as before"
+	self assert: seen size equals: 1.
+	self deny: (instance activations anySatisfy: [ :each | each node == b and: [ each state ~= #cancelled ] ]).
+	self assertConsistent: instance
+]
+
+{ #category : 'tests' }
+FBDeferredChoiceTest >> testABranchThatCannotWaitIsRejected [
+	"a branch that would run at once decides nothing and lets the others go on with it:
+	the choice refuses to start"
+	| seen a task instance |
+	seen := OrderedCollection new.
+	a := self externalWaitNamed: 'a'.
+	task := FBTask named: 'task'.
+	instance := (self choiceBetween: a and: task into: seen) newInstance.
+	instance executeWith: 1.
+	self assert: instance state equals: #error.
+	self assert: instance openWaitingEntries isEmpty.
+	self assert: seen isEmpty.
+	self assert: ((instance events detect: [ :each | each isKindOf: FBNodeActivationErrored ]) message includesSubstring: 'waits')
+]

source/FluxBase-Core/FBBaseExecutor.class.st

@@ -237,6 +237,18 @@ FBBaseExecutor >> takeTokenForSuspended: anObject [
 	^ self takeTokenForSuspended: anObject satisfying: [ :each | true ]
 ]
 
+{ #category : 'accessing' }
+FBBaseExecutor >> tokenForSuspended: anObject satisfying: aBlock [
+	"the first token stored for anObject that aBlock accepts, left where it is - the one
+	#takeTokenForSuspended:satisfying: would take. For a look before the taking
+	(#resume:withToken:value:)"
+	| tokens |
+	suspendedTokens ifNil: [ ^ nil ].
+	tokens := suspendedTokens at: anObject ifAbsent: [ ^ nil ].
+	(tokens isKindOf: Collection) ifFalse: [ ^ (aBlock value: tokens) ifTrue: [ tokens ] ifFalse: [ nil ] ].
+	^ tokens detect: aBlock ifNone: [ nil ]
+]
+
 { #category : 'accessing' }
 FBBaseExecutor >> takeTokenForSuspended: anObject satisfying: aBlock [
 	"answer and forget the first token stored for anObject that aBlock accepts, or nil.
@@ -334,10 +346,14 @@ FBBaseExecutor >> resume: aSuspension withToken: aToken value: aValue [
 	submitted on it"
 
 	| token |
-	token := self
-		takeTokenForSuspended: aSuspension
-		satisfying: [ :each | aToken isNil or: [ each == aToken ] ].
-	token := aToken ifNil: [ token ].
+	token := aToken ifNil: [ self tokenForSuspended: aSuspension satisfying: [ :each | true ] ].
+	"a wait going on is what decides a deferred choice its strand lies on: the branches
+	that did not go on fall now (FBFlowInstance>>decideDeferredChoiceWonBy:). Before the
+	wait itself is taken out: the withdrawal settles the executors, and one that saw
+	nothing pending and nothing suspended - the wait going on being in flight - would
+	take itself for finished and close its scope under the resumed strand"
+	token ifNotNil: [ flowInstance decideDeferredChoiceWonBy: token ].
+	self takeTokenForSuspended: aSuspension satisfying: [ :each | aToken isNil or: [ each == aToken ] ].
 	flowInstance dropWaitingEntriesMatching: aSuspension token: token.
 	suspended remove: aSuspension.
 	^ aSuspension resumeIn: self withToken: token value: aValue

source/FluxBase-Core/FBDeferredChoice.class.st

@@ -0,0 +1,43 @@
+"
+I am a deferred choice (WCP-16; BPMN's event-based gateway): a split whose branches each
+lead into a waiting spot - a condition wait, a trigger target such as a form, a remote
+flow - and of which exactly one goes on: the first one whose wait is resumed. The others
+fall the moment the choice is decided - queued or parked, in this instance or in a sub
+flow a branch called - the way a cancelled strand falls (FBFlowInstance>>
+decideDeferredChoice:wonBy:), and an FBDeferredChoiceDecided records which branch won.
+Behind me the branches may merge or end on their own; only the winner ever gets there.
+
+The decision is made where a wait goes on, FBBaseExecutor>>resume:withToken:value: - the
+one continuation path - and by a condition wait whose condition already holds when it is
+entered (FBWaitForConditionBehavior). There is no tie: the engine is single-threaded, the
+first impulse wins, and a second one finds its wait gone and is silent, like a double
+submit.
+
+Every branch has to lead into a node that parks (FBNode>>parksOnEntry); a branch that
+would run at once cannot take part in a choice, since steps run to rest before any impulse
+from outside arrives, and my behavior refuses such a model. I open a pass like every
+split (FBSplitExecution), which is what remembers whether I was decided and where my
+branches began (#origin).
+"
+Class {
+	#name : 'FBDeferredChoice',
+	#superclass : 'FBSplitGateway',
+	#category : 'FluxBase-Core-Model',
+	#package : 'FluxBase-Core',
+	#tag : 'Model'
+}
+
+{ #category : 'instance creation' }
+FBDeferredChoice class >> typeName [
+	^ #deferredChoice
+]
+
+{ #category : 'accessing' }
+FBDeferredChoice >> defaultBehaviorClass [
+	^ FBDeferredChoiceBehavior
+]
+
+{ #category : 'testing' }
+FBDeferredChoice >> isDeferredChoice [
+	^ true
+]

source/FluxBase-Core/FBDeferredChoiceBehavior.class.st

@@ -0,0 +1,18 @@
+Class {
+	#name : 'FBDeferredChoiceBehavior',
+	#superclass : 'FBAndSplitBehavior',
+	#category : 'FluxBase-Core-Behavior',
+	#package : 'FluxBase-Core',
+	#tag : 'Behavior'
+}
+
+{ #category : 'execution' }
+FBDeferredChoiceBehavior >> execute [
+	"a deferred choice starts every branch like an AND-split does, and each has to park
+	at once: a branch that ran on would decide nothing and let the others go on with it"
+	activation node outgoing do: [ :transition |
+		transition target parksOnEntry ifFalse: [
+			FBError signal: 'every branch of a deferred choice has to lead into a node that waits; ',
+				transition target printString, ' does not' ] ].
+	super execute
+]