code: def test_limit_concurrency(shutdown_only):
      ray.init(num_cpus=1)
  
      block_task = Semaphore.remote(0)
      block_driver = Semaphore.remote(0)
  
      ray.get(
          [block_task.locked.remote(), block_driver.locked.remote()]
      )
  
      @ray.remote(num_cpus=1)
      def foo():
          ray.get(block_driver.release.remote())
          ray.get(block_task.acquire.remote())
  
      refs = [foo.remote() for _ in range(20)]
  
      block_driver_refs = [
          block_driver.acquire.remote() for _ in range(20)
      ]
  
      # Some of the tasks will run since we relax the cap, but not all because it
      # should take exponentially long for the cap to be increased.
      ready, not_ready = ray.wait(
          block_driver_refs, timeout=10, num_returns=20
      )
      assert len(not_ready) >= 1
  
      # Now the first instance of foo finishes, so the second starts to run.
      ray.get([block_task.release.remote() for _ in range(19)])
  
      ready, not_ready = ray.wait(
          block_driver_refs, timeout=10, num_returns=20
      )
      assert len(not_ready) == 0
  
      ready, not_ready = ray.wait(refs, num_returns=20, timeout=15)
      assert len(ready) == 19
      assert len(not_ready) == 1
  
file_path: ray/python/ray/tests/test_worker_capping.py
imports: import asyncio
  import os
  import pytest
  import sys
  import tempfile
  import time
  import ray
  from ray._private.test_utils import Semaphore
label_desc: Use image node_id 4 for calling a global function with example usage: test_limit_concurrency(shutdown_only) without return types