o ٷi @sddlZddlZddlmZddlmZddlZzddlZddlm Z ddl m Z dZ ddlmZWneyEdZdZ dZ dZ eZYnweeZd d Zd d Zed$ddZddZd%dedBfddZd&dededBdedefddZddZd%ddZdd Zd!efd"d#ZdS)'N)contextmanager)Any)Queue) PlacementGroupSchedulingStrategyT)PlacementGroupFcCs"trtrdSdtjvrdSdS)z;Check if Ray is initialized without hard dependency on Ray.TRAY_RAYLET_PIDF) RAY_AVAILABLErayis_initializedosenvironr r Y/home/ubuntu/.local/lib/python3.10/site-packages/vllm_omni/distributed/ray_utils/utils.pyis_ray_initializeds  rcCsRd}|D]}t|ttfr|D]}||9}qq||9}qtjg|d}||S)z] Calculate total bytes for a tensor allocation, handling nested tuples in size args. )dtype) isinstancetuplelisttorchtensor element_size) size_argsr num_elementssinnerrr r rcalculate_total_bytes)s  rccshd}d}t}||k}t|dd}|r |r |r d}|j}d|_z dVW|r,||_dSdS|r3||_w)a Context manager to temporarily disable pin_memory if running in Ray and the allocation size exceeds the threshold. This is a workaround for Ray workers often having low ulimit -l (locked memory), causing OS call failed errors when allocating large pinned buffers. F pin_memoryTN)rgetattrr)obj size_bytes thresholdshould_disableold_pinin_rayis_large is_pinnedr r r maybe_disable_pin_memory_for_ray9s     r(cCststdddS)Nz(ray is required for worker_backend='ray'cSs tddS)Nr)maxsize)RayQueuer r r r`s z%get_ray_queue_class..)r ImportErrorr r r rget_ray_queue_class]sr-addresscCsHts tddSts"ddtjddii}tj|d|ddSdS)Nz.Ray is not available, skipping initialization.env_vars PYTHONPATHT)r.ignore_reinit_error runtime_env) rloggerwarningr r r r getinit)r.r3r r rinitialize_ray_clustercs r8PACKnumber_of_stagesstrategyreturncCsdtstdtstdt|ddt|D}tjj ||d}t | t d|S)aCreate a placement group for the given number of stages. Args: number_of_stages: The number of stages to create the placement group for. strategy: The strategy to use for the placement group. Returns: The placement group. z,ray is required for creating placement groupzJ[Orchestrator] Ray is not initialized. Initializing with default settings.cSsg|]}dddqS)g?)GPUCPUr ).0_r r r ~sz*create_placement_group..)r;z*[Orchestrator] Ray Placement Group created) rr,r r r4r5r8rangeutilplacement_groupr6readyinfo)r:r.r;bundlespgr r rcreate_placement_groupns  rIc CsZ|r)tr+z tj|WdSty(}ztd|WYd}~dSd}~wwdSdS)Nz"Failed to remove placement group: )rr rCremove_placement_group Exceptionr4r5)rHer r rrJsrJcCs|rt|dSdS)zETry to clean up Ray resources including placement group and shutdown.N)rJ)rHr r r try_close_rays rMc CsX|r(tr*zt|WdSty'}ztd|WYd}~dSd}~wwdSdS)NzFailed to kill actor: )rr killrKr4r5)actorrLr r rkill_ray_actorsrPplacement_group_bundle_indexcOsrtstdtjddGddd}|jt||ddtjddid d d }|j j|g|Ri||S) Nz&ray is required for starting ray actorr)num_gpusc@seZdZddZdS)z+start_ray_actor..OmniStageRayWorkerc_s||i|SNr )selffuncargskwargsr r rrunsz/start_ray_actor..OmniStageRayWorker.runN)__name__ __module__ __qualname__rXr r r rOmniStageRayWorkers r\)rDrQr0r11)r/CUDA_LAUNCH_BLOCKING)scheduling_strategyr3) rr,r remoteoptionsrr r r6rX)worker_entry_fnrDrQrVrWr\ worker_actorr r rstart_ray_actors rd)rrS)Nr9) loggingr contextlibrtypingrrr ray.util.queuerr*ray.util.scheduling_strategiesrrray.util.placement_grouprr, getLoggerrYr4rrr(r-strr8intrIrJrMrPrdr r r rs@        #