B ÊÒÐ_“ã@sdZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZddlZddlmmZddlmZddlZejdeddd „Zd d „ZGd d „d ƒZdd„Zdd„Zdd„Zedkreejdd…ƒZ ee ƒZ!e! "¡e j#dkre! $¡Z%dS)a‡ Functions related to batch job submission. Should be run with a file containing jobs to submit as a batch. Optional parameters for job submission are, --memory, --walltime, and --queuename. Generates job scripts, runs them, and waits for output files to be written before exiting. This script has functions to support submitting two different job types: a script as a job or a batch file consisting of multiple parallel tasks. submitting a script as a job is done calling the function: submit_script However submitting a batch file (calling submit_batch_jobs) can be done in 5 different ways with or without launcher. Two environmental variables have to be set: JOB_SUBMISSION_SCHEME and QUEUENAME (set in '~/accounts/platforms_defaults.bash') QUEUENAME has defaults based on platforms. comment/uncomment or introduce a new one JOB_SUBMISSION_SCHEME: it can have one of these options: singleTask ---> submit each task of a batch file separately in a job multiTask_singleNode ---> distribute tasks of a batch file into jobs with one node multiTask_multiNode ---> submit tasks of a batch file in one job with required number of nodes launcher_multiTask_singleNode ---> distribute tasks of a batch file into jobs with one node, submit with launcher launcher_multiTask_multiNode ---> submit tasks of a batch file in one job with required number of nodes using launcher éN)Ú message_rsmas)Úqueue_config_fileÚsupported_platforms)ÚdatetimeÚignore)ÚcategorycCsÄtjtjd}| dd¡}|jdtdd|jddtd d d |jd d ddd|jddddd|jddddd|jdddddd|jdd td!d"d |jd#d$d%d&d'|jd(d)d%d*d'|S)+a Creates an argument parser for parsing parameters for batch job submission. Required parameter: file to batch create Optional parameters: template file, memory, walltime, and queue name :return: ArgumentParser object for parsing command line batch job submission )Úformatter_classz Input FilezFile/Dataset to displayÚfilezThe file to batch create)ÚtypeÚhelpz --templateÚcustom_template_filez template filez&custom template with option settings. )Údestr Úmetavarr z--memoryÚmemoryz MEMORY (KB)z4Amount of memory to allocate, specified in kilobytes)r rr z --walltimeÚ wall_timezWALLTIME (HH:MM)z+Amount of wall time to use, in HH:MM formatz--queueÚqueueÚQUEUEzName of queue to submit job toz--outdirÚout_dirÚ run_filesZOUTDIRzoutput directory for run files)r Údefaultrr z --numBurstsÚ num_burstsznumber of burstsz&number of bursts to calculate walltimez --writeonlyÚ writeonlyÚ store_truez"Write job files without submitting)r Úactionr z--remoraÚremoraz!use remora to get job information)ÚargparseÚArgumentParserÚRawTextHelpFormatterÚadd_argument_groupÚ add_argumentÚstrÚint)ÚparserÚgroup©r$úÚ email_notifÚ job_filesÚglobr3ÚtemplateÚprintÚformatÚinps)ÚselfrVÚkZdem_filer$r$r%Ú__init__esB, zJOB_SUBMIT.__init__NÚFalsecCsÀtj |j¡s4tj |j¡r(t |j¡t |j¡|dk rB||_tj |d¡d}|d  dd„|dd…Dƒ¡7}|j |ddg|_ |j ||||j|j d |d kr¼|j|j |jd dS) aŸ Submits a single script as a job. (compare to submit_batch_jobs for several tasks given in run_file) :param job_name: Name of job. :param job_file_name: Name of job file. :param argv: Command line arguments for running job. :param email_notif: If email notifications should be on or not. Defaults to true. :return job number of the script that was submitted Nrú css|]}|dkr|VqdS)z--submitNr$)Ú.0Úflagr$r$r%ú ¤sz+JOB_SUBMIT.submit_script..r'Úscript)Újob_type)r3Únumber_of_nodesrZ)r3)r*r-Úisdirr3ÚisfileÚremoveÚmakedirsrPr1r/Úget_memory_walltimerQÚwrite_single_job_filer>Úsubmit_and_check_job_status)rWÚjob_nameÚ job_file_nameÚargvrPZ writeOnlyÚ command_liner$r$r%Ú submit_script’s     zJOB_SUBMIT.submit_scriptc Csl|dkr|j}|dk r||_t |jdj||jd¡g|_|jt kr6t d |j ¡ƒ|j |ddt |dƒ}| ¡}t|ƒ}WdQRXt t |t|jƒ|j|j¡¡}d|jkrÈ| |¡nnd |jkr|d }tj |¡}|j||t|ƒ||jd } |j |j| ||d ¡nd |jkr6| |||¡t|jƒdkrhx |jD]} t  d | ¡¡qNWdS)zã creates jobs based on scheduler :param batch_file: batch job name :param email_notif: If email notifications should be on or not. Defaults to true. :return: True if running on a cluster Nz.job_submission.py {a} --outdir {b} --writeonly)ÚaÚbz Working on a {} machine ... Úbatch)r`ÚrZ singleTaskZmultiTask_multiNodeZ_0)Únumber_of_tasksrar3)Ú batch_fileZmultiTask_singleNoderz chmod +x {})!r rPrÚlogr3rUrrQrErrTrFrfÚopenÚ readlinesÚlenÚnpr!ÚceilÚfloatrOrHrIrDÚwrite_batch_singletask_jobsr*r-r1Úget_job_file_linesÚappendÚadd_tasks_to_job_file_linesÚ split_jobsÚsystem) rWrsrPÚfÚtasksrrraÚbatch_file_nameriÚjob_file_linesÚjob_filer$r$r%Úwrite_batch_jobs±s:          zJOB_SUBMIT.write_batch_jobsc Cs–t|jƒdkr$|j|j|jddStdƒt d¡}t|dƒH}| ¡}|j dkrZd}nd }x$|D]}t  ||¡|tj d<qdWWd QRXd Sd S) zâ submit jobs based on scheduler :param batch_file: batch job name :param email_notif: If email notifications should be on or not. Defaults to true. :return: True if running on a cluster r)r3Tz! Working on a single machine ... ÚPATHrqr8z-export PATH=$ISCE_STACK/stripmapStack:$PATH; z)export PATH=$ISCE_STACK/topsStack:$PATH; NF) rwrQrhrrTr*r+rurvr7r€Úenviron)rWrsÚ system_pathrZ command_linesÚcmdrlr$r$r%Úsubmit_batch_jobsæs    zJOB_SUBMIT.submit_batch_jobscCsd}|jdkr"dtj ||¡}nš|jdkr@dtj ||¡}n||jdkr¬tjddtjdj ¡  d ¡}|  d ¡s||  d ¡r’d   tj ||¡¡}q¼d   tj ||¡¡}d}nt d  |j¡ƒ‚tj |tjdd}|rüt d|  d ¡¡}ttdd„|Dƒƒƒ}nd}|S)zí Submit a single job (to bsub or qsub). Used by submit_jobs_individually and submit_job_with_launcher and submit_script. :param job_file_name: Name of job file to submit. :return: Job number of submission TÚLSFzbsub < ÚPBSzqsub < ÚSLURMÚhostname)ÚshellÚstdoutzutf-8ÚloginÚcometz sbatch {}z{}Fz"ERROR: scheduler {0} not supported)Ústderrrz\d+cSsg|] }t|ƒ‘qSr$)r!)r\Úxr$r$r%ú sz0JOB_SUBMIT.submit_single_job..ÚNone)rFr*r-r/Ú subprocessÚPopenÚPIPEr‘ÚreadÚdecodeÚ startswithrUÚ ExceptionÚ check_outputÚSTDOUTÚreÚfindallr Úmax)rWrjr3Zjob_num_existsÚcommandrZ output_jobÚ job_numberr$r$r%Úsubmit_single_jobs$   zJOB_SUBMIT.submit_single_jobr'c Cs¦|j||||d}| d¡|jr@| d¡| d|d¡n| d|d¡d |¡}ttj ||¡dƒ}| |¡WdQRX|j  |¡t  d  |¡¡dS) a Writes a job file for a single job. :param job_name: Name of job. :param job_file_name: Name of job file. :param command_line: Command line containing process to run. :param work_dir: working or output directory )r3raz freez module load remoraz remora Ú z{0}.jobzw+Nz chmod +x {}) r|r}rrUrur*r-r/Ú writelinesrQr€)rWrirjrlr3rar„r…r$r$r%rg#s     z JOB_SUBMIT.write_single_job_filec Csnt|ƒ}| ¡}WdQRXxLt|ƒD]@\}}tj |¡ tj¡ddt|ƒ}|j ||||j dq&WdS)ab Iterates through jobs in input file and writes a job file for each job using the specified scheduler. This function is used for batch jobs in pegasus (LSF) to split the tasks into multiple jobs :param batch_file: File containing batch of jobs for which we are creating job files. :return: List of job file names. NéÿÿÿÿÚ_)r3) rurvÚ enumerater*r-r.r2Úsepr rgr)rWrsÚ input_fileZjob_listÚirlrjr$r$r%r{@s  $z&JOB_SUBMIT.write_batch_singletask_jobsc Cstg}g}g}x|D]ˆ}t d tj ||¡¡¡| ||¡}tj |d | d¡d|¡¡}tj |d | d¡d|¡¡} | |¡| |¡| | ¡qWd} d} d} t  d¡|j dkr–g} tj |d ¡}xt ||ƒD]\}}|d ksàd }xì|d krât d  ||¡¡t  d¡t |d ƒ}|  ¡}t|ƒdkrBwøWdQRXd|dkshd|dkrœtd || ¡ƒ| | d7} t  | d¡| d7} qød|dkr°d}qød|dkrÎd}|  |¡qød}td |¡ƒ‚qøWqàWt| ƒdkröxŒ| D]„}t |¡}tj|dd}t ||¡t t ¡d¡}|dtj |¡d|d|}t |jd d!ƒ}| |¡WdQRXqüW|j| |jd"n`x^t ||ƒD]P\}}d |kr¢x:tj |¡sîtd || ¡ƒ| | d7} t  | ¡q¶Wq¢Wxx|D]p}t | d¡dd#¡}xP|D]H}t|d$ƒt|d%ƒt|d&ƒt|d'ƒg}t  |¡ !¡rtd( |¡ƒ‚qWqüWdS))aƒ Writes a single job file for launcher to submit as array. This is used to submit jobs in slurm or sge where launcher is available (compare to submit_jobs_individually used on pegasus with LSF) :return: :param batch_file: File containing tasks that we are submitting. :param work_dir: the directory to check outputs and error files of job z chmod +x {}z{}_{}.or=rz{}_{}.eé<érŽZ job_statusr—Úwaitz!sacct --format="State" -j {} > {}rqéNZPENDINGZRUNNINGz/Waiting for job {} output file after {} minutesr'Z COMPLETEDÚcompleteZTIMEOUTÚtimeoutÚfailedz'Error: {} job was terminated with Errorg333333ó?)Úfactorz %Y%m%d:%H-%Mz: re-running: z: z --> z/run_files/rerun.logrn)r3z*.ezSegmentation faultZAbortedÚERRORÚErrorzError terminating job: {})"r*r€rUr-r/r¦r2r}ÚtimeÚsleeprFÚziprurvrwrTÚ RuntimeErrorÚputilsÚextract_walltime_from_job_fileÚmultiply_walltimeÚreplace_walltime_in_job_filerÚstrftimeÚnowr1r3r¨rhÚexistsrRÚcheck_words_in_filerxÚarrayÚany)rWrQr3Z job_numbersZjobs_outZjobs_errrjZjob_numÚoutÚerrr®Z wait_time_secZtotal_wait_time_minZrerun_job_filesZjob_status_filer¥Zjob_statZ stat_fileÚstatusrÚ new_wall_timeÚdateStrÚstringZrerunÚ error_filesÚerrfileZjob_exitr$r$r%rhPs†                   $     z&JOB_SUBMIT.submit_and_check_job_statuscs |}d}x(|t|jƒkr0|d}t ||¡}q Wtt t|ƒ|¡ƒ‰t|j||jƒ}x<|ˆkr˜|t|jƒkr”|d7}tt t|ƒ|¡ƒ‰q^Pq^Wx(|ˆkrÂ|d}t|j||jƒ}qœWt ˆ|¡|_|dkrêt d  |¡ƒtj dt|ƒˆ…  ¡}‡fdd„|Dƒ}t|ƒ|d<xzt ||ƒD]l\} } | | ¡} |d  | ¡} tj | ¡} |j| | | | ||jd}|j||| | …| d }|j |¡q,Wd S) zÝ splits the batch file tasks into multiple jobs with one node :param batch_file: :param tasks: :param number_of_nodes: Total number of nodes required for all tasks :return: r'z|Note: Number of jobs exceed the numbers allowed per queue for jobs with 1 node... Number of Nodes per job are adjusted to {}rcsg|] }|ˆ‘qSr$r$)r\r•)Únumber_of_parallel_tasksr$r%r–Ñsz)JOB_SUBMIT.split_jobs..r©z_{})rrrar3)rsN)r!rJrxryrwrKrMÚmathÚ!number_of_parallel_tasks_per_noderTrUÚogridÚtolistr»Úindexr*r-r1r|rr~rQr})rWrsr‚raZnumber_of_jobsZnumber_of_nodes_per_jobZnumber_of_limited_memory_tasksZ start_linesZ end_linesZ start_lineZend_lineZ job_countrƒrir„rjr$)rÏr%rªs@       zJOB_SUBMIT.split_jobsrpc CsŠtjdd}|dkr {} 2>{} r§rrªz_$LAUNCHER_JID.oz_$LAUNCHER_JID.ezw+Nz module load remoraz module load launcherz export OMP_NUM_THREADS={0}z export PATH={0}:$PATHz export LAUNCHER_WORKDIR={0}z export LAUNCHER_PPN={0} z export LAUNCHER_JOB_FILE={0} z remora $LAUNCHER_DIR/paramrun z mv remora_$SLURM_JOB_ID remora_z_$SLURM_JOB_ID z $LAUNCHER_DIR/paramrun z{} > {} 2>{} & z_{}.oz_{}.ezC export LD_PRELOAD=/home1/apps/tacc-patches/python_cacher/myopen.soz export OMP_NUM_THREADS={0}z wait)rFr˜r™ršr‘r›rœrrUrDr½Ú$extract_config_file_from_task_stringÚ)extract_date_string_from_config_file_namer}r2r*r-r.rÃrdrur¨rrOrBrrÑr1r/r«)rWr„r‚rsZ do_launcherrrjZtasks_with_outputÚlinerÕÚ date_stringZbatch_fZjob_fÚcountr$r$r%r~ys^                   z&JOB_SUBMIT.add_tasks_to_job_file_lines)NrZ)NN)N)Nr')N)rp)r'r'N)N)Ú__name__Ú __module__Ú __qualname__Ú__doc__rYrmr†r‹r¦rgr{rhrrfr|r~r$r$r$r%r6`s-  5 !  Z9 = Yr6c sNt|dƒ}| ¡}WdQRX‡fdd„|Dƒ}t d|¡dkrFdSdSdS) zÈ Checks for existence of a specific word in a file :param errfile: The file to be checked :param eword: The word to search for in the file :return: True if the word is in the file rqNcsg|] }ˆ|k‘qSr$r$)r\Úitem)Úewordr$r%r–Üsz'check_words_in_file..r'rTF)rurvrxÚsum)rÎrñrÚlinesZ check_ewordr$)rñr%rÄÑs  rÄc CsLt|ƒ}|d}|dkr2t d¡r.t d¡}nd}tjddtjdj ¡ d¡}x t D]}||krj|}PqVd}qVW|j r„|j |d <nt d ¡rœt d ¡|d <|d |d |d |d |d |ddœ}xN|  ¡D]B}||dksÐ|dkrút ||ƒ||<qÐ|dksÐt ||ƒ||<qÐW|t kr˜ttdƒ}| ¡} WdQRXx(| D] } |  d¡rD| d ¡} PqDWx,| D]"} |  d¡sp|  |¡rp|  ¡} | |  d ¡} |ddkrº| |d<| |dkr€|ddkrìt | |  d ¡ƒ|d<|ddkrt | |  d ¡ƒ|d<|ddkr4t | |  d ¡ƒ|d<|ddkrXt | |  d¡ƒ|d<|ddkr|t | |  d ¡ƒ|d<Pn| dkrqpnPqpW|dkr¨d}n$|dkr¸d}n|d krÈd!}nd}dd"d#d#d$d#g}d}x8| ¡D],\}}||dkrê||||<|d#7}qêW||||d|d|d|d|d|df }|S)%NÚjob_submission_schemeÚautoZJOB_SUBMISSION_SCHEMEZlauncher_multiTask_singleNodez hostname -fT)rr‘zutf-8Ú QUEUENAMEÚ CPUS_PER_NODEÚTHREADS_PER_COREÚMAX_JOBS_PER_QUEUEÚWALLTIME_FACTORÚ MEM_PER_NODE)rGrHrIrJrLrKrLrGrqZ PLATFORM_NAMErú#)rõÚNONEr—rHrIrJrKr—)Ú stampede2Úfronterar“rŽ)ÚpegasusrŒ)Ú eos_sanghoonÚbeijing_serverÚ deqing_serverrrär'i€>)Ú"auto_template_not_existing_optionsr*r+r˜r™ršr‘r›rœrrr@rzr!rurrvrr2rÔÚitems)r4rSrDrÚplatformrEZ check_autoÚkeyrróréZ queue_headerZ split_valuesZ default_queuerFZdef_autor®ÚvalueZout_putsr$r$r%rCäs’                  rCcCsndddddddg}|jrPdd lm}||jƒj}x8|D]}||kr6d ||<q6Wni}x|D] }d ||<qZW|S) Nrör÷rørùrúrûrôr)ÚTemplaterõ)r Úminsar.objects.dataset_templater Úoptions)r4Z job_optionsr rSÚoptionr$r$r%rBs      rÚ__main__r'F)&rïr*Úsysr˜rr¹rRÚnumpyrxrÐÚminsar.objectsrÚminsar.objects.auto_defaultsrrÚwarningsZminsar.utils.process_utilitiesÚutilsÚprocess_utilitiesr½rr¡ÚfilterwarningsÚ FutureWarningr&r5r6rÄrCrrìrkZPARAMSÚjob_objr†rr‹rÉr$r$r$r%Ús>  u^