code: def _unrunnable_nodes(
      node_by_id: collections.OrderedDict[
          str, node_proto_view.NodeProtoView
      ],
      failed_node_ids: Set[str],
  ) -> Set[str]:
      """Returns node_ids of all unrunnable descendant nodes for each member of the given failed_node_ids set."""
  
      unrunnable = set()
      queue = collections.deque()
  
      for failed_node_id in failed_node_ids:
          for node_with_upstream_failure in node_by_id[
              failed_node_id
          ].downstream_nodes:
              # Nodes with a upstream success optional trigger strategy can make
              # progress despite a failed upstream node.
              if (
                  node_by_id[
                      node_with_upstream_failure
                  ].execution_options.strategy
                  not in _UPSTREAM_SUCCESS_OPTIONAL_STRATEGIES
              ):
                  queue.append(node_with_upstream_failure)
  
      while queue:
          q_node_id = queue.popleft()
          node = node_by_id[q_node_id]
          start_node = (
              node.execution_options.resource_lifetime.lifetime_start
          )
          if (
              node.execution_options.strategy
              == pipeline_pb2.NodeExecutionOptions.LIFETIME_END_WHEN_SUBGRAPH_CANNOT_PROGRESS
              and not (
                  start_node in failed_node_ids
                  or start_node in unrunnable
              )
          ):
              logging.info(
                  "%s is an end node that may still be run since its start node %s"
                  " was neither failed nor unrunnable. Not marking the end node nor"
                  " its descendants as unrunnable due to the failures of %s.",
                  q_node_id,
                  start_node,
                  ", ".join(failed_node_ids),
              )
              continue
          if q_node_id not in unrunnable:
              queue.extend(node_by_id[q_node_id].downstream_nodes)
              unrunnable.add(q_node_id)
  
      # Lazy nodes whose descendents are all unrunnable are also unrunnable, so we
      # need to add them here.
      # We go over the dictionary in reverse order so that lazy nodes that are
      # downstream of other lazy nodes are checked in (reverse) order.
      for node_id, node in reversed(node_by_id.items()):
          if (
              node.execution_options.strategy
              in _LAZY_TRIGGER_STRATEGIES
              and node.downstream_nodes
              and all(
                  downstream in unrunnable
                  for downstream in node.downstream_nodes
              )
          ):
              unrunnable.add(node_id)
      return unrunnable
  
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 5 for calling a global function with example usage: _unrunnable_nodes(node_by_id, failed_node_ids) and returns: unrunnable