Skip to content

Commit 082fbea

Browse files
jaychiaJay Chia
andauthored
[FEAT] Use detached named actor for RayRunner (#2296)
Uses a detached named actor for the RayRunner. This means that if multiple Daft executions are run from different clients on a Ray cluster, they will all run on the same Ray actor. Co-authored-by: Jay Chia <jaychia94@gmail.com@users.noreply.github.com>
1 parent 17a25e2 commit 082fbea

1 file changed

Lines changed: 9 additions & 1 deletion

File tree

daft/runners/ray_runner.py

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -672,6 +672,10 @@ def place_in_queue(item):
672672
pbar.close()
673673

674674

675+
SCHEDULER_ACTOR_NAME = "scheduler"
676+
SCHEDULER_ACTOR_NAMESPACE = "daft"
677+
678+
675679
@ray.remote(num_cpus=1)
676680
class SchedulerActor(Scheduler):
677681
def __init__(self, *n, **kw) -> None:
@@ -740,7 +744,11 @@ def __init__(
740744

741745
if isinstance(self.ray_context, ray.client_builder.ClientContext):
742746
# Run scheduler remotely if the cluster is connected remotely.
743-
self.scheduler_actor = SchedulerActor.remote( # type: ignore
747+
self.scheduler_actor = SchedulerActor.options( # type: ignore
748+
name=SCHEDULER_ACTOR_NAME,
749+
namespace=SCHEDULER_ACTOR_NAMESPACE,
750+
get_if_exists=True,
751+
).remote( # type: ignore
744752
max_task_backlog=max_task_backlog,
745753
use_ray_tqdm=True,
746754
)

0 commit comments

Comments
 (0)