?iTKddlZddlZddlZddlZddlZddlZddlZddlZddlZ ddl Z ddl Z ddl m Z ddlmZmZmZmZdede j&e j(edffdZGd d ej,ZGd d ej0ej2ZGd deZdede j&efdZdedede j&edefdZdefdZ GddejBZ"GddejFZ$ejJde jLee jNffdZ( dGdede jRee fdede jTeded ed!e jTed"e jTed#e jTed$e jTed%e jTed&e jTed'e jTed(e jTe j&ed)e jTe j&ed*e jTed+e jTed,e jTed-ed.e jTed/e jTed0e jTed1e jTed2e jTed3e jTed4e jTed5e jTed6e jTed7e jTe jRe+efd8ed9ed:e+d;e jTede+defJd?Z-d@e.defdAZ/dBedCe jNdefdDZ0dEe j&edefdFZ1y)HN)Path)corejob_environmentloggerutilsjob_idreturn.c fd}tj||}|]|jd}|jdjd}|Dcgc] }t |g|jdz"c}S|jdd^}}|s|fgSt t |d}||fgScc}w) zlReads formated job id and returns a tuple with format: (main_id, [array_index, [final_array_index]) z:(?P\d+)_\[(?P(\d+(-\d+)?(,)?)+)(\%\d+)?\]main_idarrays,-_r)researchgroupsplittuplestrint) r patternmatchmain array_ranges array_ranger array_id array_nums i/Volumes/fast/ai/experiments/voice-extract-mac/.venv/lib/python3.12/site-packages/submitit/slurm/slurm.py read_job_idr!sLG IIgv &E {{9%{{8,2237 JVW,;tf{00556,WW#\\#q1(J< HQK() )$%%Xs%B.ceZdZdejej efdZd dededefdZdeje efdejeejeefffdZ y) SlurmInfoWatcherr c|j|jz Dchc]}|jdd}}|sygd}|D]}|jdt |g |Scc}w)Nrr)sacctz-ozJobID,State,NodeListz --parsable2z-j) _registered _finishedrextendr)selfxto_checkcommandjids r _make_commandzSlurmInfoWatcher._make_command+sj.2-=-=-NO-NAGGCLO-NOHC NND#c(+ , PsA%r modecR|j||}|jdxsdS)aSReturns the state of the job State of finished jobs are cached (use watcher.clear() to remove all cache) Parameters ---------- job_id: int id of the job on the cluster mode: str one of "force" (forces a call), "standard" (calls regularly) or "cache" (does not call) )r/StateUNKNOWN)get_infoget)r)r r/infos r get_statezSlurmInfoWatcher.get_state6s*}}V$}/xx -I-stringc t|ts|j}|j}t |dkriS|dj d}i}|ddD]}t ||j dDcic]\}}||j}}}|d} | rd| vrK t| } | D]a} ||d j| dd<t | d k\s)tt| dt| ddzD]} ||| dd | <c|Scc}}w#t$r*} tjd| d | tYd} ~ d} ~ wwxYw) zNReads the output of sacct and returns a dictionary containing main informationrr|rNJobID.zCould not interpret z# correctly (please open an issue): r) isinstancerdecode splitlineslenrzipstripr! ExceptionwarningswarnDeprecationWarningjoinranger)r)r8linesnames all_statsliner*ystatsr multi_split_job_ide split_job_idindexs r read_infozSlurmInfoWatcher.read_infoDsw&#&]]_F!!# u:>Ias#57 !"ID.1%C.IJ.IdaQ \.IEJ7^FSF] %0%8"!3 #((<#345|$)!&s<?';Sa=QTU=U!VBG \!_$5Qug">?"W !3*)K   *6(2VWXVYZ\n  s=D( D## E, EEN)standard) __name__ __module__ __qualname__tpOptionalListrr.r6UnionbytesDictrTr7r r#r#*su r{{2773<8  . .3 . . 4bggcSVhFWAW9Xr7r#c4eZdZdZedZddeddfdZy) SlurmJobscanceliX)delay_stimeoutr Ncd|jdg}|stj|dgztj|tjgzy)zSends preemption or timeout signal to the job (for testing purpose) Parameter --------- timeout: bool Whether to trigger a job time-out (if False, it triggers preemption) rbz--signalSIGTERMN)r subprocess check_callSlurmJobEnvironmentUSR_SIG)r)rdcmds r _interruptzSlurmJob._interruptjsI$++z2  ! !# "3 4c%8%@%@$AABr7)F)rVrWrX_cancel_commandr#watcherboolrlr_r7r rarafs'Os+G C$ C4 Cr7rac eZdZy)SlurmParseExceptionN)rVrWrXr_r7r rqrqysr7rq suffix_partscg}|jdD]s}d|vr\|jd\}}t|}tt|t|dzD]}|j |d|c|j |u|S)z_Parse the a suffix formatted like "1-3,5,8" into the list of numeric values 1,2,3,5,8. rrr0)rrArIrappend)rrsuffixes suffix_partlowhigh int_lengthnums r _expand_id_suffixr|}sH#))#. + #))#.ICSJSXs4y1}53q n"576 OOK (/ Or7 node_listposparsedcdg}|t|kr||}|dk(r|j||dzS|dk(rD|jd|}t||dz|}|Dcgc]}|D]}||z }}}|dz}n t |D] \} }||z|| <|dz }|t|kr|j||Scc}}w)zzParse a node group of the form PREFIX[1-3,5,8] and return the position in the string at which the parsing stopped rr[])rAr(rSr| enumerate) r}r~rprefixesclast_posrvprefixsuffixis r _parse_node_grouprstH I  cN 8 MM( #7N 8 sC0H(37X)FGH6>VhFX6XhHVQ,C&x0 6$qj 1 1HC I  MM( JWsB=c d}g}|t|krt|||}|t|kr|S#t$r}td|d||d}~wwxYw)Nrz-Unrecognized format for SLURM_JOB_NODELIST: '')rAr ValueErrorrq)r}r~rrQs r _parse_node_listrsoj!C N"#IsF;CC N" j!$QR[Q\\]"^`abhiijs-22 AA  Ac deZdZddddddddd d Zd ed d fdZed eje fdZ y )ri SLURM_JOB_ID SLURM_NTASKSSLURM_JOB_NUM_NODES SLURM_NODEIDSLURM_JOB_NODELIST SLURM_PROCID SLURM_LOCALIDSLURM_ARRAY_JOB_IDSLURM_ARRAY_TASK_ID) r num_tasks num_nodesnodenodes global_rank local_rank array_job_id array_task_id countdownr Nc|j}tjdd|gdtjj d|d|dy)Nscontrolrequeue<)rdz Requeued job z (z remaining timeouts))r rgrhr get_loggerr5)r)rr-s r _requeuezSlurmJobEnvironment._requeuesIkkz9c:BG  =R {BV!WXr7ctjj|jdd}|s |jgSt |S)aParse the content of the "SLURM_JOB_NODELIST" environment variable, which gives access to the list of hostnames that are part of the current job. In SLURM, the node list is formatted NODE_GROUP_1,NODE_GROUP_2,...,NODE_GROUP_N where each node group is formatted as: PREFIX[1-3,5,8] to define the hosts: [PREFIX1, PREFIX2, PREFIX3, PREFIX5, PREFIX8]. Link: https://hpcc.umd.edu/hpcc/help/slurmenv.html rr)osenvironr4_envhostnamer)r)r}s r hostnameszSlurmJobEnvironment.hostnamess;JJNN499W#5r: MM? " **r7) rVrWrXrrrpropertyrYr[rrr_r7r riris] #*%%%,. DY#Y$Y +2773<++r7ric eZdZdZeZ ddejee fde de dejeddf fd Z edej fd Zedej$efd Zd ej(eej*fdej(eej*fffd Zd ej*ddffd Zdej0ej4dej0ej6ej*ffd ZedefdZdededefdZde fdZ de dej0efdZ!e"deje#efdefdZ$ede fdZ%xZ&S) SlurmExecutoraPSlurm job executor This class is used to hold the parameters to run a job on slurm. In practice, it will create a batch file in the specified directory for each job, and pickle the task function and parameters. At completion, the job will also pickle the output. Logs are also dumped in the same directory. Parameters ---------- folder: Path/str folder for storing job submission/output and logs. max_num_timeout: int Maximum number of time the job can be requeued after timeout (if the instance is derived from helpers.Checkpointable) python: Optional[str] Command to launch python. This allow to use singularity for example. Caller is responsible to provide a valid shell command here. By default reuse the current python executable Note ---- - be aware that the log/output folder will be full of logs and pickled objects very fast, it may need cleaning. - the folder needs to point to a directory shared through the cluster. This is typically not the case for your tmp! If you try to use it, slurm will fail silently (since it will not even be able to log stderr. - use update_parameters to specify custom parameters (n_gpus etc...). If you input erroneous parameters, an error will print all parameters available for you. Nfoldermax_num_timeoutmax_pickle_size_gbpythonr ct|||||#tjtj n||_|jdkDs tdy)N)rrrz;Could not detect "srun", are you indeed on a slurm cluster?) super__init__shlexquotesys executableraffinity RuntimeError)r)rrrr __class__s r rzSlurmExecutor.__init__s[  +1  6<^ekk#..1 }}"\] ]#r7cddddddddS) Njob_nametimememr cpus_per_task gpus_per_nodentasks_per_node)name timeout_minmem_gbrrrtasks_per_noder_clss r _equivalence_dictzSlurmExecutor._equivalence_dicts#!,,/  r7c(ttS)4Parameters that can be set through update_parameters)set_get_default_parametersrs r _valid_parameterszSlurmExecutor._valid_parameterss*,--r7paramscPt||}d|vrt|d|d<|S)Nr)r_convert_parameters _convert_mem)r)rrs r rz!SlurmExecutor._convert_parameterss0,V4 F?(7F5M r7kwargsc ,t}tt|t|z }|r@djdt|j D}t d|d|t dd|jd|t|$di|y) aUpdates sbatch submission file parameters Parameters ---------- See slurm documentation for most parameters. Most useful parameters are: time, mem, gpus_per_node, cpus_per_task, partition Below are the parameters that differ from slurm documentation: signal_delay_s: int delay between the kill signal and the actual kill of the slurm job. setup: list a list of command to run in sbatch befure running srun array_parallelism: int number of map tasks that will be executed in parallel Raises ------ ValueError In case an erroneous keyword argument is added, a list of all eligible parameters is printed, with their default values Note ---- Best practice (as far as Quip is concerned): cpus_per_task=2x (number of data workers + gpus_per_task) You can use cpus_per_gpu=2 (requires using gpus_per_task and not gpus_per_node) z - c3DK|]\}}|dt|dyw)z (default: )N)repr).0r*rNs r z.9s("dKc41aaS DG9A#>Kcs zUnavailable parameter(s): z Valid parameters are: - z nothing to dor,rNr_) rsortedrrHitemsr_make_sbatch_stringrr_internal_update_parameters)r)rdefaultsin_valid_parametersr8rs r rz)SlurmExecutor._internal_update_parameterss6+,$S[3x=%@A ]]"d6RZR`R`RbKc"ddF,-@,AA^_e^fg  RODKKR6R +5f5r7delayed_submissionsc t|dk(rt| |Stjj |j }|jdd|jjdd}g}|D]d}|tjjdz }|j||j|j||j!|ft|}t#|j |j}|j$d i|j||jd<|j'|j)|j*} t-t/| j0} t/|D cgc])} t3|j | j4d| | +} } t7| |D]!\} }| j8j;|d #| Scc} w) NrT)parentsexist_okrz.pkl map_countr)rr taskssubmitted_pickler_)rAr_internal_process_submissionsrJobPathsget_first_id_independent_folderrmkdir parametersr4uuiduuid4hex set_timeoutrdumprurupdate_parameters _throttle_submit_command_submitit_command_strlistrIrrar rBpathsmove_temporary_file)r)rrr pickle_pathsd pickle_pathnarray_ex first_job tasks_idsajobsjobrs r rz+SlurmExecutor._internal_process_submissionsAs " #q (789LM M?? L TD 1oo))&!4  $A djjl&6&6%7t#<&>t?Y?Y&Z y2234 ejklem+ em`aHDKK93C3C2DAaS0IQZ [em + !$D, 7 C II ) )+7I J!8 + s.G0c dj|jdtjt |j gS)N z-u -m submitit.core._submit)rHrrrrr)r)s r rz#SlurmExecutor._submitit_command_stras3xx&CU[[QTUYU`U`QaEbcddr7r,uidcFtd||jd|jS)Nrr_)rrr)r)r,r s r _make_submission_file_textz(SlurmExecutor._make_submission_file_textes"Z74;;Z$//ZZr7c|jjdd}td|jjdd}||zS)Nrrr)rr4max)r)rrs r _num_taskszSlurmExecutor._num_taskshsA__((!4!!T__%8%89JA%NO~%%r7submission_file_pathcdt|gS)Nsbatch)r)r)rs r _make_submission_commandz&SlurmExecutor._make_submission_commandms#2344r7r8ct|ts|j}tjd|}|t j d|d|jdS)z3Returns the job ID from the output of sbatch stringzjob (?P[0-9]+)z'Could not make sense of sbatch output "zg" Job instance will not be able to fetch status (you may however set the job job_id manually if needed)id)r>rr?rrrFailedSubmissionErrorr)r8outputs r #_get_job_id_from_submission_commandz1SlurmExecutor._get_job_id_from_submission_commandpsg&#&]]_F0&9 >--9&BJJ  ||D!!r7c4tjddSdS)Nsrunr)shutilwhichrs r rzSlurmExecutor.affinity~s\\&)1r8q8r7)r=g?N)'rVrWrX__doc__ra job_classrYr\rrrfloatrZr classmethodrEquivalenceDictrSetrr^Anyrrr[rDelayedSubmissionJobrrrr rr staticmethodr]rr __classcell__)rs@r rrs:I !$'#' ^d#^^" ^  C ^  ^   $"6"6    ."&&+.."''#rvv+*>2773PRPVPV;CW$6BFF$6t$6L#%775+B+B#C "&&! "@esee[#[C[C[&C& 5T5bggcl5 "BHHUCZ4H "S " "999r7rctjt}t|jt |j  d|j }|Dcic] \}}|dvs ||c}}Scc}}w)rN>rr,r)inspectgetfullargspecrrBargsrAr)specszippedkeyvals r rrse  " "#6 7E S0023U^^ DF%+ ]Vcs:\/\CHV ]] ]s  A,#A,r,rr partitionrrrr cpus_per_gpunum_gpusr gpus_per_taskqossetupteardownr mem_per_gpu mem_per_cpusignal_delay_scomment constraintexcludeaccountgres mail_type mail_usernodelist dependency exclusivearray_parallelismwckeystderr_to_stdoutradditional_parameters srun_argsuse_srunc$gd}$tjD%&cic]\}%}&|& |%|$vs|%|&}'}%}&tjd||'d<| *t j d|'j dd|'d<d|'vrd |'vrt j d tj| }(t|(j})t|(j}*| Pt| tr| sJd | d z dt| ||'d<|)jdd})|*jdd}*|)jdd|'d<|s|*jdd|'d<d|'d<|!|'j!|!gd}+t#|'D] }%|+j%t'|%|'|%"| |+ddg| zz }+|#r2|rgnd|*g},|"g}"t)ddd|)g|,|"}-dj+|-|f}|+dd d!|dgz }+| |+dd"g|zz }+d#j+|+Scc}&}%w)$aCreates the content of an sbatch file with provided parameters Parameters ---------- See slurm sbatch documentation for most parameters: https://slurm.schedmd.com/sbatch.html Below are the parameters that differ from slurm documentation: folder: str/Path folder where print logs and error logs will be written signal_delay_s: int delay between the kill signal and the actual kill of the slurm job. setup: list a list of command to run in sbatch before running srun teardown: list a list of command to run in sbatch after running srun map_size: int number of simultaneous map/array jobs allowed additional_parameters: dict Forces any parameter to a given value in sbatch. This can be useful to add parameters which are not currently available in submitit. Eg: {"mail-user": "blublu@fb.com", "mail-type": "BEGIN"} srun_args: List[str] Add each argument in the list to the srun call Raises ------ ValueError In case an erroneous keyword argument is added, a list of all eligible parameters is printed, with their default values ) nonslurmrr,rrErHr6r7r:rGrIrJ@signalzY"num_gpus" is deprecated, please use "gpus_per_node" instead (overwritting with num_gpus)r3rrr2r4zP"cpus_per_gpu" requires to set "gpus_per_task" to work (and not "gpus_per_node"))rz0-r%arrayz%jz%A_%az%trtrerrorruz open-mode)z #!/bin/bashrz # Parametersrz# setupz--errorrz --unbufferedz--outputr z # commandzexport SUBMITIT_EXECUTOR=slurmz # teardown )localsrrirjrErFpoprrrstdoutstderrr>rminreplaceupdaterru_as_sbatch_flag _shlex_joinrH).r,rrr1rrrrr2r3rr4r5r6r7rr8r9r:r;r<r=r>r?r@rArBrCrDrErFrGrrHrIrJrLkvrrrUrVrJ stderr_flagssrun_cmds. r rrs]L H$*8>>#3[#341aq}RZIZ!Q$#3J[199:!N;KLJx g '1nnZ&C ?##z(I hi NN& )E  F  F)S)i77 "9q=/3yBS3T2UV 7g.g.!>>$4Jx $nnT37 7&J{(/0 /E J  _Q 1 67   "i5((.rIv3F  I Ff\f\efg((Hg./  ( E "l#h.. 99U u\s HHHrc\|t|k(rt|dSt|dzdS)NGBiMB)r)rs r rrs6 Vf+b!!&4- ! $$r7r/valuec|jdd}|durd|Stjt|}d|d|S)NrrTz #SBATCH --=)rXrrr)r/rcs r rZrZ#sK ++c3 C }C5!! KKE #E uAeW %%r7 split_commandc2djd|DS)z9Same as shlex.join, but that was only added in Python 3.8r c3FK|]}tj|yw)N)rr)rargs r rz_shlex_join...s> EKK$ s!)rH)rfs r r[r[,s 88> > >>r7)"submititNrrNNNNNNNNNNNNZNNNNNNNNNNrjFNNNT)2 functoolsr*rrrrrgrtypingrYrrEpathlibrrrrrrr[Tupler! InfoWatcherr#r&RrarDrqr|rrrJobEnvironmentriPicklingExecutorr lru_cacher^r$rr\rZroIterablerr rrZr[r_r7r rws   77&&c(: ;&&9t''9xCtxxC& )  C BGGCL  3 .jj!+/88!+Hm9D))m9`^bff!5^^"&(,&*%)!%&*&* '+*. $($( $#' $ $!"&"&!%#'26 ""&?C/3IN N HHS$Y NN{{3 N  N  N[[%N;;s#N++c"Nkk#N;;s#N;;s#N S N ;;rwws| $Nkk"''#,'N S !N"S!#N$S!%N&'N([[ )N* C +N,[[ -N.[[ /N0 ++c 1N2{{33N4{{35N6kk#7N8 C 9N:{{288D#I./;N<=N> ?N@ANB{{3CND;;rwwsBFF{';