[pull] master from ray-project:master - #1192
Merged
Merged
Conversation
`TestEnvRunnerCallbacks.test_callbacks_on_sample_timesteps` and the
sibling `test_callbacks_on_sample_rollout` asserted
len(on_episode_start_calls) == sum(is_done) + num_envs
which is off by one whenever an episode terminates exactly on the final
sampled timestep. In that case the env runner eagerly creates the
replacement episode (firing `on_episode_created`) but does not start it:
`on_episode_start` fires only once the replacement receives its reset
observation on the *next* sampled timestep, which never comes within
this `sample()` call. With unseeded `random_actions=True`, CartPole
episode lengths are stochastic, so this boundary is hit intermittently
-- the source of the flakiness observed on master.
Signed-off-by: Artur Niederfahrenhorst <artur@anyscale.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
### Description Deprecate `ray_remote_args_fn` across the public Ray Data transformation APIs. - Emits a `RayDeprecationWarning` when `ray_remote_args_fn` is passed to `Dataset.map`, `map_batches`, `flat_map`, `filter`, or `GroupedData.map_groups` - Marks the argument for removal in Ray 2.64. - Renames `_map_batches_without_batch_size_validation` to the developer API `map_batches_internal`. - Routes internal Ray Data LLM stages through `map_batches_internal` so they don't trigger a user-facing deprecation warning. - Adds a warning to the documentation section that recommends creating placement groups through `ray_remote_args_fn`. --- Follow-up to #64631 --------- Signed-off-by: Alex Chien <alexchien130@gmail.com> Co-authored-by: Balaji Veeramani <bveeramani@berkeley.edu>
#63858) ## Why are these changes needed? Adds `Dataset.with_columns(...)`, the plural counterpart to `with_column`, so multiple expression columns can be added or overwritten in a single projection instead of chaining several `with_column` calls (which duplicates work). `with_column` now delegates to `with_columns` for the projection path. This is the first PR of the split proposed in #63838. Per discussion there with @goutamvenkat-anyscale and @AyushKashyapII, the `UnnestExpr` / multi-output projection-engine changes are intentionally **out of scope here** and will follow in a separate PR. ## Changes - `Dataset.with_columns(exprs: Dict[str, Expr], *, compute=None, **ray_remote_args)` — builds a single `Project` with `StarExpr()` plus one aliased expression per entry. Column order follows the mapping's insertion order. - `Dataset.with_column` becomes a thin wrapper over `with_columns` for the standard projection path. - Tests in `test_with_column.py` covering multiple columns, order preservation, overwriting an existing column, the `with_column` delegation (regression guard), and input validation. ## Notes / open questions for reviewers - **`DownloadExpr` handling**: I deliberately left `with_column`'s existing `DownloadExpr` branch intact rather than generalizing download semantics into `with_columns`, to keep this PR scoped and avoid any regression. `with_columns` handles the expression-projection case only. Happy to revisit if you'd prefer download support in the plural API. - **Validation**: `with_columns` raises `ValueError` on a non-dict or empty mapping and `TypeError` on a non-`Expr` value. Flagging in case you'd prefer different behaviour (e.g. silently allowing an empty dict as a no-op). ## Checks Ran `python/ray/data/tests/test_with_column.py` locally against a python-only Ray dev build. 98 passed (the existing `with_column` suite plus the new `with_columns` cases). Verified the new tests actually execute with PyArrow ≥ 20. Closes part of #63838. --------- Signed-off-by: odncode <nnajiodera2@gmail.com>
The docstring said `use_datasource_v2` defaults to False and that "V1
remains the production path while V2 bakes". The default is True:
DEFAULT_USE_DATASOURCE_V2 = env_bool("RAY_DATA_USE_DATASOURCE_V2", True)
so `read_parquet()` has been going through DataSourceV2 by default, and
the docstring sends readers looking for an env var to enable something
that is already on.
Also note the flag's scope, which was easy to over-read:
`use_datasource_v2` is consulted in exactly one place (`read_parquet()`
in read_api.py); every other read API reads through V1 no matter how it
is set.
Docstring only, no behavior change.
> Thank you for contributing to Ray! 🚀
> Please review the [Ray Contribution
Guide](https://docs.ray.io/en/master/ray-contribute/getting-involved.html)
before opening a pull request.
---------
Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…Data context (#65103) ## Description As title ## Related issues > Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234". ## Additional information > Optional: Add implementation details, API changes, usage examples, screenshots, etc. --------- Signed-off-by: You-Cheng Lin <mses010108@gmail.com> Signed-off-by: You-Cheng Lin <106612301+owenowenisme@users.noreply.github.com> Co-authored-by: Balaji Veeramani <bveeramani@berkeley.edu>
…ting it (#65040) ## Why This is an **alternative** to #64514. Same goal — no `@PublicAPI` subpackage silently escapes the code↔docs consistency walk — but it walks `ray.data.llm` instead of allowlisting it. #64514 allowlists `ray.data.llm` on the premise that its eager `import transformers` (via the vLLM/SGLang engine processor configs) can't be imported in the docbuild image. That premise predates the `_mock_uninstalled_backends` wrapper now on `master`, which mocks exactly the backends the docbuild image lacks. Under that mock, `ray.data.llm` imports cleanly and can be a walk root like `ray.serve.llm` — so its documented surface gets checked rather than merely excluded. ## What - Add `ray.data.llm` to the data team `head_modules`. Its surface is documented in `doc/source/data/api/llm.rst`, reachable from `api.rst`'s toctree, so the walk checks the code surface against the real docs. - Run the cross-team coverage guard **inside** `_mock_uninstalled_backends` so its coverage check and import probes see the same mocks as the walks; a promoted head like `ray.data.llm` is counted as covered, not flagged. - Empty `UNWALKED_ANNOTATED_ALLOWLIST`: both known escapees (`ray.serve.llm`, `ray.data.llm`) are now walked. The list stays as the reviewed record for any future subpackage that genuinely can't be a walk root. - Decouple the guard decision-core unit tests from real module names. ## Relationship to #64514 Pick one, not both. This branch carries #64514's guard commit plus one delta commit; if this approach is preferred, the two squash into one before un-drafting. ## Checks The decisive check is **doc: check API doc consistency** — it exercises the mocked `ray.data.llm` import path this approach depends on, against the real docbuild image. Kept as a draft until that check is confirmed green. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Douglas Strodtman <douglas@anyscale.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…65096) ## Description I found it quite difficult to understand the autoscaling coordinator code without type aliasing, so adding it now. No behavior changes ## Motivation I don't think applying type aliases to everything in Ray Data is standard, just for the a) terms that can conflate with each other and b) for abstract interfaces. Declarations like so `self._subcluster_selectors: Dict[str, Optional[Dict[str, str]]] = {})` or `cluster_node_resources: Dict[Optional[str], List[ResourceDict]] = {}` and it's not clear what each str represents. And if we need to change the definition of the type for a abstract class parameter it can be done easily by modifying the type alias ## Related issues > Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234". ## Additional information > Optional: Add implementation details, API changes, usage examples, screenshots, etc. Signed-off-by: iamjustinhsu <jhsu@anyscale.com> Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
…protection to fiber stacks (#64772) # Description On Python 3.14 + Linux, every async-actor task permanently leaks ~518 KiB of live malloc (the per-task `asyncio.Task`, `concurrent.futures.Future`, Cython coroutine + scopes, and two msgpack `Packer`s with 256 KiB internal buffers). Closes #63290 ### Root cause **1. CPython 3.14 changed how it avoids stack overflow when freeing objects.** Freeing one object can recursively free many others (a dict frees its values, which free their contents, …), and each level is a nested C call. To keep that from overflowing the C stack, CPython has long had a safety mechanism (the "trashcan"): when it decides it's too deep, it doesn't free the object right away. Instead it parks the object on a per-thread *delete-later* list and drains the list once there's stack headroom again. Up to 3.13, "too deep" was a simple recursion counter. In 3.14 it's decided by comparing the actual machine **stack pointer** against the stack bounds CPython recorded for the thread when it attached (from pthreads, on Linux). **2. Ray async actors don't run task code on the thread's normal stack.** Each task executes on a small 256 KiB boost fiber stack allocated elsewhere in memory. The problem is that CPython still thinks the thread runs on its original pthread stack. So while a task runs on a fiber, every "am I near the stack limit?" check compares the fiber's stack pointer against the *pthread* stack's bounds. On Linux, fiber stacks happen to be allocated at lower addresses than the pthread stack, so CPython concludes the stack is hopelessly overflowed and parks **every** object freed during the task (including return-value serialization and end-of-task cleanup) on the delete-later list. That list is only ever drained by a later free on the same thread state at a healthy stack margin, which never happens here as the Ray thread only runs on fibers and Ray creates a fresh Python thread state per task and destroys it at task end. This means that CPython destroys a thread state **without draining its delete-later list** and the parked objects are orphaned permanently. That's the leak. Why the confusing symptoms: - `boost::make_fcontext` in the issue's flamegraphs just marks *where* the leaked allocations were made (on a fiber stack); the fiber stacks themselves are freed correctly. - macOS is unaffected only by luck: fiber stacks there land at *higher* addresses than the pthread stack, so the check passes. - 3.13 and earlier are unaffected because their trashcan uses the counter, not the stack pointer. ### Fix CPython 3.14.2 added an official API for exactly this situation: `PyUnstable_ThreadState_SetStackProtection` (python/cpython#141661) lets an embedder tell CPython "this thread is currently executing on *this* stack." We call it with the fiber's stack bounds: - at async-actor task entry in `task_execution_handler`, and - whenever a fiber resumes after `YieldCurrentFiber` (concurrent fibers share the thread state, so each must re-register its own stack). With the bounds correct, the near-limit check returns to normal behavior: objects are freed immediately, and the rare genuinely-deep free is parked and then properly drained. Implementation notes: the symbol is looked up via `dlsym`, so `_raylet` still imports on 3.14.0/3.14.1 (fix skipped there; those releases have a more severe, since-fixed stack-check bug anyway, python/cpython#141944). No-op below 3.14 (preprocessor-gated) and on Windows. Stack bounds are derived from the current stack pointer minus a conservative allowance for stack already used, so the protection errs toward triggering slightly early rather than missing an overflow. Side benefit: fibers gain real C-stack overflow protection (RecursionError) on 3.14, which they currently lack entirely (`boost::fibers::fixedsize_stack` has no guard pages). Also makes `FiberState::kStackSize` public so the anchoring uses the real fiber stack size. ## Related issue number Closes #63290. Supersedes #63284 (same diagnosis direction, but hand-rolled `_PyThreadStateImpl` offsets, a deliberate `gilstate_counter` leak that freezes non-main threads, and a crash premise that CPython 3.14.2 already fixed upstream). ## Checks - Verified with a locally built cp314 Linux (aarch64, python:3.14.6 docker) wheel: - refcount probe: **+4.00 refs/task → 0.00/task** (100 tasks) - `__del__` deferral probe: dealloc during return serialization on the fiber **deferred → immediate** - live-malloc probe (`mallinfo2`, 300 tasks/shape): **~518 KiB/task → ~3 KiB/task** across async call → dict/bytes, async generator, sync generator on async actor - reporter-shaped streaming workload (400 tasks, 10 concurrent sessions): live-malloc delta **0.2 MB total**, fiber-sized mapped regions 0 → 0 - async-actor smoke: correctness (echo, state, async generators, recursion), concurrency (20 overlapping 0.5 s sleeps in 0.51 s) - throughput A/B (500 sequential echo tasks, 3 runs fixed / 2 runs baseline, same container image): fixed 5624–6076 tasks/s vs unpatched 4551–4825 tasks/s meaning no regression (the unpatched build is slower while leaking) - baseline (unpatched) wheel from the same tree reproduces the bug: +4.00 refs/task, fiber dealloc deferred=True note: fable did a majority of the heavy lifting in this investigation with prompting on what to check next and validate the solution --------- Signed-off-by: Mark Towers <mark@anyscale.com> Signed-off-by: myan <myan@anyscale.com> Co-authored-by: Mark Towers <mark@anyscale.com> Co-authored-by: myan <myan@anyscale.com> Co-authored-by: Mengjin Yan <mengjinyan3@gmail.com>
…es (#65137) ## Description The `PrometheusTimeseries` test util only merges scrapes and never drops series that disappear from `/metrics`. A mid-transition pending sample can outlive the agent-side series after gauge TTL and keep the summed count too high and cause flaky. This PR uses `flush=True` on each poll (same idea as `test_stale_view_cleanup_when_job_exits`) so we only assert on the current scrape. windows://python/ray/tests:test_task_metrics passes <img width="1662" height="678" alt="image" src="https://github.com/user-attachments/assets/e0e9a9a1-b712-4fbd-9ad6-b06018368039" /> --------- Signed-off-by: Rueian Huang <rueiancsie@gmail.com>
…DataContext.max_consecutive_actor_init_deaths (#64846) ## Why are these changes needed? A single actor whose `__init__` fails (e.g. an LLM engine failing to boot) kills the whole Dataset job: the readiness ref resolves with `ActorDiedError`, `_ActorPool.pending_to_running` re-raises it, and the metadata-task branch of `process_completed_tasks` has no guard (`max_errored_blocks` only protects the `DataOpTask` branch), so the exception fails the execution. `max_restarts=-1` (Ray Data's own default) cannot help because Ray Core does not restart actors whose creation task failed — the death is a `USER_ERROR` and `gcs_actor_manager` zeroes `remaining_restarts`. At an N-actor pool, any per-boot flake probability p becomes job-death probability 1-(1-p)^N: at 1,600 vLLM engines, three separate boot-flake classes (port races, VRAM residue, ramp-wave timeouts) each repeatedly killed full production runs. This PR catches the failure at the pool boundary (`_task_done_callback`) and charges it against a new per-operator budget, `DataContext.max_consecutive_actor_init_deaths` (env `RAY_DATA_MAX_CONSECUTIVE_ACTOR_INIT_DEATHS`; default `0` preserves today's fail-fast behavior; `-1` = unlimited). The counter is **consecutive**: it resets whenever any actor of the operator initializes successfully — the same semantics as the existing `actor_init_max_retries`. Sporadic boot flakes in a progressing pipeline are tolerated indefinitely within the budget, while a systemically broken UDF (bad model path) never resets the counter and exhausts the budget, so no separate systemic-failure heuristic or hardcoded threshold is needed. Tolerated failures are logged with a running count and the actor autoscaler starts a replacement (the pool is below target size, and the below-min branch runs before the no-input guard, so no explicit re-provision is needed). Counting is per actor incarnation, after any in-actor `actor_init_retry_on_errors` retries. ## Production-scale evidence Two byte-identical 1,600-engine / 70M-caption video-captioning runs on Anyscale (CoreWeave, RTX PRO 6000 fleet), same input corpus and knobs; the only variable is boot-failure handling (run against the earlier cumulative-budget revision of this patch; the consecutive-counting revision is strictly more permissive for the interleaved-flake pattern observed): | | Baseline (app-level workaround that swallows pending_to_running errors) | This patch (workaround disabled, budget=200) | |---|---|---| | Outcome | SUCCEEDED | SUCCEEDED | | Captions written | 69,982,028 | 69,982,028 (identical) | | Wall time | 5,525 s | 5,778 s | | Real engine-boot failures absorbed | 2 | 8 (single ramp-wave burst; every one logged and the pool recovered to its full 1,500 floor) | A separate fault-injection job validated both semantics deterministically on the same stack: 3 injected creation-task failures with budget 5 → dataset completes with correct output; budget 1 → tolerates one, then re-raises the original `ActorDiedError` after exactly 2 incarnations. Companion PR for the GCS-side half of the same incident: #64845. ## Related issue number None found — searched open PRs/issues for actor init/boot failure tolerance before opening; no in-flight work. Adjacent context: the vLLM batch stage's `os._exit(1)` workaround handles post-boot engine death via `max_restarts` (see #59522), but nothing covers creation-task failures. ## Checks - [x] I've signed off every commit (DCO). - [x] I've run pre-commit on the changed files. - [x] Testing strategy: - New tests in `python/ray/data/tests/test_actor_pool_map_operator.py`: `test_actor_init_death_budget` (parametrized: budget covers / exceeded / unlimited), `test_actor_init_death_budget_with_in_actor_retries` (proves incarnation-level counting), `test_on_actor_init_death_unit` (raise timing against the consecutive budget, reset-on-success, re-raise identity). - `python -m pytest python/ray/data/tests/test_actor_pool_map_operator.py -k "init_death or init_failure"` → 8 passed (pytest 7.4.4). Existing `test_actor_init_failure_retry` and `test_actor_pool_map_operator_init` pass unchanged. Per the repo's AI contribution policy: AI assistance was used to develop this change; every changed line was reviewed, the tests above were run locally, and the change was exercised end-to-end at production scale as described. --------- Signed-off-by: xyuzh <xinyzng@gmail.com>
## Description `ResourceRequest::operator==` / `operator!=` compared only `resources_`, ignoring the other two fields of the class, `requires_object_store_memory_` and `label_selector_`. Two requests with identical quantities but a different object-store-memory flag or label selector compared equal, so any dedup, change-detection, or map-key use of `ResourceRequest` silently conflated requests that are not interchangeable. `operator==` now compares all three fields (`LabelSelector` already provides `operator==`), and `operator!=` is defined as its negation. `operator<=` / `operator>=` are unchanged — comparing quantities only is their intended resource-containment semantics. Added a test constructing requests with equal quantities that differ only by the label selector or the object-store-memory flag and asserting inequality; it fails against the previous `operator==`. ## Related issues Fixes #64837 ## Additional information `bazel test //src/ray/common/scheduling/tests:resource_request_test` passes, and the adjacent scheduling suites (`resource_set_test`, `resource_instance_set_test`, `cluster_resource_scheduler_test`, `cluster_resource_manager_test`) still pass. Verified the new test fails when the fix is reverted. Signed-off-by: yangjie01 <yangjie01@baidu.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
11 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
See Commits and Changes for more details.
Created by
pull[bot] (v2.0.0-alpha.4)
Can you help keep this open source service alive? 💖 Please sponsor : )