class_docstr: LocalTaskJob runs a single task instance.
class_docstr_tok: ["LocalTaskJob","runs","a","single","task","instance","."]
code: def _execute(self) -> int | None:
      from airflow.task.task_runner import get_task_runner
  
      self.task_runner = get_task_runner(self)
  
      def signal_handler(signum, frame):
          """Set kill signal handler."""
          self.log.error("Received SIGTERM. Terminating subprocesses")
          self.task_runner.terminate()
          self.handle_task_exit(128 + signum)
  
      def segfault_signal_handler(signum, frame):
          """Set sigmentation violation signal handler."""
          self.log.critical(SIGSEGV_MESSAGE)
          self.task_runner.terminate()
          self.handle_task_exit(128 + signum)
          raise AirflowException("Segmentation Fault detected.")
  
      def sigusr2_debug_handler(signum, frame):
          import sys
          import threading
          import traceback
  
          id2name = {th.ident: th.name for th in threading.enumerate()}
          for threadId, stack in sys._current_frames().items():
              print(id2name[threadId])
              traceback.print_stack(f=stack)
  
      signal.signal(signal.SIGSEGV, segfault_signal_handler)
      signal.signal(signal.SIGTERM, signal_handler)
  
      if not IS_WINDOWS:
          # This is not supported on Windows systems
          signal.signal(signal.SIGUSR2, sigusr2_debug_handler)
  
      if not self.task_instance.check_and_change_state_before_execution(
          mark_success=self.mark_success,
          ignore_all_deps=self.ignore_all_deps,
          ignore_depends_on_past=self.ignore_depends_on_past,
          wait_for_past_depends_before_skipping=self.wait_for_past_depends_before_skipping,
          ignore_task_deps=self.ignore_task_deps,
          ignore_ti_state=self.ignore_ti_state,
          job_id=str(self.job.id),
          pool=self.pool,
          external_executor_id=self.external_executor_id,
      ):
          self.log.info("Task is not able to be run")
          return None
  
      return_code = None
      try:
          self.task_runner.start()
          local_task_job_heartbeat_sec = conf.getint(
              "scheduler", "local_task_job_heartbeat_sec"
          )
          if local_task_job_heartbeat_sec < 1:
              heartbeat_time_limit = conf.getint(
                  "scheduler", "scheduler_zombie_task_threshold"
              )
          else:
              heartbeat_time_limit = local_task_job_heartbeat_sec
  
          # LocalTaskJob should not run callbacks, which are handled by TaskInstance._run_raw_task
          # 1, LocalTaskJob does not parse DAG, thus cannot run callbacks
          # 2, The run_as_user of LocalTaskJob is likely not same as the TaskInstance._run_raw_task.
          # When run_as_user is specified, the process owner of the LocalTaskJob must be sudoable.
          # It is not secure to run callbacks with sudoable users.
  
          # If _run_raw_task receives SIGKILL, scheduler will mark it as zombie and invoke callbacks
          # If LocalTaskJob receives SIGTERM, LocalTaskJob passes SIGTERM to _run_raw_task
          # If the state of task_instance is changed, LocalTaskJob sends SIGTERM to _run_raw_task
          while not self.terminating:
              # Monitor the task to see if it's done. Wait in a syscall
              # (`os.wait`) for as long as possible so we notice the
              # subprocess finishing as quick as we can
              max_wait_time = max(
                  0,  # Make sure this value is never negative,
                  min(
                      (
                          heartbeat_time_limit
                          - (
                              timezone.utcnow()
                              - self.job.latest_heartbeat
                          ).total_seconds()
                          * 0.75
                      ),
                      self.job.heartrate
                      if self.job.heartrate is not None
                      else heartbeat_time_limit,
                  ),
              )
  
              return_code = self.task_runner.return_code(
                  timeout=max_wait_time
              )
              if return_code is not None:
                  self.handle_task_exit(return_code)
                  return return_code
  
              perform_heartbeat(
                  job=self.job,
                  heartbeat_callback=self.heartbeat_callback,
                  only_if_necessary=False,
              )
  
              # If it's been too long since we've heartbeat, then it's possible that
              # the scheduler rescheduled this task, so kill launched processes.
              # This can only really happen if the worker can't read the DB for a long time
              time_since_last_heartbeat = (
                  timezone.utcnow() - self.job.latest_heartbeat
              ).total_seconds()
              if time_since_last_heartbeat > heartbeat_time_limit:
                  Stats.incr(
                      "local_task_job_prolonged_heartbeat_failure", 1, 1
                  )
                  self.log.error("Heartbeat time limit exceeded!")
                  raise AirflowException(
                      f"Time since last heartbeat({time_since_last_heartbeat:.2f}s) exceeded limit "
                      f"({heartbeat_time_limit}s)."
                  )
          return return_code
      finally:
          self.on_kill()
  
imports: from __future__ import annotations
  import signal
  from typing import TYPE_CHECKING
  import psutil
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException
  from airflow.jobs.base_job_runner import BaseJobRunner
  from airflow.jobs.job import perform_heartbeat
  from airflow.models.taskinstance import TaskReturnCode
  from airflow.stats import Stats
  from airflow.utils import timezone
  from airflow.utils.log.file_task_handler import _set_task_deferred_context_var
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.net import get_hostname
  from airflow.utils.platform import IS_WINDOWS
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 2 for calling the LocalTaskJobRunner obj's underlying member method code with example usage: obj._execute() and returns: None, return_code, return_code