keras/src/trainers/data_adapters/py_dataset_adapter.py/OrderedEnqueuer/_run
class OrderedEnqueuer:    def _run(self):
        """Submits request to the executor and queue the `Future` objects."""
        try:
            num_batches = self.py_dataset.num_batches
            indices = (
                range(num_batches)
                if num_batches is not None
                else itertools.count()
            )
            if self.shuffle and num_batches is not None:
                indices = list(indices)
                random.shuffle(indices)
            self._send_py_dataset()  # Share the initial py_dataset
            while True:
                with closing(self.executor_fn(_SHARED_SEQUENCES)) as executor:
                    for i in indices:
                        if self.stop_signal.is_set():
                            return

                        self.queue.put(
                            executor.apply_async(get_index, (self.uid, i)),
                            block=True,
                        )

                    # Done with the current epoch, waiting for the final batches
                    self._wait_queue()

                    if self.stop_signal.is_set():
                        # We're done
                        return

                # Call the internal on epoch end and epoch begin.
                self.py_dataset.on_epoch_end()
                # The first on_epoch_begin call was already made,
                # prior to the start of the Enqueuer.
                self.py_dataset.on_epoch_begin()
                self._send_py_dataset()  # Update the pool
        except Exception as e:
            self.queue.put(e)  # Report exception