code: def test_blocking_actor_task(shutdown_only):
      ray.init(
          num_cpus=1,
          num_gpus=1,
          object_store_memory=int(150 * 1024 * 1024),
      )
  
      @ray.remote(num_gpus=1)
      def f():
          return 1
  
      @ray.remote
      class Foo:
          def __init__(self):
              pass
  
          def blocking_method(self):
              ray.get(f.remote())
  
      # Make sure we can execute a blocking actor method even if there is
      # only one CPU.
      actor = Foo.remote()
      ray.get(actor.blocking_method.remote())
  
      @ray.remote(num_cpus=1)
      class CPUFoo:
          def __init__(self):
              pass
  
          def blocking_method(self):
              ray.get(f.remote())
  
      # Make sure that lifetime CPU resources are not released when actors
      # block.
      actor = CPUFoo.remote()
      x_id = actor.blocking_method.remote()
      ready_ids, remaining_ids = ray.wait([x_id], timeout=1.0)
      assert ready_ids == []
      assert remaining_ids == [x_id]
  
      @ray.remote(num_gpus=1)
      class GPUFoo:
          def __init__(self):
              pass
  
          def blocking_method(self):
              ray.get(f.remote())
  
      # Make sure that GPU resources are not released when actors block.
      actor = GPUFoo.remote()
      x_id = actor.blocking_method.remote()
      ready_ids, remaining_ids = ray.wait([x_id], timeout=1.0)
      assert ready_ids == []
      assert remaining_ids == [x_id]
  
imports: import collections
  import os
  import sys
  import time
  import pytest
  import ray
  import ray.cluster_utils
label_desc: Use image node_id 10 for calling a global function with example usage: test_blocking_actor_task(shutdown_only) and returns: int