Skip to content

stream: optimize gather hot path and polish for upstream#2848

Draft
He-Pin wants to merge 1 commit into
apache:mainfrom
He-Pin:hepin-gather-statefulmap-coverage-v2
Draft

stream: optimize gather hot path and polish for upstream#2848
He-Pin wants to merge 1 commit into
apache:mainfrom
He-Pin:hepin-gather-statefulmap-coverage-v2

Conversation

@He-Pin

@He-Pin He-Pin commented Apr 6, 2026

Copy link
Copy Markdown
Member

Summary

This PR applies deep code review and optimizations on top of the gather operator implementation. All changes are targeted at making the contribution meet Apache Pekko CONTRIBUTING standards.

Hot path optimizations (Ops.scala)

  • Eliminate hasCallbackFirst boolean field — use callbackFirst eq null as sentinel, saving 2 writes + 1 read per element
  • Inline pushCallbackSingle into onPushSingle — remove 1 virtual dispatch per element on the common path
  • Extract cold afterPushFinalAction() — so the JIT optimises the hot branch independently
  • Remove hadContext parameter from maybeRunFinalAction/pushPending* — check needInvokeOnCompleteCallback directly
  • Reset multiMode = false after overflow is fully drained, so subsequent single-output calls return to the fast path

Documentation & API polish

  • Fix @since version: unify to 1.3.0 across all DSL files (was inconsistently 2.0.0)
  • Fix Java docs examples: use collector.push() instead of collector.apply() (idiomatic API)
  • Rewrite gather.md to remove internal PR references and add third example (distinctUntilChanged)
  • Improve Scala/Java docs examples with bufferUntilChanged and distinctUntilChanged patterns
  • Remove internal Chinese evaluation doc (not appropriate for upstream)

Test & build fixes

  • Fix distinctUntilChanged test: remove dead lastElement = Some(elem) in the duplicate case
  • Fix missing Keep and Flow imports in FlowGatherSpec
  • Use explicit imports in ZipWithIndexBenchmark (avoid wildcard import)

Test plan

  • All 30 FlowGatherSpec tests pass
  • Java gather tests pass
  • stream/compile succeeds
  • Commit message follows Apache Pekko convention (stream: ...)

@He-Pin He-Pin marked this pull request as draft April 6, 2026 20:16
pjfanning
pjfanning requested changes Apr 7, 2026
View reviewed changes
Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
*
* '''Cancels when''' downstream cancels
*
* @since 1.3.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

incorrect change

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/Gather.scala
* The collector is only valid while the current [[Gatherer]] callback is running.
* Emitted elements MUST NOT be `null`.
*
* @since 1.3.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

2.0.0

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/Gather.scala
* A new gatherer instance is created for each materialization and on each supervision restart.
* It can keep mutable state in fields or via captured variables.
*
* @since 1.3.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

2.0.0

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/Gather.scala
* the `onComplete` callback.
*
* @param f the one-to-one transformation function
* @since 1.3.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

all the 1.3.0 in this class or elsewhere in this PR should be 2.0.0

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
/**
* Create a `Source` from an `Optional` value, emitting the value if it is present.
*
* @since 1.3.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

incorrect change

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
*
* '''Cancels when''' downstream cancels
*
* @since 1.3.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

all of these 1.3.0s are wrong - should be 2.0.0

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
* '''Cancels when''' downstream cancels
*
* @since 1.3.0
* @since 2.0.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

incorrect change

Comment thread stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
* '''Cancels when''' downstream cancels
*
* @since 1.3.0
* @since 2.0.0

@pjfanning pjfanning Apr 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why are all the mapOptions being changed? Change back to 1.3.0

@pjfanning pjfanning added this to the 2.0.0-M2 milestone Apr 7, 2026
@pjfanning pjfanning modified the milestones: 2.0.0-M2, 2.0.0-M3 Apr 17, 2026
@pjfanning pjfanning modified the milestones: 2.0.0-M3, 2.0.0-maybes May 29, 2026
He-Pin added a commit that referenced this pull request Jun 15, 2026
@He-Pin
fix: correct relative path to actors.md in gather operator docs
e23a4fb 
Motivation:
The Paradox doc build fails with "Unknown page [../../actors.md]"
because gather.md is three directory levels below the paradox root.

Modification:
Change ../../actors.md to ../../../actors.md, matching the pattern
used in neighboring ask.md.

Result:
Paradox doc build succeeds.

Tests:
Not run - docs only (path fix verified by comparing ask.md in same directory)

References:
Refs #2848
He-Pin added a commit to He-Pin/incubator-pekko that referenced this pull request Jun 15, 2026
@He-Pin
fix: correct relative path to actors.md in gather operator docs
599a39f 
Motivation:
The Paradox doc build fails with "Unknown page [../../actors.md]"
because gather.md is three directory levels below the paradox root.

Modification:
Change ../../actors.md to ../../../actors.md, matching the pattern
used in neighboring ask.md.

Result:
Paradox doc build succeeds.

Tests:
Not run - docs only (path fix verified by comparing ask.md in same directory)

References:
Refs apache#2848
@He-Pin He-Pin force-pushed the hepin-gather-statefulmap-coverage-v2 branch from 93f35dc to 599a39f Compare June 15, 2026 04:07
@He-Pin
stream: optimize gather hot path and polish for upstream
da7b38e 
Motivation:
The gather operator's hot path had unnecessary overhead and its docs
needed refinement for upstream contribution quality.

Modification:
- Optimize gather hot path: eliminate boolean field, inline callbacks
- Add comprehensive test spec (849 lines) and JMH benchmarks
- Refine docs, @SInCE markers, and API annotations
- Apply scalafmt and javafmt formatting
- Address deep review findings

Result:
Improved gather performance and production-quality test coverage.

Tests:
- sbt "stream / Test / testOnly *GatherSpec"

References:
Refs apache#2848
@He-Pin He-Pin force-pushed the hepin-gather-statefulmap-coverage-v2 branch from 599a39f to da7b38e Compare July 8, 2026 18:21
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@pjfanning pjfanning pjfanning requested changes

Requested changes must be addressed to merge this pull request.

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

2.0.0-maybes

Development

Successfully merging this pull request may close these issues.

2 participants

@He-Pin @pjfanning