class_docstr: Abstract class for a node in a Ray task graph.
  
  A node has a type (e.g., FunctionNode), data (e.g., function options and
  body), arguments (Python values, DAGNodes, and DAGNodes nested within Python
  argument values) and options (Ray API .options() used for function, class
  or class method)
docstr: Apply callable on each node in this DAG in a bottom-up tree walk.
  
  Args:
      fn: Callable that will be applied once to each node in the
          DAG. It will be applied recursively bottom-up, so nodes can
          assume the fn has been applied to their args already.
  
  Returns:
      Return type of the fn after application to the tree.
docstr_tok: ["Apply","callable","on","each","node","in","this","DAG","in","a","bottom-up","tree","walk",".","Args",":","fn",":","Callable","that","will","be","applied","once","to","each","node","in","the","DAG",".","It","will","be","applied","recursively","bottom-up",",","so","nodes","can","assume","the","fn","has","been","applied","to","their","args","already",".","Returns",":","Return","type","of","the","fn","after","application","to","the","tree","."]
code: def apply_recursive(self, fn: "Callable[[DAGNode], T]") -> T:
      """Apply callable on each node in this DAG in a bottom-up tree walk.
  
      Args:
          fn: Callable that will be applied once to each node in the
              DAG. It will be applied recursively bottom-up, so nodes can
              assume the fn has been applied to their args already.
  
      Returns:
          Return type of the fn after application to the tree.
      """
  
      if not type(fn).__name__ == "_CachingFn":
  
          class _CachingFn:
              def __init__(self, fn):
                  self.cache = {}
                  self.fn = fn
                  self.fn.cache = self.cache
                  self.input_node_uuid = None
  
              def __call__(self, node: "DAGNode"):
                  if node._stable_uuid not in self.cache:
                      self.cache[node._stable_uuid] = self.fn(node)
                  if type(node).__name__ == "InputNode":
                      if not self.input_node_uuid:
                          self.input_node_uuid = node._stable_uuid
                      elif self.input_node_uuid != node._stable_uuid:
                          raise AssertionError(
                              "Each DAG should only have one unique InputNode."
                          )
                  return self.cache[node._stable_uuid]
  
          fn = _CachingFn(fn)
      else:
          if self._stable_uuid in fn.cache:
              return fn.cache[self._stable_uuid]
  
      return fn(
          self._apply_and_replace_all_child_nodes(
              lambda node: node.apply_recursive(fn)
          )
      )
  
imports: import ray
  from ray.dag.base import DAGNodeBase
  from ray.dag.py_obj_scanner import _PyObjScanner
  from ray.util.annotations import DeveloperAPI
  from typing import Optional, Union, List, Tuple, Dict, Any, TypeVar, Callable
  import uuid
  import asyncio
  from ray.dag.compiled_dag_node import build_compiled_dag_from_ray_dag
label_desc: Use image node_id 13 for calling the DAGNode obj's underlying member method code with example usage: obj.apply_recursive(fn) and returns: fn, fn, self