code: def test_back_pressure(shutdown_only_with_initialization_check):
      ray.init()
  
      signal_actor = Semaphore.options(max_pending_calls=10).remote(
          value=0
      )
  
      try:
          for i in range(10):
              signal_actor.acquire.remote()
      except ray.exceptions.PendingCallsLimitExceeded:
          assert False
  
      with pytest.raises(ray.exceptions.PendingCallsLimitExceeded):
          signal_actor.acquire.remote()
  
      @ray.remote
      def release(signal_actor):
          ray.get(signal_actor.release.remote())
          return 1
  
      # Release signal actor through common task,
      # because actor tasks will be back pressured
      for i in range(10):
          ray.get(release.remote(signal_actor))
  
      # Check whether we can call remote actor normally after
      # back presssure released.
      try:
          signal_actor.acquire.remote()
      except ray.exceptions.PendingCallsLimitExceeded:
          assert False
  
      ray.shutdown()
  
imports: import os
  import sys
  import time
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import Semaphore, enable_external_redis, client_test_enabled, run_string_as_driver, wait_for_condition, get_gcs_memory_used, run_string_as_driver_nonblocking
  from ray.experimental.internal_kv import _internal_kv_list
  from ray.tests.conftest import call_ray_start
  import subprocess
  import psutil
label_desc: Use image node_id 2 for calling a global function with example usage: test_back_pressure(shutdown_only_with_initialization_check) and returns: int