docstr: Submit one round of tasks. If we already have the max concurrent rounds
  in flight, first wait for the oldest round of tasks to finish.
docstr_tok: ["Submit","one","round","of","tasks",".","If","we","already","have","the","max","concurrent","rounds","in","flight",",","first","wait","for","the","oldest","round","of","tasks","to","finish","."]
code: def __next__(self):
      """
      Submit one round of tasks. If we already have the max concurrent rounds
      in flight, first wait for the oldest round of tasks to finish.
      """
      prev_metadata = []
      if all(len(r) == 0 for r in self._rounds):
          raise StopIteration
  
      if len(self._rounds) >= self._max_concurrent_rounds:
          prev_metadata_refs = self._rounds.pop(0)
          if prev_metadata_refs:
              if self._progress_bar is not None:
                  prev_metadata = (
                      self._progress_bar.fetch_until_complete(
                          prev_metadata_refs
                      )
                  )
              else:
                  prev_metadata = ray.get(prev_metadata_refs)
  
      self._submit_round()
  
      return prev_metadata
  
imports: import logging
  import math
  from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.shuffle import ShuffleOp
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 3 for calling the _PipelinedStageExecutor obj's underlying member method code with example usage: obj.__next__() and returns: prev_metadata