code: def test_nested_tasks(shutdown_only):
      ray.init(num_cpus=1)
  
      @ray.remote
      class Counter:
          def __init__(self):
              self.count = 0
  
          def inc(self):
              self.count += 1
              # Since we relex the cap after a timeout we can have slightly more
              # than 1 task. We should never have 20 though since that takes 2^20
              # * 10ms time.
              assert self.count < 20
  
          def dec(self):
              self.count -= 1
  
      counter = Counter.remote()
  
      @ray.remote(num_cpus=1)
      def g():
          return None
  
      @ray.remote(num_cpus=1)
      def f():
          ray.get(counter.inc.remote())
          res = ray.get(g.remote())
          ray.get(counter.dec.remote())
          return res
  
      ready, _ = ray.wait(
          [f.remote() for _ in range(1000)],
          timeout=60.0,
          num_returns=1000,
      )
      assert len(ready) == 1000, len(ready)
      # Ensure the assertion in `inc` didn't fail.
      ray.get(ready)
  
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 1 for calling a global function with example usage: test_nested_tasks(shutdown_only) and returns: None, res