def stop(self, drain_queue_and_join=True):
        """Stops running threads and wait for them to exit, if necessary.

        This method is thread safe and is called from various threads. Note that
        the `drain_queue_and_join` argument must be set correctly.
        It is safe to call this method multiple times, extra calls are ignored.

        Args:
            drain_queue_and_join: set to True to drain the queue of pending
                items and wait for the worker thread to complete. Set to False
                if invoked from a worker thread to avoid deadlocks. Note that
                setting this to False means this enqueuer won't be reused.
        """
        with self.start_stop_lock:
            if not self.running:
                return
            self.running = False

            if drain_queue_and_join:
                # Drain the `future_queue` and put items in `ready_queue` for
                # the next run.
                while True:
                    try:
                        value = self.future_queue.get(block=True, timeout=0.1)
                        if isinstance(value, Exception):
                            raise value  # Propagate exception from other thread
                        inputs = value.get()
                        self.future_queue.task_done()
                        if inputs is not None:
                            self.ready_queue.put(inputs)
                    except queue.Empty:
                        break
                self.run_thread.join()

            self.run_thread = None
            _SHARED_SEQUENCES[self.uid] = None
Stops running threads and wait for them to exit, if necessary.

        This method is thread safe and is called from various threads. Note that
        the `drain_queue_and_join` argument must be set correctly.
        It is safe to call this method multiple times, extra calls are ignored.

        Args:
            drain_queue_and_join: set to True to drain the queue of pending
                items and wait for the worker thread to complete. Set to False
                if invoked from a worker thread to avoid deadlocks. Note that
                setting this to False means this enqueuer won't be reused.
        
keras/src/trainers/data_adapters/py_dataset_adapter.py