code: def test_shared_output(shared_ray_instance):
      """Verify when an upstream task output is shared by
      multi output, the upstream task runs only once.
      """
  
      @ray.remote
      def shared_f():
          return 1
  
      @ray.remote
      def g(input):
          return input + 1
  
      @ray.remote
      def h(input):
          return input + 2
  
      x = shared_f.bind()
      dag = MultiOutputNode([g.bind(x), h.bind(x)])
  
      ray.get(dag.execute()) == [2, 3]
  
      # Verify f ran only once.
      def verify():
          tasks = list_tasks(filters=[("name", "=", "shared_f")])
          return len(tasks) == 1
  
      wait_for_condition(verify)
  
file_path: ray/python/ray/dag/tests/test_output_node.py
imports: import pytest
  import ray
  from ray.dag.input_node import InputNode
  from ray.dag.output_node import MultiOutputNode
  from ray.util.state import list_tasks
  from ray._private.test_utils import wait_for_condition
label_desc: Use image node_id 5 for calling a global function with example usage: test_shared_output(shared_ray_instance) and returns: int, unknown, unknown, unknown