code: def test_schedule_actor_and_normal_task(ray_start_cluster_enabled):
      cluster = ray_start_cluster_enabled
      cluster.add_node(
          memory=1024**3,
          _system_config={"gcs_actor_scheduling_enabled": True},
      )
      ray.init(address=cluster.address)
      cluster.wait_for_nodes()
  
      @ray.remote(memory=600 * 1024**2, num_cpus=0.01)
      class Foo:
          def method(self):
              return 2
  
      @ray.remote(memory=600 * 1024**2, num_cpus=0.01)
      def fun(singal1, signal_actor2):
          signal_actor2.send.remote()
          ray.get(singal1.wait.remote())
          return 1
  
      singal1 = SignalActor.remote()
      signal2 = SignalActor.remote()
  
      o1 = fun.remote(singal1, signal2)
      # Make sure the normal task is executing.
      ray.get(signal2.wait.remote())
  
      # The normal task is blocked now.
      # Try to create actor and make sure this actor is not created for the time
      # being.
      foo = Foo.remote()
      o2 = foo.method.remote()
      ready_list, remaining_list = ray.wait([o2], timeout=2)
      assert len(ready_list) == 0 and len(remaining_list) == 1
  
      # Send a signal to unblock the normal task execution.
      ray.get(singal1.send.remote())
  
      # Check the result of normal task.
      assert ray.get(o1) == 1
  
      # Make sure the actor is created.
      assert ray.get(o2) == 2
  
file_path: ray/python/ray/tests/test_advanced_5.py
imports: import logging
  import sys
  import time
  import numpy
  import pytest
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, client_test_enabled
label_desc: Use image node_id 2 for calling a global function with example usage: test_schedule_actor_and_normal_task(ray_start_cluster_enabled) and returns: int, int