o ciR@sUdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z ddlmZmZmZmZmZmZmZmZddlZddlZddlmZddlmmmZddl mmm!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/m0Z0dd l1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;ddlm?Z?ddl@mAZAerddlBmCZCddlDmEZEeFeGZHdZIdZJdZKeLeMd<dZNdZOdZPdeQdeQfddZRGdddZSdeTfd d!ZUd"ejVdeWfd#d$ZXdS)%zThis file includes the Worker class which sits on the client side. It implements the Ray API functions that are forwarded through grpc calls to the server. N) defaultdictFuture) TYPE_CHECKINGAnyCallableDictListOptionalTupleUnion)%DEFAULT_CLIENT_RECONNECT_GRACE_PERIOD)upload_py_modules_if_needed)upload_working_dir_if_needed)pickle)GetTimeoutError) JobConfig)dumps_from_clientloads_from_server) GRPC_OPTIONSGRPC_UNRECOVERABLE_ERRORS INT32_MAXOBJECT_TRANSFER_WARNING_SIZEClientActorClassClientActorHandleClientActorRefClientObjectRefClientRemoteFunc ClientStub) DataClient)LogstreamClient)log_once) ActorClass)RemoteFunctiong@MAX_BLOCKING_OPERATION_TIME_Sizkhttps://docs.google.com/document/d/167rnnDFIVRhHhK4mznEIemOtj63IOhtIPvSYaPgI4Fg/edit#heading=h.f7ins22n6nylzkhttps://docs.google.com/document/d/167rnnDFIVRhHhK4mznEIemOtj63IOhtIPvSYaPgI4Fg/edit#heading=h.1afmymq455wutimeoutreturncCs|d}|tkr t}|S)Nr$)MAX_TIMEOUT_SECr'r+J/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/client/worker.pybackoffJsr-c@seZdZ     ddededeeeefded ee j f d d Z ddd dZ de j d efddZded efddZdejd efddZdefddZde jfddZddZded eejgdfd dfd!d"Zdd#d$eed efd%d&Zdeed$efd'd(Zddd)d*ed+ee fd,d-Z! dd*ed.ee fd/d0Z"d1dd2d3d4eed5ed$ed6ed eeeeeff d7d8Z#d ee$fd9d:Z%d;ej&d5eed ee$fded dfd?d@Z(d>ed dfdAdBZ)d>ed dfdCdDZ*dEdFZ+ ddGedHeed e fdIdJZ,dKe dLed dfdMdNZ-dOedPedQed dfdRdSZ. ddTej/j0d$eefdUdVZ1dWedHeed efdXdYZ2dWedHeed efdZd[Z3dWed\ed]edHeed ef d^d_Z4dWed`edHeed efdadbZ5dcedHeed eefdddeZ6dfedged dfdhdiZ7djed ee8eeffdkdlZ9d efdmdnZ:dd efdodpZ;d efdqdrZ< ddse=dtee8eeffdudvZ>dKdwd efdxdyZ?dzd{d efd|d}Z@dWed d~fddZAdWed efddZBd efddZCdS)WorkerFNconn_strsecuremetadataconnection_retries _credentialscCs&t|_d|jfg|r |ng|_d|_d|_tjj|_i|_ |p+t j dd dv|_||_||_|dur>||_d|_nd|_t|_dt j vrQtt j d|_|jdk|_d |_d |_|d|_d |_t||j|j|_tt|_t||j|_|j t!j"d |_#d|_$t%&|_'d|_(dS) aInitializes the worker side grpc client. Args: conn_str: The host:port connection string for the ray server. secure: whether to use SSL secure channel or not. metadata: additional metadata passed in the grpc request headers. connection_retries: Number of times to attempt to reconnect to the ray server if it doesn't respond immediately. Setting to 0 tries at least once. For infinite retries, catch the ConnectionError exception. _credentials: gprc channel credentials. Default ones will be used if None. client_idN RAY_USE_TLS01trueT!RAY_CLIENT_RECONNECT_GRACE_PERIODrF))make_client_id _client_idr3channelservergrpcChannelConnectivityIDLE _conn_state _convertedosenvirongetlower_secure _conn_str_connection_retriesr5r _reconnect_grace_periodint_reconnect_enabled _in_shutdown_has_connected_connect_channel_serverside_ray_initializedr data_clientrreference_countr log_clientset_logstream_levelloggingINFOclosed!total_outbound_message_size_bytes threadingLock _req_id_lock_req_id)selfr1r2r3r4r5r+r+r,__init__RsD        zWorker.__init__r(c Cs<|jdur|j|j|j|jrI|jdur|j}n!tjdd dvr:t j j \}}}tj|||d}nt}tj|j|td|_n tj|jtd|_|j|jt}d}t}d} |t|jd ksn|r|d 7}|jrvnt|} |r| |jkrd |_td |jd z"t|jj|d t|j|_ t!|"} | rWn]t#|Wn=tj$yt%&d|dYn+tj'y} zt%&d|dt%&d| (t#|WYd} ~ nd} ~ wwt%&d|d|st)|}|t|jd ksn|sn| sd |_t*drt+,dtddS)z Attempts to connect to the server specified by conn_str. If reconnecting after an RPC error, cleans up the old channel and continues to attempt to connect until the grace period is over. Nr7r8r9)certificate_chain private_keyroot_certificatesoptionsrFTz:Failed to reconnect within the reconnection grace period (zs)r*zCouldn't connect channel in z seconds, retryingz+Ray client server unavailable, retrying in zs...zReceived when checking init: z8Waiting for Ray to become ready on the server, retry in ray_client_security_groupszRay Client connection timed out. Ensure that the Ray Client port on the head node is reachable from your local machine. See https://docs.ray.io/en/latest/cluster/ray-client.html#step-2-check-ports for more information.zray client connection timeout)-r? unsubscribe_on_channel_state_changecloserJr5rFrGrHrIray_private tls_utilsload_certs_from_envrAssl_channel_credentialssecure_channelrKrinsecure_channel subscribetimeINITIAL_TIMEOUT_SECmaxrLrPrMConnectionErrorchannel_ready_futureresultray_client_pb2_grpcRayletDriverStubr@bool ping_serversleepFutureTimeoutErrorloggerdebugRpcErrordetailsr-r!warningswarn) r` reconnecting credentialsserver_cert_chainrcca_cert start_time conn_attemptsr' service_ready elapsed_timeer+r+r,rRs        1 zWorker._connect_channelrcCsL|jsdS|jr dS|tvrdS|tjjkr$|}|dkr$dSdS)zz Returns True if the RPC error can be recovered from and a retry is appropriate, false otherwise. FzException serializing request!T)rOrPcoderrA StatusCodeINTERNALr)r`rrr+r+r,_can_reconnects zWorker._can_reconnect stub_namec Os~|js;z t|j||i|WStjy-}z||r(tdWYd}~qd}~wty:tdYqwt d)z Calls the stub specified by stub_name (Schedule, WaitObject, etc...). If a recoverable error occurrs while calling the stub, attempts to retry the RPC. ?NClient is shutting down.) rPgetattrr@rArrrtr~ ValueErrorrw)r`rargskwargsrr+r+r, _call_stubs     zWorker._call_stubreqc osd}|js|d|_zM|jj|g|Ri|D]<}|j|kr/td|jd|jdq|d|jkrDtd|jd|j d|j}|V||j dkrUWdSqWdSt j yv}z| |rqtd WYd}~qd}~wtytd Yqwtd ) z Calls the stub for GetObject on the underlying server stub. If a recoverable error occurs while streaming the response, attempts to retry the get starting from the first chunk that hasn't been received. rgzReceived a repeated chunk z from request .zReceived chunk z when we expected Nrr)rPstart_chunk_idr@ GetObjectchunk_idrrreq_id RuntimeErrorlast_seen_chunk total_chunksrArrrtr~rrw)r`rrrrchunkrr+r+r,_get_object_iterator5sH          zWorker._get_object_iteratorcCsx|js|Stt}|j|jd7_|jtkrd|_t|j}Wdn1s-wY|d|fd|fgS)aT Adds a unique req_id and the current thread's identifier to the metadata. These values are useful for preventing mutating operations from being replayed on the server side in the event that the client must retry a requsest. Args: metadata: the gRPC metadata to append the IDs to rgN thread_idr)rOstrr\ get_identr^r_r)r`r3rrr+r+r,_add_ids_to_metadataes    zWorker._add_ids_to_metadata conn_statecCstd|||_dS)Nz"client gRPC channel state change: )rrrD)r`rr+r+r,rjys zWorker._on_channel_state_changec CsHz|j}Wntjy}zt|d}~ww|j|j|j|jdS)N) num_clientspython_version ray_version ray_commit) rTConnectionInforArdecode_exceptionrrrr)r`datarr+r+r,connection_info}szWorker.connection_inforefcallbackcCs$tj|jgdd}|j||dS)NT)ids asynchronous)ray_client_pb2 GetRequestidrTRegisterGetCallback)r`rrrr+r+r,register_callbackszWorker.register_callbackr*r'cCst|tr |s gS|}nt|tr|g}ntdt||dur$d}nt|}t}dtj vr8t tj d} |rHt |t |td}n|}z| ||}Wntyl|rbt|krbtd|Ynwq9t|t|krtdt|t|t|tr|d}|S)Nz>Can't get something that's not a list of IDs or just an ID: %s(RAY_CLIENT_MAX_BLOCKING_OPERATION_TIME_STgMbP?zInternal retry for get {}zzWorker._get..)rr'r3Failed to deserialize {}#client_object_transfer_size_warningi@z'Ray Client is attempting to retrieve a z.2fzy GiB object over the network, which may be slow. Consider serializing the object to a file and using S3 or rsync instead.r$) stacklevel)rr bytearrayrr3valid cloudpickleloadserrorrUnpicklingError TypeErrorr exceptionr total_sizerr!rr UserWarningextendrrArrr) r`rr'rrresprerrsize_gbrr+r+r,rs>  z Worker._get) client_ref_id_ownerrrcCs,t|tr tdt||j}||||S)NzCalling 'put' on an ObjectRef is not allowed (similarly, returning an ObjectRef from a remote function is not allowed). If you really want to do this, you can wrap the ObjectRef in a list and call 'put' on it (or return it).)rrrrr> _put_pickled)r`valrrrr+r+r,puts  z Worker.putownerc Cs~tj|d}|dur ||_|dur|jj|_|j|}|js:zt |j t j tfy9td|j wt|jS)N)rr)r PutRequestr actor_refrowner_idrT PutObjectrrrrrrrrrrr)r`rrrrrr+r+r,rs     zWorker._put_pickledrgT) num_returnsr' fetch_local object_refsrrc Cst|tstdt||D]}t|ts tdt|qdd|D||dur.|nd|jd}tjd i|}|jd||j d}|j sLt d d d|j D} d d|j D} | | fS) Nz/wait() expected a list of ClientObjectRef, got z?wait() expected a list of ClientObjectRef, got list containing cSrr+r)r object_refr+r+r,rrzWorker.wait..r) object_idsrr'r6 WaitObjectrz.Client Wait request failed. Reference invalid?cSg|]}t|qSr+rrrr+r+r,r!cSrr+rrr+r+r,r$rr+)rrrrrr>r WaitRequestrr3rrready_object_idsremaining_object_ids) r`rrr'rrrrrclient_ready_object_idsclient_remaining_object_idsr+r+r,waits8    z Worker.waitcOsJ|}t||f|j|_|}|dkrd}|dkrtd|||S)Ndynamicr streamingzdStreaming actor methods (num_returns="streaming") are not currently supported when using Ray Client.)_prepare_client_taskrr>r _num_returnsr_call_schedule_for_task)r`instancerrtaskrr+r+r, call_remote*s zWorker.call_remotercstd|jd|jd|j|j|_|durd}|dkr#dddtDdtt j t fddffd d }|j |||j|7_|jtkrbtd rbtd td ttS)NzScheduling task  rgrcSsg|]}tqSr+r)r_r+r+r,rDrz2Worker._call_schedule_for_task..rr(c s t|trt|tjrt|}D]}||qdS|j}|jsMzt |j }Wnt j t fy@}z|}WYd}~nd}~wwD]}||qCdSt|jkrstddt|j}t|jD] \}}||qgdSt|jD] \}}||qydS)Nz Expected z returns but received )rrrArr set_exception task_ticketrrrrrrrr return_idsrzip set_result)rfutureticketexe_newexcraw_id id_futuresnum_return_refsr+r, populate_idsFs:      z4Worker._call_schedule_for_task..populate_ids%client_communication_overhead_warningaaMore than 10MB of messages have been created to schedule tasks on the server. This can be slow on Ray Client due to communication overhead over the network. If you're running many fine-grained tasks, consider running them inside a single remote function. See the section on "Too fine-grained tasks" in the Ray Design Patterns document for more details: z. If your functions frequently use large objects, consider storing the objects remotely with ray.put. An example of this is shown in the "Closure capture of large / unserializable object" section of the Ray Design Patterns document, available here: )rrnamer payload_idr>r6ranger r DataResponserrTScheduler[ByteSizeMESSAGE_SIZE_THRESHOLDr!rr$DESIGN_PATTERN_FINE_GRAIN_TASKS_LINK!DESIGN_PATTERN_LARGE_OBJECTS_LINKr)r`rrrr+rr,r9s.""   zWorker._call_schedule_for_taskrcCsD|jrdS|j|d8<|j|dkr |||j|=dSdS)Nrgr)rZrU_release_serverr`rr+r+r, call_release}s  zWorker.call_releasecCs<|jdurtd||jtj|gddSdS)Nz Releasing )r)rTrrhex ReleaseObjectrReleaseRequestr$r+r+r,r#s zWorker._release_servercCs*td||j|d7<dS)Nz Retaining rg)rrr&rUr$r+r+r, call_retainszWorker.call_retaincCsDd|_d|_|j|jd|_|jr |jd|_dSdS)NT)rPrZrTrkrVr@r?r`r+r+r,rks    z Worker.closer namespacecCst}tjj|_||_|pd|_tgif|j|_| |d}t |dks(Jt t |ddd}|j r?td|d|S)Nr/rgrT)weak_refz ActorID for z is empty)r ClientTask NAMED_ACTORrrr+rr>rrrrrris_nilr)r`rr+rfutureshandler+r+r, get_actors    zWorker.get_actoractor no_restartc Cst|tstdt|tj}|jj |_ ||_ tj|d}|j |_ z |j |WdStjy?}zt|d}~ww)Nz.ray.kill() only supported for actors. Got: {}.)r3)rrrrrrTerminateRequestActorTerminaterrr4r>r6rT TerminaterArr)r`r3r4 term_actortermrr+r+r,terminate_actors     zWorker.terminate_actorobjforce recursivec Cst|tstdt|dtj}|j|_||_||_ tj|d}|j |_ z |j |WdStjyB}zt|d}~ww)Nzr6rTr7rArr)r`r;r<r= term_objectr9rr+r+r,terminate_tasks$   zWorker.terminate_taskreq_typecCs`t}||_|jj|||jd}|ddkr t|jj }|S|ddkr*|j St |j S)N)r'r3 response_typeresource_tableruntime_context) rClusterInfoRequestrr@ ClusterInfor3 WhichOneofdictrDtablerEjsonr)r`rBr'rr output_dictr+r+r,get_cluster_infos  zWorker.get_cluster_infokeyc Cs\tj||d}z |jd||jd}Wntjy#}zt|d}~ww|dr,|jSdS)NrNr+KVGetrvalue) r KVGetRequestrr3rArrHasFieldrQr`rNr+rrrr+r+r,internal_kv_gets zWorker.internal_kv_getc CsLtj||d}z |jd||jd}W|jStjy%}zt|d}~ww)NrOKVExistsr)rKVExistsRequestrr3rArrexistsrTr+r+r,internal_kv_existsszWorker.internal_kv_existsrQ overwritec CsZtj||||d}||j}z |jd||d}W|jStjy,}zt|d}~ww)N)rNrQrZr+KVPutr) r KVPutRequestrr3rrArralready_exists) r`rNrQrZr+rr3rrr+r+r,internal_kv_puts zWorker.internal_kv_put del_by_prefixc CsXtj|||d}||j}z |jd||d}W|jStjy+}zt|d}~ww)N)rNr_r+KVDelr) r KVDelRequestrr3rrArr deleted_num)r`rNr_r+rr3rrr+r+r,internal_kv_dels zWorker.internal_kv_delprefixc CsHztj||d}|jd||jdjWStjy#}zt|d}~ww)N)rdr+KVListr)r KVListRequestrr3keysrArr)r`rdr+rrr+r+r,internal_kv_list szWorker.internal_kv_listuri expiration_scCs$tj||d}|jd||jddS)N)rirjPinRuntimeEnvURIr)rClientPinRuntimeEnvURIRequestrr3)r`rirjrr+r+r,pin_runtime_env_uriszWorker.pin_runtime_env_uriall_namespacescCs tj|d}t|j|jS)N)rn)rClientListNamedActorsRequestrKrrTListNamedActors actors_json)r`rnrr+r+r,list_named_actorss zWorker.list_named_actorscCs2|r |jdur dS|js|tjj|_|jS)NF) is_connectedr@rSrMrClusterInfoTypeIS_INITIALIZEDr*r+r+r,is_initializedszWorker.is_initializedcCs2|jdurtd|jtjj|d}|duSdS)zSimple health check. Piggybacks the IS_INITIALIZED call to check if the server provides an actual response. NzPinging server.r*F)r@rrrMrrtPING)r`r'ryr+r+r,r}+s  zWorker.ping_servercCs|j o|jSN)rPrQr*r+r+r,rs9szWorker.is_connected job_configray_init_kwargsc Cs|duri}z^|durd}n9t(}|jpi}t||td}t||td}|dd|j|ddWdn1s=wYt |}|j t j |t ||jd}|jsbtd|jWdStjyu}zt|d}~ww)zInitialize the serverN)rexcludesT)validate)ryrzreconnect_grace_periodz$Initialization failure from server: )tempfileTemporaryDirectory runtime_envrrrpopset_runtime_envrdumpsrTInitr InitRequestrKrMokConnectionAbortedErrormsgrArr)r`ryrzserialized_job_configtmp_dirrresponserr+r+r, _server_init<sB     zWorker._server_initr"cCs*tj}|jj}t||jd|j|<|S)z@Register a ClientActorClass for the ActorClass and return a UUIDre)uuiduuid4r&__ray_metadata__modified_classr_default_optionsrE)r`r3rNclsr+r+r,_convert_actorcs zWorker._convert_actorfuncr#cCs$tj}t|j|jd|j|<|S)z@Register a ClientRemoteFunc for the ActorClass and return a UUIDre)rrr&r _functionrrE)r`rrNr+r+r,_convert_functionjs  zWorker._convert_functionrcCs |j|S)z)Given a UUID, return the converted objectrEr`rNr+r+r,_get_convertedr zWorker._get_convertedcCs ||jvS)zACheck if a key UUID is present in the store of converted objects.rrr+r+r,_converted_key_existsvrzWorker._converted_key_existscCs t||jSrx)rr>)r`rr+r+r,_dumps_from_clientzs zWorker._dumps_from_client)r/FNr0N)F)r(Nrx)D__name__ __module__ __qualname__rr|r r rNr rAChannelCredentialsrarRrrrrrrrrrBrjrrrrrrrHrbytesrrrrrrr-rr%r#r)rkr2r:rArtTypeEnumrMrUrYr^rcrhrmrrrrvr}rsrrrrrrrr+r+r+r,r.Qs" Oe 0  0"   & D            'r.cCst}|jSrx)rrr&rr+r+r,r=~sr=rcCs4|tjjkrtd|St|}t|S)NzGRPC connection failed: ) rrArABORTEDrwbase64standard_b64decoderr)rrr+r+r,rsr)Y__doc__rrKrXrFr~r\rtrr collectionsrconcurrent.futuresrtypingrrrrr r r r rAray._private.tls_utilsrlray.cloudpickler!ray.core.generated.ray_client_pb2core generatedr&ray.core.generated.ray_client_pb2_grpcrzray._private.ray_constantsr #ray._private.runtime_env.py_modulesr$ray._private.runtime_env.working_dirrray.cloudpickle.compatrray.exceptionsrray.job_configrray.util.client.client_picklerrrray.util.client.commonrrrrrrrrrrray.util.client.dataclientrray.util.client.logsclientr ray.util.debugr! ray.actorr"ray.remote_functionr# getLoggerrrrur)r&r__annotations__r r!r"rNr-r.rr=rrrr+r+r+r,sb  (       0      3