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.
There is a race condition where the failure message is sent but resubscription hasn't been established. In this case, we'll lose this message.
This PR fixes the test case only by making sure resubscription is done before killing the worker.
We are not going to fix this issue for the first version of GCS HA because:
1. Right now, only failed worker will be reported and it'll be used to eagerly kill the unnecessary workers. From correctness, it's ok.
2. worker/task support is not P0 for GCS HA (Ray Serve). Actor doesn't have this issue since the subscription for the actor is by actor id and raylet will fetch the actor status when resubscribing.
Things are not working:
- we won't kill workers which are not useful anymore. So for a very extreme case (the worker hangs), it'll have a resource leak.
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.
Related issue number
Fixes#24267.
Previously we pinned the controller on the head node because it was the SPOF, now with GCS HA work that will no longer be the case so it should be able to be run on any node. We still prefer the head node for non-HA cases.
It seems `_InactiveRpcError` is not saved as the last exception. Raise an explicit error when publishing fails after retries.
For log and error publishing, dropping messages should be tolerable so log the exception instead.
The previous implementation of TorchPredictor failed when the model outputted an array for each sample (for example outputting logits). This is because Pandas Dataframes cannot hold numpy arrays of more than 1 element. We have to convert the outermost numpy array returned by the model to a list so that it be can stored in a Pandas Dataframe.
The newly added test fails under the old implementation.
For mac and windows, we default to start the head node with 127.0.0.1 to avoid security pop-ups and assume people won't use cluster mode a lot. However, when people do want to create a mac/windows cluster, our connection message is confusing since we cannot connect to the head node that's listening on 127.0.0.1. This PR tells people what to do in this case (e.g. use node-ip-address).
This PR adds a fast path for `sync_dir_between_nodes` that gets triggered if both source IP and target IP are the same. It uses simple `shutil` operations instead of packing and unpacking to improve performance.
The documentation says
> Consider using .map_batches() for better performance (you can implement filter by dropping records).
but there aren't any examples of how to do so.
I have seen errors where the prometheus view data dictionary is changed when iterating over it. So make a copy (which is atomic) before iterating.
Also, use a separate thread for GCS client heartbeat RPC. This avoids the issue of missing heartbeat when raylet is too busy.
This PR supports various output formatting. By default, we support yaml format. But this can be changed depending on the UX research we will conduct in the future.
1cb0f4b51a5799e0360a66db01000000:
actor_id: 1cb0f4b51a5799e0360a66db01000000
class_name: A
state: ALIVE
f90ba34fa27f79a808b4b5aa01000000:
actor_id: f90ba34fa27f79a808b4b5aa01000000
class_name: A
state: ALIVE
Table format is not supported yet. We will support this once we enhance the API output (which I will create an initial API review soon).
This PR adds a flag to enable creating nodes in the main thread in the autoscaler. The flag is turned on for the KubeRay node provider. KubeRay already uses a flag to disable node updater threads -- with the changes from this PR, the autoscaler becomes single-threaded when launching on KubeRay.
When an object is under reconstruction, pull manager keeps the bundle request active with no timeout, which may block the next bundle request that's needed for the object reconstruction. As a result, we have deadlock.
For example, task 1 takes object A as argument and returns object B, task 2 takes object B as argument. When we run task 2, pull manager will add B to the queue and then B is lost. In this case, task 1 is re-submitted and A is added the the pull manager queue after B (assuming both tasks are scheduled to the same node). Due to limited available object store memory, A cannot be activated until B is pulled but B cannot be pulled until A is pulled and B is reconstructed.
The solution is that if an active pull request has pending-creation objects, pull manager will deactivates it until creation is done. This way, we will free object store memory occupied by the current active pull request so that next requests can proceed and potentially unblock the object creation.
This is the 1st PR to remove the code path of multiple core workers in one process. This PR is aiming to remove the flags and APIs related to `num_workers`.
After this PR checking in, we needn't to consider the multiple core workers any longer.
The further following PRs are related to the deeper logic refactor, like eliminating the gap between core worker and core worker process, removing the logic related to multiple workers from workerpool, gcs and etc.
**BREAK CHANGE**
This PR removes these APIs:
- Ray.wrapRunnable();
- Ray.wrapCallable();
- Ray.setAsyncContext();
- Ray.getAsyncContext();
And the following APIs are not allowed to invoke in a user-created thread in local mode:
- Ray.getRuntimeContext().getCurrentActorId();
- Ray.getRuntimeContext().getCurrentTaskId()
Note that this PR shouldn't be merged to 1.x.
Instead of letting Datasets implicitly use cluster resources in the margins of explicit allocations of other libraries, such as Tune, Datasets should provide an option for explicitly allocating resources for a Datasets workload for users that want to box Datasets in. This PR adds such an explicit resource allocation option, via exposing a top-level scheduling strategy on the DatasetContext with which a placement group can be given.
Tune resource bookkeeping was broken. Specifically, this is what happened in the repro provided in #24259:
- Only one PG can be scheduled per time
- We staged resources for trial 1
- We run trial 1
- We stage resources for trial 2
- We pause trial 1, caching the placement group
- This removes the staged PG for trial 2 (as we return the PG for trial 1)
- However, now we `reconcile_placement_groups`, which re-stages a PG
- both trial 1 and trial 2 are now not in `RayTrialExecutor._staged_trials`
- A staging future is still there because of the reconciliation
This PR fixes this problem in two ways. `has_resources_per_trial` will check a) also for staging futures for the specific trial, and b) will also consider cached placement groups.
Generally, resource management in Tune is convoluted and hard to debug, and several classes share bookkeeping responsibilities (runner, executor, pg manager). We should refactor this.
`df.transform` has undefined behavior when the passed in function mutates the dataframe, as mentioned in the pandas docs. This is because I believe the implementation iterates through slices of the dataframe and passes these slices to the provided function. This "gotcha" exposes an implementation to users who are using `BatchMapper`.
It's pretty common to have preprocessors that mutates the dataframe, for example our own test does the following
```
def add_and_modify_udf(df: "pd.DataFrame"):
df["new_col"] = df["old_column"] + 1
df["to_be_modified"] *= 2
return df
```
so instead of using `df.transform`, we instead do `self.fn(df)`. As the `df` is the output of Ray Datasets `iter_batches`, the provided function can safely mutate the dataset.
A newly planned version of the Serve schema (used in the REST API and CLI) requires the user to pass in their deployment graph's`import_path` and optionally a runtime_env containing that graph. This new schema can then pick up any `init_args` and `init_kwargs` values directly from the graph, instead of requiring them to be serialized and passed explicitly into the REST request.
This change:
* Adds the `import_path` and `runtime_env` fields to the `ServeApplicationSchema`.
* Updates or disables outdated unit tests.
Follow-up changes should:
* Update the status schemas (i.e. `DeploymentStatusSchema` and `ServeApplicationStatusSchema`).
* Remove deployment-level `import_path`s.
* Process the new `import_path` and `runtime_env` fields instead of silently ignoring them.
* Remove `init_args` and `init_kwargs` from `DeploymentSchema` afterwards.
Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>
This PR adds support for tensor columns in the to_tf() and to_torch() APIs.
For Torch, this involves an explicit extension array check and (zero-copy) conversion of the tensor column to a NumPy array before converting the column to a Torch tensor.
For TensorFlow, this involves bypassing df.values when converting tensor feature columns to NumPy arrays, instead manually creating a single NumPy array from the column Series.
In both cases, I think that the UX around heterogeneous feature columns and squeezing the column dimension could be improved, but I'm saving that for a future PR.
A follow-up PR from this one: https://github.com/ray-project/ray/pull/24628
In the previous PR, it fixed the resubscribing issue for raylet. But there is also core worker which needs to do resubscribing.
There are two ways of doing resubscribe:
1. When the client-side detects any failure, it'll do resubscribing.
2. Server side will ask the client to do resubscribing.
1) is a cleaner and better solution. However, it's a little bit hard due to the following reasons:
- We are using long-polling, so for some extreme cases, we won't be able to detect the failure. For example, the client-side received the message, but before it sends another request, the server-side restarts, and the client will miss the opportunity of detecting the failure. This could happen if we have a standby GCS that starts very fast and somehow the client-side has a lot of traffic and runs very slow.
- The current gRPC framework doesn't give the user a way to handle failure which might need some refactoring on this one.
We can go with this way once we have gRPC streaming.
This PR is implementing 2) which includes three parts:
- raylet: (https://github.com/ray-project/ray/pull/24628)
- core worker: (this pr)
- python
Correctness: whenever when a worker started, it'll register to raylet immediately (sync call) before connecting to GCS. So, we just need to send all restart rpcs to registered workers and it should work because:
- if the worker just started and hasn't registered with the raylet: it's ok, because the worker hasn't connected with GCS yet, so no need to do resubscribing.
- if the worker has registered with the rayelt: it's covered by the code path here.
Adds a Dataset.split_proportionately method that allows the user to split a dataset using proportions. This is a very common use-case for eg. train-test splitting. The implementation is a thin wrapper over Dataset.split_at_indices.
Additionally, this PR adds a ray.ml.train_test_split function intended to provide a familiar API to ML practitioners.
