Time streaming exec scheduling

python/ray/data/_internal/execution/streaming_executor.py

@@ -208,8 +208,15 @@ def run(self):
         """
         try:
             # Run scheduling loop until complete.
-            while self._scheduling_loop_step(self._topology) and not self._shutdown:
-                pass
+            while True:
+                # use process_time to avoid timing ray.wait in _scheduling_loop_step
+                t_start = time.process_time()
+                continue_sched = self._scheduling_loop_step(self._topology)
+                self._initial_stats.streaming_exec_schedule_s.add(
+                    time.process_time() - t_start
+                )
+                if not continue_sched or self._shutdown:
+                    break
         except Exception as e:
             # Propagate it to the result iterator.
             self._output_node.mark_finished(e)

python/ray/data/_internal/stats.py

@@ -601,6 +601,9 @@ def __init__(
         self.needs_stats_actor = needs_stats_actor
         self.stats_uuid = stats_uuid
 
+        # Streaming executor stats
+        self.streaming_exec_schedule_s: Timer = Timer()
+
         # Iteration stats, filled out if the user iterates over the dataset.
         self.iter_wait_s: Timer = Timer()
         self.iter_get_s: Timer = Timer()