o پi@s8dZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z mZmZmZmZddlZddlZddlZddlmZmZmZmZmZmZmZmZddlmZm Z ddl!m"Z"m#Z#m$Z$ddl%m&Z&e'e(Z)Gdd d Z*Gd d d Z+ej,Gd d d Z-GdddZ.ddZ/dS)z gRPC Request Manager - Orchestrates request lifecycle without tokenization. Mimics TokenizerManager's state management and ZMQ communication patterns. N)AnyAsyncGeneratorDictListOptionalUnion)AbortReqBatchEmbeddingOutputBatchTokenIDOutputGetLoadsReqInputGetLoadsReqOutputHealthCheckOutputTokenizedEmbeddingReqInputTokenizedGenerateReqInput)PortArgs ServerArgs)get_or_create_event_loopget_zmq_socketkill_process_tree)get_exception_tracebackc@sReZdZdZdZddejdefddZefde d e e fd d Z d e fd dZ dS)_GrpcCommunicatorz Communicator for request/response patterns with scheduler. Thread-safe and handles the async request/response cycle with proper timeout handling to prevent hangs if the scheduler becomes unresponsive. g>@senderfan_outcCs&||_||_d|_d|_t|_dSN)_sender_fan_out _result_event_result_valuesasyncioLock_lock)selfrrr#X/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/grpc/grpc_request_manager.py__init__1s z_GrpcCommunicator.__init__timeoutreturnc s|j4IdH:t|_g|_|r|j|z!tj|j|dIdH|jWd|_d|_WdIdHSd|_d|_w1IdHsHwYdS)aW Send request and wait for response(s). Args: obj: Request object to send to scheduler timeout: Maximum time to wait for response (seconds) Returns: List of response objects from scheduler(s) Raises: asyncio.TimeoutError: If no response within timeout N)r&) r!rEventrrr send_pyobjwait_forwait)r"objr&r#r#r$__call__8s  z_GrpcCommunicator.__call__recv_objcCsJ|jdur|jdur!|j|t|j|jkr#|jdSdSdSdS)z Handle received response from scheduler. Called by handle_loop when a matching response type is received. Safe to call even if no request is pending (will be ignored). N)rrappendlenrset)r"r.r#r#r$ handle_recvXs  z_GrpcCommunicator.handle_recvN)r)__name__ __module__ __qualname____doc__DEFAULT_TIMEOUTzmqSocketintr%floatrrr-r2r#r#r#r$r's  rc@s,eZdZdZddZd ddZd ddZdS) GrpcSignalHandlerzTMinimal signal handler for gRPC server - delegates real crash handling to scheduler.cCs ||_dSr) grpc_manager)r"r=r#r#r$r%hs zGrpcSignalHandler.__init__NcCs$td|d|dd|j_dS)z7Handle SIGTERM by gracefully shutting down gRPC server.zSIGTERM received. signum=z frame=z. Shutting down gRPC server...TN)loggerwarningr=gracefully_exitr"signumframer#r#r$sigterm_handlerks z!GrpcSignalHandler.sigterm_handlercCs(tdtdttdddS)z-Handle SIGQUIT from failed scheduler process.zUReceived SIGQUIT from scheduler process. Scheduler failed, shutting down gRPC server.zLNote: Crash dumps are handled by the scheduler process, not the gRPC server.Tinclude_parentN)r>errorinforosgetpidrAr#r#r$running_phase_sigquit_handlerrsz/GrpcSignalHandler.running_phase_sigquit_handlerNN)r3r4r5r6r%rDrKr#r#r#r$r<es  r<c@seZdZUdZeed<eejj ed<e j ed<e ed<e j ed<eeefed<eed<d Zeed <d Zeed <d Zeed <d Zeed<d Zeed<d Zeed<dZe ed<dZe ed<ejedZeeed<ejedZ eeed<ejedZ!eeed<ejedZ"eeed<ejedZ#eeed<ejedZ$eeeed<ejedZ%eeeed<ejedZ&eeeed<ejedZ'eeeed<dZ(eeed<dZ)e ed <dS)! GrpcReqStatez"State tracking for a gRPC request. request_id grpc_context out_queuefinishedeventr, created_time finished_timefirst_token_time last_timerlast_completion_tokensfinished_time_perffirst_token_time_perfFstream_finishedinput_logprobs_sent)default_factory output_idsinput_token_logprobs_valinput_token_logprobs_idxoutput_token_logprobs_valoutput_token_logprobs_idxinput_top_logprobs_valinput_top_logprobs_idxoutput_top_logprobs_valoutput_top_logprobs_idxN session_idis_session_request)*r3r4r5r6str__annotations__rgrpcaioServicerContextrQueueboolr(rrrr;rUrVrWrXr:rYrZr[r\ dataclassesfieldlistr^rr_r`rarbrcrdrerfrgrhr#r#r#r$rM~s8           rMc @seZdZdZ d6dedefddZ  d7dedee d ee j j d e eeeefdffd d Z  d7dedee d ee j j fd dZde fddZ d6dedee d ejfddZde d efddZddZdededefddZdefddZde fddZ!d e"fd!d"Z#d#e$fd$d%Z%d&d'Z&d(d)Z'd*d+Z(d ee e)ffd,d-Z* d6d.ee d/eed ee+fd0d1Z,d2d3Z-d4d5Z.dS)8GrpcRequestManagerzx Manages gRPC request lifecycle, mimicking TokenizerManager's orchestration behaviors without tokenization. N server_args port_argscCs||_||_tjd|_t|jtj|jdd|_ t|jtj |j dd|_ i|_ t|_d|_d|_d|_d|_t|_t|_g|_d|_||_t|j |jd|_td|jd|j |jrotd |j dSdS) z$Initialize the gRPC request manager.T)bindFN)rz2GrpcRequestManager initialized with ZMQ IPC: recv=z, send=z6Bootstrap server initialized for disaggregation mode: )!rtrur8rContextcontextrPULLdetokenizer_ipc_namerecv_from_schedulerPUSHscheduler_input_ipc_namesend_to_scheduler rid_to_stater1 asyncio_tasksr@no_create_loop event_loopis_pause Condition is_pause_condtimelast_receive_tstampcrash_dump_request_listcrash_dump_performedbootstrap_serverrdp_sizeget_loads_communicatorr>rHdisaggregation_mode)r"rtrurr#r#r$r%sJ  zGrpcRequestManager.__init__r,rNrOr'c Cst|jdd}|dkr ||||2z 3dHW}|Vq6dStd|d|dur6dtj}n|}td|t|}t|j|_d|j_ d|j_ |||d |2z3dHW}q^6td |td |d g} g} t |D](} t|} t|j| _d| j_ |d | } | | | || | |qt|dd}|std|dg}| D]}|2z 3dHW}| |q6q|VdStd|dddt | D}i}| D] }t|}|||<q|rPtj|tjdIdH\}}|D]?}||}z-|IdH}t|tr0|dd}||vr0|||d<|Vt|}|||<Wq tyKYq w|sdSdS)a Submit a generation request to the scheduler with n>1 parallel sampling support. This method implements the same two-phase approach as tokenizer_manager.py: 1. Phase 1: Send prefix caching request (max_new_tokens=0) 2. Phase 2: Send n generation requests that reuse the cached prefix Yields individual responses for streaming, or aggregated responses for non-streaming. nrNzMultiple sampling request (n=z), using two-phase approachgrpc-z$Phase 1: Caching prefix for request rz-prefixz%Phase 1 completed: Prefix cached for zPhase 2: Generating z parallel requests-streamFzNon-streaming mode: collecting z responseszStreaming mode: multiplexing z streamscSsi|]\}}||qSr#r#).0iridr#r#r$ @sz7GrpcRequestManager.generate_request..) return_whenrNindex)getattrsampling_params_handle_single_requestr>debuguuiduuid4hexcopymax_new_tokensrranger/ enumerater create_task __anext__r+keysFIRST_COMPLETEDpop isinstancedictgetStopAsyncIteration)r"r,rNrOrresponsebase_request_id prefix_obj_ generators request_idsrgen_objgen_request_id is_stream responses generator rid_to_indextask_maptaskdone response_rid next_taskr#r#r$generate_requests                  z#GrpcRequestManager.generate_requestc Cs>|dur dtj}||_t||tdt|td}t |dr1|j r1|j j |_ d|_ ||j |<||z]||IdHt|dd} z*|jIdH}|rY|Vt|trt|ddrt|sr|}|j|d <|VWnWntjytd |d ||IdHwqKW||dS||w) z@Handle a single request - core implementation without n>1 logic.NrFrNrOrPrQrRr,rSsession_paramsTrrQ token_idszRequest z cancelled by client)rrrrrMrrnr(rhasattrrrgrhrrecord_request_for_crash_dump_send_to_schedulerrrPrrrrr^CancelledErrorr>rH abort_request_cleanup_request_state)r"r,rNrOstaterrfinal_responser#r#r$rcsR       z)GrpcRequestManager._handle_single_requestcCs||jvr |j|=dSdS)z9Clean up local request state (does not notify scheduler).N)r)r"rNr#r#r$rs  z)GrpcRequestManager._cleanup_request_statec sdur dtj|_tdtdt|tdj <t z  |IdHWnt yQ}zj = |WYd}~Sd}~wwfdd}t|S)z Submit an embedding request to the scheduler. Returns a future that will contain the embedding result. Nz grpc-embed-Frc sz=zjIdHjIdH}|Wnty0}z |WYd}~nd}~wwWjvr=j=dSdSjvrHj=wr)rRr+rPr set_result Exception set_exceptionr)resultefuturerNr"rr#r$wait_for_results   z=GrpcRequestManager.embedding_request..wait_for_result)rrrrrMrrnr(rrFuturerrrr)r"r,rNrrr#rr$embedding_requests4    z$GrpcRequestManager.embedding_requestc s|drdS|j|}|rd|_d|_td|dt|d}z||IdHtd|WdSt yR}zt d |WYd}~dSd}~ww) zAbort a running request. Sends abort request to scheduler and marks local state as finished to stop processing any further outputs from the scheduler. HEALTH_CHECKFTzMarked request z as aborted locally)rNz$Sent abort to scheduler for request z+Failed to send abort request to scheduler: ) startswithrrrQr[r>rrrrrG)r"rNr abort_reqrr#r#r$rs$   z GrpcRequestManager.abort_requestc s|jsz|jIdH}t|_|jr@|j4IdH|jr+|jIdH|js WdIdHn 1IdHs;wYt|t rN| |IdHn@t|t r\| |IdHn2t|t rj||IdHn$t|trx||IdHnt|tr|j|n tdt|Wnjtjjy|jrYdSYqtjjy}z%|jrtd|WYd}~dStd|dtWYd}~dSd}~wty}ztd|dt|jrWYd}~dSWYd}~nd}~ww|jrdSdS)zt Main event loop - processes outputs from scheduler. Mimics TokenizerManager's handle_loop. NzUnknown output type: z&ZMQ recv interrupted during shutdown: zZMQ error in handle loop:  zHandle loop error: )r@r| recv_pyobjrrrrr+rr _handle_batch_outputr _handle_embedding_outputr _handle_health_check_outputr_handle_abort_reqr rr2r>r?typer8rGAgainZMQErrorrrr)r"r.rr#r#r$ handle_loopsX (      zGrpcRequestManager.handle_loopr batch_out batch_indexcCs|jdurdSt|jdkr |j|j||j|j||j|j||j|j||jjdkret|jdkrQ|j|j||j |j ||j |j ||j |j |dSdS)z Convert and accumulate logprobs from batch output to state. Follows the same logic as tokenizer_manager.convert_logprob_style. Nr) r_r0extendr`rarbr,top_logprobs_numrcrdrerf)r"rrrr#r#r$_convert_logprob_style0s<  z)GrpcRequestManager._convert_logprob_stylec sg}g}t}t}tjD]"\}|jvrqj|}|jr.td|q|jdkr9||_||_ ||_ |j rEj ngj duj rTj ndjr]jndjrfjndj rqj nddd}|jjr||jjr|jjdkr|jr|jjr|js|j|j|j|jd|d<d |_n|jjs|d r|j|j|j|jd|d<|jjrjrtjkr|jjrfd d } j| d | d| dd|d<n|d r|j|j|j|jd|d<|dr|j |d| |j!"||d r5d |_||_#||_$d |_%|j&'fdd} | t()| |q|rFt(j*|dd iIdHdSdS)z.Handle batch generation output from scheduler.z$Skipping output for aborted request rTNr) prompt_tokenscompletion_tokens cached_tokens finish_reason)rNrrQ meta_info)token_logprobs_valtoken_logprobs_idxtop_logprobs_valtop_logprobs_idxinput_logprobsTrQcs(t|d}|rt|kr|SgSr)rr0) attr_name source_list)rrr#r$get_parts  z9GrpcRequestManager._handle_batch_output..get_partrbrerfoutput_logprobsrcs,tdIdH|jvrj|=dSdS)Ng@)rsleepr)rNr"r#r$cleanups   z8GrpcRequestManager._handle_batch_output..cleanupreturn_exceptions)+r perf_counterrridsrrQr>rrVrZrWr^finished_reasonsrrrr,return_logprobrlogprob_start_lenr_rr\r`rcrdrar0rbrerfrr/rPputrUrYr[rRr1rrgather) r"r put_tasks cleanup_tasksnownow_perf_counterrr output_datarrr#)rrr"r$rbs                z'GrpcRequestManager._handle_batch_outputcst|jD]D\}}||jvrq|j|}||j||jr"|j|nd|jr+|j|ndd}|j|IdHd|_t |_ t |_ |j qdS)z-Handle batch embedding output from scheduler.rN)rN embeddingrrT)rrr embeddingsrrrPrrQrrUrrYrRr1)r"rrrrrr#r#r$rs$      z+GrpcRequestManager._handle_embedding_output health_outcs|j}||jvrtd|dS|j|}|dt|dr"|jndt|dr+|jndd}|j|IdHd|_ t |_ t |_ |jdS) z*Handle health check output from scheduler.z)Health check output for unknown request: NT output_strrrstop)rNhealthy output_textr)rrr>r?rrrrPrrQrrUrrYrRr1)r"rrrrr#r#r$r s$    z.GrpcRequestManager._handle_health_check_outputr.cs|jdr dS|j|jvrtd|jddS|j|j}d|_d|_|jr=|j|jddd|j|jdd }n|jdd|jd d d d d dd }|j |IdH|j td|jdS)aHandle abort request from scheduler. The scheduler sends AbortReq back to notify us that a request was aborted, either due to explicit abort_request() call or scheduler-initiated abort (priority preemption, queue full, KV cache pressure, etc). rNzAbort request for zB not in local state (may have already finished or not started yet)TmessagezRequest aborted)idr)rNrGrQrabortzAbort before prefill)rr r)r rrrzHandled abort request for ) rrrr>rrQr[finished_reasonrrPrrRr1)r"r.rabort_responser#r#r$r,sB       z$GrpcRequestManager._handle_abort_reqc s@z |j|WdSty}z td|d}~ww)z(Send an object to the scheduler via ZMQ.zFailed to send to scheduler: N)rr)rr>rG)r"r,rr#r#r$risz%GrpcRequestManager._send_to_schedulercCs<t|jdkr|jtt|ddt|jddSdS)z(Record request for potential crash dump.drunknown)rrNrN)r0rr/rrrr3)r"r,r#r#r$rqs z0GrpcRequestManager.record_request_for_crash_dumpc shtdd|_t|jD] }|s|q|jr*tjt|jddiIdHt|j D]\}}|j sL|j dddIdHd|_ |jq1|jr^tjt|jddiIdH|jrtdzt|jdrt|jjr}|jIdHn|jWnty}ztd |WYd}~nd}~ww|j|j|jtd dS) z(Gracefully shutdown the request manager.z Shutting down GrpcRequestManagerTrNzServer shutting down)rGshutdownzShutting down bootstrap serverrz&Error shutting down bootstrap server: z$GrpcRequestManager shutdown complete)r>rHr@rrrrcancelrrritemsrQrPrrRr1rriscoroutinefunctionrrr?r|closerryterm)r"rrrrr#r#r$r|sF         zGrpcRequestManager.shutdowncCst|j|j|jdS)z)Get server information for health checks.)active_requestspausedlast_receive_time)r0rrrrr#r#r$get_server_infosz"GrpcRequestManager.get_server_infoincludedp_rankcs<t|d}||IdH}durfdd|D}|S)a Get comprehensive load metrics from the scheduler. This method uses the communicator pattern to send GetLoadsReqInput to the scheduler and wait for GetLoadsReqOutput responses. Args: include: List of metric sections to include (core, memory, spec, lora, disagg, queues, all) dp_rank: Optional DP rank filter (None for all ranks) Returns: List of GetLoadsReqOutput objects, one per scheduler/DP rank )rrNcsg|] }|jkr|qSr#r)rrrr#r$ sz0GrpcRequestManager.get_loads..)r r)r"rrreqresultsr#rr$ get_loadss  zGrpcRequestManager.get_loadscCs|jrdSd|_t}|j|t|j||_t t ur6t |}| t j|j| t j|j|j|t|jdS)zWAutomatically create and start the handle_loop task, matching TokenizerManager pattern.NT)rrraddrprint_exception_wrapperrr threadingcurrent_thread main_threadr<add_signal_handlersignalSIGTERMrDSIGQUITrKsigterm_watchdog)r"loopsignal_handlerr#r#r$auto_create_handle_loops"z*GrpcRequestManager.auto_create_handle_loopcs&|jstdIdH|jrdSdS)zIWatchdog to handle SIGTERM gracefully, matching TokenizerManager pattern.g?N)r@rrrr#r#r$r.sz#GrpcRequestManager.sigterm_watchdogrrL)/r3r4r5r6rrr%rrrirkrlrmrrrrrrrrrrrrorrrMr r:rrr rr rrrrrrrrr r$r1r.r#r#r#r$rss  @  x  @ 83 2 != 0   rscszz |IdHWdSty<t}td|t|dr,t|jtr,|jt t ddt dYdSw)zt Sometimes an asyncio function does not print exception. We do another wrapper to handle the exception. Nz%GrpcRequestManager hit an exception: __self__TrEr)rrr>rGrrr2rsdump_requests_before_crashrrIrJsysexit)func tracebackr#r#r$r&s  r&)0r6rrrploggingrIr+r4r'rrtypingrrrrrrrkr8 zmq.asynciosglang.srt.managers.io_structrr r r r r rrsglang.srt.server_argsrrsglang.srt.utilsrrr sglang.utilsr getLoggerr3r>rr< dataclassrMrsr&r#r#r#r$s@ (   >, I