[RFC] Add release_resources flag to ray.get/wait

[ericl]

In IO-intensive use cases, the application may not want ray.get() to release resources. For example, if object spilling is on, ray.get() may be requiring remote object reads from external storage. In this case, the worker should be still considered "busy", and releasing resources just results in the needless creation of more workers that can cause thrashing.

Example:

@ray.remote
def read(refs):
   for r in refs:  # assume refs are pointing to objects spilled to external storage
      ray.get(r, release_resources=False)

ray.get([read.remote(refs) for refs in refs_list])

Without the release_resources=False flag, the above program will spawn too many workers and cause Ray to crash.

Alternatives:

• A possible alternative is to optimize ray.wait/get to only release resources when waiting on task completion, instead of IO transfers. However, this mechanism would be more complex to implement.
• Implement as an experimental flag for now.

[robertnishihara]

A possible alternative is to optimize ray.wait/get to only release resources when waiting on task completion, instead of IO transfers. However, this mechanism would be more complex to implement.

Would this alternative just always do the right thing? That sounds very preferable.

As much as possible, I'd prefer for things to just work out of the box. My concern with this flag is that the code users write won't work the first time. It will fail because too many workers are started. Then they'll have to debug and figure out what went wrong. It could be hard to make the connection between too many workers and the ray.get behavior, but say they make that connection because we have a good error message, then they'll have to figure out the right place to add the flag, and then it will work.

[ericl]

Yeah, the concern is that it could be tricky or even impossible to implement in a performant way. For example, suppose we had to check object metadata prior to releasing resources, that could add a lot of overhead to every get call. There might be some nice solution though.

[stephanie-wang]

I think we could do the optimization just based on whether the value is ready in the in-process store. So on ray.get(o):

1. Wait for o to appear in in-process store. # Release resources.
2. Wait for o to appear in local object store. # Hold resources.

Also makes sense to do that in general, I think, since we do use actually use resources in step 2.

The only thing is that this can still behave poorly for borrowed object refs since it relies on an RPC to the owner during step 1. But it should be okay as long as the RPC is not too delayed.

This seems simple enough that we could just try implementing it.

[robertnishihara]

@stephanie-wang, regarding

1. Wait for o to appear in in-process store. # Release resources.
2. Wait for o to appear in local object store. # Hold resources.

I'm a bit confused about what those two scenarios mean. Is it the same as saying the following?

1. If the object hasn't been created in the system yet, release the resources.
2. If the object exists somewhere in the system (including on disk) but not on the local machine, then hold the resources.

[stephanie-wang]

@stephanie-wang, regarding

1. Wait for o to appear in in-process store. # Release resources.
2. Wait for o to appear in local object store. # Hold resources.

I'm a bit confused about what those two scenarios mean. Is it the same as saying the following?

1. If the object hasn't been created in the system yet, release the resources.

2. If the object exists somewhere in the system (including on disk) but not on the local machine, then hold the resources.

Yes.

[ericl]

Ok, so formalizing the proposal a bit more, suppose we call Get(oids) or Wait(oids), this currently has two phases:

// phase 1
in_plasma_oids = local_memory_store->Get(oids)   // always done, all objects are "created" at this point
// phase 2
plasma_store->Get(in_plasma_oids)  // only for objects where data is actually in plasma

Currently both phases can release resources. The proposal is to change it so that only phase 1 can release resources. Phase 2 will never release resources. Is that right?

The only thing is that this can still behave poorly for borrowed object refs since it relies on an RPC to the owner during step 1. But it should be okay as long as the RPC is not too delayed.

IIUC, this is a race condition where we might spuriously unblock right?

Update: I actually tried this out on top of Sang's PR (setting release_resources=false for all NotifyDirectCallBlocked RPCs in plasma_store_provider.cc), and it seems to do the trick. The race condition didn't seem to have any considerable effect in this workload.

[edoakes]

Agree with @robertnishihara that we should try to do this without user input if at all possible.

@ericl the proposal sounds good to me -- this allows better "fair sharing" of network B/W when fetching large objects within the cluster as well. There are basically two common cases here and for both it seems preferable to hold resources:

1. The object is small, in which case network transfer overhead/latency will be low so you don't want the overhead of releasing/acquiring resources.
2. The object is large, in which case network transfer overhead will be high so the worker actually is consuming significant resources (albeit not directly the resources specified in the scheduling logic).

There may also be some degenerate case where the remote raylet is very busy but the local one isn't so the object transfer for a large object is delayed without consuming local resources, but it seems best to optimize for the common cases.

To clarify, phase 1 will always release resources, right? (assuming the object isn't already available locally before the call)

[ericl]

Right, I also agree the new behavior should be much better. The possibility of regression is there, but unlikely (and we can make it a feature flag for now to be safe).

To clarify, phase 1 will always release resources, right? (assuming the object isn't already available locally before the call)

The current behavior will be unchanged (it will release if there are any "pending" objects).

@rkooo567 , for the spilling PR I think the only ramifications are:

• we should keep the release_resources flag internal (not expose in worker.py)
• in a followup PR, add a ray config release_resources_during_plasma_fetch = false that defaults release_resources to false for plasma store provider to implement the proposal above

[stephanie-wang]

About this:

Update: I actually tried this out on top of Sang's PR (setting release_resources=false for all NotifyDirectCallBlocked RPCs in plasma_store_provider.cc), and it seems to do the trick. The race condition didn't seem to have any considerable effect in this workload.

I'm thinking now that it would make sense to split the handlers for releasing resources vs fetching objects in the raylet. It no longer makes sense to keep them in one handler since we'll always set release_resources=false when fetching objects anyway. Also, this has been a brittle/complicated part of the NodeManager code for a while so it'd be nice to make it more sane. So now it will be:

1. Wait for o to appear in in-process store. # Send ReleaseResources/ReacquireResources RPC to raylet. No fetching.
2. Wait for o to appear in local object store. # Send Fetch/CancelFetch to raylet, resources still held.

[rkooo567]

@ericl I believe it is done? Please reopen if not