diff --git a/source/FluxBase-Core-Tests/FBWaitingRegistryTest.class.st b/source/FluxBase-Core-Tests/FBWaitingRegistryTest.class.st index 57b02a5..28d0dae 100644 --- a/source/FluxBase-Core-Tests/FBWaitingRegistryTest.class.st +++ b/source/FluxBase-Core-Tests/FBWaitingRegistryTest.class.st @@ -875,3 +875,40 @@ FBWaitingRegistryTest >> testAReservationOfAFinishedIterationFallsWithTheIterati self assert: (instance waitingEntryFor: handedOut first correlationId) isNil. self assert: (instance waitingEntryFor: handedOut second correlationId) identicalTo: handedOut second ] + +{ #category : 'tests' } +FBWaitingRegistryTest >> testResumingOneOfTwoIterationsWaitingAtOneNodeLeavesTheOtherItsAddress [ + "two iterations of a parallel scope wait at the very same node, each registered under + its own token. Signalling the first has to resume the first alone: the second keeps + its address and is resumed by its own signal. The resume used to drop every entry + naming the suspension, so the second link answered a silent false while its token + still waited - one flow form per list entry, and the first submit spent the other + links. The transition-shaped wait (a trigger target) goes the same way through + FBBaseExecutor>>resume:withToken:value:" + | start fork wait flow instance first second | + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + wait := fork addOutgoingNode: (FBExternalWait new condition: [ :each | false ]; yourself). + (wait addOutgoingNode: (FBParallelJoinNode named: 'scope join')) + addOutgoingNode: FBEndEvent new. + flow := FBFlow new start: start. + instance := flow newInstance. + instance executeWith: #( 10 20 ). + self assert: instance openWaitingEntries size equals: 2. + first := instance openWaitingEntries first. + second := instance openWaitingEntries second. + self assert: first suspension identicalTo: second suspension. + self deny: first token identicalTo: second token. + + self assert: (instance signal: first correlationId with: 1). + + "the second iteration still waits, under its own address and its own token" + self assert: instance openWaitingEntries asArray equals: { second }. + self assert: (instance waitingEntryFor: second correlationId) identicalTo: second. + self assert: instance suspended size equals: 1. + + self assert: (instance signal: second correlationId with: 2). + instance execute. + self assert: instance state equals: #done. + self assert: instance openWaitingEntries isEmpty +] diff --git a/source/FluxBase-Core/FBBaseExecutor.class.st b/source/FluxBase-Core/FBBaseExecutor.class.st index 5de1db5..181c6fa 100644 --- a/source/FluxBase-Core/FBBaseExecutor.class.st +++ b/source/FluxBase-Core/FBBaseExecutor.class.st @@ -303,10 +303,18 @@ FBBaseExecutor >> resume: aSuspension withToken: aToken value: aValue [ fresh activation, a node that parked inside its own execution still has one. So FBTransition and FBNode each implement #resumeIn:withToken:value:. - The registry entries naming aSuspension go first: a wait that is over has no address - any more, whichever way it was continued. Asked of my own instance's registry, not - through #openWaitingEntries - the entry was registered on the instance whose executor - holds the suspension (FBBaseActivation>>waitForNode:withToken:), and #openWaitingEntries + The registry entry of the wait that goes on falls with it: a wait that is over has no + address any more, whichever way it was continued. The entry of that wait, not every + entry naming aSuspension - two iterations of a parallel scope park the very same + transition or node, each with an entry on its own token, and the first one resumed + must not take the second one's address with it (that is what it did until 2026-09-10: + the second link answered a silent false while its token still waited). Which entry is + the wait's is told by the token, so the token is taken first and the entry dropped + after it, by token; only a resume that ends up with no token at all - a suspension + parked before tokens were kept - drops every entry of the suspension, as there is + nothing to tell them apart by. Asked of my own instance's registry, not through + #openWaitingEntries - the entry was registered on the instance whose executor holds + the suspension (FBBaseActivation>>waitForNode:withToken:), and #openWaitingEntries answers openness, which is momentarily false in the middle of a resume: between removing the last suspension and activating what it goes on with, nothing is pending and nothing suspended, and #isDone is true. A reservation is not an entry of the @@ -326,12 +334,13 @@ FBBaseExecutor >> resume: aSuspension withToken: aToken value: aValue [ submitted on it" | token | - flowInstance dropWaitingEntriesMatching: aSuspension. - suspended remove: aSuspension. token := self takeTokenForSuspended: aSuspension satisfying: [ :each | aToken isNil or: [ each == aToken ] ]. - ^ aSuspension resumeIn: self withToken: (aToken ifNil: [ token ]) value: aValue + token := aToken ifNil: [ token ]. + flowInstance dropWaitingEntriesMatching: aSuspension token: token. + suspended remove: aSuspension. + ^ aSuspension resumeIn: self withToken: token value: aValue ] { #category : 'as yet unclassified' } diff --git a/source/FluxBase-Core/FBFlowInstance.class.st b/source/FluxBase-Core/FBFlowInstance.class.st index a189c85..26883a8 100644 --- a/source/FluxBase-Core/FBFlowInstance.class.st +++ b/source/FluxBase-Core/FBFlowInstance.class.st @@ -188,9 +188,14 @@ FBFlowInstance >> descendFrom: aToken [ ] { #category : 'executing' } -FBFlowInstance >> dropWaitingEntriesMatching: aSuspension [ - "forget every address of mine that names aSuspension - the step a continuation makes - before it takes the suspension out, see FBBaseExecutor>>resume:withToken:value:. +FBFlowInstance >> dropWaitingEntriesMatching: aSuspension token: aToken [ + "forget the address of mine that names the wait going on: aSuspension, parked on + aToken - the step a continuation makes as it takes the suspension out, see + FBBaseExecutor>>resume:withToken:value:. The token tells the wait's entry apart from + the entries of other strands parked at the same suspension - two iterations of a + parallel scope wait at one transition or node, one entry each - which keep theirs. + With no token to go by (nil: a suspension parked before tokens were kept) every entry + of the suspension falls, there being nothing to tell them apart. Mine only, by design: an entry is registered on the instance whose executor holds the suspension (FBBaseActivation>>waitForNode:withToken:), and the executor resuming is @@ -200,7 +205,9 @@ FBFlowInstance >> dropWaitingEntriesMatching: aSuspension [ registered here - then there is nothing to forget and nothing is created" (waitingRegistry ifNil: [ ^ self ]) entries - select: [ :each | each matchesSuspension: aSuspension ] + select: [ :each | + (each matchesSuspension: aSuspension) + and: [ aToken isNil or: [ each token == aToken ] ] ] thenDo: [ :each | waitingRegistry takeEntryFor: each correlationId ] ]