o bi@sUddlZddlZddlZddlmZmZddlZddlmZddl m Z dZ e dZ dZ ejdddd Gd d d ZeZejed <d dZdS)N)DictList) DataContext)NodeAffinitySchedulingStrategy<g333333?)num_cpus max_restartsmax_task_retriesc@sXeZdZdZddZddZdeedefdd Z d d Z d eefd dZ ddZ dS)AutoscalingRequesteraActor to make resource requests to autoscaler for the datasets. The resource requests are set to timeout after RESOURCE_REQUEST_TIMEOUT seconds. For those live requests, we keep track of the last request made for each execution, which overrides all previous requests it made; then sum the requested amounts across all executions as the final request to the autoscaler. csBi_t_tj_fdd}tj|dd_ j dS)Ncs tttjjqN)timesleepPURGE_INTERVALrayget _self_handlepurge_expired_requestsremoteselfrf/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/execution/autoscaling_requester.pypurge_thread_run)s z7AutoscalingRequester.__init__..purge_thread_runT)targetdaemon) _resource_requestsRESOURCE_REQUEST_TIMEOUT_timeoutrget_runtime_context current_actorr threadingThread _purge_threadstart)rrrrr__init__ s   zAutoscalingRequester.__init__cCs |tjjj|ddSN)bundles)_purger autoscalersdkrequest_resources_aggregate_requestsrrrrr3sz+AutoscalingRequester.purge_expired_requestsreq execution_idcCs8||t|jf|j|<tjjj|ddSr') r)rrrrr*r+r,r-)rr.r/rrrr,7s   z&AutoscalingRequester.request_resourcescCs<t}t|jD]\}\}}||kr|j|q dSr )rlistritemspop)rnowk_trrrr)Ds  zAutoscalingRequester._purgereturncCsg}|jD] \}\}}||qdd}||}|dkrBt}d|vrB||dkrBtt|d|}|ddig||S)NcSs&d}|D] }d|vr||d7}q|S)NrCPUr)r.r rrrrget_cpusPs  z:AutoscalingRequester._aggregate_requests..get_cpusrr8)rr1extendrcluster_resourcesmathceilARTIFICIAL_CPU_SCALING_FACTOR)rr.r5r9r:r totaldeltarrrr-Ks  z(AutoscalingRequester._aggregate_requestscCs ||_dS)z&Set the timeout. This is for test onlyN)r)rttlrrr_test_set_timeouths z&AutoscalingRequester._test_set_timeoutN) __name__ __module__ __qualname____doc__r&rrrstrr,r)r-rDrrrrr s  r _autoscaling_requester_lockcCsbt}|j}ttdd}ttj dddd|d WdS1s*wYdS)NF)softr Tdetached)name namespace get_if_existslifetimescheduling_strategy) r get_currentrQrrr get_node_idrJr optionsr)ctxrQrrr)get_or_create_autoscaling_requester_actorrs  $rV)r>r"rtypingrrrray.data.contextrray.util.scheduling_strategiesrrrr@rr RLockrJ__annotations__rVrrrrs   X