o `۷iE@sddlZddlZddlmZmZmZmZddlZddlm Z ddl m Z m Z m Z mZddlmZddlmZddlmZmZmZeeZedd d ed efd d Zedd d eefddZedd d eefddZedd d efddZedd  d*ded ededeeee fd ef ddZ!edd   d+ded edeeee fded eeeee fff ddZ"edd  d,deded ed!ed eeeff d"d#Z#edd Gd$d%d%Z$edd e  d+ded&ed'eeee fded e$f d(d)Z%dS)-N)DictListOptionalTuple)TPUAcceleratorManager)VALID_TPU_TYPESget_chips_per_hostget_num_chips_from_topologyreserve_tpu_slice)client_mode_wrap) PublicAPI)PlacementGroupplacement_groupremove_placement_groupalpha) stabilityaccelerator_typereturncCsb|}|dr|dd}n|dr|dd}n|}|tvr/td|dttd|S)aExtracts the version from the accelerator type. Args: accelerator_type: The full accelerator type string (e.g. "TPU-V6E"). Returns: The version string (e.g. "v6e"). Raises: ValueError: If the accelerator type is invalid. ztpu-tpuvzInvalid accelerator_type: z. Must be one of z/ or start with 'TPU-' followed by a valid type.)lower startswithreplacer ValueErrorlist)raccel_type_lowerversionrB/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/util/tpu.pyget_tpu_version_from_types  r cCst}|dkr d}|S)z Return the name of the TPU pod that the worker is a part of. Returns: The name of the TPU pod. Returns None if not part of a TPU pod. rN)rget_current_node_tpu_name)tpu_namerrrget_current_pod_name7sr#cCtS)z Count the number of workers associated with the TPU pod that the worker belongs to. Returns: The total number of workers in the TPU pod. Returns None if the worker is not part of a TPU pod. )r"get_num_workers_in_current_tpu_podrrrrget_current_pod_worker_countEs r&cCr$)z Return the number of TPU chips on the node. Returns: The total number of chips on the TPU node. Returns 0 if none are found. )r!get_current_node_num_acceleratorsrrrrget_num_tpu_chips_on_nodeQsr(topology num_workersresources_per_workercCsZ|r|sdSzt|||dd\}}|dkrWdStdt||WSty,YdSw)a Calculates the number of slices needed to accommodate the specified number of workers. Args: topology: The TPU topology string. accelerator_type: The accelerator type string. num_workers: The desired number of workers. resources_per_worker: Optional dict of resources per worker. Returns: The number of slices required. Returns 1 if inputs are invalid or incomplete. r)rresources_per_unit num_slicesr)get_tpu_worker_resourcesmaxmathceil Exception)r)rr*r+workers_per_slice_rrrget_tpu_num_slices_for_workers[s  r7r,r.r/c Cst|}t||}t|}||}|r|ni}d|vr!d|d<d|vr)||d<|d} | dkr5td|| dkrFtd|d| d|| dkrWtd | d |d t|| } | |fS) aF Calculates the number of workers and the resources required for each worker to run based on a TPU topology. Args: topology: The TPU topology string. accelerator_type: The accelerator string. resources_per_unit: Optional manual override for resources per unit. If unspecified, the number of TPU chips in a host is assumed. num_slices: The number of TPU slices. Returns: A tuple containing: - num_workers: Total workers required. - unit_resources: The resource dictionary for a single worker. CPUr,TPUrzTPU resources must be positive.z Total chips (z,) not divisible by TPUs requested per unit (z).z$The requested resources per bundle (zE TPU chips) do not divide evenly into the chips available per slice (zi). This configuration results in an uneven distribution of workers across slices, which is not supported.)r rr copyrint) r)rr.r/accelerator_versionchips_per_hosttotal_chips_per_slicetotal_chips_availablefinal_resources tpus_per_unitr*rrrr0s4    r08081coordinator_addressslice_idcoordinator_portcCs||t|t|dS)a Returns the environment variables required for JAX multi-slice coordination. Args: coordinator_address: The IP address or hostname of the coordinator. num_slices: The total number of slices in the cluster. slice_id: The index of the current slice. coordinator_port: The port the coordinator is listening on. Returns: A dictionary mapping environment variable names to their values. )MEGASCALE_COORDINATOR_ADDRESSMEGASCALE_PORTMEGASCALE_NUM_SLICESMEGASCALE_SLICE_ID)str)rCr/rDrErrrget_tpu_coordinator_env_varss rKc@sbeZdZdZ     d,dededeeeefd ed ed eed efd dZ defddZ ddZ   d-d ed ed eede fddZ ede fddZedefddZedefddZedefddZedefddZedefd d!Zedefd"d#Zedee fd$d%Zedeeeeffd&d'Zedeeeffd(d)Zd*d+ZdS).SlicePlacementGroupa. A handle to a placement group reservation for a TPU slice. The following definitions are added for clarity: - Accelerator type: A string describing the accelerator type and version (e.g. TPU-V2, TPU-V6E). - Accelerator version: The accelerator generation only (e.g. v6e, v5p, v5litepod). - Pod type: The TPU accelerator version and the number of chips in a topology. (e.g. v6e-128, v5p-8). - Accelerator topology: The physical topology representing the structure (e.g. 2x2x2, 16x16). Args: topology: The TPU topology string (e.g. "2x2x2"). accelerator_version: The TPU accelerator generation (e.g. "v6e", "v5p", "v4"). resources_per_bundle: Optionally specify the resources to include in every worker bundle. strategy: PlacementGroup parameter. The strategy to create the placement group. Currently default to "SPREAD" - "PACK": Packs Bundles into as few nodes as possible. - "SPREAD": Places Bundles across distinct nodes as even as possible. - "STRICT_PACK": Packs Bundles into one node. The group is not allowed to span multiple nodes. - "STRICT_SPREAD": Packs Bundles across distinct nodes. lifetime: PlacementGroup parameter. Either `None`, which defaults to the placement group will fate share with its creator and will be deleted once its creator is dead, or "detached", which means the placement group will live as a global object independent of the creator. num_slices: Number of TPU slices in the SlicePlacementGroup. Defaults to 1 when unspecified. Examples: .. testcode:: python :skipif: True import ray from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy from ray.util.tpu import SlicePlacementGroup slice_handle = SlicePlacementGroup(topology="4x4", accelerator_version="v6e") slice_pg = slice_handle.placement_group ray.get(slice_pg.ready(), timeout=10) @ray.remote(num_cpus=0, resources={'TPU': 4}) def spmd_task(world, rank): print(f"Current TPU is rank {rank} of {world}") tasks = [ spmd_task.options( scheduling_strategy=PlacementGroupSchedulingStrategy( placement_group=slice_pg, ) ).remote(world=4, rank=i) for i in range(slice_handle.num_hosts) ] NSPREADrr,r)r<resources_per_bundlestrategynamelifetimer/c Cs||_||_|pi|_||_t|j|j||jd\|_|_t |j|j|_ t |j}t d||j } | |j|_ g|_g|_|d|_|||||_dS)Nr-r,)stripr _topology_accelerator_version_resources_per_bundle _num_slicesr0 _num_bundles_bundle_resourcesr_chips_per_hostr r1 _num_hosts _head_pgs_bundle_label_selector_validate_tpu_config_placement_group_reserve_slice) selfr)r<rNrOrPrQr/ total_chipshosts_per_slicerrr__init__s2     zSlicePlacementGroup.__init__cCs |tvrtd|dtdS)NzInvalid accelerator version: z. Must be one of: )rr)r`r<rrr_accelerator_version_checkHs z.SlicePlacementGroup._accelerator_version_checkcCs<||jtj|j|jdstd|jd|jddS)N)tpu_accelerator_version tpu_topologyzInvalid accelerator topology: 'z' for accelerator version: '')rdr<r!is_valid_tpu_accelerator_topologyrSrr`rrrr]Ns  z(SlicePlacementGroup._validate_tpu_configrc sg_g}jj}dj}zStjD]@}tj|}|s2t djdjd|d|\} } j | t j j| i} j| g||fddt|D7}qt||||jd} | WStyqw) z8Performs the two-step scheduling to reserve a TPU slice.zTPU-z'Failed to reserve TPU slice. Requested z slice(s) of topology 'z' with accelerator type 'zE'. Ensure that sufficient TPU resources are available in the cluster.csg|]}jqSr)rXr:).0r6rirr |s z6SlicePlacementGroup._reserve_slice..)bundlesrOrPrQbundle_label_selector)r\rWrVr<upperranger/r rS RuntimeErrorr[appendray_rayletRAY_NODE_TPU_SLICE_NAME_KEYextendrr4shutdown) r`rOrPrQrlbundles_per_slicerr6 reservation slice_namehead_pgselectorpgrrirr_[sD        z"SlicePlacementGroup._reserve_slicecC|jS)z%The underlying PlacementGroup object.)r^rirrrrz#SlicePlacementGroup.placement_groupcCr})z0The number of chips per host for this TPU slice.)rYrirrrr=sz"SlicePlacementGroup.chips_per_hostcCr})z5The total number of hosts in the SlicePlacementGroup.)rZrirrr num_hostsr~zSlicePlacementGroup.num_hostscCr})z7The total number of bundles in the SlicePlacementGroup.)rWrirrr num_bundlesr~zSlicePlacementGroup.num_bundlescCr})z'The physical topology of the TPU slice.)rSrirrrr)r~zSlicePlacementGroup.topologycCr})z&The TPU accelerator type of the slice.)rTrirrrr<r~z'SlicePlacementGroup.accelerator_versioncCr})z8The number of TPU slices this SlicePlacementGroup spans.)rVrirrrr/r~zSlicePlacementGroup.num_slicescCr})z1The internal head PGs used to reserve the slices.)r[rirrrhead_placement_groupsr~z)SlicePlacementGroup.head_placement_groupscCr})z1The bundle label selector list for the worker PG.)r\rirrrrmr~z)SlicePlacementGroup.bundle_label_selectorcCr})z/The resources that are assigned to each bundle.)rXrirrrbundle_resourcesr~z$SlicePlacementGroup.bundle_resourcescCs4|jr t|jd|_|jD]}t|qg|_dS)z=Removes the worker placement group and all internal head PGs.N)r^rr[)r`rzrrrrvs     zSlicePlacementGroup.shutdown)NrMrNr,)rMrN)__name__ __module__ __qualname____doc__rJrrfloatr;rcrdr]r r_propertyrr=rrr)r<r/rrrmrrvrrrrrLsr= - 2 rLr<rNcKstd||||d|S)a9Asynchronously creates a PlacementGroup for a TPU slice. A slice placement group reserves num_slices TPU slice(s) and creates a placement group for scheduling tasks or actors. Args: topology: The desired TPU pod topology (e.g. "4x4", "2x8"). accelerator_version: The TPU accelerator generation, (e.g. "v4", "v5p", "v6e"). resources_per_bundle: Specify the number of resources to reserve per bundle. When unspecified, SlicePlacementGroup defaults to reserving 1 bundle per TPU host in a topology, with the bundle resources set to the number of TPU in a host. Ex: Specifying {"TPU": 1} for a 4x4 topology would result in 16 bundles, each with 1 TPU. If resources_per_bundle=None for the same topology, there would be 4 bundles with 4 TPU each. num_slices: The number of tpu slices within the placement group **kwargs: Additional arguments for the placement group, such as 'name', 'lifetime', or 'strategy'. Returns: The handle for the created SlicePlacementGroup. )r)r<rNr/Nr)rL)r)r<rNr/kwargsrrrslice_placement_groupsr)N)Nr,)rB)&loggingr2typingrrrrrrray._private.acceleratorsrray._private.accelerators.tpurrr r ray._private.client_mode_hookr ray.util.annotationsr ray.util.placement_groupr rr getLoggerrloggerrJr r#r;r&r(rr7r0rKrLrrrrrs        ( @  j