[4/N] Support async actor and async generator interface. (#35584) (#3…

…5752)

NOTE: It is a big PR, but most of code is testing (200~300 lines I believe)

This is the fourth PR to support the streaming generator.

The detailed design and API proposal can be found from https://docs.google.com/document/d/1hAASLe2sCoay23raqxqwJdSDiJWNMcNhlTwWJXsJOU4/edit#heading=h.w91y1fgnpu0m.
The Execution plan can be found from https://docs.google.com/document/d/1hAASLe2sCoay23raqxqwJdSDiJWNMcNhlTwWJXsJOU4/edit#heading=h.kxktymq5ihf7.
There will be 4 PRs to enable streaming generator for Ray Serve (phase 1).

This PR -> introduce cpp interfaces to handle intermediate task return [1/N] Streaming Generator. Cpp interfaces and implementation #35291
Support core worker APIs + cython generator interface. [2/N] Streaming Generator. Support core worker APIs + cython generator interface. #35324
E2e integration [3/N] Streaming Generator. E2e integration #35325 (review)
Support async actors [4/N] Support async actor and async generator interface. #35382 < ---- This PR
adds an async actor execution support to the generator implementation (basically keep posting generator.anext to the event loop)
Impelements a standard async generator interface from Python. (anext and aiter)

python/ray/_private/async_compat.py

@@ -19,10 +19,15 @@ def get_new_event_loop():
         return asyncio.new_event_loop()
 
 
+def is_async_func(func):
+    """Return True if the function is an async or async generator method."""
+    return inspect.iscoroutinefunction(func) or inspect.isasyncgenfunction(func)
+
+
 def sync_to_async(func):
     """Convert a blocking function to async function"""
 
-    if inspect.iscoroutinefunction(func):
+    if is_async_func(func):
         return func
 
     async def wrapper(*args, **kwargs):

python/ray/_private/ray_perf.py

@@ -280,7 +280,6 @@ def async_actor_multi():
         ray.get([async_actor_work.remote(a) for _ in range(m)])
 
     results += timeit("n:n async-actor calls async", async_actor_multi, m * n)
-    ray.shutdown()
 
     NUM_PGS = 100
     NUM_BUNDLES = 1

python/ray/_raylet.pxd

@@ -39,7 +39,8 @@ from ray.includes.libcoreworker cimport (
 
 from ray.includes.unique_ids cimport (
     CObjectID,
-    CActorID
+    CActorID,
+    CTaskID,
 )
 from ray.includes.function_descriptor cimport (
     CFunctionDescriptor,
@@ -154,6 +155,13 @@ cdef class CoreWorker:
     cdef python_scheduling_strategy_to_c(
         self, python_scheduling_strategy,
         CSchedulingStrategy *c_scheduling_strategy)
+    cdef CObjectID allocate_dynamic_return_id_for_generator(
+            self,
+            const CAddress &owner_address,
+            const CTaskID &task_id,
+            return_size,
+            generator_index,
+            is_async_actor)
 
 cdef class FunctionDescriptor:
     cdef: