o ÔÙ¾i{ã@s~ddlZddlmZddlZddlZddlmZddlmZee ƒZ defdd„Z Gdd „d ƒZ Gd d „d ƒZ e ƒZe ƒZdS) éN)ÚAny)Ú ServerArgs)Ú init_loggerÚ server_argsc Ãstj ¡}| tj¡}d|j›}| |¡t d|›¡ z&|  ¡IdH}t   |¡}t d¡t   |¡IdH}| t  |¡¡IdHWn8ty~}z,tjd|›ddz| t  dt|ƒd œ¡¡IdHWn tysYnwWYd}~nd}~wwq ) z• This function runs as a background task in the FastAPI process. It listens for TCP requests from offline clients (e.g., DiffGenerator). ztcp://*:z,ZMQ Broker is listening for offline jobs on TNz-Broker received an offline job from a client.zError in ZMQ Broker: )Úexc_infoÚerror)ÚstatusÚmessage)ÚzmqÚasyncioÚContextÚsocketÚREPÚ broker_portÚbindÚloggerÚinfoÚrecvÚpickleÚloadsÚasync_scheduler_clientÚforwardÚsendÚdumpsÚ ExceptionrÚstr)rÚctxr Úbroker_endpointÚpayloadÚ request_batchÚresponse_batchÚe©r"úb/home/ubuntu/.local/lib/python3.10/site-packages/sglang/multimodal_gen/runtime/scheduler_client.pyÚrun_zeromq_broker s.€      $ ÿ€ûór$c@óNeZdZdZdd„Zdefdd„Zdedefd d „Zde fd d „Z d d„Z dS)ÚSchedulerClientz£ A synchronous, singleton client for communicating with the Scheduler service. Designed for use in DiffGenerator, where synchronous usage is preferred cCsd|_d|_d|_dS©N)ÚcontextÚscheduler_socketr©Úselfr"r"r#Ú__init__5s zSchedulerClient.__init__rcCsŒ|jdur|jjst d¡| ¡||_t ¡|_|j tj ¡|_ |j   tj d¡|j   tj d¡|jj}|j  |¡t d|›¡dS)Nz8SchedulerClient is already initialized. Re-initializing.ré€[z2SchedulerClient connected to backend scheduler at )r(ÚclosedrÚwarningÚcloserr r r ÚREQr)Ú setsockoptÚLINGERÚRCVTIMEOÚscheduler_endpointÚconnectÚdebug)r+rr5r"r"r#Ú initialize:s   ÿzSchedulerClient.initializeÚbatchÚreturncCsBz|j |¡|j ¡}|WStjjy t d¡tdƒ‚w)úESends a batch or request to the scheduler and waits for the response.ú,Timeout waiting for response from scheduler.ú"Scheduler did not respond in time.)r)Ú send_pyobjÚ recv_pyobjr rÚAgainrÚ TimeoutError)r+r9Ú output_batchr"r"r#rOs   þzSchedulerClient.forwardcCs²|jdus |jjrt d¡dS|j tj¡}| tjd¡| tj d¡|j j }z*z|  |¡|  ddi¡| ¡WW| ¡dStjjySYW| ¡dSw| ¡w© zS Checks if the scheduler server is alive using a temporary socket. Nz'Cannot ping: client is not initialized.FriÐÚmethodÚpingT)r(r.rrr r r1r2r3r4rr5r6r>r?r0r@©r+Ú ping_socketÚendpointr"r"r#rEYs$   ý ý zSchedulerClient.pingcCs4|jr |j ¡d|_|jr|j ¡d|_dSdS©z-Closes the socket and terminates the context.N)r)r0r(Útermr*r"r"r#r0qs   þzSchedulerClient.closeN© Ú__name__Ú __module__Ú __qualname__Ú__doc__r,rr8rrÚboolrEr0r"r"r"r#r&/s r&c@r%)ÚAsyncSchedulerClienta? An asynchronous, singleton client for communicating with the Scheduler service. Designed for use in asynchronous environments like FastAPI entrypoints. To support high concurrency, it creates a new REQ socket for each request rather than sharing a single one (which would cause ZMQ state errors). cCsd|_d|_dSr')r(rr*r"r"r#r,„s zAsyncSchedulerClient.__init__rcCsD|jdur|jjst d¡| ¡||_tj ¡|_t  d¡dS)Nz=AsyncSchedulerClient is already initialized. Re-initializing.z9AsyncSchedulerClient initialized with zmq.asyncio.Context) r(r.rr/r0rr r r r7)r+rr"r"r#r8ˆsÿ zAsyncSchedulerClient.initializer9r:cÃs¼|jdur tdƒ‚|j tj¡}| tjd¡| tjd¡|jj }|  |¡z0z|  t   |¡¡IdH| ¡IdH}t  |¡WW| ¡StjjyXt d¡tdƒ‚w| ¡w)r;NzAAsyncSchedulerClient is not initialized. Call initialize() first.rr-r<r=)r(Ú RuntimeErrorr r r1r2r3r4rr5r6rrrrrr0rr@rrA)r+r9r rHrr"r"r#r“s(€ ÿ   ü þ zAsyncSchedulerClient.forwardcÃsÆ|jdus |jjrt d¡dS|j tj¡}| tjd¡| tj d¡|j j }z3z!|  |¡|  t ddi¡¡IdH| ¡IdHWW| ¡dStjjy]YW| ¡dSw| ¡wrC)r(r.rrr r r1r2r3r4rr5r6rrrrr0r@rFr"r"r#rE­s&€   ý ý zAsyncSchedulerClient.pingcCs|jr |j ¡d|_dSdSrI)r(rJr*r"r"r#r0Ås  þzAsyncSchedulerClient.closeNrKr"r"r"r#rQ{s  rQ)rÚtypingrr Ú zmq.asyncioÚ)sglang.multimodal_gen.runtime.server_argsrÚ1sglang.multimodal_gen.runtime.utils.logging_utilsrrLrr$r&rQrÚsync_scheduler_clientr"r"r"r#Ús   "LR