code: def _estimate_avail_cpus(cur_pg: Optional["PlacementGroup"]) -> int:
      """Estimates the available CPU parallelism for this Dataset in the cluster.
  
      If we aren't in a placement group, this is trivially the number of CPUs in the
      cluster. Otherwise, we try to calculate how large the placement group is relative
      to the size of the cluster.
  
      Args:
          cur_pg: The current placement group, if any.
      """
      cluster_cpus = int(ray.cluster_resources().get("CPU", 1))
      cluster_gpus = int(ray.cluster_resources().get("GPU", 0))
  
      # If we're in a placement group, we shouldn't assume the entire cluster's
      # resources are available for us to use. Estimate an upper bound on what's
      # reasonable to assume is available for datasets to use.
      if cur_pg:
          pg_cpus = 0
          for bundle in cur_pg.bundle_specs:
              # Calculate the proportion of the cluster this placement group "takes up".
              # Then scale our cluster_cpus proportionally to avoid over-parallelizing
              # if there are many parallel Tune trials using the cluster.
              cpu_fraction = bundle.get("CPU", 0) / max(1, cluster_cpus)
              gpu_fraction = bundle.get("GPU", 0) / max(1, cluster_gpus)
              max_fraction = max(cpu_fraction, gpu_fraction)
              # Over-parallelize by up to a factor of 2, but no more than that. It's
              # preferrable to over-estimate than under-estimate.
              pg_cpus += 2 * int(max_fraction * cluster_cpus)
  
          return min(cluster_cpus, pg_cpus)
  
      return cluster_cpus
  
imports: import importlib
  import logging
  import os
  import pathlib
  import random
  import sys
  import threading
  import time
  import urllib.parse
  from collections import deque
  from types import ModuleType
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, TypeVar, Union
  import numpy
  import ray
  from ray._private.utils import _get_pyarrow_version
  from ray.data._internal.arrow_ops.transform_pyarrow import unify_schemas
  from ray.data.context import WARN_PREFIX, DataContext
label_desc: Use image node_id 5 for calling a global function with example usage: _estimate_avail_cpus(cur_pg) and returns: cluster_cpus, min