U /e @sddlmZddlZddlZddlZddlZddlZddlmZddl Z ddl Z ddl m Z ddl mZddlmZmZddlmZddlmZdd lmZeeZGd d d eZdddd d dddejf dddddddddddd ddZdS)) annotationsN)Any) CPU_COUNT)WINDOWS)ProcessInterface SpecCluster)nprocesses_nthreads) Scheduler)parse_memory_limitcs~eZdZUdZded<ded<ded<ded<d ed <ddddd ddfdd Zddfdd Zddfdd ZZS)SubprocessWorkeraeA local Dask worker running in a dedicated subprocess Parameters ---------- scheduler: Address of the scheduler worker_class: Python class to use to create the worker, defaults to 'distributed.Nanny' name: Name of the worker worker_kwargs: Keywords to pass on to the ``Worker`` class constructor str scheduler worker_classdict worker_kwargs str | Nonenamez!asyncio.subprocess.Process | Noneprocessdistributed.NannyN dict | NoneNone)r rrrreturncsBtr td||_||_||_t|p(i|_d|_t dS)Nz*SubprocessWorker does not support Windows.) r RuntimeErrorr rrcopyrrsuper__init__)selfr rrr __class__A/tmp/pip-unpacked-wheel-g426oqom/distributed/deploy/subprocess.pyr-szSubprocessWorker.__init__)rc sFtdd|jdtd|j|jdiIdH|_t IdHdS)NZdaskspecz--specr)clsopts) asyncioZcreate_subprocess_execr jsondumpsrrrrstart)rrrr r'>s zSubprocessWorker.startcsj|jrP|jjdkrPt|jjjddD] }|q(|j|jIdHd|_t IdHdS)NT) recursive) r returncodepsutilProcesspidchildrenkillwaitrclose)rchildrrr r0Hs  zSubprocessWorker.close)rNN) __name__ __module__ __qualname____doc____annotations__rr'r0 __classcell__rrrr r s  r z:8787rrintrr z int | Nonerr) hostscheduler_portscheduler_kwargsdashboard_addressr n_workersthreads_per_workerr silence_logskwargsrc  s$tr td|sd}|pi}|p"i}|dkr>|dkr>t\}}|dkr\|dk r\tdt|}|r|dkrtdttt|}|rd|krtdd|t d|d<|dk st t |||dk |d|}t |||d |}t |d } t||d d fd d t|D} tf| | d|d| S)awCreate in-process scheduler and workers running in dedicated subprocesses This creates a "cluster" of a scheduler running in the current process and workers running in dedicated subprocesses. .. warning:: This function is experimental Parameters ---------- host: Host address on which the scheduler will listen, defaults to localhost scheduler_port: Port fo the scheduler, defaults to 0 to choose a random port scheduler_kwargs: Keywords to pass on to scheduler dashboard_address: Address on which to listen for the Bokeh diagnostics server like 'localhost:8787' or '0.0.0.0:8787', defaults to ':8787' Set to ``None`` to disable the dashboard. Use ':0' for a random port. worker_class: Worker class to instantiate workers from, defaults to 'distributed.Nanny' n_workers: Number of workers to start threads: Number of threads per each worker worker_kwargs: Keywords to pass on to the ``Worker`` class constructor silence_logs: Level of logs to print out to stdout, defaults to ``logging.WARN`` Use a falsy value like False or None to disable log silencing. Examples -------- >>> cluster = SubprocessCluster() # Create a subprocess cluster #doctest: +SKIP >>> cluster # doctest: +SKIP SubprocessCluster(SubprocessCluster, 'tcp://127.0.0.1:61207', workers=5, threads=10, memory=16.00 GiB) >>> c = Client(cluster) # connect to subprocess cluster # doctest: +SKIP Scale the cluster to three workers >>> cluster.scale(3) # doctest: +SKIP z+SubprocessCluster does not support Windows.z 127.0.0.1NZ memory_limitauto)logger)r9portZ dashboardr<)r9Znthreadsr?)r"options)rrcsi|] }|qSrr).0iworkerrr sz%SubprocessCluster..SubprocessCluster)workersr rIrr?)rrrmaxrr8mathceilr rCAssertionErrortoolzmerger r ranger) r9r:r;r<rr=r>rr?r@r rLrrHr rKRsb<       rK) __future__rr$rr%loggingrNtypingrr*rQZ dask.systemrZdistributed.compatibilityrZdistributed.deploy.specrrZdistributed.deploy.utilsrZdistributed.schedulerr Zdistributed.worker_memoryr getLoggerr2rCr WARNrKrrrr s4        ;