%`i`2ddlZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z mZmZddlZddlmZddlmZe ddZejeZded ed ed efd Zd ed eeeeffdZGddejZGddeZGddeZ GddeZ!ea"Gdde Z#Gdde Z$dZ%dZ&dZ'de ed e efd!Z(d"e efd#Z)de efd$Z*dS)%N) namedtuple)IOListOptionalTupleUnion)DEFAULT_OBJECT_PREFIX) ObjectRef ParsedURLzbase_url, offset, sizeurloffsetsizereturnc|d|d|S)atMethods to create a URL with offset. When ray spills objects, it fuses multiple objects into one file to optimize the performance. That says, each object needs to keep tracking of its own special url to store metadata. This method creates an url_with_offset, which is used internally by Ray. Created url_with_offset can be passed to the self._get_base_url method to parse the filename used to store files. Example) file://path/to/file?offset=""&size="" Args: url: url to the object stored in the external storage. offset: Offset from the beginning of the file to the first bytes of this object. size: Size of the object that is stored in the url. It is used to calculate the last offset. Returns: url_with_offset stored internally to find objects from external storage. z?offset=z&size=r r rs q/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/_private/external_storage.pycreate_url_with_offsetrs 4 / /6 / / / //url_with_offsetctj|}tj|j}|dd}d|vsd|vrtd|t|dd}t|dd}t|||S)aParse url_with_offset to retrieve information. base_url is the url where the object ref is stored in the external storage. Args: url_with_offset: url created by create_url_with_offset. Returns: named tuple of base_url, offset, and size. ?rr rzFailed to parse URL: )base_urlr r) urllibparseurlparseparse_qsquerygeturlsplit ValueErrorintr )r parsed_result query_dictrr rs rparse_url_with_offsetr%1sL))/::M&&}':;;J##%%++C003Hz!!V:%=%=BBBCCC H%a( ) )F z&!!$ % %D hvD A A AArc TeZdZdZdZdZdZeddZdZ d Z d e d e e d e ed ede ef dZdZejde efdZejd e e de edefdZejde efdZejdZdS)ExternalStorageaThe base class for external storage. This class provides some useful functions for zero-copy object put/get from plasma store. Also it specifies the interface for object spilling. When inheriting this class, please make sure to implement validation logic inside __init__ method. When ray instance starts, it will instantiating external storage to validate the config. Raises: ValueError: when given configuration for the external storage is invalid. cd|_dSN) _core_workerselfs r__init__zExternalStorage.__init__[sAErrray._raylet.CoreWorkerc:|jtjjj}t j}|jsRt jdt j|z |jkrtd|jd|jR|j |_|jS)a*Get the core_worker initialized in this process. In rare cases, the core worker may not be fully initialized by the time an I/O worker begins to execute an operation because there is no explicit flag set to indicate that the Python layer is ready to execute tasks. NgMbP?z4CoreWorker didn't initialize within grace period of zs.) r,ray_privateworker global_workertime connectedsleepCORE_WORKER_INIT_GRACE_PERIOD_S RuntimeError core_worker)r.r4starts rr;zExternalStorage.core_worker`s   $\(6FIKKE&  5!!!9;;&)MMM&D?DDD& !' 2D   rc6|j|Sr+)r; get_if_local)r. object_refss r_get_objects_from_storez'ExternalStorage._get_objects_from_storews ,,[999rcB|j|||||dSr+)r;put_file_like_object)r.metadata data_size file_like object_ref owner_addresss r_put_object_to_storez$ExternalStorage._put_object_to_store~s4 -- iJ      rfr?owner_addressesr c8g}d}||}t|||D]Y\}\} } } } t| } t| }| :t| dkr'd|d}t || dnt| }| dd|ddz|ddz| z| z|rt | ndz}t|}|j| z|z|z|ksJ||}||ksJt||| }| | ||z }[| |S) aFuse all given objects into a given file handle. Args: f: File handle to fusion all given object refs. object_refs: Object references to fusion to a single file. owner_addresses: Owner addresses for the provided objects. url: url where the object ref is stored in the external storage. Return: List of urls_with_offset of fused objects. The order of returned keys are equivalent to the one with given object_refs. rNzObject z does not exist.little byteorderrr) r@ziplenhexr!to_bytes memoryview HEADER_LENGTHwriterappendencodeflush)r.rIr?rJr keysr ray_object_pairsrefbufrC_rG address_len metadata_lenerrorbuf_lenpayload payload_len written_bytesrs r_write_multiple_objectsz'ExternalStorage._write_multiple_objectss"77 DD69 )?7 7  $ $ 2C##x]m,,Kx==L{s8}}11=#''))=== ''';aaCHHG$$Q($;;''X'>>?""1"99:    '.6:c???3 8 g,,K"[0<?'I[XXXXGGG,,M K////4]O KK..00 1 1 1 m #FF   rc^||z|z|jz}||krtd|d|ddS)aCheck whether or not the obtained_data_size is as expected. Args: address_len: Length of the address. metadata_len: Actual metadata length of the object. buffer_len: Actual buffer length of the object. obtained_data_size: Data size specified in the url_with_offset. Raises: ValueError: If obtained_data_size is different from address_len + metadata_len + buffer_len + 24 (first 8 bytes to store length). zObtained data has a size of z., although it is supposed to have the size of .N)rUr!)r.r_r` buffer_lenobtained_data_sizedata_size_in_bytess r _size_checkzExternalStorage._size_checksj , & 3d6H H  !3 3 31/A11-111  4 3rcdS)aNSpill objects to the external storage. Objects are specified by their object refs. Args: object_refs: The list of the refs of the objects to be spilled. owner_addresses: Owner addresses for the provided objects. Returns: A list of internal URLs with object offset. Nrr.r?rJs r spill_objectszExternalStorage.spill_objectsrurl_with_offset_listcdS)aRestore objects from the external storage. Args: object_refs: List of object IDs (note that it is not ref). url_with_offset_list: List of url_with_offset. Returns: The total number of bytes restored. Nrr.r?rqs rrestore_spilled_objectsz'ExternalStorage.restore_spilled_objectsrprurlscdS)zDelete objects that are spilled to the external storage. Args: urls: URLs that store spilled object files. NOTE: This function should not fail if some of the urls do not exist. Nrr.rus rdelete_spilled_objectsz&ExternalStorage.delete_spilled_objectsrprcdS)zDestroy external storage when a head node is down. NOTE: This is currently working when the cluster is started by ray.init Nrr-s rdestroy_external_storagez(ExternalStorage.destroy_external_storagerprN)rr0)__name__ __module__ __qualname____doc__rUr9r/propertyr;r@rHrrr strrfrlabcabstractmethodror"rtrxrzrrrr'r'Hs  M&'#EEE !!!X!,:::   33"&y/3DHI3TW3 c3333j.   T#Y          ?  BFs)          49          rr') metaclasscNeZdZdZdeefdZdZdeefdZdZ dS) NullStoragez>-00  >$>>>  ! ( ( 7 7 7 74())S-@-@@@@@)/(8C@U>O1/BBM$-H")Fh%% v!nnQVVAYY(nKK "~~affQii8~LL ..h.GG  lG]EWXXX  !{ 3 3 66,//))gq*m                s1DFF F ruc|D]M}t|j} tj|>#t $rYJwxYwdSr+)r%rrrremoveFileNotFoundError)r.rur rs rrxz(FileSystemStorage.delete_spilled_objectslsg  C(66?D  $$       sA A AcD|jD]}||dSr+)r_destroy_external_storager.rs rrzz*FileSystemStorage.destroy_external_storageus5"3 ; ;N  * *> : : : : ; ;rctj|rm tj|n6#t $rYn*t $rtdYdSwxYwtj|kdSdS)NzuError cleaning up spill files. You might still have remaining spilled objects inside `ray_spilled_objects` directory.) rrisdirshutilrmtreer Exceptionlogger exceptionrs rrz+FileSystemStorage._destroy_external_storageys gmmN++   n----%        F   gmmN++     s6 A)#A)(A)r+)r{r|r}r~rrrrr"r/ror rtrxrzr __classcell__rs@rrrs&* *X*X*Xc49n-*Xc] *X*X*X*X*X*XX VT#Y V V V V ?BFs)649;;;rrceZdZdZ ddedepedeffd Zdeefd Z d ee d eefd Z d eefdZ dZ xZS)ExternalStorageSmartOpenImplaThe external storage class implemented by smart_open. (https://github.com/RaRe-Technologies/smart_open) Smart open supports multiple backend with the same APIs. To use this implementation, you should pre-create the given uri. For example, if your uri is a local file path, you should pre-create the directory. Args: uri: Storage URI used for smart open. prefix: Prefix of objects that are stored. override_transport_params: Overriding the default value of transport_params for smart-open library. Raises: ModuleNotFoundError: If it fails to setup. For example, if smart open library is not downloaded, this will fail. Nrurioverride_transport_paramsct ddlm}n$#t$r}t d|d}~wwxYw| Jdt |t r|g}t |ts Jdt |ts Jdd|D}t||_ |j s!t|r Jd||_ nd |D|_ t|j t|ksJtjdt|j |_t"d ||_|pi|_|j r0ddl}|d |_d |j|d|_ni|_|j|jdS)Nrrz{Smart open is chosen to be a object spilling external storage, but smart_open and boto3 is not downloaded. Original error: z.uri should be provided to use object spilling.z/uri must be a single string or list of strings.rc8g|]}|dS)zs3://) startswith.0us r z9ExternalStorageSmartOpenImpl.__init__..s$888qQ\\'**888rz'all uri's must be s3 or none can be s3.c8g|]}|dS)/)striprs rrz9ExternalStorageSmartOpenImpl.__init__..s"4441!''#,,444rr^s3) service_nameT) defer_seekresourcer)rr/ smart_openrModuleNotFoundErrorrrrr"all is_for_s3any_urisrQrr_current_uri_indexr prefixrboto3rrtransport_paramsupdate) r.rrrrre uri_is_s3rrs rr/z%ExternalStorageSmartOpenImpl.__init__s   ' ' ' ' ' ' '"   %:67::   P c3   %C#t$$WW&WWW$+s++NN-NNN+88C888 Y~ 59~~ P P'P P P%DJJ44444DJ4:#c((****"("21c$*oo"F"F.:::: )B)Hb& > ' LLL nn$n77DG # G*%%D ! ! %'D ! $$T%CDDDDDs* A AA rcnt|dkrgSddlm}|jdzt|jz|_|j|j}|jdt |}|d|}||d|j5}|||||cdddS#1swxYwYdS)Nrrr)-rr)moder) rQrrrrrrrrf)r.r?rJrrkeyr rEs rroz*ExternalStorageSmartOpenImpl.spill_objectss* {  q I######$(#:Q#>#dj//"Qj01HH9+FFHHnnsnn T !2    //;                   sB**B.1B.r?rqc :ddlm}d}tt|D]t}||}||}t |}|j} |j} || d|j5} | | t | dd} t | dd} t | dd}| | | ||j| | }||z }| | }|||| ||dddn #1swxYwYv|S)Nrrr)rrLrMrN)rrrrQrr%rr rrr"rrrlrrH)r.r?rqrrrrFrr#rr rIr_r`rbrGrCs rrtz4ExternalStorageSmartOpenImpl.restore_spilled_objectss $#####s;''((  A$QJ215<<>>O2/BBM$-H")Fht7LMMM QRv!nnQVVAYY(nKK "~~affQii8~LL ..h.GG  lG]EWXXX !{ 3 3  66,//))gq*m                s:DFF F rucdSr+rrws rrxz3ExternalStorageSmartOpenImpl.delete_spilled_objects rcdSr+rr-s rrzz5ExternalStorageSmartOpenImpl.destroy_external_storagerr)Nr)r{r|r}r~rrdictr/rror rtrxrzrrs@rrrs2+/ 9E9E9E[D9E$( 9E9E9E9E9E9EvT#Y* ?BFs)@ 49           rrcDeZdZdZdeffd Zdeeffd ZxZS)UnstableFileStoragez/This class is for testing with writing failure.rc Xtj|fi|d|_d|_dS)Ng?g?)rr/ _failure_rate_partial_failure_ratior.rkwargsrs rr/zUnstableFileStorage.__init__!s8++F+++ &)###rrctj|jk}||jk}||jk}|rtd|rXtjt t |}t|d||St||S)Nz1Spilling object failed intentionally for testing.) rrrIOErrorchoicerrQrro)r.r?rJrfailedpartial_failedrrs rroz!UnstableFileStorage.spill_objects&s MOOd0 0T''T88  GMNN N  G eC $4$45566A77((RaR/JJ J77((oFF Fr r{r|r}r~rr/rrorrs@rrrs~99******* GT#Y G G G G G G G G G GrrcDeZdZdZdeffd Zdeeffd ZxZS)SlowFileStoragez/This class is for testing slow object spilling.rc Xtj|fi|d|_d|_dS)Nr))rr/ _min_delay _max_delayrs rr/zSlowFileStorage.__init__6s4++F+++rrctj|j|jz z|jz}tj|t ||Sr+)rrrr6r8rro)r.r?rJdelayrs rrozSlowFileStorage.spill_objects;sM 4?T_#DEW 5ww$$[/BBBrrrs@rrr3s~99 CT#YCCCCCCCCCCrrc| Jd|r|d}|dkrt|fi|dan|dkrt|fi|dann|dkrt|fi|danT|dkrt|fi|dan:|d krt |fi|dan t d |t atS) z3Setup the external storage according to the config.Nznode_id should be provided.type filesystemparamsrmock_distributed_fs unstable_fsslow_fszUnknown external storage type: )r_external_storagerrrr!r)configr session_name storage_types rsetup_external_storagerAs-    =    *f~ < ' ' 1' N NVH=M N N   \ ) ) <!!!(+!!  2 2 2!2' N NVH=M N N   ] * *!4G P Pvh?O P P   Y & & / L L6(;K L L  M|MMNN N'MM rc"tadSr+)rr rrrreset_external_storager_s# rc8t||S)aDSpill objects to the external storage. Objects are specified by their object refs. Args: object_refs: The list of the refs of the objects to be spilled. owner_addresses: The owner addresses of the provided object refs. Returns: A list of keys corresponding to the input object refs. )r ro)r?rJs rrorods  * *; H HHrr?rqc8t||S)zRestore objects from the external storage. Args: object_refs: List of object IDs (note that it is not ref). url_with_offset_list: List of url_with_offset. )r rt)r?rqs rrtrtqs  4 4[BV W WWrruc:t|dS)z|Delete objects that are spilled to the external storage. Args: urls: URLs that store spilled object files. N)r rx)rus rrxrx}s ,,T22222rcVtjjdt|S)zkGenerate a unqiue spill file name. Args: object_refs: objects to be spilled in this file. z-multi-)uuiduuid4rRrQ)r?s rrrs) jll 9 9s;'7'7 9 99r)+rloggingrrrr6rr collectionsrtypingrrrrrr2ray._private.ray_constantsr ray._rayletr r getLoggerr{rrr"rr%ABCMetar'rrrr rrrrrortrxrrrrrs.  """"""33333333333333 <<<<<<!!!!!! J{$< = =  8 $ $03003030000:B3B5c33GBBBB.t t t t t  t t t t n I I I I I/ I I I ||||||||~J J J J J ?J J J Z KMMGGGGG+GGG* C C C C C' C C C<&&& I I I Xi X8