code: def test_fair_queueing(shutdown_only):
      ray.init(
          num_cpus=1,
          _system_config={
              # Having parallel leases is slow in this case
              # because tasks are scheduled FIFO,
              # the more parallism we have,
              # the more workers we need to start to execute f and g tasks
              # before we can execute the first h task.
              "max_pending_lease_requests_per_scheduling_category": 1,
              "worker_cap_enabled": True,
          },
      )
  
      @ray.remote
      def h():
          return 0
  
      @ray.remote
      def g():
          return ray.get(h.remote())
  
      @ray.remote
      def f():
          return ray.get(g.remote())
  
      # This will never finish without fair queueing of {f, g, h}:
      # https://github.com/ray-project/ray/issues/3644
      timeout = 60.0
      ready, _ = ray.wait(
          [f.remote() for _ in range(1000)],
          timeout=timeout,
          num_returns=1000,
      )
      assert len(ready) == 1000, len(ready)
  
imports: import logging
  import subprocess
  import sys
  import time
  from pathlib import Path
  import os
  import pytest
  from unittest import mock
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition
  from ray.autoscaler._private.constants import RAY_PROCESSES
  import psutil
label_desc: Use image node_id 7 for calling a global function with example usage: test_fair_queueing(shutdown_only) and returns: int, ray, ray