U /eJ@sddlmZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZddlmZddlZddlmZmZmZmZdd lmZdd lmZdd lmZdd lm Z dd l!m"Z"ddl#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*e+e,Z-Gddde(Z.dS)) annotationsN)suppress) isawaitable)Any)parse)IOLoop) _deprecated format_bytesparse_timedeltatypename) get_template)PeriodicCallback)Status)Adaptive) SchedulerInfo)LogLogs LoopRunner NoOpAwaitableSyncMethodMixinformat_dashboard_link log_errorsc@seZdZUdZdZdZded<d]dd Zedd d d Z e j d dddd Z eddZ e j ddZ ddZ ddZ ddZd^ddZejfddZddZd d!Zefd"d#d$d%d&d'Zd(dd)d*d+Zd,d-Zd_d.d/Zd`d0d1Zed2d3d4d5Zd6d7Zed8d9Zd:d;Zdd?Z!d@dAZ"dBdCZ#dDdEZ$dFdGZ%dHdIZ&dJdKZ'edLd dMdNZ(edOdPZ)dQdRZ*edSdTZ+edUdVZ,edWdXZ-dYdZZ.d[d\Z/dS)bClusteraSuperclass for cluster objects This class contains common functionality for Dask Cluster manager classes. To implement this class, you must provide 1. A ``scheduler_comm`` attribute, which is a connection to the scheduler following the ``distributed.core.rpc`` API. 2. Implement ``scale``, which takes an integer and scales the cluster to that many workers, or else set ``_supports_scaling`` to False For that, you should get the following: 1. A standard ``__repr__`` 2. A live IPython widget 3. Adaptive scaling 4. Integration with dask-labextension 5. A ``scheduler_info`` attribute which contains an up-to-date copy of ``Scheduler.identity()``, which is used for much of the above 6. Methods to gather logs TNz IOLoop | None_Cluster__loopFcCst||d|_dii|_i|_d|_d|_g|_||_d|_d|_ t |dd|_ d|_ |dkrrt tdd}|tt|d|_tj|_dS)N)loop asynchronousworkerssecondsdefault)nametype)r _loop_runnerscheduler_infoperiodic_callbacks_watch_worker_status_comm_watch_worker_status_task_cluster_manager_logsquietscheduler_comm _adaptiver _sync_interval_sync_cluster_info_taskstruuidZuuid4r r# _cluster_inforcreatedstatus)selfrrr*r"Zscheduler_sync_intervalr5>/tmp/pip-unpacked-wheel-g426oqom/distributed/deploy/cluster.py__init__>s(  zCluster.__init__)returncCs |j}|dkr|jj|_}|SN)rr$r)r4rr5r5r6r^sz Cluster.looprNone)valuer8cCs*tjdtdd|dkr td||_dS)Nz'setting the loop property is deprecated) stacklevelzexpected an IOLoop, got None)warningswarnDeprecationWarning ValueErrorr)r4r;r5r5r6riscCs |jdSNr"r1r4r5r5r6r"rsz Cluster.namecCs||jd<dSrBrC)r4r"r5r5r6r"vscs|jIdH}d|_|ddiIdHt|IdH|_||_t | ||_ |jj dgidIdH}|j |t ||_|jD] }|qtj|_dS)NzCluster worker statusopZsubscribe_worker_statuscluster-manager-info)keysr )r+Z live_commr"writerreadr%r'asyncioZ ensure_future_watch_worker_statusr( get_metadatar1update_sync_cluster_infor.r&valuesstartrrunningr3)r4comminfopcr5r5r6_startzs"   zCluster._startcsd}d}d|j}|jtjkrz&|jjdg|jdIdHd}WnJtj k r^YqYn2t k r|d7}||krt j ddd YnXt ||jd |}t|IdHqdS) Nr rF)rGr;rzWFailed to sync cluster info multiple times - perhaps there's a connection issue? Error:T)exc_infog?)r-r3rrQr+Z set_metadatar1copyrJCancelledError Exceptionloggerwarningminsleep)r4Z err_countZwarn_atZ max_intervalintervalr5r5r6rNs*    zCluster._sync_cluster_infoc s|jtjkrdStj|_tt|jW5QRX|jrL|j IdH|j r^|j IdH|j r|j tt j|j IdHW5QRX|jr|jIdH|jD] }|qtj|_dSr9)r3rclosedclosingrAttributeErrorr,stopr'closer(r.cancelrJrZr+Z close_rpcr&rO)r4rTr5r5r6_closes$      zCluster._closec CsL|jtjkr|jrtSdStt|j|j|dW5QRSQRXdS)N)Zcallback_timeout) r3rrarrr RuntimeErrorsyncrg)r4timeoutr5r5r6res   z Cluster.closecCs^t|dtjtjkrZz t|}Wn$tk rDdt|}YnX|d|t|ddS)Nr3zwith a broken __repr__ zunclosed cluster )source)getattrrrareprr[object__repr__ResourceWarning)r4_warnZself_rr5r5r6__del__s  zCluster.__del__c slz|IdH}Wntk r*YqZYnXt |D]\}}|||q8W5QRXq|IdHdS)z>Listen to scheduler for updates on adding and removing workersN)rIOSErrorr_update_worker_statusre)r4rRZmsgsrEmsgr5r5r6rKs  zCluster._watch_worker_statuscCsV|dkr0|d}|jd||j|n"|dkrF|jd|=n td||dS)Naddrremovez Invalid op)popr%rMrA)r4rErurr5r5r6rts zCluster._update_worker_statusztype[Adaptive]rr)rkwargsr8c KsPtt|jW5QRXt|ds.i|_|j|||f|j|_|jS)zTurn on adaptivity For keyword arguments see dask.distributed.Adaptive Examples -------- >>> cluster.adapt(minimum=0, maximum=10, interval='500ms') _adaptive_options)rrcr,rdhasattrrzrM)r4rryr5r5r6adapts   z Cluster.adaptint)nr8cCs tdS)zScale cluster to n workers Parameters ---------- n : int Target number of workers Examples -------- >>> cluster.scale(10) # scale cluster to ten workers N)NotImplementedError)r4r~r5r5r6scale s z Cluster.scalecCs(|jtj|f|js$t|dS)aCLog a message. Output a message to the user and also store for future retrieval. For use in subclasses where initialisation may take a while and it would be beneficial to feed back to the user. Examples -------- >>> self._log("Submitted job X to batch scheduler") N)r)appenddatetimenowr*print)r4logr5r5r6_logs z Cluster._logc st}|r(tddd|jD|d<|rX|jIdH}tddd|D|d<|r|dkrhd}|jj|dIdH}|D]$\}}tdd d|D||<q|S) N css|]}|dVqdS)rNr5).0liner5r5r6 /sz$Cluster._get_logs..rcss|]\}}|VqdSr9r5rlevelrr5r5r6r4sZ SchedulerT)rcss|]\}}|VqdSr9r5rr5r5r6r;s)rrjoinr)r+get_logsZ worker_logsitems) r4cluster schedulerrlogsLdkvr5r5r6 _get_logs*szCluster._get_logscCs|j|j|||dS)aeReturn logs for the cluster, scheduler and workers Parameters ---------- cluster : boolean Whether or not to collect logs for the cluster manager scheduler : boolean Whether or not to collect logs for the scheduler workers : boolean or Iterable[str], optional A list of worker addresses to select. Defaults to all workers if `True` or no workers if `False` Returns ------- logs: Dict[str] A dictionary of logs, with one item for the scheduler and one for each worker )rrr)rir)r4rrrr5r5r6r?s zCluster.get_logsr)Z use_insteadcOs |j||Sr9)r)r4argsryr5r5r6rVsz Cluster.logscCsJddlm}z |}|r*|j|kr*|WSWntk r@YnX||S)zReturn client for the cluster If a client has already been initialized for the cluster, return that otherwise initialize a new client object. r)Client)Zdistributed.clientrcurrentrrA)r4rZcurrent_clientr5r5r6 get_clientZs  zCluster.get_clientcCs\z|jdd}Wntk r(YdSX|jdddddd}t||SdS) NZservicesZ dashboardz://r/r:)r%KeyErrorscheduler_addresssplitr)r4porthostr5r5r6dashboard_linkjs $zCluster.dashboard_linkcCs|jr|jjrd}nd}t|jd}t|drJtdd|jD}nt|dr`t|j}n|}||krp|n |d|}d|d |d S) NrZManualr worker_speccss&|]}d|krdn t|dVqdS)grouprN)len)rZeachr5r5r6r{sz*Cluster._scaling_status..z / zM
Scaling mode: zB
Workers: z$
) r,Zperiodic_callbackrr%r{sumrrOr)r4moder requestedZ worker_countr5r5r6_scaling_statusts"    zCluster._scaling_statuscszjWStk rYnXz,ddlm}m}m}m}m}m}m }m }Wnt k rfd_YdSX|dd} | j rJ|dd| d|d| d} |dd | d|dd | d|d | d} ||| g|| gg|d d d} d| _| dd| ddfdd} | | tfdd}| |n|d} ||}|| gg|_|dd|dd|_fddttjjdddfdd}j||S) z3Create IPython widget for display within a notebookr)HTML AccordionButtonHBoxIntTextLayoutTabVBoxNZ150px)widthZWorkers) descriptionlayoutZScaleZMinimumZMaximumZAdaptZ500px)Z min_width)rzManual ScalingrzAdaptive ScalingcsjjjddS)N)minimummaximum)r|r;)b)rrr4rMr5r6adapt_cbsz!Cluster._widget..adapt_cbc s8j}ttjW5QRX|dSr9)r;rrcr,rdr)rr~)requestr4rMr5r6scale_cbs   z!Cluster._widget..scale_cbrrZScalingcs__dSr9) _repr_html_r;rr5) scale_statusr4r3r5r6rMs zCluster._widget..updatez(distributed.deploy.cluster-repr-intervalmsrcs$td}|jd<|dS)Niz cluster-repr)r r&rP)rT)cluster_repr_intervalr4rMr5r6installs z Cluster._widget..install)Z_cached_widgetrc ipywidgetsrrrrrrrr ImportErrorr_supports_scalingZselected_index set_titleZon_clickrrchildrenr daskconfiggetrZ add_callback)r4rrrrrrrrrrr|Z accordionrrtabrr5)rrrrrr4r3rMr6_widgetsV,             zCluster._widgetcCsRz|j}Wntk r&d}YnXtdjt|j|j|jd|j||dS)NzScheduler not started yet.zcluster.html.j2r)r#r"rrscheduler_info_reprcluster_status) r%rrcr renderr#__name__r"r)r4rrr5r5r6rs zCluster._repr_html_cKsddlm}|}|rvddl}t|jtdkrh|jf|p@i}t||d<||d<||ddq||f|nt||d }||dddS) z%Display the cluster rich IPython reprr)displayNz8.0.0 text/plain text/htmlT)raw)rr) ZIPython.displayrrr parse_version __version__Z_repr_mimebundle_rmr)r4ryrZwidgetrZ mimebundler5r5r6_ipython_display_s   zCluster._ipython_display_cCs ||jSr9)ri __aenter__rDr5r5r6 __enter__ szCluster.__enter__cCs||j|||Sr9)ri __aexit__)r4exc_type exc_value tracebackr5r5r6__exit__ szCluster.__exit__ccs|Sr9r5rDr5r5r6 __await__szCluster.__await__cs|IdH|Sr9r5rDr5r5r6rs zCluster.__aenter__cs|}t|r|IdHdSr9)rer)r4rrrfr5r5r6rszCluster.__aexit__r/cCs|js dS|jjS)Nz)r+addressrDr5r5r6rszCluster.scheduler_addresscCst|dt|jS)N_name)rlr#rrDr5r5r6_cluster_class_name"szCluster._cluster_class_namec Cszd|j|j|jt|jdtdd|jdDf}dd|jdD}t|rn|dtt|7}|d7}|S) Nz!%s(%s, %r, workers=%d, threads=%drcss|]}|dVqdS)ZnthreadsNr5rwr5r5r6r,sz#Cluster.__repr__..cSsg|] }|dqS)Z memory_limitr5rr5r5r6 /sz$Cluster.__repr__..z , memory=)) rr"rrr%rrOallr )r4textZmemoryr5r5r6ro&s zCluster.__repr__cCs t|jSr9setrrDr5r5r6plan6sz Cluster.plancCs t|jSr9rrDr5r5r6r:szCluster.requestedcCsdd|jdDS)NcSsh|] }|dqS)r"r5)rrr5r5r6 @sz#Cluster.observed..r)r%rOrDr5r5r6observed>szCluster.observedcCst|t|ko|j|jkSr9)r#r")r4otherr5r5r6__eq__BszCluster.__eq__cCst|Sr9)idrDr5r5r6__hash__EszCluster.__hash__)FNFNr)N)TTT)TTT)N)0r __module__ __qualname____doc__rr__annotations__r7propertyrsetterr"rUrNrgrer>r?rrrKrtrr|rrrrrrrrrrrrrrrrrrrrorrrrrr5r5r5r6r$sp      %      U     r)/ __future__rrJrloggingr0r> contextlibrinspectrtypingrZpackaging.versionrrZtornado.iolooprZ dask.configrZ dask.utilsrr r r Z dask.widgetsr Zdistributed.compatibilityr Zdistributed.corerZdistributed.deploy.adaptiverZdistributed.objectsrZdistributed.utilsrrrrrrr getLoggerrr\rr5r5r5r6s(           $