o xioh@sdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z mZddlmZddlmZddlmZddlmZmZddlZddlZdd lmZdd lmZdd lmZdd l m!Z!dd l"m#Z#m$Z$ddl%m&Z&m'Z'ddl(m)Z)erddl*m+m,Z,ddl-m.Z.ddl*m/Z/m0Z0e1e2Z3ej4ddddZ5dZ6GdddeZ7GdddeZ8eGdddZ9eGdddZ:Gdd d e Z;dS)!zAbstract Scheduler class.) annotationsN)ABCabstractmethod)Iterator) dataclass)Enum) TYPE_CHECKINGAny) CommError) launch_add) LaunchError)SchedulerError)create_sweep_command_argsmake_launch_sweep_entrypoint)event_loop_thread_exec%strip_resource_args_and_template_vars) generate_idApi) QueuedRunRunzsched:cyan)fg g@c@s,eZdZdZdZdZdZdZdZdZ dZ d S) SchedulerStaterN) __name__ __module__ __qualname__PENDINGSTARTINGRUNNING FLUSH_RUNS COMPLETEDFAILEDSTOPPED CANCELLEDr-r-U/home/ubuntu/.local/lib/python3.10/site-packages/wandb/sdk/launch/sweeps/scheduler.pyr1src@sTeZdZdZdZdZdZdZdZdZ dZ d Z dddZ ddddZ ed ddZdS)!RunState)runningalive)pendingr1) preemptingr1)crasheddead)failedr5)killedr5)finishedr5) preemptedr5)unknownr1clsr argslistkwdsreturncOst|}|d|_|S)Nr)object__new___value_)r;r<r>objr-r-r.rAIs  zRunState.__new__r:_strlifeNonecCs ||_dSN_life)selfrDrFr-r-r.__init__Ns zRunState.__init__boolcCs |jdkS)Nr1rIrKr-r-r.is_aliveQ zRunState.is_aliveN)r;r r<r=r>r r?r/)r:)rDrErFrEr?rGr?rM)r"r#r$r'r% PREEMPTINGCRASHEDr*KILLEDFINISHED PREEMPTEDUNKNOWNrArLpropertyrOr-r-r-r.r/<s  r/c@seZdZUded<ded<dS)_Workerdict[str, Any] agent_configrEagent_idN)r"r#r$__annotations__r-r-r-r.rYVs  rYc@sPeZdZUded<ded<ejZded<dZded <dZd ed <dZ d ed <dS)SweepRunrEidint worker_idr/stateNzpublic.QueuedRun | None queued_runzdict[str, Any] | Noner<zlist[str] | Nonelogs) r"r#r$r]r/r'rbrcr<rdr-r-r-r.r^\s   r^c@seZdZdZdZdZgdZddddddddwddZedxddZ edyddZ edyd d!Z edyd"d#Z edyd$d%Z edzd'd(Zejd{d*d(Zed|d,d-Zed|d.d/Zed}d0d1Zed~d3d4Zed~d5d6Zdd8d9Zdyd:d;Zdd=d>Zdyd?d@ZdydAdBZdydCdDZddGdHZd|dIdJZdydKdLZddNdOZddRdSZ dydTdUZ!ddXdYZ"dydZd[Z#dyd\d]Z$dd_d`Z%ddbdcZ&e'j(fddfdgZ)ddhdiZ*ddkdlZ+ddndoZ,ddsdtZ-ddudvZ.dS) SchedulerzPA controller/agent that populates a Launch RunQueue from a hyperparameter sweep.zplaceholder-uri-schedulerzsweep-controller)wandb schedulerWANDB_SWEEP_IDN) polling_sleepsweep_identityproject project_queue num_workersapirr< Any | Noneri float | Nonerj str | Nonerkrlrmrnint | str | Nonekwargsc Osddl} ddlm} ||_| |_|p tjdp |dp |j |_ |p.tjdp.|d|_ |p3d|_ t j|_z?|jj|d|j |j d } | d t jjkrSt j|_| | d |_|| d |_|jdkrwttd |jd|j Wnty} z ttd|d| d} ~ ww| |_i|_t|_ |dur|nt!|_"||_#i|_$|%|_&|p|j&j'did}t(|)rt*|nd|_+|j&j'di|_,dS)Nrr WANDB_ENTITYrk WANDB_PROJECTrlzempty-sweep-idz{})rkrlrbconfigrunszFound z previous valid runs for sweep zException when finding sweep (z) rgrnsettings)-yamlwandb.apis.publicr_api _public_apiosenvirongetrzdefault_entity_entity_project _sweep_idrr%_statesweepr,name safe_load _sweep_config_get_num_runs_launched_num_runs_launchedrftermlog LOG_PREFIX Exceptionr _kwargs_runs threadingLock_threading_lockDEFAULT_POLLING_SLEEP_polling_sleep_project_queue_workers_init_wandb_run _wandb_runrwrEisdigitr` _num_workers_settings_config)rKrorirjrkrlrmrnr<rtr{ PublicApiresper-r-r.rLmsb        zScheduler.__init__rar`r?SweepRun | NonecCdS)zCalled when worker available.Nr-)rKrar-r-r._get_next_sweep_runzScheduler._get_next_sweep_runrGcCr)zCalled every polling loop.Nr-rNr-r-r._pollrzScheduler._pollcCdSrHr-rNr-r-r._exitzScheduler._exitcCrrHr-rNr-r-r. _load_staterzScheduler._load_statecCrrHr-rNr-r-r. _save_staterzScheduler._save_statercCsttd|jj|jS)NzScheduler state is )_loggerdebugrrrrNr-r-r.rbszScheduler.statevaluecCs*ttd|jjd|j||_dS)NzScheduler was z is )rrrrbrr)rKrr-r-r.rbs  rMcCs|jtjtjtjtjfvSrH)rbrr)r*r+r,rNr-r-r.rOs zScheduler.is_alivecCs"|jd}|s dS|j|k}|S)z/False if under user-specified cap on # of runs.run_capF)rrr)rKr at_runcapr-r-r.rs  zScheduler.at_runcapcCs t|jSrH)lenrrNr-r-r.num_active_runsrPzScheduler.num_active_runsdict[int, _Worker]cCs,i}|D] \}}|j|j||j<q|S)zReturns dict of id:worker already assigned to a launch run. runs should always have a worker_id, but are created before workers are assigned to the run ) _yield_runsrra)rK busy_workersrDrr-r-r.rszScheduler.busy_workerscs*tjdkr iSfddjDS)z6Returns dict of id:worker ready to launch another run.rcs i|] \}}|jvr||qSr-)r).0_idwrNr-r. sz/Scheduler.available_workers..)rritemsrNr-rNr.available_workerss  zScheduler.available_workers wandb.RuncCs,tjdd}tjd|jd|j|d}|S)z8Controls resume or init logic for a scheduler wandb run.T)disable_job_creationz Scheduler.allow)rresumerwrz)rfSettingsinitrr)rKrzrunr-r-r.rs  zScheduler._init_wandb_runcCs tj|_dS)zStop the sweep.N)rr+rrNr-r-r. stop_sweeps zScheduler.stop_sweeperrcCstj|_|r t|dS)z%Fail the sweep w/ optional exception.N)rr*rr )rKrr-r-r. fail_sweepszScheduler.fail_sweepcCsttd|js ttd|jjd|dSt j |_ | s6ttd|dS| t||dS)zAStart a scheduler, confirms prerequisites, begins execution loop.zScheduler starting.zSweep already in end state (z ). Exiting...Nz1No 'job' or 'image_uri' loaded from sweep config.)rfrrrO termerrorrbrlowerexitrr&r_try_load_executablerasyncior_register_agentsrNr-r-r.starts" zScheduler.startc Csttdtj|_z ||jsnttd|| |jtj krA|j dkr:ttdnrt |jq |jD]`}|jr[ttd|jdtj |_nJz ||}|sfWn?Wn0tyw}zt|d}~wty}zttd |tj|_WYd}~nd}~ww||r|jd 7_qDt |jqWn8tyttd tj|_|YdSty}zttd |tj|_|d}~ww|jtj kr|jrtj|_|dS) zMain run function.zScheduler runningTzPolling for new runs to launchrzDone polling on runs, exitingzSweep at run_cap ()NzFailed to get next sweep run: rz-Scheduler received KeyboardInterrupt. Exitingz Scheduler failed with exception )rfrrrr'rb_update_scheduler_run_staterO_update_run_statesrr(rtimesleeprrrrrr rrr*_add_to_launch_queueKeyboardInterrupttermwarnr+rr))rKrarrr-r-r.r/sx         ) z Scheduler.runc Cs|z|WntyttdtYnwd}|jt j kr0| dd}n2|jt j kr>| dd}n$|jt j t jfvrS| dd}|nt j|_| d d }|ttd ||jdS) NzFailed to save state: PAUSEDpausedrU completedCANCELED cancelledrSr4z Scheduler )rrrrfrr traceback format_excrbrr(_set_sweep_stater)r,r+ _stop_runsr*rrfinish)rKstatusr-r-r.rms2         zScheduler.exitrxlist[dict[str, Any]]c CsZd}|D]&}|dddvr&|ds&td|dd|dd |q|d 7}q|S) z.Returns the number of valid runs in the sweep.rrbr)r7r4summaryMetricszexcluding run: rz with state: z from run cap r)rrr)rKrxcountrr-r-r.rs z Scheduler._get_num_runs_launchedcCsx|jdr4z|j|jd}ttd|jdWdSty3t tt YdSwt |jdS)zoCheck existence of valid executable for a run. logs and returns False when job is unreachable jobzSuccessfully loaded job (z) in schedulerFT image_uri) rrr~rrfrrrrrrrrM)rK _job_artifactr-r-r.rs  zScheduler._try_load_executablec sg}t|jj}t|jD]H}ttd|dz|t d||j |j |j d}| |WqtyV}ztd||d|WYd}~qd}~wwtj|IdH}t|D]\}}t||dd|j|<qcdS)Nz Starting AgentHeartbeat worker (r-)rj project_namerkzfailed to register agent: r_)r[r\)rr}register_agentrangerrrrsocket gethostnamerrrappendrrrgather enumeraterYr) rKtasksrraworkerrfinished_tasksidxr[r-r-r.rs2 zScheduler._register_agentsIterator[tuple[str, SweepRun]]ccs>|j|jEdHWddS1swYdS)z)Thread-safe way to iterate over the runs.N)rrrrNr-r-r.rs"zScheduler._yield_runsruns_to_remove list[str]cCsT|j|D]}ttd|d|j|=qWddS1s#wYdS)zHelper for removing runs from memory. Can be overloaded to prevent deletion of runs, which is useful for debugging or when polling on completed runs. zCleaning up finished run (rN)rrfrrr)rKrrun_idr-r-r. _cleanup_runss  "zScheduler._cleanup_runscCsdg}|D] \}}||g7}q|D]}ttd|d||s/ttd|dqdS)NzStopping run (rzFailed to stop run ()rrfrr _stop_runr)rK to_deleterrDr-r-r.rs  zScheduler._stop_runsrrEc Cs||jvrtd|d|jdS|j|}|j|=|js+td|jddS|jjs1dStd|d|j d|j   d }z|j j|d }|r^ttd |d WdSWdSty~}ztd |d|WYd}~dSd}~ww)z.Stops a run and removes it from the scheduler.zrun: z not in _runs: Fz2tried to _stop_run but run not queued yet (run_id:rTzRun:v1::utf-8)rz Stopped run .zerror stopping run (): N)rrrrcr_rbrObase64standard_b64encoderrencodedecoder}stop_runrfrrr)rKrrencoded_run_idsuccessrr-r-r.rs:    zScheduler._stop_runc Cs||jj}|tjkrtj|_n|tjtj fvrtj|_n |tj kr'tj |_z |j |j|j|j}WntyO}ztd|WYd}~dSd}~ww|dkrZtj |_dS|dvrdtj|_dS|dkrntj|_dSdS)zGUpdate the scheduler state from state of scheduler run and sweep state.zsweep state error: NrU)r,r+r)_get_run_staterr_r/rTrr+rbr*rSrUr)r}get_sweep_staterrrrrrr,r()rKrb sweep_staterr-r-r.rs.        z%Scheduler._update_scheduler_run_statec Csg}|D]S\}}|||j|_z |jr|jjnd}Wnttfy;}ztd|d}WYd}~nd}~ww|jjrD|dkrYtd|d|jd|d| |q| |dS)ztIterate through runs. Get state from backend and deletes runs if not in running state. Threadsafe. Nz Failed to get queued_run.state: r6(z ) states: (z, r) rrrbrcr r rrrOrr)rKrrr rqi_staterr-r-r.rs  zScheduler._update_run_states list[Any]c sz6|j|j}|s gWS|j|jd|jd|}|jdd|jdgd}fdd|D}|WStyQ}zt d|WYd }~gSd }~ww) zUse the public api to get metrics from a run. Uses the metric name found in the sweep config, any misspellings will result in an empty list. /metricr_step)keyscsg|]}|qSr-r-rx metric_namer-r. Asz3Scheduler._get_metrics_from_run..z[_get_metrics_from_run] N) rrcr~rrkrlr scan_historyrrr)rKrrcapi_runhistorymetricsrr-rr._get_metrics_from_run1s  zScheduler._get_metrics_from_runrZc Cs\z|j|j|j|}|r|WSWiSty-}ztd|WYd}~iSd}~ww)z+Use the public api to get info about a run.z[_get_run_info] N)r} get_run_inforrrrr)rKrinforr-r-r. _get_run_infoHs zScheduler._get_run_infoprev_run_stater/c Csd}z|j|j|j|}t|}W|StyQ}z2td|d||tjkr.z*Sweep command contains unresolved macros: z', see launch docs for supported macros.)rr<rrrr rindexcopydeepcopyrrwupdaterfr)rKrr< entry_point macro_argspidx launch_config unresolvedr-r-r._make_entry_and_launch_configs8        z'Scheduler._make_entry_and_launch_configc Csz|jdp |jd}|jd}|jdp|}|dur-|dur-ttd|jd|dur<|durr?r@rArjr<z Added run (z ) to queue (T)rrrr rr_r;rfrr3r4rrwr`rrr rrrrrcr/r'rbrr) rKr_job_sweep_config_uri _image_urir6r9_job_launch_config _priorityrrcr-r-r.rsT       zScheduler._add_to_launch_queue)rorr<rprirqrjrrrkrrrlrrrmrrrnrsrtrp)rar`r?r)r?rG)r?r)rrr?rGrQ)r?r`)r?r)r?r)rrrr?rG)rxrr?r`)r?r)rrr?rG)rrEr?rM)rrEr?r )rrEr?rZ)rrErr/r?r/)r?rZ)rbrEr?rG)rrEr?rE)rr^r?r*)rr^r?rM)/r"r#r$__doc__PLACEHOLDER_URISWEEP_JOB_TYPE ENTRYPOINTrLrrrrrrrXrbsetterrOrrrrrrrrrrrrrrrrrrrrrr/rWrr$rr)r;rr-r-r-r.refsxM                 >      "       #re)sN