code: def fuse(
      dsk,
      keys=None,
      dependencies=None,
      ave_width=_default,
      max_width=_default,
      max_height=_default,
      max_depth_new_edges=_default,
      rename_keys=_default,
      fuse_subgraphs=_default,
  ):
      """Fuse tasks that form reductions; more advanced than ``fuse_linear``
  
      This trades parallelism opportunities for faster scheduling by making tasks
      less granular.  It can replace ``fuse_linear`` in optimization passes.
  
      This optimization applies to all reductions--tasks that have at most one
      dependent--so it may be viewed as fusing "multiple input, single output"
      groups of tasks into a single task.  There are many parameters to fine
      tune the behavior, which are described below.  ``ave_width`` is the
      natural parameter with which to compare parallelism to granularity, so
      it should always be specified.  Reasonable values for other parameters
      will be determined using ``ave_width`` if necessary.
  
      Parameters
      ----------
      dsk: dict
          dask graph
      keys: list or set, optional
          Keys that must remain in the returned dask graph
      dependencies: dict, optional
          {key: [list-of-keys]}.  Must be a list to provide count of each key
          This optional input often comes from ``cull``
      ave_width: float (default 1)
          Upper limit for ``width = num_nodes / height``, a good measure of
          parallelizability.
          dask.config key: ``optimization.fuse.ave-width``
      max_width: int (default infinite)
          Don't fuse if total width is greater than this.
          dask.config key: ``optimization.fuse.max-width``
      max_height: int or None (default None)
          Don't fuse more than this many levels. Set to None to dynamically
          adjust to ``1.5 + ave_width * log(ave_width + 1)``.
          dask.config key: ``optimization.fuse.max-height``
      max_depth_new_edges: int or None (default None)
          Don't fuse if new dependencies are added after this many levels.
          Set to None to dynamically adjust to ave_width * 1.5.
          dask.config key: ``optimization.fuse.max-depth-new-edges``
      rename_keys: bool or func, optional (default True)
          Whether to rename the fused keys with ``default_fused_keys_renamer``
          or not.  Renaming fused keys can keep the graph more understandable
          and comprehensive, but it comes at the cost of additional processing.
          If False, then the top-most key will be used.  For advanced usage, a
          function to create the new name is also accepted.
          dask.config key: ``optimization.fuse.rename-keys``
      fuse_subgraphs : bool or None, optional (default None)
          Whether to fuse multiple tasks into ``SubgraphCallable`` objects.
          Set to None to let the default optimizer of individual dask collections decide.
          If no collection-specific default exists, None defaults to False.
          dask.config key: ``optimization.fuse.subgraphs``
  
      Returns
      -------
      dsk
          output graph with keys fused
      dependencies
          dict mapping dependencies after fusion.  Useful side effect to accelerate other
          downstream optimizations.
      """
  
      # Perform low-level fusion unless the user has
      # specified False explicitly.
      if config.get("optimization.fuse.active") is False:
          return dsk, dependencies
  
      if keys is not None and not isinstance(keys, set):
          if not isinstance(keys, list):
              keys = [keys]
          keys = set(flatten(keys))
  
      # Read defaults from dask.yaml and/or user-defined config file
      if ave_width is _default:
          ave_width = config.get("optimization.fuse.ave-width")
          assert ave_width is not _default
      if max_height is _default:
          max_height = config.get("optimization.fuse.max-height")
          assert max_height is not _default
      if max_depth_new_edges is _default:
          max_depth_new_edges = config.get(
              "optimization.fuse.max-depth-new-edges"
          )
          assert max_depth_new_edges is not _default
      if max_depth_new_edges is None:
          max_depth_new_edges = ave_width * 1.5
      if max_width is _default:
          max_width = config.get("optimization.fuse.max-width")
          assert max_width is not _default
      if max_width is None:
          max_width = 1.5 + ave_width * math.log(ave_width + 1)
      if fuse_subgraphs is _default:
          fuse_subgraphs = config.get("optimization.fuse.subgraphs")
          assert fuse_subgraphs is not _default
      if fuse_subgraphs is None:
          fuse_subgraphs = False
  
      if not ave_width or not max_height:
          return dsk, dependencies
  
      if rename_keys is _default:
          rename_keys = config.get("optimization.fuse.rename-keys")
          assert rename_keys is not _default
      if rename_keys is True:
          key_renamer = default_fused_keys_renamer
      elif rename_keys is False:
          key_renamer = None
      elif not callable(rename_keys):
          raise TypeError("rename_keys must be a boolean or callable")
      else:
          key_renamer = rename_keys
      rename_keys = key_renamer is not None
  
      if dependencies is None:
          deps = {
              k: get_dependencies(dsk, k, as_list=True) for k in dsk
          }
      else:
          deps = {
              k: v
              if isinstance(v, list)
              else get_dependencies(dsk, k, as_list=True)
              for k, v in dependencies.items()
          }
  
      rdeps = {}
      for k, vals in deps.items():
          for v in vals:
              if v not in rdeps:
                  rdeps[v] = [k]
              else:
                  rdeps[v].append(k)
          deps[k] = set(vals)
  
      reducible = {k for k, vals in rdeps.items() if len(vals) == 1}
      if keys:
          reducible -= keys
  
      for k, v in dsk.items():
          if type(v) is not tuple and not isinstance(
              v, (numbers.Number, str)
          ):
              reducible.discard(k)
  
      if not reducible and (
          not fuse_subgraphs
          or all(len(set(v)) != 1 for v in rdeps.values())
      ):
          # Quick return if there's nothing to do. Only progress if there's tasks
          # fusible by the main `fuse`, or by `fuse_subgraphs` if enabled.
          return dsk, deps
  
      rv = dsk.copy()
      fused_trees = {}
      # These are the stacks we use to store data as we traverse the graph
      info_stack = []
      children_stack = []
      # For speed
      deps_pop = deps.pop
      reducible_add = reducible.add
      reducible_pop = reducible.pop
      reducible_remove = reducible.remove
      fused_trees_pop = fused_trees.pop
      info_stack_append = info_stack.append
      info_stack_pop = info_stack.pop
      children_stack_append = children_stack.append
      children_stack_extend = children_stack.extend
      children_stack_pop = children_stack.pop
      while reducible:
          parent = reducible_pop()
          reducible_add(parent)
          while parent in reducible:
              # Go to the top
              parent = rdeps[parent][0]
          children_stack_append(parent)
          children_stack_extend(reducible & deps[parent])
          while True:
              child = children_stack[-1]
              if child != parent:
                  children = reducible & deps[child]
                  while children:
                      # Depth-first search
                      children_stack_extend(children)
                      parent = child
                      child = children_stack[-1]
                      children = reducible & deps[child]
                  children_stack_pop()
                  # This is a leaf node in the reduction region
                  # key, task, fused_keys, height, width, number of nodes, fudge, set of edges
                  info_stack_append(
                      (
                          child,
                          rv[child],
                          [child] if rename_keys else None,
                          1,
                          1,
                          1,
                          0,
                          deps[child] - reducible,
                      )
                  )
              else:
                  children_stack_pop()
                  # Calculate metrics and fuse as appropriate
                  deps_parent = deps[parent]
                  edges = deps_parent - reducible
                  children = deps_parent - edges
                  num_children = len(children)
  
                  if num_children == 1:
                      (
                          child_key,
                          child_task,
                          child_keys,
                          height,
                          width,
                          num_nodes,
                          fudge,
                          children_edges,
                      ) = info_stack_pop()
                      num_children_edges = len(children_edges)
  
                      if fudge > num_children_edges - 1 >= 0:
                          fudge = num_children_edges - 1
                      edges |= children_edges
                      no_new_edges = len(edges) == num_children_edges
                      if not no_new_edges:
                          fudge += 1
                      if (
                          (num_nodes + fudge) / height <= ave_width
                          and
                          # Sanity check; don't go too deep if new levels introduce new edge dependencies
                          (no_new_edges or height < max_depth_new_edges)
                      ):
                          # Perform substitutions as we go
                          val = subs(dsk[parent], child_key, child_task)
                          deps_parent.remove(child_key)
                          deps_parent |= deps_pop(child_key)
                          del rv[child_key]
                          reducible_remove(child_key)
                          if rename_keys:
                              child_keys.append(parent)
                              fused_trees[parent] = child_keys
                              fused_trees_pop(child_key, None)
  
                          if children_stack:
                              if no_new_edges:
                                  # Linear fuse
                                  info_stack_append(
                                      (
                                          parent,
                                          val,
                                          child_keys,
                                          height,
                                          width,
                                          num_nodes,
                                          fudge,
                                          edges,
                                      )
                                  )
                              else:
                                  info_stack_append(
                                      (
                                          parent,
                                          val,
                                          child_keys,
                                          height + 1,
                                          width,
                                          num_nodes + 1,
                                          fudge,
                                          edges,
                                      )
                                  )
                          else:
                              rv[parent] = val
                              break
                      else:
                          rv[child_key] = child_task
                          reducible_remove(child_key)
                          if children_stack:
                              # Allow the parent to be fused, but only under strict circumstances.
                              # Ensure that linear chains may still be fused.
                              if fudge > int(ave_width - 1):
                                  fudge = int(ave_width - 1)
                              # This task *implicitly* depends on `edges`
                              info_stack_append(
                                  (
                                      parent,
                                      rv[parent],
                                      [parent] if rename_keys else None,
                                      1,
                                      width,
                                      1,
                                      fudge,
                                      edges,
                                  )
                              )
                          else:
                              break
                  else:
                      child_keys = []
                      height = 1
                      width = 0
                      num_single_nodes = 0
                      num_nodes = 0
                      fudge = 0
                      children_edges = set()
                      max_num_edges = 0
                      children_info = info_stack[-num_children:]
                      del info_stack[-num_children:]
                      for (
                          _,
                          _,
                          _,
                          cur_height,
                          cur_width,
                          cur_num_nodes,
                          cur_fudge,
                          cur_edges,
                      ) in children_info:
                          if cur_height == 1:
                              num_single_nodes += 1
                          elif cur_height > height:
                              height = cur_height
                          width += cur_width
                          num_nodes += cur_num_nodes
                          fudge += cur_fudge
                          if len(cur_edges) > max_num_edges:
                              max_num_edges = len(cur_edges)
                          children_edges |= cur_edges
                      # Fudge factor to account for possible parallelism with the boundaries
                      num_children_edges = len(children_edges)
                      fudge += min(
                          num_children - 1,
                          max(0, num_children_edges - max_num_edges),
                      )
  
                      if fudge > num_children_edges - 1 >= 0:
                          fudge = num_children_edges - 1
                      edges |= children_edges
                      no_new_edges = len(edges) == num_children_edges
                      if not no_new_edges:
                          fudge += 1
                      if (
                          (num_nodes + fudge) / height <= ave_width
                          and num_single_nodes <= ave_width
                          and width <= max_width
                          and height <= max_height
                          and
                          # Sanity check; don't go too deep if new levels introduce new edge dependencies
                          (no_new_edges or height < max_depth_new_edges)
                      ):
                          # Perform substitutions as we go
                          val = dsk[parent]
                          children_deps = set()
                          for child_info in children_info:
                              cur_child = child_info[0]
                              val = subs(val, cur_child, child_info[1])
                              del rv[cur_child]
                              children_deps |= deps_pop(cur_child)
                              reducible_remove(cur_child)
                              if rename_keys:
                                  fused_trees_pop(cur_child, None)
                                  child_keys.extend(child_info[2])
                          deps_parent -= children
                          deps_parent |= children_deps
  
                          if rename_keys:
                              child_keys.append(parent)
                              fused_trees[parent] = child_keys
  
                          if children_stack:
                              info_stack_append(
                                  (
                                      parent,
                                      val,
                                      child_keys,
                                      height + 1,
                                      width,
                                      num_nodes + 1,
                                      fudge,
                                      edges,
                                  )
                              )
                          else:
                              rv[parent] = val
                              break
                      else:
                          for child_info in children_info:
                              rv[child_info[0]] = child_info[1]
                              reducible_remove(child_info[0])
                          if children_stack:
                              # Allow the parent to be fused, but only under strict circumstances.
                              # Ensure that linear chains may still be fused.
                              if width > max_width:
                                  width = max_width
                              if fudge > int(ave_width - 1):
                                  fudge = int(ave_width - 1)
                              # key, task, height, width, number of nodes, fudge, set of edges
                              # This task *implicitly* depends on `edges`
                              info_stack_append(
                                  (
                                      parent,
                                      rv[parent],
                                      [parent] if rename_keys else None,
                                      1,
                                      width,
                                      1,
                                      fudge,
                                      edges,
                                  )
                              )
                          else:
                              break
                  # Traverse upwards
                  parent = rdeps[parent][0]
  
      if fuse_subgraphs:
          _inplace_fuse_subgraphs(
              rv, keys, deps, fused_trees, rename_keys
          )
  
      if key_renamer:
          for root_key, fused_keys in fused_trees.items():
              alias = key_renamer(fused_keys)
              if alias is not None and alias not in rv:
                  rv[alias] = rv[root_key]
                  rv[root_key] = alias
                  deps[alias] = deps[root_key]
                  deps[root_key] = {alias}
  
      return rv, deps
  
imports: from __future__ import annotations
  import math
  import numbers
  import uuid
  from enum import Enum
  from dask import config, core, utils
  from dask.core import flatten, get_dependencies, ishashable, istask, reverse_dict, subs, toposort
label_desc: Use image node_id 10 for calling a global function with example usage: fuse(dsk, keys, dependencies, ave_width, max_width, max_height, max_depth_new_edges, rename_keys, fuse_subgraphs) and returns: rv,  deps, dsk,  dependencies, dsk,  dependencies, dsk,  deps