U /e;@sddlmZddlZddlZddlZddlZddlZddlmZddl m Z ddl Z ddl Z ddl mZmZeeZGdddeZGdd d eZGd d d eZd d ddddddddddddddhZd(ddd d d!d!d"d#d$d% d&d'ZdS))) annotationsN)dumps)Any)ProcessInterface SpecClustercs8eZdZdZfddZfddZfddZZS)ProcesszaA superclass for SSH Workers and Nannies See Also -------- Worker Scheduler c sd|_d|_tjf|dSN) connectionprocsuper__init__)selfkwargs __class__:/tmp/pip-unpacked-wheel-g426oqom/distributed/deploy/ssh.pyr szProcess.__init__cs.|js tt||jjtIdHdSr)r AssertionErrorweakreffinalizer killr startr rrrr!s  z Process.startcs4|jr|j|jr |jtIdHdSr)r rr closer rrrrr(s   z Process.close)__name__ __module__ __qualname____doc__r rr __classcell__rrrrrs  rcsTeZdZdZddddddfdd Zed d Zejd d Zfd dZZ S)Workera*A Remote Dask Worker controlled by SSH Parameters ---------- scheduler: str The address of the scheduler address: str The hostname where we should run this worker worker_class: str The python class to use to create the worker. connect_options: dict kwargs to be passed to asyncssh connections remote_python: str Path to Python on remote node to run this worker. kwargs: dict These will be passed through the dask worker CLI to the dask.distributed.Worker class deprecateddistributed.NannyNstrdict) scheduleraddressconnect_optionsrc st|dkrtd||_||_||_||_t||_| |_ ||_ | ddk rp| ddk rptdn<| ddk rt dt|jdd|_n|jdd|_dS)Nr sworker_module has been deprecated in favor of worker_class. Please specify a Python class rather than a CLI module.nprocs n_workersz=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 r ValueErrorr%r$ worker_classr&copyrname remote_pythongetwarningswarn FutureWarningpopr)) r r$r%r&r worker_moduler,r/Zloopr.rrrr Ds.  zWorker.__init__cCstdt|jSNz[The nprocs attribute will be removed in a future release. It has been renamed to n_workers.r1r2r3r)rrrrr(ms z Worker.nprocscCstdt||_dSr6r7)r valuerrrr(vs c sz ddl}Wntk r(tdYnX|jjfjIdH_jdIdH}|jdkrxdt j t j j }n<jdIdH}|jdkrdt j t j j }nt djstj_d|jd d jd d tfd dtjDg}j|IdH_d}|jkrfjjIdH}|sBt dt|d|kr|d7}qtd|tIdHdS)NrhDask's SSHCluster requires the `asyncssh` package to be installed. Please install it using pip or conda.uname%env DASK_INTERNAL_INHERIT_CONFIG="{}" cmd /c ver&set DASK_INTERNAL_INHERIT_CONFIG={} &&z;Worker failed to set DASK_INTERNAL_INHERIT_CONFIG variable  -mdistributed.cli.dask_spec--spec'%s'csi|]}|jjdqS)clsopts)r,r).0irrr s z Worker.start..Worker failed to startz worker atr*%s) asyncssh ImportErrorconnectr%r&r run exit_statusformatdaskconfig serialize global_config Exceptionr/sys executablejoinr$rranger)create_processr stderrreadlinestriploggerinfodebugr r)r rKresultset_envcmdZstarted_workerslinerrrrsb          z Worker.start)r r!NNN) rrrrr propertyr(setterrrrrrrr0s)  rcs:eZdZdZd dddddfdd Zfd d ZZS) ScheduleraA Remote Dask Scheduler controlled by SSH Parameters ---------- address: str The hostname where we should run this worker connect_options: dict kwargs to be passed to asyncssh connections remote_python: str Path to Python on remote node to run this scheduler. kwargs: dict These will be passed through the dask scheduler CLI to the dask.distributed.Scheduler class Nr"r#z str | Noner%r&rr/cs,t||_||_||_|p$tj|_dSr)r r r%rr&rVrWr/)r r%r&rr/rrrr s  zScheduler.__init__c snz ddl}Wntk r(tdYnXtd|j|jf|jIdH|_|jdIdH}|j dkrd t j t j j}n<|jdIdH}|j dkrd t j t j j}ntdd ||jd d d d td|jdg}|j|IdH|_|jjIdH}|stdt|d|kr|dd|_qNqtd|tIdHdS)Nrr9zCreated Scheduler Connectionr:r;r<r=z>Scheduler failed to set DASK_INTERNAL_INHERIT_CONFIG variable r>r?r@rArBzdistributed.SchedulerrCrIz Scheduler atz Scheduler at:r*rJ)rKrLr^r`rMr%r&r rNrOrPrQrRrSrTrUrXr/rrrZr r[r\r]r_splitr r)r rKrarbrcrdrrrrsP        zScheduler.start)N)rrrrr rrrrrrrgsrgZscheduler_addrZscheduler_port worker_addrsZnthreadsr(r)Z ssh_usernameZssh_portZssh_private_keyZnohostZlogdirr/Z memory_limitZ worker_portZ nanny_portZremote_dask_workerr r!zlist[str] | Nonezdict | list[dict] | Nonez dict | Noner"zstr | list[str] | Nonerr) hostsr&worker_optionsscheduler_optionsr5r,r/rreturnc  s2pipi|pi}|dkr(tdt|t@r`ddlm}td|d||f|S|sxtdt|dt t rt t |krt d t t rt t |krt d t |dt tr҈nd|t t rdnd d } fd dt|ddD} t| | fddi|S)a Deploy a Dask cluster using SSH The SSHCluster function deploys a Dask Scheduler and Workers for you on a set of machine addresses that you provide. The first address will be used for the scheduler while the rest will be used for the workers (feel free to repeat the first hostname if you want to have the scheduler and worker co-habitate one machine.) You may configure the scheduler and workers by passing ``scheduler_options`` and ``worker_options`` dictionary keywords. See the ``dask.distributed.Scheduler`` and ``dask.distributed.Worker`` classes for details on the available options, but the defaults should work in most situations. You may configure your use of SSH itself using the ``connect_options`` keyword, which passes values to the ``asyncssh.connect`` function. For more information on these see the documentation for the ``asyncssh`` library https://asyncssh.readthedocs.io . Parameters ---------- hosts List of hostnames or addresses on which to launch our cluster. The first will be used for the scheduler and the rest for workers. connect_options Keywords to pass through to :func:`asyncssh.connect`. This could include things such as ``port``, ``username``, ``password`` or ``known_hosts``. See docs for :func:`asyncssh.connect` and :class:`asyncssh.SSHClientConnectionOptions` for full information. If a list it must have the same length as ``hosts``. worker_options Keywords to pass on to workers. scheduler_options Keywords to pass on to scheduler. worker_class The python class to use to create the worker(s). remote_python Path to Python on remote nodes. Examples -------- Create a cluster with one worker: >>> from dask.distributed import Client, SSHCluster >>> cluster = SSHCluster(["localhost", "localhost"]) >>> client = Client(cluster) Create a cluster with three workers, each with two threads and host the dashdoard on port 8797: >>> from dask.distributed import Client, SSHCluster >>> cluster = SSHCluster( ... ["localhost", "localhost", "localhost", "localhost"], ... connect_options={"known_hosts": None}, ... worker_options={"nthreads": 2}, ... scheduler_options={"port": 0, "dashboard_address": ":8797"} ... ) >>> client = Client(cluster) Create a cluster with two workers on each host: >>> from dask.distributed import Client, SSHCluster >>> cluster = SSHCluster( ... ["localhost", "localhost", "localhost", "localhost"], ... connect_options={"known_hosts": None}, ... worker_options={"nthreads": 2, "n_workers": 2}, ... scheduler_options={"port": 0, "dashboard_address": ":8797"} ... ) >>> client = Client(cluster) An example using a different worker class, in particular the ``CUDAWorker`` from the ``dask-cuda`` project: >>> from dask.distributed import Client, SSHCluster >>> cluster = SSHCluster( ... ["localhost", "hostwithgpus", "anothergpuhost"], ... connect_options={"known_hosts": None}, ... scheduler_options={"port": 0, "dashboard_address": ":8797"}, ... worker_class="dask_cuda.CUDAWorker") >>> client = Client(cluster) See Also -------- dask.distributed.Scheduler dask.distributed.Worker asyncssh.connect r r'r) SSHClusterzNote that the SSHCluster API has been replaced. We're routing you to the older implementation. This will be removed in the futurerjz(`hosts` must be a non empty list, value z found.zYWhen specifying a list of connect_options you must provide a dictionary for each address.zQWhen specifying a list of remote_python you must provide a path for each address.rhrDoptionsc sTi|]L\}}|t|ttr n |dttrD|dnddqS)r*)r%r&rr,r/rp)r isinstancer#list)rFrGhostr&r/r,rlrrrHs zSSHCluster..r*Nr.ro)r+setold_cluster_kwargsZdistributed.deploy.old_sshror1r2 setdefaultreprrrrslen RuntimeErrorrgr# enumerater) rkr&rlrmr5r,r/rZ OldSSHClusterr$workersrrurro+sRa      ro)NNNNr r!N) __future__rr-loggingrVr1rjsonrtypingrrQZ dask.configZdistributed.deploy.specrr getLoggerrr^rrrgrwrorrrrsP    U