code: def _collect_fuse(
      graph: ChunkGraph,
      node: ChunkType,
      graph_results: Set[ChunkType],
      cached_can_fuse,
  ):
      fuse_graph = ChunkGraph()
      fuse_graph.add_node(node)
      fuse_heads = []
      fuse_tails = []
      tail_reduction_node = None
  
      stack = [node]
      # Do a full search of sub graph even the fuse tails > 1
      while len(stack) != 0:
          node = stack.pop()
          is_head = graph.count_predecessors(node) == 0
          for n in graph.iter_predecessors(node):
              can_fuse = cached_can_fuse(n)
              if can_fuse is False or can_fuse is REDUCTION:
                  is_head = True
              elif not fuse_graph.contains(n):
                  stack.append(n)
                  fuse_graph.add_node(n)
              else:
                  fuse_graph.add_edge(n, node)
          if is_head:
              fuse_heads.append(node)
          # Skip the successors of tail reduction node.
          if node is tail_reduction_node:
              continue
          is_tail = (
              graph.count_successors(node) == 0 or node in graph_results
          )
          for n in graph.iter_successors(node):
              can_fuse = cached_can_fuse(n)
              if can_fuse is False:
                  is_tail = True
              elif can_fuse is REDUCTION:
                  if tail_reduction_node is None:
                      tail_reduction_node = n
                      fuse_tails.append(n)
                      stack.append(n)
                      fuse_graph.add_node(n)
                  elif n is tail_reduction_node:
                      fuse_graph.add_edge(node, n)
                  else:
                      is_tail = True
              elif not fuse_graph.contains(n):
                  stack.append(n)
                  fuse_graph.add_node(n)
              else:
                  fuse_graph.add_edge(node, n)
          if is_tail:
              fuse_tails.append(node)
  
      return _Fuse(fuse_graph, fuse_heads, fuse_tails)
  
file_path: mars/mars/optimization/physical/numexpr.py
imports: import dataclasses
  import functools
  import logging
  from typing import List, Set
  import numpy
  from ...core import ChunkType, ChunkGraph
  from ...tensor import arithmetic
  from ...tensor import reduction
  from ...tensor.fuse import TensorNeFuseChunk
  from ...tensor.fuse.numexpr import NUMEXPR_INSTALLED
  from .core import RuntimeOptimizer, register_optimizer
label_desc: Use image node_id 2 for calling a global function with example usage: _collect_fuse(graph, node, graph_results, cached_can_fuse) and returns: _Fuse