&`iDddlZddlZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl Z ddl mZddlmZmZddlmZdZdZeddeedfd ed ed ed ed ede edede edede de e je je jfe e jddfffdZ d-ded ed ede edede e jf dZddeedfd ed ed ed ed ede edede edede de e je je jfffdZdedeee eefffdZe j eZ! d.dede ede eee eeffdej"fdZ#dedefdZ$e j GddZ%d e jdee&ddffd!Z'd"d#d$ej"d%ed&eddfd'Z(e j d(e jd%eddfd)Z)dd"d*ded%ede ed&eddf d+Z*e j e*Z+d efd,Z,dS)/N)Dict GeneratorListOptionalTupleUnion) TempFileLock)_force_on_node_get_node_id_from_node_ip) DeveloperAPIi@@F source_ip source_path target_ip target_path force_allexcludechunk_size_bytesmax_size_bytesreturn_futuresreturnc ||krt||||||||| S||krt|||||} |r| ddfS| SdS)aDSynchronize directory on source node to directory on target node. Per default, this function will collect information about already existing files in the target directory. Only files that differ in either mtime or filesize will be transferred, unless ``force_all=True``. If ``source_ip==target_ip``, shutil will be used to copy the directory. Otherwise, the directory will be packed and sent through the Ray Object Store to the target node. Args: source_ip: IP of source node. source_path: Path to directory on source node. target_ip: IP of target node. target_path: Path to directory on target node. force_all: If True, all files will be transferred (not just differing files). Ignored if ``source_ip==target_ip``. exclude: Pattern of files to exclude, e.g. ``["*/checkpoint_*]`` to exclude trial checkpoints. chunk_size_bytes: Chunk size for data transfer. Ignored if ``source_ip==target_ip``. max_size_bytes: If packed data exceeds this value, raise an error before transfer. If ``None``, no limit is enforced. Ignored if ``source_ip==target_ip``. return_futures: If True, returns a tuple of the unpack future, the pack actor, and the files_stats future. If False (default) will block until synchronization finished and return None. Returns: None, or Tuple of unpack future, pack actor, and files_stats future. If ``source_ip==target_ip``, pack actor and files_stats future will be None. ) rrrrrrrrr)iprrrrN)!_sync_dir_between_different_nodes_sync_dir_on_same_node) rrrrrrrrrrets p/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/tune/utils/file_transfer.pysync_dir_between_nodesrsbI0##-))      # #$##)      #d? "  $ #rct|}tjdddit|}||||}|r|St j|S)aHSynchronize directory to another directory on the same node. Per default, this function will collect information about already existing files in the target directory. All files will be copied over. Args: ip: IP of the node. source_path: Path to source directory. target_path: Path to target directory. exclude: Pattern of files to exclude, e.g. ``["*/checkpoint_*]`` to exclude trial checkpoints. return_futures: If True, returns a future of the copy task. Returns: None, or future of the copy task. num_cpusr) source_dir target_dirr)r _remote_copy_diroptionsr remoterayget)rrrrrnode_id copy_on_node copy_futures rrru2(++G#+RRQR.:Q:QRRL%%;&K 7;  rc t|} t|} tjdddit| } t jdddit| } |rd} n3t jdddit| |} | || |||}| ||}|r||| fStj|S)aSynchronize directory on source node to directory on target node. Per default, this function will collect information about already existing files in the target directory. Only files that differ in either mtime or filesize will be transferred, unless ``force_all=True``. Args: source_ip: IP of source node. source_path: Path to directory on source node. target_ip: IP of target node. target_path: Path to directory on target node. force_all: If True, all files will be transferred (not just differing files). exclude: Pattern of files to exclude, e.g. ``["*/checkpoint_*]`` to exclude trial checkpoints. chunk_size_bytes: Chunk size for data transfer. max_size_bytes: If packed data exceeds this value, raise an error before transfer. If ``None``, no limit is enforced. return_futures: If True, returns a tuple of the unpack future, the pack actor, and the files_stats future. If False (default) will block until synchronization finished and return None. Returns: None, or Tuple of unpack future, pack actor, and files_stats future. r!rN)r" files_statsrrrr$) r _PackActorr&r _unpack_from_actor%_remote_get_recursive_files_and_statsr'r(r))rrrrrrrrrsource_node_idtarget_node_idpack_actor_on_source_nodeunpack_on_target_noder. pack_actor unpack_futures rrrs8J/y99N.y99N * 2!!!$^44!!/6$^44 ;C   (88  &   +11)% 2J*00[IIM6j+55 7= ! !!rpathcvi}tj|dD]\}}}tj||}|D]v} tj||}tjtj||}|j|jf||<g#t$rYswxYw|S)aTReturn dict of files mapping to stats in ``path``. This function scans a directory ``path`` recursively and returns a dict mapping each contained file to a tuple of (mtime, filesize). mtime and filesize are returned from ``os.lstat`` and are usually a floating point number (timestamp) and an int (filesize in bytes). Ftopdown) oswalkr8relpathjoinlstatst_mtimest_sizeFileNotFoundError) r8r.rootdirsfilesrel_rootfilekeystats r_get_recursive_files_and_statsrKsKWT5999  dE7??4..  D gll8T22x T3 7 788#'=$,#> C  $       sA#B(( B54B5r"r.c dtdtffd }tj}t j|dtj5}|ss||ddnm|pi}||dd tj |d D]9\}}}tj ||} |D]X} tj | | } |tj || | d Y|D]} tj | | } tj tj || } | j| jf}|| rn| |vr || |kr|tj || | ;d d d n #1swxYwY|S) a_Pack whole directory contents into an uncompressed tarfile. This function accepts a ``files_stats`` argument. If given, only files whose stats differ from these stats will be packed. The main use case for this is that we can collect information about files already existing in the target directory, and only pack files that have been updated. This is similar to how cloud syncing utilities decide which files to transfer. Args: source_dir: Path to local directory to pack into tarfile. exclude: Pattern of files to exclude, e.g. ``["*/checkpoint_*]`` to exclude trial checkpoints. files_stats: Dict of relative filenames mapping to a tuple of (mtime, filesize). Only files that differ from these stats will be packed. Returns: Tarfile as a stream object. candidatercJsdSD]}tj||rdSdS)NFT)fnmatch)rMexclrs r_should_excludez"_pack_dir.._should_excludesD 5  Dy$// tt urw)fileobjmodeformatT)arcname recursiveFr:)rWN)strboolioBytesIOtarfileopen PAX_FORMATaddr<r=r8r>r?r@rArB)r"rr.rQstreamtarrDrErFrGdirrIrHrJ file_stats ` r _pack_dirresO634Z\\F f3w7I J J JHc H7 H GGJdG ; ; ; ;%+K GGJeG < < <%'WZ%G%G%G H H!dE7??4<<YYC',,x55CGGBGLLS993RWGXXXX! H HD',,x66C8BGLLS$A$ABBD $ t| ;I&s++! k))k#.>).K.K GGBGLLS993GGGGG H!HHHHHHHHHHHHHHH> Ms F G""G&)G& num_bytesc.t|dz ddS)Nr z.2fGiB)float)rfs r _gib_stringrj*s!I )** 3 3 3 33rc eZdZdZddeefdedeedee ee e e ffde dee f dZ d efd Zd eeddffd Zd eefd ZdS) r/aActor wrapping around a packing job. This actor is used for chunking the packed data into smaller chunks that can be transferred via the object store more efficiently. The actor will start packing the directory when initialized, and separate chunks can be received by calling the remote ``next()`` task. Args: source_dir: Path to local directory to pack into tarfile. exclude: Pattern of files to exclude, e.g. ``["*/checkpoint_*]`` to exclude trial checkpoints. files_stats: Dict of relative filenames mapping to a tuple of (mtime, filesize). Only files that differ from these stats will be packed. chunk_size_bytes: Cut bytes stream into chunks of this size in bytes. max_size_bytes: If packed data exceeds this value, raise an error before transfer. If ``None``, no limit is enforced. Nr"rr.rrc <t||||_|jdd|j}|r9||kr3t d|dt |dt |d||_||_d|_dS)N)r"rr.rzPacked directory z content has a size of z, which exceeds the limit of z. Please check the directory contents. If you want to transfer everything, you can increase or disable the limit by passing the `max_size` argument.) reraseektell RuntimeErrorrj chunk_sizemax_sizeiter)selfr"rr.rr file_sizes r__init__z_PackActor.__init__Ds !7     AK$$&&  i.88LJLLy))LL!.11LLL +&  rrc4|jSN)ragetvaluerts r get_full_dataz_PackActor.get_full_data`s{##%%%rc#K|jd|j|j}|r'|V|j|j}|%dSdS)Nr)rarnreadrq)rtdatas r_chunk_generatorz_PackActor._chunk_generatorcsy {00 5JJJ;##DO44D 5 5 5 5 5rc|js&t||_ t|jS#t$rYdSwxYwrx)rsrnext StopIterationrzs rrz_PackActor.nextjs\y 6T224455DI  ?? "   44 sA AA)__name__ __module__ __qualname____doc___DEFAULT_CHUNK_SIZE_BYTES_DEFAULT_MAX_SIZE_BYTESrYrrrrriintrvbytesr{rrrr$rrr/r/.s.#'>B 9(? $d3eSj(9#9:;    ! 8&u&&&&5)E4,=">5555huorr/actorc#pK tj|j}|dS|V4)z0Iterate over actor task and return as generator.TN)r(r)rr')rbuffers r _iter_remoterss@**,,-- > F rT_retryrar#rcX|dtj|} t |dd5t j|5}||dddn #1swxYwYddddS#1swxYwYdS#t$rxt |d5 dddn #1swxYwYtj |s*|rt||dYdStd|d YdSwxYw) z,Unpack tarfile stream into target directory.r.locktimeout)rSNFrTarget directory u does not exist and couldn't be recreated. Please raise an issue on GitHub: https://github.com/ray-project/ray/issues) rnr<r8normpathr r]r^ extractall TimeoutErrorexists _unpack_dirrp)rar#rrbs rrr|s) KKNNN!!*--JZ... : : : + +f--- +z*** + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +  Z... / /                  w~~j))  FJu======="@ @@@    sB' B B6 BB B B B B'BB'!B"B''D)C D)C D)C 6D)D)(D)r6ctj}t|D]}||t ||dS)z8Iterate over chunks received from pack actor and unpack.)r#N)r[r\rwriter)r6r#rars rr0r0sTZ\\Fz** V:......r)rrctj|} t|dd5t |d}rfd}|}t j||ddddS#1swxYwYdS#t$rxt|d5 dddn #1swxYwYtj|s*|rt|dYdStd |d YdSwxYw) z"Copy dir with shutil on the actor.rrrNct}tj|}|D]S}tj||}D].}t j||r||n/T|Srx)setr<r8r>r?rOr`) r8names ignored_namesrel_pathnamerMrPrr"s r_ignorez_copy_dir.._ignores$'EEM!wtZ@@H %&&$&GLL4$@$@ $+&&D&y$??& - 1 1$ 7 7 7 %&)(r)ignoreFrrr) r<r8rr _delete_path_unsafeshutilcopytreerr _copy_dirrp)r"r#rr _ignore_funcrs` ` rrrs!!*--J'Z... : : : I I  + + +L ' ) ) ) ) ) ) ' OJ < H H H H' I I I I I I I I I I I I I I I I I I(  Z... / /                  w~~j))  *j???????"@ @@@    s^B3A7* B7A;;B>A;?BD B." D.B2 2D5B2 66D/DDctj|rJtj|rt j|ntj|dSdS)z1Delete path (files and directories). No filelock.TF)r<r8risdirrrmtreeremove)rs rrrs[ w~~k"" 7== % % # M+ & & & & Ik " " "t 5r)NF)NN)-rOr[r<rr]typingrrrrrrr(ray.air._internal.filelockr ray.air.util.noder r ray.util.annotationsr rrrYrZr ObjectRefActorIDrrrrirKr'r1r\rerjr/rrrr0rr%rr$rrrs @@@@@@@@@@@@@@@@ 333333GGGGGGGG-------0 "5$; FFFFF F  F d^ FFSMFF  #-cm 34 #-t #$&FFFFZ# # #  # # # d^ #  # cm # # # # V"5$; B"B"B"B"B" B"  B" d^ B"B"SMB"B" 4s}ck3=@A ABB"B"B"B"Jc5;L6L1M4)3 3Q(R(R% #:>DDD d^D$sE%*$5567DZ DDDDN454S4444AAAAAAA AH  %t2C(DHL PT:/3;/C/D/// /# 00000d^ 0  0  0000h3:i((Sr