U /ek6@s UddlmZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlmZddlmZmZmZddlm Z m!Z!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(m)Z)m*Z*m+Z+m,Z,ddl-Z-ddl.Z.dd l/m0Z0m1Z1m2Z2dd l3m4Z4dd l5m6Z6ddl7Z7dd l8m9Z9m:Z:m;Z;dd l8mm?Z?ddl@mAZAmBZBmCZCddlDmEZEddlFmGZGddlHmIZImJZJddlKmLZLddlMmNZNmOZOmPZPmQZQmRZRmSZSddlTmUZUddlVmWZWddlXm%Z%ddlYmZZZddl[m\Z\ddl]m^Z^ddl_m`Z`ddlambZcddldmeZeddlfmgZgmhZhmiZimjZjmkZkmlZlmmZmmnZnddlompZpmqZqdd lrmsZsmtZtmuZumvZvmwZwddlrmbZxdd!lrmyZyz ddlzZ7Wne{k rYnXzdd"l|m}Z}Wn$e{k r6d#d$d%d&Z}YnXe ~eZd'd(e jjjDZd)Zehd*d+d,d-iZd.d/Ze-jd0d1Ze-jd2d3Zee7jjZd4d5Zd6d7Zd8d9Zd:d;Zdd?Zd@dAZdBdCZdDdEZdFdGZd|dIdJZd}dKdLZd~dMdNZddPdQZddRdSZddTdUZdVdWZdXdYZdZd[d\d]d^ZGd_d`d`Ze%fdadbZeeZdcedd<e$ZGdedfdfZdgdhZdidjZddkdlZdmdnZddodpZdqdrZemdsdtZe ddudvZejddwdxZe-jdydzZe-jd{d|Ze-jd}d~Ze-jddZe-jddZe-jddZeZe-jddZe dddZe-jddZe-jddZe-jddZddZddZe dddZedfddddddZdeqddfddddddddddZdd[dddZdd[dddZdd[dddZddZddedeqd-ddddOddd-dfdddddd#ddddddd#dddddZddddd[dddZdd[dddZddd[dddZe ddd#dddddŜddDŽZΐdddʄZϐddd̄Ze1dd΄ZeуrddЄZn e-jӠdѡZҐdddԄZdefddքZאdddلZؐdddۄZdd݄Zdd߄ZddZe e jdfddZe ddZe ddZe ddZejejejedZddZddZddZddZddZdddZdddZddZddZe ddZe ddZdddddZdd[ddd Ze dd#ddd dddZe ddZe ddZe dddZe-jddZGdddeWZGdddeGZGdddeOZdd Zdd!d"Zd-d d#d$d$d#d#d[d%d&d'Zd$dd(d)d*ZGd+d,d,eEZd-dd#d.d/d0Ze d1d2d1d2d3d4d5d6Zd7d8Ze-jd9d:d;d<Zdd=d>d?d=d@dAdBZGdCdDdDeqZGdEdFdFeqZGdGdHdHeqZ e d-dIdJd#d dKdLdMZ e dNdOdPdQdRZ dSdTddUdVdd[dWdXdYZ dSdTdZdJddÐd[d\d]d^Z e-jd_d`Ze-jdadbgdcdddeZe-dfdgZe-dhdiZdjddkdldmZddjd2dndodpdqZddjd2drdodsdtZdudvZeZGdwdxdxZdjdxdydzd{ZdS() annotationsN) defaultdict)Callable CollectionMapping)contextmanager nullcontextsuppress)countsleep)IOAny GeneratorIteratorLiteral)assocmemoizemerge)AsyncHTTPClient)IOLoop)Event Schedulersystem)versions) BatchedSend)Client_global_clientsdefault_client)Comm)TCP)MACOSWINDOWS)initialize_logging)CommClosedErrorConnectionPoolStatusclean_exceptionconnectrpc) SpecCluster) WorkerPlugin)time)Nanny) ServerNode)enable_proctitle_on_children) deserialize) TaskState)Security) DequeHandler_offload_executorget_ipget_ipv6get_mp_contextiscoroutinefunction log_errorsreset_logger_locks)WORKER_ANY_RUNNINGWorker)ComputeTaskEventExecuteInvalidTransition SecedeEventStateMachineEvent) WorkerState) is_debuggingboolreturncCs tdk SN)sysgettracerJrJ:/tmp/pip-unpacked-wheel-g426oqom/distributed/utils_test.pyrC`srCcCs$i|]\}}t|tjr||jqSrJ) isinstanceloggingLoggerlevel).0nameloggerrJrJrK is rScCsdSrGrJrJrJrJrKprUz1distributed.scheduler.active-memory-manager.startFcs"tjD]}|jddIdHqdS)NFZ executor_wait)r< _instancesclose)workerrJrJrKcleanup_global_workersxs r[cCs|SrGrJ)loop_in_threadrJrJrKloop}sr]c #stjtjjddd}tfdd}|t|}tj|fD]^}|kr\}}z |VW5| |j XqL||krL|W5QRW5QRdSqLW5QRXW5QRXdS)Nz test IOLoop)Zthread_name_prefixcs0t}t}||f|IdHdSrG)rcurrentasynciorZ set_resultwait)io_loop stop_eventZ loop_startedrJrKrunszloop_in_thread..run) concurrentfuturesZFutureZThreadPoolExecutorconfig_for_cluster_testssubmit_run_and_close_tornadoZ as_completedresult add_callbackset)cleanupZtpereZranfrbrcrJrdrKr\s&     r\cCs$tjjtjjttdSrG)daskconfigclearupdatecopydeepcopyoriginal_configrJrJrJrK reset_configs rwcstfdd}|S)z A decorator to disable debug facilities during timing-sensitive tests. Warning: this doesn't affect already created IOLoops. c sDtjd}|dk rtjd=z||WS|dk r>|tjd<XdS)NPYTHONASYNCIODEBUG)osenvironget)argskwargsZold_asyncio_debugfuncrJrKwrappeds znodebug..wrapped) functoolswraps)rrrJr~rKnodebugs rcCs$tjd|_|jdk r tjd=dS)za A setup_module() that you can install in a test module to disable debug facilities. rxN)ryrzr{_old_asyncio_debugmodulerJrJrKnodebug_setup_modules rcCs|jdk r|jtjd<dS)ze A teardown_module() that you can install in a test module to reenable debug facilities. Nrx)rryrzrrJrJrKnodebug_teardown_modules rcCs|dSNr^rJxrJrJrKincsrcCs|dSrrJrrJrJrKdecsrcCs||SrGrJryrJrJrKmulsrcCs||SrGrJrrJrJrKdivsrcCs tddS)Nzhello!) RuntimeErrorrrJrJrKthrowssrcCs|dSNrJrrJrJrKdoublesr{Gz?cCst||dSrr rdelayrJrJrKslowincsrcCst||dSrr rrJrJrKslowdecsrcCst|d|Srr rrJrJrK slowdoublesrr^cCs"ddlm}t|||dS)Nr)randomr^)rr )rZscalerrJrJrK randomincs rcCst|||SrGr )rrrrJrJrKslowaddsrcCst|t|SrG)r sum)seqrrJrJrKslowsumsrcOs0|dd}t|t|dkr(|dS|SdS)Nrrr^r)r{r len)r|r}rrJrJrK slowidentity s   rc Cs$||dW5QRSQRXdSrrJ)rlockrJrJrKlock_incsrrNone)eventrFcCs |dSrG)ra)rrJrJrKblock_on_eventsrc@seZdZdZddZdS)_UnhashableCallableNcCs|dSrrJ)selfrrJrJrK__call__sz_UnhashableCallable.__call__)__name__ __module__ __qualname____hash__rrJrJrJrKrsrcCs||}||krq dS)z* Burn CPU for *duration* seconds. NrJ)durationZtimerdeadlinerJrJrKrun_for#s  rzdefaultdict[str, int] _varying_dictc@seZdZddZddZdS) _ModuleSlotcCs||_||_dSrG)modnameslotname)rrrrJrJrK__init__2sz_ModuleSlot.__init__cCsttj|j|jSrG)getattrrHmodulesrrrrJrJrKr{6sz_ModuleSlot.getN)rrrrr{rJrJrJrKr1srcs&ttdttfdd}|S)zi Return a function that returns a result (or raises an exception) from *items* at each call. rcsN}|}|tkr"tn(|}|d|<t|trF|n|SdSr)r{r IndexErrorrL Exception)dctiritemskeyZslotrJrKrFs   zvarying..func)rrnext_varying_key_gen)rrrJrrKvarying:s  rcCsdd}|ttt|fS)zi Like *varying*, but return the full specification for a map() call on multiple items lists. c_s |||SrGrJ)rr|r}rJrJrKapply\szmap_varying..apply)listmapr)Z itemslistsrrJrJrK map_varyingVsrcst|IdH|dSr)r`r rrJrJrKasyncincbsrc s:dfdd}zt|WSjddXdS)NcstIdHSrG)rr_rJr|async_fnr}Z tornado_looprJrKinner_fnjsz(_run_and_close_tornado..inner_fnT)Zall_fds)rYr`re)rr|r}rrJrrKrjgs rjc s6tj| fdd}t|W5QRXdS)Nc sz tfdddIdH}Wn:tk rZ}ztD]}|q:W5d}~XYn*XtD]}|jqd|IdHdS)NT 127.0.0.1)validatehostport)rrrangeputaddressfinished) schedulerexc_r}nputsrqrJrKrxs  zrun_scheduler.._)rprqrmrj)rrrqrr}rrJrrK run_schedulerus rc sltj|Vddlmtt2|fdd}tt rTt |W5QRXW5QRXdS)Nr)r<c szt}zfddiIdH}Wn0tk rV}z||fW5d}~XYn X||jf|IdHdSNrT)rygetpidrrrrpidrZrr<r}rscheduler_addrrJrKrs zrun_worker.._) rprqrm distributedr<r:r9r{rLstrrjr scheduler_qrqr}rrJrrK run_workers  rc sFtj|0|fdd}ttr8t|W5QRXdS)Nc szt}ztfddiIdH}Wn0tk rV}z||fW5d}~XYn X||jf|IdHdSr)ryrr-rrrrrr}rrrJrKrs zrun_nanny.._)rprqrmr{rLrrjrrJrrK run_nannys  rc#sJtjdtddttjdVfddfdd}||dS)Nz/check_active_rpc is deprecated - use gen_test()r) stacklevelcstdttjdSNz!some RPCs left active by test: %spytestfailrmr)activerJ active_beforerJrKrszcheck_active_rpc..failcs tfdddIdHdS)NcstttjdkSNrrrmr)rrJrrJrKrUrVz0check_active_rpc..wait..timeout fail_func)async_wait_forrJractive_rpc_timeoutrrJrKras  zcheck_active_rpc..wait)warningswarnDeprecationWarningrmr)rZrun_sync)r]rrarJrrKcheck_active_rpcs  rcs<ttjdVfdd}tfdd||dIdHdS)NcstdttjdSrrrJrrJrKrsz _acheck_active_rpc..failcstttjdkSrrrJrrJrKrUrVz$_acheck_active_rpc..r)rmr)rr)rrrJrrK_acheck_active_rpcs   rc cs$t\}}||fVW5QRXdSrG)cluster)r]rworkersrJrJrKcluster_fixtures rcCs |\}}|SrGrJrrrrJrJrKssrcCs|\}}|dSrrJrrJrJrKasrcCs|\}}|dSrrJrrJrJrKbsrc cs.|\}}t|d|d }|VW5QRXdSNr)r]rr]rrrclientrJrJrKr sr ccsR|j}|r|j|V|j}|r<|s<|jn|sN|rN|jdS)zSync client with the Active Memory Manager (AMM) turned off. This works regardless of the AMM being on or off in the dask config. N)Zammrunningstopstart)r beforeafterrJrJrK client_no_amms    rc cs.|\}}t|d|d }|VW5QRXdSrrrrJrJrKclient_secondarysrc ks\|pt}t|pid|}t|p"id|}tf||d|\}}||fVW5QRXdS)Nsecurity) worker_kwargsscheduler_kwargs)tls_only_securityrr)rrrr}rrrJrJrKtls_cluster_context%s rc cs(t|d\}}||fVW5QRXdS)N)r)r)r]rrrrJrJrK tls_cluster3src cs0|\}}t|d||d }|VW5QRXdS)Nr)rr]r)rr]rrrr rJrJrK tls_client9srcCstSrG)rrJrJrJrKr@srcCs0|D] }|q|D]}||qdSrG)killjoinrY) processesprocrJrJrK_kill_join_processesEs  rcCs|||jdSrG)rYZ join_threadZ_writer)rrJrJrK _close_queueOsrr c #sB|pi}|pi}|pi}ttddtt|rJt}nt}t}g}|t |t } |t | t j dt| d|f|dd} | || i} t } |t | tD]P} tdtjd|}t j d|| | |f|d}|||d |i| |j<q| }t|tr>||tD].} | \}}t|trj||| |d <qJtz|d }d |d iWntk riYnXfdd}t|d idd| DfVW5QRXz t}Wntk rYn X| W5QRXW5QRXW5QRXdS)NTcheckzDask cluster test: Schedulerr^)rQtargetr|r}daemon)nthreads memory_limitzDask cluster test: Worker)rQr!r|r}rrrZconnection_argsr c s^tf4IdH:}|IdH}t|kr2qJtdkrtdqW5QIdHRXdS)NzTimeout on cluster creation)r)Zncores_runningrr,r)rr#nworkersZ rpc_kwargsZsaddrr rJrKwait_for_workerss  z!cluster..wait_for_workerscSs$g|]}|dt|ddqS)rr)rr)weakrefrefrPwrJrJrK szcluster..)!r/check_process_leakcheck_instancesrhrr contextlib ExitStackcallbackrr7QueuerProcessrr appendrrr MEMORY_LIMITrr{rLrr,Zget_connection_argsKeyErrorrjvaluesr ValueErrorrY)r'ZnannyrrrrqZ _run_workerstackrrrZworkers_by_pidrrr}rZsaddr_or_exceptionrZaddr_or_exceptionrr(r rJr&rKrUs "                  rfloatzdict[str, Any] | NonezCallable[[Callable], Callable])r clean_kwargsrFcs<pistdtrdfddfdd}|S)aCoroutine test @pytest.mark.parametrize("param", [1, 2, 3]) @gen_test(timeout=5) async def test_foo(param) await ... # use tornado coroutines @gen_test(timeout=5) async def test_foo(): await ... # use tornado coroutines \timeout should always be set and it should be smaller than the global one frompytest-timeoutc sPt4IdH2tt|||IdHW5QIdHRSQIdHRXdSrG)rr`wait_for create_task)rr|r}rrJrKasync_fn_outers z gen_test..async_fn_outercs8tttffdd}t|_|S)Ncs"tstdtf||S)Nz,gen_test only works for coroutine functions.)r8rrj)r|r})rBrrJrK test_funcsz&gen_test.._..test_func)rrrhcleaninspect signature __signature__)rrC)rBr<r~rKrs  zgen_test.._)AssertionErrorrC)rr<rrJ)rBr<rrKgen_tests rIz-list[tuple[str, int] | tuple[str, int, dict]]rz Security | dict[str, Any] | Noneztype[ServerNode]z"tuple[Scheduler, list[ServerNode]])r#rrr<rrrFcs|pi}pitfdd|d|IdHfddt|D}tj|IdHt}tjt|kstddjDstddj Dr t d IdHt|d kr`tjd d|DIdH IdHt t ttd q`|fS) NTr)rrrrc sPg|]H\}}jf|d|d|ddt|dkrDt|dnqS)r^Tr)r#rQrrrr)rrr)rPrZncorer<rrrrJrKr-s z!start_cluster..css|]}|jtjkVqdSrG)statusr&r )rPwsrJrJrK sz start_cluster..css|]}|jdkVqdSrG)comm)rPrNrJrJrKrMs{Gz?rTcss|]}|jddVqdS)r^rAN)rYr+rJrJrKrMszCluster creation timeout)r enumerater`gatherr,rranyr8Z stream_commsr rY check_invalid_worker_transitionscheck_invalid_task_statescheck_worker_fail_hard TimeoutError)r#rrr<rrrr rJrJrK start_clusters<  rWr)rrFcCs^|jdsdS|jdD]*\}}|d}td|ttf|qtdt|jddS)Nzinvalid-worker-transitionrZWorker:z Invalid worker transitions found)eventsr{popprintr?r9r)rrmsgrZrJrJrKrS"s    rScCsb|jdsdS|jdD]:\}}td|dtd|d|dD] }t|qFqtddS)Nzinvalid-worker-task-staterXrZzState:statestoryzInvalid worker task state)rYr{r[r9)rrr\linerJrJrKrT0s   rTcCs|jdsdS|jdD]z\}}|}|d}t|dj|dj|d<t|dj|dj|d<td|tf|\}}}|st | |qdS)Nzworker-fail-hardrZ exception tracebackz Failed worker) rYr{rtrZr0headerframesr[r'rHwith_traceback)rrr\rZrtbrJrJrKrU=s   rUcsbtdddtjfdd|DIdH|IdH|t|t|t|dS)NzClosing out test clusterc s,ttjtt|IdHW5QRXdSrG)r r`rVr$EnvironmentErrorrY)r,rJrJrK end_workerOszend_cluster..end_workerc3s|]}|VqdSrGrJr+rgrJrKrMSszend_cluster..) rRdebugr`rQrYr rSrTrU)rrrJrhrK end_clusterLs rjrZtest_cluster_dumpz4list[tuple[str, int] | tuple[str, int, dict]] | Nonezstr | Literal[False])r#rrrr<r rr client_kwargsrrqr<allow_unclosedcluster_dump_directoryrFcsddlm dkrddg p"i p*ip2ip:ipBi sPtdtrZd ttddd d   tttjd d d  fd d}|S)Nrr)rr^)rrr=r>Fz:0P)Z dashboardZdashboard_addresstransition_counter_max)r$Z death_timeoutrocststdttfdditf  fddt}dgdt }r| dd|j ||j fdd |j Dd _S) Nz/gen_cluster only works for coroutine functions.z!distributed.comm.timeouts.connectZ5sc sjtj fddtjfdd f ddfdd}t|S) Nc sHr>|jfdd4IdH }|VW5QIdHRXndVdS)NT)rZ asynchronous)r)rc)rr rkrrJrK_client_factoryszBgen_cluster.._..test_func.._client_factoryc sg}d}ztdD]}z"tdIdH\}}WnRtk r}z4tjd|j j d|ddd t dIdHW5d}~XYqX||dd<qq|dkrtd ||fVW5t||IdHttdIdHXdS) NFr^<)rr<rrzFailed to start gen_cluster: : z ; retryingT)exc_infozCould not start cluster) rjr`r?r[rrWrrRerror __class__rr )rrrrLe)r<r#rrrrrJrK_cluster_factorys4 " zCgen_cluster.._..test_func.._cluster_factoryc sd}tj4IdH\}}|4IdHb}|g|}|dk rZ|g|}zF|}t|}tt| }|IdH}t|f|Wntjk r>|st t }|j |drt ||jdIdH||stdIdHqt|f|td d|dYn\tjjk rXYnBtk rrtdst ||jdIdHYnXW5QIdHRXW5QIdHRXz t}Wntk rYnX|jddIdHd d } zft!} t!| d kr6t"#| s"qZtd IdHqrNt$d| n t%d| W5tjt|D]:} t| d drtz| jWnt k rYnXqtX|W5QRSQRXdS)N)file)rrL output_dir func_namerOzTest timeout after z6s. ========== Test stack trace starts here ========== xfailT)fastcSs"ddtjDddtDS)NcSsg|]}|s|qSrJclosedrPrqrJrJrKr- szegen_cluster.._..test_func..async_fn..get_unclosed..cSsg|]}|jdkr|qSrrKrrJrJrKr- s )rrXrr8rJrJrJrK get_unclosedszQgen_cluster.._..test_func..async_fn..get_uncloseddatars皙?zUnclosed Comms: Unclosed Comms)&rprqrmr`r@r?Zshieldvalidate_staterVrHioStringIO print_stackdump_cluster_statercancelZ cancelledr getvaluerr}rhas_pytestmarkrr9_closerrXrrrrrOSErrorr,gcZcollectr[r) rkrrrqr|coroZtaskZcoro2bufferrr,r ) rrryrlrmrqrr} outer_argsrCrrJrKrs        (   z;gen_cluster.._..test_func..async_fnc srtd4IdHPrDtjddIdHW5QIdHRSIdHW5QIdHRSQIdHRXdS)N)rrrA)rr`r?rJ)rrrrJrKrB?s,zAgen_cluster.._..test_func..async_fn_outer)r0asynccontextmanagerrj)rr}rB)rr<rrlr rkrmrqrr#rrrrCrr)rrryrr}rrKrCs  fz)gen_cluster.._..test_funcr^rcsg|]\}}|jkr|qSrJ) arguments)rPrQp)boundrJrKr-Os z*gen_cluster.._..) parameters)r8rrrrhrDrErFrinsert bind_partialreplacerrrG)rZorig_sigr|rr<rrlr<r rkrmrqr#rrrrr)rrrCrKrs$ 0    zgen_cluster.._)rrrHrCrdictrr6)r#rrrr<r rrrkrrqr<rlrmrrJrrK gen_cluster[sD  (6rzlist[ServerNode])rrLr{r|rFc s|}t}|r"t|dtr2dd|D}n*|jddiddIdH}d d|D}|||d }tj|d d tj ||d }t |d} t || W5QRXtd|dS)zA variant of Client.dump_cluster_state, which does not rely on any of the below to work: - Having a client at all - Client->Scheduler comms - Scheduler->Worker comms (unless using Nannies) rcSsi|]}|j|qSrJ)r_to_dictr+rJrJrKrSjsz&dump_cluster_state..opZ dump_staterF)r\on_errorNcSs(i|] \}}|t|tr t|n|qSrJ)rLrreprrPkvrJrJrKrSms)rrrT)exist_okz.yamlr,zDumped cluster state to )rversion_module get_versionsrLr< broadcastrrymakedirspathropenyamlZ safe_dumpr[) rrLr{r|Zscheduler_infoZ versions_infoZ workers_infor]fnamefhrJrJrKr[s"  rzScheduler | Worker | Nanny)serversrFcGsB|D]8}t|tr"|jr"|qt|tr|jjr|qdS)zRun validate_state() on the Scheduler and all the Workers of the cluster. Excludes workers wrapped by Nannies and workers manually started by the test. N)rLrrrr<r])rrrJrJrKr~s  rzsubprocess.Popen)rterminate_timeoutrFcCsf|dkrbtjdr&|tjn |tjz|j |dW5tt | W5QRXXdS)NwinrA) pollrHplatform startswith send_signalsignalZCTRL_BREAK_EVENTSIGINTr rr communicate)rrrJrJrK_terminate_processs    rz list[str]rz!Iterator[subprocess.Popen[bytes]])r|capture_outputr kill_timeoutr}rFcks6|rtj|d<tj|d<tjdr.tj|d<t|}tjdr^tj tj d|d|d<ntj tj d|d|d<tj |f|}z |VW5zt||W5|j |d\}}|rtd |jd |d tt|tr|n||r$td |jd |d tt|tr|n|XXW5QRXd S)aStart a shell command in a subprocess. Yields a subprocess.Popen object. On exit, the subprocess is terminated. Parameters ---------- args: list[str] Command line arguments capture_output: bool, default False Set to True if you need to read output from the subprocess. Stdout and stderr will both be piped to ``proc.stdout``. If False, the subprocess will write to stdout/stderr normally. When True, the test could deadlock if the stdout pipe's buffer gets full (Linux default is 65536 bytes; macOS and Windows may be smaller). Therefore, you may need to periodically read from ``proc.stdout``, or use ``proc.communicate``. All the deadlock warnings apply from https://docs.python.org/3/library/subprocess.html#subprocess.Popen.stderr. Note that ``proc.communicate`` is called automatically when the contextmanager exits. Calling code must not call ``proc.communicate`` in a separate thread, since it's not thread-safe. When captured, the stdout/stderr of the process is always printed when the process exits for easier test debugging. terminate_timeout: optional, default 30 When the contextmanager exits, SIGINT is sent to the subprocess. ``terminate_timeout`` sets how many seconds to wait for the subprocess to terminate after that. If the timeout expires, SIGKILL is sent to the subprocess (which cannot be blocked); see ``kill_timeout``. If this timeout expires, `subprocess.TimeoutExpired` is raised. kill_timeout: optional, default 10 When the contextmanger exits, if the subprocess does not shut down after ``terminate_timeout`` seconds in response to SIGINT, SIGKILL is sent to the subprocess (which cannot be blocked). ``kill_timeout`` controls how long to wait after SIGKILL to join the process. If this timeout expires, `subprocess.TimeoutExpired` is raised. kwargs: optional optional arguments to subprocess.Popen stdoutstderrr creationflagsScriptsrbinrAz------ stdout: returncode z,  ------z------ stderr: returncode N) subprocessPIPESTDOUTrHrrZCREATE_NEW_PROCESS_GROUPrryrrprefixPopenrr[ returncoderLbytesdecoder)r|rrrr}routerrrJrJrKpopens*2      rrcCsHt|}|sDt|t|kr |dk r0|td|dq dSNzcondition not reached until z seconds)r,r rr predicaterrZperiodrrJrJrKr?s  r?csPt|}|sLt|IdHt|kr |dk r8|td|dq dSr)r,r`r rrrrJrJrKrs  rcCstddkrdSd}}zdzDttjtj}|d|dt| dd}WW"dSt k rzYW dSXW5|dk r||dk r|XdS) z Return whether IPv6 is locally functional. This doesn't guarantee IPv6 is properly configured outside of localhost. Z DISABLE_IPV61FN)z::rr%rT) rygetenvrYsocketAF_INET6 SOCK_STREAMbindlistencreate_connection getsocknamer)ZservclirJrJrKhas_ipv6s   rcCs|SrGrJ)rCrJrJrK requires_ipv6srz ipv6 required?cs&t|fd|i|IdH}|dS)zh Check that it is possible to connect to the distributed *addr* within the given *timeout*. rN)r(abort)addrrr}rNrJrJrKassert_can_connectsrc s<t|(t|fd|i|IdH}|W5QRXdS)zj Check that it is impossible to connect to the distributed *addr* within the given *timeout*. rN)rraisesr(r)rrZexception_classr}rNrJrJrKassert_cannot_connect$s rtcpcsvtd||ff|td|t|ff|g}trb|td||ff|td|t|ff|g7}tj|IdHdS)zT Check that the local *port* is reachable from all IPv4 and IPv6 addresses. %s://127.0.0.1:%d %s://%s:%d %s://[::1]:%d %s://[%s]:%dN)rr5rr6r`rQrprotocolr}rgrJrJrK&assert_can_connect_from_everywhere_4_60srcsvtd||ff|td|t|ff|g}trb|td||ff|td|t|ff|g7}tj|IdHdS)zK Check that the local *port* is reachable from all IPv4 addresses. rrrrN)rr5rrr6r`rQrrJrJrK$assert_can_connect_from_everywhere_4@srcs|td|f|g}tdkr8|tdt|ff|g7}trh|td|f|tdt|ff|g7}tj|IdHdS)zR Check that the local *port* is only reachable from local IPv4 addresses. tcp://127.0.0.1:%dr tcp://%s:%dtcp://[::1]:%d tcp://[%s]:%dN)rr5rrr6r`rQrr}rgrJrJrKassert_can_connect_locally_4Rs rcsfts ttd|f|tdt|ff|td|f|tdt|ff|g}tj|IdHdS)zK Check that the local *port* is reachable from all IPv6 addresses. rrrrNrrHrr5rr6r`rQrrJrJrK$assert_can_connect_from_everywhere_6as rcsxts ttd|f|tdt|ff|td|f|g}tdkrd|tdt|ff|g7}tj|IdHdS)zR Check that the local *port* is only reachable from local IPv6 addresses. rrrz::1rNrrrJrJrKassert_can_connect_locally_6os  rc cst|trt|}|j}|jdd}|dk r<|j}||_t}t |g|jdd<| |z |VW5||jdd<| ||dk r||_XdS)z%Capture output from the given Logger.N) rLrrM getLoggerrOhandlers propagaterr StreamHandlersetLevel)rRrOrZ orig_levelZ orig_handlersZorig_propagatesiorJrJrKcaptured_loggers      rccs:t|tjst|j}t|_z |jVW5||_XdS)z4Capture output from the given logging.StreamHandler.N)rLrMrrHstreamrr)handler orig_streamrJrJrKcaptured_handlers   rc csxddlm}tjj}t|}z8||t|tj||t|dVW5|||t|XdS)z6 Temporarily change configuration dictionary. r)defaultsN) distributed.configrrprqrtrurrrsr#) new_configrrqZ orig_configrJrJrKrs    rc csddl}tjd}tjdd\}}z\t|d}|| |W5QRX|tjd<z dVW5|rt|tjd<ntjd=XW5t|XdS)zH Temporarily change configuration file to match dictionary *c*. rNZ DASK_CONFIGz dask-configrr,) rryrzr{tempfilemkstempremovefdopenwritedump)rqrZold_filefdrrorJrJrKnew_config_files    r testscCs&tjt|}tj|s"t||S)z; Get the path to one of the test TLS certificates. )ryrr certs_direxistsrH)filenamerrJrJrKget_certsrcCs6td}td}ddd|d|id|id|idiiiS)z= A functional TLS configuration with our test certs. tls-ca-cert.pemztls-key-cert.pemrrNZtlscert)zca-filer rrZ)r)ca_fileZkeycertrJrJrK tls_configsrcCst}d|ddd<|S)zg A functional TLS configuration with our test certs, disallowing plain TCP communications. TrrNzrequire-encryption)r)rqrJrJrKtls_only_configsrc Cs tt t}W5QRX|S)z: A Security object with proper TLS configuration. )rrr2secrJrJrK tls_securitys rc Cs*tt t}W5QRX|js&t|S)zg A Security object with proper TLS configuration and disallowing plain TCP communications. )rrr2Zrequire_encryptionrHrrJrJrKrs  r tls-cert.pem tls-key.pemrcCs<tjtjjt|d}d|_tj|_|t|t||SN)cafileF) sslcreate_default_contextPurpose CLIENT_AUTHrcheck_hostname CERT_REQUIRED verify_modeload_cert_chaincertfilekeyfilerctxrJrJrKget_server_ssl_contexts r*cCs<tjtjjt|d}d|_tj|_|t|t||Sr) rrr  SERVER_AUTHrr"r#r$r%r&rJrJrKget_client_ssl_contexts r,c Csxtd}z0||\}}||kr8|||t||fWn8tk rr}ztd|d|W5d}~XYnXdS)Nresourcezrlimit too low (z) and can't be increased: )r importorskipZ getrlimitZ setrlimitminrskip)limitZdesiredr-ZsofthardrxrJrJrK bump_rlimit"s r3cKs&|dddgtfdtd|S)Nr#)tls://127.0.0.1r^)r4rr4)rr) setdefaultrr)r}rJrJrKgen_tls_cluster,sr6c csbtj}tj}z dVW5ttjD]\}}||kr"tj|=q"tjD]}||krFtj|=qFXdSrG)rHrrrPkeys)Z old_modulesold_pathrelemrJrJrKsave_sys_modules3s  r:c #stdVt}fddtD}|s2qntdt|dkrddlm}t}zft|dg}t|d D]F\}}| d |d t|d |d | d | ||j qxW5~Xt jd |ddqdS)z7Context manager to ensure we haven't leaked any threadsNcs"g|]}|krd|jkr|qS)zwatch message queuerQ)rPthreadZactive_threads_startrJrKr-Ks z%check_thread_leak..rOr%r)profilez! thread(s) were leaked from test r^z#------ Call stack of leaked thread /rtr F)Zpytrace) threadingrPr,r rr>rH_current_framesrr5rZ call_stackidentrr)r Z bad_threadsr>rclinesrr<rJr=rKcheck_thread_leakBs.   rFzlist[multiprocessing.Process])rrFcCsHt}t}|sgS|t|}|dkr2|S|dj|dqdS)zWait until timeout for get_mp_context().active_children() to terminate. Return list of active subprocesses after the timeout expired. rrAN)r,r7active_childrenr)rt0childrenZ join_timeoutrJrJrKwait_active_childrenks rJcCs^t}|D] }|qt|d}|D] }|q*tdd}|rZtd|tsZtdS)zSend SIGTERM to get_mp_context().active_children(), wait up to 3 seconds for processes to die, then send SIGKILL to the survivors rArTz(Leaked unkillable children processes: %sN) r7rG terminaterJrrRwarningr"rH)rrIrrJrJrKterm_or_kill_active_children}s      rMT(zIterator[None])r  check_timeout term_timeoutrFc csFt|dz*dV|r2t|d}|r2td|W5t|dXdS)aTerminate any currently-running subprocesses at both the beginning and end of this context Parameters ---------- check : bool, optional If True, raise AssertionError if any processes survive at the exit check_timeout: float, optional Wait up to these many seconds for subprocesses to terminate before failing term_timeout: float, optional After sending SIGTERM to a subprocess, wait up to these many seconds before sending SIGKILL rANzTest leaked subprocesses: )rMrJrH)r rPrQrIrJrJrKr.s  r.c cstjtjtjtjtjtjtjt jt t jdVt }t t rtdt |dksntqnt tjD]B}tt0|jj|jdd|jtkr|j|jW5QRXqtjt }tddtjDr&tdt |dkstqtjtdD],}tddt jDrZqntdq8d d t jD}t jtd |td dt jDstd dt jDtddtjDstttjtjt jtdS)Ng?rFrWcss|]}|jdkVqdS)rNrrrJrJrKrMsz"check_instances..r%css|]}|VqdSrGrrrJrJrKrMscSsg|]}|s|qSrJrrrJrJrKr-sz#check_instances..rcss$|]}|jtjtjtjhkVqdSrG)rKr&rinitfailedrPnrJrJrKrMscSsi|] }||jqSrJrrTrJrJrKrSsz#check_instances..css|]}|jtjkVqdSrG)rKr&rrrJrJrKrMs)rrXrrr<rr*Z_initialized_clientsSchedulerTaskStateWorkerTaskStater-rrr,rmr rHr rr]rlrYrKr;rRrallr9rr3Zclear_all_instances)r r,rLrJrJrKr/sZ                     r/c ksZttjjtdddf|.tD]\}}t | |q,dVW5QRXdS)zNSet recommended config values for tests that create or interact with clusters.z500 msF)Zlocal_directoryzdistributed.admin.tick.intervalz"distributed.worker.profile.enabledN) rwrprqrmr gettempdirlogging_levelsrrMrr)Z extra_configrQrOrJrJrKrhs rhc csbtd|rtnt>t|d(|r2tnt dVW5QRXW5QRXW5QRXdS)Nr)r`Zset_event_looprFrr.r/)threadsZ instancesrrJrJrKrDs   rDc cst dVW5QRXdSrG)rDrJrJrJrKrnsrnc@s eZdZdZddZddZdS)TaskStateMetadataPluginz)WorkPlugin to populate TaskState.metadatacCs|jj|_dSrG)r]tasks)rrZrJrJrKsetup szTaskStateMetadataPlugin.setupcKsH|j|}|dkr(|dkr(t|jd<n|dkrD|dkrDt|jd<dS)Nready executing start_timeZmemoryZ stop_time)r^r,metadata)rrr finishr}tsrJrJrK transitions  z"TaskStateMetadataPlugin.transitionN)rrr__doc__r_rfrJrJrJrKr] sr]c@s8eZdZddZddZd ddZdd d Zd d ZdS) LockedCommcCs0||_||_||_||_||_t|ts,tdSrG) write_event write_queue read_event read_queuerNrLr rH)rrNrkrlrirjrJrJrKrs zLockedComm.__init__cCs t|j|SrG)rrNrrQrJrJrK __getattr__"szLockedComm.__getattr__NmessagecsN|jr |j|jj|fIdH|jr6|jIdH|jj|||dIdHS)N) serializersr)rjrrN peer_addressrirar rr\rprrJrJrKr %s zLockedComm.writecsN|jj|dIdH}|jr4|j|jj|fIdH|jrJ|jIdH|S)N) deserializers)rNreadrlrrqrkra)rrsr\rJrJrKrt,s zLockedComm.readcs|jIdHdSrG)rNrYrrJrJrKrY4szLockedComm.close)Nro)N)rrrrrnr rtrYrJrJrJrKrhs   rhc@s2eZdZdZd ddZddZddZd d ZdS) _LockedCommPoolaHA ConnectionPool wrapper to intercept network traffic between servers This wrapper can be attached to a running server to intercept outgoing read or write requests in test environments. Examples -------- >>> w = await Worker(...) >>> read_event = asyncio.Event() >>> read_queue = asyncio.Queue() >>> w.rpc = _LockedCommPool( w.rpc, read_event=read_event, read_queue=read_queue, ) # It might be necessary to remove all existing comms # if the wrapped pool has been used before >>> w.rpc.remove(remote_address) >>> async def ping_pong(): return await w.rpc(remote_address).ping() >>> with pytest.raises(asyncio.TimeoutError): >>> await asyncio.wait_for(ping_pong(), 0.01) >>> read_event.set() >>> await ping_pong() NcCs"||_||_||_||_||_dSrG)rirjrkrlpool)rrvrkrlrirjrJrJrKrSs z_LockedCommPool.__init__cCs t|j|SrG)rrvrmrJrJrKrn\sz_LockedCommPool.__getattr__cs,|jj||IdH}t||j|j|j|jSrG)rvr(rhrkrlrirj)rr|r}rNrJrJrKr(_sz_LockedCommPool.connectcs|jIdHdSrG)rvrYrrJrJrKrYesz_LockedCommPool.close)NNNN)rrrrgrrnr(rYrJrJrJrKru8s rucCsBtdz tWn&tk r<tr6tjddYnXdS)zWork around https://github.com/dask/distributed/issues/5601 where any test that inits Security.temporary() crashes on MacOS GitHub Actions CI Z cryptographyzdistributed#5601)reasonN)rr.r2 temporary ImportErrorr!r}rJrJrJrKxfail_ssl_issue5601is   rzc Cst}d}|D]}zt|dks(tdt|ts:tdt|dtrP|dsXtdt|dtsntd|r||dkstd |d |dkr|d ksntd |d}Wqtk r}z"td |d|dt|W5d}~XYqXqdS)aLTest that a story is well formed. Parameters ---------- story: list[tuple] Output of Worker.story ordered_timestamps: bool, optional If False, timestamps are not required to be monotically increasing. Useful for asserting stories composed from the scheduler and multiple workers grz too shortz not a tuplezstimulus_id not a stringzTimestamp is not a floatz*Timestamps are not monotonically ascendingr>r^zTimestamps is too oldzMalformed story event: z Problem: z . in story: N)r,rrHrLtuplerr; _format_story)r^ordered_timestampsnowZprev_tsevrrJrJrKassert_valid_storyvs * r)strictrz list[tuple])r^expectrrrFc Cst||ddd}zX|r0t|t|kr0tt|}|D],}t|}|||sh||dd|r@q._valid_event..)rrXzip)r ev_expectrJrJrK _valid_eventsz"assert_story.._valid_eventNr{zassert_story(strict=z) failed story: z expect: )rr StopIterationiterrrHr~)r^rrrrZstory_itrrrJrJrK assert_storys&9  r)r^rFcCs |sdSdddd|DS)Nz (empty story)z- z - css|]}t|VqdSrG)r)rPrrJrJrKrMsz _format_story..)r)r^rJrJrKr~sr~c@s@eZdZdZdZddZddZddZd d d Zdd d Z dS) BrokenCommr@cCsdSrGrJrrJrJrKrYszBrokenComm.closecCsdS)NTrJrrJrJrKrszBrokenComm.closedcCsdSrGrJrrJrJrKrszBrokenComm.abortNcCs tdSrGr)rrsrJrJrKrtszBrokenComm.readcCs tdSrGrrrrJrJrKr szBrokenComm.write)N)NN) rrrrqZ local_addressrYrrrtr rJrJrJrKrs rr)rCrQrFcs"t|dg}tfdd|DS)zReturn True if the test function is marked by the given @pytest.mark.; False otherwise. FIXME doesn't work with individually marked parameters inside @pytest.mark.parametrize Z pytestmarkc3s|]}|jkVqdSrGr;)rPmarkr;rJrKrMsz!has_pytestmark..)rrR)rCrQZmarksrJr;rKrs rz5type[BaseException] | tuple[type[BaseException], ...]z str | NonezGenerator[(None, None, None)])expected_exceptionmatchexpected_cause match_causerFc csrtj||d }dVW5QRX|j}|js0tt|j|s@||rnt|t|jsntd|d|jddS)zyContextmanager to assert that a certain exception with cause was raised Parameters ---------- exc_type: )rNz Pattern ``z`` not found in ``z``) rrvalue __cause__rHrLresearchr)rrrrrurrJrJrKraises_with_causes   rcCs|d|d}t|dS)aUCX exception handler for `ucx_loop` during test. Prints the exception and its message. Parameters ---------- loop: object Reference to the running event loop context: dict Dictionary containing exception details. r`roN)r{r[)r]contextr\rJrJrKucx_exception_handler-s rfunction)Zscopeccstd}t}|t||V||ddl}d|j j _ ddl m }|}|jrddl}|j}|jdS)zAllows UCX to cancel progress tasks before closing event loop. When UCX tasks are not completed in time (e.g., by unexpected Endpoint closure), clean up tasks before closing the event loop to prevent unwanted errors from being raised. ucprN)has_cuda_context)rr.r`Znew_event_loopZset_exception_handlerrresetrYZdistributed.comm.ucxrNZucxrZdistributed.diagnostics.nvmlrZ has_contextZ numba.cudaZcudaZcurrent_contextZdevice)rr]rrr)ZnumbarJrJrKucx_loop?s     rrzIO[bytes] | Nonez int | None)rr max_linesrFcCsZ|std}|dk r0||kr0t|d|d|}t|||krL|S|d7}q dS)z Read lines from an IO stream until the match is found, and return the matching line. Prints each line to test stdout for easier debugging of failures. rNz not found in z( log lines. See test stdout for details.r^)rHreadliner[)rrrrr_rJrJrKwait_for_log_linebsrcs,eZdZdZfddZfddZZS)BlockedGatherDepaA Worker that sets event `in_gather_dep` the first time it enters the gather_dep method and then does not initiate any comms, thus leaving the task(s) in flight indefinitely, until the test sets `block_gather_dep` Example ------- .. code-block:: python @gen_test() async def test1(s, a, b): async with BlockedGatherDep(s.address) as x: # [do something to cause x to fetch data from a or b] await x.in_gather_dep.wait() # [do something that must happen while the tasks are in flight] x.block_gather_dep.set() # [from this moment on, x is a regular worker] See also -------- BlockedGetData BlockedExecute cs&t|_t|_tj||dSrG)r`r in_gather_depblock_gather_depsuperrrr|r}rwrJrKrs  zBlockedGatherDep.__init__cs.|j|jIdHtj||IdHSrG)rrmrrar gather_deprrrJrKrs zBlockedGatherDep.gather_dep)rrrrgrr __classcell__rJrJrrKrxs rcs,eZdZdZfddZfddZZS)BlockedGetDataaA Worker that sets event `in_get_data` the first time it enters the get_data method and then does not answer the comms, thus leaving the task(s) in flight indefinitely, until the test sets `block_get_data` See also -------- BlockedGatherDep BlockedExecute cs&t|_t|_tj||dSrG)r`r in_get_datablock_get_datarrrrrJrKrs  zBlockedGetData.__init__cs4|j|jIdHtj|f||IdHSrG)rrmrrarget_data)rrNr|r}rrJrKrs zBlockedGetData.get_data)rrrrgrrrrJrJrrKrs rcsJeZdZdZfddZddddfdd Zd d d fd d ZZS)BlockedExecuteaA Worker that sets event `in_execute` the first time it enters the execute method and then does not proceed, thus leaving the task in executing state indefinitely, until the test sets `block_execute`. After that, the worker sets `in_deserialize_task` to simulate the moment when a large run_spec is being deserialized in a separate thread. The worker will block again until the test sets `block_deserialize_task`. Finally, the worker sets `in_execute_exit` when execute() terminates, but before the worker state has processed its exit callback. The worker will block one last time until the test sets `block_execute_exit`. Note ---- In the vast majority of the test cases, it is simpler and more readable to just submit to a regular Worker a task that blocks on a distributed.Event: .. code-block:: python def f(in_task, block_task): in_task.set() block_task.wait() in_task = distributed.Event() block_task = distributed.Event() fut = c.submit(f, in_task, block_task) await in_task.wait() await block_task.set() See also -------- BlockedGatherDep BlockedGetData csNt|_t|_t|_t|_t|_t|_tj ||dSrG) r`r in_execute block_executein_deserialize_taskblock_deserialize_taskin_execute_exitblock_execute_exitrrrrrJrKrs      zBlockedExecute.__init__rrA)r stimulus_idrFcsV|j|jIdHztj||dIdHWS|j|jIdHXdS)N)r)rrmrrarrrexecute)rrrrrJrKrs   zBlockedExecute.executerWz&tuple[Callable, tuple, dict[str, Any]])rerFcs,|j|jIdHt|IdHSrG)rrmrrar_maybe_deserialize_task)rrerrJrKrs z&BlockedExecute._maybe_deserialize_task)rrrrgrrrrrJrJrrKrs#  r) jump_startr<)r,rrFccsN|jj}|jj}d|j_d|j_dV||j_||j_|rJtj|_tj|_dS)aPrevent any task from transitioning from fetch to flight on the worker while inside the context, simulating a situation where the worker's network comms are saturated. This is not the same as setting the worker to Status=paused, which would also inform the Scheduler and prevent further tasks to be enqueued on the worker. Parameters ---------- w: Worker The Worker on which tasks will not transition from fetch to flight jump_start: bool If False, tasks will remain in fetch state after exiting the context, until something else triggers ensure_communicating. If True, trigger ensure_communicating on exit; this simulates e.g. an unrelated worker moving out of in_flight_workers. rN)r]Ztransfer_incoming_count_limitZ*transfer_incoming_bytes_throttle_thresholdr&ZpausedrKr )r,rZold_count_limitZ old_thresholdrJrJrKfreeze_data_fetchingsrrzIterator[LockedComm])bcommrFccsp|r t|jst|jr$t|j}t}t}t|dd|||_}z |VW5|||_XdS)a Contextmanager blocking writes to a `BatchedSend` from sending over the network. The returned `LockedComm` object can be used for control flow and inspection via its ``read_event``, ``read_queue``, ``write_event``, and ``write_queue`` attributes. On exit, any writes that were blocked are un-blocked, and the original comm of the `BatchedSend` is restored. N)rrHrNr`rr3rhrm)rZ orig_commrirjZ locked_commrJrJrKfreeze_batched_send s"    rrO)intervalzstr | Collection[str]zWorker | Scheduler)rr] dask_workerrrFc st|tr|jj}nt|tr&|j}nt|t|tr>|f}t|dkrZtt t |nt|}z,||ksz||j|krt |IdHqdWnft j t jfk r||krd|d||jd|jd|}nd|d|j}t|YnXdS)zWait for a task to appear on a Worker or on the Scheduler and to be in a specific state or one of a set of possible states. r^Nztasks[z].state=z on z; expected state=z] not found on )rLr<r]r^r TypeErrorrrrrrr`r ZCancelledErrorrVrr[)rr]rrr^Z state_strr\rJrJrKwait_for_state- s$    $"rz=type[StateMachineEvent] | tuple[type[StateMachineEvent], ...]rA)type_rrmatchesrFcsr|jj}d}|r\|d|k r\|d}|D]2t|s8q(tfdd|Dr(Sq(t|IdHq dS)zEWait for a specific stimulus to appear in the log of the WorkerState.Nr|c3s |]\}}t||kVqdSrG)rrrrJrKrMe sz$wait_for_stimulus..)r]Z stimulus_logrLrXrr`r )rrrrlogZlast_evrJrrKwait_for_stimulusU s  rccstddd}|V|dS)zAn empty WorkerStatez 127.0.0.1:1rn)rroN)rBr)r]rJrJrKrLj s rLra long-running)paramsccsddi|_ddi|_|tjdddidd}|tdddgksDt|jdkrb|tddd d |j dj |jksxt|Vd S) zA WorkerState running a single task 'x' with resources {R: 1}. The task may or may not raise secede(); the tests using this fixture runs twice. Rr^rZcompute)rZresource_restrictionsr)rrrg?Zsecede)rZcompute_durationrN) Zavailable_resourcesZtotal_resourcesZhandle_stimulusr=dummyr>rHparamr@r^r])rLrequestZ instructionsrJrJrKws_with_running_taskr s     rcCs |jjSrG)nodeZnodeid)rrJrJrK name_of_test src cst}tdd}dg}t|tkrz@t.}|D]}|||dq6W5QRWqW5QRXWqtk r}z.|jtjkrt d|t dWYqW5d}~XYqXqt d|dVdS)Nc ssPttjtj6}|tjtjd|d|f|d|VW5QRXdS)Nr^r@)rAF_INETr setsockopt SOL_SOCKET SO_REUSEADDRrr)rrrJrJrK _bind_port s  z*requires_default_ports.._bind_portiR")rz4Address already in use. Waiting before running test r^z)Default ports didn't open up in time for ) r,r _TEST_TIMEOUTr0r1 enter_contextrerrnoZ EADDRINUSEr[r rV)rr rZ default_portsr:rrrJrJrKrequires_default_ports s&   rint)rrFcs8t}|d|dIdH}|jdks,t|jdS)Nzhttp://localhost:z/metricsutf8)rfetchcoderHbodyr)r http_clientresponserJrJrKfetch_metrics_body srzdict[str, Any])rrrFcs4ddlm}t|IdH}fdd||D}|S)Nrtext_string_to_metric_familiescs(i|] }dks|jr|j|qSrG)rQrrPfamilyrrJrKrS s  z!fetch_metrics..)prometheus_client.parserrr)rrrtxtZfamiliesrJrrK fetch_metrics s   rzset[str]cs<ddlm}t|IdH}tjfdd||D}|S)z Get all the names of samples returned by Prometheus. This mostly matches list of metric families, but when there's `foo` (gauge) and `foo_total` (count) these will both have `foo` as the family. rrNcs0g|](}dks|jrdd|jDqS)NcSsh|] }|jqSrJr;)rPsamplerJrJrK sz8fetch_metrics_sample_names...)rQrZsamplesrrrJrKr- s z.fetch_metrics_sample_names..)rrrrmunion)rrrrZ sample_namesrJrrKfetch_metrics_sample_names s  rcCsGddd}t|S)Nc@seZdZddZdS)z'_get_gc_overhead.._CustomObjectcSsdSrrJrrJrJrK __sizeof__ sz2_get_gc_overhead.._CustomObject.__sizeof__N)rrrrrJrJrJrK _CustomObject sr)rH getsizeof)rrJrJrK_get_gc_overhead src@s.eZdZdZdddddZdddd Zd S) SizeOfzT An object that returns exactly nbytes when inspected by dask.sizeof.sizeof rrnbytesrFcCsHt|tstdt||tkr:tdtd|d|t|_dS)Nz$Expected integer for nbytes but got zExpected a value larger than z integer but got .)rLrrtype _size_objr9_nbytes)rrrJrJrKr s zSizeOf.__init__rEcCs|jSrG)rrrJrJrKr szSizeOf.__sizeof__N)rrrrgrrrJrJrJrKr s rrcCst|S)zEA function that emulates exactly nbytes on the worker data structure.)r)rrJrJrK gen_nbytes sr)r)r)r)r^)r)r)r)r)r^)r^)NNN)rFNrNN)FrTr)Nr)Nr)r)r)r)rrr)rrr)TrNrO)TTT)T)r)N)N( __future__rr`concurrent.futuresrfr0rtrrrrErrMlogging.configmultiprocessingryrrrrrrHrrBrr) collectionsrcollections.abcrrrrrr itertoolsr r,r typingr rrrrrrZtlzrrrZtornado.httpclientrZtornado.iolooprrprrrrrrZdistributed.batchedrZdistributed.clientrrrZdistributed.commrZdistributed.comm.tcpr Zdistributed.compatibilityr!r"rr#Zdistributed.corer$r%r&r'r(r)Zdistributed.deployr*Zdistributed.diagnostics.pluginr+Zdistributed.metricsZdistributed.nannyr-Zdistributed.noder.Zdistributed.proctitler/Zdistributed.protocolr0Zdistributed.schedulerr1rVZdistributed.securityr2Zdistributed.utilsr3r4r5r6r7r8r9r:Zdistributed.workerr;r<Z distributed.worker_state_machiner=r>r?r@rArWrBZ dask.arrayryZpytest_timeoutrCrrrRrootmanager loggerDictrr[rrirkZNO_AMMr[Zfixturer]r\rurqrvrwrrrrrrrrrrrrrrrrrrrrrr__annotations__rrrrrrjrrrrrrrrrrr rrqrrrrrrrrrIrWrSrTrUrjrrrrrr?rrrrr0rrfrrrrrrINFOrrrr rabspathrdirname__file__rrrrrrr*r,r3r6r:rFrJrMr.r/rhrDrnr]rhrurzrrr~rrrrrrrrrrrrrrLrrrrrrrrrrrJrJrJrKsH                     (                                   g27 ,# M                 ( ;1 &&V *#&#?("(,* $ $