From e3ff899a10ffe7e497129e1d89ae4980881d8657 Mon Sep 17 00:00:00 2001 From: Norbert Hartl Date: Thu, 10 Sep 2026 14:52:13 +0200 Subject: [PATCH] A parallel scope's step asks and records like the root executor's FBParallelExecutor>>step now does the two things FBFlowExecutor>>step does around a transition: it asks the instance first (FBFlowInstance>>aboutToTransition:, where ApptiveBase counts its rate limits - a loop used to bypass them) and records an FBTransitionFollowed (a scope's log had no transitions at all). Both only for a transition actually followed, since inside a scope the branch condition is decided at the step; a refusal raised from the hook leaves the work item queued, as in the root executor. FBProbingFlowInstance is a test double that records what it is asked and can refuse. Co-Authored-By: Claude Fable 5.1 --- .../FBParallelExecutorTest.class.st | 60 +++++++++++++++++++ .../FBProbingFlowInstance.class.st | 32 ++++++++++ .../FluxBase-Core/FBParallelExecutor.class.st | 16 +++-- 3 files changed, 104 insertions(+), 4 deletions(-) create mode 100644 source/FluxBase-Core-Tests/FBProbingFlowInstance.class.st diff --git a/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st b/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st index 31a089b..33bef61 100644 --- a/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st +++ b/source/FluxBase-Core-Tests/FBParallelExecutorTest.class.st @@ -392,3 +392,63 @@ FBParallelExecutorTest >> testTheScopeJoinIsFoundPastAnInnerLoop [ self assert: instance state equals: #done. self assert: seen asArray equals: { { #( 2 4 ). nil } } ] + +{ #category : 'tests' } +FBParallelExecutorTest >> testEveryIterationRecordsTheTransitionItFollows [ + "a transition followed inside a parallel scope is recorded like one followed by the + root executor - once per iteration that follows it, not for an iteration its + condition holds back. The scope's log used to have no transitions at all" + | seen instance body followed | + seen := OrderedCollection new. + instance := self loopOver: #( 1 2 3 ) bodyWhen: [ :x | x < 3 ] into: seen. + body := instance flow nodes detect: [ :each | each name = 'body' ]. + followed := instance events select: [ :each | + (each isKindOf: FBTransitionFollowed) and: [ each transition target == body ] ]. + self assert: followed size equals: 2. + self assert: (followed collect: [ :each | each value ]) asArray equals: #( 1 2 ). + self assert: instance state equals: #done +] + +{ #category : 'tests' } +FBParallelExecutorTest >> testTheInstanceIsAskedBeforeEachIterationsTransition [ + "FBFlowInstance>>aboutToTransition: is asked once per iteration that follows the + body's transition - what ApptiveBase counts its rate limits with, and what a loop + used to bypass" + | start fork body instance flow | + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + body := (FBPluggableTask named: 'body') block: [ :act | act value: act argument ]. + fork --> (body when: [ :x | x < 3 ]). + (body addOutgoingNode: (FBParallelJoinNode named: 'scope join')) addOutgoingNode: FBEndEvent new. + flow := FBFlow new start: start. + instance := FBProbingFlowInstance new flow: flow. + instance executeWith: #( 1 2 3 ). + self assert: instance state equals: #done. + self assert: (instance asked count: [ :each | each target == body ]) equals: 2 +] + +{ #category : 'tests' } +FBParallelExecutorTest >> testARefusedTransitionLeavesTheIterationsWorkQueued [ + "the hook refuses the second iteration's transition: the error comes out of the + step, the work item stays queued as it does in the root executor, and nothing of + that iteration ran" + | start fork body instance flow ran | + ran := OrderedCollection new. + start := FBStartEvent new. + fork := start addOutgoingNode: (FBParallelNode named: 'fork'). + body := (FBPluggableTask named: 'body') block: [ :act | ran add: act argument. act value: act argument ]. + fork --> body. + (body addOutgoingNode: (FBParallelJoinNode named: 'scope join')) addOutgoingNode: FBEndEvent new. + flow := FBFlow new start: start. + instance := FBProbingFlowInstance new flow: flow. + instance refuseWith: [ :transition | + (transition target == body and: [ instance executor currentIteration = 2 ]) + ifTrue: [ Error signal: 'refused' ] ]. + instance initialValue: #( 1 2 3 ). + self should: [ instance execute ] raise: Error. + self assert: ran asArray equals: #( 1 ). + self assert: instance executor pending size equals: 1. + self assert: instance executor pending first transition target identicalTo: body. + self deny: (instance events anySatisfy: [ :each | + (each isKindOf: FBTransitionFollowed) and: [ each transition target == body and: [ each value = 2 ] ] ]) +] diff --git a/source/FluxBase-Core-Tests/FBProbingFlowInstance.class.st b/source/FluxBase-Core-Tests/FBProbingFlowInstance.class.st new file mode 100644 index 0000000..f091fb6 --- /dev/null +++ b/source/FluxBase-Core-Tests/FBProbingFlowInstance.class.st @@ -0,0 +1,32 @@ +" +A flow instance for tests that records what the engine asks it before a transition +(FBFlowInstance>>aboutToTransition:) and can refuse one, the way ApptiveBase's instance +refuses a transition whose rate limit is used up. +" +Class { + #name : 'FBProbingFlowInstance', + #superclass : 'FBFlowInstance', + #instVars : [ + 'asked', + 'refusal' + ], + #category : 'FluxBase-Core-Tests', + #package : 'FluxBase-Core-Tests' +} + +{ #category : 'hooks' } +FBProbingFlowInstance >> aboutToTransition: aTransition [ + self asked add: aTransition. + refusal ifNotNil: [ :block | block value: aTransition ] +] + +{ #category : 'accessing' } +FBProbingFlowInstance >> asked [ + ^ asked ifNil: [ asked := OrderedCollection new ] +] + +{ #category : 'accessing' } +FBProbingFlowInstance >> refuseWith: aBlock [ + "aBlock gets every transition asked about and may raise" + refusal := aBlock +] diff --git a/source/FluxBase-Core/FBParallelExecutor.class.st b/source/FluxBase-Core/FBParallelExecutor.class.st index ea51f61..7a4e89c 100644 --- a/source/FluxBase-Core/FBParallelExecutor.class.st +++ b/source/FluxBase-Core/FBParallelExecutor.class.st @@ -390,16 +390,24 @@ FBParallelExecutor >> shouldFollowTransition: transition with: value [ { #category : 'services' } FBParallelExecutor >> step [ - | workItem transition lastValue | - workItem := pending removeFirst. + | workItem transition lastValue followed | + workItem := pending first. transition := workItem transition. "the iteration this step belongs to, read off the work item's token: a body seed (no token yet, see #scopeToken) is the newest iteration's, everything else carries the token of the iteration that queued it" current := workItem token ifNil: [ index ] ifNotNil: [ :token | self iterationOf: token ]. lastValue := transition source lastValueIn: self. - - (self shouldFollowTransition: transition with: lastValue) ifTrue: [ + followed := self shouldFollowTransition: transition with: lastValue. + "the same two things FBFlowExecutor>>step does around a transition: the instance is + asked before it is taken - ApptiveBase counts its rate limits there, and a refusal + raised from the hook leaves the work item queued, as it does in the root executor - + and the transition is recorded. Only for a transition actually followed: inside a + scope the branch condition is decided here, not at queueing time" + followed ifTrue: [ flowInstance aboutToTransition: transition ]. + pending removeFirst. + followed ifTrue: [ + flowInstance addEvent: (FBTransitionFollowed new transition: transition; value: lastValue). self activateTransition: transition with: lastValue