class_docstr: Generator implementation class for SyncPipelineTaskGenerator.
class_docstr_tok: ["Generator","implementation","class","for","SyncPipelineTaskGenerator","."]
docstr: Returns `True` if all upstream nodes are either COMPLETE or unrunnable.
code: def _lifetime_end_when_subgraph_cannot_progress(
      self,
      node: node_proto_view.NodeProtoView,
      successful_node_ids: Set[str],
      unrunnable_node_ids: Set[str],
      failed_nodes_dict: Mapping[str, status_lib.Status],
  ) -> bool:
      """Returns `True` if all upstream nodes are either COMPLETE or unrunnable."""
      if not (
          start_node := node.execution_options.resource_lifetime.lifetime_start
      ):
          raise ValueError(
              f"Node {node.node_info.id} has trigger strategy"
              " LIFETIME_END_WHEN_SUBGRAPH_CANNOT_PROGRESS but no lifetime_start."
          )
      # If the start node was not successful we will never trigger the end node.
      if start_node not in successful_node_ids:
          return False
  
      # Otherwise, the end node should run if none of its upstream nodes are
      # runnable.
  
      # All nodes not in this set are runnable.
      complete_or_unrunnable_nodes = (
          successful_node_ids
          | unrunnable_node_ids
          | failed_nodes_dict.keys()
      )
  
      # Any potentially runnable upstream nodes are the upstream nodes that are
      # not complete or unrunnable.
      runnable_upstream_node_ids = (
          set(node.upstream_nodes) - complete_or_unrunnable_nodes
      )
      logging.info(
          "[LIFETIME_END_WHEN_SUBGRAPH_CANNOT_PROGRESS trigger check]"
          " for node %s,"
          " complete_or_unrunnable nodes: %s, runnable upstream nodes: %s",
          node.node_info.id,
          complete_or_unrunnable_nodes,
          runnable_upstream_node_ids,
      )
      # If this set is empty then the end node should run, otherwise it needs to
      # wait.
      return not runnable_upstream_node_ids
  
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 12 for calling the _Generator obj's underlying member method code with example usage: obj._lifetime_end_when_subgraph_cannot_progress(node, successful_node_ids, unrunnable_node_ids, failed_nodes_dict) and returns: unknown, False