[core] Support detached/GCS owned objects

[ericl]

Overview

In the current ownership model, all created objects are owned by some worker executing the job. This means however that when a job exits, all created objects become unavailable.

In certain cases it is desirable to share objects between jobs (e.g., shared cache), without creating objects explicitly from a detached actor.

We have a current internal API to allow the owner of an object created with ray.put() to be assigned to a specific actor, e.g.:

ray.put(data, _owner=actor_handle)

This means the object created will fate-share with that actor instead of the current worker process. This means that it's currently possible to create global objects by creating a named detached actor, and setting the owner to that actor.

However, it would be nice to support ray.put(data, _owner="global") to avoid the need for that hack, and allow the object to be truly HA (e.g., tracked durably in HA GCS storage).

[fishbone]

I have a concern about this method.
In this way, it looks like we treat the global storage as a key-value (with schema) storage and after running ray.util.put_global_ref(obj_ref, lookup_key="objfoo1") we store objfoo1 -> obj_ref in the storage which might leads to memory leak. It looks like makes memory management complicated. It's different from the actor which when it terminates it'll release all the resources. Besides if we do it in this way, we probably want to introduce namespace to avoid conflict and to prevent some object from being accidentally being updated by other apps.

If we have a global actor per group of task, and it's served for this purpose, will it just work? Will it be simpler? Do we need to support move ownership for this?

[ericl]

I agree binding the lifetime of an object to an actor might be the way to go. The way I see it there are probably a few use cases:
(1) pin object permanently, manual deletion
(2) allow object to be evicted if under memory pressure (caching case)
(3) bind lifetime of object to another actor/task (similar to ownership transfer)

Perhaps we could create the following API? Perhaps we don't need to give objects an explicit name, just allow their lifetime to be changed:

ray.util.set_object_lifetime(obj_ref, lifetime="detached"|"lru"|<actorRef>)

# can only implement <actorRef> case initially

Btw, actually doing ownership transfer is pretty hard, since the owner metadata is propagated inline with the object reference. You'd need to add some kind of transfer table to handle that. Assuming the number of objects that require transfer is quite small, it would be easier to instead only allow ownership to be transferred to the GCS. This is simpler since it means owner lookups can always fall back to GCS on owner failure.

[fishbone]

I think my concern here is two things: 1) how to manage all the detached variables; 2) how to do better isolation between groups of jobs.

how to manage all the detached variables
We plan to make ray run in a cluster, and developers can submit their job to the cluster. If someone created many detached variables by using (1), it'd prevent other jobs from running due to OOM. And it's not easy to recover from this since we don't have extra info about all these detached variables.
Besides, if we do ray.util.set_object_lifetime(obj_ref, lifetime="detached"|"lru"|<actorRef>) how can another job get the obj_ref? It should be a variable shared between jobs, but if one job exit, how can it be transferred from one to another?

Here I assume one job to be a .py file and will be executed by ray submit job.py or python job.py. Please let me know if I'm wrong.

how to do better isolation between groups of jobs
If we use the variable name to do communication between jobs (if I understand the concept of the job correctly), the variable name will be a global variable visible to all jobs in the cluster. If we have a series of jobs and run them with different inputs, there will be conflicts.

I thought of introducing job groups of something else, but then it looks like one driver job with several sub-jobs. So sub-jobs(separate py file) can run in the driver job (main py file), and the driver job will have a reference of the shared data, which will work.

I think something is missing here in my brain. Please share more details about this.

[ericl]

Another user ask in #12748

[ericl]

If someone created many detached variables by using (1), it'd prevent other jobs from running due to OOM. And it's not easy to recover from this since we don't have extra info about all these detached variables.

This isn't something we need to worry about though--- if the user pins too much, it's their problem. With an autoscaling cluster, we can also spill / add more nodes to handle this case.

Besides, if we do ray.util.set_object_lifetime(obj_ref, lifetime="detached"|"lru"|) how can another job get the obj_ref? It should be a variable shared between jobs, but if one job exit, how can it be transferred from one to another?

One way is that the object refs can be transferred through a named actor. Perhaps we can try this to see if it satisfies use cases before adding another API.

[ericl]

I thought of introducing job groups of something else, but then it looks like one driver job with several sub-jobs. So sub-jobs(separate py file) can run in the driver job (main py file), and the driver job will have a reference of the shared data, which will work.

Btw, in the RayDP/Spark on Ray use case, the variable just needs to be transferred within the job, so there isn't the use case of multiple jobs yet. For caching use cases, the variables can be transferred to named actors global across many jobs. So I think changing lifetime is sufficient for both these cases, but perhaps there are others that require more functionality.

[fishbone]

I think I misunderstood some parts. So let me just summarize it here and please let me know if I was wrong:

• Having a named actor shared across jobs to exchange data between them
• If the user wants to do the cross job communication, pass the ownership to the actor
• Multiple eviction policy can be supported here (detached, lru, binding it to anther job/actor)

Another thing I might want to be clarified here is what's a job? Is it a python script? Some demo code about what needs to be achieved should be helpful for me to understand it.

This isn't something we need to worry about though--- if the user pins too much, it's their problem. With an autoscaling cluster, we can also spill / add more nodes to handle this case.

I'm thinking that it's a cluster and multiple users will want to submit the job in the same cluster. If someone uses too much and we can't have a way to group the related variables, it's going to be a mess. But if it's through a named actor and jobs users have different named actors, then it's OK. We can limit memory usage through these actors to protect the cluster.

[ericl]

That sounds right to me. A couple clarifications: a Python script/driver defines a job, since we generate a unique job on connecting to a cluster. So in the Spark on Ray case, all actors run would be all within a single job. Re: memory usage, the isolation here is not very well developed, but note this is a general problem in Ray and not something specific to detached objects. Hence, it is out of scope of this feature itself.

…

On Thu, Dec 10, 2020, 5:26 PM Yi Cheng ***@***.***> wrote: I think I misunderstood some parts. So let me just summarize it here and please let me know if I was wrong: - Having a named actor shared across jobs to exchange data between them - If the user wants to do the cross job communication, pass the ownership to the actor - Multiple eviction policy can be supported here (detached, lru, binding it to anther job/actor) Another thing I might want to be clarified here is what's a job? Is it a python script? Some demo code about what needs to be achieved should be helpful for me to understand it. This isn't something we need to worry about though--- if the user pins too much, it's their problem. With an autoscaling cluster, we can also spill / add more nodes to handle this case. I'm thinking that it's a cluster and multiple users will want to submit the job in the same cluster. If someone uses too much and we can't have a way to group the related variables, it's going to be a mess. But if it's through a named actor and jobs users have different named actors, then it's OK. We can limit memory usage through these actors to protect the cluster. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#12635 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAADUSXOM64NL5Q6GBZZ3FDSUFYMRANCNFSM4UOG4ZQQ> .

[DonYum]

Both @ericl and @ahbone 's opinions are impressive.

As @ahbone say, the management of the detached objs looks like shouldn't be given to the user. This will mess the code of user.
However, In practice, we do need a 'method' to make some long-life objs.

Yes, using detached named actor to customize a 'detached objs' is a method. But it's to complete, and is implicit.

As a novice of ray, I spent hours to find some method to 'detach' a plasma obj In docs/issues and discussions, but I failed.
I have to submit a new issue #12748 which lead me to here.
The process was too painful. Along the way, I always thinking, why not provide such an API. Is this a technical problem?
...

In my opinion, ray core is like a toolbox, providing enough APIs is our responsibility, and how customers use it is another thing that we may not consider too much.

Hope my ideas will help.
Thanks a lot!

[fishbone]

@DonYum Thanks for your ideas here, and I agree this is a useful feature to support.
My concern here is not about providing such an API/feature but how we we should support this feature better. Early discussion of it will save us effort in the future.
What worries me here is actually how to manage all these long-life objects. Suppose we have a lot of long-life objects created, now the user wants to free some objects through ray client, it's hard to know which one to free since they are all ids (maybe I was wrong, not an experienced ray user). Usually it might be useful for us to free all the global variables created by some job when they are not needed anymore. Binding them to a named actor sounds to me like a potential way to go.

@ericl Here for RayDP, as you mentioned that:

So in the Spark on Ray case, all actors run would be all within a single job.

In this way, it looks like all objects can be passed back to the driver. Why detached objects will help here?
If you have some sample code about the usecase we'd like to support, it'll be helpful.