U xbOI@srddlZddlZddlZddlZddlZddddgZdddd d gZdd dZdddZ dddZ GdddZ dS)NlsfpbsslurmlocalZOMP_NUM_THREADSZOPENBLAS_NUM_THREADSZMKL_NUM_THREADSZNUMEXPR_NUM_THREADSZVECLIB_MAXIMUM_THREADSxFcCs,|}|dkr|gS|\}}}}||||}} |dkrB|} n| } tt| |} t| d} tt| | }g} t|D]~} |dkr|| | }|| | d}t||}| ||||gq|| | }|| | d}t||}| ||||gq|r(td || |td | || S)aSDivide the input box into `num_split` different sub_boxes. :param box: [x0, y0, x1, y1]: list[int] of size 4 :param num_split: int, the number of sub_boxes to split a box into :param dimension: str = 'y' or 'x', the dimension along which to split the boxes :return: sub_boxes: list(list(4 int)), the splited sub boxes y z/split along {} dimension ({:d}) into {:d} boxesz, with each box up to {:d} in {} dimension) lowerintnpceilmaxrangeminappendprintformat)box num_split dimension print_msgx0y0x1y1lengthwidthdim_sizestep sub_boxesir0r1c0c1r&M/home/exouser/operations/rsmas_insar/sources/MintPy/mintpy/objects/cluster.pysplit_box2sub_boxess4        r(TcCsj|rtdti}tD]}tj|d||<q|rft|}tD]"}|tj|<|rBtd||qB|S)alimit/set the number of threads for all environmental variables to the given value and save/return the original value for backup purpose. Link: https://stackoverflow.com/questions/30791550 Parameters: num_threads - str, number of threads Set to None to return without changing env variables Returns: num_threads_dict - dict, dictionary of the original number of threads z save the original settings of {}N set {} = {})rrNUM_THREADS_ENV_LISTosenvirongetstr) num_threadsrnum_threads_dictkeyr&r&r'set_num_threadsOs  r2cCs||rtdt|D]\\}}|tjkr|dkrXtj||rvtd|q|tj|<|rtd||qdS)z?Set back the number of threads for all environmental variables.z(roll back to the original settings of {}Nzremove env variable {}r))rrr*itemsr+r,keyspop)r0rr1valuer&r&r'roll_back_num_threadsks  r7c@s^eZdZdZdddZddZddZd d Zd d Zd dZ e ddZ ddZ ddZ dS) DaskClustera Generic dask cluster wrapper for parallel processing in blocks. This object takes in a computing function for one block in space. For the computing function: 1. the output is always several matrices (or None) and one box. 2. the number of matrices may vary for different applications/functions. 3. all matrices will be in 2D in size of (len, wid) or 3D in size of (n, len, wid), thus, the last two dimension (in space) will be the same. This charateristics allows the automatic result collection without prior knowledge of the computing funciton, thus being a generic wrapper. Check ifgram_inversion.py as an example. NcKs|||_||_||_||_||j|j|_||j|jd<td|j|jdk rltd|jd|_ d|_ dS)aiInitiate object :param cluster_type: str, cluster to use (local, slurm, lsf, pbs) :param num_worker: str, number of workers to use :param config_name: str, the name of configuratino section :other param **kwargs: dask configuration parameters e.g. config_name: str, the user specified config name to use config_namezinput Dask cluster type: {}Nzinput Dask config name: {}) r cluster_type num_workerr9cluster_kwargsformat_num_workerformat_config_namerrclusterclient)selfr:r;r9kwargsr&r&r'__init__s   zDaskCluster.__init__c Cstd|jdkr(ddlm}||_nddl}|jdkrL|jf|j|_nZ|jdkrh|jf|j|_n>|jdkr|j f|j|_n"d |j}|d t 7}t |td |j d }|rtd d}||j d W5QRXdS)zInitiate the clusterzinitiate Dask clusterrr) LocalClusterNrrrzun-recognized input cluster: {}z supported clusters: {} Fz dask_command_run_from_python.txtw)rr:dask.distributedrDr? dask_jobqueueZ LSFClusterr<Z PBSClusterZ SLURMClusterr CLUSTER_LIST ValueErrorZ job_scriptopenwrite)rArDrHmsg debug_modefr&r&r'rKs&        zDaskCluster.openc Csddlm}|d}t||jddd}t||_td|jtd|j|j|jtd ||j|_ |j j d d | |||\}}| ||||}|S) aWrapper function encapsulating submit_workers and compile_workers. For a generic result collection without prior knowledge of the computing function, we assume that the output of "func" is: several 2D or 3D matrices + a box :param func: function, a python function to run in parallel :param func_data: dict, a dictionary of the argument to pass to the function :param results: list[numpy.ndarray], arrays of the appropriate structure representing the final output of processed box (need to be in the same order as the function passed in submit_workers returns in) :return: results: tuple(numpy.ndarray), the processed results of the box r)ClientrrF)rrrzCsplit patch into {} sub boxes in x direction for workers to processz scale Dask cluster to {} workerszinitiate Dask clientT)check) rGrPr(r;lenrrr?scaler@ get_versions submit_jobcollect_result) rAfunc func_dataresultsrPrr futuressubmission_timer&r&r'runs   zDaskCluster.runc Cs`t}g}t|D]B\}}td||||d<|jj|f|ddi}||q||fS)aSubmit dask workers to the networking client that run the specified function (func) on the specified data (func_data). Each dask worker is in charge of a small subbox of the main box. :param func: function, a python function to run in parallel :param func_data: dict, a dictionary of the argument to pass to the function :param sub_boxes: list(np.nd.array), list of boxes to be computed in parallel :return futures: list(dask.Future), list of futures representing future dask worker calculations :return submission_time: time, the time of submission of the dask workers (used to determine worker runtimes as a performance diagnostic) z-submit a job to the worker for sub box {}: {}rretries)time enumeraterrr@submitr) rArWrXr r[rZr!sub_boxfuturer&r&r'rUs  zDaskCluster.submit_jobc CsZddlm}d}||ddD]6\}}|d7}t|} td|d| dd |d } | \} } } }| |d8} | |d8} | |d8} ||d8}t|d d D]\}}|d k r|j}|d kr|||d d d d | || | f<q|d kr|||d d | || | f<q|dkr8|||| || | f<qd|}|d7}t|qq|S)a$Compile results from completed workers and recompiles their sub outputs into the output for the complete box being worked on. :param futures: list(dask.Future), list of futures representing future dask worker calculations :param results: list[numpy.ndarray], arrays of the appropriate structure representing the final output of processed box (need to be in the same order as the function passed in submit_workers returns in) :param box: numpy.ndarray, the initial complete box being processed :param submission_time: time, the time of submission of the dask workers (used to determine worker runtimes as a performance diagnostic) :return: results: tuple(numpy.ndarray), the processed results of the box r) as_completedT) with_resultsrz FUTURE #z complete. Time used: z.0fz secondsNr^z*worker result has unexpected dimension: {}z! it should be either 2 or 3 or 4!)rGrdr_rr`ndimr Exception)rArZrYrr[rdZ num_futurercZ sub_resultsZsub_trbrrrrr! sub_resultZnum_dimrMr&r&r'rVsN            zDaskCluster.collect_resultcCs0|jtd|jtd|dS)zPClose connections to dask client and cluster and moves dask output/error files. zclose dask clientzclose dask clusterN)r@closerr?move_dask_stdout_stderr_files)rAr&r&r'rlRs   zDaskCluster.closecCs|dkrt}d|}|dkr available cores: {}rhz+change number of worker to {} and continue z5numWorker = all is NOT supported for cluster type: {}) r+ cpu_countr.rrendswithr floatrJr)r:r;Znum_corerMr&r&r'r=cs*   zDaskCluster.format_num_workercCsddl}|jdkrd|_|jS|jdkr8td|j|_t|jd}|j|kr|jd}d|j|}|d|j7}t||j|_|jS) aVFormat dask config_name property based on presence or absence of user specified config name. :return: config_name: str, the config_name formatted as follows: - the user specified config name if its exists in $DASK_CONFIG/dask.yaml - the default cluster_type config in $DASK_CONFIG/dask.yaml rNrzFinput config name is None, thus use the default (same as cluster type)Zjobqueueconfigz+Dask configuration "{}" was not found in {}z* Falling back to default config name: "{}") daskr:r9rlistrtr-r4r)rAruZ config_namesZconfig_locationrMr&r&r'r>s    zDaskCluster.format_config_namecCstd}td}td}t|||dkr6dSd}d}||fD]$}tj|r`t|t|qF||D]}t||qt|D]}t||qdS)zEMove *o and *e files produced by dask into stdout and sderr directoryz*.oz*.ez!dask_command_run_from_python.txt*rNZ stdout_daskZ stderr_dask) globrRr+pathisdirshutilrmtreemkdirmove)rAZ stdout_filesZ stderr_filesZ job_filesZ stdout_folderZ stderr_folderZstd_diritemr&r&r'rms        z)DaskCluster.move_dask_stdout_stderr_files)N)__name__ __module__ __qualname____doc__rCrKr\rUrVrl staticmethodr=r>rmr&r&r&r'r8s )'5 7r8)rF)NT)T) r+r_rwrznumpyr rIr*r(r2r7r8r&r&r&r' s  0