class_docstr: Generator implementation class for AsyncPipelineTaskGenerator.
class_docstr_tok: ["Generator","implementation","class","for","AsyncPipelineTaskGenerator","."]
code: def __call__(self) -> List[task_lib.Task]:
      result = []
      for node in [
          node_proto_view.get_view(n) for n in self._pipeline.nodes
      ]:
          node_uid = task_lib.NodeUid.from_node(self._pipeline, node)
          node_id = node.node_info.id
  
          with self._pipeline_state:
              node_state = self._pipeline_state.get_node_state(node_uid)
              if node_state.state in (
                  pstate.NodeState.STOPPING,
                  pstate.NodeState.STOPPED,
                  pstate.NodeState.PAUSING,
                  pstate.NodeState.PAUSED,
                  pstate.NodeState.FAILED,
              ):
                  logging.info(
                      "Ignoring node in state '%s' for task generation: %s",
                      node_state.state,
                      node_uid,
                  )
                  continue
  
          # If this is a pure service node, there is no ExecNodeTask to generate
          # but we ensure node services and check service status.
          service_status = self._ensure_node_services_if_pure(
              node_id, node_state.backfill_token
          )
          if service_status is not None:
              if (
                  node_state.backfill_token
                  and service_status.code
                  == service_jobs.ServiceStatusCode.SUCCESS
              ):
                  # Transitions ExampleGen node to STOPPED state and service job to
                  # STATE_STOPPED when backfill completes.
                  logging.info(
                      "Stopping ExampleGen: %s ; Backfill with token: %s completed",
                      node_id,
                      node_state.backfill_token,
                  )
                  result.append(
                      task_lib.UpdateNodeStateTask(
                          node_uid=node_uid,
                          state=pstate.NodeState.STOPPED,
                          backfill_token="",
                      )
                  )
                  # The service job already completes with success but we still need to
                  # update the in-memory state.
                  self._service_job_manager.stop_node_services(
                      self._pipeline_state, node_id
                  )
              elif (
                  service_status.code
                  != service_jobs.ServiceStatusCode.RUNNING
              ):
                  error_msg = (
                      f"associated service job failed; node uid: {node_uid}; error"
                      f" message: {service_status.msg}"
                  )
                  result.append(
                      task_lib.UpdateNodeStateTask(
                          node_uid=node_uid,
                          state=pstate.NodeState.FAILED,
                          status=status_lib.Status(
                              code=status_lib.Code.UNKNOWN,
                              message=error_msg,
                          ),
                          backfill_token="",
                      )
                  )
              elif node_state.state != pstate.NodeState.RUNNING:
                  result.append(
                      task_lib.UpdateNodeStateTask(
                          node_uid=node_uid,
                          state=pstate.NodeState.RUNNING,
                          backfill_token=node_state.backfill_token,
                      )
                  )
              continue
  
          # For mixed service nodes, we ensure node services and check service
          # status; the node is aborted if its service jobs have failed.
          service_status = self._ensure_node_services_if_mixed(
              node.node_info.id
          )
          if service_status is not None:
              if (
                  service_status.code
                  != service_jobs.ServiceStatusCode.RUNNING
              ):
                  error_msg = (
                      f"associated service job failed; node uid: {node_uid}; error"
                      f" message: {service_status.msg}"
                  )
                  result.append(
                      task_lib.UpdateNodeStateTask(
                          node_uid=node_uid,
                          state=pstate.NodeState.FAILED,
                          status=status_lib.Status(
                              code=status_lib.Code.UNKNOWN,
                              message=error_msg,
                          ),
                      )
                  )
                  continue
  
          # If a task for the node is already tracked by the task queue, it need
          # not be considered for generation again.
          if self._is_task_id_tracked_fn(
              task_lib.exec_node_task_id_from_node(self._pipeline, node)
          ):
              continue
  
          result.extend(
              self._generate_tasks_for_node(
                  self._mlmd_handle, node, node_state.backfill_token
              )
          )
      return result
  
imports: import sys
  import traceback
  from typing import Callable, List, Optional
  from absl import logging
  from tfx.orchestration import metadata
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import event_observer
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 2 for calling the _Generator obj's underlying member method code with example usage: obj.__call__() and returns: result