update to actor list limit to fix issue 803

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

@@ -21,6 +21,8 @@
 from ray.exceptions import RayError
 from ray.util.actor_pool import ActorPool
 
+MAX_LIST = 10000  # Max number of actors returned by list
+
 
 class RayUtils:
     """
@@ -110,15 +112,21 @@ def operator() -> ActorHandle:
             return clazz.options(**actor_options).remote(params)
 
         cls_name = clazz.__class__.__name__.replace('ActorClass(', '').replace(')','')
+        current = len(list_actors(filters=[("class_name", "=", cls_name), ("state", "=", "ALIVE")], limit=MAX_LIST))
         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 = current + n_actors
+        if overall < MAX_LIST:
+            n_list = min(overall + 10, MAX_LIST)
+            for i in range(120):
+                time.sleep(1)
+                alive = list_actors(filters=[("class_name", "=", cls_name), ("state", "=", "ALIVE")], limit=n_list)
+                if len(actors) >= len(alive) - current:
+                    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")
+        else:
+            print(f"Overall number of actors of class {clazz.__name__} is {overall}. Skipping readiness check")
 
     @staticmethod
     def process_files(