o پi\@sdZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z m Z mZddlZddlZddlZddlmZddlmZddlmZmZmZmZmZddlmZmZdd lm Z dd l!m"Z"dd l#m$Z$m%Z%m&Z&dd l'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-dd l.m/Z/ddl0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6ddl7m8Z8ddl9m:Z:ddl;me?Z@Gddde ZAGdddZBGdddZCe fde&de%de fddZDdS)HA controller that dispatches requests to multiple data parallel workers.N)Enumauto)CallableListOptional)envs)compute_dp_attention_world_info)ActiveRanksOutput BlockReqInputTokenizedEmbeddingReqInputTokenizedGenerateReqInputWatchLoadUpdateReq)Req RequestStage)run_scheduler_process)start_cpu_monitor_thread)!DP_ATTENTION_HANDSHAKE_PORT_DELTAPortArgs ServerArgs)process_tracing_init trace_get_proc_propagate_context trace_set_proc_propagate_contexttrace_set_thread_infotrace_slice_endtrace_slice_start) numa_utils) bind_portconfigure_ipv6configure_loggerget_zmq_socketkill_itself_when_parent_diedmaybe_reindex_device_id)TorchMemorySaverAdapter)Watchdog)TypeBasedDispatcherget_exception_tracebackc@s:eZdZdZeZeZeZeZe de fddZ dS)LoadBalanceMethodzLoad balance method.methodc Cs<|}z||WSty}ztd||d}~ww)NzInvalid load balance method: )upperKeyError ValueError)clsr(excr.`/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/managers/data_parallel_controller.pyfrom_strNs zLoadBalanceMethod.from_strN) __name__ __module__ __qualname____doc__r ROUND_ROBINFOLLOW_BOOTSTRAP_ROOMTOTAL_REQUESTS TOTAL_TOKENS classmethodstrr0r.r.r.r/r'Fsr'c@s6eZdZdefddZdefddZdefdd Zd S) DPBudgetdp_sizecCs"||_dg||_dg||_dS)Nr)r<total_requests total_tokens)selfr<r.r.r/__init__Xs zDPBudget.__init__ load_updatecCs,|jD]}|j|j|j<|j|j|j<qdS)zUpdate the budget.N)loadsnum_reqsr=dp_rank num_tokensr>)r?rAloadr.r.r/ update_budget]s zDPBudget.update_budgetr(cs^|tjkrjtj}n|tjkr"ttjfddd}ndSj|d7<|S)Ncsj|j|fSN)r>r=)ir?r.r/jsz#DPBudget.dispatch..)key)r'r7r=indexminr8ranger<)r?r( target_rankr.rJr/dispatchcs   zDPBudget.dispatchN) r1r2r3intr@rrGr'rRr.r.r.r/r;Wsr;c @s~eZdZdZdedededdfddZd d Zd d Z d dZ de fddZ de fddZddZddZdededededejf ddZ d8dedeeedeefdd Zd!ed"edeedeefd#d$Zd!ed%edeefd&d'Zdedefd(d)Z d8dedededeedeeef d*d+Zde fd,d-Zde fd.d/Zde fd0d1Zde fd2d3Z de fd4d5Z!d6d7Z"dS)9DataParallelControllerr server_args port_argsrun_scheduler_process_funcreturnNcCs.||_||_t|j|_||_d|_td|j |_ |j dkr,t |j tj |jd|_d|_tj|jtj|jtj|jtj|ji}||j|_t|j |_t|_g|_dg|j |_ dg|j |_!|j"rq|#|||j$|_%n |&||d|_%|'t(j)d|j*dt+j,-d|_.|j/rt0ddSdS)NrrMFTrT) debug_namewatchdog_timeoutsofttest_stuck_timedata_parallel_controller)1rUrVr'r0load_balance_methodrWglobal_balance_idzmqContextr<context node_rankr PULLscheduler_input_ipc_namerecv_from_tokenizerround_robin_counterr5round_robin_schedulerr6follow_bootstrap_room_schedulerr7total_requests_schedulerr8total_tokens_scheduler dispatchingr; dp_budget threadingLockenv_lockscheduler_procsworkersstatusenable_dp_attentionlaunch_dp_attention_schedulerstp_sizecontrol_message_steplaunch_dp_schedulersinit_dispatcherr$createsoft_watchdog_timeoutrSGLANG_TEST_STUCK_DP_CONTROLLERget soft_watchdogenable_metricsr)r?rUrVrWdispatch_lookupr.r.r/r@wsN        zDataParallelController.__init__cCs,t|jD]\}}|j|r||qdSrH) enumeraterrrs send_pyobj)r?objrIworkerr.r.r/send_to_all_workerss   z*DataParallelController.send_to_all_workerscCs&|jdd|jD]}||q dSrH)rrrwr)r?rrr.r.r/send_control_messages z+DataParallelController.send_control_messagecCs|j|dSrH)rmrG)r?rr.r.r/handle_load_update_reqsz-DataParallelController.handle_load_update_reqrankscCs |j|_dSrH)rs)r?rr.r.r/update_active_rankss z*DataParallelController.update_active_ranksreqcCs\|jjrt|j|jttj|jt|j|_| ||jjr,t tj|jdddSdS)NT)thread_finish_flag) rU enable_tracerrid trace_contextrr DC_DISPATCHrrlrr?rr.r.r/dispatching_with_traces  z-DataParallelController.dispatching_with_tracecCsDtt|jft|jft|jft|jft|j fg|_ |j |j dSrH) r%r rr r rrrr r_request_dispatcheradd_fallback_fnrrJr.r.r/rys z&DataParallelController.init_dispatcherc Csd}g}g}g}t|jD]O}t|}|j|_|j|_|t|jt } || t j |j ||||| fd} || ||j |j|j7}|jdkr\t|jtj|jd|j|<q |D]} | q_|D]} | qh|D]} | qqdS)NrtargetargsT)rPr<rinit_newtokenizer_ipc_namedetokenizer_ipc_nameappendr nccl_portrnEventThread#launch_tensor_parallel_group_threadrvpp_size gpu_id_steprcr rbr`PUSHrerrclosestartwait) r?rUrV base_gpu_idthreadssockets ready_eventsrD tmp_port_args ready_eventthreadsockeventr.r.r/rxsD         z+DataParallelController.launch_dp_schedulersrrDrcCs&|||||| tdq )NTi')launch_tensor_parallel_groupsettimesleep)r?rUrVrrDrr.r.r/r s  z:DataParallelController.launch_tensor_parallel_group_thread worker_portscCs|jdurd|jt}n.|jdr(t|j\}}d|dt|t}n|jd\}}d|dt|t}|jdkrK|||j d|S| ||jS)aBroadcast worker ports from node 0 to all other nodes. Node 0 acts as the server, waiting for all other nodes to connect and sending them the pre-allocated worker ports. Other nodes act as clients, connecting to node 0 to receive their copy of the worker ports. Args: server_args: Server arguments containing node configuration. worker_ports: Pre-allocated worker ports to broadcast. Returns: List of worker ports (same on all nodes after broadcast). Nztcp://127.0.0.1:[ztcp://:rrM) dist_init_addrportr startswithrrSsplitrc_broadcast_ports_as_servernnodes_receive_ports_as_client)r?rUrendpointrhostr.r.r/_broadcast_worker_portss    z.DataParallelController._broadcast_worker_portsrexpected_clientscCstd|dtd|t|jtj|d}z9d}||krH|}td||||d7}td|d |d ||ks!td |W| S| w) z+Broadcast worker ports to all client nodes.zBroadcasting worker ports to z client nodeszWorker ports: TrzReceived handshake from node rMzSent worker ports to /z nodeszWorker port broadcast completed) loggerdebugr rbr`REPrecvdecoderr)r?rrr rep_socketconnected_clients client_rankr.r.r/r?s"   z1DataParallelController._broadcast_ports_as_serverrccCstdt|jtj|d}|tjd|tjdz2z | t | | }tdt |d|WW|StjyNtdtdw|w)z*Receive worker ports from the server node.z,Connecting to node 0 to receive worker portsFi' z Received z worker ports from node 0z,Timeout waiting for worker ports from node 0z9Failed to receive worker ports from node 0 within timeout)rrr rbr`REQ setsockoptRCVTIMEOSNDTIMEOsendr:encode recv_pyobjlenrAgainerror RuntimeError)r?rrc req_socketrr.r.r/r[s"    z/DataParallelController._receive_ports_as_clientcCsg}|jdkr1t|jD]$}t|jtj}||d|d|j|<t d|dd|q | ||r8|nd}| ||dd|dS)NrrMzAssigned port z to worker ) rcrPr<r rbr`rrrrrrrr)r?rUrVrrDport_and_socketbroadcasted_portsr.r.r/russ   z5DataParallelController.launch_dp_attention_schedulerscCs|jstd|d|dtj|jd}g}t|j|jd}t|j|jd} t ||j | ||j | d} | } |j | } t | |j | | |j | d} d}d}| D]}| D]}|}|jr~t |j||j |j |j\}}}t|||}|j|_tjdd\}}|j|||| || |j}|jr|j nd}|j ||j}|||j}||j |j}||j |j|j |j|j}|j^t|I}tj|j||||||||||f d }|"t|| |Wdn1swYWdn 1s wYWdn 1swYWdn 1s-wY|j || |q]qYg}t t!|D] }| ||"qG|dd |_#|dd |_$dS) Nz Launch DPz starting at GPU #.)enablerMrF)duplexrmax_total_num_tokensmax_req_input_len)%rtrinfor#rzenable_memory_savermaxrrrPrcrvr r< attn_cp_sizerrrmpPiperr moe_dp_sizeep_sizerpr"ProcessrWconfigure_subprocessrrrqrrrrr)r?rUrVrrDrmemory_saver_adapterscheduler_pipe_readerspp_size_per_nodennodes_per_pp_rank pp_rank_rangennodes_per_tp_grouptp_size_per_node tp_rank_range attn_cp_rank moe_dp_rankpp_ranktp_rankrank_port_args_readerwritergpu_id attn_dp_size attn_tp_size moe_ep_rankprocscheduler_inforIr.r.r/rs            Kz3DataParallelController.launch_tensor_parallel_groupcCs6|jdurtd|j|j|j|dSdS)NzDirect routing to DP rank TF)data_parallel_rankrrrrrrr.r.r/maybe_external_dp_rank_routings z5DataParallelController.maybe_external_dp_rank_routingcCsr||rdS |j|jr-td|j|j|j||jdt|j|_dS|jdt|j|_q)NTzChoose worker rM)rrsrgrrrrrrrr.r.r/rhs    z,DataParallelController.round_robin_schedulercCsx||rdS|jdur!|jjdkr!|j|_|jdt|j|_|jdus*Jd|jt|j}|j||dS)NfakerMzreq.bootstrap_room should not be None. Do not send requests directly to prefill or decode instances; send to the router instead.)rbootstrap_roomrUdisaggregation_transfer_backendrgrrrr)r?rrQr.r.r/ris     z6DataParallelController.follow_bootstrap_room_schedulercC0||rdS|jtj}|j||dSrH)rrmrRr'r7rrrr?r target_workerr.r.r/rj& z/DataParallelController.total_requests_schedulercCrrH)rrmrRr'r8rrrrr.r.r/rk,rz-DataParallelController.total_tokens_schedulercCsD |jz |jtj}Wn tjyYnw||qqrH)r~feedrfrr`NOBLOCKZMQErrorr)r?recv_reqr.r.r/ event_loop2s  z!DataParallelController.event_looprH)#r1r2r3r4rrrr@rrrr rrrryrxrSrnrrrrrr:rrrurrrhrirjrkr r.r.r.r/rTts @  .   #    t rTrUrVrWc Cstdttt}t||jr4t |j dd}|j dkr)d}n|j dkr0d}t |z4t |||}|d|j|jd |jd krO||jD]}|td |jd |jqRWdStyt}td ||tjYdSw)Nz sglang::data_parallel_controllersglangz DP ControllerprefillzPrefill DP ControllerrzDecode DP Controllerready)rsrrrz$Scheduler or DataParallelController z terminated with z)DataParallelController hit an exception: ) setproctitle faulthandlerrr!psutilrparentrrrotlp_traces_endpointdisaggregation_moderrTrrrrcr rqjoinrrpidexitcode Exceptionr& send_signalsignalSIGQUIT) rUrV pipe_writerrWparent_process thread_label controllerr tracebackr.r.r/$run_data_parallel_controller_process=sH        r)Er4rloggingmultiprocessingrrrnrenumrrtypingrrrrr r`sglang.srt.environrsglang.srt.layers.dp_attentionr sglang.srt.managers.io_structr r r r r"sglang.srt.managers.schedule_batchrrsglang.srt.managers.schedulerrsglang.srt.metrics.cpu_monitorrsglang.srt.server_argsrrrsglang.srt.tracing.tracerrrrrrsglang.srt.utilsrsglang.srt.utils.commonrrrr r!r"+sglang.srt.utils.torch_memory_saver_adapterr#sglang.srt.utils.watchdogr$ sglang.utilsr%r& getLoggerr1rr'r;rTrr.r.r.r/sP           P