o rri@sddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z ddlmZmZejdddZGd d d ZGd d d ejeZGd ddeeZGdddejeZdZGddde ZGdddejZGdddeZdS)N)Path) TypedDict)loggerutilsRT) covariantc @s eZdZdZd deddfddZdejee fdej e ej e e fffd d Z dej ej e fd d Zd!de de de fddZedefddZd"ddZd!de de dej e e ffddZd!de de defddZde ddfddZd"ddZde ddfddZdS)# InfoWatcheraAn instance of this class is shared by all jobs, and is in charge of calling slurm to check status for all jobs at once (so as not to overload it). It is also in charge of dealing with errors. Cluster is called at 0s, 2s, 4s, 8s etc... in the begginning of jobs, then at least every delay_s (default: 60) Parameters ---------- delay_s: int Maximum delay before each non-forced call to the cluster. <delay_sreturnNcCs<||_t|_t|_i|_d|_d|_td|_d|_ dS)N-infr) _delay_sset _registered _finished _info_dict_output _start_timefloat_last_status_check _num_calls)selfr rF/home/ubuntu/.local/lib/python3.10/site-packages/submitit/core/core.py__init__'s  zInfoWatcher.__init__stringcCtNNotImplementedError)rrrrr read_info1zInfoWatcher.read_infocCrr r!rrrr _make_command4r$zInfoWatcher._make_commandstandardjob_idmodecCrr r!rr(r)rrr get_state7r$zInfoWatcher.get_statecC|jS)zNumber of calls to sacct)rr%rrr num_calls:zInfoWatcher.num_callscCs,t|_t|_td|_i|_d|_dS)zsClears cache. This should hopefully not be used. If you have to use it, please add a github issue. rr N) rr_timetimerrrrrr%rrrclear?s    zInfoWatcher.clearcCs<|durtd||jvr|||||j|iS)adReturns a dict containing info about the job. State of finished jobs are cached (use watcher.clear() to remove all cache) Parameters ---------- job_id: str id of the job on the cluster mode: str one of "force" (forces a call), "standard" (calls regularly) or "cache" (does not call) Nz$Cannot call sacct without a slurm id) RuntimeErrorr register_jobupdate_if_long_enoughrgetr*rrrget_infoIs     zInfoWatcher.get_infocCs"|j||d}gd}||vS)aReturns whether the job is finished. Parameters ---------- job_id: str id of the job on the cluster mode: str one of "force" (forces a call), "standard" (calls regularly) or "cache" (does not call) r))READYPENDINGRUNNINGUNKNOWNREQUEUED COMPLETING PREEMPTED)r+upper)rr(r)state incompleterrris_done\s  zInfoWatcher.is_donecCsn|dvsJ|dkr dSt|j}t|j}t|jtd|d}|dkr+d}||kr5|dSdS)zUpdates if forced to, or if the delay is reached (Force-updates with less than 1ms delay are ignored) Also checks for finished jobs )r'forcecacherDNrCgMbP?)r/r0rrminrmaxupdate)rr)last_check_deltalast_job_delta refresh_delayrrrr4js  z!InfoWatcher.update_if_long_enoughc Cs|}|dur dS|jd7_ztd|jdd|tj|dd|_ Wn!t yM}zt d|jd|d WYd}~nd}~ww|j ||j t|_|j|j}|D]}|j|d d rt|j|qedS) zz Job.__init__..)r(task_id)_job_idtuple_tasks _sub_jobs_cancel_at_deletionlenrJobPathsrx_pathsr/r0rr_register_in_watcher)rrnr(rorrurrs    z Job.__init__cCs |jds|j|jdSdS)Nr)r{watcherr3r(r%rrrrs zJob._register_in_watchercCr,r )ryr%rrrr(z Job.job_idcCr,r )rr%rrrpathsrz Job.pathscCs|jsdSt|jS)z&Returns the number of tasks in the Jobr)r|r~r%rrr num_taskss z Job.num_taskscCs.|jjsJd|jjtj|jjS)zMReturns the submitted object, with attributes `function`, `args` and `kwargs`z#Cannot find job submission pickle: )rsubmitted_pickleexistsrDelayedSubmissionloadr%rrr submissions  zJob.submissionTvaluezJob[R]cCs ||_|S)aBSets whether the job deletion in the python environment triggers cancellation of the corresponding job in the cluster By default, jobs are not cancelled unless this method is called to turn the option on. Parameters ---------- value: bool if True, the cluster job will be cancelled at the instance deletion, if False, it will not. Returns ------- Job the current job (for chaining at submission for instance: "job = executor.submit(...).cancel_at_deletion()") )r})rrrrrcancel_at_deletionszJob.cancel_at_deletionrxcCsFd|kr |jksntd|d|jd|js|S|j|S)aeReturns a given sub-Job (task). Parameters ---------- task_id The id of the task. Must be between 0 and self.num_tasks Returns ------- job The sub_job. You can call all Job methods on it (done, stdout, ...) If the job doesn't have sub jobs, return the job itself. rztask_id z must be between 0 and r)r ValueErrorr|)rrxrrrtasks   zJob.taskcheckcCs&|rtjntj|j|jgdddS)zCancels the job Parameters ---------- check: bool whether to wait for completion and check that the command worked FrMN)rR check_callcall_cancel_commandr()rrrrrcancels z Job.cancelcCs|}|jr Jd|dS)Nz4You should use `results()` if your job has subtasks.r)resultsr|)rrrrrresult sz Job.resultcCsT||jrdd|jDS|\}}|dkr'|}|dur%td||gS)a_Waits for and outputs the result of the submitted function Returns ------- output the output of the submitted function. If the job has several tasks, it will return the output of every tasks in a List Raises ------ Exception Any exception raised by the job cSsg|] }tt|qSr)rbcastrrrssub_jobrrrrv szJob.results..errorNzUnknown job exception)waitr|_get_outcome_and_result exceptionr2)routcomer job_exceptionrrrrs z Job.resultscCs||jrdd|jD}dd|D}|sdS|dSz|\}}Wntjy;}z|WYd}~Sd}~ww|dkr^td|jd|d |jd |jd |jjd |jj SdS) aWaits for completion and returns (not raise) the exception containing the error log of the job Returns ------- Exception/None the exception if any was raised during the job. If the job has several tasks, it returns the exception of the task with smallest id that failed. Raises ------ UncompletedJobError In case the job never completed cSg|]}|qSr)rrrrrrv=z!Job.exception..cSg|]}|dur|qSr r)rsrXrrrrv?sNrrz Job (task=z>) failed during processing with trace: ---------------------- zA ---------------------- You can check full logs with 'job.stderr(z)' and 'job.stdout(z)'or at paths: - z - ) rr|rrUncompletedJobErrorFailedJobErrorrxrstderrstdout)rall_exceptions exceptionsrtracerXrrrr*s< z Job.exceptioncCs|jrJd|jj}|j}z ||jWn ty%|d9}Ynwt }|jj sKt ||krKt d|jj sKt ||ks8|jj sd|j d|jd|jj d|jdg}|}|rv|d d |gn,|jj rtjd d t|jjgd d}|d|jjd |gn |d|jjtd|z t|jj }W|Stytd|jj dt dt|jj }Y|Sw)avGetter for the output of the submitted function. Returns ------- outcome the outcome of the job: either "error" or "success" result the output of the submitted function Raises ------ UncompletedJobError if the job is not finished or failed outside of the job (from slurm) z(This should not be called for a meta-jobrErzJob z (task: z ) with path z$has not produced any output (state: )zError stream produced:z(----------------------------------------tailz-40zutf-8)encodingz*No error stream produced. Look at stdout: z)No output/error stream produced ! Check:  zEOFError on file z, trying again in 2s)r|rrn_results_timeout_schmodstatst_modePermissionErrorr/r0 result_picklersleepr(rxr@rextendrrRrSr[appendrrrQ pickle_loadEOFErrorwarningswarn)rptimeout start_waitmessagelogoutputrrrrTsF        zJob._get_outcome_and_resultcCs"|std|rdSdS)zWait while no result find is found and the state is either PENDING or RUNNING. The state is checked from slurm at least every min and the result path every second. rN)doner/rr%rrrrs zJob.waitF force_checkcCs||jr tdd|jDS|jj}z ||jWn ty$Ynw|jj r-dS|j j |j |r6dnddrszJob.done..TrCr'r7F) r|allrrnrrrOSErrorrrrrBr()rrrrrrrs  zJob.donecCst|jdkr dS|jdS)Nrr)r~r{r%rrrrxsz Job.task_idcCs|jj|jddS)z+State of the job (does not force an update)r'r7)rr+r(r%rrrr@sz Job.staterCr)cCs|jj|j|dS)z9Returns informations about the job as a dict (sacct call)r7)rr6r()rr)rrrr6sz Job.get_infonamecCs|jj|jjd}||vrtd|dt|||s#dS||d }|}Wd|S1s:wY|S)zReturns a string with the content of the log file or None if the file does not exist yet Parameter --------- name: str either "stdout" or "stderr" )rrz Unknown "z", available are Nr) rrrrlistkeysropenread)rrrfrrrr_get_logs_strings    zJob._get_logs_stringcC@|jrdd|jD}dd|D}|sdSd|S|dS)zoReturns a string with the content of the print log file or None if the file does not exist yet cSrr)rrrrrrvrzJob.stdout..cSrr rrssrrrrvNrrr|rQr)rstdout_stdout_not_nonerrrrs  z Job.stdoutcCr)zoReturns a string with the content of the error log file or None if the file does not exist yet cSrr)rrrrrrvrzJob.stderr..cSrr rrrrrrvrNrrr)rstderr_stderr_not_nonerrrrs  z Job.stderrAsyncJobProxy[R]cCst|S)zSReturns a proxy object that provides asyncio methods for this Job. ) AsyncJobProxyr%rrr awaitablesz Job.awaitablec Csld}z|j}Wnty#}ztd|WYd}~nd}~ww|jjd|jd|jd|dS)Nr;zBypassing state error: z) r@rTrrOrUrrr]r(rx)rr@rXrrr__repr__s  $z Job.__repr__cCs0|jr|jj|jdds|jdddSdSdS)NrDr7F)r)r}rrBr(rr%rrr__del__s z Job.__del__cCr,r )__dict__r%rrr __getstate__szJob.__getstate__r@cCs|j||dS)z7Make sure jobs are registered when loaded from a pickleN)rrHr)rr@rrr __setstate__s  zJob.__setstate__)rmr\)TF)rC)r r)5r]r^r_r`rrr rrbrcrr[Sequencerarrrhr(rrrrrrrirrrrrrgrrfrrrTupleAnyrrrrxr@rer6rrrrrrrrrrrrrjsH ,  "* 6 &  "rjc@sTeZdZdZdddZdedejfdd Zd e ejdd fd d Z defddZ d S) DelayedJoba Represents a Job that have been queue for submission by an executor, but hasn't yet been scheduled. Typically obtained by calling `ex.submit` within a `ex.batch()` context Trying to read the attributes of the job will, by default, fail. But if you passed `ex.batch(allow_implicit_submission=True)` then the attribute read will in fact force the job submission, and you'll obtain a real job instead. exExecutorcC ||_dSr )_submitit_executor)rrrrrrs zDelayedJob.__init__rr cCsV|dkrdS|jd}|jstd||jtks&Jd|d|dt||S)Nr}FrzMAccesssing job attributes is forbidden within 'with executor.batch()' contextz Executor z didn't properly submitted z !)r_allow_implicit_submissionsAttributeError_submit_delayed_batchrrrgetattr)rrrrrr __getattr__s   zDelayedJob.__getattr__new_jobNcCs(|jdd|j|j|j|_dS)Nr)rpoprHrr)rrrrr_promote/s zDelayedJob._promotecCs t|Sr )objectrr%rrrr6 zDelayedJob.__repr__)rr) r]r^r_r`rr[rbrrrjrrrrrrr s  rc@seZdZdeefddZddejee fddfdd Z ddejee fdefd d Z ddejee fdej efd d Z ddejee fdejejfddZdS)rjobcCrr )r)rrrrrr;rzAsyncJobProxy.__init__r poll_intervalr Ncs.|jst|IdH|jrdSdS)z&same as wait() but with asyncio sleep.N)rrasynciorrrrrrr>s zAsyncJobProxy.waitc||IdH|jS)aasyncio version of the result() method. Wait asynchornously for the result to be available by polling the self.done() method. Parameters ---------- poll_interval: int or float how often to check if the result is available, in seconds N)rrrrrrrrCs zAsyncJobProxy.resultcr)a'asyncio version of the results() method. Waits asynchornously for ALL the results to be available by polling the self.done() method. Parameters ---------- poll_interval: int or float how often to check if the result is available, in seconds N)rrrrrrrrNs zAsyncJobProxy.resultsc#sJjjdkrtfddtjjDEdHtVdS)aZawaits for all tasks results concurrently. Note that the order of results is not guaranteed to match the order of the tasks anymore as the earliest task coming back might not be the first one you sent. Returns ------- an iterable of Awaitables that can be awaited on to get the earliest result available of the remaining tasks. Parameters ---------- poll_interval: int or float how often to check if the result is available, in seconds (see https://docs.python.org/3/library/asyncio-task.html#asyncio.as_completed) rcs"g|] }j|qSr)rrrr)rsirrrrrvms"z6AsyncJobProxy.results_as_completed..N)rrr as_completedrange ensure_futurerrrrrresults_as_completed\s   z"AsyncJobProxy.results_as_completed)r)r]r^r_rjrrrbrcrarrrrgrIteratorrFuturerrrrrr:s $ *rzInteractions with jobs are not allowed within "with executor.batch()" context (submissions/creations only happens at exit time).c@sJeZdZUdZeed<eed<eed<eed<eed<eed<eed<d S) EquivalenceDictz@Gives the specific name of the params shared across all plugins.r timeout_minmem_gbnodes cpus_per_task gpus_per_nodetasks_per_nodeN)r]r^r_r`r[__annotations__rrrrrzs  rc @seZdZUdZeZejeeje d<d-dej e e fdej eje ejffddZede fd d Zejd.d edejdfd dZd/ddZdejdefdejdejdeefddZejdejejdejeejfddZ dejdefdej!ejdejeefddZ"dej#ejgefdejeefddZ$dejddfd d!Z%edej e&fd"d#Z'edej(e fd$d%Z)d&eje ejfdeje ejffd'd(Z*dejddfd)d*Z+ede,fd+d,Z-dS)0rzBase job executor. Parameters ---------- folder: Path/str folder for storing job submission/output and logs. job_classNrn parameterscCs4t||_|durin||_d|_d|_dS)NF)r expanduserabsoluternr_delayed_batchr)rrnrrrrrs zExecutor.__init__r cCs*|j}|dr|dtd }|S)Nr)r]endswithr~lower)clsnrrrrs z Executor.nameFallow_implicit_submissionsc cst||_|jdur tdg|_z%zdVWnty+}z td|d}~ww|Wd|_dSd|_w)aCreates a context within which all submissions are packed into a job array. By default the array submissions happens when leaving the context Parameter --------- allow_implicit_submissions: bool submits the current batch whenever a job attribute is accessed instead of raising an exception Example ------- jobs = [] with executor.batch(): for k in range(12): jobs.append(executor.submit(add, k, 1)) Raises ------ AttributeError if trying to access a job instance attribute while the batch is not exited, and intermediate submissions are not allowed. Nz8Nesting "with executor.batch()" contexts is not allowed.zOCaught error within "with executor.batch()" context, submissions are dropped. )rr r2rTrrOrr)rrrXrrrbatchs    zExecutor.batchcCsl|jdusJ|js|jstjdtddSt|j\}}||}t||D] \}}||q'g|_dS)Nz>No submission happened during "with executor.batch()" context.)category)r rrrRuntimeWarningzip_internal_process_submissionsr)rjobs submissionsnew_jobsjnew_jrrrrs   zExecutor._submit_delayed_batchfn.argskwargscOsdtj|g|Ri|}|jdurt|}|j||fn||gd}t|tur0td|S)NrzEExecutors should never return a base Job class (implementation issue)) rrr rrrtyperjr2)rrrrdsrrrrsubmits  zExecutor.submitdelayed_submissionscCdSr r)rr rrrrz&Executor._internal_process_submissionsiterablecs:fddt|D}t|dkrtdgS||S)aA distributed equivalent of the map() built-in function Parameters ---------- fn: callable function to compute *iterable: Iterable lists of arguments that are passed as arguments to fn. Returns ------- List[Job] A list of Job instances. Example ------- a = [1, 2, 3] b = [10, 20, 30] executor.map_array(add, a, b) # jobs will compute 1 + 10, 2 + 20, 3 + 30 csg|] }tjg|RqSrrr)rsrrrrrvrwz&Executor.map_array..rReceived an empty job array)rr~rrr)rrr#rrr%r map_arrays    zExecutor.map_arrayfnscCs2dd|D}t|dkrtdgS||S)aSubmit a list of job. This is useful when submiting different Checkpointable functions. Be mindful that all those functions will be run with the same requirements (cpus, gpus, timeout, ...). So try to make group of similar function calls. Parameters ---------- fns: list of callable functions to compute. Those functions must not need any argument. Tyically those are "Checkpointable" instance whose arguments have been specified in the constructor, or partial functions. Returns ------- List[Job] A list of Job instances. Example ------- a_vals = [1, 2, 3] b_vals = [10, 20, 30] fns = [functools.partial(int.__add__, a, b) for (a, b) in zip (a_vals, b_vals)] executor.submit_array(fns) # jobs will compute 1 + 10, 2 + 20, 3 + 30 cSsg|]}t|qSrr$)rsrrrrrvsz)Executor.submit_array..rr&)r~rrr)rr(rrrr submit_arrays    zExecutor.submit_arraycKs&|jdur td|jdi|dS)zUpdate submision parameters.NzPChanging parameters within batch context "with executor.batch():" is not allowedr)r r2_internal_update_parametersrrrrrupdate_parameters$s zExecutor.update_parameterscCr!r rr rrr_equivalence_dict,zExecutor._equivalence_dictcCstS)z4Parameters that can be set through update_parameters)rr-rrr_valid_parameters0r.zExecutor._valid_parametersparamscsBttjtjttf|dur|Sfdd|DS)zConvert generic parameters to their specific equivalent. This has to be called **before** calling `update_parameters`. The default implementation only renames the key using `_equivalence_dict`. Ncsi|] \}}|||qSr)r5)rsrtveq_dictrr >rwz0Executor._convert_parameters..)rbrrfrer[r.items)rr1rr3r_convert_parameters5s zExecutor._convert_parameterscKs|j|dS)zUpdate submission parameters.N)rrHr+rrrr*@sz$Executor._internal_update_parameterscCdS)aThe 'score' of this executor on the current environment. -> -1 means unavailable -> 0 means available but won't be started unless asked (eg debug executor) -> 1 means available -> 2 means available and is a highly scalable executor (cluster) rrr-rrraffinityDs zExecutor.affinityr rr\).r]r^r_r`rjrrbTyperrrcr[rrfrer classmethodr contextlibcontextmanagerrirrrCallablerrabcabstractmethodrgrrrIterabler'rr)r,rr.Setr0r7r*rar9rrrrrs6 0 &,   2,* rc s:eZdZdZ d%dejeefdede ddffd d Z d ej e j dej eejfd d Zd&ddZedefddZdedeejfddZdededdfddZejdefddZejdededefddZejdedej efddZeejd ejeefdefd!d"Zededdfd#d$ZZS)'PicklingExecutoraBase job executor. Parameters ---------- folder: Path/str folder for storing job submission/output and logs. max_num_timeout: int maximum number of timeouts after which submitit will not reschedule the job. Note: only callable implementing a checkpoint method are rescheduled in case of timeout. max_pickle_size_gb: float maximum size of pickles in GB allowed for a submission. Note: during a batch submission, this is the estimated sum of all pickles. ?rnmax_num_timeoutmax_pickle_size_gbr Ncs(t|||_||_d|_d|_dS)Ng?r)superrrFrG _throttling_last_job_submitted)rrnrFrGrqrrr`s  zPicklingExecutor.__init__r c Cs |}|j|r |dndd}g}d}|D]v}tj}tj|j |d}|j j ddd| ||j |||rrd}t|} |jd} | | |jkrr|d| d | d d |jd } | d 7} | d7} t| |t|_||j} | j|d|| q|S)aSubmits a task to the cluster. Parameters ---------- fn: callable The function to compute *args: any positional argument for the function **kwargs: any named argument for the function Returns ------- Job A Job instance, providing access to the job information, including the output of the function once it is computed. rTz.pkl)parentsexist_okFi@zSubmitting an estimated z x z.2fz > zGB of objects zV(function and arguments) through pickle (this can be slow / overload the file system).zbIf this is the intended behavior, you should update executor.max_pickle_size_gb to a larger value r)r.rr5uuiduuid4hexrrget_first_id_independent_folderrnparentmkdir set_timeoutrFdumpr~rst_sizerGunlinkr2 _throttler/r0rJ_submit_command_submitit_command_strrmove_temporary_filer) rr r4rr check_sizedelayedtmp_uuid pickle_pathnumsizemsgrrrrris4     z.PicklingExecutor._internal_process_submissionscCs<t|j|jkrt|jt|j|jks dSdSr )r/r0rJrIrr%rrrrXs zPicklingExecutor._throttlecCr8)Nrkrr%rrrrZr"z&PicklingExecutor._submitit_command_strrWc Cstj}tj|jd|d}|d}|| ||Wdn1s+wY| |}tj |dd}| |}t t|}|j|j||d} | jj|dd d || j||| jj| S) aSubmits a command to the cluster It is recommended not to use this function since the Job instance assumes pickle files will be created at the end of the job, and hence it will not work correctly. You may use a CommandFunction as argument to the submit function instead. The only problem with this latter solution is that stdout is buffered, and you will therefore not be able to monitor the logs in real time. Parameters ---------- command: str a command string Returns ------- Job A Job instance, providing access to the crun job information. Since it has no output, some methods will not be efficient z.submission_file_z.shwNF)verboserpsubmission_fileT)keep_as_symlink)rNrOrPrrrQrnrwrite_make_submission_file_text_make_submission_commandCommandFunction#_get_job_id_from_submission_commandrr _num_tasksrrr[ _write_job_idr(_set_job_permissions) rrWr^submission_file_pathr command_listrr( tasks_idsrrrrrYs    z PicklingExecutor._submit_commandr(uidcCr8)zWrite the job id in a file named {job-independent folder}/parent_job_id_{uid}. This can create files read by plugins to get the job_id of the parent job Nr)rr(rrrrrrmszPicklingExecutor._write_job_idcCr)z1Returns the number of tasks associated to the jobr!r%rrrrlr"zPicklingExecutor._num_taskscCr)z~Creates the text of a file which will be created and run for the submission (for slurm, this is sbatch file). r!)rrWrrrrrrhsz+PicklingExecutor._make_submission_file_textrocCr)zCreate the submission command.r!)rrorrrrir"z)PicklingExecutor._make_submission_commandrcCr)z=Recover the job id from the output of the submission command.r!)rrrrrksz4PicklingExecutor._get_job_id_from_submission_commandcCr!r r)rnrrrrnr/z%PicklingExecutor._set_job_permissions)rDrEr\) r]r^r_r`rbrcrr[rarrrgrrrjrrrXrhrZrYrmr?r@rlrhri staticmethodrdrkrn __classcell__rrrqrrCPsB   -$ rC)r?rr<rRr0r/typingrbrNrpathlibrtyping_extensionsrrrTypeVarrr Genericrjrr_MSGrABCrrCrrrrs2  s-;F