He-Pin commented on PR #3461: URL: https://github.com/apache/pekko/pull/3461#issuecomment-5407057410
Thanks @pjfanning for the careful review — both concerns are addressed in 56783632b8. **1. activeStage save/restore — done, with a test.** You were right to flag it. One interesting empirical detail: in the minimal topology the clobbering self-heals, because the last chased event of execute() is the pull back to the source port, and processPull sets activeStage to that port's owner — i.e. this logic again. But with a longer fused upstream chain and an event limit that is too small, execute() stops mid-chain and activeStage is left pointing at an intermediate stage, so any activeStage-dependent helper used by the next handler invocation (e.g. the default InHandler/OutHandler termination methods) would act on the wrong stage. Fixed by saving/restoring activeStage around execute(), and both the longer-upstream-chain test and a dedicated attribution test assert `currentInterpreter.activeStage eq logic` for every dispatched message. **2. The limit is now justified — and you were right that 2 was not enough.** I added exactly the test you suggested (three map stages), in both directions: - *Longer upstream chain* (maps between source and boundary): **fails with limit 2, passes with the new limit** — this is the case execute() can and must cover, since the whole push/pull round trip stays inside one shell. - *Longer downstream chain* (maps after the boundary): **fails with any limit** (verified up to 1024). The reason is structural, not a matter of budget: demand crosses the async boundary via BatchingActorInputBoundary, which requests upstream in asynchronous batches of size/2. While the drain loop occupies the source shell's actor thread, no RequestMore can be processed, so bursts larger than the outstanding demand batch can drop — which is the documented at-most-once semantics of the getStageActor + isAvailable pattern. The downstream-chain test therefore bursts within the initial demand batch (8 < 16) and documents this boundary. The magic 2 is now `LazyDispatch.ReplenishEventLimit = 16` with a comment explaining the cost model (a few event hops per fused stage per direction; leftovers stay queued for the interpreter's outer run loop). I also fixed a latent startup race in the original test: messages were sent right after the stage actor materialized, but before initial demand had necessarily propagated to the source port. The tests now wait for the first onPull before bursting. Tests: FusingSpec (4 new/updated tests, 34/34 with the whole suite), StageActorRefSpec, ActorRefSourceSpec, StreamRefsSpec all green; stream MiMa clean. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
