0Fie DddlmZddlZddlZddlZddlZddlZddlZddlm Z ddl Z ddl Z ddl m Z ddlmZddlmZmZddlmZddlmZejeZGd d eejZGd d eZGd deZddddddddejf d$d#ZdS)%) annotationsN)Any) CPU_COUNT)WINDOWS)ProcessInterface SpecCluster)nprocesses_nthreads)parse_memory_limitcdeZdZUded<fdZd fd Zejd dZd fd Z xZ S) Subprocessz!asyncio.subprocess.Process | Noneprocessctrtdd|_tdS)Nz$Subprocess does not support Windows.)r RuntimeErrorr super__init__self __class__s =lib/python3.11/site-packages/distributed/deploy/subprocess.pyrzSubprocess.__init__s>  GEFF F  returnNonecK|d{Vtd{VdSN)_startrstartrs rrzSubprocess.start"sQkkmmggmmoorc KdS)zStart the subprocessN)rs rrzSubprocess._start&s rcK|jr|jjtj|jjdD]}||j|jd{Vd|_t d{VdS)NT) recursive) r returncodepsutilProcesspidchildrenkill communicaterclose)rchildrs rr(zSubprocess.close*s < -DL3; (899BBTBRR   L     ,**,, , , , , , , , ggmmoorrr) __name__ __module__ __qualname____annotations__rrabcabstractmethodrr( __classcell__rs@rr r s.... ####rr cDeZdZUdZded<ded< d d fd Zd ZxZS) SubprocessSchedulerzA local Dask scheduler running in a dedicated subprocess Parameters ---------- scheduler_kwargs: Keywords to pass on to the ``Scheduler`` class constructor dictscheduler_kwargs str | NoneaddressN dict | NonecZ|pi|_tdSr)r6rr)rr6rs rrzSubprocessScheduler.__init__@s.!1 6B rcKdddtjdi|jdg}td|t j|dt jj id{V|_ |j j d{V }|std t|d |vr3|d d |_nt|dS) Ndaskspec--speczdistributed.Schedulerclsopts stderrTzScheduler failed to startz Scheduler atz Scheduler at:)jsondumpsr6loggerinfojoinasynciocreate_subprocess_exec subprocessPIPEr rCreadlinedecodestriprsplitr8debug)rcmdlines rrzSubprocessScheduler._startGs^    J/9RD??? KK % % %%%#zz/::1=CCEE    Trr)r6r9r+r,r-__doc__r.rrr1r2s@rr4r44s|)-rr4c^eZdZUdZded<ded<ded<ded< ddfd ZddZxZS)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 r7namestr scheduler worker_classr5 worker_kwargsdistributed.NannyNr9rrc||_||_||_tj|pi|_t dSr)rYr[r\copyr]rr)rr[r\rYr]rs rrzSubprocessWorker.__init__usN "(!Y}':;; rc Kdd|jdtj|ji|jdg}t d|tj |d{V|_ dS)Nr<r=r>r?rB) r[rErFr\r]rGrHrIrJrKr )rrSs rrzSubprocessWorker._starts   N  Jt0:PT=O:PQQ R R    CHHSMM"""$;SAAAAAAA r)r^NN) r[rZr\rZrYr7r]r9rrr*rUr2s@rrXrXas  NNN 0%)        B B B B B B B BrrXz:8787r^hostr7scheduler_portintr6r9dashboard_addressrZr\ n_workers int | Nonethreads_per_workerr] silence_logskwargsrrrc ` trtd|sd}|pi}|pi}||t\}}||tdt|z}|r9|7tdt t jt|z }|rd|vrtdd|t|d<|Jtj |||du|d|}tj |||d |}td |id } t||d d  fd t|D} td| | d|d| S)aLCreate a scheduler and workers that run in dedicated subprocesses This creates a "cluster" of a scheduler 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.1NrD memory_limitauto)rG)rbport dashboardre)rbnthreadsrir6)r@options)r\r]ci|]}|Srr).0iworkers r z%SubprocessCluster..s333Qq&333rSubprocessCluster)workersr[rurYrir)rrr maxrrdmathceilr rGtoolzmerger4rXranger) rbrcr6rer\rfrhr]rirjr[rxrus @rrwrwsvJHIII !'RM'-2/7(;(=(=% %/;9(::;; K'/ C )i2G(H(H$I$IJJ ^=88(: Ay) ) ) n%  {"*$6!2   K*(   M#  0 I $0=QQF4333% "2"2333G   !       r)rbr7rcrdr6r9rerZr\rZrfrgrhrgr]r9rirdrjrrr) __future__rr/rJr`rEloggingrztypingrr"r| dask.systemrdistributed.compatibilityrdistributed.deploy.specrrdistributed.deploy.utilsr distributed.worker_memoryr getLoggerr+rGABCr r4rXWARNrwrrrrs""""""    !!!!!!------AAAAAAAA888888888888  8 $ $!378*********Z*B*B*B*B*Bz*B*B*B\$($+ %)!% tttttttr