[core] Split raylet cython file into multiple files (DynamicObjectRefGenerator)

python/ray/_private/object_ref_generator.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+from ray.util.annotations import DeveloperAPI
+from typing import Iterator, List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import ray
+
+
+@DeveloperAPI
+class DynamicObjectRefGenerator:
+    def __init__(self, refs: List["ray.ObjectRef"]):
+        # TODO(swang): As an optimization, can also store the generator
+        # ObjectID so that we don't need to keep individual ref counts for the
+        # inner ObjectRefs.
+        self._refs: List["ray.ObjectRef"] = refs
+
+    def __iter__(self) -> Iterator("ray.ObjectRef"):
+        for ref in self._refs:
+            yield ref
+
+    def __len__(self) -> int:
+        return len(self._refs)

python/ray/_raylet.pyx

@@ -224,6 +224,7 @@ import ray.core.generated.common_pb2 as common_pb2
 import ray._private.memory_monitor as memory_monitor
 import ray._private.profiling as profiling
 from ray._private.utils import decode, DeferSigint
+from ray._private.object_ref_generator import DynamicObjectRefGenerator
 from ray.util.annotations import PublicAPI
 
 cimport cpython
@@ -276,22 +277,6 @@ async_task_name = contextvars.ContextVar('async_task_name', default=None)
 async_task_function_name = contextvars.ContextVar('async_task_function_name',
                                                   default=None)
 
-
-class DynamicObjectRefGenerator:
-    def __init__(self, refs):
-        # TODO(swang): As an optimization, can also store the generator
-        # ObjectID so that we don't need to keep individual ref counts for the
-        # inner ObjectRefs.
-        self._refs = refs
-
-    def __iter__(self):
-        while self._refs:
-            yield self._refs.pop(0)
-
-    def __len__(self):
-        return len(self._refs)
-
-
 class ObjectRefGenerator:
     """A generator to obtain object references
     from a task in a streaming manner.