o cis @sddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z mZmZmZmZmZmZmZmZddlZddlZddlmZddlmZddlmZdd lmZdd l m!Z!dd l"m#Z#m$Z$m%Z%m&Z&m'Z'dd l(m)Z)dd l*m+Z+ddl,m-Z-ddl.m/Z/m0Z0ddl1m2Z2ddl3m4Z4m5Z5m6Z6m7Z7m8Z8ddl9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBmCZCmDZDeEe8ZFe ddGdddZGdeHde%fddZI dOdee d eJdee$fd!d"ZKGd#d$d$ZLd%d&ZMGd'd(d(e'ZNGd)d*d*ZOd+eddfd,d-ZPd.ejdeQfd/d0ZRGd1d2d2ZSd3e de!gfd4d5ZTGd6d7d7ZUd8e#d3eede#fd9d:ZVd;d<d8e#d=e@d>ejWd?eQdejXf d@dAZYdBeZdCe[dDe\de:fdEdFZ]dGe:dHeQdee$fdIdJZ^d=e@de@fdKdLZ_d=e@de@fdMdNZ`dS)PN)deque)deepcopy) dataclass) AnyAsyncGenerator AwaitableCallableListOptionalTupleTypeUnion)FastAPI)jsonable_encoder)version)MutableHeaders) Middleware)ASGIAppMessageReceiveScopeSend)Config) LifespanOn) IS_PYDANTIC_2) RayActorError RayTaskError)RequestMetadata)#RAY_SERVE_HTTP_KEEP_ALIVE_TIMEOUT_S)RAY_SERVE_HTTP_PROXY_CALLBACK_IMPORT_PATH&RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_SSERVE_HTTP_REQUEST_ID_HEADERSERVE_LOGGER_NAME)ResponseStatus)call_function_from_import_pathgenerate_request_idserve_encoders) HTTPOptions)BackPressureErrorDeploymentUnavailableErrorRayServeExceptionT)frozenc@sPeZdZUeed<eed<eed<deeeeffddZde j j fddZ d S) ASGIArgsscopereceivesendreturncCs|j|j|jfSN)r-r.r/selfr4P/home/ubuntu/.local/lib/python3.10/site-packages/ray/serve/_private/http_util.py to_args_tupleAszASGIArgs.to_args_tuplecCstjj|Sr1) starletterequestsRequestr6r2r4r4r5to_starlette_requestDszASGIArgs.to_starlette_requestN) __name__ __module__ __qualname__r__annotations__rrr r6r7r8r9r:r4r4r4r5r,;s r,serialized_bodyr0csdfdd}|S)zAReturns an ASGI receiver that returns the provided buffered body.Fcs,rt}|IdHddddS)NT http.requestF)bodytype more_body)asyncioEventwait) block_foreverreceivedr?r4r5 mock_receiveQs  z0make_buffered_asgi_receive..mock_receiver4)r?rJr4rHr5make_buffered_asgi_receiveJs rKobj status_codecCsd}d}|dur d}d}n&t|tr|}d}nt|tr$|d}d}ntjt|tddd}d }d |d |ggd d |dgS)zSerializes the provided object and converts it to ASGI messages. These ASGI messages can be sent via an ASGI `send` interface to comprise an HTTP response. Ns text/plainzutf-8stext/plain; charset=utf-8)custom_encoder),:) separatorssapplication/jsonhttp.response.starts content-type)rBstatusheaderszhttp.response.body)rBrA) isinstancebytesstrencodejsondumpsrr&)rMrNrA content_typer4r4r5convert_object_to_asgi_messagesas0    r^c@s"eZdZdZdddZddZdS) ResponseaQASGI compliant response class. It is expected to be called in async context and pass along `scope, receive, send` as in ASGI spec. >>> from ray.serve.http_util import Response # doctest: +SKIP >>> scope, receive = ... # doctest: +SKIP >>> await Response({"k": "v"}).send(scope, receive, send) # doctest: +SKIP NrLcCst||d|_dS)zConstruct a HTTP Response based on input type. Args: content: Any JSON serializable object. status_code (int, optional): Default status code is 200. )rMrNN)r^ _messages)r3contentrNr4r4r5__init__s zResponse.__init__cs |jD] }||IdHqdSr1)r`)r3r-r.r/messager4r4r5r/s z Response.sendNrL)r;r<r=__doc__rbr/r4r4r4r5r_s r_csNg}d}|r"|IdH}|ddksJ|d}||d|sd|S)NTrBr@rCrArO)appendjoin)r-r.r/ body_bufferrCrcr4r4r5receive_http_bodys  ric@seZdZdZddZddZdefddZd efd d Z d efd d Z ddZ de efddZ defddZdejdee edffddZdS) MessageQueueaqQueue enables polling for received or sent messages. Implements the ASGI `Send` interface. This class: - Is *NOT* thread safe and should only be accessed from a single asyncio event loop. - Assumes a single consumer of the queue (concurrent calls to `get_messages_nowait` and `wait_for_message` is undefined behavior). cCs"t|_t|_d|_d|_dS)NF)r_message_queuerDrE_new_message_event_closed_errorr2r4r4r5rbs  zMessageQueue.__init__cCsd|_|jdS)zClose the queue, rejecting new messages. Once the queue is closed, existing messages will be returned from `get_messages_nowait` and subsequent calls to `wait_for_message` will always return immediately. TN)rmrlsetr2r4r4r5closeszMessageQueue.closeecC ||_dSr1)rn)r3rqr4r4r5 set_error zMessageQueue.set_errorrccCs|j||jdSr1)rkrfrlror3rcr4r4r5 put_nowaits zMessageQueue.put_nowaitcs|jrtd||dS)zSend a message, putting it on the queue. `RuntimeError` is raised if the queue has been closed using `.close()`. z6New messages cannot be sent after the queue is closed.N)rm RuntimeErrorrvrur4r4r5__call__szMessageQueue.__call__cs |js|jIdHdSdS)a Wait until at least one new message is available. If a message is available, this method will return immediately on each call until `get_messages_nowait` is called. After the queue is closed using `.close()`, this will always return immediately. N)rmrlrFr2r4r4r5wait_for_messages zMessageQueue.wait_for_messager0cCs>g}t|jdkr||jt|jdks |j|S)aReturns all messages that are currently available (non-blocking). At least one message will be present if `wait_for_message` had previously returned and a subsequent call to `wait_for_message` blocks until at least one new message is available. r)lenrkrfpopleftrlclear)r3messagesr4r4r5get_messages_nowaits  z MessageQueue.get_messages_nowaitcs|jr|j|jIdHt|jdkr,|j}t|jdkr*|js*|j|St|jdkr9|jr9|jt|jdkrE|jrGtdSdS)aThis blocks until a message is ready. This method should not be used together with get_messages_nowait. Please use either `get_one_message` or `get_messages_nowait`. Raises: StopAsyncIteration: if the queue is closed and there are no more messages. Exception (self._error): if there are no more messages in the queue and an error has been set. Nr) rnrlrFrzrkr{rmr|StopAsyncIteration)r3msgr4r4r5get_one_messages   zMessageQueue.get_one_messagecall_futNcCsd}zL t|}tj||gtjdIdH\}}|}|r$|V||vr)nq|}|dur5|dW|s>||durL|sN|dSdSdS|sX||dure|sf|www)aRepeatedly consume messages from the queue and yield them. This is used to fetch queue messages in the system event loop in a thread-safe manner. Args: call_fut: The async Future pointing to the task from the user code event loop that is pushing messages onto the queue. Yields: List[Any]: Messages from the queue. NT) return_when) rD create_taskryrFFIRST_COMPLETEDr~ exceptiondonecancel)r3rwait_for_msg_taskr_r}rqr4r4r5fetch_messages_from_queues8   z&MessageQueue.fetch_messages_from_queue)r;r<r=rerbrp BaseExceptionrsrrvrxryr r~rrDFuturerrrr4r4r4r5rjs    rjc @sleZdZdZdededeegeeffddZ de fdd Z e de jfd d Zd d Zde fddZdS)ASGIReceiveProxyzProxies ASGI receive from an actor. The `receive_asgi_messages` callback will be called repeatedly to fetch messages until a disconnect message is received. r-request_metadatareceive_asgi_messagescCs&|d|_d|_||_||_d|_dS)NrB)_type_queue_request_metadata_receive_asgi_messages_disconnect_message)r3r-rrr4r4r5rbIs  zASGIReceiveProxy.__init__r0cCs|jdkr dddSddiS)a7Return the appropriate disconnect message based on the connection type. HTTP ASGI spec: https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event WS ASGI spec: https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event-ws websocketwebsocket.disconnecti)rBcoderBhttp.disconnect)rr2r4r4r5_get_default_disconnect_messageVs z0ASGIReceiveProxy._get_default_disconnect_messagecCs|jdur t|_|jSr1)rrDQueuer2r4r4r5queuehs  zASGIReceiveProxy.queuec s z8||jIdH}t|trt|}n t|tr|n|g}|D]}|j||ddvr8||_ WdSq#Wn/t yQ| }|j|||_ YdSt yi}z |j|WYd}~dSd}~wwq)aFetch messages repeatedly until a disconnect message is received. If a disconnect message is received, this function exits and returns it. If an exception occurs, it will be raised on the next __call__ and no more messages will be received. TNrB>rr) rrrWrXpickleloadslistrrvrKeyErrorr Exception)r3pickled_messagesr}rcrqr4r4r5fetch_until_disconnectos<       z'ASGIReceiveProxy.fetch_until_disconnectcs>|jr|jdur|jS|jIdH}t|tr||S)zReturn the next message once available. This will repeatedly return a disconnect message once it's been received. N)remptyrgetrWrrur4r4r5rxs zASGIReceiveProxy.__call__N)r;r<r=rerrrrrXrbrrpropertyrDrrrrxr4r4r4r5rBs  )rclscsddlm}m}ddlmmdd}fdd|jD}|}|D]O}|j||j}t |} t | j } t| dkrFtd| d} | j||d } | gd d| d d D} | j| d }||j_|j_|j|q%||t |jD].}t|fsqtst|r|jr|jj|j_t|jdd }|d ur|kr|qfdd|jD|jd d <d S)a_Transform the `cls`'s methods and class annotations to FastAPI routes. Modified from https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py Usage: >>> from fastapi import FastAPI >>> app = FastAPI() # doctest: +SKIP >>> class A: # doctest: +SKIP ... @app.route("/{i}") # doctest: +SKIP ... def func(self, i: int) -> str: # doctest: +SKIP ... return self.dep + i # doctest: +SKIP >>> # just running the app won't work, here. >>> make_fastapi_class_based_view(app, A) # doctest: +SKIP >>> # now app can be run properly r) APIRouterDepends)APIRouteAPIWebSocketRoutecSsddlm}|jS)Nrserve)rayrget_replica_contextservable_objectrr4r4r5get_current_servable_instances  zDmake_fastapi_class_based_view..get_current_servable_instancecs,g|]}t|frj|jjvr|qSr4)rWr=endpoint).0route)rrrr4r5 s   z1make_fastapi_class_based_view..zOMethods in FastAPI class-based view must have ``self`` as their first argument.)defaultcSsg|] }|jtjjdqS))kind)replaceinspect Parameter KEYWORD_ONLY)r parameterr4r4r5rsN) parameters _serve_clscsg|]}|vr|qSr4r4)rr)routes_to_remover4r5r s)fastapirrfastapi.routingrrroutesremoverr signaturerrvaluesrzr*r __signature__rrfinclude_routerrWrresponse_modelresponse_field outer_type_secure_cloned_response_fieldgetattr) fastapi_apprrrrclass_method_routes new_routerr old_endpoint old_signatureold_parametersold_self_parameternew_self_parameternew_parameters new_signature serve_clsr4)rrrrr5make_fastapi_class_based_viewsZ         "rsockc Csz%|tjtjdttdr|tjtjdWdS|tjddWdStyA}ztd|dWYd}~dSd}~ww) zMutate a socket object to allow multiple process listening on the same port. Returns: success: whether the setting was successful. r SO_REUSEPORTTz'Setting SO_REUSEPORT failed because of z. SO_REUSEPORT is disabled.NF) setsockoptsocket SOL_SOCKET SO_REUSEADDRhasattrrrloggerdebug)rrqr4r4r5set_socket_reuse_port s   rc @s|eZdZdZdeeeffddZedefddZ ede e fdd Z d d Z d ed edede efddZddZdS)ASGIAppReplicaWrapperz;Provides a common wrapper for replicas running an ASGI app. app_or_funccCs@t|r ||_n||_tt|jddddd|_t|j_dS)NonF)lifespan log_level log_config access_log)r isfunction _asgi_apprr_serve_asgi_lifespanr)r3rr4r4r5rb-s   zASGIAppReplicaWrapper.__init__r0cCs|jSr1rr2r4r4r5appDszASGIAppReplicaWrapper.appcCst|jtr |jjSdSr1)rWrrdocs_urlr2r4r4r5 docs_pathHs zASGIAppReplicaWrapper.docs_pathcsfddlm}||jjtjd|jIdH|jjr!tdWddS1s,wYdS)NrLoggingContextlevelz=ASGI lifespan startup failed. Check replica logs for details.) ray.serve._private.logging_utilsrrrloggingWARNINGstartup should_exitrwr3rr4r4r5_run_asgi_lifespan_startupMs "z0ASGIAppReplicaWrapper._run_asgi_lifespan_startupr-r.r/cs||||IdHdS)z Calls into the wrapped ASGI app.Nr)r3r-r.r/r4r4r5rxYs zASGIAppReplicaWrapper.__call__csVddlm}||jjtjd|jIdHWddS1s$wYdS)Nrrr)rrrrrrshutdownrr4r4r5__del__hs  "zASGIAppReplicaWrapper.__del__N)r;r<r=rer rrrbrrr rYrrrrrrxrr4r4r4r5r*s$  r middlewarescCsN|durg}t|tstd|D]}tt|ts$tdt|dq|S)a4Validate the return value of HTTP proxy callback. Middlewares should be a list of Starlette middlewares. If it is None, we will treat it as an empty list. If it is not a list, we will raise an error. If it is a list, we will check if all the items in the list are Starlette middlewares. Nz@HTTP proxy callback must return a list of Starlette middlewares.zMHTTP proxy callback must return a list of Starlette middlewares, instead got z type item in the list.)rWr ValueError issubclassrBr)r middlewarer4r4r5#validate_http_proxy_callback_returnqs  rc@s0eZdZdefddZdededefddZd S) RequestIdMiddlewarercCrrr1)_app)r3rr4r4r5rbrtzRequestIdMiddleware.__init__r-r.r/cs\t|d}|tdurt|tdtffdd }||||IdHdS)Nr-rccsJ|ddkrt|d}|d|ddkr|d<|IdHdS)NrBrTrz X-Request-IDzwebsocket.accept)rrf)rcrV request_idr/r4r5send_with_request_ids    z:RequestIdMiddleware.__call__..send_with_request_id)rrr!r%rfrr)r3r-r.r/rVr r4r r5rxs   zRequestIdMiddleware.__call__N) r;r<r=rrbrrrrxr4r4r4r5rsrrcCs`ttg|D]&}ttjtdkr|j|fi|j}q|j|g|jRi|j }q|S)zyWrap the ASGI app with the provided middlewares. The built-in RequestIdMiddleware will always be applied first. z0.35.0) rrrparser7 __version__roptionsargskwargs)rrrr4r4r5_apply_middlewaressrF)enable_so_reuseport http_options event_looprcst|jt}|rt|z ||j|jfWnty6}ztd|jd|jd|d}~wwt j t d_ t jt jfddd|j|j|j|j|d d ddd d }d d|_||j|gdS)z{Start an HTTP server to run the ASGI app. Returns a task that blocks until the server exits (e.g., due to error). zFailed to bind to address 'rR'.Nz uvicorn.errorcsSr1r4r4rr4r5z(start_asgi_http_server..ToffF) factoryhostport root_pathtimeout_keep_alivelooprrrr)configcSsdSr1r4r4r4r4r5rr)sockets)rrrrbindrrOSErrorrwrCRITICAL getLoggerruvicornServerrrkeep_alive_timeout_sinstall_signal_handlersrr)rrrrrrqserverr4rr5start_asgi_http_serversB   r+excrequest_timeout_sr cCst|trtddd|d|ddSt|tjr+d|d}t|td d|dSt|ttfrKt|t rCtj d |d d id tdd|j dSt|t t fr_tj d |d d id nt dtddt|dS)NTzRequest z timed out after zs.)ris_errorrczClient for request z" disconnected, cancelling request.izRequest failed: log_to_stderrF)extraz'Request failed due to unexpected error.i)rW TimeoutErrorr#rDCancelledErrorrinfor(r)rwarningrcrrrY)r,r-r rcr4r4r5get_http_response_statuss<      r7rUresponse_startedcCs"|s|jdvr gSt|j|jdS)N)r.r2)rN)rr^rc)rUr8r4r4r5send_http_response_on_exceptions r9cCs@t|}tpddkr t|_|jstr|jpt|_|jpg|_|S)z7Enhanced configuration with component-specific options.r)rrr(r-r rrr4r4r5$configure_http_options_with_defaults"s   r;cCs6t|}trtdtd|jttt|S)Nz1Calling user-provided callback from import path 'r)rrrr5rextendrr$r:r4r4r5configure_http_middlewares6sr=rd)arDrr[rrr collectionsrcopyr dataclassesrtypingrrrrr r r r r r7r&rrfastapi.encodersr packagingrstarlette.datastructuresrstarlette.middlewarerstarlette.typesrrrrruvicorn.configruvicorn.lifespan.onrray._common.pydantic_compatrray.exceptionsrrray.serve._private.commonrray.serve._private.constantsrrr r!r")ray.serve._private.proxy_request_responser#ray.serve._private.utilsr$r%r&ray.serve.configr'ray.serve.exceptionsr(r)r*r%rr,rXrKintr^r_rirjrrboolrrrrrAbstractEventLoopTaskr+rfloatrYr7r9r;r=r4r4r4r5s   ,              ' efG  8 &