&`iUddlZddlZddlZddlmZmZddlmZddlm Z m Z m Z m Z m Z mZddlmZmZddlmZmZddlmZmZddlmZdd lmZdd lmZdd lmZdd l m!Z!dd l"m#Z#ddl$m%Z%ddl&m'Z'e rddl(m)Z)m*Z*ej+e,Z-eddZ.ee/e0d<eefZ1GddZ2ddde3fdZ4dede/fdZ5GddeZ6Gdde6Z7dS)N)ABCabstractmethod) defaultdict) TYPE_CHECKINGCallableDictIterableListOptional)env_bool env_float)ExecutionOptionsExecutionResources)PhysicalOperatorReportsExtraResourceUsage)AllToAllOperator)HashShufflingOperatorBase)InputDataBuffer) ZipOperator) memory_string)GiB) DataContextlog_once)OpStateTopologyRAY_DATA_DEBUG_RESOURCE_MANAGER1LOG_DEBUG_TELEMETRY_FOR_RESOURCE_MANAGER_OVERRIDEceZdZdZdZeddZdZddded e ge fd e fd Z d Z d ddddefdZdZde fdZde fdZde fdZde fdZd ede fdZd edefdZd edefdZd ededefdZdefdZed)dZd edefd Zd ede e fd!Z!d edefd"Z"de#efd#Z$d ede%efd$Z&d ede%efd%Z'd edefd&Z(d edefd'Z)d(S)*ResourceManagerz@A class that manages the resource usage of a streaming executor.+RAY_DATA_OBJECT_STORE_MEMORY_LIMIT_FRACTION?g?topologyroptionsget_total_resources data_contextc||_||_||_tj|_d|_tj|_tj|_tj|_ i|_ i|_ i|_ tt|_tt|_d|_|jr5t'd|D}|rt)||j|_|j|jn!|r|jn|j|_|dS)Nrc3>K|]}|VdSN)%implements_accurate_memory_accounting).0ops /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/execution/resource_manager.py z+ResourceManager.__init__..`s?  ?A88::      ) _topology_options_get_total_resourcesrzero_global_limits_global_limits_last_update_time _global_usage_global_running_usage_global_pending_usage _op_usages_op_running_usages_op_pending_usagesrint_mem_op_internal_mem_op_outputs_op_resource_allocatorop_resource_reservation_enabledallReservationOpResourceAllocatorop_resource_reservation_ratio+override_object_store_memory_limit_fractionop_resource_allocator_enabled*DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION9DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION_NO_RESERVATION#_object_store_memory_limit_fraction)_warn_about_object_store_memory_if_needed)selfr$r%r&r' should_enables r.__init__zResourceManager.__init__Asd" $7!0577/0,/466%7%<%>%>"%7%<%>%>"FHNPNP>I=M=M=Hr?)rKr-r^mem_op_internalmem_op_outputsnext_ops r._estimate_object_store_memoryz-ResourceManager._estimate_object_store_memorys b/ * * 1*GL12:EE1133-  G >/CD NN %4b!#1R //r0ctddd|_tddd|_tddd|_|j|j|jt|j D]{\}}| | }| }|}|jrJ|jrJ|jrJ|||}||}||}t%|t&r'||||j|<||j|<||j|<|j||_|j||_|j||_|j|j_}|j"|j|jdSdS)zRecalculate resource usages.rrQN)limits)rr7r8r9r:clearr;r<reversedr1itemsupdate_resource_usagecurrent_processor_usagerunning_processor_usagepending_processor_usagerQrlcopyraraddextra_resource_usage_metricsobj_store_mem_usedr@update_budgetsr5)rKr-r^op_usageop_running_usageop_pending_usageused_object_stores r. update_usageszResourceManager.update_usagessn 01a88%71a%@%@"%71a%@%@"  %%''' %%'''"$."6"6"8"899& J& JIB  $ $ & & &1133H!99;; !99;; 3 3 33'; ; ;;'; ; ;; $ B B2u M M }}9J}KKH/44$5 5   "788 8 R4466777"*DOB *:D #B '*:D #B '"&!3!7!7!A!AD )-)C)G)G **D &*.)C)G)G **D & .6-IBK * *  & 2  ' 6 6* 7      3 2r0c|jS)z5Return the global resource usage at the current time.)r7rKs r.get_global_usagez ResourceManager.get_global_usages !!r0c|jS)z=Return the global running resource usage at the current time.)r8rs r.get_global_running_usagez(ResourceManager.get_global_running_usage ))r0c|jS)z=Return the global pending resource usage at the current time.)r9rs r.get_global_pending_usagez(ResourceManager.get_global_pending_usagerr0ctj|jz |jkr|jStj|_|jj}|jj}|}|j}| |j |z}| | ||_|jS)aReturn the global resource limits at the current time. This method autodetects any unspecified execution resource limits based on the current cluster size, refreshing these values periodically to support cluster autoscaling. rn) timer6GLOBAL_LIMITS_UPDATE_INTERVAL_Sr5r2resource_limitsexclude_resourcesr3rIrwrQminsubtract)rKdefault_limitsexcludetotal_resourcesdefault_mem_fractions r.get_global_limitsz!ResourceManager.get_global_limitss IKK$> >2 3 3& &/3y{{,6-13355#G).. / C"!#/  -00AAJJ7SS""r0c|j|S)zDReturn the resource usage of the given operator at the current time.)r:rKr-s r. get_op_usagezResourceManager.get_op_usage sr""r0c|j|S)zFReturn the memory usage of the internal buffers of the given operator.)r>rs r.get_mem_op_internalz#ResourceManager.get_mem_op_internals$R((r0c|j|S)z=Return the memory usage of the outputs of the given operator.)r?rs r.get_mem_op_outputsz"ResourceManager.get_mem_op_outputss#B''r0verbosec ||jvrd}nh|j|jdd}|j|jr|d|j|jddz }|d|j|dz }tt}|rH|dt ||d t ||d z }|j|j |}|r7|d |jdz }|d |jdz }|d |d z }|j |}|r|d|jdz }|d |jdz }|d |z }t|jtr6t |jj |d}|d |d z }|S)zbReturn a human-readable string representation of the resource usage of the given operator.zn/arSz CPUz, z GPUz object storeNz (in=z,out=)z , alloc=(cpu=z,gpu=z ,obj_store=z, budget=(cpu=r)r;cpugpuobject_store_memory_strrrrrr@get_allocation get_budgetrarC_output_budgetsrY)rKr-r usage_str allocationbudgetreserved_for_outputs r.get_op_usage_strz ResourceManager.get_op_usage_strs_ T, , ,II226:DDDDI&r*. LK$"9""="AKKKKK bd5b9QQSSbbb bI = HGG  D E d&>&>r&B&BCCEE$T%<%g|]}||S)r)r,r-rKs r. z4ResourceManager.get_eligible_ops..fs,GGGrt/B/B2/F/FGGGGr0)r1rs`r.get_eligible_opsz ResourceManager.get_eligible_opsesGGGGT^GGGGr0c#K|jD]6}||s|V||Ed{V7dS)zGet the downstream ineligible operators of the given operator. E.g., - "cur_map->downstream_map" will return an empty list. - "cur_map->limit1->limit2->downstream_map" will return [limit1, limit2]. N)rfrget_downstream_ineligible_opsrKr-rks r.rz-ResourceManager.get_downstream_ineligible_opshso- G GG&&w// G ==gFFFFFFFFF G Gr0c#K|jD]7}||r|V||Ed{V8dSa Get the downstream eligible operators of the given operator, ignoring intermediate ineligible operators. E.g., - "cur_map->downstream_map" will return [downstream_map]. - "cur_map->limit1->limit2->downstream_map" will return [downstream_map]. N)rfrget_downstream_eligible_opsrs r.rz+ResourceManager.get_downstream_eligible_opsvss- E EG""7++ E ;;GDDDDDDDDDD  E Er0cj|}|tfd|Dz }|S)zsGet the outputs memory usage of the given operator, including the downstream ineligible operators. c3LK|]}|jVdSr*)rrQ)r,rkrKs r.r/zTResourceManager.get_op_outputs_object_store_usage_with_downstream..sH     g & & :      r0)r?sumr)rKr-op_outputs_usages` r.1get_op_outputs_object_store_usage_with_downstreamzAResourceManager.get_op_outputs_object_store_usage_with_downstreamsf /3C    ==bAA       r0c>td|jDS)z>Check if the operator has a downstream materializing operator.c3@K|]}t|tVdSr*raMATERIALIZING_OPERATORSr,rks r.r/zBResourceManager.has_materializing_downstream_op..sA   w 7 8 8      r0)anyrfrs r.has_materializing_downstream_opz/ResourceManager.has_materializing_downstream_ops3  1      r0N)r_r)*__name__ __module__ __qualname____doc__rr rGrHrrrrrMrJr=rlrrrrrrrrrboolstrrrFpropertyrrr rrr rr rrrrrr0r.r r /sRJJ'(# 2;5s22.AE=1919"19&b*<&<= 19 " 19191919f:0$0-60 00006888t""4""""**<******<****##5####2#/#4F####)&6)3))))(%5(#((((+#3++#++++Z7t7777+++X+  0@ S    :-:(;M2N:::: !1 d    H$'7"8HHHH G" G " # G G G GE"E " #EEEE  "        2B t      r0r r$rr_c~t|D],\}}t|r|s|cS-dS)N) enumerate_is_shuffle_op completed)r$idxr-s r._get_first_pending_shuffle_oprsKX&&R "   bllnn JJJ 2r0r-c:t|ttfSr*)rarrr-s r.rrs b+-FG H HHr0c eZdZdZGddZddZedefdZed e d e fd Z ed e d e e efd e e e fd ee fdZed e d eefdZed e d ee fdZed e d eefdZd ee fdZed e d e fdZd e d ee fdZd e d e fdZdS)raAn interface for dynamic operator resource allocation. This interface allows dynamically allocating available resources to each operator, limiting how many tasks each operator can submit, and how much data each operator can read from its running tasks. cPeZdZdZdZdZdZdZdefdZ e dede fd Z d S) OpResourceAllocator.IdleDetectoraUtility class for detecting idle operators. Note, stalling can happen when there are less resources than Data executor expects. E.g., when some resources are preempted by non-Data code, see `test_no_deadlock_on_resource_contention` as an example. This class is used to detect potential stalling and allow the execution to make progress. g$@gN@Fctt|_td|_td|_dS)Nc(tjSr*rrr0r.z;OpResourceAllocator.IdleDetector.__init__..s  r0c(tjSr*rrr0r.rz;OpResourceAllocator.IdleDetector.__init__..s 49;;r0)rr=last_num_outputslast_output_timelast_detection_timers r.rMz)OpResourceAllocator.IdleDetector.__init__s@$/$4$4D !$/0C0C$D$DD !'23F3F'G'GD $ $ $r0r-c6tj}||j|z |jkrl|jj}||j|kr||j|<||j|<||j|<n0||j|<||||j|z dSdSNTF)rrDETECTION_INTERVAL_Srbnum_task_outputs_generatedrr"print_warning_if_idle_for_too_long)rKr-cur_timecur_num_outputss r. detect_idlez,OpResourceAllocator.IdleDetector.detect_idlesy{{H$22669RRR"$*"G"T%:2%>>>0?D)"-08D)"-3;D,R003;D,R0;;Ht'.(sX   R##B'' -22666  766r0)rr1r)rKr s`@r._get_eligible_opsz%OpResourceAllocator._get_eligible_ops&sN'DT^'T'T$     $T^44    r0cV| o| Srrrs r.r z#OpResourceAllocator._is_op_eligible2s4&&(( ( 0--///  r0c#K|jD]7}||r|V||Ed{V8dSr)rfr _get_downstream_eligible_opsrs r.rz0OpResourceAllocator._get_downstream_eligible_ops<ss- F FG##G,, F <>rBB  M++M:: tt "..}== tt ur0N)r$r)rrrrrrMrrr|rrrrr=r rrrrr r  staticmethodr r rrrr0r.rrsV? ? ? ? ? ? ? ? B2222 #   ^  &6 4   ^     ""24F"FG  $((8#(=#>  #    ^   - (;M2N   ^  $4 #   ^  !1 h?Q6R   ^  4(8#9      ,    \ F"F " #FFFF " r0rc eZdZdZdedeffd ZdeefdZ de fdZ d ede fd Z d edee fd Zd edeefd Zd edee fd Zd edeee fdeeefdeefdZde fdZxZS)rCaAn OpResourceAllocator implementation that reserves resources for each operator. This class reserves memory and CPU resources for eligible operators, and considers runtime resource usages to limit the resources that each operator can use. It works in the following way: 1. An operator is eligible for resource reservation, if it has enabled throttling and hasn't completed. Ineligible operators are not throttled, but their usage will be accounted for their upstream eligible operators. E.g., for such a dataset "map1->limit->map2->streaming_split", we'll treat "map1->limit" as a group and "map2->streaming_split" as another group. 2. For each eligible operator, we reserve `reservation_ratio * global_resources / num_eligible_ops` resources, half of which is reserved only for the operator outputs, excluding pending task outputs. 3. Non-reserved resources are shared among all operators. 4. In each scheduling iteration, each eligible operator will get "remaining of their own reserved resources" + "remaining of shared resources / num_eligible_ops" resources. The `reservation_ratio` is set to 50% by default. Users can tune this value to adjust how aggressive or conservative the resource allocation is. A higher value will make the resource allocation more even, but may lead to underutilization and worse performance. And vice versa. resource_managerreservation_ratiocDt|j||_||_d|jcxkrdksnJi|_i|_tj|_ i|_ i|_ i|_ | |_dS)Ng?)superrMr1_resource_manager_reservation_ratio _op_reserved_reserved_for_op_outputsrr4 _total_shared _op_budgetsr_reserved_min_resourcesrr)rKrr __class__s r.rMz'ReservationOpResourceAllocator.__init__s )3444!1"3d-4444444444HJHJ%/466GI>@ FH$"//11r0r_cg}g}|jD]I}|s3|jD]+}|r||,J|D]Q}|t |j|||Rt t|S)a Resource reservation is based on the number of eligible operators. However, there might be completed operators that still have blocks in their output queue, which we need to exclude them from the reservation. And we also need to exclude the downstream ineligible operators. E.g., for the following pipeline: ``` map1 (completed, but still has blocks in its output queue) -> limit1 (ineligible, not completed) -> map2 (not completed) -> limit2 -> map3 ``` The reservation is based on the number of eligible operators (map2 and map3), but we need to exclude map1 and limit1 from the reservation. ) r1rinput_dependenciesappendextendlistrrset)rKlast_completed_opsops_to_exclude_from_reservationr-deps r._get_ineligible_ops_with_usagez=ReservationOpResourceAllocator._get_ineligible_ops_with_usages *,'. 7 7B,,.. 7077C11337*11#666% 7 7B + 2 2T+II"MMNN    , 2 22 6 6 6 6C788999r0roc |j}|j|j|jt |dkrdS|}||j t |z }t|D]\}}tddt|j dz d}||}|\} } | | | }| || }|||dr d|j|<ncd|j|<tdd| j }|dkr=t'dt)|rt*d|d ||j|<|j |j|<||} || }| tj}||_dS) Nrr!T)ignore_object_store_memoryFlow_resource_warning_z6Cluster resources are not enough to run any task from z8. The job may hang forever unless the cluster scales up.)rrrrprr!lenrwscalerrrmaxrQrmin_max_resource_requirementsrrxsatisfies_limitridrZr[r4r) rKro eligible_ops remainingdefault_reservedindexr-reserved_for_outputsreserved_for_tasksmin_resource_usagemax_resource_usageop_total_reserveds r._update_reservationz2ReservationOpResourceAllocator._update_reservations->>@@  !!! %++--- $**,,, |   ! ! FKKMM "<<(?3|CTCT(UVV"<004 A4 AIE2$61c*>BAFF$$ "2!:!:;O!P!P 575U5U5W5W 2  2!-%7%;%;4,,..>>vFFFr0c6|j|Sr*)r rYrs r.rz)ReservationOpResourceAllocator.get_budgets##B'''r0c6|j|Sr*)rrYrs r.rz0ReservationOpResourceAllocator.get_output_budgets#''+++r0c(tjSr*)rr4rs r.rz-ReservationOpResourceAllocator.get_allocations!&(((r0rrcr||jvrdS|j|j}|j|}|t |j||z dz }t j|r ||j|<dSt|}|dksJ|dkr| |rd}||j|<|S)Nrr!) r rQrrr3rmathisinfrr=r)rKr-rrresrs r.rz.sA7$;<<r0inf)r,rrrr3rr4r@r rprr1rrrrrwrrrqr2rBrxis_zeror5is_non_negativer4rrrfr)rKroop_to_exclude_from_reservation completed_opcompleted_op_usageremaining_sharedr7r- op_mem_usagerr} op_reservedop_reserved_remainingop_reserved_exceededi op_shared to_borrowtarget_num_gpus r.r|z-ReservationOpResourceAllocator.update_budgets3sz *.)L)L)N)N&: ; ;L!%!7!D!D\!R!R __%788FZZ 2 7 9 9::FF 33F;;    ->>@@ |   ! ! F - O OBL D2FFrJJ JL $5gg    C 043PQS3T TVWXX XL-::2>>CC$0DH+B/K$/$8$8$B$B$F$F"'))%% !$9D R $,#4#4[#A#A#E#E"'))$$  0889MNN  +//0B0G0I0IJJx 5566) ) EAr(..sc,6G6G!6K/LMMI --//$*2.229==>>',..//  $$&& 59==+C+C+S+S ,, 5&MM)44 /88CC #3355    8  5//11!481<<"%J!7!D!DR!H!H!LL"" "# $((3388^8LL  R   B!5 (,'7';'@'@(-e (A(( $   r0)rrrrr rrMr rr,rr@rrr rr=rrrrr| __classcell__)r"s@r.rCrChs222U222222>:5E0F::::>E'*<E'E'E'E'NG&6G4GGGG(-((;M2N((((,$4,#,,,,)!1)h?Q6R)))) ""24F"FG  $((8#(=#>  # 6n#nnnnnnnnr0rC)8loggingrGrabcrr collectionsrtypingrrrr r r ray._private.ray_constantsr r 9ray.data._internal.execution.interfaces.execution_optionsrr9ray.data._internal.execution.interfaces.physical_operatorrr=ray.data._internal.execution.operators.base_physical_operatorr3ray.data._internal.execution.operators.hash_shuffler8ray.data._internal.execution.operators.input_data_bufferr3ray.data._internal.execution.operators.zip_operatorr!ray.data._internal.execution.utilrray.data._internal.utilrrUrrVr5ray.data._internal.execution.streaming_executor_staterr getLoggerrrZrr__annotations__rr r=rrrrCrr0r.rls ##############JJJJJJJJJJJJJJJJ::::::::UTTTTTKKKKKK;;;;;;''''''((((((######XWWWWWWWW  8 $ $EMH%tEE18D>,[9k k k k k k k k \ J3I'IDIIII|||||#|||~yyyyy%8yyyyyr0