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 _run_trial_async(self, trial):
      def finish_trial_run(result_or_e):
          if not isinstance(result_or_e, BaseException):
              self._finish_trial_run(
                  is_success=True,
                  is_cancelled=self.trials._fmin_cancelled,
                  trial=trial,
                  data=result_or_e,
              )
              logger.debug(
                  "trial {tid} task thread exits normally and writes results "
                  "back correctly.".format(tid=trial["tid"])
              )
          else:
              self._finish_trial_run(
                  is_success=False,
                  is_cancelled=self.trials._fmin_cancelled,
                  trial=trial,
                  data=result_or_e,
              )
              logger.debug(
                  "trial {tid} task thread catches an exception and writes the "
                  "info back correctly.".format(tid=trial["tid"])
              )
  
      def run_task_thread():
          local_eval_function, local_space = (
              self.eval_function,
              self.space,
          )
          params = self._get_spec_from_trial(trial)
  
          def run_task_on_executor(_):
              import traceback
  
              domain = base.Domain(
                  local_eval_function,
                  local_space,
                  pass_expr_memo_ctrl=None,
              )
  
              try:
                  result = domain.evaluate(
                      params, ctrl=None, attach_attachments=False
                  )
                  yield result
              except BaseException as e:
                  # Because the traceback is not pickable, we need format it and pass it back
                  # to driver
                  _traceback_string = traceback.format_exc()
                  logger.error(_traceback_string)
                  e._tb_str = _traceback_string
                  yield e
  
          try:
              worker_rdd = self.spark.sparkContext.parallelize([0], 1)
              if self.resource_profile:
                  worker_rdd = worker_rdd.withResources(
                      self.resource_profile
                  )
              if self.trials._spark_supports_job_cancelling:
                  if self.trials._spark_pinned_threads_enabled:
                      spark_context = self.spark.sparkContext
                      spark_context.setLocalProperty(
                          "spark.jobGroup.id", self._job_group_id
                      )
                      spark_context.setLocalProperty(
                          "spark.job.description", self._job_desc
                      )
                      spark_context.setLocalProperty(
                          "spark.job.interruptOnCancel",
                          str(self._job_interrupt_on_cancel).lower(),
                      )
                      result_or_e = worker_rdd.mapPartitions(
                          run_task_on_executor
                      ).collect()[0]
                  else:
                      result_or_e = worker_rdd.mapPartitions(
                          run_task_on_executor
                      ).collectWithJobGroup(
                          self._job_group_id,
                          self._job_desc,
                          self._job_interrupt_on_cancel,
                      )[
                          0
                      ]
              else:
                  result_or_e = worker_rdd.mapPartitions(
                      run_task_on_executor
                  ).collect()[0]
          except BaseException as e:
              # I recommend to catch all exceptions here, it can make the program more robust.
              # There're several possible reasons lead to raising exception here.
              # so I use `except BaseException` here.
              #
              # If cancelled flag is set, it represent we need to cancel all running tasks,
              # Otherwise it represent the task failed.
              finish_trial_run(e)
          else:
              # The exceptions captured in run_task_on_executor would be returned in the result_or_e
              finish_trial_run(result_or_e)
  
      if self.trials._spark_pinned_threads_enabled:
          try:
              # pylint: disable=no-name-in-module,import-outside-toplevel
              from pyspark import inheritable_thread_target
  
              run_task_thread = inheritable_thread_target(
                  run_task_thread
              )
          except ImportError:
              pass
  
      task_thread = threading.Thread(target=run_task_thread)
      task_thread.setDaemon(True)
      task_thread.start()
      self._task_threads.add(task_thread)
  
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 11 for calling the _SparkFMinState obj's underlying member method code with example usage: obj._run_trial_async(trial) without return types