code: def test_on_completed_callback_refcount(
      ray_start_regular_shared, raise_in_callback
  ):
      """Check that the _on_completed callback is ref counted properly."""
      signal = SignalActor.remote()
  
      def callback(result):
          if raise_in_callback:
              raise Exception("ruh-roh")
  
      @ray.remote
      def wait():
          ray.get(signal.wait.remote())
  
      ref = wait.remote()
  
      initial_refcount = sys.getrefcount(callback)
      ref._on_completed(callback)
  
      # Python ref count should be incremented to avoid the callback being GC'd while the
      # C++ core worker holds a ref to it.
      assert sys.getrefcount(callback) > initial_refcount
  
      # Trigger the task to finish so the callback should execute.
      ray.get(signal.send.remote())
  
      # Now the refcount should drop back down to the initial count.
      wait_for_condition(
          lambda: sys.getrefcount(callback) == initial_refcount
      )
  
imports: import asyncio
  import os
  import sys
  import threading
  import time
  import pytest
  import ray
  from ray._private.client_mode_hook import client_mode_should_convert
  from ray._private.test_utils import SignalActor, kill_actor_and_wait_for_failure, wait_for_condition, wait_for_pid_to_exit
label_desc: Use image node_id 8 for calling a global function with example usage: test_on_completed_callback_refcount(ray_start_regular_shared, raise_in_callback) without return types