feat(ray): support ReusePolicy — run Ray tasks on shared, reusable RayClusters - #1308
Merged
Conversation
A ray task with reusable=ReusePolicy(...) is retyped to fastray so the backend runs it on a shared, reusable RayCluster instead of an ephemeral one. replicas maps to the number of shared clusters; only replicas=1 is supported for now and anything else raises BadConfiguration. Other plugin task types still reject ReusePolicy. Signed-off-by: Kevin Su <pingsutw@apache.org>
Ray is the only plugin with a reusable backend (a shared RayCluster); other plugin tasks still reject ReusePolicy. Signed-off-by: Kevin Su <pingsutw@apache.org>
pingsutw
marked this pull request as draft
July 16, 2026 00:51
… into the ray plugin Core no longer special-cases Ray. Plugins opt in to reuse with a supports_reuse_policy class attribute (checked by the TaskEnvironment guard) and an apply_reuse_policy(task_template) hook that task serialization dispatches to instead of add_reusable. The ray plugin implements both: it validates that the policy requests exactly one replica (one shared RayCluster) and retypes the task to fastray for the shared-cluster backend plugin, leaving the Ray custom spec untouched. Signed-off-by: Kevin Su <pingsutw@apache.org>
…e change Drop the fastray retype: the task type stays "ray", and apply_reuse_policy records the validated policy under a reusePolicy key in the task's custom (RayJob) spec. The field is unknown to the RayJob proto and ignored by its unmarshalling; backends that support shared RayClusters route on its presence, and everything else runs the task as a plain ephemeral ray job. Signed-off-by: Kevin Su <pingsutw@apache.org>
pingsutw
changed the title
Drop the apply_reuse_policy hook; the ray plugin now records the reuse policy directly in the RayJob custom spec, using the same field names as the actor (fast-task) reuse spec so the backend can share its parsing. task_serde skips the default fast-task retyping when the plugin class sets supports_reuse_policy. Claude-Session: https://claude.ai/code/session_01XuokhQkvv6UzxfSYrLpMTr Signed-off-by: Kevin Su <pingsutw@apache.org>
…worker scaling min_replica_count/replica_count scale the shared cluster's worker groups and scaledown_ttl becomes the Ray autoscaler idle timeout, so multiple replicas are meaningful and allowed. Signed-off-by: Kevin Su <pingsutw@apache.org>
replicas means the number of shared RayClusters (only 1 supported) and idle_ttl means shutting down the entire idle cluster; worker scaling stays under the Ray plugin config, not the reuse policy. Signed-off-by: Kevin Su <pingsutw@apache.org>
concurrency and scaledown_ttl have no meaning for a shared RayCluster — Ray schedules work on resource availability and worker scaling belongs to RayJobConfig. Raise BadConfiguration for non-default values instead of silently ignoring them. (The dataclass defaults are indistinguishable from explicitly passed defaults, so those pass.) Signed-off-by: Kevin Su <pingsutw@apache.org>
Signed-off-by: Kevin Su <pingsutw@apache.org>
Claude-Session: https://claude.ai/code/session_01XuokhQkvv6UzxfSYrLpMTr Signed-off-by: Kevin Su <pingsutw@apache.org>
pingsutw
marked this pull request as ready for review
July 20, 2026 17:06
Signed-off-by: Kevin Su <pingsutw@apache.org>
cosmicBboy
approved these changes
Jul 24, 2026
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 join this conversation on GitHub.
Already have an account?
Sign in to comment
2 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.
Why
ReusePolicytoday only supports container/pod tasks (reusable actors). Ray tasks always pay full RayCluster cold-start cost on every run, even when many runs share an identical cluster spec.This PR lets users opt a Ray task environment into a shared, reusable RayCluster:
What changed
Core stays plugin-agnostic, and the task type is unchanged — reuse is signaled through the task's custom spec:
supports_reuse_policyclass attribute (checked by theTaskEnvironmentreusable+plugin_config guard) and anapply_reuse_policy(task_template)hook that task serialization dispatches to instead ofadd_reusable. No Ray knowledge inreuse.py.replicasmaps to the number of shared clusters; only 1 supported for now, otherwiseBadConfiguration) and records it under areusePolicykey in the task's custom (RayJob) spec. The field is unknown to the RayJob proto and ignored by its unmarshalling — the task type staysray, so backends without shared-cluster support simply run the task as a normal ephemeral Ray job.reusePolicy: shared RayCluster (keyed by the Ray spec hash, so identical specs reuse one cluster) when present, ephemeral otherwise. The generic backend plugin interface lives in feat: add ClusterPlugin interface for shared-cluster plugins flyte#7460.examples/plugins/ray_reusable_cluster.py.Tests
tests/flyte/internal/runtime/test_reuse.py— core behavior unchanged (9 passing).plugins/ray/tests/test_task.py— policy recorded in custom with type unchanged; multi-replica rejection (9 passing).