o ci@s8ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z mZddlmZmZddlmZmZddlmZddlmZddlmZmZmZmZmZmZm Z m!Z!ddl"Z#ddl$m%Z%dd l&m'Z'dd l(m)Z)dd l*m+Z+m,Z,m-Z-m.Z.ddl/Z/dd l/m0Z0dd l1m2Z2ddl3m4Z4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;mZ>m?Z?m@Z@mAZAddlBmCZCddlDmEZEmFZFmGZGmHZHmIZImJZJmKZKmLZLmMZMmNZNmOZOmPZPmQZQmRZRmSZSddlTmUZUmVZVddlWmXZXmYZYmZZZm[Z[m\Z\ddl]m^Z^m_Z_m`Z`maZambZbddlcmdZdmeZeddlfmgZgddlhmiZimjZjmkZkddllmmZmddlnmoZoddlpmqZqddlrmsZsddltmuZumvZvmwZwddlxmyZyezeRZ{e eCemee|ee}ee~fZd e~d!efd"d#ZGd$d%d%Zee~gdfZGd&d'd'e ZGd(d)d)eZGd*d+d+ZeGd,d-d-ZGd.d/d/ZdS)0N)ABCabstractmethod) defaultdictdeque)asynccontextmanagercontextmanager) dataclass) import_module)AnyAsyncGeneratorCallableDict GeneratorOptionalTupleUnion) to_thread)Request) Starlette)ASGIAppReceiveScopeSend) cloudpickle)get_or_create_event_loop) ActorClass ActorHandle)RemoteFunction)metrics) DeploymentID ReplicaIDReplicaQueueLengthInfoRequestMetadataServeComponentTypeStreamingHTTPRequest gRPCRequest)DeploymentConfig)GRPC_CONTEXT_ARG_NAMEHEALTH_CHECK_METHOD/RAY_SERVE_COLLECT_AUTOSCALING_METRICS_ON_HANDLE$RAY_SERVE_METRICS_EXPORT_INTERVAL_MS4RAY_SERVE_REPLICA_AUTOSCALING_METRIC_RECORD_PERIOD_S&RAY_SERVE_REQUEST_PATH_LOG_BUFFER_SIZE RAY_SERVE_RUN_SYNC_IN_THREADPOOL(RAY_SERVE_RUN_SYNC_IN_THREADPOOL_WARNING*RAY_SERVE_RUN_USER_CODE_IN_SEPARATE_THREADRECONFIGURE_METHODREQUEST_LATENCY_BUCKETS_MSREQUEST_ROUTING_STATS_METHODSERVE_CONTROLLER_NAMESERVE_LOGGER_NAMESERVE_NAMESPACE)create_replica_implcreate_replica_metrics_manager)ASGIAppReplicaWrapperASGIArgsASGIReceiveProxy MessageQueueResponse)access_log_msg configure_component_cpu_profilerconfigure_component_logger#configure_component_memory_profilerget_component_logger_file_path)InMemoryMetricsStore MetricsPusher)get_asgi_route_name) Semaphoreget_component_file_nameparse_import_path)DeploymentVersion)AutoscalingConfig)_get_in_flight_requests) Deployment)BackPressureErrorDeploymentUnavailableErrorRayServeException) LoggingConfig import_pathreturncCsjt|\}}tt||}t|tr|j}|St|tr"|jj}|St|t r3t d|d|j }|S)NzThe import path "zm" contains a decorated Serve deployment. The decorator's settings are ignored when deploying via import path.) rGgetattrr isinstancer _functionr__ray_metadata__modified_classrKloggerwarning func_or_class)rP module_name attr_namedeployment_defr]N/home/ubuntu/.local/lib/python3.10/site-packages/ray/serve/_private/replica.py%_load_deployment_def_from_import_pathss     r_c@seZdZdZdZdZdZdedej de e de fd d Z d d Zd dZddZde fddZde e fddZdedefddZdedefddZdefddZdedede fd d!Zdeeeffd"d#Zd'd%d&Zd$S)(ReplicaMetricsManagera"Manages metrics for the replica. A variety of metrics are managed: - Fine-grained metrics are set for every request. - Autoscaling statistics are periodically pushed to the controller. - Queue length metrics are periodically recorded as user-facing gauges. push_metrics_to_controllerrecord_metrics set_replica_request_metric_gauge replica_id event_loopautoscaling_configingresscCs||_t|_t|_||_||_tjt t d|_ d|_ t dk|_t d|_tjddd|_|jtjddd d |_|jrDtt|_tjd d d d |_|jrUtt|_td ttjddtd d|_|jrott|_ tj!ddd|_"|#||jr|$|%dSdS)N) namespacerserve_deployment_replica_startszCThe number of times this replica has been restarted due to failure.) description serve_deployment_request_counterz?The number of queries that have been processed in this replica.)route)rktag_keysserve_deployment_error_counterzZ3d?d@Z4dAdBZ5dCdDZ6dEdFZ7d ee effdGdHZ8dS)I ReplicaBaserdr\ init_args init_kwargsdeployment_configversionrg route_prefixc s|_|_|j_|_|_|_jj_jj r(jj dj_jj _ jj t_t|||jttdd_tfdd_d_t_d_d_d_d_jddt|jjj |d_!d_"d_#dS)N_F) deployment_idrun_sync_methods_in_threadpool run_user_code_in_separate_threadlocal_testing_modecsjSr)max_ongoing_requestsr]rr]r^{sz&ReplicaBase.__init__..servable_object)rdrerfrg)$_versionrsr_deployment_id_deployment_configrw _route_prefixname_component_nameapp_name unique_id _component_id_configure_logger_and_profilerslogging_configr _event_loopUserCallableWrapperr-r/_user_callable_wrapperrE _semaphore_user_callable_initializedrLock_user_callable_initialized_lock_initialization_latency_healthy_shutting_down_user_callable_asgi_app_set_internal_replica_contextr7rf_metrics_manager_port _docs_path) rrdr\rrrrrgrr]rr^rWsL       zReplicaBase.__init__rQcCs|jjSr)rrrr]r]r^rz ReplicaBase.max_ongoing_requestscC |jSr)rrrr]r]r^r z$ReplicaBase.get_num_ongoing_requestscCs|jj|j|j|j|jfSr)rrrrrrr]r]r^ get_metadatas zReplicaBase.get_metadataNrrcCstjjj|j||jddS)N)rdrr)rxservecontextrrsr)rrr]r]r^rs  z)ReplicaBase._set_internal_replica_contextrcCsp|duri}t|trtdi|}ttj|j|j|tdt tj|j|jdt tj|j|jd\|_ |_ dS)N)component_typecomponent_name component_idr buffer_size)rrrr]) rSdictrOr?r#REPLICArrr,r@r> cpu_profilercpu_profiler_log)rrr]r]r^rs* z+ReplicaBase._configure_logger_and_profilersrcCs |j Sr)rlockedrr]r]r^_can_accept_requests zReplicaBase._can_accept_request request_argscCs\|j}|jdur,|d}z t|j|j}Wnty%d}tdYnw|dur,|}|S)zGet the matched route string for ASGI apps to be used in logs & metrics. If this replica does not wrap an ASGI app or there is no matching for the request, returns the existing route from the request metadata. NrzFailed unexpectedly trying to get route name for request. Routes in metric tags and log messages may be inaccurate. Please file a GitHub issue containing this traceback.)rmrrD asgi_scoperrWr)rrr rmreq matched_router]r]r^_maybe_get_http_routes    z!ReplicaBase._maybe_get_http_routec #st}d}ddtffdd }z|VWn9tjy2}z|}|||WYd}~n#d}~wtyP}z|}td|||WYd}~nd}~wwt|d}| ||||durh|ddS)Nscs|dSrr])r status_coder]r^_status_code_callbackrzEReplicaBase._handle_errors_and_metrics.._status_code_callbackzRequest failed.ri) rrrCancelledError_on_request_cancelledrrWr_on_request_failed_record_errors_and_metrics)rr start_timeuser_exceptionrerr]rr^_handle_errors_and_metricss0  z&ReplicaBase._handle_errors_and_metricsrrrc Csz|j}|j}|j}|durd}n t|tjrd}nd}tjt|p!d|p$||p'||dddid|j j |||dud dS) NOK CANCELLEDERRORCALL)methodrmstatusrserve_access_logTextra)rmrr) _http_methodrm call_methodrSrrrWinfor=rr) rrrrr http_method http_router$ status_strr]r]r^rs,   z&ReplicaBase._record_errors_and_metricsrequest_kwargscCs|jr6t|dkrt|dtsJ|d}|j}t|||j}|dd|_| |||_ ||f}||fS|j rat|dkrFt|dt sHJ|d}|j |j}|jf}|jr_t|jini}||fS)NrrrWS)is_http_requestlenrSr$r r:receive_asgi_messagesgetr#r rmis_grpc_requestr%rget_user_method_infor$user_request_prototakes_grpc_context_kwargr' grpc_context)rrr r)requestscopereceive method_infor]r]r^_unpack_proxy_args3s6 zReplicaBase._unpack_proxy_argsc s||||\}}||:||4IdH|j|||IdHWdIdHWdS1IdHsr?r@ is_streamingrAr;) rrr r)limitrBr5r6rCrDr]r]r^handle_request_with_rejections\   *"z)ReplicaBase.handle_request_with_rejectioncstr)NotImplementedErrorrr]r]r^_on_initializedszReplicaBase._on_initializedc sza|j4IdHEt|_|js.|jIdH|_|jr$|jjj|_ | IdHd|_|rD|j |j IdH|j |jIdHWdIdHn 1IdHsTwY|IdHWdStypttdwNT)rr_initialization_start_timerrinitialize_callabler _callable docs_pathrrO set_sync_method_threadpool_limitrcall_reconfigure user_config check_healthr RuntimeError traceback format_excrrr]r]r^ initializes2   ( zReplicaBase.initializecszI|j|jjk}|j|jjk}||_t|j||_|j|j|r*| |j|j |j IdH|r@|j |jIdH|j|j jdWdStyXttdwNr)rWrrrHfrom_deployment_versionrrrrfrrrUrrVr user_callablerrYrZr[)rruser_config_changedlogging_config_changedr]r]r^ reconfigures<     zReplicaBase.reconfigurercCdSrr]rrrr]r]r^rsz!ReplicaBase._on_request_cancelledcCrdrr]rer]r]r^rszReplicaBase._on_request_failedcCrdrr]rr]r]r^r9szReplicaBase._wrap_requestc Cst|j4IdH%z|j|dVW|j|n|j|wWdIdHdS1IdHs3wYdSr)rrrrrr]r]r^r:s .zReplicaBase._start_requestcs`|jj} t|IdH|j}|dkr$td|d|dn tjddd id dSq) zWait for any ongoing requests to finish. Sleep for a grace period before the first time we check the number of ongoing requests to allow the notification to remove this replica to propagate to callers first. TNrzWaiting for an additional z!s to shut down because there are z ongoing requests.z,Graceful shutdown complete; replica exiting.rGFr!)rgraceful_shutdown_wait_loop_srrrrrWr%)rwait_loop_period_srIr]r]r^_drain_ongoing_requestss" z#ReplicaBase._drain_ongoing_requestscsPz |jIdHWn|jrtdntdY|jIdHdS)NzJ__del__ ran before replica finished initializing, and raised an exception.z__del__ raised an exception.)rcall_destructorrrWrrrrr]r]r^r(s zReplicaBase.shutdowncs.d|_|jr|IdH|IdHdSrP)rrrhrrr]r]r^perform_graceful_shutdown8s z%ReplicaBase.perform_graceful_shutdownc sZz|j}|dur|IdHd|_WdSty,}z tdd|_|dd}~ww)NTzReplica health check failed.F)rcall_user_health_checkrrrWrXrfrr]r]r^rXBs    zReplicaBase.check_healthc sPz|j}|dur|IdHWSiWSty'}ztd|dd}~ww)Nz$Replica record routing stats failed.)rcall_user_record_routing_statsrrWrXrlr]r]r^record_routing_statsOs   z ReplicaBase.record_routing_stats)9rrrr r rr r&rHrrrpropertyrrrReplicaMetadatarrrrOrr"rr rr rrStatusCodeCallbackr BaseExceptionrrr8bytesr<r rErMrrOr]rcrrrrrr9rr:rhrrjrXror]r]r]r^rVs D       "  &      /  $    rc@sZeZdZddZdedejfddZdedefdd Z e ded e e d d ffd d Z d S)Replicacs4|j|jjd|jdurt|j|_dSdSr^)rrr`rrrQrr]r]r^rO[s zReplica._on_initializedmetadatarcCsJtjj|j}|D]}|q t|j}|D]}|qdS)z#Recursively cancels child requests.N)rxrr _get_requests_pending_assignmentinternal_request_idvaluescancelrJ)rrvrrequests_pending_assignmenttaskin_flight_requestsreplica_resultr]r]r^res     zReplica._on_request_cancelledrcCs tjjrtjjdSdSr)rxutilpdb$_is_ray_debugger_post_mortem_enabled _post_mortemrer]r]r^rvs zReplica._on_request_failedrQNc csltjjjtjjj|j|j|j|j j |j |j d| | }|VWddS1s/wYdS)zContext manager that wraps user method calls. 1) Sets the request context var with appropriate metadata. 2) Records the access log message (if not disabled). 3) Records per-request metrics via the metrics manager. )rmrJ_internal_request_idrmultiplexed_model_idr3N)rxrr_serve_request_contextr_RequestContextrmrJrxrrrr3r)rrrBr]r]r^r9zs  "zReplica._wrap_request)rrrrOr"rrrrrrrrrr9r]r]r]r^ruZs   ruc@sJeZdZdZdededededededed efd d Z d e fd dZ de fddZ defddZ d0dedeedefddZddZdeeeffddZdefddZded eedeeeeffd!d"Zdedeeeffd#d$Zdedeedffd%d&Zdedeedffd'd(Zd)edefd*d+Zd,d-Z defd.d/Z!dS)1 ReplicaActora?Actor definition for replicas of Ray Serve deployments. This class defines the interface that the controller and deployment handles (i.e., from proxies and other replicas) use to interact with a replica. All interaction with the user-provided callable is done via the `UserCallableWrapper` class. rdserialized_deployment_defserialized_init_argsserialized_init_kwargsdeployment_config_proto_bytesrrgrc sRt|} t|} t| trt| } t|| t|t|| |||d|_dS)N)rdr\rrrrrgr) r&from_proto_bytesrloadsrSrr_r6 _replica_impl) rrdrrrrrrgrrr\r]r]r^rs"    zReplicaActor.__init__handlecCs|jdSr)pongr)rrr]r]r^push_proxy_handleszReplicaActor.push_proxy_handlerQcCr)zFetch the number of ongoing requests at this replica (queue length). This runs on a separate thread (using a Ray concurrency group) so it will not be blocked by user code. )rrrr]r]r^rs z%ReplicaActor.get_num_ongoing_requestscs>tttttjtj t fS)apoke the replica to check whether it's alive. When calling this method on an ActorHandle, it will complete as soon as the actor has started running. We use this mechanism to detect when a replica has been allocated a worker slot. At this time, the replica can transition from PENDING_ALLOCATION to PENDING_INITIALIZATION startup state. Returns: The PID, actor ID, node ID, node IP, and log filepath id of the replica. ) osgetpidrxget_runtime_context get_actor_id get_worker_id get_node_idrget_node_ip_addressget_node_instance_idrArr]r]r^ is_allocateds   zReplicaActor.is_allocatedNr_aftercs|j|IdH|jS)zHandles initializing the replica. Returns: 3-tuple containing 1. DeploymentConfig of the replica 2. DeploymentVersion of the replica 3. Initialization duration in seconds N)rr]r)rrrr]r]r^initialize_and_get_metadatas z(ReplicaActor.initialize_and_get_metadatac|jIdHdSr)rrXrr]r]r^rXzReplicaActor.check_healthcs|jIdHSr)rrorr]r]r^rosz!ReplicaActor.record_routing_statscs|j|IdH|jSr)rrcrr\r]r]r^rcs zReplicaActor.reconfigurepickled_request_metadatar cCs.t|}|js |jrt|df}||fS)Nr)r?rr+r/)rrr rr]r]r^_preprocess_request_argss  z%ReplicaActor._preprocess_request_argscsJ|||\}}|jj|g|Ri|IdH}|jr#|j|f}|S)z$Entrypoint for `stream=False` calls.N)rrr<r/r3SerializeToStringrrr r)rrDr]r]r^r<s zReplicaActor.handle_requestcOsZ|||\}}|jj|g|Ri|2z3dHW}|jr&|j|f}|Vq6dSr=)rrrEr/r3rrr]r]r^rEsz%ReplicaActor.handle_request_streamingcOsr|||\}}|jj|g|Ri|2z3dHW}t|tr(t|Vq|jr2|j| f}|Vq6dS)aKEntrypoint for all requests with strict max_ongoing_requests enforcement. The first response from this generator is always a system message indicating if the request was accepted (the replica has capacity for the request) or rejected (the replica is already at max_ongoing_requests). For non-streaming requests, there will only be one more message, the unary result of the user request handler. For streaming requests, the subsequent messages will be the results of the user request handler (which must be a generator). N) rrrMrSr!r?r@r/r3rrr]r]r^rM%s" z*ReplicaActor.handle_request_with_rejectionproto_request_metadatacsTddlm}||}t|j|j|j|j|jd}|jj |g|Ri|IdHS)Nr)r")rJrxr$rrm) ray.serve.generated.serve_pb2r" FromStringrJrxr$rrmrr<)rrr r)RequestMetadataProtoprotorr]r]r^handle_request_from_javaEs"   z%ReplicaActor.handle_request_from_javacrr)rrjrr]r]r^rj[rz&ReplicaActor.perform_graceful_shutdowncCs|jdur9ddl}|jt|jd}||jj|Wdn1s'wYtd|jd|jSt ddS)zuSaves CPU profiling data, if CPU profiling is enabled. Logs a warning if CPU profiling is disabled. Nrwbz Saved CPU profile data to file ""zAttempted to save CPU profile data, but failed because no CPU profiler was running! Enable CPU profiling by enabling the RAY_SERVE_ENABLE_CPU_PROFILING env var.) rmarshalsnapshot_statsopenrdumpstatsrWr%error)rrrmr]r]r^_save_cpu_profile_data^s  z#ReplicaActor._save_cpu_profile_data)NN)"rrrrr rtrHrrrrrrrrr&rr rqrrXr rorcrr"rr<r rErMrrjrr]r]r]r^rs|             rc@sTeZdZUdZeed<eed<eed<eed<eed<edededdfd d Z d S) UserMethodInfoz4Wrapper for a user method and its relevant metadata.callabler is_asgi_apptakes_any_argsr2crQcCs,t|j}|||j|t|dkt|vdS)Nr)rrrrr2)inspect signature parametersrr,r')clsrrparamsr]r]r^ from_callable~s  zUserMethodInfo.from_callableN) rrrrr __annotations__rr classmethodrr]r]r]r^rts rc@seZdZdZeefZdedede de de de de fd d Z e d ejfd d Zded efddZedefddZded efddZdedefddZdddddddedeeed ee eefd!e d"eed#ee d eee ffd$d%Ze d eefd&d'ZdVd(d)Zed eefd*d+Zdefd,d-Z d ee!j"j#fd.d/Z$d ee!j"j#fd0d1Z%ed2d3Z&ed e eeffd4d5Z'ed6efd7d8Z(ded9ed!e d:e d;e d"eedeed efde*d?e+d@e,dAe-d ef dBdCZ.ed9ed@e,dAe-dDe/d ef dEdFZ0d>e*dGeedHe eefd e1edffdIdJZ2eddKd>e*dGeedHe eefdLeed ee1edff dMdNZ3ed>e*dGeedHe eefd efdOdPZ4dQe5fdRdSZ6edTdUZ7dS)WrzLWraps a user-provided callable that is used to handle requests to a replica.r\rrrrrrc st|st|stdt|d|_|_|_t|_|_ |_ d_ |_ |_ d_i_d_j rXt_fdd}tjd|d_jdSt_dS)NzBdeployment_def must be a function or class. Instead, its type was rFFcstjjdSr)rset_event_loop_user_code_event_loop run_foreverr]rr]r^_run_user_code_event_loops z?UserCallableWrapper.__init__.._run_user_code_event_loopT)daemontarget)r isfunctionisclass TypeErrortype_deployment_def _init_args _init_kwargs _is_functionr_local_testing_mode_destructor_called_run_sync_methods_in_threadpool!_run_user_code_in_separate_thread _warned_about_sync_method_change_cached_user_method_inforSrnew_event_loopr threadingThread_user_code_event_loop_threadrget_running_loop) rr\rrrrrrrr]rr^rs8   zUserCallableWrapper.__init__rQcCrr)rrr]r]r^rerzUserCallableWrapper.event_looprmcs2ts Jdtdtffdd }|S)aDecorator to run a coroutine method on the user code event loop. The method will be modified to be a sync function that returns a `asyncio.Future` if user code is running in a separate event loop. Otherwise, it will return the coroutine directly. z7_run_user_code can only be used on coroutine functions.rQcsB|g|Ri|}|jrt||j}|jr|St|S|Sr)rrrun_coroutine_threadsaferr wrap_future)rargskwargscorofutrmr]r^wrappers z3UserCallableWrapper._run_user_code..wrapper)riscoroutinefunction functoolswrapsr )rmrr]rr^_run_user_codes z"UserCallableWrapper._run_user_coderLcs|t_dSr)rcurrent_default_thread_limiter total_tokens)rrLr]r]r^rUsz4UserCallableWrapper.set_sync_method_threadpool_limit method_namecs|jvr j|Sjrj}n(tj|rtj|}nfdd}tt|tj}td|d|dt j |t jt d}|j|<|S)zGet UserMethodInfo for the provided call method name. This method is cached to avoid repeated expensive calls to `inspect.signature`. cs&|drdSttj|sdSdS)N__FT) startswithrrRrS)attrrr]r^callable_method_filters zHUserCallableWrapper.get_user_method_info..callable_method_filterzTried to call a method 'z*' that does not exist. Available methods: rF)r) rrrShasattrrRlistfilterdirrNrrrSr8)rr user_methodrmethodsr%r]rr^r0s&      z(UserCallableWrapper.get_user_method_inforD asgi_argscsP|\}}}t|tjjr||||IdHdSt||||IdHdS)aHandle the result from user code and send it over the ASGI interface. If the result is already a Response type, it is sent directly. Otherwise, it is converted to a custom Response type that handles serialization for common Python objects. N) to_args_tuplerS starlette responsesr<send)rrDrr5r6rr]r]r^_send_user_result_over_asgi s  z/UserCallableWrapper._send_user_result_over_asgiNF)rrrKgenerator_result_callback'run_sync_methods_in_threadpool_overriderrrrKrrc sd}dur ntdurnt|dur|jn|}tp'to2tp1t } | ra|ratrKd}|sKt dj dfdd} t | IdH} | |fS| rw|j sw|durwd|_ ttjj di} t| r| IdH} | |fS) a7Call the callable with the provided arguments. This is a convenience wrapper that will work for `def`, `async def`, generator, and async generator functions. Returns the result and a boolean indicating if the result was a sync generator that has already been consumed. FNTMethod 'f' returned a generator. You must use `handle.options(stream=True)` to call generators on a deployment.cs,i}r|D]}|q d}|Srr])rDrrrr is_generatorrr]r^ run_callableJs  z;UserCallableWrapper._call_func_or_gen..run_callable)r)tuplerrrrismethodrisasyncgenfunctionisgeneratorfunctionrrrrun_syncrwarningswarnr.format iscoroutine) rrrrrKrrsync_gen_consumedrun_sync_in_threadpoolis_sync_methodrrDr]rr^_call_func_or_gensP     z%UserCallableWrapper._call_func_or_gencCrr)rSrr]r]r^r`lrz!UserCallableWrapper.user_callablecsPjj}dtdtffdd }jD]}|||qjIdHdS)Nrexccs |Sr)handle_exception)rr rr]r^r yrzGUserCallableWrapper._initialize_asgi_callable..handle_exception)rSapprrservice_unavailable_exceptionsadd_exception_handler_run_asgi_lifespan_startup)rrr r r]rr^_initialize_asgi_callableps z-UserCallableWrapper._initialize_asgi_callablecs|jdur tdtjdddid|jr|j|_n%|j|j|_|j|jj|j |j ddIdHt |jt r@| IdHt|jtd|_t|jtd|_tjdddidt |jt rc|jjSdS) zInitialize the user callable. If the callable is an ASGI app wrapper (e.g., using @serve.ingress), returns the ASGI app object, which may be used *read only* by the caller. Nz/initialize_callable should only be called once.zStarted initializing replica.rGFr!)rrrzFinished initializing replica.)rSrYrWr%rr__new__r rrrrSr8rrRr(_user_health_checkr2_user_record_routing_statsrrr]r]r^rRs>     z'UserCallableWrapper.initialize_callablecCs|jdur td|ddS)Nz-`initialize_callable` must be called before `z`.)rSrY)rrr]r]r^_raise_if_not_initializeds  z-UserCallableWrapper._raise_if_not_initializedcC |d|jdur|SdS)Nrk)rr_call_user_health_checkrr]r]r^rks  z*UserCallableWrapper.call_user_health_checkcCr)Nrn)rr_call_user_record_routing_statsrr]r]r^rns  z2UserCallableWrapper.call_user_record_routing_statscs||jIdHdSr)r rrr]r]r^rsz+UserCallableWrapper._call_user_health_checkcs||jIdH\}}|Sr)r r)rrDrr]r]r^rsz3UserCallableWrapper._call_user_record_routing_statsrWcsn|d|dur5|jrtdt|jts$td|jdtd|jt |jt|fdIdHdSdS)NrVz1deployment_def must be a class to use user_configz%user_config specified but deployment z missing z method)r) rr ValueErrorrrSr0rNrr rR)rrWr]r]r^rVs,   z$UserCallableWrapper.call_reconfigureuser_method_infor+rc st|}t|} |rK|r|D]} || q|S| r,|2z 3dHW} || q6|S|r<|js<|||IdH|S|sI|sItd|jd|S|rQJd|sU| r^td|jd|S)aiPostprocess the result of a user method. User methods can be regular unary functions or return a sync or async generator. This method will raise an exception if the result is not of the expected type (e.g., non-generator for streaming requests or generator for unary requests). Generator outputs will be written to the `generator_result_callback`. Note that HTTP requests are an exception: they are *always* streaming requests, but for ASGI apps (like FastAPI), the actual method will be a regular function implementing the ASGI `__call__` protocol. NCalled method 'G' with `handle.options(stream=True)` but it did not return a generator.z4All HTTP requests go through the streaming codepath.rr)r isgenerator isasyncgenrrrr) rrDrrKr+rrr result_is_genresult_is_async_genrr]r]r^_handle_user_method_results<       z.UserCallableWrapper._handle_user_method_resultrrBr5r6c st||j}|jr$tdtffdd }|||||}ndtffdd }t|||||}d} |2z3dHW} |s[| d} d}| dd kr[|t | d | Vq?6dS) Nitemcsj|dSrcall_soon_threadsafe put_nowaitr# result_queuesystem_event_loopr]r^enqueue@sz9UserCallableWrapper.call_http_entrypoint..enqueuecs|dSr)r&r')r)r]r^r+HsFrTrzhttp.response.startr) r;r0r$rrrr _call_http_entrypointrfetch_messages_from_queuer) rrrBr5r6rr+ call_futurefirst_message_peekedmessagesmsgr]r(r^r>1s,  z(UserCallableWrapper.call_http_entrypointrc sd|dtjd|jddddd|jr|||f}n|js%t}n tj |||f}d}z9t | }|j |j|id|d IdH\}}|j||dd||t|||d IdH} |durg|sg|| WSty} z"|js|| } || t|||IdH|dur|s|d} ~ wt jy|dur|st|jd s|w) zCall an HTTP entrypoint. `send` is used to communicate the results of streaming responses. Raises any exception raised by the user code so it can be propagated as a `RayTaskError`. r,%Started executing request to method ''.FTrGr r!N)rrrKr)rKr+rrrset_max_batch_size)rrWr%rrrrrrequestsrrrfetch_until_disconnectr rr"r9donerzrr rrr) rrr5r6rr  receive_taskrDr final_resultrresponser]r]r^r,`sb          z)UserCallableWrapper._call_http_entrypointr r)c s|js||||IdH}|2z 3dHW}|Vq6dSttdtffdd }|j||||d}|2z3dHW}|D]} | VqDq<6dS)zCalls a user method for a streaming call and yields its results. The user method is called in an asyncio `Task` and places its results on a `result_queue`. This method pulls and yields from the `result_queue`. Nr#csj|dSrr$r'r(r]r^_enqueue_thread_safeszEUserCallableWrapper.call_user_generator.._enqueue_thread_safer+)r_call_user_generatorr;rrr r-) rrr r)genrDr<r.r0r1r]r(r^rAs,  z'UserCallableWrapper.call_user_generatorr=r+cs|ddur ntdurnt||jjtp*to5t p4t  }t j dj dddddd ttdfffd d fd d }rp|rp|jrpt|IdHdSrfdd}|IdHdSS)zCall a user generator. The `generator_result_callback` is used to communicate the results of generator methods. Raises any exception raised by the user code so it can be propagated as a `RayTaskError`. r>Nr2r3FTr4r!rQcs|i}t|r|IdH}t|r!|D]}|VqdSt|r5|2z 3dHW}|Vq(6dStdjdNrr)rrrrrrr?rD)rr r)rr]r^_call_generator_asyncs     zGUserCallableWrapper._call_user_generator.._call_generator_asynccs@i}t|r|D]}|qdStdjdr@)rrrrrA)rr+r r)rr]r^_call_generator_sync s   zFUserCallableWrapper._call_user_generator.._call_generator_synccs$2z 3dHW}|q6dSrr])rD)rBr+r]r^gen_coro_wrappers zBUserCallableWrapper._call_user_generator..gen_coro_wrapper)rrrr0r$rrrrrrrWr%rr r rrr)rrr r)r+r rCrDr])rBrr+r r)rr^r>s0      z(UserCallableWrapper._call_user_generatorcs~|dtjd|jddddd||j}|j|j||ddId H\}}t|s4t |r=t d |j d |S) zCall a (unary) user method. Raises any exception raised by the user code so it can be propagated as a `RayTaskError`. r;r2r3FTr4r!)rrrKNrr) rrWr%r$r0r rrrrrr)rrr r)rrDrr]r]r^r;!s$    z$UserCallableWrapper.call_user_methodr cCs.t||jrtjj|jddStjjdddS)NirzInternal Server Errori)rSrrrr<message)rr r]r]r^r Cs z$UserCallableWrapper.handle_exceptionc s|jdur tddS|jrdSd|_z)t|jdr(|j|jjddIdHt|jdrrr,r rAr>r;rr rir]r]r]r^rs 5$   P 7    = / N   &  K  !r)rconcurrent.futuresrIrrloggingrr?rrrZrabcrr collectionsrr contextlibrr dataclassesr importlibr typingr r r r rrrrstarlette.responsesranyiorfastapirstarlette.applicationsrstarlette.typesrrrrrxrray._common.utilsr ray.actorrrray.remote_functionr ray.serverray.serve._private.commonrr r!r"r#r$r%ray.serve._private.configr&ray.serve._private.constantsr'r(r)r*r+r,r-r.r/r0r1r2r3r4r5ray.serve._private.default_implr6r7ray.serve._private.http_utilr8r9r:r;r< ray.serve._private.logging_utilsr=r>r?r@rA ray.serve._private.metrics_utilsrBrC1ray.serve._private.thirdparty.get_asgi_route_namerDray.serve._private.utilsrErFrGray.serve._private.versionrHray.serve.configrIray.serve.contextrJray.serve.deploymentrKray.serve.exceptionsrLrMrNray.serve.schemarO getLoggerrWrrrrqr_r`rrrrurrrr]r]r]r^s  (       $ D        L9b