update to actor list limit to fix issue 803

data-processing-lib/ray/src/data_processing_ray/runtime/ray/ray_utils.py

@@ -16,11 +16,13 @@
 
 import ray
 from ray.experimental.state.api import list_actors
-from data_processing.utils import GB, UnrecoverableException
+from data_processing.utils import GB, UnrecoverableException, get_logger
 from ray.actor import ActorHandle
 from ray.exceptions import RayError
 from ray.util.actor_pool import ActorPool
 
+MAX_LIST = 10000  # Max number of actors returned by list
+
 
 class RayUtils:
     """
@@ -109,16 +111,35 @@ def operator() -> ActorHandle:
             time.sleep(creation_delay)
             return clazz.options(**actor_options).remote(params)
 
+        logger = get_logger(__name__)
         cls_name = clazz.__class__.__name__.replace('ActorClass(', '').replace(')','')
+        current = list_actors(filters=[("class_name", "=", cls_name), ("state", "=", "ALIVE")], limit=MAX_LIST)
+        c_len = len(current)
         actors = [operator() for _ in range(n_actors)]
-        for i in range(120):
-            time.sleep(1)
-            alive = list_actors(filters=[("class_name", "=", cls_name), ("state", "=", "ALIVE")])
-            if len(actors) == len(alive):
-                return actors
-        # failed - raise an exception
-        print(f"created {actors}, alive {alive}")
-        raise UnrecoverableException(f"out of {len(actors)} created actors only {len(alive)} alive")
+        overall = c_len + n_actors
+        if overall < MAX_LIST:
+            n_list = min(overall + 10, MAX_LIST)
+            alive = []
+            for i in range(120):
+                time.sleep(1)
+                alive = list_actors(filters=[("class_name", "=", cls_name), ("state", "=", "ALIVE")], limit=n_list)
+                if len(alive) >= n_actors + c_len:
+                    return actors
+            # failed
+            if len(alive) >= n_actors/2 + c_len:
+                # At least half of the actors were created
+                logger.info(f"created {n_actors}, alive {len(alive)} Running with less actors")
+                created_ids = [item.actor_id for item in alive if item not in current]
+                return [
+                    actor for actor in actors
+                    if (str(actor._ray_actor_id).
+                        replace('ActorID(', '').replace(')','') in created_ids)]
+            else:
+                # too few actors created
+                raise UnrecoverableException(f"Created {n_actors}, alive {len(alive)}. Too few actors were created")
+        else:
+            raise UnrecoverableException(f"Overall number of actors of class {clazz.__name__} is {overall}. "
+                                         f"Too many actors to create, greater then {MAX_LIST}")
 
     @staticmethod
     def process_files(