&`i|ddlZddlZddlZddlmZmZddlmZmZm Z m Z m Z m Z m Z ddlZddlZddlmZddlmZddlmZmZddlmZddlmZdd lmZmZejeZ d e!d d fd Z"edgdZ#GddZ$GddeZ%edGddeZ&eGddeZ'edGddeZ(dS)N) defaultdict namedtuple)AnyDictListOptionalSetTupleUnion)SerializedObject)utils)ChannelInterfaceChannelOutputType)IntraProcessChannel)get_self_actor) DeveloperAPI PublicAPIbuffer_size_bytesreturn ray.ObjectRefctjjj}|d|z} ||dd}n3#tjj$rt dwxYw|S)a Create a channel that can be read and written through Ray's shared-memory object store. The channel has no buffer, so the writer will block until reader(s) have read the previous value. A writer and colocated readers can communicate via a shared memory buffer. If the readers are remote, then RPC is used to synchronize the writer and readers' buffers. Args: buffer_size_bytes: The initial buffer size in bytes for messages that can be passed between tasks in the DAG. The buffers will be automatically resized if larger messages are written to the channel. Returns: Channel: A wrapper around ray.ObjectRef. 0NT) owner_address_is_experimental_channelzXPut failed since the value was either too large or the store was full of pinned objects.) ray_privateworker global_workercheck_connected put_object exceptionsObjectStoreFullErrorloggerinfo)selfrrvalue object_refs /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/experimental/channel/shared_memory_channel.py_create_channel_refr)s.\ .F  $ $E && '   > .  0      s A 0A: ReaderRefInfo reader_refref_owner_actor_idnum_reader_actorsc.eZdZdZdeeeffdZdS)_ResizeChannela_ When a channel must be resized, the channel backing store must be resized on both the writer and the reader nodes. The writer first resizes its own backing store. The writer then uses an instance of this class as a sentinel value to tell the reader to resize its own backing store. The class instance is sent through the channel. _node_id_to_reader_ref_infoc||_dS)z\ Args: _node_id_to_reader_ref_info: A node id to ReaderRefInfo. N)r1)r%r1s r(__init__z_ResizeChannel.__init__Ps,G(((N)__name__ __module__ __qualname____doc__rstrr*r3r4r(r0r0HsKG%)#}*<%=GGGGGGr4r0c eZdZddddeedeeffdZ d deddeedefd eed d fd Z xZ S)SharedMemoryTypeN)rnum_shm_buffersrr=ctddlm}|}||j}||_|d}||_dS)a Args: buffer_size_bytes: The initial buffer size in bytes for messages that can be passed between tasks in the DAG. The buffers will be automatically resized if larger messages are written to the channel. num_shm_buffers: The number of shared memory buffers per channel. Note: In the case of multiple nodes, we only support 1 shared memory buffer. r) DAGContextN)superr3ray.dagr? get_currentr_num_shm_buffers)r%rr=r?ctx __class__s r(r3zSharedMemoryType.__init__\sn &&&&&&$$&&  $ # 5 !2  "O /r4writerray.actor.ActorHandlereader_and_node_listdriver_actor_idrChannelc0t|||j|S)a Instantiate a ChannelInterface class that can be used to pass data of this type. Args: writer: The actor that may write to the channel. None signifies the driver. reader_and_node_list: A list of tuples, where each tuple contains a reader actor handle and the node ID where the actor is located. driver_actor_id: If this channel is read by a driver and that driver is an actual actor, this will be the actor ID of that driver actor. Returns: A ChannelInterface that can be used to pass data of this type. )CompositeChannelrD)r%rGrIrJs r(create_channelzSharedMemoryType.create_channelys%*   !     r4N) r5r6r7rintr3rr r9rN __classcell__)rFs@r(r<r<[s,0)- 000$C=0"# 000000B*.   01 #5)@#)E#FG "#           r4r<alpha) stabilityc2eZdZdZ d'deejjdee de fdee e e fded d ed d eee efd edefdZd ee efded fdZde fdZedZd(dZd(dZedejjdee de fde dd dee efdededdfdZdZde fdZde de fd Zd)d!ed"eeddfd#Zd)d"eedefd$Zd)d"eeddfd%Z d(d&Z!dS)*rKz\ A wrapper type for ray.ObjectRef. Currently supports ray.get but not ray.wait. NFrGrIrHtyp_writer_node_idz ray.NodeID _writer_refrr1_writer_registered_reader_registeredc Dt|dksJ|D]&\} } t| tjjsJ'|t }n%t|t rt |}t d} |j| krtd| d||_ ||_ ||_ tj j j|_|j||_||_|pi|_t)t*|_|j D]%\} } |j| | &d|_|t3} || ksJtj|_t=||j|_| |jn.| Jd | Jd||_||_||_|jdksJd }|j!D]:\} }|"| r|xjt|z c_8d };|r|xjd z c_|jdksJ|#|j|_$dS) a Create a channel that can be read and written by co-located Ray processes. Anyone may write to or read from the channel. The channel has no buffer, so the writer will block until reader(s) have read the previous value. Args: writer: The actor that may write to the channel. None signifies the driver. reader_and_node_list: A list of tuples, where each tuple contains a reader actor handle and the node ID where the actor is located. typ: Type information about the values passed through the channel. Either an integer representing the max buffer size in bytes allowed, or a SharedMemoryType. Returns: Channel: A wrapper around ray.ObjectRef. rN)rz8typ.buffer_size_bytes must be at least MIN_BUFFER_SIZE (z bytes)z<_writer_node_id must also be passed to the constructor when zW_node_id_to_reader_ref_info must also be passed to the constructor when _writer_ref is.FTr@)%len isinstanceractor ActorHandler<rPr ValueError_writer_reader_and_node_list_typrrr_workerrrXrYr1rlist_node_id_to_readersappend_num_local_readersrruntime_contextget_runtime_context get_node_idrVr)rW_create_reader_refsitems is_local_node_get_local_reader_ref_local_reader_ref)r%rGrIrUrVrWr1rXrYreader_MIN_BUFFER_SIZEnode_id self_actorremote_node_existsreaderss r(r3zChannel.__init__s8'((1,,,,- = =IFAfci&;<< < << < ;"$$CC S ! ! :"S999Cd))  ? 2 2-#---   %9" |*8  $$&&&"4"4 ( -2 ( HSSWGXGX #9 = =OFG  $W - 4 4V < < < <#$  ())JZ''''#7799EEGG   349NOOD   $ $S%: ; ; ; ; ++M,++ .::';:: +D #2D /JD ,&!++++" $ 8 > > @ @ * * GW!!'** *''3w<<7'''%)""  )  # #q ( # #&****<@ > @ @   GW%%g.. !((,(H(N(N(P(P > > $G_!!'** .= "7 + +  EE   "   #'r4c|jrdS|jD]>\}}||r$|jj|j?d|_dSNT)rYr1rmrnrdr$experimental_channel_register_readerr,)r%rtrys r(ensure_registered_as_readerz#Channel.ensure_registered_as_readerYs  "  F(,(H(N(N(P(P   $G_!!'**  (MM#.#'r4 writer_refnode_id_to_reader_ref_infowriter_registeredreader_registeredc 4t||||||||}|S)N)rVrWr1rXrYrK) rGrIrUwriter_node_idrrrrchans r(_deserialize_reader_channelz#Channel._deserialize_reader_channelds7  *"(B00     r4c |jJ|j|j|j|j|j|j|j|j|jffSrO) r1rrarbrcrVrWrXrYr%s r( __reduce__zChannel.__reduce__{sT/;;;/ L  & I     ,  #  # 2  r4c(d|jd|jdS)Nz$Channel(_node_id_to_reader_ref_info=z, _writer_ref=))r1rWrs r(__str__zChannel.__str__s0 /43S / /+ / / / r4serialized_value timeout_msc|jt|jz}||jjkr||j_|j}t ||jj|_||jj||j |_ t|j }|j |}|j j|||j|dSdSrO) total_bytesr\metadatarcrrWr)rlror1rpr0rdget_serialization_context serializer#experimental_channel_put_serializedrh)r%rrsizeprev_writer_refspecial_messagespecial_message_serializeds r(_resize_channel_if_neededz!Channel._resize_channel_if_neededs +c2B2K.L.LL $)- - -*.DI '#.O249TUUD   $ $TY%@ A A A%)%?%?0&&D " -T-MNNO 6688BB?SS ' L $ H H*'      ' . -r4r&timeoutc.|||dks|dks Jd|t|dznd}t|ts |j|}n#t$rp}tj }tj ||dt|d|}t||d}~wwxYw|}tj}||||9|ttj|z dzz}t%|d}|jj||j|j|dS)Nr#Timeout must be non-negative or -1.r[) print_filez"Could not serialize the put value z: )rrPr]r rdrr TypeErrorioStringIOrutilinspect_serializabilityreprgetvaluetime monotonicrmaxrrrWrh) r%r&rrresiomsg start_times r(writez Channel.writes ((*** Ow!||w"}}} 00=} <-4,?S4(((R %!122 % ,#'<#I#I#K#K#U#U$$   , , ,kmm0030GGG(E{{((||~~((  nn!+ , % ^%%  &&'7DDD   #t~//*<DEE EJZ++J  DD     #       s,B DA+C;;DcL||dks|dks Jd|tj}|j|jg|ddd}t |tr|j|_| |j|_d|_ ||)|tj|z z}t|d}|j|jg|ddd}|S)NrrrT)rreturn_exceptionsF) rrrrd get_objectsrpr]r0r1rorYr)r%rrrets r(readz Channel.readsA Ow!||w"}}} 00=} < ((***^%% l&&  # $g'    c> * * /2/ND ,%)%?%?0&&D "',D #  , , . . ."4>++j88gq//,**'('T+C r4c||dks|dks Jd||j|jg|dddS)NrrrT)rrskip_deserialization)rrdrrpr%rs r(release_bufferzChannel.release_bufferss Ow!||w"}}} 00=} < ((***    # $"!% !     r4cd|jj|jd}|jD]}||rd}|r||j D]&}|jj|j 'dS)zp Close this channel by setting the error bit on both the writer_ref and the reader_ref. FTN) rdrexperimental_channel_set_errorrWrfkeysrnrr1valuesr,)r%is_local_node_readerrtrys r(closez Channel.closes  ??@PQQQ$/4466 , ,G!!'** ,'+$  /  , , . . .#?FFHH  O L $ C C*      r4)NNNNFFrNrO)"r5r6r7r8rrr^r_rr r9r rPr<rr*boolr3rorl staticmethodrnrrrrrrrfloatrrrrr:r4r(rKrKs7;2615JN#(#(s s ./s #5)@#)E#FGs eC!112 3 s ",/ s o. s &.d3 3E.F%Gs !s !s s s s j+/]0B+C / "#+#+#+#+#+JRR\R'''', ' ' ' ' %"5)@#)E#FG $  %)m);$<    \,         #3J$ $ 3$ %$ D$ $ $ $ LHUOs8   huo       r4rKc eZdZdZ ddeejjdee de fde dee e e ffdZd Zd Zdd ed eed dfdZdd eed efdZdd eefdZddZedZedZdS)BufferedSharedMemoryChannelaA channel that can be read and written by Ray processes. It creates `num_shm_buffers` number of buffers and allows buffered read and write APIs. I.e., read and write APIs are non-blocking as long as it can write to next buffer or read from a next buffer. See `read` and `write` APIs for more details. Args: writer: The actor that may write to the channel. None signifies the driver. reader_and_node_list: A list of tuples, where each tuple contains a reader actor handle and the node ID where the actor is located. Note that currently we only support this for readers on the same node as the writer. num_shm_buffers: Number of shared memory buffers to read/write. typ: Type information about the values passed through the channel. Either an integer representing the max buffer size in bytes allowed, or a SharedMemoryType. NrGrIrHr=rUcz||_fdt|D|_d|_d|_dS)Nc2g|]}tSr:r).0rrrIrUrGs r( z8BufferedSharedMemoryChannel.__init__..1s8    F0# 6 6   r4r)rDrange_buffers_next_write_index_next_read_index)r%rGrIr=rUs `` `r(r3z$BufferedSharedMemoryChannel.__init__)se!0      ?++     "# !r4cB|jD]}|dS)z^ Check whether the process is a valid writer. This method must be idempotent. N)rrr%buffers r(rz7BufferedSharedMemoryChannel.ensure_registered_as_writer<4m 1 1F  . . 0 0 0 0 1 1r4cB|jD]}|dS)z^ Check whether the process is a valid reader. This method must be idempotent. N)rrrs r(rz7BufferedSharedMemoryChannel.ensure_registered_as_readerCrr4r&rrc||jdksJ|j|j|||xjdz c_|xj|jzc_dS)a5Write a value to a channel. If the next buffer is available, it returns immediately. If the next buffer is not read by downstream consumers, it blocks until a buffer is available to write. If a buffer is not available within timeout, it raises RayChannelTimeoutError. rr@N)rrrrrrD)r%r&rs r(rz!BufferedSharedMemoryChannel.writeJs{ ((***$)))) d,-33E7CCC !# $"77r4c||jdksJ|j|j|}|xjdz c_|xj|jzc_|S)a8Read a value from a channel. If the next buffer is available, it returns immediately. If the next buffer is not written by an upstream producer, it blocks until a buffer is available to read. If a buffer is not available within timeout, it raises RayChannelTimeoutError. rr@)rrrrrrD)r%routputs r(rz BufferedSharedMemoryChannel.readYsw ((***%****t45::7CC " !66 r4c|jdksJ|j|j||xjdz c_|xj|jzc_dS)aRelease the native buffer of the channel to allow the buffer to be reused for future data. If the next buffer is available, it returns immediately. If the next buffer is not written by an upstream producer, it blocks until a buffer is available to be released. If a buffer is not available within timeout, it raises RayChannelTimeoutError. rr@N)rrrrrDrs r(rz*BufferedSharedMemoryChannel.release_bufferisg%**** d+,;;GDDD " !66r4cB|jD]}|dSrO)rrrs r(rz!BufferedSharedMemoryChannel.closexs,m  F LLNNNN  r4c|jSrO)rrs r(next_write_indexz,BufferedSharedMemoryChannel.next_write_index|s %%r4c|jSrO)rrs r(next_read_indexz+BufferedSharedMemoryChannel.next_read_indexs $$r4rOr)r5r6r7r8rrr^r_rr r9rPr r<r3rrrrrrrrpropertyrrr:r4r(rrsu.7; ""./"#5)@#)E#FG" " eC!112 3 """"&111111 8 83 8% 8D 8 8 8 8HUOs 7 7huo 7 7 7 7&&X&%%X%%%r4rceZdZdZ ddeejjdee de fde dee d ee ej efd eeed ed efd Zdejjde fdZddZddZdZde fdZddedeeddfdZddeedefdZddeefdZde fdZddZdS)rMat Can be used to send data to different readers via different channels. For example, if the reader is in the same worker process as the writer, the data can be sent via IntraProcessChannel. If the reader is in a different worker process, the data can be sent via shared memory channel. Args: writer: The actor that may write to the channel. None signifies the driver. reader_and_node_list: A list of tuples, where each tuple contains a reader actor handle and the node ID where the actor is located. num_shm_buffers: The number of shared memory buffers per channel. Note: In the case of multiple nodes, we only support 1 shared memory buffer. driver_actor_id: If this channel is read by a driver and that driver is an actual actor, this will be the actor ID of that driver actor. NFrGrIrHr=rJ _channel_dict _channelsrXrYc ||_||_||_||_||_||_|pi|_|p t|_|jrdStj |j|j\} } t| } | dkrNtd} |j | ||j} | |j| <tjtj|j| \}}t|dkrWt#|j||}|j ||D]$\}}||} ||j| <%t|dkrVt%|j|}|j ||D]&\}}||} ||j| <%dSdS)Nrr@) num_readers)rarbrD_driver_actor_idrXrYrsetrr split_readers_by_localityr\radd _get_actor_idsplit_actors_by_node_localityget_actor_noderrK)r%rGrIr=rJrrrXrYremote_reader_and_node_listlocal_reader_and_node_listnum_local_readers local_channelactor_idreaders_same_nodereaders_different_noderemote_channelrqrrs r(r3zCompositeChannel.__init__s %9" / /"4"4*0b"+cee >  F  +DL$:T U U ' & :;; q 0A>>>M N  } - - -))$,77H+8D x (  /   . .0K    "  ! !Q & &8 /N N  ~ . . .. > > --f55/="8,, % & &! + +$T\3IJJN N  ~ . . .3 > > --f55/="8,, , + > >r4rqrc4|jSrO)r~hex)r%rqs r(rzCompositeChannel._get_actor_ids##%%%r4cb|jrdS|jD]}|d|_dSr)rXrrr%channels r(rz,CompositeChannel.ensure_registered_as_writerF  "  F~ 2 2G  / / 1 1 1 1"&r4cb|jrdS|jD]}|d|_dSr)rYrrrs r(rz,CompositeChannel.ensure_registered_as_readerrr4c tt|j|j|j|j|j|j|j|jffSrO) rMrarbrDrrrrXrYrs r(rzCompositeChannel.__reduce__sB L  &  !  !   N  #  # "  r4c,dd|jDdS)NzCompositeChannel(_channels=c,g|]}t|Sr:)r9)rrs r(rz,CompositeChannel.__str__..s;;;G ;;;r4r)rrs r(rzCompositeChannel.__str__s. ?;;DN;;; ? ? ? r4r&rcn||jD]}|||dSrO)rrr)r%r&rrs r(rzCompositeChannel.writesF ((***~ * *G MM% ) ) ) ) * *r4c||j||SrO)rr_resolve_actor_idrrs r(rzCompositeChannel.reads< ((***!$"8"8":":;@@IIIr4c||j||dSrO)rrr rrs r(rzCompositeChannel.release_buffer sB ((*** 411334CCGLLLLLr4ctj}| ||jkrBt |jdksJ|jdd}||}|S)Nr@r)rrj get_actor_idrr\rbr)r%rdriver_proxy_actors r(r z"CompositeChannel._resolve_actor_ids{*,,99;;  x4+@@@t122a7777!%!;A!>q!A ))*<==Hr4cB|jD]}|dSrO)rrrs r(rzCompositeChannel.closes,~  G MMOOOO  r4)NNNFFrrO)r5r6r7r8rrr^r_rr r9rPrrrr rr3rrrrrrrrrrr rr:r4r(rMrMs,*.GK59#(#(B>B>./B>#5)@#)E#FGB> B> "# B>  S[2B%B CD B>C 012B>!B>!B>B>B>B>H&CI$9&c&&&&''''''''         **3*%*D**** JJHUOJsJJJJMMhuoMMMM 3    r4rM))rloggingr collectionsrrtypingrrrrr r r rray.exceptions ray._rayletr ray.experimental.channelr ray.experimental.channel.commonrr.ray.experimental.channel.intra_process_channelrray.experimental.channel.utilsrray.util.annotationsrr getLoggerr5r#rPr)r*r0r<rKrrMr:r4r(rs  ////////?????????????????? ((((((******OOOOOOOONNNNNN99999988888888  8 $ $&&&&&&Z NNN GGGGGGGG&8 8 8 8 8 (8 8 8 v W{{{{{{{{| n%n%n%n%n%"2n%n%n%b WTTTTT'TTTTTr4