Skip to content

Scope collective ID pool per query - #24547

Open
Jesse-Bakker wants to merge 4 commits into
NVIDIA:mainfrom
Jesse-Bakker:collective-id-contextvar
Open

Jesse-Bakker wants to merge 4 commits into
NVIDIA:mainfrom
Jesse-Bakker:collective-id-contextvar

Conversation

@Jesse-Bakker

Copy link
Copy Markdown

Description

The pool of vacant collective IDs was process-global. The global pool's
state depends on every query the process has run or is running. So, a
process that first runs a query as part of one cluster and then as part
of another can allocate different collective IDs.
I see two possible solutions:

  • Return the collective IDs to the global pool in reverse order. This
    will make sequential queries work (assuming they always return all their
    IDs properly through the context manager)
  • Scope the collective IDs per query through a ContextVar, making their
    ID pools independent

I went for the latter approach, as it seems more robust.

Checklist

  • I am familiar with the Contributing Guidelines.
  • New or existing tests cover these changes.
  • The documentation is up to date with these changes.

@copy-pr-bot

copy-pr-bot Bot commented Oct 12, 2026

Copy link
Copy Markdown

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

@github-actions github-actions Bot added Python Affects Python cuDF API. cudf-polars Issues specific to cudf-polars labels Oct 12, 2026
@Jesse-Bakker
Jesse-Bakker force-pushed the collective-id-contextvar branch from afdf8e9 to fe1d521 Compare October 12, 2026 09:05
@Jesse-Bakker
Jesse-Bakker marked this pull request as ready for review October 12, 2026 09:05
@Jesse-Bakker
Jesse-Bakker requested a review from a team as a code owner October 12, 2026 09:05
@coderabbitai

coderabbitai Bot commented Oct 12, 2026 •

Copy link
Copy Markdown

Review in Change Stack →

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info
⚙️ Run configuration
  • Configuration used: Repository: NVIDIA/cudf/.coderabbit.yaml
  • Review profile: CHILL
  • Plan: Enterprise
  • Run ID: fad455da-e5d4-46f9-b7d9-268d9d1a61ea
📥 Commits

Reviewing files that changed from the base of the PR and between 45b76fc and 94dc15c.

📒 Files selected for processing: 3
  • docs/cudf/source/cudf_polars/api.md
  • python/cudf_polars/cudf_polars/engine/core.py
  • python/cudf_polars/cudf_polars/streaming/actor_graph/collectives/common.py

Included review availability: This review used your included allowance. Your plan provides up to 12 included reviews per hour; 11 remain after this review.

📝 Summary

Summary by CodeRabbit

  • Bug Fixes
    • Queries running concurrently now receive isolated collective operation IDs, preventing reservations from one query or thread from affecting another.
    • ID allocation remains consistent when query scopes are nested or restored, preserving the expected sequence of IDs.
  • Documentation
    • Added the collective ID scope helper to the SPMD reference documentation.

Collective ID allocation now uses context-local pools instead of a process-global pool. collective_id_scope() creates and restores a pool. evaluate_on_rank runs its evaluation within that scope. Tests cover nested scopes, separate threads, and rank evaluation.

Changes: Query-scoped collective IDs

Layer / File(s) Summary
Context-local pool and reservation behavior
python/cudf_polars/cudf_polars/streaming/actor_graph/collectives/*, docs/cudf/source/cudf_polars/api.md
Adds context-local pool selection and collective_id_scope(). ReserveOpIDs and reserve_op_id() acquire and release IDs through the selected pool. The package exports the scope, and the API documentation lists it.
Rank evaluation scope and validation
python/cudf_polars/cudf_polars/engine/core.py, python/cudf_polars/tests/streaming/test_collective_ids.py
evaluate_on_rank enters a fresh scope and delegates the existing evaluation pipeline to _evaluate_on_rank. Tests cover independent, nested, and threaded scopes, and the evaluation entry point.

Priority: ➖ Normal

Estimated code review effort: 3 (Moderate) | ~20 minutes

Change: Bug fix

Suggested reviewers: wence-, rjzamora

Merge Risk: 🟡 Moderate · up to 94dc1

Concurrent queries on the same SPMD engine may interfere with one another’s collective operations. Resolve or explicitly accept that risk before merging.

Pre-merge checks: Passed 4 · Failed 1

Failed Failed checks: 1 warning

Check name Status Explanation Resolution
Docstring Coverage Warning Docstring coverage is 36.84% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 19 functions across 4 files. (1 skipped: … Write docstrings for the functions missing them to satisfy the coverage threshold.
Passed Passed checks: 4 passed
Check name Status Explanation
Title check Passed The title clearly and concisely describes the main change: scoping the collective ID pool per query.
Description check Passed The description explains the process-global pool problem, the ContextVar-based solution, and the related tests and documentation updates.
Linked Issues check Passed Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check Passed Check skipped because no linked issues were found for this pull request.
Full details: Docstring Coverage

Explanation

Docstring coverage is 36.84% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 19 functions across 4 files. (1 skipped: 1 unsupported.)

  • Fix all pre-merge checks with AI
✨ Finishing Touches
🧪 Generate unit tests (beta)
  • Create a new PR
  • Autofix · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts

Comment @coderabbitai help to get the list of available commands.

@wence-

wence- commented Oct 12, 2026

Copy link
Copy Markdown
Contributor

/ok to test fe1d521

@wence- wence- added improvement Improvement / enhancement to an existing function non-breaking Non-breaking change labels Oct 12, 2026

@wence- wence- 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.

Please run pre-commit to fix the minor CI style errors.

Also a question about the safety of using the same communicator with two concurrent queries.

Comment on lines +927 to +939
# Each query reserves its collective IDs from its own pool, so that the
# IDs agree across ranks even if this process concurrently takes part in
# other queries (possibly on other communicators).
with collective_id_scope():
return _evaluate_on_rank(
ctx,
comm,
py_executor,
ir,
config_options,
quent_query_worker_state=quent_query_worker_state,
query_id=query_id,
)

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.

question: What would happen if you ran two queries on the same communicator concurrently with this setup? Because each query generates its own id scope, I think that you would get overlapping ids, and message passing could get mixed up.

Should we key the id scope by the communicator?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Running concurrent queries on the same communicator would probably fail miserably, whereas before this change, it would probably work as long as the queries are started in the same order on all ranks. Keying the scope by communicator would work for that, but again, only if the queries are started in the same order on all ranks. It should be possible for a caller to sequence that using a queue and a barrier though.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Hmm, no, it also depends on release timing and each rank can time that slightly differently, so that can still race and cause mismatched collective ids I think.

The pool of vacant collective IDs was process-global. The global pool's
state depends on every query the process has run or is running. So, a
process that first runs a query as part of one cluster and then as part
of another can allocate different collective IDs.
I see two possible solutions:
 - Return the collective IDs to the global pool in reverse order. This
   will make sequential queries work (assuming they always return all their
   IDs properly through the context manager)
 - Scope the collective IDs per query through a ContextVar, making their
   ID pools independent

I went for the latter approach, as it seems more robust.
@Jesse-Bakker
Jesse-Bakker force-pushed the collective-id-contextvar branch from fe1d521 to 45b76fc Compare October 12, 2026 16:54
@TomAugspurger

Copy link
Copy Markdown
Contributor

/ok to test 45b76fc

@TomAugspurger

Copy link
Copy Markdown
Contributor

/ok to test 94dc15c

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

Labels

cudf-polars Issues specific to cudf-polars improvement Improvement / enhancement to an existing function non-breaking Non-breaking change Python Affects Python cuDF API.

Projects

Status: Todo

Development

Successfully merging this pull request may close these issues.

3 participants