o wOiY@sUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z mZddlmZddlmZmZmZmZmZmZddlmZddlmZmZmZmZmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(ddl)m*Z*e*Z+de,d e,fd d Z-d Z.e,e/d <Gdddej0Z1Gddde1Z2e,Z3e,Z4e,Z5eGdddZ6GdddZ7d#ddZ8eGdddZ9eGdddZ:Gddde$Z;GdddZNAc@s(eZdZdZejdedefddZdS) ImageFetcherz Downloads and sets up an image onto the localhost. This is only needed for ``LocalhostScheduler`` since typically real schedulers will do this on-behalf of the user. imagercCst)zi Pulls the given image and returns a path to the pulled image on the local host. )NotImplementedErrorselfr'r"r"r#fetch@szImageFetcher.fetchN)__name__ __module__ __qualname____doc__abcabstractmethodr r+r"r"r"r#r&9sr&c@s"eZdZdZdedefddZdS)LocalDirectoryImageFetchera Interprets the image name as the path to a directory on local host. Does not "fetch" (e.g. download) anything. Used in conjunction with ``LocalScheduler`` to run local binaries. The image name must be an absolute path and must exist. Example: 1. ``fetch(Image(name="/tmp/foobar"))`` returns ``/tmp/foobar`` 2. ``fetch(Image(name="foobar"))`` raises ``ValueError`` 2. ``fetch(Image(name="/tmp/dir/that/does/not_exist"))`` raises ``ValueError`` r'rcCs<tj|std|dtj|std|d|S)z Raises: ValueError - if the image name is not an absolute dir and if it does not exist or is not a directory zInvalid image name: z%, image name must be an absolute pathz&, does not exist or is not a directory)ospathisabs ValueErrorisdirr)r"r"r#r+Xs    z LocalDirectoryImageFetcher.fetchN)r,r-r.r/r r+r"r"r"r#r2Isr2c@sreZdZUdZeed<eed<ejed<e e ed<e e ed<e ed<dd d Z de fd d Zde fddZd S) _LocalReplicazD Contains information about a locally running role replica. role_name replica_idprocstdoutstderr error_filerNcCs<|j|j|jr|j|jr|jdSdS)z terminates the underlying process for this replica closes stdout and stderr file handles safe to call multiple times N)r; terminatewaitr<closer=r*r"r"r#r?s   z_LocalReplica.terminatecCs|jduSN)r;pollrBr"r"r#is_alivesz_LocalReplica.is_alivecCs|rdS|jjdkS)NFr)rEr; returncoderBr"r"r#faileds z_LocalReplica.failedrN)r,r-r.r/RoleName__annotations__int subprocessPopenr rr r?boolrErGr"r"r"r#r8rs     r8c@seZdZdZdedefddZdededd fd d Zd edd fd dZ dddZ de efddZ defddZ dddZddZd S)_LocalApplicationz Container object used by ``LocalhostScheduler`` to group the pids that form an application. Each replica of a role in the application is a process and has a pid. idlog_dircCs$||_||_i|_tj|_d|_dS)N)rPrQ role_replicasrPENDINGstate last_updated)r*rPrQr"r"r#__init__s  z_LocalApplication.__init__r9replicarNcCs|j|g}||dSrC)rS setdefaultappend)r*r9rXprocsr"r"r# add_replicasz_LocalApplication.add_replicarUcCst|_||_dSrC)timerVrU)r*rUr"r"r# set_states  z_LocalApplication.set_statecCs&|jD] }|D]}|q qdS)z terminates all procs associated with this app, and closes any resources (e.g. log file handles) safe to call multiple times N)rSvaluesr?)r*replicasrr"r"r#r?s  z_LocalApplication.terminatecCsXd}tj}|jD]}|D]}tj|jsqtj|j}||kr(|}|j}qq |SrC) sysmaxsizerSr_r3r4existsr>getmtime)r*r> min_timestampr`rXmtimer"r"r#_get_error_filesz!_LocalApplication._get_error_filecCsN|}|stSt|d}tt|WdS1s wYdS)Nra)rhropenjsondumpsload)r*r>fr"r"r#get_structured_error_msgs  $z*_LocalApplication.get_structured_error_msgc Cs|dttfdd}i}|jD])\}}g}|D]}|j|jj|jj||j ||j |j d}| |q|||<q|j |j|jj|j|d}tj|dd} ttj|jdd  } | | Wd n1smwYtd |j d | d S) aA terminates all procs associated with this app, and closes any resources (e.g. log file handles) and if log_dir has been specified, writes a SUCCESS file indicating that the log files have been flushed and closed and ready to read. NOT safe to call multiple times! std_iocSs|r|jSdS)Nz )name)ror"r"r#_fmt_io_filenamesz1_LocalApplication.close.._fmt_io_filename)r:pidexitcoder<r=r>)app_idrQ final_staterVroles)indentSUCCESSwNzSuccessfully closed app_id: z. )r?r rrSitemsr:r;rrrFr<r=r>rZrPrQrUrprVrjrkrir3r4joinwriteloginfo) r*rq roles_infor9r` replicas_inforX replica_infoapp_infoinfo_strfpr"r"r#rAs4     z_LocalApplication.closecCsXi}|jD]\}}||g}|D] }||jjqqd|jd|jd|dS)Nz{app_id:z, state:z , pid_map:})rSr{rYrZr;rrrPrU)r* role_to_pidr9r`pidsrar"r"r#__repr__s z_LocalApplication.__repr__rH)r,r-r.r/r rWr8r\rr^r?r rhrnrArr"r"r"r#rOs     0rOcCs td}d}||tjdS)z Sets PR_SET_PDEATHSIG to ensure a child process is terminated appropriately. See http://stackoverflow.com/questions/1884941/ for more information. For libc.so.6 read http://www.linux-m68k.org/faq/glibcinfo.html z libc.so.6N)ctypesCDLLprctlsignalSIGTERM)libcPR_SET_PDEATHSIGr"r"r#_pr_set_pdeathsig s rc@sFeZdZUdZeeed<eeefed<eeed<eeed<dS) ReplicaParamzS Holds ``LocalScheduler._popen()``parameters for each replica of the role. argsenvr<r=N) r,r-r.r/rr rJrr r"r"r"r#rs   rc@sJeZdZUdZeed<eed<eee e fed<eee efed<dS) PopenRequestzf Holds parameters to create a subprocess for each replica of each role of an application. rtrQ role_params role_log_dirsN) r,r-r.r/AppIdrJr rrIrrr"r"r"r#r&s rcs~eZdZdZd5dedeffdd Zdefdd Zd e d e dd fd dZ de ee ffddZdede fddZdefddZdeefddZdedededefddZdededeeeffdd Zd!edefd"d#Zd e dedeefd$d%Zd e dedefd&d'Zdedee fd(d)Z! * d6deded+ed,eed-ee"d.ee"de#fd/d0Z$dedd fd1d2Z%d3d4Z&Z'S)7LocalSchedulera} Schedules on localhost. Containers are modeled as processes and certain properties of the container that are either not relevant or that cannot be enforced for localhost runs are ignored. Properties that are ignored: 1. Resource requirements 2. Container limit enforcements 3. Retry policies 4. Retry counts (no retries supported) 5. Deployment preferences ..note:: Use this scheduler sparingly since an application that runs successfully on a session backed by this scheduler may not work on an actual production cluster using a different scheduler. d session_name cache_sizecs,t|i|_|dkrtd||_dS)Nrz$cache size must be greater than zero)superrW_appsr6 _cache_size)r*rr __class__r"r#rWJs  zLocalScheduler.__init__rcCs.t}|jdtddd|jdtddd|S)N image_fetcherzimage fetcher typedir)type_helpdefaultrQz0dir to write stdout/stderr log files of replicas)rrr)raddr )r*optsr"r"r#run_optsTszLocalScheduler.run_optsapp schedulerNcCsdSrCr")r*rrr"r"r# _validate_szLocalScheduler._validatecCs dtiS)Nr)r2rBr"r"r# _img_fetcherscs zLocalScheduler._img_fetcherscfgcCsF|d}|}||d}|s!td|d||||S)Nrz Unsupported image fetcher type: z. Must be one of: )getrrkeysr)r*rimg_fetcher_typefetchers img_fetcherr"r"r#_get_img_fetcherfs  zLocalScheduler._get_img_fetchercCsttj}d}|jD]\}}t|jr|j|kr|}q |r,|j|=td|ddStdt |jddS)a7 Evicts one least recently used element from the apps cache. LRU is defined as the oldest app in a terminal state (e.g. oldest finished app). Returns: ``True`` if an entry was evicted, ``False`` if no entries could be evicted (e.g. all apps are running) Nzevicting app: z, from local scheduler cacheTzno apps evicted, all z apps are runningF) rbrcrr{rrUrVr~debuglen)r*lru_time lru_app_idrtrr"r"r# _evict_lruss   zLocalScheduler._evict_lrufilecCsF|sdStj|rtd|dtjtj|ddt|ddS)z Given a file name, opens the file for write and returns the IO. If no file name is given, then returns ``None`` Raises a ``FileExistsError`` if the file is already present. Nz log file: zT already exists, specify a different log_dir, app_name, or remove the file and retryT)exist_okrz)mode)r3r4isfileFileExistsErrormakedirsdirnameri)r*rr"r"r# _get_file_ios   zLocalScheduler._get_file_ior9r:replica_paramsc Cs||j}||j}tj}||j|d}tj t |ddd}t d|d|d|t j|j|||td} t||| |||d S) z Same as ``subprocess.Popen(**popen_kwargs)`` but is able to take ``stdout`` and ``stderr`` as file name ``str`` rather than a file-like obj. TORCHELASTIC_ERROR_FILErwPrxwidthzRunning z (replica z): )rrr<r= preexec_fn)r<r=r>)rr<r=r3environcopyupdaterpprintpformatrr~rrLrMrrr8) r*r9r:rstdout_stderr_rr> args_pfmtr;r"r"r#_popens,    zLocalScheduler._popenrtcCs<|d}d}|stjdd}d}tjt||j||fS)a- Returns the log dir and a bool (should_redirect_std). We redirect stdout/err to a log file ONLY if the log_dir is user-provided in the cfg 1. if cfg.get("log_dir") -> (user-specified log dir, True) 2. if not cfg.get("log_dir") -> (autogen tmp log dir, False) rQTtsm_)prefixF)rtempfilemkdtempr3r4r|r r)r*rtr base_log_dir redirect_stdr"r"r#_get_app_log_dirs  zLocalScheduler._get_app_log_dir dryrun_infoc Cst|j|jkr|std|jd|j}|j}|j}||jvs'Jdt |t ||}|j D]/}|j |}|j |}tt|D]} || } || } t | ||| | } ||| qHq6||j|<|S)NzApp cache size (z#) exceeded. Increase the cache sizez7no app_id collisons expected since uuid4 suffix is used)rrrr IndexErrorrequestrtrQr3rrOrrrrangerr\) r*rrrt app_log_dir local_appr9rrr:rreplica_log_dirrXr"r"r#schedules2        zLocalScheduler.schedulecCs|||}t|ddS)NcSstj|dddS)Nrwrr)rr)pr"r"r#sz/LocalScheduler._submit_dryrun..)_to_popen_requestr )r*rrrr"r"r#_submit_dryruns zLocalScheduler._submit_dryrunc Cs$t|j}||}|||\}}i}i}|jD]q} || jg} || jg} | j} || j} t j | | j }t | jD]J}|gt| j| |t|}t j || jt|}dt j |di| j}d}d}|rzt j |d}t j |d}| t||||| |q?qt||||S)zK Converts the application and cfg into a ``PopenRequest``. rz error.jsonNz stdout.log stderr.log)r$rprrrvrY containerr+r'r3r4r| entrypointr num_replicasr substituterr rrZrr)r*rrrtrrrrrrolerreplica_log_dirsrimg_rootcmdr:rrenv_varsr<r=r"r"r#rs>    z LocalScheduler._to_popen_requestc Cs||jvrdS|j|}|}t|jr|j}n1d}d}|jD]}|D]}||O}||O}q&q"|reZdZdedededefddZddZd d Zd d Zd S)rrtrrrcCs.||_t||_||_d|_||_d|_dS)NF)_app_idrecompile_regex _log_file_log_fp _scheduler _app_finished)r*rtrrrr"r"r#rWws   zLogIterator.__init__cCs0|j|j}|rt|jrd|_dSd|_dS)NTF)r rrrrUr)r*descr"r"r#_check_finisheds  zLogIterator._check_finishedcCsV |tj|jrt|jd|_ |S|jr%td|j d|jt dq)NTrarz finished without writing: r) rr3r4rr rir rrrr]sleeprBr"r"r#__iter__s  zLogIterator.__iter__cCsX |j}|s|jr|jttd|n|d}t |j |r+|Sq)NTr ) r readlinerrA StopIterationr]rrrstriprmatchr )r*liner"r"r#__next__s     zLogIterator.__next__N) r,r-r.r rrWrrrr"r"r"r#rvs   rrcKst||dddS)Nrr)rr)rr)rkwargsr"r"r#create_schedulers rrH)>r0rrjr3rrrrLrbrr]r dataclassesrrrtypingrrrrr r uuidr torchelastic.tsm.driver.apir r rrrrrrrrrrrrrtorchelastic.utils.loggingrr~r r$r%rJABCr&r2rAppNamerIr8rOrrrrrrr"r"r"r#sN    D  $' s  A9