From 89c6e025923f48db19fb847df3805864ba8ef30c Mon Sep 17 00:00:00 2001 From: Norbert Hartl Date: Thu, 10 Sep 2026 12:11:10 +0200 Subject: [PATCH] A resume drops only the resumed strand's address Two iterations of a parallel scope park the very same transition or node, each registered under its own token. Resuming one dropped every registry entry naming the suspension, so the other iteration's address was gone while its token still waited: its signal answered a silent false. One flow form per list entry, and the first submit spent the other links. FBBaseExecutor>>resume:withToken:value: now takes the token first and drops the entry of that token only (FBFlowInstance>> dropWaitingEntriesMatching:token:). A resume that ends up without a token - a suspension parked before tokens were kept - still drops every entry of the suspension, there being nothing to tell them apart by. Co-Authored-By: Claude Fable 5.1 --- .../FBWaitingRegistryTest.class.st | 37 +++++++++++++++++++ source/FluxBase-Core/FBBaseExecutor.class.st | 23 ++++++++---- source/FluxBase-Core/FBFlowInstance.class.st | 15 ++++++-- 3 files changed, 64 insertions(+), 11 deletions(-) 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 ] ]