o 1 iBB@s*ddlZddlZddlZddlZddlZddlZddlZddlZddlZe dZ ddZ ddZ ddZ dd d d d Zd dZddZ d9ddZddZddZddZddZddZd Zd Zd!d"Zd#d$Zd%d&Zd'Zd(Zd)Zd*Zd+Z d,Z!d-d.Z"d/d0Z#d1d2Z$d3d4Z%d5d6Z&d7d8Z'dS):Nzray.util.spark.utilscCs dtjvS)NZDATABRICKS_RUNTIME_VERSION)osenvironrr`/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/util/spark/utils.pyis_in_databricks_runtimes rcCs,d|}d|}d|d|d|dS)N zCommand z failed with return code z", tail output are included below.  )join)cmd return_codetail_output_dequeZcmd_strZ tail_outputrrrgen_cmd_exec_failure_msgs  rcCsD|jdd}t|dd}|d}ddddd}|||S) Nzspark.executor.memoryZ1gii@l)kmgt)confgetlowerint)sparkZ value_strZ value_numZ value_unitZunit_maprrr*get_configured_spark_executor_memory_bytess rT) extra_env synchronousc st|hd}|rtdt||dd}|dur(|dur(td|dur.|nitj|}tj |f|dtj tj d|t j dd fd d }tj|d d |safS}|dkrqtt||dS)a A convenience wrapper of `subprocess.Popen` for running a command from a Python script. If `synchronous` is True, wait until the process terminated and if subprocess return code is not 0, raise error containing last 100 lines output. If `synchronous` is False, return an `Popen` instance and a deque instance holding tail outputs. The subprocess stdout / stderr output will be streamly redirected to current process stdout. >textstdoutstderrz`kwargs` cannot contain envNz5`extra_env` and `env` cannot be used at the same timeT)r rrrd)maxlencs&jD] }|tj|qdSN)rappendsyswrite)lineprocessr rrredirect_log_thread_fnNs  z(exec_cmd..redirect_log_thread_fnr)targetargsr)setkeys intersection ValueErrorlistpoprr subprocessPopenPIPESTDOUT collectionsdeque threadingThreadstartwait RuntimeErrorr)r rrkwargsZillegal_kwargsr r*r rr(rexec_cmd(s8   r?cCs^ddl}ddlm}|||j|j}|||fdkWdS1s(wYdS)Nr)closing)socket contextlibr@AF_INET SOCK_STREAM connect_ex)hostportrAr@sockrrris_port_in_usebs  $rIcCsDt}t||kr t||rdStdt||ks dS)NTF)timerIsleep)rFrGtimeoutZbeg_timerrr_wait_service_upjs  rNrr!cCs\t}|pg}t|D]}|||}||vrq t||s"|Sq td|d|d)z! Get random unused port. z!Get available port between range z and z failed.)random SystemRandomrangerandintrIr=)rFZmin_portZmax_port max_retriesZ exclude_listrng_rGrrrget_random_unused_portus   rWcCs(ddlm}|}|durtd|S)Nr) SparkSessionzSpark session haven't been initiated yet. Please use `SparkSession.builder` to create a spark session and connect to a spark cluster.)Z pyspark.sqlrXZgetActiveSessionr=)rXZ spark_sessionrrrget_spark_sessions rYcCs |jdS)Nzspark.driver.host)rr)rrrr!get_spark_application_driver_hosts rZcCsV|j}|dur"dd}|dgd||||jS|| S)z0Gets the current max number of concurrent tasks.NcSsdSr#r)rVrrr dummpy_mappersz3get_max_num_concurrent_tasks..dummpy_mapperrJ) Z_jscsc parallelizeZ withResourcesmapcollectZmaxNumConcurrentTasksZ_java_resource_profileZresourceProfileManagerZdefaultResourceProfile)Z spark_contextZresource_profileZsscr[rrrget_max_num_concurrent_taskss   r`cCs*ddl}ttjvrttjtS|jSNr)psutil)RAY_ON_SPARK_WORKER_PHYSICAL_MEMORY_BYTESrrrvirtual_memorytotal)rbrrr'_get_spark_worker_total_physical_memorys  rfcCs,ddl}ttjvrttjtS|djSNrz/dev/shm)shutil'RAY_ON_SPARK_WORKER_SHARED_MEMORY_BYTESrrr disk_usagere)rhrrr%_get_spark_worker_total_shared_memorys  rkg?c Csddl}ddl}ttjvrttjt}n|j}|t}t tjvr+ttjt }n| dj}|t}t ||||\}}}|durHt |||fSrg)rhrb)RAY_ON_SPARK_DRIVER_PHYSICAL_MEMORY_BYTESrrrrdre'_RAY_ON_SPARK_NODE_MEMORY_BUFFER_OFFSET'RAY_ON_SPARK_DRIVER_SHARED_MEMORY_BYTESrj_calc_mem_per_ray_node_loggerwarning) configured_heap_memory_bytesconfigured_object_store_bytesrhrbZavailable_physical_memZavailable_shared_memheap_mem_bytesobject_store_bytes warning_msgrrrcalc_mem_ray_head_nodes.      rwcCs.t||t}t||t}t||||Sr#)rrmro)Znum_task_slotsphysical_mem_bytesshared_mem_bytesrrrsavailable_physical_mem_per_nodeavailable_shared_mem_per_noderrr_calc_mem_per_ray_worker_nodes  r|c Csddlm}m}d}|p||}tjds||kr|}|t}||kr(|}d}||kr?||kr7d|d}nd|d}|}t|}|durNt||} nt|} | ||fS) Nr)&DEFAULT_OBJECT_STORE_MEMORY_PROPORTION!OBJECT_STORE_MINIMUM_MEMORY_BYTESZ#RAY_OBJECT_STORE_ALLOW_SLOW_STORAGEzzYour configured `object_store_memory_per_node` value is too high and it is capped by 80% of per-Ray node allocated memory.zYour operating system is configured with too small /dev/shm size, so `object_store_memory_worker_node` value is configured to minimal size (z- bytes),Please increase system /dev/shm size.zYou configured too small Ray node object store memory size, so `object_store_memory_worker_node` value is configured to minimal size (zI bytes),Please increase 'object_store_memory_worker_node' argument value.)Zray._private.ray_constantsr}r~rrr0_RAY_ON_SPARK_MAX_OBJECT_STORE_MEMORY_PROPORTIONr) rzr{rrrsr}r~rvruZobject_store_bytes_upper_boundrtrrrros@  roRAY_ON_SPARK_WORKER_CPU_CORESRAY_ON_SPARK_WORKER_GPU_NUMrcrirlrncCs(ddl}ttjvrttjtS|Sra)multiprocessingrrrr cpu_count)rrrr_get_cpu_coresSs rc Csttjvr ttjtStddurdSztjdddddd}t|j dWSt yG}zt dt|WYd}~dSd}~ww)Nz nvidia-smirz1nvidia-smi --query-gpu=name --format=csv,noheaderT)shellcheckrcapture_outputr zU'nvidia-smi --query-gpu=name --format=csv,noheader' command execution failed, error: )rrrrrhwhichr3runlenrstripsplit Exceptionrpinforepr)Zcompleted_procerrr_get_num_physical_gpusas, rcCsd||krtd|d|d||}|dkr0||kr&td|d|d|||kr0||}|S)Nzscpu number per Ray worker node should be <= spark worker node CPU cores, you set cpu number per Ray worker node to z* but spark worker node CPU core number is .rz|gpu number per Ray worker node should be <= spark worker node GPU number, you set GPU devices number per Ray worker node to z- but spark worker node GPU devices number is )r0)num_cpusnum_gpusnum_cpus_per_nodenum_gpus_per_nodenum_ray_node_slotsrrr_get_local_ray_node_slots~s* rc CsVt}|dkr t}nd}t||||}t}t}t|||||\} } } | | d| fS)z Returns tuple of ( ray_worker_node_heap_mem_bytes, ray_worker_node_object_store_bytes, error_message, # always None warning_message, ) rN)rrrrfrkr|) rrheap_memory_per_nodeobject_store_memory_per_noderrrrxryZray_worker_node_heap_mem_bytesZ"ray_worker_node_object_store_bytesrvrrr"_get_avail_mem_per_ray_worker_nodes2 rc sjfdd}|jdgd|d\}}}} |dur(td|d| dur1t| ||fS)a< Return the available heap memory and object store memory for each ray worker, and error / warning message if it has. Return value is a tuple of (ray_worker_node_heap_mem_bytes, ray_worker_node_object_store_bytes, error_message, warning_message) NB: We have one ray node per spark task. c sdztWSty1}zddl}d||j}ddt||dfWYd}~Sd}~ww)Nrr r)rr tracebackr format_tb __traceback__r)rVrrZ trace_msgrrrrrrmappers z1get_avail_mem_per_ray_worker_node..mapperrJrNz:Inferring ray worker node available memory failed, error: a. You can bypass this error by setting following spark configs: spark.executorEnv.RAY_ON_SPARK_WORKER_CPU_CORES, spark.executorEnv.RAY_ON_SPARK_WORKER_GPU_NUM, spark.executorEnv.RAY_ON_SPARK_WORKER_PHYSICAL_MEMORY_BYTES, spark.executorEnv.RAY_ON_SPARK_WORKER_SHARED_MEMORY_BYTES.)Z sparkContextr]r^r_r=rprq) rrrrrrZ'inferred_ray_worker_node_heap_mem_bytesZ+inferred_ray_worker_node_object_store_byteserrrvrrr!get_avail_mem_per_ray_worker_nodes   rcs:dtjvrddtjddDfdd|DS|S)NZCUDA_VISIBLE_DEVICEScSsg|]}t|qSr)rr).0devrrr s z9get_spark_task_assigned_physical_gpus..,csg|]}|qSrr)raddrZvisible_cuda_dev_listrrrs)rrr)Z gpu_addr_listrrr%get_spark_task_assigned_physical_gpuss r)rrOr!N)(r7loggingrrPrhr3r%r9rK getLoggerrprrrr?rIrNrWrYrZr`rfrkrrmrwr|rorrrcrirlrnrrrrrrrrrrsT   :   &E. ;