U /e<@sddlmZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl m Z ddlmZddlmZddlm Z eeZGd d d Zd d Zdd dZdddZGdddZdS)) annotationsN)Queue)Thread)sleep)merge)gen)timec@s,eZdZdZdZdZdZdZdZdZ dZ d S) bcolorszzzzzzzzN) __name__ __module__ __qualname__HEADERZOKBLUEZOKGREENWARNINGFAILENDCBOLDZ UNDERLINErr>/tmp/pip-unpacked-wheel-g426oqom/distributed/deploy/old_ssh.pyr sr c s\ddl}ddlmddlm}m}|}||d}zDt d t j |j ddddd d d d Wq^Wq>||fk rZ}ztd tjd jddddtjttjdt|tjt|d7}|dkr"td tjdtjtdtdtjd|dtjtdW5d}~XYq>Xq>tdjddd|jdddd d\}jdfddfd d!fd"d#}|} d$rtd%| |rqqt} t| d&krH d'|r<qHtd%q!|!dS)(Nr) PipeTimeout)PasswordRequiredException SSHExceptionparamikoaddress ssh_usernamessh_portssh_private_keyT)hostnameusernameportZ key_filenamecompresstimeoutZbanner_timeoutz[ dask ssh ] : zDSSH connection error when connecting to {addr}:{port} to run '{cmd}'cmd)addrrr"z, SSH reported this exception: z/SSH connection failed after 3 retries. Exiting.z zRetrying... (attempt z/3)z[ {label} ] : {cmd}label)r&r"z$SHELL -i -c '')Zget_pty皙?c srzR}t|dkrP|}td|ddjd|d}q Wntjfk rlYnXdS)z< Read stdout stream, time out if necessary. rzstdout from ssh channel: %s output_queuez[ {label} ] : {output}r&)r&outputN) readlinelenrstriploggerdebugputformatsocketr!line)rcmd_dictstdoutrrread_from_stdout}s  z#async_ssh..read_from_stdoutc sz`}t|dkr^|}td|ddjddtj|tj }q Wnt j fk rzYnXdS)z< Read stderr stream, time out if necessary. rzstderr from ssh channel: %sr)[ {label} ] : r&r&N) r+r,r-r.r/r0r1r rrr2r!r3)rr5stderrrrread_from_stderrs"  z#async_ssh..read_from_stderrcsVrR}ddjddtjdt|tjdSdS)zp Communicate a little bit, without blocking too long. Return True if the command ended. r)r8r&r9z'remote process exited with exit status TN)Zexit_status_readyZrecv_exit_statusr0r1r rstrr) exit_status)channelr5r;r7rr communicates zasync_ssh..communicate input_queueg?g@)"rZparamiko.buffered_piperZparamiko.ssh_exceptionrrZ SSHClientZset_missing_host_key_policyZ AutoAddPolicylogging getLoggersetLevelWARNconnectprintr rr1rr< traceback print_excos_exitrZ exec_commandr> settimeoutZ get_transportemptyZ send_ignorersendclose) r5rrrsshretriesestdinr? transportstartr)rr>r5r;r7r:r6r async_ssh"s           rVc Csdj|p tj|d}|dk r@d|d|}|dj|||d7}tjd|d|tj}t} t} ||||| | |||d } tt| gd } d | _ | t | d | iS) Nz8{python} -m distributed.cli.dask_scheduler --port {port})pythonr mkdir -p  && z,&> {logdir}/dask_scheduler_{addr}:{port}.log)r#rlogdirz scheduler :) r"r&rrr@r)rrrtargetargsTthread) r1sys executabler rrrrrVdaemonrUr) rZr#rrrr remote_pythonr"r&r@r)r5r_rrrstart_schedulers8 rddistributed.cli.dask_workerc Csd|dkrdnd}| s |d7}| r,|d7}| r8|d7}| rD|d7}|j| pPtj||||||| | | d }|dk r|d j|d 7}|dk rd |d |}|dj||d7}d|}t}t}||||||||d}tt|gd}d|_|t|d|iS)NzX{python} -m {remote_dask_worker} {scheduler_addr}:{scheduler_port} --nthreads {nthreads}r$z --nworkers {n_workers}z --host {worker_addr}z --memory-limit {memory_limit}z --worker-port {worker_port}z --nanny-port {nanny_port}) rWremote_dask_workerscheduler_addrscheduler_port worker_addrnthreads n_workers memory_limit worker_port nanny_portz$ --local-directory {local_directory})local_directoryrXrYz%&> {logdir}/dask_scheduler_{addr}.log)r#rZzworker )r"r&rr@r)rrrr\Tr_) r1r`rarrrVrbrUr)rZrhrirjrkrlrrrnohostrmrnrorcrgrpr"r&r@r)r5r_rrr start_workersd   rrc @sreZdZdddZejdd Zed d Zej d d Zed dZ ddZ ddZ ddZ ddZddZdS) SSHClusterrNFrecKsJ||_||_||_|dd}|r:tdd||dk rT|dk rTtdn&|dk rnt dt |}n |dkrzd}||_ ||_ ||_ ||_| |_| |_| |_| |_||_||_||_ddl}| dk rtj| d|jd } ttjd j| d tj| |_ g|_!t"| |||||| |_#g|_$|D]}|%|q4dS) Nnprocsz.__init__() got an unexpected keyword argument z, z=Both nprocs and n_workers were specified. Use n_workers only.zZThe nprocs argument will be removed in a future release. It has been renamed to n_workers.r$rz dask-ssh_z%Y-%m-%d_%H:%M:%SzaOutput will be redirected to logfiles stored locally on individual worker nodes under "{logdir}".)rZ)&rhrirkpop TypeErrorjoinkeys ValueErrorwarningswarn FutureWarningrlrrrrqrcrmrnrorgrpdatetimerJpathnowstrftimerGr rr1rrZthreadsrd schedulerworkers add_worker)selfrhriZ worker_addrsrkrlrrrrqrZrcrmrnrorgrpkwargsrur~r#rrr__init__Sst   zSSHCluster.__init__cCsdSNrrrrr_startszSSHCluster._startcCstdt|jSNz[The nprocs attribute will be removed in a future release. It has been renamed to n_workers.r{r|r}rlrrrrrus zSSHCluster.nprocscCstdt||_dSrr)rvaluerrrrus cCsd|j|jfS)Nz%s:%d)rhrirrrrscheduler_addressszSSHCluster.scheduler_addresscCs^|jg|j}z6|D]"}|dst|dqqtdqWntk rXYnXdS)Nr)r()rrrMrGgetrKeyboardInterruptrZ all_processesprocessrrrmonitor_remote_processess z#SSHCluster.monitor_remote_processescCsP|jt|j|j|j||j|j|j|j |j |j |j |j |j|j|j|jdSr)rappendrrrZrhrirkrlrrrrqrmrnrorcrgrp)rrrrrrs(zSSHCluster.add_workercCs6|jg|j}|D]}|dd|dqdS)Nr@shutdownr_)rrr0rxrrrrrszSSHCluster.shutdowncCs|Srrrrrr __enter__szSSHCluster.__enter__cCs |dSr)r)rexc_type exc_valuerHrrr__exit__szSSHCluster.__exit__) rNNrtNFNNNNNreN)r r r rr coroutinerpropertyrusetterrrrrrrrrrrrsRs6 _    rs)N)NreN) __future__rrBrJr2r`rHr{queuer threadingrrrZtlzrZtornadorZdistributed.metricsrCr r.r rVrdrrrsrrrrs.         1 8 W