o ciu@shddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z ddl mZddlmZmZmZmZmZmZmZddlZddlZddlZddlmmmZddlmmmZddlm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&dd l'm(Z(dd l)m*Z*dd l+m,Z,dd l-m.Z.dd l/m0Z0ddl1m2Z2m3Z3m4Z4m5Z5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>m?Z?ddl@mAZAeBeCZDe(ddZEddZFGdddejGZHddZIdeJfddZKdejLdeeeJeffd d!ZMd1d"d#ZNd$d%ZOd2d'd(ZPd1d)d*ZQd+eeJdee&fd,d-ZRd.d/ZSeCd0kr2eSdSdS)3N) defaultdict)futures)AnyCallableDictListOptionalSetUnion) cloudpickle) ray_constants)disable_client_hook) GcsClient) env_integer) setup_logger)%canonicalize_bootstrap_address_or_die)add_port_to_grpc_server JobConfig)CLIENT_SERVER_MAX_THREADS GRPC_OPTIONSOBJECT_TRANSFER_CHUNK_SIZEClientServerHandle ResponseCache) DataServicer)LogstreamServicer) serve_proxier)dumps_from_serverloads_from_client)current_serverTIMEOUT_FOR_SPECIFIC_SERVER_Scstfdd}|S)z Decorator for gRPC stubs. Before calling the real stubs, checks if there's an existing entry in the caches. If there is, then return the cached entry. Otherwise, call the real function and use the real cache c st|d}tfdd|Dr|||Sd}d}td}|j|}|||}|durMt|trK|t j j | t |||Sz|||} Wn"tyw} z|||| |t j j | t | d} ~ ww|||| | S)N) client_id thread_idreq_idc3s|]}|vVqdSN).0imetadatar&Q/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/client/server/server.py <sz7_use_response_cache..wrapper..r"r#r$)dictinvocation_metadataanyintresponse_caches check_cache isinstance Exceptionset_codegrpc StatusCodeFAILED_PRECONDITION set_detailsstr update_cache) selfrequestcontext expected_idsr"r#r$response_cache cached_entryrespefuncr)r+wrapper8s4      z$_use_response_cache..wrapper) functoolswraps)rErFr&rDr+_use_response_cache1s$rIc @seZdZdefddZ dLdejdejfddZe dLdej fd d Z dLdej fd d Z e dLdejfd dZdLdejfddZdLdejfddZ dLdejfddZdLdejfddZdLdefddZdededefddZddZdd Zd!d"Zd#d$Z e dLd%d&Z! dLdej"ded'e#d(e$j%de&ej'f d)d*Z(dej"fd+d,Z)dej"defd-d.Z* dLdej+dej,fd/d0Z- dLd1e.ee/fd2eded3efd4d5Z0dLdej1fd6d7Z2 dLd8ej3d9e4e5d:e6ee5fdej7fd;d<Z8 dLd8ej3d9e4e5d:e6ee5fdej7fd=d>Z9 dLd8ej3d9e4e5d:e6ee5fdej7fd?d@Z: dLd8ej3d9e4e5d:e6ee5fdej7fdAdBZ; dLd8ej3dej7fdCdDZj?fdFdGZ@dededEe&e6fdHdIZAdJdKZBdS)MRayletServicerray_connect_handlercCsVtt|_tt|_i|_i|_tt|_i|_t|_ t |_ ||_ tt|_dS)zuConstruct a raylet service Args: ray_connect_handler: Function to connect to ray cluster N)rr- object_refsclient_side_ref_map function_refs actor_refsset actor_ownersregistered_actor_classes named_actors threadingLock state_lockrKrr1)r<rKr&r&r+__init__as    zRayletServicer.__init__Nr=returnc Cs||jr t|j}d|_nd}d}tStr$tjjj }|j }nr\current_job_configrc extra_kwargsrCjob_uriscurrent_job_urisr&r&r+InittsZ      zRayletServicer.Initc Csz$ttjjj|j|j|j|jd}Wdn1swYWnt y=}z t ||d}WYd}~nd}~wwt j |dS)N) overwrite namespaceF)already_exists) r r` experimental internal_kv_internal_kv_putkeyvalueryrzr4return_exception_in_contextrk KVPutResponse)r<r=r>r{rCr&r&r+KVPuts   zRayletServicer.KVPutc Cz ttjjj|j|jd}Wdn1swYWnty9}z t||d}WYd}~nd}~wwt j |dS)Nrz)r) r r`r|r}_internal_kv_getrrzr4rrk KVGetResponse)r<r=r>rrCr&r&r+KVGet  zRayletServicer.KVGetc Csz"ttjjj|j|j|jd}Wdn1swYWnty;}z t ||d}WYd}~nd}~wwt j |dS)N) del_by_prefixrzr) deleted_num) r r`r|r}_internal_kv_delrrrzr4rrk KVDelResponse)r<r=r>rrCr&r&r+KVDels  zRayletServicer.KVDelc Csz ttjjj|j|jd}Wdn1swYWnty9}z t||g}WYd}~nd}~wwt j |dS)Nr)keys) r r`r|r}_internal_kv_listprefixrzr4rrkKVListResponse)r<r=r>rrCr&r&r+KVListrzRayletServicer.KVListc Cr)NrF)exists) r r`r|r}_internal_kv_existsrrzr4rrkKVExistsResponse)r<r=r>rrCr&r&r+KVExistsrzRayletServicer.KVExistscCsHttjj|jd}Wdn1swYtjt|dS)N)all_namespaces) actors_json) r r`utillist_named_actorsrrkClientListNamedActorsResponsergdumps)r<r=r>actorsr&r&r+ListNamedActorss zRayletServicer.ListNamedActorscCst}|j|_|jtjjkrz.RayletServicer.ClusterInfo..)tablecSrr&rrr&r&r+rr)!rkClusterInfoResponsetypeClusterInfoTypeCLUSTER_RESOURCESr r`cluster_resourcesitemsresource_tableCopyFrom ResourceTableAVAILABLE_RESOURCESavailable_resourcesRUNTIME_CONTEXTRuntimeContextget_runtime_context_commonutils hex_to_binary get_job_idjob_id get_node_idnode_idrz-should_capture_child_tasks_in_placement_groupcapture_client_tasks gcs_addressget_runtime_env_string runtime_envruntime_context_return_debug_cluster_inforg)r<r=r>rB resourcesfloat_resourcesctxrtcr&r&r+ ClusterInfosP        zRayletServicer.ClusterInfocCsd}|jtjjkrt}n6|jtjjkrt}n*|jtjjkr&t }n|jtjj kr0i}n|jtjj kr@dtj j i}ntdt|S)z9Handle ClusterInfo requests that only return a json blob.N dashboard_urlzUnsupported cluster info type)rrkrNODESr`nodesIS_INITIALIZEDraTIMELINEtimelinePING DASHBOARD_URLrbrcget_dashboard_url TypeErrorrgr)r<r=r>datar&r&r+rs    z)RayletServicer._return_debug_cluster_infor"idcCs|jo||jvr,||j|vr,td|d||j||= WddS||jvri||j|vritd|d||j||||r`td||j|= WddS WddS1suwYdS)NzReleasing object z for TzReleasing actor Deleting reference to actor F) rVrLridebughexrQremove_can_remove_actor_refrO)r<r"rr&r&r+release(s$    $zRayletServicer.releasecCsF|j||||Wdn1swYtdSr%)rV_release_objects_release_actorsgccollect)r<r"r&r&r+ release_all;s    zRayletServicer.release_allcs,tfdd|jD }|o|jvS)Nc3s|]}|vVqdSr%r&)r' actor_listactor_id_bytesr&r+r,Fs z7RayletServicer._can_remove_actor_ref..)r/rQvaluesrS)r<rno_ownerr&rr+rEs  z$RayletServicer._can_remove_actor_refcCsr||jvrtd|dSt|j|}|j|=||jvr#|j|=||jvr,|j|=td|d|dS)Nz%Releasing client with no references: Released all z objects for client )rLrirrsrMr1)r<r"countr&r&r+rKs   zRayletServicer._release_objectscCs||jvrtd|dSd}|j|}|D]}|d7}||r2td||j|=qtd|d|dS)Nz!Releasing client with no actors: rrrz actors for client: )rQrirpoprrrO)r<r"ractors_to_removeid_bytesr&r&r+rWs   zRayletServicer._release_actorsc Cs0|ddkrJz+|j|j|jj}ttj||jj|jj dWdn1s,wYWn_t yI}z t ||WYd}~nMd}~ww|ddkrz%|j |j j}ttj||j jdWdn1spwYWnt y}z t ||WYd}~n d}~wwtdtjddS) Nterminate_type task_object)force recursiveactor) no_restartzEClient requested termination without providing a valid terminate_typeTr[) WhichOneofrLr"rrr r`cancelrrr4rrOrkillr RuntimeErrorrkTerminateResponse)r<reqr> object_refrC actor_refr&r&r+ Terminatefs< zRayletServicer.Terminater$ result_queuec stjdkrtdjd}j|d}|s0tjdttd|ddSz4t d |t d t d dffd d }| | WdWdS1s]wYWdSty} ztjdt| dWYd} ~ Sd} ~ ww)zAttempts to schedule a callback to push the GetResponse to the main loop when the desired object is ready. If there is some failure in scheduling, a GetResponse will be immediately returned. rz3Async get() must have exactly 1 Object ID. Actual: rNFzClientObjectRef with id z not found for client validerrorz async get: %sresultrXc szLt|}t|}|dksJdt|t}tj|D])}|t}t||dt}tj d||||||d}tj |d} |q Wd St yu} ztj dt | d}tj |d}  | WYd } ~ d Sd } ~ ww) zPushes GetResponses to the main DataPath loop to send to the client. This is called when the object is ready on the server side.r&Serialized object cannot be zero bytesrTrrchunk_id total_chunks total_size)getr$FrN)rrsmathceilrrangestart_chunk_idminrk GetResponse DataResponseputr4r r) r serializedrrrstartendget_resp chunk_respexcrBr"r$r=rr<r&r+send_get_responses@     z;RayletServicer._async_get_object..send_get_response)rsids ValueErrorrLrrkr r rrirr r _on_completedr4) r<r=r"r$rr>ridrefrrCr&rr+_async_get_objects8    "(' z RayletServicer._async_get_objectccsTt|}|d}|durtjdttddVdS|||EdHdS)Nr"Fz.client_id is not specified in request metadatar) r-r.rrkr r rr _get_object)r<r=r>r*r"r&r&r+ GetObjects   zRayletServicer.GetObjectc csdg}|jD](}|j||d}|r||qtjdttd|d|dVdSz$t d|t t j||j d}Wdn1sMwYWntyq}ztjdt|dVWYd}~dSd}~wwt|||}t|} | dksJdt| t} t|j| D]} | t} t| | d t} tjd || | | | | d VqdS) NFzClientObjectRef z is not found for client rzget: %s)timeoutrrrTr)rrLrappendrkr r rrrirr r`rr4rrsrrrr r r )r<r=r" objectrefsrrrrCrrrrrrr&r&r+rsV      zRayletServicer._get_objectcCs||j|jd|j|S)z#gRPC entrypoint for unary PutObject) _put_objectr client_ref_idowner_id)r<r=r>r&r&r+ PutObjectszRayletServicer.PutObjectrr$r%c Csz+t||}|r|j|}nd}ttj||d}Wdn1s%wYWn"tyM} ztdtj ddt | dWYd} ~ Sd} ~ ww||j || <t|dkrf| |j||<td|tj | d d S) aPut an object in the cluster with ray.put() via gRPC. Args: data: Pickled data. Can either be bytearray if this is called from the dataservicer, or bytes if called from PutObject. client_ref_id: The id associated with this object on the client. client_id: The client who owns this data, for tracking when to delete this reference. owner_id: The owner id of the object. context: gRPC context. N)_ownerz Put failed:rF)rrrrzput: %sT)rr)rrOr r`rr4rirjrk PutResponser rrLbinaryrsrMr) r<rr$r"r%r>objowner objectrefrCr&r&r+r#s*     zRayletServicer._put_objectc Cs$g}|jD]}||j|jvrtdt|||j|j|q|j}|j}z%tt j |||dkr8|ndd\}}Wdn1sHwYWn tyn} zt d| t jddWYd} ~ Sd} ~ wwt dt|t|fdd |D} d d |D} t jd | | d S) Nz4Asking for a ref not associated with this client: %s) num_returnsrz Exception F)rz wait: %s %scSg|]}|qSr&r))r'ready_object_refr&r&r+ Esz-RayletServicer.WaitObject..cSr/r&r0)r'remaining_object_refr&r&r+r2HsT)rready_object_idsremaining_object_ids) object_idsrLr"r4r:r r.rr r`waitrirrk WaitResponser) r<r=r>rLrr.rready_object_refsremaining_object_refsrCr4r5r&r&r+ WaitObject-sJ   zRayletServicer.WaitObjecttaskarglistkwargsc Cs4td|jtjj|jfzftX|jtjj kr%| ||||}n:|jtjj kr5| ||||}n*|jtjj krE|||||}n|jtjjkrS|||}n tdtjj|jd|_|WdWS1sowYWdSty}ztjdddtjdt|dWYd}~Sd}~ww)Nzschedule: %s %sz$Unimplemented Schedule task type: %sTzCaught schedule exception)exc_infoFr)rirnamerk ClientTaskRemoteExecTypeNamerr FUNCTION_schedule_functionACTOR_schedule_actorMETHOD_schedule_method NAMED_ACTOR_schedule_named_actorNotImplementedErrorrr4ClientTaskTicketr r)r<r<r=r>r>rrCr&r&r+ScheduleRs<( zRayletServicer.Schedulec Csv|j|j}|durtdt||j}t|j}|dur&|jdi|}|j|i|}| ||j } t j | dS)Nz7Can't run an actor the server doesn't have a handle for return_idsr&) rOr payload_idr4getattrr@decode_optionsoptionsremoteunify_and_track_outputsr"rkrM) r<r<r=r>r> actor_handlemethodoptsoutputrr&r&r+rIts   zRayletServicer._schedule_methodcCs||j|jt|j}t|j}|dur|jdi|}t||j|i|}Wdn1s4wY||j|j <|j |j |j t j|j gdSNrOr&)lookup_or_register_actorrQr"rSbaseline_optionsrTrrUrO _actor_idr)rQaddrkrM)r<r<r=r>r> remote_classrYrr&r&r+rGs  zRayletServicer._schedule_actorc Cs||j|jt|j}t|j}|dur|jdi|}t||j|i|}Wdn1s4wY|||j}t j |dSr[) lookup_or_register_funcrQr"rSr]rTrrUrVrkrM) r<r<r=r>r> remote_funcrYrZrr&r&r+rEs   z!RayletServicer._schedule_functioncCsvt|jdks Jt|j|jpd}|j}||jvr"||j|<|j |j  ||j |t j|jgdS)NrrO)rsrQr` get_actorr@rzr^r)rOrQr"r_rSrkrM)r<r<r>r bin_actor_idr&r&r+rKs    z$RayletServicer._schedule_named_actorrTcCstB||jvr>|j||}t|}t|std|dus(t|dkr1t ||j|<n tj di|||j|<Wdn1sHwY|j|S)Nz6Attempting to register function that isn't a function.rr&) r rNrLr`rinspect isfunctionr4rsrU)r<rr"rTfuncrefrEr&r&r+ras    z&RayletServicer.lookup_or_register_funccCstA||jvr=|j||}t|}t|std|dus(t|dkr.t |}n tj di||}||j|<Wdn1sGwY|j|S)Nz0Attempting to schedule actor that isn't a class.rr&) r rRrLr`rreisclassr4rsrU)r<rr"rTactor_class_ref actor_class reg_classr&r&r+r\s      z'RayletServicer.lookup_or_register_actorcCsp|durg}n t|tr|}n|g}|D]}||j|vr'td|||j||<qdd|DS)NzAlready saw object_ref cSr/r&r0)r'outr&r&r+r2sz:RayletServicer.unify_and_track_outputs..)r3listr)rLriwarning)r<rZr"outputsrlr&r&r+rVs z&RayletServicer.unify_and_track_outputsr%)C__name__ __module__ __qualname__rrWrk InitRequestrlrxrIrrrrrrrrrrrrrrr:rbytesboolrrrrrr GetRequestr0queueQueuerr rrr PutRequestr(r&r bytearrayr#r8r;rArrrrMrNrIrGrErKr`remote_functionRemoteFunctionrar\rVr&r&r&r+rJ`s 4      '     H (   (*  '          rJcCs,|dur|t||tjjdSdSr%)r9encode_exceptionr5r6r7ABORTED)errr>r&r&r+rsrrXcCst|}t|Sr%)r rbase64standard_b64encodedecode)rjrr&r&r+r}s r}rTcCs(|jsdSt|j}t|tsJ|Sr%)pickled_optionsr]r^r3r-)rTrYr&r&r+rSs  rScCs d dtdtttffdd}|p|}tjtjtddt d}t |}t |}t }t ||t ||t ||t||t||||d}||S) Nr\rhc[sVttstjdd|i|WdSWddS1s$wYdS)Nr\r&)r r`rainitr\rhr&r&r+default_connect_handlers "z&serve..default_connect_handlerray_client_server) max_workersthread_name_prefix)rT) task_servicer data_servicer logs_servicer grpc_serverr%)rrr:rr6serverrThreadPoolExecutorrrrJrrray_client_pb2_grpc"add_RayletDriverServicer_to_server(add_RayletDataStreamerServicer_to_server'add_RayletLogStreamerServicer_to_serverrrr)connection_strrKrrrrrcurrent_handler&r&r+serves:      rcs\ttjiWdn1swYdfdd }t||d}|fS)Ncs trStjd|iS)Nr\)r`rarrargsinfor>r&r+rK%sz+init_and_serve..ray_connect_handler)rKr%)r r`rr)rrr>rK server_handler&rr+init_and_serve s  rFcCs@|dtt|WddS1swYdS)Nr)stopr r`shutdown)r_exiting_interpreterr&r&r+shutdown_with_server1s  "rcsddtffdd }|S)Nr\csXr rtjd|d|dStjd|d|dStjdd|i|dS)N)address_redis_username_redis_passwordr\)rr\r\r&)r`rrrredis_passwordredis_usernamer&r+rK8s z/create_ray_handler..ray_connect_handlerr%r)rrrrKr&rr+create_ray_handler7srrcCst|}t|dS)zr Try to create a gcs client based on the command line args or by autodetecting a running Ray cluster. r)rrrr&r&r+try_create_gcs_clientJs rc Cs`ddl}|}|jdtddd|jddtdd d|jd tgd d d |jddtdd|jddtdd|jddtdd|jddtddd|\}}ttjtj t |j |j |j }d|j|jf}t|}|j rw||j d}td|d||jd krt||j |j |j |jd}nt||}zt} dti} z"tjjst|j } tjj| tjjjd t !| tj"d!Wn$t#y} zt$d"|jd#|j t%| WYd} ~ nd} ~ wwt&d$|jd%kr|j'j(dkrt}n|d$8}|dkr t)|d&dkr|tkrt|d'qt)y/|*dYdSw)(Nrz--hostz0.0.0.0zHost IP to bind to)rdefaulthelpz-pz--porti'zPort to bind toz--mode)proxylegacyspecific-serverr)rchoicesrz --addressFz Address to use to connect to Ray)requiredrrz--redis-usernamez username for connecting to Redisz--redis-passwordz Password for connecting to Redisz--runtime-env-agent-addressz8The port to use for connecting to the runtime_env_agent.)rrrrz%s:%dz****zStarting Ray Client server on z, args )rrruntime_env_agent_addressTtimerr[z ] Failed to put health check on rrz idle checks before shutdown.)+argparseArgumentParser add_argumentr:r0parse_known_argsrr LOGGER_LEVEL LOGGER_FORMATrrrrhostportreplacerirmoderrrr rr`r|r}_internal_kv_initializedr_initialize_internal_kvr~rgrKV_NAMESPACE_HEALTHCHECKr4rrjsleepr num_clientsKeyboardInterruptr) rparserr_rKhostportargs_strridle_checks_remaining health_report gcs_clientrCr&r&r+mainSs            "r__main__r%)F)TrrGrrergloggingrr]rwrTr collectionsr concurrentrtypingrrrrrr r r6r`ray._private.state!ray.core.generated.ray_client_pb2core generatedrk&ray.core.generated.ray_client_pb2_grpcrr ray._privater ray._private.client_mode_hookr ray._rayletrray._private.ray_constantsrray._private.ray_loggingrray._private.servicesrray._private.tls_utilsrray.job_configrray.util.client.commonrrrrr#ray.util.client.server.dataservicerr"ray.util.client.server.logservicerrray.util.client.server.proxierr%ray.util.client.server.server_picklerrr#ray.util.client.server.server_stubsr getLoggerrprir rIRayletDriverServicerrJrr:r} TaskOptionsrSrrrrrrr&r&r&r+sn  $               /   !   c