code: def test_dependency_refcounts(ray_start_regular):
      @ray.remote
      def one_dep(dep, signal=None, fail=False):
          if signal is not None:
              ray.get(signal.wait.remote())
          if fail:
              raise Exception("failed on purpose")
  
      @ray.remote
      def one_dep_large(dep, signal=None):
          if signal is not None:
              ray.get(signal.wait.remote())
          # This will be spilled to plasma.
          return np.zeros(10 * 1024 * 1024, dtype=np.uint8)
  
      # Test that regular plasma dependency refcounts are decremented once the
      # task finishes.
      signal = SignalActor.remote()
      large_dep = ray.put(np.zeros(10 * 1024 * 1024, dtype=np.uint8))
      result = one_dep.remote(large_dep, signal=signal)
      check_refcounts({large_dep: (1, 1), result: (1, 0)})
      ray.get(signal.send.remote())
      # Reference count should be removed once the task finishes.
      check_refcounts({large_dep: (1, 0), result: (1, 0)})
      del large_dep, result
      check_refcounts({})
  
      # Test that inlined dependency refcounts are decremented once they are
      # inlined.
      signal = SignalActor.remote()
      dep = one_dep.remote(None, signal=signal)
      check_refcounts({dep: (1, 0)})
      result = one_dep.remote(dep)
      check_refcounts({dep: (1, 1), result: (1, 0)})
      ray.get(signal.send.remote())
      # Reference count should be removed as soon as the dependency is inlined.
      check_refcounts({dep: (1, 0), result: (1, 0)})
      del dep, result
      check_refcounts({})
  
      # Test that spilled plasma dependency refcounts are decremented once
      # the task finishes.
      signal1, signal2 = SignalActor.remote(), SignalActor.remote()
      dep = one_dep_large.remote(None, signal=signal1)
      check_refcounts({dep: (1, 0)})
      result = one_dep.remote(dep, signal=signal2)
      check_refcounts({dep: (1, 1), result: (1, 0)})
      ray.get(signal1.send.remote())
      ray.get(dep, timeout=10)
      # Reference count should remain because the dependency is in plasma.
      check_refcounts({dep: (1, 1), result: (1, 0)})
      ray.get(signal2.send.remote())
      # Reference count should be removed because the task finished.
      check_refcounts({dep: (1, 0), result: (1, 0)})
      del dep, result
      check_refcounts({})
  
      # Test that regular plasma dependency refcounts are decremented if a task
      # fails.
      signal = SignalActor.remote()
      large_dep = ray.put(np.zeros(10 * 1024 * 1024, dtype=np.uint8))
      result = one_dep.remote(large_dep, signal=signal, fail=True)
      check_refcounts({large_dep: (1, 1), result: (1, 0)})
      ray.get(signal.send.remote())
      # Reference count should be removed once the task finishes.
      check_refcounts({large_dep: (1, 0), result: (1, 0)})
      del large_dep, result
      check_refcounts({})
  
      # Test that spilled plasma dependency refcounts are decremented if a task
      # fails.
      signal1, signal2 = SignalActor.remote(), SignalActor.remote()
      dep = one_dep_large.remote(None, signal=signal1)
      check_refcounts({dep: (1, 0)})
      result = one_dep.remote(dep, signal=signal2, fail=True)
      check_refcounts({dep: (1, 1), result: (1, 0)})
      ray.get(signal1.send.remote())
      ray.get(dep, timeout=10)
      # Reference count should remain because the dependency is in plasma.
      check_refcounts({dep: (1, 1), result: (1, 0)})
      ray.get(signal2.send.remote())
      # Reference count should be removed because the task finished.
      check_refcounts({dep: (1, 0), result: (1, 0)})
      del dep, result
      check_refcounts({})
  
imports: import copy
  import logging
  import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, convert_actor_state, kill_actor_and_wait_for_failure, put_object, wait_for_condition, skip_flaky_core_test_premerge
label_desc: Use image node_id 6 for calling a global function with example usage: test_dependency_refcounts(ray_start_regular) and returns: np