class_docstr: Generator implementation class for SyncPipelineTaskGenerator.
class_docstr_tok: ["Generator","implementation","class","for","SyncPipelineTaskGenerator","."]
code: def __call__(self) -> List[task_lib.Task]:
      layers = _topsorted_layers(self._pipeline)
      exec_node_tasks = []
      update_node_state_tasks = []
      successful_node_ids = set()
      failed_nodes_dict: Dict[str, status_lib.Status] = {}
      finalize_pipeline_task = None
      lazily_evaluated_node_ids = set()
  
      # Loop over all nodes before deciding scheduling so we have full knowledge
      # of all the completed/lazy nodes.
      for layer in layers:
          for node in layer:
              node_id = node.node_info.id
              node_uid = task_lib.NodeUid.from_node(
                  self._pipeline, node
              )
              node_state = self._node_state_by_node_uid[node_uid]
              self._node_proto_view_by_node_id[node_id] = node
  
              if (
                  node.execution_options.strategy
                  in _LAZY_TRIGGER_STRATEGIES
              ):
                  lazily_evaluated_node_ids.add(node.node_info.id)
              if node_state.is_success() or (
                  node_state.is_failure()
                  and node.execution_options.node_success_optional
              ):
                  successful_node_ids.add(node_id)
              elif node_state.is_failure():
                  failed_nodes_dict[node_id] = node_state.status
  
      # Collect nodes that cannot be run because they have a failed ancestor.
      unrunnable_node_ids = _unrunnable_nodes(
          self._node_proto_view_by_node_id,
          set(failed_nodes_dict.keys()),
      )
  
      for layer_nodes in layers:
          for node in layer_nodes:
              node_id = node.node_info.id
              if node_id in successful_node_ids:
                  continue
              if node_id in failed_nodes_dict:
                  continue
              if not self._trigger_strategy_satisfied(
                  node,
                  successful_node_ids,
                  failed_nodes_dict,
                  lazily_evaluated_node_ids,
                  unrunnable_node_ids,
              ):
                  continue
              tasks = self._generate_tasks_for_node(node)
              for task in tasks:
                  if isinstance(task, task_lib.UpdateNodeStateTask):
                      if pstate.is_node_state_success(task.state) or (
                          pstate.is_node_state_failure(task.state)
                          and node.execution_options.node_success_optional
                      ):
                          successful_node_ids.add(node_id)
                      elif pstate.is_node_state_failure(task.state):
                          failed_nodes_dict[node_id] = task.status
                          # While the pipeline can still proceed depending on the trigger
                          # strategy of descending nodes, the fail fast option should only
                          # be used together with ALL_UPSTREAM_NODES_SUCCEEDED since it will
                          # fail the pipeline if any node fails.
                          if self._fail_fast:
                              finalize_pipeline_task = self._abort_task(
                                  failed_nodes_dict
                              )
                      update_node_state_tasks.append(task)
                  elif isinstance(task, task_lib.ExecNodeTask):
                      exec_node_tasks.append(task)
  
              # TODO(b/308161293): Remove this and check for updates in later layers
              # as well.
              if finalize_pipeline_task:
                  break
          if finalize_pipeline_task:
              break
  
      # Always update node states if possible.
      result = update_node_state_tasks
      # If finalize_pipeline_task is set here then we should be in fail_fast
      # mode. Will only update node states and finalize pipeline, ignoring other
      # tasks.
      if finalize_pipeline_task:
          result.append(finalize_pipeline_task)
          return result
  
      # Because we can find newly failed nodes from UpdateNodeStateTask
      # recompute all unrunnable nodes so we can fail the pipeline in this
      # loop.
      # Note that because we only ever append to failed_nodes_dict this set
      # is guaranteed to contain at least the unrunnable nodes we originally
      # computed.
      unrunnable_node_ids = _unrunnable_nodes(
          self._node_proto_view_by_node_id,
          set(failed_nodes_dict.keys()),
      )
  
      # Nodes that are still runnable have neither succeeded nor failed, don't
      # have a failed ancestor, or have a triggering strategy that ignores
      # upstream failures.
      runnable_node_ids = self._node_proto_view_by_node_id.keys() - (
          unrunnable_node_ids
          | successful_node_ids
          | failed_nodes_dict.keys()
      )
  
      # If there are no more runnable nodes, then we finalize the pipeline,
      # otherwise run our exec_node tasks,
      if not runnable_node_ids:
          logging.info(
              "No more runnable nodes in pipeline, finalizing. Succesful nodes: %s,"
              " failed nodes: %s, unrunnable nodes: %s.",
              successful_node_ids,
              failed_nodes_dict.keys(),
              unrunnable_node_ids,
          )
          if failed_nodes_dict:
              result.append(self._abort_task(failed_nodes_dict))
          else:
              result.append(
                  task_lib.FinalizePipelineTask(
                      pipeline_uid=self._pipeline_state.pipeline_uid,
                      status=status_lib.Status(code=status_lib.Code.OK),
                  )
              )
      else:
          result.extend(exec_node_tasks)
  
      return result
  
imports: import collections
  import textwrap
  from typing import Callable, Dict, List, Mapping, Optional, Set
  from absl import logging
  from tfx.orchestration import node_proto_view
  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.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from tfx.utils import topsort
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 3 for calling the _Generator obj's underlying member method code with example usage: obj.__call__() and returns: result, result