o ۷i{@sdZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z mZeeZdZdZGd d d Zd ed efd dZdS)aG Tensor Parallelism worker for distributed inference. This module implements a simple broadcast-based mechanism for TP serving: - Rank 0 receives HTTP requests and broadcasts them to all ranks - All ranks execute inference synchronously - Rank 0 collects and returns the result Inspired by SGLang's distributed architecture. N)current_platform)GenerateRequestGenerateResponse ModelManageri,c@s\eZdZdZdededefddZeddZd d Z d d Z d dZ de de fddZdS) TPCoordinatorzv Coordinator for Tensor Parallelism inference. Runs on rank 0 and broadcasts requests to all TP workers. model_managerrank world_sizecCsT||_||_||_t|_t|_d|_d|_ t d|d|| dS)NFz TPCoordinator initialized: rank=z , world_size=) r r r time_last_broadcast_time threadingLock_heartbeat_lock_stop_heartbeat_heartbeat_threadloggerinfo_start_heartbeat)selfr r r rO/home/ubuntu/vllm_env/lib/python3.10/site-packages/cache_dit/serve/tp_worker.py__init__$s   zTPCoordinator.__init__cCs|jjS)z=Expose the underlying model_manager's pipe for compatibility.)r piperrrrr/szTPCoordinator.pipecCs |jS)z8Get model information from the underlying model manager.)r get_model_inforrrrr4s zTPCoordinator.get_model_infocs<fdd}tj|dd_jtdtddS)Nc sjsbttjJtjtkrNztjtgtj t j d}t j |ddt_tdWntyM}ztd|WYd}~nd}~wwWdn1sXwYjrdSdS)NdtypedevicersrczHeartbeat sent to workerszHeartbeat failed: )rr sleepHEARTBEAT_INTERVALrrtorchtensorHEARTBEAT_SIZElongr device_typedist broadcastrdebug Exceptionerror) size_tensorerrrheartbeat_loop9s(  z6TPCoordinator._start_heartbeat..heartbeat_loopT)targetdaemonz#Heartbeat thread started (interval=zs))rThreadrstartrrr$)rr1rrrr8s  zTPCoordinator._start_heartbeatcCs"d|_|jr|jjdddSdS)NTr)timeout)rrjoinrrrrstopNszTPCoordinator.stoprequestreturnc Cs|jYtt|}t|}tj|gtjtj d}t j |dd||j d|j |j }tj |tjtj d}|d|tj|tjdt j |ddt|_Wdn1s_wY|d|}t|}|j|} | S)z} Generate images using TP. This method broadcasts the request to all ranks and collects the result. rrr!rN)r)rr synchronizepickledumpslenr%r&r(r)r*r+r zerosuint8copy_ frombufferr rcpunumpytobytesloadsr generate) rr9 request_data request_sizer/ padded_sizerequest_tensorbroadcasted_request_databroadcasted_requestresponserrrrGSs,      zTPCoordinator.generateN)__name__ __module__ __qualname____doc__rintrpropertyrrrr8rrrGrrrrr s r r r c Cs td|d ztttjdgtjtjd}tj |dd| }|t kr4t d|dWq |t d t t }tj|tjtjd}tj |dd|d |}t|}t d|d ||}t d|d Wntytd|d Yd Sty} z@dt| vsdt| vrtd|d| tWYd } ~ d Std|dt| jd| t dWYd } ~ n-d } ~ wt!y} ztd|dt| jd| t dWYd } ~ nd } ~ wwq )zb Worker loop for TP ranks > 0. Receives requests from rank 0 and executes inference. z TP worker z! started, waiting for requests...Trrr!zRank z received heartbeatrNz executing inference...z inference completedz shutting down...NCCLr6z NCCL error: z runtime error: z: g?z error: )"rrrr;r%r&r(r)r*r+itemr'r,get_world_sizer?r@rCrDrEr<rFrGKeyboardInterrupt RuntimeErrorstrlowerr.destroy_process_group exceptiontyperOr r#r-) r r r/rIrJrKrHr9_r0rrr run_tp_worker{sN    ""r`)rRloggingr<r rr%torch.distributed distributedr* platformsrr rrr getLoggerrOrr$r'r rSr`rrrrs    ^