o biȁ@s.ddlZddlZddlZddlZddlmZmZddlmZddl m Z m Z m Z m Z mZmZddlmZmZddlmZmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!e rldd l"m#Z#m$Z$e%e&Z'ej()dddkZ*eefZ+GdddZ,GdddeZ-Gddde-Z.dS)N)ABCabstractmethod) defaultdict) TYPE_CHECKINGCallableDictIterableListOptional)ExecutionOptionsExecutionResources)PhysicalOperatorReportsExtraResourceUsage)AllToAllOperator)InputDataBuffer) ZipOperator) memory_string) DataContextlog_once)OpStateTopologyRAY_DATA_DEBUG_RESOURCE_MANAGER01c@seZdZdZdZdZdZdddedege fd e fd d Z d d Z ddddde fddZddZde fddZde fddZde fddZde fddZdede fdd Zdedefd!d"Zdefd#d$Zed-d&d'Zdedee fd(d)Zdedee fd*d+Zd,S).ResourceManagerz@A class that manages the resource usage of a streaming executor. ?g?topologyroptionsget_total_resources data_contextcCs||_||_||_t|_d|_t|_t|_t|_ i|_ i|_ i|_ t t|_t t|_t|_i|_i|_d|_|jrTtdd|D}|rTt||j|_|jdur\|jn |rc|jn|j|_|dS)Nrcss|]}|VqdSN)%implements_accurate_memory_accounting.0opr'a/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/execution/resource_manager.py Zs z+ResourceManager.__init__..) _topology_options_get_total_resourcesr zero_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_outputsDEBUG_RESOURCE_MANAGER_debug_downstream_fraction_downstream_object_store_memory_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)selfrrr r! should_enabler'r'r(__init__6s@        zResourceManager.__init__c Csddl}ddlm}ddlm}|sdS|}|dd}|dd}|dkrP||}|dkrR|drTt |d |d d |d d d|d d ddSdSdSdS)zDWarn if object store memory is configured below 50% of total memory.rN) WARN_PREFIXrmemoryobject_store_memoryr$ray_data_object_store_memory_warningz. Ray's object store is configured to use only z.1%z of available memory (geA.1fz GB out of a!GB total). For optimal Ray Data performance, we recommend setting the object store to at least 50% of available memory. You can do this by setting the 'object_store_memory' parameter when calling ray.init() or by setting the RAY_DEFAULT_OBJECT_STORE_MEMORY_PROPORTION environment variable.) rayray.data.contextrKray.util.debugris_initializedcluster_resourcesgetloggerwarning)rHrPrKrrT total_memoryrMobject_store_fractionr'r'r(rGns0     z9ResourceManager._warn_about_object_store_memory_if_neededr&r staterreturncCsjt|trdS|jjp d}||jj7}|}|jD] }||jj|jj7}q||j |<||j |<||S)Nr) isinstancermetrics"obj_store_mem_pending_task_outputsobj_store_mem_internal_outqueueoutqueue_memory_usageoutput_dependenciesobj_store_mem_internal_inqueue!obj_store_mem_pending_task_inputsr7r8)rHr&rZmem_op_internalmem_op_outputsnext_opr'r'r(_estimate_object_store_memorys      z-ResourceManager._estimate_object_store_memoryc Cstddd|_tddd|_tddd|_|j|j|j|j|j d}t |j }t |j D]\}}||}|}|}|jrUJ|jrZJ|jr_J||||_||||_t|try||||j|<||j|<||j|<|j||_|j||_|j||_d|td|d}|d7}td||j|<|jj|j |<|j|j_q<|jdur|jdSdS)zRecalculate resource usages.r?N)r r0r1r2r3clearr4r5r;r<lenr*reverseditemsupdate_resource_usagecurrent_processor_usagerunning_processor_usagepending_processor_usagerMrgr\raddextra_resource_usagemaxmin_metricsobj_store_mem_usedr= update_usages) rHnum_ops_so_far num_ops_totalr&rZop_usageop_running_usageop_pending_usagefr'r'r(rxsZ               zResourceManager.update_usagescC|jS)z5Return the global resource usage at the current time.)r0rHr'r'r(get_global_usagez ResourceManager.get_global_usagecCr)z=Return the global running resource usage at the current time.)r1rr'r'r(get_global_running_usagerz(ResourceManager.get_global_running_usagecCr)z=Return the global pending resource usage at the current time.)r2rr'r'r(get_global_pending_usagerz(ResourceManager.get_global_pending_usagecCsht|j|jkr |jSt|_|jj}|jj}|}|j}|j |9_ | | ||_|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. ) timer/GLOBAL_LIMITS_UPDATE_INTERVAL_Sr.r+resource_limitsexclude_resourcesr,rFrMrusubtract)rHdefault_limitsexcludetotal_resourcesdefault_mem_fractionr'r'r(get_global_limitss  z!ResourceManager.get_global_limitscCs |j|S)zDReturn the resource usage of the given operator at the current time.)r3rHr&r'r'r( get_op_usage  zResourceManager.get_op_usagecCs|j|jdd}|j|jr|d|j|jdd7}|d|j|d7}|jr|dt|j|dt|j|d7}t|j t r||j j vr|j j |}|d |jd7}|d |jd7}|d |7}t|j j |d }|d|d7}|S) zbReturn a human-readable string representation of the resource usage of the given operator.rOz CPUz, z GPUz object storez (in=z,out=)z, budget=(cpu=z,gpu=z ,obj_store=r)r4cpugpuobject_store_memory_strr:rr7r8r\r=r@ _op_budgets_output_budgetsrU)rHr& usage_strbudgetreserved_for_outputr'r'r(get_op_usage_strs0     z ResourceManager.get_op_usage_strcCs |jduS)z.Return whether OpResourceAllocator is enabled.Nr=rr'r'r(rC+rz-ResourceManager.op_resource_allocator_enabledOpResourceAllocatorcCs|jdusJ|jS)zReturn the OpResourceAllocator.Nrrr'r'r(op_resource_allocator/sz%ResourceManager.op_resource_allocatorcC|jdurdS|j|SzqReturn the maximum bytes of pending task outputs can be read for the given operator. None means no limit.N)r=max_task_output_bytes_to_readrr'r'r(r5  z-ResourceManager.max_task_output_bytes_to_readcCrz_Return the budget for the given operator, or None if the operator has unlimited budget.N)r= get_budgetrr'r'r(r<rzResourceManager.get_budgetN)r[r)__name__ __module__ __qualname____doc__rrDrEr rr rrJrGr6rgrxrrrrr rstrrboolrCpropertyrr rrr'r'r'r(r&sD  8 B rc@s^eZdZdZdefddZeddZedede e fd d Z edede e fd d Z d S)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. resource_managercCs ||_dSr")_resource_manager)rHrr'r'r(rJLs zOpResourceAllocator.__init__cCdS)z#Callback to update resource usages.Nr'rr'r'r(rxOsz!OpResourceAllocator.update_usagesr&r[cCrrr'rr'r'r(rTz1OpResourceAllocator.max_task_output_bytes_to_readcCrrr'rr'r'r(rZrzOpResourceAllocator.get_budgetN)rrrrrrJrrxr r r6rr rr'r'r'r(rDs rcseZdZdZGdddZdedeffdd Zded e fd d Z d e efd d Z ddZ ded eefddZded e fddZded efddZded eefddZded eefddZded eefddZddZZS)r@aAn 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. c@sHeZdZdZdZdZdZddZdefdd Z e ded e fd d Z d S)z+ReservationOpResourceAllocator.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@FcCs*tt|_tdd|_tdd|_dS)NcStSr"rr'r'r'r(zFReservationOpResourceAllocator.IdleDetector.__init__..cSrr"rr'r'r'r(rr)rr6last_num_outputslast_output_timelast_detection_timerr'r'r(rJs z4ReservationOpResourceAllocator.IdleDetector.__init__r&cCs|t}||j||jkr<|jj}||j|kr*||j|<||j|<||j|<dS||j|<||||j|dSdSNTF)rrDETECTION_INTERVAL_Sr]num_task_outputs_generatedrr"print_warning_if_idle_for_too_long)rHr&cur_timecur_num_outputsr'r'r( detect_idles    z7ReservationOpResourceAllocator.IdleDetector.detect_idle idle_timecCs:||jks|jr dSd|_d|d|d}t|dS)z4Print a warning if an operator is idle for too long.NTz Operator z# is running but has no outputs for a seconds. Execution may be slower than expected. Ignore this warning if your UDF is expected to be slow. Otherwise, this can happen when there are fewer cluster resources available to Ray Data than expected. If you have non-Data tasks or actors running in the cluster, exclude their resources from Ray Data with `DataContext.get_current().execution_options.exclude_resources`. This message will only print once.)WARN_ON_IDLE_TIME_S _warn_printedrVrW)clsr&rmsgr'r'r(rs  zNReservationOpResourceAllocator.IdleDetector.print_warning_if_idle_for_too_longN) rrrrrrrrJr r classmethodfloatrr'r'r'r( IdleDetector{s rrreservation_ratiocsft|||_d|jkrdksJJi|_i|_t|_i|_i|_ i|_ | |_ dS)Ngrh) superrJ_reservation_ratio _op_reserved_reserved_for_op_outputsr r- _total_sharedrr_reserved_min_resourcesr_idle_detector)rHrr __class__r'r(rJs  z'ReservationOpResourceAllocator.__init__r&r[cCs| o | S)z2Whether the op is eligible for memory reservation.)throttling_disabledexecution_finishedrr'r'r(_is_op_eligibles z.ReservationOpResourceAllocator._is_op_eligiblecsfddjjDS)Ncsg|] }|r|qSr')rr$rr'r( s  zDReservationOpResourceAllocator._get_eligible_ops..)rr*rr'rr(_get_eligible_opss z0ReservationOpResourceAllocator._get_eligible_opsc CsV|j}|}|j|j|j|}t|dkr$dS| |j t|}t |D]s\}}t ddt |jdd}|\}} ||} | |} | | } | |j|ddrgd|j|<n"d|j|<t dd|j} |dkrtdt|rtd|d | |j|<|j|j|<| |} || }| t }q2||_dS) NrriT)ignore_object_store_memoryFlow_resource_warning_z7Cluster resources are not engough to run any task from z8. The job may hang forever unless the cluster scales up.)rrrrrjrrcopyrkscaler enumerater rtrMmin_max_resource_requirementsrrurrsatisfies_limitridrVrWr-r) rH global_limits eligible_ops remainingdefault_reservedindexr&reserved_for_outputsmin_resource_usagemax_resource_usagereserved_for_tasksop_total_reservedr'r'r(_update_reservationsJ                   z2ReservationOpResourceAllocator._update_reservationcCs|j|dSr")rrUrr'r'r(r+sz)ReservationOpResourceAllocator.get_budgetcCs6||D]}|j|sdS|j|rdSqdSr)_get_downstream_eligible_opsrrrrHr&rfr'r'r(-_should_unblock_streaming_output_backpressure.s  zLReservationOpResourceAllocator._should_unblock_streaming_output_backpressurecs0jj|}|tfdd|D7}|S)zsGet the outputs memory usage of the given operator, including the downstream ineligible operators. c3s|] }j|jVqdSr")rrrMr%rfrr'r(r)Is  zWReservationOpResourceAllocator._get_op_outputs_usage_with_downstream..)rr8sum_get_downstream_ineligible_ops)rHr&op_outputs_usager'rr(%_get_op_outputs_usage_with_downstreamBs  zDReservationOpResourceAllocator._get_op_outputs_usage_with_downstreamcCs||jvrdS|j|j}||}|t|j||d7}t|r*||j|<dSt|}|dks4J|dkr?| |r?d}||j|<|S)Nrri) rrMrrtrmathisinfrr6r)rHr&resrr'r'r(rOs       zdownstream_map" will return an empty list. - "cur_map->limit1->limit2->downstream_map" will return [limit1, limit2]. N)rarrrr'r'r(ras z=ReservationOpResourceAllocator._get_downstream_ineligible_opsccs4|jD]}||r|Vq||EdHqdS)a 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)rarrrr'r'r(ros  z;ReservationOpResourceAllocator._get_downstream_eligible_opsc Cs||j|}t|dkrdS|j}|D]K}d}||jj|7}||}|t ||j |d7}|j | }||_ |j|}|| t}||j|<|| t} || }q| t}tt|D]V\} }|dt|| } ||j||  t} | s| | |r| | } || }|sJ||| | f|j|| |j|<td|j|_qs|D]}tdd|jDrtd|j|_ qdS)Nrrhinfcss|]}t|tVqdSr")r\MATERIALIZING_OPERATORSrr'r'r(r)s  z?ReservationOpResourceAllocator.update_usages..)rrrjrrkrrr7rrtrrrrMrrr r-rrlrincremental_resource_usagerris_zeroris_non_negativerranyra) rHrremaining_sharedr& op_mem_usagerr{ op_reservedop_reserved_remainingop_reserved_exceededi op_shared to_borrowr'r'r(rxsb            z,ReservationOpResourceAllocator.update_usages)rrrrrrrrJr rrr rrr r rrrr6rrrrrx __classcell__r'r'rr(r@as4? E    r@)/loggingrosrabcrr collectionsrtypingrrrrr r 9ray.data._internal.execution.interfaces.execution_optionsr r 9ray.data._internal.execution.interfaces.physical_operatorr r=ray.data._internal.execution.operators.base_physical_operatorr8ray.data._internal.execution.operators.input_data_bufferr3ray.data._internal.execution.operators.zip_operatorr!ray.data._internal.execution.utilrrQrrRr5ray.data._internal.execution.streaming_executor_staterr getLoggerrrVenvironrUr9rrrr@r'r'r'r(s2