Redo for PR #24698:
This fixes two bugs in data locality:
When a dependent task is already in the CoreWorker's queue, we ran the data locality policy to choose a raylet before we added the first location for the dependency, so it would appear as if the dependency was not available anywhere.
The locality policy did not take into account spilled locations.
Added C++ unit tests and Python tests for the above.
Split test_reconstruction to avoid test timeout. I believe this was happening because the data locality fix was causing extra scheduler load in a couple of the reconstruction stress tests.
When loading the data from GCS, for detached actors, we treat it the same as normal actors.
But the detached actor lives beyond the job's scope and should be loaded even when the job is finished.
This PR fixed it.
This fixes two bugs in Datasets push-based shuffle:
Scheduling strategy specified by the caller was not getting propagated correctly to the map stage in push-based shuffle. This is because the map and reduce stages shared the same ray.remote options dict, and we deleted the caller-specified scheduling strategy from the reduce stage so that we could specify a NodeAffinitySchedulingStrategy instead.
We were only reporting partial stats for the merge stage.
Related issue number
Issue 1 is necessary for performance at large-scale (#24480).
This makes it possible to use an NFS file system that is shared on a cluster for runtime_env working directories.
Co-authored-by: shrekris-anyscale <92341594+shrekris-anyscale@users.noreply.github.com>
Co-authored-by: Eric Liang <ekhliang@gmail.com>
closes#24475
Current deployment graph has big perf issues compare with using plain deployment handle, mostly because overhead of DAGNode traversal mechanism. We need this mechanism to empower DAG API, specially deeply nested objects in args where we rely on pickling; But meanwhile the nature of each execution becomes re-creating and replacing every `DAGNode` instances involved upon each execution, that incurs overhead.
Some overhead is inevitable due to pickling and executing DAGNode python code, but they could be quite minimal. As I profiled earlier, pickling itself is quite fast for our benchmarks at magnitude of microseconds.
Meanwhile the elephant in the room is DeploymentNode and its relatives are doing too much work in constructor that's beyond necessary, thus slowing everything down. So the fix is as simple as
1) Introduce a new set of executor dag node types that contains absolute minimal information that only preserves the DAG structure with traversal mechanism, and ability to call relevant deployment handles.
2) Add a simple new pass in our build() that generates and replaces nodes with executor dag to produce a final executor dag to run the graph.
Current ray dag -> serve dag mixed a lot of stuff related to deployment generation and init args, in longer term we should remove them but our correctness depends on it so i rather leave it as separate PR.
### Current 10 node chain with deployment graph `.bind()`
```
chain_length: 10, num_clients: 1
latency_mean_ms: 41.05, latency_std_ms: 15.18
throughput_mean_tps: 27.5, throughput_std_tps: 3.2
```
### Using raw deployment handle without dag overhead
```
chain_length: 10, num_clients: 1
latency_mean_ms: 20.39, latency_std_ms: 4.57
throughput_mean_tps: 51.9, throughput_std_tps: 1.04
```
### After this PR:
```
chain_length: 10, num_clients: 1
latency_mean_ms: 20.35, latency_std_ms: 0.87
throughput_mean_tps: 48.4, throughput_std_tps: 1.43
```
This PR consolidates the Ray Train and Tune checkpoint managers. These concepts previously did something very similar but in different modules. To simplify maintenance in the future, we've consolidated the common core.
- This PR keeps full compatibility with the previous interfaces and implementations. This means that for now, Train and Tune will have separate CheckpointManagers that both extend the common core
- This PR prepares Tune to move to a CheckpointStrategy object
- In follow-up PRs, we can further unify interfacing with the common core, possibly removing any train- or tune-specific adjustments (e.g. moving to setup on init rather on runtime for Ray Train)
The consolidation is split into three PRs:
1. This PR - adds a common checkpoint manager class.
2. #24772 - based on this PR, adds the integration for Ray Train
3. #24430 - based on #24772, adds the integration for Ray Tune
Follow up: #24017
Briefly, wandb service is still in experimental stage, and is not ready to be released as an integration without extensive testing. Hence, we are interested in rolling back the update to the integration we made recently, until this feature is ready to be shipped.
This is a follow-up PRs of https://github.com/ray-project/ray/pull/24813 and https://github.com/ray-project/ray/pull/24628
Unlike the change in cpp layer, where the resubscription is done by GCS broadcast a request to raylet/core_worker and the client-side do the resubscription, in the python layer, we detect the failure in the client-side.
In case of a failure, the protocol is:
1. call subscribe
2. if timeout when doing resubscribe, throw an exception and this will crash the system. This is ok because when GCS has been down for a time longer than expected, we expect the ray cluster to be down.
3. continue to poll once subscribe ok.
However, there is an extreme case where things might be broken: the client might miss detecting a failure.
This could happen if the long-polling has been returned and the python layer is doing its own work. And before it sends another long-polling, GCS restarts and recovered.
Here we are not going to take care of this case because:
1. usually GCS is going to take several seconds to be up and the python layer's work is simply pushing data into a queue (sync version). For the async version, it's only used in Dashboard which is not a critical component.
2. pubsub in python layer is not doing critical work: it handles logs/errors for ray job;
3. for the dashboard, it can just restart to fix the issue.
A known issue here is that we might miss logs in case of GCS failure due to the following reasons:
- py's pubsub is only doing best effort publishing. If it failed too many times, it'll skip publishing the message (lose messages from producer side)
- if message is pushed to GCS, but the worker hasn't done resubscription yet, the pushed message will be lost (lose messages from consumer side)
We think it's reasonable and valid behavior given that the logs are not defined to be a critical component and we'd like to simplify the design of pubsub in GCS.
Another things is `run_functions_on_all_workers`. We'll plan to stop using it within ray core and deprecate it in the longer term. But it won't cause a problem for the current cases because:
1. It's only set in driver and we don't support creating a new driver when GCS is down.
2. When GCS is down, we don't support starting new ray workers.
And `run_functions_on_all_workers` is only used when we initialize driver/workers.
Packages are uploaded to the GCS for `runtime_env`. These packages are garbage collected when their refcount becomes zero.
The problem is the reference doesn't get incremented until the job starts, which happens after the package is uploaded. It's possible for the package's refcount to go to zero in between the upload and when the job starts, causing the package to be deleted before it's needed by the job. It's likely the cause of https://github.com/ray-project/ray/issues/23423.
We can't just increment the refcount at the time of upload, because if the script is killed before the job is started (e.g. via Ctrl-C) then the reference will never be decremented and the package will never be deleted.
The solution in this PR is to increment the refcount at the time of upload, but automatically decrement after a configurable timeout (default 30s). This should be enough time for the job to start. When the job starts, it increments the refcount as usual and decrements it when the job finishes or is killed.
Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>
Looking at past failures of dataset_shuffle_push_based_random_shuffle_1tb and when running it on my own, I noticed that raylets are killed because GCS was not able to respond to it in time. It seems at the beginning of the run, there is a huge CPU spike which starved GCS out of CPU. With the same spirit of adjusting workers to higher OOM scores, we can give workers higher niceness so they yield CPU to GCS, Raylet and other user processes.
I ran dataset_shuffle_push_based_random_shuffle_1tb a few time which no longer sees raylet death because of GCS CPU starvation. But there are other issues making the test fail which I will continue to investigate.
commit 40774ac219
Author: Qing Wang <kingchin1218@gmail.com>
Date: Tue May 17 11:33:59 2022 +0800
Minor changes for Java runtime env. (#24840)
Introduced an extra log message that spams stdout when running with Tune. Move this log line to debug and add an e2e test check.
When assigning an owner for an object (different from the current worker), such as:
```python
ray.put(vaule, _owner = ACTORHANDLE)
```
Object Manager holds the wrong owner's address and updates location info to the wrong worker, making `ray.get` slow. the current master will get Timeout in this new test case.
Why are these changes needed?
Add API stability annotations for datasource classes, and add a linter to check all data classes have appropriate annotations.
This test is not running well in Redis mode. Given that the other tests are ok, I'd like to only disable this one instead of revert the whole commit to making sure the other tests don't have regression.
`linux://python/ray/tests:test_runtime_env_plugin::test_plugin_timeout`
Now the status of subscribing to Actors in Actor Manager is eager mode, that is to say, when worker A passes List<ActorHandler> as an input parameter to another worker B, worker B will immediately subscribe to the status of all Actors in this list when constructing, even if worker B has not yet used these actors.
Assuming that a graph job has 1000 actors, and each actor has a List of the graph, then this job has nearly 100w subscription relationships. When the job goes offline, the 1000 actor processes will be killed, the redis-server will instantly receive the disconnect event from the 1000 actor processes, each event will trigger 1000 unsubscribexxx operations in the freeClient, causing the redis-server to get stuck.
We suggest to change this eager mode to lazy mode, and only initiate subscription when `SubmitActorTask`, which can reduce many unnecessary subscription relationships.
The microbenchmark (Left is this PR, Right is master branch)
in python3.10, it fixed a bug that a interactively defined class was tagged with a wrong type during inspection; which now throws OSError. detailed pr python/cpython#27171
we need to handle this case properly in otherwise ray actor definition will throw in interactive mode. please refer to #25026 for repo.
Adds file locking to prevent parallel file system operations to Tune/AIR syncing functions.
Co-authored-by: Kai Fricke <krfricke@users.noreply.github.com>
Weights and biases creates a wandb directory to collect intermediate logs and artifacts before uploading them. This directory should be in the respective trial directories. This also means we can re-enable auto resuming.
When `create_scheduler("pb2", ....)` is run a `TuneError` exception is raised. See referenced issue below for details.
In addition to the fix, introduced a new test (`ray/tune/tests/test_api.py::ShimCreationTest.testCreateAllSchedulers`) to confirm that `tune.create_scheduler()` will work with all documented schedulers.
Note: `tesCreateAllTestSchedulers` is a superset of what is covered in `testCreateScheduer`. It may be reasonable to retire the later test.
This PR makes several improvements to the Datasets' tensor story. See the issues for each item for more details.
- Automatically infer tensor blocks (single-column tables representing a single tensor) when returning NumPy ndarrays from map_batches(), map(), and flat_map().
- Automatically infer tensor columns when building tabular blocks in general.
- Fixes shuffling and sorting for tensor columns
This should improve the UX/efficiency of the following:
- Working with pure-tensor datasets in general.
- Mapping tensor UDFs over pure-tensor, a better foundation for tensor-native preprocessing for end-users and AIR.
Since ray supports Redis as a storage backend, we should ensure the code path with Redis as storage is still being covered e2e.
The tests don't run for a while after we switch to memory mode by default. This PR tries to fix this and make it run with every commit.
In the future, if we support more and more storage backends, this should be revised to be more efficient and selective. But now I think the cost should be ok.
This PR is part of GCS HA testing-related work.
* [runtime env] runtime env inheritance refactor (#22244)
Runtime Environments is already GA in Ray 1.6.0. The latest doc is [here](https://docs.ray.io/en/master/ray-core/handling-dependencies.html#runtime-environments). And now, we already supported a [inheritance](https://docs.ray.io/en/master/ray-core/handling-dependencies.html#inheritance) behavior as follows (copied from the doc):
- The runtime_env["env_vars"] field will be merged with the runtime_env["env_vars"] field of the parent. This allows for environment variables set in the parent’s runtime environment to be automatically propagated to the child, even if new environment variables are set in the child’s runtime environment.
- Every other field in the runtime_env will be overridden by the child, not merged. For example, if runtime_env["py_modules"] is specified, it will replace the runtime_env["py_modules"] field of the parent.
We think this runtime env merging logic is so complex and confusing to users because users can't know the final runtime env before the jobs are run.
Current PR tries to do a refactor and change the behavior of Runtime Environments inheritance. Here is the new behavior:
- **If there is no runtime env option when we create actor, inherit the parent runtime env.**
- **Otherwise, use the optional runtime env directly and don't do the merging.**
Add a new API named `ray.runtime_env.get_current_runtime_env()` to get the parent runtime env and modify this dict by yourself. Like:
```Actor.options(runtime_env=ray.runtime_env.get_current_runtime_env().update({"X": "Y"}))```
This new API also can be used in ray client.
This PR adds precise reason details regarding worker failures. All information is available either by
- ray list workers
- exceptions from actor failures.
Here's an example when the actor is killed by a SIGKILL (e.g., OOM killer)
```
RayActorError: The actor died unexpectedly before finishing this task.
class_name: G
actor_id: e818d2f0521a334daf03540701000000
pid: 61251
namespace: 674a49b2-5b9b-4fcc-b6e1-5a1d4b9400d2
ip: 127.0.0.1
The actor is dead because its worker process has died. Worker exit type: UNEXPECTED_SYSTEM_EXIT Worker exit detail: Worker unexpectedly exits with a connection error code 2. End of file. There are some potential root causes. (1) The process is killed by SIGKILL by OOM killer due to high memory usage. (2) ray stop --force is called. (3) The worker is crashed unexpectedly due to SIGSEGV or other unexpected errors.
```
## Design
Worker failures are reported by Raylet from 2 paths.
(1) When the core worker calls `Disconnect`.
(2) When the worker is unexpectedly killed, the socket is closed, raylet reports the worker failures.
The PR ensures all worker failures are reported through Disconnect while it includes more detailed information to its metadata.
## Exit types
Previously, the worker exit types are arbitrary and not correctly categorized. This PR reduces the number of worker exit types while it includes details of each exit type so that users can easily figure out the root cause of worker crashes.
### Status quo
- SYSTEM ERROR EXIT
- Failure from the connection (core worker dead)
- Unexpected exception or exit with exit_code !=0 on core worker
- Direct call failure
- INTENDED EXIT
- Shutdown driver
- Exit_actor
- exit(0)
- Actor kill request
- Task cancel request
- UNUSED_RESOURCE_REMOVED
- Upon GCS restart, it kills bundles that are not registered to GCS to synchronize the state
- PG_REMOVED
- When pg is removed, all workers fate share
- CREATION_TASK (INIT ERROR)
- When actor init has an error
- IDLE
- When worker is idle and num workers > soft limit (by default num cpus)
- NODE DIED
- Only can detect when the node of the owner is dead (need improvement)
### New proposal
Remove unnecessary states and add “details” field. We can categorize failures by 4 types
- UNEXPECTED_SYSTEM_ERROR_EXIT
- When the worker is crashed unexpectedly
- Failure from the connection (core worker dead)
- Unexpected exception or exit with exit_code !=0 on core worker
- Node died
- Direct call failure
- INTENDED_USER_EXIT.
- When the worker is requested to be killed by users. No workflow required. Just correctly store the state.
- Shutdown driver
- Exit_actor
- exit(0)
- Actor kill request
- Task cancel request
- INTENDED_SYSTEM_EXIT
- When the worker is requested to be killed by system (without explicit user request)
- Unused resource removed
- Pg removed
- Idle
- ACTOR_INIT_FAILURE (CREATION_TASK_FAILED)
- When the actor init is failed, we fate share the process with the actor.
- Actor init failed
## Limitation (Follow up)
Worker failures are not reported under following circumstances
- Worker is failed before it registers its information to GCS (it is usually from critical system bug, and extremely uncommon).
- Node is failed. In this case, we should track Node ID -> Worker ID mapping at GCS and when the node is failed, we should record worker metadata.
I will create issues to track these problems.
