code: def test_async_callback(ray_start_regular_shared):
      global_set = set()
  
      ref = ray.put(None)
      ref._on_completed(lambda _: global_set.add("completed-1"))
      wait_for_condition(lambda: "completed-1" in global_set)
  
      signal = SignalActor.remote()
  
      @ray.remote
      def wait():
          ray.get(signal.wait.remote())
  
      ref = wait.remote()
      ref._on_completed(lambda _: global_set.add("completed-2"))
      assert "completed-2" not in global_set
      signal.send.remote()
      wait_for_condition(lambda: "completed-2" in global_set)
  
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 7 for calling a global function with example usage: test_async_callback(ray_start_regular_shared) without return types