class_docstr: Class for managing threads which run concurrent Spark jobs.
  
  This maintains a primary dispatcher thread, plus 1 thread per Hyperopt trial.
  Each trial's thread runs 1 Spark job with 1 task.
code: def launch_dispatcher(self):
      def run_dispatcher():
          prev_num_ok_trials = (
              0  # number of trials that succeeded with STATUS_OK
          )
          start_time = timeit.default_timer()
  
          while not self._fmin_done:
              new_tasks = self._poll_new_tasks()
  
              for trial in new_tasks:
                  self._run_trial_async(trial)
  
              cur_time = timeit.default_timer()
              elapsed_time = cur_time - start_time
  
              # check early stopping condition if it is defined
              if self.early_stop_fn:
                  curr_num_ok_trials = 0
                  for trial in self.trials:
                      if trial["result"]["status"] == STATUS_OK:
                          curr_num_ok_trials += 1
  
                  if curr_num_ok_trials > prev_num_ok_trials:
                      # check early stopping condition
                      stop, kwargs = self.early_stop_fn(
                          self.trials, *self.early_stop_args
                      )
                      self.early_stop_args = kwargs
  
                      if stop:
                          # same logic as timeouts
                          self.trials._fmin_cancelled = True
                          self.trials._fmin_cancelled_reason = (
                              "early stopping condition"
                          )
                          self._cancel_running_trials()
                          logger.warning(
                              "fmin cancelled because of "
                              + self.trials._fmin_cancelled_reason
                          )
                      prev_num_ok_trials = curr_num_ok_trials
  
              # In the future, timeout checking logic could be moved to `fmin`.
              # For now, timeouts are specific to SparkTrials.
              # When a timeout happens:
              #  - Set `trials._fmin_cancelled` flag to be True.
              #  - FMinIter checks this flag and exits if it is set to True.
              if (
                  self.trials.timeout is not None
                  and elapsed_time > self.trials.timeout
                  and not self.trials._fmin_cancelled
              ):
                  self.trials._fmin_cancelled = True
                  self.trials._fmin_cancelled_reason = (
                      "fmin run timeout"
                  )
                  self._cancel_running_trials()
                  logger.warning(
                      "fmin cancelled because of "
                      + self.trials._fmin_cancelled_reason
                  )
  
              time.sleep(1)
  
          if self.trials._fmin_cancelled:
              # Because cancelling fmin triggered, warn that the dispatcher won't launch
              # more trial tasks.
              logger.warning(
                  "fmin is cancelled, so new trials will not be launched."
              )
  
          logger.debug("dispatcher thread exits normally.")
  
      self._dispatcher_thread = threading.Thread(target=run_dispatcher)
      self._dispatcher_thread.setDaemon(True)
      self._dispatcher_thread.start()
  
imports: import copy
  import os
  import threading
  import time
  import timeit
  import traceback
  from hyperopt import base, fmin, Trials
  from hyperopt.base import validate_timeout, validate_loss_threshold, STATUS_OK
  from hyperopt.utils import coarse_utcnow, _get_logger, _get_random_id
label_desc: Use image node_id 6 for calling the _SparkFMinState obj's underlying member method code with example usage: obj.launch_dispatcher() without return types