def start_evaluator(
        self,
        restart: bool = False,
        fork: bool = False,
        disable_tqdm: bool = False,
        run_location: Optional[feedback_schema.FeedbackRunLocation] = None,
        return_when_done: bool = False,
    ) -> Optional[Union[Process, Thread]]:
        """
        Start a deferred feedback function evaluation thread or process.

        Args:
            restart: If set, will stop the existing evaluator before starting a
                new one.

            fork: If set, will start the evaluator in a new process instead of a
                thread. NOT CURRENTLY SUPPORTED.

            disable_tqdm: If set, will disable progress bar logging from the evaluator.

            run_location: Run only the evaluations corresponding to run_location.

            return_when_done: Instead of running asynchronously, will block until no feedbacks remain.

        Returns:
            If return_when_done is True, then returns None. Otherwise, the started process or thread
                that is executing the deferred feedback evaluator.

        Relevant constants:
            [RETRY_RUNNING_SECONDS][trulens.core.session.TruSession.RETRY_RUNNING_SECONDS]

            [RETRY_FAILED_SECONDS][trulens.core.session.TruSession.RETRY_FAILED_SECONDS]

            [DEFERRED_NUM_RUNS][trulens.core.session.TruSession.DEFERRED_NUM_RUNS]

            [MAX_THREADS][trulens.core.utils.threading.TP.MAX_THREADS]
        """
        if is_otel_tracing_enabled():
            raise ValueError(
                "Deferred evaluator not supported with OTEL tracing!"
            )

        assert not fork, "Fork mode not yet implemented."
        assert (
            (not fork) or (not return_when_done)
        ), "fork=True implies running asynchronously but return_when_done=True does not!"

        if self._evaluator_proc is not None:
            if restart:
                self.stop_evaluator()
            else:
                raise RuntimeError(
                    "Evaluator is already running in this process."
                )

        if not fork:
            self._evaluator_stop = threading.Event()

        def runloop(stop_when_none_left: bool = False):
            assert self._evaluator_stop is not None

            print(
                f"Will keep max of "
                f"{self.DEFERRED_NUM_RUNS} feedback(s) running."
            )
            print(
                f"Tasks are spread among max of "
                f"{threading_utils.TP.MAX_THREADS} thread(s)."
            )
            print(
                f"Will rerun running feedbacks after "
                f"{text_utils.format_seconds(self.RETRY_RUNNING_SECONDS)}."
            )
            print(
                f"Will rerun failed feedbacks after "
                f"{text_utils.format_seconds(self.RETRY_FAILED_SECONDS)}."
            )

            total = 0

            # TODO: a lot of the time we say `if tqdm`, but shouldn't we say `if not disable_tqdm`?
            if tqdm:
                # Getting total counts from the database to start off the tqdm
                # progress bar initial values so that they offer accurate
                # predictions initially after restarting the process.
                queue_stats = self.connector.db.get_feedback_count_by_status()
                queue_done = (
                    queue_stats.get(feedback_schema.FeedbackResultStatus.DONE)
                    or 0
                )
                queue_total = sum(queue_stats.values())

                # Show the overall counts from the database, not just what has been
                # looked at so far.
                tqdm_status = tqdm(
                    desc="Feedback Status",
                    initial=queue_done,
                    unit="feedbacks",
                    total=queue_total,
                    postfix={
                        status.name: count
                        for status, count in queue_stats.items()
                    },
                    disable=disable_tqdm,
                )

                # Show the status of the results so far.
                tqdm_total = tqdm(
                    desc="Done Runs",
                    initial=0,
                    unit="runs",
                    disable=disable_tqdm,
                )

                # Show what is being waited for right now.
                tqdm_waiting = tqdm(
                    desc="Waiting for Runs",
                    initial=0,
                    unit="runs",
                    disable=disable_tqdm,
                )

            runs_stats = defaultdict(int)

            futures_map: Dict[
                Future[feedback_schema.FeedbackResult], pandas.Series
            ] = dict()

            while fork or not self._evaluator_stop.is_set():
                if len(futures_map) < self.DEFERRED_NUM_RUNS:
                    # Get some new evals to run if some already completed by now.
                    new_futures: List[
                        Tuple[
                            pandas.Series,
                            Future[feedback_schema.FeedbackResult],
                        ]
                    ] = core_feedback.Feedback.evaluate_deferred(
                        limit=self.DEFERRED_NUM_RUNS - len(futures_map),
                        shuffle=True,
                        session=self,
                        run_location=run_location,
                    )

                    # Will likely get some of the same ones that already have running.
                    for row, fut in new_futures:
                        if fut in futures_map:
                            # If the future is already in our set, check whether
                            # its status has changed and if so, note it in the
                            # runs_stats.
                            if futures_map[fut].status != row.status:
                                runs_stats[row.status.name] += 1

                        futures_map[fut] = row
                        total += 1

                    if tqdm:
                        tqdm_total.total = total
                        tqdm_total.refresh()

                if tqdm:
                    tqdm_waiting.total = self.DEFERRED_NUM_RUNS
                    tqdm_waiting.n = len(futures_map)
                    tqdm_waiting.refresh()

                # Note whether we have waited for some futures in this
                # iteration. Will control some extra wait time if there is no
                # work.
                did_wait = False

                if len(futures_map) > 0:
                    did_wait = True

                    futures_copy = list(futures_map.keys())

                    try:
                        for fut in futures.as_completed(
                            futures_copy, timeout=10
                        ):
                            del futures_map[fut]

                            if tqdm:
                                tqdm_waiting.update(-1)
                                tqdm_total.update(1)

                            feedback_result = fut.result()
                            runs_stats[feedback_result.status.name] += 1

                    except futures.TimeoutError:
                        pass

                if tqdm:
                    tqdm_total.set_postfix({
                        name: count for name, count in runs_stats.items()
                    })

                    queue_stats = (
                        self.connector.db.get_feedback_count_by_status()
                    )
                    queue_done = (
                        queue_stats.get(
                            feedback_schema.FeedbackResultStatus.DONE
                        )
                        or 0
                    )
                    queue_total = sum(queue_stats.values())

                    tqdm_status.n = queue_done
                    tqdm_status.total = queue_total
                    tqdm_status.set_postfix({
                        status.name: count
                        for status, count in queue_stats.items()
                    })

                # Check if any of the running futures should be stopped.
                futures_copy = list(futures_map.keys())
                for fut in futures_copy:
                    row = futures_map[fut]

                    if fut.running():
                        # Not checking status here as this will be not yet be set
                        # correctly. The computation in the future updates the
                        # database but this object is outdated.

                        elapsed = datetime.now().timestamp() - row.last_ts
                        if elapsed > self.RETRY_RUNNING_SECONDS:
                            fut.cancel()

                            # Not an actual status, but would be nice to
                            # indicate cancellations in run stats:
                            runs_stats["CANCELLED"] += 1

                            del futures_map[fut]

                if not did_wait:
                    if stop_when_none_left:
                        break
                    # Nothing to run/is running, wait a bit.
                    if fork:
                        sleep(10)
                    else:
                        self._evaluator_stop.wait(10)

            print("Evaluator stopped.")

        if return_when_done:
            runloop(stop_when_none_left=True)
            return None
        else:
            if fork:
                proc = Process(target=runloop)
            else:
                proc = threading_utils.Thread(target=runloop)
                proc.daemon = True
            # Start a persistent thread or process that evaluates feedback functions.
            self._evaluator_proc = proc
            proc.start()
            return proc

        Start a deferred feedback function evaluation thread or process.

        Args:
            restart: If set, will stop the existing evaluator before starting a
                new one.

            fork: If set, will start the evaluator in a new process instead of a
                thread. NOT CURRENTLY SUPPORTED.

            disable_tqdm: If set, will disable progress bar logging from the evaluator.

            run_location: Run only the evaluations corresponding to run_location.

            return_when_done: Instead of running asynchronously, will block until no feedbacks remain.

        Returns:
            If return_when_done is True, then returns None. Otherwise, the started process or thread
                that is executing the deferred feedback evaluator.

        Relevant constants:
            [RETRY_RUNNING_SECONDS][trulens.core.session.TruSession.RETRY_RUNNING_SECONDS]

            [RETRY_FAILED_SECONDS][trulens.core.session.TruSession.RETRY_FAILED_SECONDS]

            [DEFERRED_NUM_RUNS][trulens.core.session.TruSession.DEFERRED_NUM_RUNS]

            [MAX_THREADS][trulens.core.utils.threading.TP.MAX_THREADS]
        
src/core/trulens/core/session.py