Skip to content

[core][move semantics][8/n] Streaming generator support for _consume_once - #67223

Open
karticam wants to merge 1 commit into
karticam/owner-side-flowfrom
karticam/streaming-generator-consume-once
Open

karticam wants to merge 1 commit into
karticam/owner-side-flowfrom
karticam/streaming-generator-consume-once

Conversation

@karticam

@karticam karticam commented Oct 21, 2026 •

Copy link
Copy Markdown
Contributor

This is PR 8 of a series of PRs to implement move semantics for plasma objects in core.

The complete design doc is here: https://docs.google.com/document/d/1ZJhgRmwVyKfNCR21CWf6xMcc0B6HJwSPQsfebTRvnnE/edit?tab=t.0#heading=h.7wcod1f9afpg

Until now _consume_once=True was rejected on streaming generator actor methods (#66620). With this PR it is accepted, and every object the generator yields can be moved to exactly one consumer, the same way a plain return can.

Changes

  1. The generator object carries a flag. A streaming generator task has one return: the generator object. The objects it yields don't exist yet when the task is submitted. So AddPendingTask doesn't make the generator object MOVABLE. Instead it sets a new Reference field, yields_consume_once_, on it when a streaming generator object is created with _consume_once method option, through a new AddOwnedObject parameter. The generator object itself stays NOT_MOVABLE.
  2. Each yielded object is created MOVABLE. OwnDynamicStreamingTaskReturnRef passes MOVABLE as the initial move state when the generator object has the flag. Every creation path goes through it.
  3. The rejection is lifted. ActorMethod._remote no longer rejects streaming generator methods.

Nothing else changes for the generated object refs. Everything in the previous PRs apply to it.

Every yield of the generator becomes consume-once, including small ones kept in the owner's memory store.

A generator that runs again, after an actor restart or for lineage reconstruction, reports every item again, and the owner records the new run's node as each item's primary. For an item that was already moved, the consumer's node keeps its copy too, so the item has two copies until it goes out of scope. Master does the same today with the old copy of any item a generator reports again, so this PR doesn't change it.

@karticam
karticam added this pull request to stack #66768 October 21, 2026 00:52
@karticam karticam added core Issues that should be addressed in Ray Core go add ONLY when ready to merge, run all tests labels Oct 21, 2026
@karticam
karticam marked this pull request as ready for review October 21, 2026 02:06
@karticam
karticam requested a review from a team as a code owner October 21, 2026 02:06
@karticam karticam changed the title [core][move semantics][8/n] Streaming generator actor methods accept _consume_once [core][move semantics][8/n] Streaming generator support for _consume_once Oct 21, 2026

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Code Review

This pull request implements support for consume_once on streaming generator actor methods in Ray. Previously, _consume_once=True was not supported for streaming generators. With these changes, each object yielded by a streaming generator with consume_once enabled is created as MOVABLE (allowing it to be moved to the consumer's node and freed from the producer's node), while the generator object itself remains NOT_MOVABLE. The implementation updates the C++ ReferenceCounter and TaskManager to track and propagate the yields_consume_once property, and adds comprehensive unit and integration tests in both Python and C++ to verify the behavior. There are no review comments to address, so no further feedback is provided.

@cursor cursor Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Fix All in Cursor

❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.

Reviewed by Cursor Bugbot for commit 063560a. Configure here.

Comment thread src/ray/core_worker/reference_counter.h Resolved
@karticam
karticam force-pushed the karticam/streaming-generator-consume-once branch from 063560a to 700eaa8 Compare October 21, 2026 05:14
@karticam
karticam force-pushed the karticam/streaming-generator-consume-once branch from 700eaa8 to 1245c19 Compare October 21, 2026 08:35
@karticam
karticam force-pushed the karticam/streaming-generator-consume-once branch from 1245c19 to 1bb2992 Compare November 3, 2026 03:00
@karticam
karticam force-pushed the karticam/streaming-generator-consume-once branch from 1bb2992 to 2e9f91e Compare November 3, 2026 03:07
…ble objects

Signed-off-by: kartica.modi <kartica.modi@anyscale.com>
@karticam
karticam force-pushed the karticam/streaming-generator-consume-once branch from 2e9f91e to 62e98dc Compare November 14, 2026 02:36

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Issues that should be addressed in Ray Core go add ONLY when ready to merge, run all tests

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants