o rri)9@sdZddlmZmZddlmZmZddlZddlm Z ddl Z ddl Z ddl Z ddlZddlZddlmZddlZddlmZdd lmZmZdd lmZdd lmZdd lmZdd lmZm Z e!e"Z#GdddZ$GdddZ%de&fddZ'eGdddZ(GdddZ)dS)z)Scheduling and job monitoring utilities. )contextmanager ExitStack) dataclassfieldN)Path)SlurmJob)git_save) SlurmConfig SubmitRules)get_distrib_spec) DecoratedMain)try_load)XP_get_sigc@s,eZdZdedejefddZddZdS)_SubmitItTargetmainargvcCs:|||_t}t|jtjd<|tjdd<|dS)NRANKr) get_xpxpr strrankosenvironsysr)selfrrspecr=/home/ubuntu/.local/lib/python3.10/site-packages/dora/shep.py__call__#s  z_SubmitItTarget.__call__cOs>tjdkr|jjr|jjtjj|g|Ri|S)Nr) r rrrendezvous_fileexistsunlinksubmitithelpersDelayedSubmission)rargskwargsrrr checkpoint,s   z_SubmitItTarget.checkpointN) __name__ __module__ __qualname__r tpSequencerr r)rrrrr"s rc@sXeZdZdZdefddZedefddZdd d Z dd d Z ed dZ ddZ dS)Sheepz[ A Sheep is a specific run for a given XP. Sheeps are managed by the Shepherd. rcCsN||_d|_d|_|jr%t|j}t|tr |\|_|_dS||_dSdSN)rjob _other_jobs _job_filer"r isinstancetuple)rrcontentrrr__init__9s    zSheep.__init__returncCs|jj|jjjjSr0)rfolderdorashepjob_filerrrrr3EszSheep._job_filestandardcCsZ|jdurdS|jj|jj|}|dkr$|jr$tdd|jDr$d}|dr+dS|S)z1Return the current state of the `Sheep`. NUNKNOWNcss|]}|jdkVqdS)r?N)state.0r1rrr PszSheep.state..MISSING CANCELLED)r1watcher get_statejob_idr2any startswith)rmoder@rrrr@Is  z Sheep.statecCs"|jdurdS|jj|jj|S)zDReturn True if the job is no longer running on the cluster. NT)r1rFis_donerH)rrKrrrrLYs z Sheep.is_donecCs$|jdur|jj|jjdSdS)z)Return the path to the main log. Nz _0_log.out)r1rr$rHr=rrrlog`s z Sheep.logcCsPd|jjd|d}|jdur|d|jjd7}|d|jjd7}|S)NzSheep(z, state=z, zsid=zargv=))rsigr@r1rHr)routrrr__repr__hs  zSheep.__repr__N)r>) r*r+r,__doc__rr7propertyrr3r@rLrMrQrrrrr/4s     r/xcCsdS)z7No logging logging function, passed to `Shepherd`. Nr)rTrrrno_logqsrUc@s.eZdZUeed<eedZej e ed<dS) _JobArray slurm_config)default_factorysheepsN) r*r+r,r __annotations__rlistrYr-Listr/rrrrrVws rVc@sPeZdZdZefdedejegdffddZ dej ede fd d Z d edej e fd d Zdedej e fddZddZedefddZde dedefddZdejfddZddZedefdd Zedefd!d"Zedefd#d$Zd%ejefd&d'Z d(ed)ededej!fd*d+Z"d,d-Z#ed(efd.d/Z$d0e%fd1d2Z&dS)3Shepherdz Takes care of the little jobs. Args: main (DecoratedMain): main function decorated by Dora. log (callable): log function, if provided should take a single string argument. rrMNcCs`||_|jjddd|jjddd|jjddd||_d|_d|_g|_g|_ | dS)NT)exist_okparentsF) r_by_idmkdir_orphans_arraysrM _in_job_array_existing_git_clone _to_cancel _to_submit_check_orphans)rrrMrrrr7s zShepherd.__init__rr8cCs"t|trJ|j|}t|S)z Given a list of of arguments, return the matching `Sheep`, which will contain both information on the `dora.xp.XP`, and on the latest job associated with that XP. )r4rrrr/)rrrrrrget_sheep_from_argvs zShepherd.get_sheep_from_argvrOcCs|j|}t|S)zj Returns a `Sheep` given the XP signature, if any exists, otherwise returns None. )rget_xp_from_sigr/)rrOrrrrget_sheep_from_sigs zShepherd.get_sheep_from_sigrHcCs4|j|}|r|j}|j|}t|SdS)zu Returns the `Sheep` associated with the given `job_id`. If no sheep is found, returns None. N)r` is_symlinkresolvenamerrjr/)rrHlinkrOrrrrget_sheep_from_job_ids   zShepherd.get_sheep_from_job_idcCstjdS)zB Force an update of all job states with submitit. N)rrFupdater=rrrrqszShepherd.updaterWccs>|jrJ|jt|d|_z dVWd|_dSd|_w)z*Context manager to launch XP in job array.TNF)rdrgappendrV)rrWrrr job_arrays zShepherd.job_arraysheeprulescCs|jdurM|}|dkr|jrtd|jjd|_n/|dvr4td|jjd|jr3d|_n|jrMtd|jjd|||jd|_|jdurr|j s]|j t |||j d j ksgJ|j d j |dSdS) z Decides whether to schedule a new job for the given sheep, based on the rules given in `rules`. Jobs are actually only scheduled once the `commit()` method is called. N COMPLETEDz"Ignoring previously completed job )FAILEDrE OUT_OF_MEMORYTIMEOUTrD NODE_FAILz Previous job z failed or was canceledzCancelling previous job z with status )r1r@ replace_doneloggerdebugrHretryreplace cancel_lazyrdrgrrrVrWrY)rrtrWrur@rrrmaybe_submit_lazys,   zShepherd.maybe_submit_lazyr1cCs|j|dS)z] Cancel a job. The job is actually cancelled only when `commit()` is called. N)rfrr)rr1rrrrszShepherd.cancel_lazycCsH|jr ||jg|_d|_|jr"|jd}|||jsdSdS)zu Commit all changes registered so far with either `maybe_submit_lazy()` and `cancel_lazy()`. Nr)rf_cancelrergpop_submit)rrsrrrcommits   zShepherd.commitcC|jjj|jjjjSr0)rr:dirr;by_idr=rrrr`zShepherd._by_idcCrr0)rr:rr;orphansr=rrrrbrzShepherd._orphanscCrr0)rr:rr;arraysr=rrrrcrzShepherd._arraysjobscCs8dgdd|D}tdd|tj|dddS)NscancelcSsg|]}|jqSr)rHrArrr sz$Shepherd._cancel..z Running %s Tcheck)r}r~joinsprun)rr cancel_cmdrrrrszShepherd._cancelrnr9c Csdtjd<t|j}tj||dd}|j}|dkr.|ddkr%td|d|d<d}n|}d |d<|j }|rE|j |} | d |d <d ||d <|j r`d |d<|j dur_||j |d<n||d<|j durn|j |d<|d=|d=|d=|d=t d||jd|dd||S)N1SLURM_KILL_BAD_EXITmax_num_timeout)r9rrz.Can only take <= 8 gpus, or multiple of 8 gpusnodesrGBmemzgpu:gresntasks_per_node cpus_per_taskgpus mem_per_gpu cpus_per_gpuone_task_per_nodezSlurm parameters %rT)job_namestderr_to_stdoutr)rrdict__dict__r$ SlurmExecutorrr ValueErrorrrrrr}r~update_parameters) rrnr9rWr(executorr gpus_per_noderrrrr_get_submitit_executorsL          zShepherd._get_submitit_executorc Cs|jD]H}|j}td|dtjddtd|dddgd d d }d d |j d D}|rItd|dtjdg|d d| qdS)zCheck for orphaned jobs.zFound dirty tag z`, meaning a job might have been scheduled but Dora or Slurm crashed before the job id was saved.squeuez-uz-nz-oz%iz-hT)capture_outputrcSsg|]}|r|qSrr)rBlinerrrr5z+Shepherd._check_orphans.. zFound orphan job ids z , will cancelrrN)rbiterdirrnr}warningrrrgetloginstdoutdecodestripsplitr#)rdirtyrnprocidsrrrrh-s zShepherd._check_orphansccs4|j|}|z dVW|dS|w)z,Context manager to enter a potential orphan.N)rbtouchr#)rrntokenrrr _enter_orphan;s  zShepherd._enter_orphanrsc s|j}|j}|s dSt|dk}|d}|j|j|jjjtfdd|Ds/Jd|r=t t dd|D}n|jj }|rL|jj d|}n|jj d |}|r\|j |}n|jj}|jd d |D]} | j} |j| | jr}| jqh||||} g} r|jdurt|j|_||tH} r|jdusJ| t|j|r| | |jD]!} r|jdusJt| j|j| | t|j| jjqWdn1swYt || D]x\} }t!"|| ft#| j$d t%&d |j'|| _(| | _)|j*|j'}|}|+| jj,-|rA| jj,|j }|r<|-|-ks;Jn|+|| jj.}|rN||+||j/| j}|0d|j'd| jj d|qWddS1sxwYdS)Nrrc3s|] }|jjjkVqdSr0)rr:r )rBother use_git_saverrrCOsz#Shepherd._submit..z?All jobs inside an array must have the same value for git_save.cSsg|]}|jjqSr)rrO)rBrtrrrrSrz$Shepherd._submit.._array__T)r^wbzCreated job with id %szScheduled job z for sheep /)1rYrWlenrinit_xprr:r allrsortedrOrnrc _xp_submititrar!r"r#rre get_new_clonerr enter_context enter_clonebatch assign_clonerrsubmitrrzippickledumpopenr3r}r~rHr1r2r` symlink_tor9rm_latest_submititget_namerM)rrsrYrWis_arrayfirstname_sigrnsubmitit_folderrtrrrstackr1ro submitit_linklatestrrrrEs               $$zShepherd._submit)'r*r+r,rRrUr r-Callablerr7r.r/riOptionalrkrprqrr rsr rr$rrrrSrr`rbrcr\rrrrhrrVrrrrrr]}s8"     ( r])*rR contextlibrr dataclassesrrloggingpathlibrrr subprocessrrtypingr-r$rr confr r distribr rr utilsrrrr getLoggerr*r}rr/rrUrVr]rrrrs2       =