&`iYzdZddlZddlZddlZddlZddlZddlmZddlm Z m Z m Z m Z m Z mZddlZddlmcmcmZddlmcmcmZddlmZmZmZddlmZe rddlmZejeZ e eej!e"fgdfZ#dZ$d ej%fd Z&d ej%fd Z'Gd d Z(GddZ)dS)zjThis file implements a threaded stream controller to abstract a data stream back to the ray clientserver. N) OrderedDict) TYPE_CHECKINGAnyCallableDictOptionalUnion) INT32_MAXOBJECT_TRANSFER_CHUNK_SIZEOBJECT_TRANSFER_WARNING_SIZE)log_once)Worker reqc #*K|jj}t|}|dks Jd|tkr3t dr$|dz }t jd|ddttj |tz }td|D]z}|tz}t||dztz}tj|jj|||||||jj }tj|j| V{d S) aS Chunks a put request. Doing this lazily is important for large objects, since taking slices of bytes objects does a copy. This means if we immediately materialized every chunk of a large object and inserted them into the result_queue, we would effectively double the memory needed on the client to handle the put. r%Cannot chunk object with missing dataclient_object_put_size_warning@z#Ray Client is attempting to send a .2fa GiB object over the network, which may be slow. Consider serializing the object and using a remote URI to transfer via S3 or Google Cloud Storage instead. Documentation for doing this can be found here: https://docs.ray.io/en/latest/handling-dependencies.html#remote-uris) client_ref_iddatachunk_id total_chunks total_sizeowner_id)req_idputN)rrlenr r warningswarn UserWarningmathceilr rangeminray_client_pb2 PutRequestrr DataRequestr) r request_datarsize_gbrrstartendchunks n/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/client/dataclient.py chunk_putr0"sT7>>B>>>111h(771u$  S D S S S    9Z*DDEEL!\** G G55*x!|/IIJJ)'/eCi(%!W%    ( FFFFFFF G Gc #K|jj}t|}|dks Jdtj|t z }t d|D]}|t z}t||dzt z}tj |jj |jj |jj |jj |jj|jj|jj||||| }tj|j|VdS)aY Chunks a client task. Doing this lazily is important with large arguments, since taking slices of bytes objects does a copy. This means if we immediately materialized every chunk of a large argument and inserted them into the result_queue, we would effectively double the memory needed on the client to handle the task. rrr) typename payload_id client_idoptionsbaseline_options namespacerrr)rtaskN)r:rrr#r$r r%r&r' ClientTaskr3r4r5r6r7r8r9r)r)rr*rrrr,r-r.s r/ chunk_taskr<Ls8=L\""J >>>B>>>9Z*DDEEL!\**HH55*x!|/IIJJ)x*h(H$ X6h(eCi(%    ( GGGGGGGHHr1cVeZdZdZdedejfdZdeej e fde fdZ dS) ChunkCollectora This object collects chunks from async get requests via __call__, and calls the underlying callback when the object is fully received, or if an exception while retrieving the object occurs. This is not used in synchronous gets (synchronous gets interact with the raylet servicer directly, not through the datapath). __call__ returns true once the underlying call back has been called. callbackrequestcVt|_||_d|_||_dS)N) bytearrayrr?last_seen_chunkr@)selfr?r@s r/__init__zChunkCollector.__init__ys)KK   ! r1responsereturncXt|tr||dS|j}|js||dS|jt kr8tdr)|jdz }tj d|ddt|j }|j }||j dzkr;|j |||_ |j dz|jj_n||j dzkrVd|d |j dzd |j}t$||t)|dSt$d |d |jd |j |jdz kr||j dSdS)NT#client_object_transfer_size_warningrz'Ray Client is attempting to retrieve a rzy GiB object over the network, which may be slow. Consider serializing the object to a file and using rsync or S3 instead.rzReceived chunk z when we expected z for request zReceived a repeated chunk z from request .F) isinstance Exceptionr?getvalidrr r r r!r"rrrDextendr@start_chunk_idrloggerwarning RuntimeErrordebugr)rErGget_respr+ chunk_datarmsgs r/__call__zChunkCollector.__call__s h * *  MM( # # #4<~  MM( # # #4  != = =( 1C C =)E1G M-H---    ] $ t+a/ / / I  Z ( ( (#+D /3.BQ.FDL  + + ,q0 0 0L(LL'!+LL:B/LL  NN3    MM,s++ , , ,4 LL3X33 (333      5 9 9 9 MM$) $ $ $45r1N) __name__ __module__ __qualname____doc__ResponseCallabler'r)rFr DataResponserMboolrYr1r/r>r>mst   !1 NcheZdZdddedefdZdefdZdej fdZ d Z d&d Z d e dd fd ZdejdefdZd&dZdedd fdZd&dZedZd&dZdejdejfdZ d'dejdeedd fdZdZ d'dej!dej"fdZ# d'dej$dej%fdZ&d'dej'fdZ( d'dej)dej*fdZ+dej)dedd fd Z, d'dej-dej.fd!Z/ d'dej0dd fd"Z1dej2defd#Z3dej4dej5fd$Z6dej7dej8fd%Z9d S)( DataClient client_workerrr6metadatac||_||_||_||_t |_tj|_ tj |j |_ | |_ i|_i|_d|_d|_d|_d|_|jdS)aInitializes a thread-safe datapath over a Ray Client gRPC channel. Args: client_worker: The Ray Client worker that manages this client client_id: the generated ID representing this client metadata: metadata to pass to gRPC requests )lockFrN)rd _client_id _metadata_start_datathread data_threadroutstanding_requests threadingLockrg Conditioncv _create_queue request_queue ready_dataasyncio_waiting_data _in_shutdown_req_id_last_exception_acknowledge_counterr,)rErdr6res r/rFzDataClient.__init__s+#!11335@MM! N$$ %49555!//11*,BD!! #$%!      r1rHc|jsJ|xjdz c_|jtkrd|_|jdksJ|jS)Nrr)rglockedrvr rEs r/_next_idzDataClient._next_ids]y!!!!!  <) # #DL|q    |r1c<tj|jdddS)Nray_client_streaming_rpcraT)targetr4argsdaemon)rmThread _data_mainr{s r/rjzDataClient._start_datathreads*?+     r1c#K |j}|dS|d}|dkrt|Ed{Vn |dkrt |Ed{Vn|Vo)NTr3rr:)rrrN WhichOneofr0r<)rErreq_types r/ _requestszDataClient._requestss $((**C{~~f--H5  $S>>))))))))V##%c??********  r1Ncd} |jjs=tj|jj}|jdt |fgz}|||d} |D]}| | t d| dS#tj$rp}||}|s;||_Yd}~t d| dS|Yd}~nd}~wwxYw|jj=n#t$$r}||_Yd}~nd}~wwxYwt d| dS#t d| wxYw)NF reconnectingT)rewait_for_readyzShutting down data channel.)rdruray_client_pb2_grpcRayletDataStreamerStubchannelristrDatapathr_process_responserRrU _shutdowngrpcRpcError_can_reconnectrw_reconnect_channelrM)rErstubre resp_streamrGes r/rzDataClient._data_mains  (5 .*A&. >nc,>O>O-P,QQ"mmNN$$%#', .$/99..x8888 LL6 7 7 7 NN     }...#'#6#6q#9#9L'/0, LL6 7 7 7 NN      ++-------- .(5 .( % % %#$D  % LL6 7 7 7 NN      LL6 7 7 7 NN    s`A/E4B??D>D9,E D94E9D>>EF E*E% F%E**F0G rGc|jdkrtd|dS|j|jvrd} |j|j}t |t r ||}n |r |||r |j|j=n*#t $rtdYnwxYw|j5|j|j vr)|r'|j |j=| |jddddS#1swxYwYdS|j5||j |j<|j ddddS#1swxYwYdS)z; Process responses from the data servicer. rzGot unawaited response NTzCallback error:)rrRrUrtrLr>rM exceptionrgrl _acknowledgersrp notify_all)rErG can_remover?s r/rzDataClient._process_response s ?a   LL=8== > > > F ?d7 7 7J 44X_Eh77'!)(!3!3JJ'HX&&&C 1(/B 4 4 4  !233333 4 7 7?d&???J?1(/B%%ho666  7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 % %3;0""$$$ % % % % % % % % % % % % % % % % % %s6AB $B32B3=8DD D)E  EErc|j|s6tdt|dStdt|dS)z Processes RPC errors that occur while reading from data stream. Returns True if the error can be recovered from, False otherwise. z$Unrecoverable error in data channel.Fz"Recoverable error in data channel.T)rdrrRerrorrU)rErs r/rzDataClient._can_reconnectDsk !0033  LL? @ @ @ LLOOO5 9::: Qtr1cD|j5d|_|j|j}i|_dddn #1swxYwY|jrtd|j}ntd}|D]}|r ||dS)z+ Shutdown the data channel TNzOFailed during this or a previous request. Exception that broke the connection: zERequest cannot be fulfilled because the data client has disconnected.)rgrurprrtvaluesrwConnectionError)rE callbackserrr?s r/rzDataClient._shutdownQsY + + $D  G   188::I(*D %  + + + + + + + + + + + + + + +   !@)-)=@@CC " C"  H    sAAAArc"|jjsdS|jsJ|xjdz c_|jt zdkrB|jtj tj |dSdS)z Puts an acknowledge request on the request queue periodically. Lock should be held before calling this. Used when an async or blocking response is received. Nrr)r) acknowledge) rd_reconnect_enabledrgrzrxACKNOWLEDGE_BATCH_SIZErrrr'r)AcknowledgeRequest)rErs r/rzDataClient._acknowledgems !4  Fy!!!!! !!Q&!!  $'= = B B   " "* . A P P P      C Br1cF |jd}n#tj$rd}YnwxYw|sztd |jdn)#t$rtdwxYwtd|j 5| |_ |j D]}|j | d d d d S#1swxYwYd S) a Attempts to reconnect the gRPC channel and resend outstanding requests. First, the server is pinged to see if the current channel still works. If the ping fails, then the current channel is closed and replaced with a new one. Once a working channel is available, a new request queue is made and filled with any outstanding requests to be resent to the server. )timeoutFzKEncountered connection issues in the data channel. Attempting to reconnect.T)rz$Failed to reconnect the data channelzReconnection succeeded!N)rd ping_serverrrrRrS_connect_channelrrUrgrqrrrlrr)rEping_succeededr@s r/rzDataClient._reconnect_channels #"/;;A;FFNN} # # #"NNN # 4 NN+    "333FFFF"   EFFF  LL2 3 3 3Y 0 0!%!3!3!5!5D 4;;== 0 0"&&w//// 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0s+22A..&B8ADDDc(tjSN)queue SimpleQueuerar1r/rqzDataClient._create_queues """r1cd}|j5d|_|j|jZt jt j}|j||jd|j |j }dddn #1swxYwY|| dSdS)NT)connection_cleanup) rgrurprrrr'r)ConnectionCleanupRequestrrkjoin)rEthreadcleanup_requests r/closezDataClient.closes Y * * $D  G   !-#1"<'5'N'P'P###"&&777"&&t,,,+) * * * * * * * * * * * * * * *   KKMMMMM  sBB&&B*-B*rcj5|_j||j<jfdj }j =j= dddn #1swxYwY|S)Nc$jvpjSr)rsru)rrEsr/z+DataClient._blocking_send..sVt%>%S$BSr1) rg_check_shutdownr|rrrrrlrpwait_forrsr)rErrrs` @r/_blocking_sendzDataClient._blocking_sendsY & &  " " "]]__FCJ   " "3 ' ' '03D %f - G  SSSSS T T T  " " "?6*D')&1   f % % % & & & & & & & & & & & & & & & sB8CCCr?c|j5||}||_||j|<||j|<|j|ddddS#1swxYwYdSr)rgrr|rrtrlrrr)rErr?rs r/ _async_sendzDataClient._async_sends Y ( (  " " "]]__FCJ08D %f -03D %f -   " "3 ' ' '  ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( (sAA33A7:A7cd|jsJ|jsdS|jt jj|jjkrdSddlm }||j |j d|j }nd}t|)Nr) disconnectzlRequest can't be sent because the Ray client has already been disconnected due to an error. Last exception: zKRequest can't be sent because the Ray client has already been disconnected.) rgrzrureleasermcurrent_threadidentrkray.utilracquirerwr)rErrXs r/rzDataClient._check_shutdownsy!!!!!   F   # % % +t/?/E E E F''''''     +*'** C%  c"""r1r@cdtj|}||}|jS)N)init)r'r)rrrEr@contextdatareqresps r/InitzDataClient.Inits;!,   ""7++yr1cdtj|}||}|jS)N)prep_runtime_env)r'r)rrrs r/PrepRuntimeEnvzDataClient.PrepRuntimeEnv s<!,$   ""7++$$r1ctjtj}||}|jS)N)connection_info)r'r)ConnectionInfoRequestrr)rErrrs r/ConnectionInfozDataClient.ConnectionInfosC ,*@BB   ""7++##r1cdtj|}||}|jS)NrN)r'r)rrNrs r/ GetObjectzDataClient.GetObject:!,   ""7++xr1ct|jdkrtd|tj|}t ||}|||dS)Nrz=RegisterGetCallback() must have exactly 1 Object ID. Actual: r)r?r@)rids ValueErrorr'r)r>r)rEr@r?r collectors r/RegisterGetCallbackzDataClient.RegisterGetCallback#s w{  q %"%% !,   #HgFFF  ),,,,,r1cdtj|}||}|jS)N)r)r'r)rrrs r/ PutObjectzDataClient.PutObject2rr1cZtj|}||dS)N)rr'r)r)rEr@rrs r/ ReleaseObjectzDataClient.ReleaseObject;s:!,    !!!!!r1c\tj|}|||dS)N)r:r)rEr@r?rs r/SchedulezDataClient.ScheduleCs1 ,'::: (+++++r1cdtj|}||}|jS)N) terminate)r'r)rrrEr@rrs r/ TerminatezDataClient.TerminateGs;(   ""3''~r1cdtj|}||}|jS)N)list_named_actors)r'r)rrrs r/ListNamedActorszDataClient.ListNamedActorsPs<(%   ""3''%%r1)rHNr):rZr[r\rlistrFintr|rmrrjrrrrrrr`rrrr staticmethodrqrr'r)r_rrr^rr InitRequest InitResponserPrepRuntimeEnvRequestPrepRuntimeEnvResponserConnectionInfoResponser GetRequest GetResponserrr( PutResponserReleaseRequestrr;rTerminateRequestTerminateResponserClientListNamedActorsRequestClientListNamedActorsResponserrar1r/rcrcs#!h#!3#!$#!#!#!#!L# 9#3       :"%#"%$"%"%"%"%H  $    834$$0$0$0$0N##\#*!-  $.04 ( (  ' (+, (  ( ( ( (###B<@%1  $FJ%%%;%  .%%%%$$n.S$$$$;?%0  # -%0 -rcrar1r/r s# ######FFFFFFFFFFFFFFFF ::::::::::::DDDDDDDDDDDD $#####.------  8 $ $U>#> #IJKTQR'G>-'G'G'G'GTHN.HHHHBJJJJJJJJZ]&]&]&]&]&]&]&]&]&]&r1