[pull] master from ray-project:master

cpp/src/ray/runtime/task/task_executor.cc

@@ -140,6 +140,7 @@ Status TaskExecutor::ExecuteTask(
     bool is_streaming_generator,
     bool retry_exception,
     int64_t generator_backpressure_num_objects,
+    int64_t num_objects_per_yield,
     const std::optional<std::string> &tensor_transport) {
   RAY_LOG(DEBUG) << "Execute task type: " << TaskType_Name(task_type)
                  << " name:" << task_name;

cpp/src/ray/runtime/task/task_executor.h

@@ -97,6 +97,7 @@ class TaskExecutor {
       bool is_streaming_generator,
       bool retry_exception,
       int64_t generator_backpressure_num_objects,
+      int64_t num_objects_per_yield,
       /* This is used by the in-actor RDT object store. However, it is only supported in
        * the Python frontend. */
       const std::optional<std::string> &tensor_transport);

python/ray/_common/ray_option_utils.py

@@ -221,6 +221,16 @@ def issubclass_safe(obj: Any, cls_: type) -> bool:
             "whenever `next` is called). Use -1 to disable this feature. "
         ),
     ),
+    "_num_objects_per_yield": Option(
+        (int, type(None)),
+        lambda x: None
+        if (x is None or x > 0)
+        else (
+            "_num_objects_per_yield is a private streaming generator option "
+            "that must be set to a positive integer."
+        ),
+        default_value=1,
+    ),
 }
 
 _actor_only_options = {