ray.wait(fetch_local=False) in asyncio

NathanTP · May 3, 2022, 12:53am

How severe does this issue affect your experience of using Ray?

Medium: It contributes to significant difficulty to complete my task, but I can work around it.

Short Version
Is there any way to use ray.wait with fetch_local=False in asyncio? From what I can tell, awaiting a reference, even as part of asyncio.wait(), will always materialize the references on the current worker.

More detailed scenario:
I’m writing what is essentially an asynchronous actor pool. It takes requests, returns a reference immediately, and then forwards the request to an actor at some unknown time in the future. The pool itself is a Ray actor with an asyncio event loop running. The problem I’m facing is that I won’t know which actor to use for any particular request until all of its inputs are ready. From what I can tell, using “await asyncio.wait(asyncio.wrap_future(inputReference.future()))” will materialize inputReference on the pool actor. This would be unpleasant if inputReference points to something big and expensive.

Possible Solutions

Ideally, I’d just be able to add some sort of “fetch_local=False” flag to the Reference’s future or asyncio.wait or something.
Another solution might be a way to specify required input references to an actor invocation without materializing them (like a dependency list or something).
In a previous version of this, I used a thread pool to wrap ray.wait with concurrent.futures.ThreadPoolExecutor and used loop.run_in_executor to make ray.wait compatible with asyncio. The problem here is that I may have many pending requests and it’s not ideal to have a ton of threads floating around. Ultimately, this design led to some nasty head of line blocking issues. I’ll try to do some more principled profiling here to give concrete numbers but I’m pretty sure I was seeing slowdowns when I added 64+ threads.

yic · June 2, 2022, 6:55am

How about this solution:
when return you return two objects (Tasks — Ray 1.12.1), one is the dummy return, and the other one is the actual result. You get the dummy one. We are sure if the dummy one is ready, the actual one should also be ready.

But yeah, I think support fetch_local is a cleaner way but it also requires more work. You can submit some issues or feature requests about this one if it’s important. Or even better, it’ll be nice if you can help contribute it

NathanTP · June 2, 2022, 5:07pm

Ya, I use the dummy return trick elsewhere in the project. In this case I can’t do it because the references come from the user of the library. They could come from anywhere. I suppose I could require the user to provide a dummy reference for me to wait on, but that’s a big burden on the user.

I’m happy to look into implementing it. I don’t know a lot about how the asyncio wrapping works. It might be as easy as messing with the future that references get wrapped in. It might be easy…or not lol. Do you have any pointers to where that is handled in Ray or maybe a reference on how to asyncio-ify stuff?

yic · June 2, 2022, 8:13pm

Another solution could be having a thread there, wrapping with futures like 3). but when you wait, you call ray.wait([obj_refs]). Basically, group all the waiting refs.

As to support async with wait, we firstly need an API for this,

github.com

ray-project/ray/blob/f8551942bf44c2b54dbad982ddcf3e69f7797dff/python/ray/includes/object_ref.pxi#L117

      
        
                self.data = CObjectID.FromBinary(<c_string>id)
            
            
@classmethod
            def nil(cls):
                return cls(CObjectID.Nil().Binary())
            
            
@classmethod
            def from_random(cls):
                return cls(CObjectID.FromRandom().Binary())
            
            
def future(self) -> concurrent.futures.Future:
                """Wrap ObjectRef with a concurrent.futures.Future
            
            
    Note that the future cancellation will not cancel the correspoding
                task when the ObjectRef representing return object of a task.
                Additionally, future.running() will always be ``False`` even if the
                underlying task is running.
                """
                py_future = concurrent.futures.Future()
            
            
    self._on_completed(

Maybe we can add future(fetch_local=True) as default one? (we need an issue for this and get approved since it’s API change).

Then, for the callback side, we should have AsyncWait implemented which is similar to this one:

github.com

ray-project/ray/blob/f8551942bf44c2b54dbad982ddcf3e69f7797dff/src/ray/core_worker/core_worker.cc#L1277

      
        
                &ready));
            RAY_CHECK(static_cast<int>(ready.size()) <= num_objects);
            if (timeout_ms > 0) {
              timeout_ms =
                  std::max(0, static_cast<int>(timeout_ms - (current_time_ms() - start_time)));
            }
            if (fetch_local) {
              RetryObjectInPlasmaErrors(
                  memory_store_, worker_context_, memory_object_ids, plasma_object_ids, ready);
              if (static_cast<int>(ready.size()) < num_objects && plasma_object_ids.size() > 0) {
                RAY_RETURN_NOT_OK(plasma_store_provider_->Wait(
                    plasma_object_ids,
                    std::min(static_cast<int>(plasma_object_ids.size()),
                             num_objects - static_cast<int>(ready.size())),
                    timeout_ms,
                    worker_context_,
                    &ready));
              }
            }
            RAY_CHECK(static_cast<int>(ready.size()) <= num_objects);

Basically, now the callback is only set for set_get_async_callback

github.com

ray-project/ray/blob/f8551942bf44c2b54dbad982ddcf3e69f7797dff/python/ray/includes/object_ref.pxi#L155

      
        
                                   "asyncio.wrap_future(ref.future()).")
                return asyncio.wrap_future(self.future())
            
            
def _on_completed(self, py_callback: Callable[[Any], None]):
                """Register a callback that will be called after Object is ready.
                If the ObjectRef is already ready, the callback will be called soon.
                The callback should take the result as the only argument. The result
                can be an exception object in case of task error.
                """
                core_worker = ray.worker.global_worker.core_worker
                core_worker.set_get_async_callback(self, py_callback)
                return self

we need to have something for set_wait_async_callback(fetch_local=False). Something like this.

NathanTP · June 2, 2022, 9:01pm

I made a feature request on github to track this:

github.com/ray-project/ray

Core

opened 09:00PM - 02 Jun 22 UTC

NathanTP

enhancement

### Description add ray.wait(..., fetch_local=False) like behavior in asyncio … At a high level, I would like to determine if a reference is ready without materializing it in an asyncio context. In a non-asyncio context this is achieved with ray.wait(fetch_local=False) but ray.wait is not available with asyncio. Possible Solutions: 1. A likely solution would be to add a 'fetch_local' kwarg to Reference.future(). 2. We could make an awaitable version of ray.wait. 3. In a previous version of this actor pool, I used a thread pool to wrap ray.wait with concurrent.futures.ThreadPoolExecutor and used loop.run_in_executor to make ray.wait compatible with asyncio. The problem here is that I may have many pending requests and it’s not ideal to have a ton of threads floating around. Ultimately, this design led to some nasty head of line blocking issues. I’ll try to do some more principled profiling here to give concrete numbers but I’m pretty sure I was seeing slowdowns when I added 64+ threads. (this feature request started life on the mailing list: https://discuss.ray.io/t/ray-wait-fetch-local-false-in-asyncio/6002) ### Use case The 'fetch_local' option for ray.wait is presumably useful to some and it makes sense to extend it to asyncio. In my case specifically, I’m writing what is essentially an asynchronous actor pool. It takes requests, returns a reference immediately, and then forwards the request to an actor at some unknown time in the future. The pool itself is a Ray actor with an asyncio event loop running. The problem I’m facing is that I won’t know which actor to use for any particular request until all of its inputs are ready. Using “await asyncio.wait(asyncio.wrap_future(inputReference.future()))” will materialize inputReference on the pool actor. This would be unpleasant if inputReference points to something big and expensive.

Topic		Replies	Views
How `ray.wait()` works Ray Core	2	401	August 22, 2022
Using ray for submitting async tasks from a FastAPI backend Ray Core	2	304	April 22, 2022
Feature request: Allow ray.wait() to do the necessary work for an instant ray.get() Ray Core	16	395	May 25, 2021
RuntimeWarning: coroutine 'Queue.put_async' was never awaited Ray Core	1	1110	June 21, 2022
Does polling `ray.wait([task_handle], timeout=0)` guarantee to eventually finish? Ray Core	11	540	April 27, 2023

ray.wait(fetch_local=False) in asyncio

Related topics