U /eE@s*UddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlZddlZddlmZmZddlmZmZmZmZmZmZddlmZddlmZddlm Z ddl!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*dd l+m,Z,m-Z-m.Z.dd l/m0Z0ddl1Z1dd l2m3Z3dd l4m5Z5ddl6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>ddl?m@Z@mAZAmBZBddlCmDZDddlEmFZFddlGmHZHmIZImJZJmKZKddlGmLZMddlNmOZOddlPmQZQddlRmSZSmTZTmUZUddlVmWZWmXZXmYZYmZZZm[Z[ddlVm\Z]ddlVm^Z^ddl_m`Z`ddlambZbddlcmdZdmeZeddlfmgZgddlhmiZiddljmkZkdd llmmZmdd!lnmoZompZpdd"lqmrZrdd#lsmtZtdd$lumvZvdd%lwmxZydd&lzm{Z{dd'lzm|Z}dd(l~mZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZdd)lmZmZmZdd*lmZmZdd+lmZdd,lmZmZmZdd-lmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZe$rdd.lmZdd/lmZdd0lamZdd1lmZed2Ze)d3ZeeZe1jàd4Zerevd5Zd6ed7<iZd8ed9<iZd8ed:<eXjeXjeXjhZd;d;d<d=d>Zd?d@dAdBZGdCdDdDeekZdDdEdFdGZddIdEdJdKZdLdMZ|GdNdOdOeӃZddPdQZdgZeFdRdSZdTdUZdddefdVdWZdXdYZeFdRdSZeܡZdZdEd[d\Zd]d^Zd_gZeojd`fdadbZdcddZdedfZdgdhZdidjZdkdlZddmd?dndodpZddqdmd?drdsdtZddvdwZejZdxdyZdzd{dd_d|d}d}d~dddddZeddfddddddddZdd}dddddZddddddZdd?ddddddZdS)) annotationsN) defaultdictdeque)Callable Collection ContainerIterableMappingMutableMapping)Executor)suppress) timedelta isawaitable) TYPE_CHECKINGAnyClassVarLiteralTextIOTypeVarcast)firstkeymappluck)IOLoop)istask) CPU_COUNT)apply format_bytesfuncname key_split parse_bytesparse_timedeltatmpdirtypename) preloadingprofileutils) BatchedSend)LRU)Commconnectget_address_host parse_address)resolve_address)address_from_user_args)OFFLOAD_THRESHOLD)PeriodicCallback randbytes to_thread)ConnectionPoolStatuscoerce_to_address error_messagepingpong)rpc) send_recv)nvml)_get_plugin_name)WorkDir WorkSpace) get_handlers)time) ServerNode) setproctitle)pickle to_serialize)PubSubWorkerExtension)Security)ShuffleWorkerExtension) safe_sizeof)ThreadPoolExecutor)secede) TimeoutError_maybe_complexget_iphas_arg import_file in_async_callis_python_shutting_downiscoroutinefunctionjson_load_robust log_errorsoffload parse_portsrecursive_to_dictsilence_logging thread_statewarn_on_duration)gather_from_workers pack_dataretry_operation)disable_gc_diagnosisenable_gc_diagnosis) get_versions) DeprecatedMemoryManagerAttributeDeprecatedMemoryMonitorWorkerMemoryManager)NO_VALUEAcquireReplicasEvent BaseWorkerCancelComputeEventComputeTaskEventDeprecatedWorkerStateAttributeExecuteFailureEventExecuteSuccessEventFindMissingEvent FreeKeysEventGatherDepBusyEventGatherDepFailureEventGatherDepNetworkFailureEventGatherDepSuccessEvent PauseEventRefreshWhoHasEventRemoveReplicasEventRescheduleEventRetryBusyWorkerEvent SecedeEventStateMachineEventStealRequestEvent TaskState UnpauseEventUpdateDataEvent WorkerState) ParamSpecClient) WorkerPlugin)NannyPTzdistributed.admin.pdb-on-err)Zpubsubshufflezdict[str, type]DEFAULT_EXTENSIONS"dict[str, Callable[[Worker], Any]]DEFAULT_METRICSDEFAULT_STARTUP_INFORMATIONzCallable[P, T])methodreturncs`djdtr:tddddfdd }n"tddd dfd d }|S) zO Decorator to close the worker if this method encounters an exception. worker-z -fail-hardzP.argszP.kwargsr)argskwargsrc s~z|f||IdHWStk rx}zB|jtjtjfkrV|dt|t|t |IdHW5d}~XYnXdSNzworker-fail-hard) Exceptionstatusr5closedclosing log_eventr7logger exception _force_closeselfrrerreason6/tmp/pip-unpacked-wheel-g426oqom/distributed/worker.pywrappers zfail_hard..wrapperrc sxz|f||WStk rr}zB|jtjtjfkrP|dt|t||j t |W5d}~XYnXdSr) rrr5rrrr7rrloop add_callbackrrrrrrs )__name__rR functoolswraps)rrrrr fail_hards  rstrrc s|z"t|jdd|ddIdHWnTttfk r<Yn<tk rv|jjrZ|jsZt j dddt dYnXdS) z Used with the fail_hard decorator defined above 1. Wait for a worker to close 2. If it doesn't, log and kill the process F)nanny executor_waitrNz[Error trying close worker in response to broken internal state. Forcibly exiting worker NOWTexc_info) asynciowait_forcloseKeyboardInterrupt SystemExit BaseExceptionstatevalidaterrcriticalos_exit)rrrrrrsrc-seZdZUdZeZded<eZded<ded<ded <d ed <d ed <ded<ded<ded<ded<ded<ded<ded<ded<ded<d ed<d ed<d ed <d ed!<d"ed#<d"ed$<d"ed%<d&ed'<d(ed)<d*ed+<d,ed-<d.ed/<d0ed1<d2ed3<d4ed5<d6Z d7ed8<d4ed9<d4ed:<d*ed;<d4ed<<d=ed><d*ed?<d@edA<d@edB<d@edC<d=edD<dEedF<dGedH<dedI<dJedK<dLedM<dNedO<dPedQ<d"edR<dSedT<d"edU<iZ dedV<dedW<dXedY<dXedZ<d=ed[<dPed\<ded]<d^ed_<d`eda<dTd6d6d6d6d6d6d6d6d6d6d6d6d6d6d6dbd6e e d6d6d6d6d6dcddd6ded6d6d6d6d6d6dcdfd6d6d6d6d6dg)d4dhd4dhdid4djdkdldmdndhdkdodpdqd4dPdrdsdsd4d4d7d4d4d=d*dd`dldldkdkdldtdudvdwdwdwd4dx*dydzZ d{ed|<ed}d~ddZeZeZeZeZeZeZeZeZeddZeddZeZeddZeZeZ eZ!eZ"eddZ#eddZ$eZ%eZ&eZ'eZ(eZ)eZ*eZ+eZ,eZ-eddZ.eddZ/eZ0eZ1eZ2eZ3eZ4edd~ddZ5ed d~ddZ6d*d~ddZ7eddZ8ddPddddZ9eddZ:eddZ;edddddZ=d*ddddZ?dEd~ddZ@ddZAddZBdedddEdfddZCdddddZDdd~ddZEd"ddddZFdd~ddZGeHdddddZIdUd*dd=ddddZJdd~dd„ZKdddĜddƄZLdVd=d ddɜdd˄ZMfdd̈́ZNeOdWd"d=d=d*d4dМdd҄ZPdXd*dԜddքZQdd؄ZRddڄZSdYdd~dd݄ZTdZdd=d4ddߜddZUd"ddddZVeOd[dd4d=ddddZWeOd*ddddZXd*d*ddddZYdddddZZeHdddfdd Z[eHdddfdd Z\d*ddddZ]dddddZ^dddddZ_d"d"dސdd*ddd d Z`eHd*d d d*dd d dZad*ddddZbd*d"ddddZceOdd~ddZdd\ddZed]ddZfd^djddddZgd_dd~ddZhdd d!d"d#ZieHd*d*dd$d%d&Zjdd'dd(d)d*d+Zkdd~d,d-Zldd~d.d/Zmd`d=d0d1d2Zndad"d@dd3d4d5Zodbd6ddd7d8Zpdd~d9d:Zqdd~d;d<Zrd*dȐd=d>d?Zsed@d~dAdBZtdcd@d@dCdDdEZud*d~dFdGZvdd~dHdIZwedJdKZxedLdMZyedNdOZzedPdQZ{edRdSZ|Z}S(dWorkeraWorker node in a Dask distributed cluster Workers perform two functions: 1. **Serve data** from a local dictionary 2. **Perform computation** on that data and on data from peers Workers keep the scheduler informed of their data and use that scheduler to gather data from other workers when necessary to perform a computation. You can start a worker with the ``dask worker`` command line application:: $ dask worker scheduler-ip:port Use the ``--help`` flag to see more options:: $ dask worker --help The rest of this docstring is about the internal state that the worker uses to manage and track internal computations. **State** **Informational State** These attributes don't change significantly during execution. * **nthreads:** ``int``: Number of nthreads used by this worker process * **executors:** ``dict[str, concurrent.futures.Executor]``: Executors used to perform computation. Always contains the default executor. * **local_directory:** ``path``: Path on local machine to store temporary files * **scheduler:** ``rpc``: Location of scheduler. See ``.ip/.port`` attributes. * **name:** ``string``: Alias * **services:** ``{str: Server}``: Auxiliary web servers running on this worker * **service_ports:** ``{str: port}``: * **transfer_outgoing_count_limit**: ``int`` The maximum number of concurrent outgoing data transfers. See also :attr:`distributed.worker_state_machine.WorkerState.transfer_incoming_count_limit`. * **batched_stream**: ``BatchedSend`` A batched stream along which we communicate to the scheduler * **log**: ``[(message)]`` A structured and queryable log. See ``Worker.story`` **Volatile State** These attributes track the progress of tasks that this worker is trying to complete. In the descriptions below a ``key`` is the name of a task that we want to compute and ``dep`` is the name of a piece of dependent data that we want to collect from others. * **threads**: ``{key: int}`` The ID of the thread on which the task ran * **active_threads**: ``{int: key}`` The keys currently running on active threads * **state**: ``WorkerState`` Encapsulated state machine. See :class:`~distributed.worker_state_machine.BaseWorker` and :class:`~distributed.worker_state_machine.WorkerState` Parameters ---------- scheduler_ip: str, optional scheduler_port: int, optional scheduler_file: str, optional host: str, optional data: MutableMapping, type, None The object to use for storage, builds a disk-backed LRU dict by default. If a callable to construct the storage object is provided, it will receive the worker's attr:``local_directory`` as an argument if the calling signature has an argument named ``worker_local_directory``. nthreads: int, optional local_directory: str, optional Directory where we place local resources name: str, optional memory_limit: int, float, string Number of bytes of memory that this worker should use. Set to zero for no limit. Set to 'auto' to calculate as system.MEMORY_LIMIT * min(1, nthreads / total_cores) Use strings or numbers like 5GB or 5e9 memory_target_fraction: float or False Fraction of memory to try to stay beneath (default: read from config key distributed.worker.memory.target) memory_spill_fraction: float or False Fraction of memory at which we start spilling to disk (default: read from config key distributed.worker.memory.spill) memory_pause_fraction: float or False Fraction of memory at which we stop running new tasks (default: read from config key distributed.worker.memory.pause) max_spill: int, string or False Limit of number of bytes to be spilled on disk. (default: read from config key distributed.worker.memory.max-spill) executor: concurrent.futures.Executor, dict[str, concurrent.futures.Executor], "offload" The executor(s) to use. Depending on the type, it has the following meanings: - Executor instance: The default executor. - Dict[str, Executor]: mapping names to Executor instances. If the "default" key isn't in the dict, a "default" executor will be created using ``ThreadPoolExecutor(nthreads)``. - Str: The string "offload", which refer to the same thread pool used for offloading communications. This results in the same thread being used for deserialization and computation. resources: dict Resources that this worker has like ``{'GPU': 2}`` nanny: str Address on which to contact nanny, if it exists lifetime: str Amount of time like "1 hour" after which we gracefully shut down the worker. This defaults to None, meaning no explicit shutdown time. lifetime_stagger: str Amount of time like "5 minutes" to stagger the lifetime value The actual lifetime will be selected uniformly at random between lifetime +/- lifetime_stagger lifetime_restart: bool Whether or not to restart a worker after it has reached its lifetime Default False kwargs: optional Additional parameters to ServerNode constructor Examples -------- Use the command line to start a worker:: $ dask scheduler Start scheduler at 127.0.0.1:8786 $ dask worker 127.0.0.1:8786 Start worker at: 127.0.0.1:1234 Registered with scheduler at: 127.0.0.1:8786 See Also -------- distributed.scheduler.Scheduler distributed.nanny.Nanny z!ClassVar[weakref.WeakSet[Worker]] _instancesz!ClassVar[weakref.WeakSet[Client]]_initialized_clientsz Nanny | Nonerzthreading.Lock_lockinttransfer_outgoing_count_limitzdict[str, int]threadsactive_threads_lockzdict[int, str]active_threadszset[str] active_keysz defaultdict[str, dict[str, Any]] profile_keysz.deque[tuple[float, dict[str, dict[str, Any]]]]profile_keys_historydict[str, Any]profile_recentz#deque[tuple[float, dict[str, Any]]]profile_historyzdeque[dict[str, Any]]transfer_incoming_logtransfer_outgoing_logtransfer_outgoing_count_totaltransfer_outgoing_bytes_totaltransfer_outgoing_bytestransfer_outgoing_countfloat bandwidthlatencyprofile_cycle_intervalr>Z workspacer=_workdirrlocal_directoryz Client | None_clientz#defaultdict[str, tuple[float, int]]bandwidth_workersz$defaultdict[type, tuple[float, int]]bandwidth_typeszlist[preloading.Preload]preloads str | Nonecontact_addressNz"int | str | Collection[int] | None _start_port _start_host _interface _protocol_dashboard_addressbool _dashboard _http_prefixz float | None death_timeoutlifetimelifetime_staggerlifetime_restartdict extensionsrFsecurityconnection_argsrrzdict[str, Executor] executorsr(batched_streamrnamescheduler_delayzdict[str, BatchedSend] stream_commsheartbeat_intervalservices service_specsrmetricsstartup_informationlow_level_profiler schedulerexecution_statezdict[str, WorkerPlugin]pluginsztuple[WorkerPlugin, ...]_pending_plugins1sF/rauto))scheduler_filenthreadsrrrr reconnectexecutor resources silence_logsrpreload preload_argvrrrrrr interfacehostportprotocoldashboard_address dashboard http_prefixrrrrrrrrtransition_counter_max memory_limitdatamemory_target_fractionmemory_spill_fractionmemory_pause_fraction scheduler_sni int | Nonez IOLoop | Nonez dict | Nonez Any | Nonez bool | Nonez:Executor | dict[str, Executor] | Literal['offload'] | Nonezdict[str, float] | Nonezlist[str] | Nonez"list[str] | list[list[str]] | Nonez Security | dict[str, Any] | Nonezdict[str, type] | Nonez%Mapping[str, Callable[[Worker], Any]]zint | Literal[False]z str | floatzMutableMapping[str, Any] | Callable[[], MutableMapping[str, Any]] | Callable[[str], MutableMapping[str, Any]] | tuple[Callable[..., MutableMapping[str, Any]], dict[str, Any]] | Nonezfloat | Literal[False] | None)* scheduler_ipscheduler_portrrrrrrrrrrrrrrrrrrrrrrrrr r rrrrrrrr r r rrrrc); s| dk r&| rtdntjdtdd|dk r>tjdtdd|_t_tj d}-tj d_ t tj d}.i_ t_i_t_ttj_td d _t_td d _| dkrtj d } td d _td d _d _d _d _d _t tj d_ tdd_!tdd_"d_#d_$|!dkrXtj d}!t%|!dd}!|!snt&'t(|stj dpt)*}t+j,-|d}t.dd*t/|_0j0j1dd_2j2j3_4W5QRX|stj d}|stj d}|dk st&|dk st&t5j6||j4d_7t%| _8|rZt9|j8d}/|/d }0nH|dkrtj d!drtj d!}0n |dkrt:|}0n t:||f}0|_;|dkr|0|_?|r$t@|\}2}3|3Ad#d$kr$|3Bd%s$td&|3|_C|_D|p8tE}| dkr`tj d'} tF| tGs`t&i_H| rvtI| d(tF|tGrtJf|}|ptJ_KtFjKtJst&jKLd)_MtNO_P_Q|+r|+jMd*<tRjStTd$d+d,d-_UtVWd krtTd$d.d,jUd/<| d0kr,jUd0jUd1<n.tF| tGrFjUX| n| dk rZ| jUd1<d1jUkrxtT|d2d,jUd1<tYd3jPd4_Z|_[d _\i_]j4t^j,krt^j,_d j4i_`|_ai_b|pi_c|_d|_e|_f|rtG|ni_g|r tG|ni_h|dkr&tj d5}|_ijjjkjljmjnotpjqtrjsjtjujvjwjxjyjzj{j|j}j~jjjjd6}4jqototototpotototjd7 }5tjf|4|5jMd8|,t|'||&|(|)|*d9_tj}6tj d:}7jjdk rH|7d;k rHtjj|7}6t|jjj j`| |-| |%|6|.d< }8t|8|0_jjjPd=_t%|dd_tjjd>}9|9jd?<tfd@ddA}9|9jdB<tjd>}9|9jdC<|_|dkrt}fdDdE|D_HtdFtj dGr~t%tj dHdd}:tj|:d>}9|9jdI<tj|!d>}9|9jdJ<|"dkrtj dK}"t%|"}"|#dkrtj dL}#t%|#}#|$dkrtj dM}$|$_|"r|"tdd$|#7}"jQj|"jdNdO|"_tjdS)PNzThe `reconnect=True` option for `Worker` has been removed. To improve cluster stability, workers now always shut down in the face of network disconnects. For details, or if this is an issue for you, see https://github.com/dask/distributed/issues/6350.zThe `reconnect` argument to `Worker` is deprecated, and will be removed in a future release. Worker reconnection is now always disabled, so passing `reconnect=False` is unnecessary. See https://github.com/dask/distributed/issues/6350 for details. stacklevelzThe `loop` argument to `Worker` is ignored, and will be removed in a future release. The Worker always binds to the current loopz'distributed.worker.connections.outgoingz'distributed.worker.connections.incomingz/distributed.worker.transfer.message-bytes-limiti)maxlenzdistributed.scheduler.validate順rzdistributed.scheduler.bandwidthcSsdSNrrrrrrr6z!Worker.__init__..cSsdSrrrrrrr8rgMbP?z distributed.worker.profile.cyclems)defaultztemporary-directoryzdask-worker-spacerzCreating scratch directories is taking a surprisingly long time. ({duration:.2f}s) This is often due to running workers on a network file system. Consider specifying a local-directory to point workers to write scratch data to a local disk.rprefixzdistributed.worker.preloadzdistributed.worker.preload-argv)Zfile_dirtimeoutaddresszscheduler-addressz://:r[z;Host address with IPv6 must be bracketed like '[::1]'; got zdistributed.worker.resources)levelworkerserver_hostnamezDask-Actor-Threads)Zthread_name_prefix)rUactorzDask-GPU-ThreadsgpurUrzDask-Default-ThreadsZ2msintervalrz$distributed.worker.profile.low-level)gatherrun run_coroutineget_data update_dataZ free_keys terminateZping upload_file call_stackr&Zprofile_metadataget_logskeysversions actor_executeactor_attributez plugin-addz plugin-removeget_monitor_infobenchmark_diskbenchmark_memorybenchmark_network get_story) rzcancel-computezacquire-replicasz compute-taskz free-keyszremove-replicasz steal-requestzrefresh-who-hasworker-status-change)handlersstream_handlersr)r rr rrrz"distributed.worker.memory.transferF) rr rrrtransfer_incoming_count_limitrr transfer_incoming_bytes_limittransfer_message_bytes_limit)rZioloopr( heartbeatcsddiS)Nop keep-alive) batched_sendrrrrrri`rI find-missingcsi|]\}}||qSrr.0r extensionrKrr 'sz#Worker.__init__..zdask worker [not started]z"distributed.worker.profile.enabledz#distributed.worker.profile.intervalr&z profile-cyclez$distributed.worker.lifetime.durationz#distributed.worker.lifetime.staggerz#distributed.worker.lifetime.restartzworker-lifetime-reachedr) ValueErrorwarningswarnDeprecationWarningr threadingLockrdaskconfiggetrr!rrrsetrrr&createrrrrrrrrrrrrrrrrr"AssertionErrorZ_setup_loggingrtempfile gettempdirrpathjoinrZr>Z _workspaceZ new_work_dirrZdir_pathrr%Zprocess_preloadsrrrSr6rsplitlenrrr-count startswithrrr isinstancerrrXrFrZget_connection_argsrrcurrentrio_loopr'_offload_executorrIrr;device_get_countupdater(rrrrsysinsertrrrrrrrrrrr.r/r0r1r2_handle_remote_stimulusrmrr8r4get_call_stack get_profileget_profile_metadatar6r7r8r9r: plugin_add plugin_remover;r<r=r>r?rgrerhrtryrshandle_worker_status_changerA__init__rcmemory_managermathinfr rr}r rfr9rr$rrr1rGperiodic_callbacks find_missingZ_addressritemsrBtrigger_profile cycle_profilerrandomZ call_laterclose_gracefullyrrradd);rrrrrrrrrrrrrrrrrrrrrrrrrrrr r rrrrrrrrr r r rrrrrrCrEcfgZscheduler_addrZprotocol_address_Z host_addressrArBrDZ transfer_incoming_bytes_fractionrpcZprofile_trigger_intervalrrKrrts.B                                                                  zWorker.__init__rcruzMutableMapping[str, Any]rcCs|jjS)a{task key: task payload} of all completed tasks, whether they were computed on this Worker or computed somewhere else and then transferred here over the network. When using the default configuration, this is a zict buffer that automatically spills to disk whenever the target threshold is exceeded. If spilling is disabled, it is a plain dict instead. It could also be a user-defined arbitrary dict-like passed when initialising the Worker or the Nanny. Worker logic should treat this opaquely and stick to the MutableMapping API. .. note:: This same collection is also available at ``self.state.data`` and ``self.memory_manager.data``. )rur rKrrrr Qsz Worker.datatransfer_incoming_bytes)targetZ*transfer_incoming_bytes_throttle_threshold data_neededtransfer_incoming_count_totalin_flight_tasks_countrErCzset[TaskState]cCs"tdtdd|jjDS)Nz\The `Worker.data_needed` attribute has been removed; use `Worker.state.data_needed[address]`cSsh|]}|D]}|q qSrr)rNtsstsrrr sz%Worker.data_needed..)rRrS FutureWarningrrvaluesrKrrrrs zWorker.data_neededcCstdtt|jjS)Nz_The `Worker.waiting_for_data_count` attribute has been removed; use `len(Worker.state.waiting)`)rRrSrrbrwaitingrKrrrwaiting_for_data_counts zWorker.waiting_for_data_countcCs|j|jkrd|jnd}d|jjd|j|d|jjdt|jd|jjd|jj d t|jj d |jj d t|jj d S) Nz, name: < z , status: z , stored: z , running: rz , ready: z, comm: z , waiting: >) r address_safe __class__rrrbr rexecuting_countrreadyrr)rrrrr__repr__sjzWorker.__repr__cCs|jjSN)Z_deque_handlerrrKrrrlogssz Worker.logszstr | Collection[str]None)topicmsgrcCs:d||d}|jtkr&||n|j|j|dS)Nz log-event)rHrr)Z thread_idrU get_identrJrr)rrrZfull_msgrrrrs zWorker.log_eventcCs|jS)z For API compatibility with Nanny)r$rKrrrworker_addressszWorker.worker_addresscCs |jdS)Nr)rrKrrrrszWorker.executorr5)valuercCs~|j}tj||dt}|||tjkrP|tjkrP|t|dn*|tjkrz|tj tj fkrz|t |ddS)zrOverride Server.status to notify the Scheduler of status changes. Also handles pausing/unpausing. zworker-status-change- stimulus_idN) rrA__set__r@_send_worker_status_changer5runninghandle_stimulusrrpausedclosing_gracefullyr{)rrZ prev_statusrrrrrs   z Worker.status)rrcCs|d|jj|ddS)Nr@)rHrr)rJ_statusr)rrrrrrs z!Worker._send_worker_status_changec s4z|jj\}}Wntk r,d\}}YnXt|jj|jt|jtt |j d|jj ||d|jj |jj |jj|j|j|jd|jd}|j}|D]:\}}d|kr|d\}}} |||i| <q|||<q|jD]L\}} z,| |} t| r| IdH} ||| Wqtk r,YqXq|S)Nr)totalworkerstypes)memoryZdisk)Zincoming_bytesincoming_countZincoming_count_totalZoutgoing_bytesoutgoing_countZoutgoing_count_total) task_countsrZ managed_bytesZ spilled_bytestransferZevent_loop_interval.)r Z spilled_totalAttributeErrorrrrrrrr$rnbytesrZtransfer_incoming_countrrrrZ_tick_interval_observedmonitorrecentrz partition setdefaultrrr) rZspilled_memoryZ spilled_diskoutZmonitor_recentkvZk0rZk1Zmetricresultrrr get_metricssL     zWorker.get_metricsc sXi}|jD]D\}}z&||}t|r2|IdH}|||<Wqtk rPYqXq|Sr)rrzrr)rrrfrrrrget_startup_informations  zWorker.get_startup_informationcCs$t|j|j|jj|jj|jjdS)N)typeidrrr ) rrrrr$rrrur rKrrridentity,s zWorker.identityexcludezContainer[str])rrcstjd}|j|tjj|j|jd}fdd|D}| || |j jd| |j jdt |dS)zDictionary representation for debugging purposes. Not type stable and not intended for roundtrips. See also -------- Worker.identity Client.dump_cluster_state distributed.utils.recursive_to_dict r)rrrXrrcsi|]\}}|kr||qSrrrNrrrrrrPGsz#Worker._to_dict..) super_to_dictrr6rWrXrrrzrjrrurW)rrinfoextrarrrr5s  zWorker._to_dict)rrcCs*|jr&|jjr&|jjs&|j|dS)aJImplements BaseWorker abstract method. Send a fire-and-forget message to the scheduler through bulk comms. If we're not currently connected to the scheduler, the message will be silently dropped! See also -------- distributed.worker_state_machine.BaseWorker.batched_send N)rcommrsend)rrrrrrJQs  zWorker.batched_sendc sjdjdt}jdkr4j_tdzNt}tjjfj IdH}d|_ t |_ |jtddjjj tjjjj ddjjDd djDtjjjjjjjtt !IdH"IdHd tj#d d gd IdH|j$d gd}|IdH}|%drPt&|dt}||d}'|||d|_(t)j*_WqWq>t+k rtdjjt,-dIdHYq>t.k rtdYq>Xq>|ddkr.d|kr|dnt/|}t0d|t1d|n@t,j2fdd|dDIdHtdjjtdj34|jd4jd4j56j7|dS)NrIrG1-------------------------------------------------zWorker->Schedulerzregister-workerFcSs"i|]}|jdkr|j|qS)r)rkey get_nbytes)rNrrrrrPzs z3Worker._register_with_scheduler..cSsi|]\}}|t|qSr)r$rrrrrPszworker-connect-)rHZreplyr$rr7rrrrnowrr rrrpidr8rrrZ server_idmsgpack serializers deserializerswarningrr@Waiting to connect to: %26sg?z&Timed out when connecting to schedulerrOKmessagez Unable to connect to scheduler: z#Unexpected response from register: c3s |]\}}j||dVqdS))rpluginNrq)rNrrrKrr sz2Worker._register_with_scheduler..zworker-pluginsz Registered to: %26s)8rxstopr@rr$rrr+rrrweakrefrefZ_serverwriterrlistr rrtasksrrztotal_resourcesrur r service_portsrrgetpidr`rrrreadrYr_update_latencyrr5rOSErrorrsleeprKreprerrorrQr.rstartrrhandle_scheduler) rr_startrfutureresponseZ_endmiddlerrrKr_register_with_schedulerds        !         zWorker._register_with_scheduler)rrcCs$|d|jd|_|d|dS)N皙?ffffff?r)r digest_metric)rrrrrrszWorker._update_latencyc sFtdjzttjjjIdHfddj Dddj DdIdH}t}|d} ||ddkrt d jd jd d IdHWdS|d |_|ddjd_jjWnNtk rtdYn0tk r@tdIdHYnXdS)Nz Heartbeat: %scs,i|]$}|jjkr|jj|jqSr)rr start_timerNrrrrrrPs z$Worker.heartbeat..cSs$i|]\}}t|dr||qS)rG)hasattrrGrMrrrrPs )r$rr executingrrrmissingz%Scheduler was unaware of this worker z. Shutting down.F)rr@zheartbeat-intervalrFrGz6Failed to communicate with scheduler during heartbeat.z6Unexpected exception during heartbeat. Closing worker.)rdebugr$r@r]rZheartbeat_workerrrrrrzrrrrrx callback_timerclearrrrr)rrendrrrrrGsF         zWorker.heartbeatr*)rrc s.z||IdHW5|jddIdHXdS)Nz)worker-handle-scheduler-connection-brokenr)rZ handle_stream)rrrrrrszWorker.handle_schedulerTz str | bytes)filenamer loadrc stj|j|fdd}t|dkr2||}nt||IdH}|rzttj Wn0t k r}zt ||W5d}~XYnXdt|dS)Nc sLt|tr|}td&}|||t|W5QRX|S)Nwb) rerencodeopenrflushrfsyncfileno)r rZ out_filenamerrfuncs   z Worker.upload_file..funci'r)rr) rr_r`rrbrUrO cache_loadsr rrrr)rrr rr rrr rr4s   zWorker.upload_filez list[str]cCs t|jSr)rr rKrrrr7sz Worker.keyszdict[str, list[str]])who_hasrcslfdd|D}t|jjdIdH\}}}j|dd|r`td|||d|dSd d iSdS) Ncs*i|]"\}}|jkr|dd|DqS)cSsg|] }t|qSr)r6)rNaddrrrr sz,Worker.gather...r rrKrrrPs z!Worker.gather..)r9whoF)r reportz4Could not find data: %s on workers: %s (who_has: %s)z partial-fail)rr7rr)rzr[r9r$r2rr)rr rZ missing_keysZmissing_workersrrKrr.s$  z Worker.gatherrdict[str, float])rrrcCsVt|r|jn |jj|d|jj|jjd}tdkrR|jj|d<|jj |d<|S)N)r) range_queryrc last_timergpu_namegpu_memory_total) rrrrrcrr;rirr)rrrrrrrr;&s     zWorker.get_monitor_infoc stIdHttj}|D]}tj|jjj d}j d}jdkrt| }t t jj|d<zj|f|IdHWnFtk r}z(t|dkr|jtjkrWY q$nW5d}~XYq$X|_qq$tdjdjttjdjd }|jjrpz d dl}Wn t k rRt!"d YnX|j#j$j%j&j'jd t j_(j)dkrj_)j*D]:}z|+IdHWn t,k rt!-d YnXq.j(zdj/j0j(j1f} Wn(t,k r j/j0j(} YnXt!2djt!2d| j)j3krXt!2dj)j45D]*\} } t!2d6| j(dt7| qbt!2djjt!2dt!2dj8j9j:j;rt!2dt<j:j;t!2dj=t>djt?j@fddjADddiIdH} dd| D} t| dkrnt| dkrft!Bd | D]}t!BtC|qP| d d!_Ajj8_DIdHES)"N)rrrrrr()ZtcpZtlsZ default_hostrzCould not start Worker on host z with port zdistributed.worker.http.routes)servermodulesr!rz4To start diagnostics web server please install Bokehr zFailed to start preloadz%s%s:%dz Start worker at: %26sz Listening to: %26sz Worker name: %26sz {:>16} at: {:>26}r%rrz Threads: %26dz Memory: %26sz Local Directory: %26szdask worker [%s]c3s|]}j|ddVqdS)F)r catch_errorsNrrNrrKrrrsz&Worker.start_unsafe..Zreturn_exceptionsTcSsg|]}t|tr|qSr)rer)rNrrrrrs z'Worker.start_unsafe..zVMultiple plugin exceptions raised. All exceptions will be logged, the first is raised.r)Fr start_unsafer_rVrr/rrrrZget_listen_argscopyrMr,rr$listenrrberrnoZ EADDRINUSEZ_start_addressrQr?rWrXrYrZstart_http_serverrrZdistributed.dashboard.worker ImportErrorrrr r(r+Zhttp_applicationZ http_serveriprrrrrZstart_servicesZlistenerr!rrrrrzformatrrrrur rrrBrr.rrrrZstart_periodic_callbacks)rZportsrZ start_addressrrZroutes distributedrZlistening_addressrrZ plugins_msgsZplugins_exceptionsexcrrKrr;s            $      zWorker.start_unsafer worker-close)r#rrrrc sljtjtjtjfkr8tdj|IdHdSjtjkrHd}t zt dj |Wn t k rt d|YnXjt krt dj|st dtj_jD] }|qtjdIdHjD]8}z|IdHWqtk rtd YqXqjD].}t|d r(|}t|r(|IdH}q(|rjrj} | j|d IdHW5QRXtd fd djD} tj dd| DIdHj!r t"fddt#j$Ds t#j%D],} d| _&| j'r| IdHn| qj()IdHj*+,j-dkr^t.dIdH/ddij0rt1t2j0t3dIdHW5QRXj4D]} | t5j6krqfdd} t7r| | ddnLzt8| | |dIdHWn2t9k r*tj:d| dd| | |dYnXqjIdHtj_t;IdHtddS)a|Close the worker Close asynchronous operations running on the worker, stop all executors and comms. If requested, this also closes the nanny. Parameters ---------- timeout Timeout in seconds for shutting down individual instructions executor_wait If True, shut down executors synchronously, otherwise asynchronously nanny If True, close the nanny reason Reason for closing the worker Returns ------- str | None None if worker already in closing state or failed, "OK" otherwise z8Attempted to close worker that is already %s. Reason: %sNFz!Stopping worker at %s. Reason: %szStopping worker. Reason: %sz%Closed worker has not yet started: %sz Not waiting on executor to closer"zFailed to tear down preloadrrzdask worker [closing]cs g|]}t|dr|qS)teardown)rr%rrKrrr s z Worker.close..css|]}t|r|VqdSrr)rNtdrrrrszWorker.close..c3s$|]}|kr|jtkr|VqdSrrWORKER_ANY_RUNNINGrNwrKrrrs TZucxg?rHz close-stream)secondscs6t|tr&|jj|j|dn |j|ddS)N)waitr#)r,)rerIZ _work_queuequeuershutdownrr,r"rr_closeAs  zWorker.close.._closer/zKCould not close executor %r by dispatching to thread. Trying synchronously.rzdask worker [closed]r)tdjdj<fdd}j|j|dS)NZ1msr,cs<tfjIdH}d|_|ddiIdH|dS)NzWorker->WorkerrHZconnection_stream)r+rrrr)rr$Zbcommrrrbatched_send_connects z3Worker.send_to_worker..batched_send_connect)rr(r_ongoing_background_tasks call_soonr)rr$rr>rr=rsend_to_workers    zWorker.send_to_workerz dict | Statusc st}|dkrj}|r8|r8t|jtjkr8|d}jtjkrNd}d}nd}|dk rj|krt dj|j||ddiSjd7_j d7_ fd d |D}t |t |krt |t |D]>} | jjkrd d lm} | tjj| j| d || <qddd |Dd} fdd |D} t| } j| 7_j| 7_t}||dkrd||t}zrz<|j| |dIdH}|j|dIdH}|dkst|Wn0tk rt dj||YnXW5j| 8_jd8_Xt}d||||pDd}j |j!|j!||d||| | || |d tj"S)Nrrz= Throttling outgoing data transfers because worker is paused.rFzUWorker %s has too many open connections to respond to data request from %s (%d/%d).%srbusycs"i|]}|jkr|j|qSrrrNrrKrrrPs z#Worker.get_data..rActorr(rcSsi|]\}}|t|qSr)rDrrrrrPs)rr cs i|]}|jj|jpdqS)rrrrrCrKrrrPs{Gzt?zget-data-load-durationrrz$failed during get data with %s -> %szget-data-send-durationg?) rrrdurationrr7r compressedr)#r@rr,Z peer_addressr$rr5rrrrrrbrZractorsdistributed.actorrErrzsumrrrrrrr\rrabortrappendrZ dont_reply)rrr7rrmax_connectionsrZ throttle_msgr rrErZbytes_per_task total_bytesrrJrrIrrKrr1s        zWorker.get_datazdict[str, object])r rrrcCs8|t|||pdtddd|DddS)Nz update-data-)r rrcSsi|]\}}|t|qSr)sizeofrrrrrPsz&Worker.update_data..r)rr)rr|r@rz)rr rrrrrr2 szWorker.update_data)rrcsz|D]N\}}||jjkr>|jj|||jj|7<n ||jj|<||jj|<qt|jj|jj|jdIdHdS)N)rr()rzrravailable_resourcesr]r set_resourcesr)rrr8ZquantityrrrrTs   zWorker.set_resourceszWorkerPlugin | bytes)rrrrc st|trtdt|}|dkr*t|}|s2t||jkrN|j|dIdH||j|<t d|t |drz"|j |d}t |r|IdH}Wn>tk r}z |st|}td|WYSd}~XYnXddiS) Nr)rzStarting Worker plugin %ssetuprFrrr)rebytesrrCloadsr<r\rrrrrrrUrrr7)rrrrrrrrrrrq+s(      zWorker.plugin_add)rrc std|z8|j|}t|drF|j|d}t|rF|IdH}Wn8tk r}zt|}t d|WYSd}~XYnXddiS)NzRemoving Worker plugin r%rFrrr) rrrpoprr%rrr7r)rrrrrrrrrrrOs    zWorker.plugin_remove)rrrcCsDtj|}|tjkr:|jtkr:td|j|||n||_dS)Nz*Invalid Worker.status transition: %s -> %s) r5lookuprrr(rrrr)rrrZ new_statusrrrrs^s  z"Worker.handle_worker_status_changeztype[StateMachineEvent]zCallable[(Ellipsis, None)])clsrcs"fdd}djd|_|S)Ncsf|}|dSr)r)reventrZrrrrus z)Worker._handle_remote_stimulus.._z_handle_remote_stimulus())r)rrZrrr\rrmrszWorker._handle_remote_stimuluszasyncio.Task[StateMachineEvent])taskrcst|dS)zOverride BaseWorker method for added validation See also -------- distributed.worker_state_machine.BaseWorker._handle_stimulus_from_task N)r_handle_stimulus_from_task)rr^rrrr_|sz!Worker._handle_stimulus_from_taskrx)stimsrc s\ztj|WnFtk rV}z(t|drD|\}}|||W5d}~XYnXdS)zOverride BaseWorker method for added validation See also -------- distributed.worker_state_machine.BaseWorker.handle_stimulus distributed.worker_state_machine.WorkerState.handle_stimulus to_eventN)rrrrrar)rr`rrrrrrrs    zWorker.handle_stimulus)rrcCs@|jj|}|jdkt|j||jjkp2||jjk||jkdS)Nr)rwaiting_for_dataheapr )rrrrbr constrainedr )rrrrrrstateofs  zWorker.stateof Iterable[str]z list[tuple])keys_or_stimulircs |jj|Sr)rstory)rrgrrrr?szWorker.get_storyrz)r7rcCsBd}|D],}|jj|}|jr0tt|jS|}q|s>t|S)aFor diagnostics, we want to attach a transfer to a single task. This task is typically the next to be executed but since we're fetching tasks for potentially many dependents, an exact match is not possible. Additionally, if a key was fetched through acquire-replicas, dependents may not be known at all. Returns ------- The task to attach startstops of this transfer to N)rrZ dependentsnextiterr\)rr7causerrrrr _get_causes  zWorker._get_cause)rrr rkr(rc sRtfdd|D}|jd|j|j|d||pBd}||}j|j|j||dj|fdd|D|||d |d krjd |d _j|\} } | || d fj|<ttt | } t | d kr| \} j | \} } | || d fj | < d|| d|jdt |dS)Nc3s|]}jj|VqdSr)rrrrrKrrrsz7Worker._update_metrics_received_data..r)actionrrsource{Gz?g@csi|]}|jj|jqSrrGrrKrrrPsz8Worker._update_metrics_received_data..)rrrrIr7rrri@Brrrztransfer-bandwidthztransfer-durationztransfer-count)rM startstopsrOrrrrrZmaprrrbrrZcountersr) rrrr rkr(rQrIrZbwZcntrtyprrKr_update_metrics_received_datasB    z$Worker._update_metrics_received_datazCollection[str])r( to_gather total_nbytesrrc s|jtkr(tjtd||dtdSz|jjd|||tft dt ||t}t |j |||jdIdH}t}|dd kr|jjd |||tft||d tdWS|dd kst||}|j|||d ||d|jjd|t|d |tft|||d dtdWStk r~t d||jjd|||tft||dtdYStk r} zJt | |jrtrddl} | tj| ||dtdWYSd} ~ XYnXdS)zImplements BaseWorker abstract method See also -------- distributed.worker_state_machine.BaseWorker.gather_dep zWorker is shutting downworker-closing-)r(rurz request-depzRequest %d keys from %s)rNrrBz busy-gatherzgather-dep-busy-rr )rrr rkr(z receive-depzgather-dep-success-)r(rur rz+Worker stream died during communication: %szreceive-dep-failedzgather-dep-network-failure-rzgather-dep-failure-)rr(rofrom_exceptionr7r@rlogrOrrrbget_data_from_workerr9r$rnr\rlrsrZrqrrrprrLOG_PDBpdb set_trace) rr(rtrurrrrrkrr{rrr gather_deps           zWorker.gather_dep)r(rcs$tdIdHt|dtdS)zWait some time, then take a peer worker out of busy state. Implements BaseWorker abstract method. See Also -------- distributed.worker_state_machine.BaseWorker.retry_busy_worker_later g333333?Nzretry-busy-worker-)r(r)rrrvr@)rr(rrrretry_busy_worker_laterEs  zWorker.retry_busy_worker_later)rrrcCst|||dS)zBImplement BaseWorker.digest_metric by calling Server.digest_metricN)rAr)rrrrrrrRszWorker.digest_metriccCs0|tdtd|jdj|jd_dS)Nz find-missing-rrGrL)rrlr@rxrrKrrrryVszWorker.find_missingcCst||||||dSN)functionrrr,r/)rrrrr,rrrrr/csz Worker.runcCst||||||dSrr)rrrrrr,rrrr0fszWorker.run_coroutine)rrc s|pi}|dd}|}|jj|}t||}t|d|}zht|rZ|||IdH} n>|r|j|jdt ||||j ||j |j IdH} n |||} dt | dWStk r} zdt | dWYSd} ~ XYnXdS) Nseparate_threadTrr*rrrrrr)rXrrKgetattrr rRrrun_in_executorrapply_function_actorrrrrDr) rr*rrrrrr rrexrrrr9is2    zWorker.actor_executec Cs\z"t|jj||}dt|dWStk rV}zdt|dWYSd}~XYnXdS)Nrrrr)rrrKrDr)rr* attributerrrrrr:s zWorker.actor_attributez&tuple[Callable, tuple, dict[str, Any]])rrcs|jdk stt}t|jtkr@ttf|jIdH\}}}nt|j\}}}t}||dkrv|jd||d|||fS)NroZ deserializermrr) run_specr\r@rRr0rU _deserializerprO)rrrrrrrrrr_maybe_deserialize_tasks  zWorker._maybe_deserialize_task)rrrc s|jtkrt|dtdS|jj|}z||IdH\}}}WnNtk r}z0tj d|ddt j ||dtdWYSd}~XYnXz|jj r|j rt|jdkst||jdk st||||\}} |jdk st|jd d } z|j| } Wn0tk r6td | d t|jYnX|j|zt|_t|rpt||| |jIdH} n^d tt| kr|j !| t"||| |j#||j$|j%|j IdH} n|j !| t&||| |jIdH} W5|j|X| d|j'|<| ddkrX|j(dk r$|j(d| d| dt)|| d| d| d| d| ddtdWS| d} t*| t+rt|j,dtdWS|jt-j.krt*| t/j0rt|rt1d|dt|j,dtdWSt2d|tt3|ddt4|dd t5| dd | d!t j | || d| dd"td#WStk r}z0tj d$|ddt j ||d%tdWYSd}~XYnXdS)&zExecute a task. Implements BaseWorker abstract method. See also -------- distributed.worker_state_machine.BaseWorker.execute rv)rrNzCould not deserialize task %sTrzrun-spec-deserialize-failed-)rZ cancelledZresumedrrzInvalid executor z; expected one of: rIthreadrH task-finishedz task-durationrrrrrztask-finished-)rrrrrrractual-exceptionz reschedule-z Async task z- cancelled during worker close; rescheduling.zcancelled-by-worker-close-zUCompute Failed Key: %s Function: %s args: %s kwargs: %s Exception: %r rFmax_lenZexception_textz task-erred-)rrrrz&Exception during execution of task %s.zexecute-unknown-error-)6rr(rur@rrrrrrrjrwrrbr\r_prepare_args_for_executionrrYrKeyErrorrQsortedrrdiscardrrRapply_function_asyncrrrrrapply_functionrrrapply_function_simplerdigestsrkre Reschedulerr5rrZCancelledErrorrrrconvert_args_to_strconvert_kwargs_to_str)rrrrrrrr#args2kwargs2rrrZtask_excrrrexecutes                        zWorker.executetupleztuple[tuple, dict[str, Any]])rrrrc Cst}i}|jD]^}|j}z|j|||<Wqtk rlddlm}|t|jj ||j ||||<YqXqt ||t t fd} t ||t t fd} t} | |dkr|jd|| d|d| || | fS)NrrD)Z key_typesrHz disk-readrzdisk-load-duration)r@Z dependenciesrr rrLrErrrKr$r\rVrrprOr) rrrrrr deprrErrrrrrr4 s   ( z"Worker._prepare_args_for_executioncCsRt|j}|jt}|_|j||f|j|t|j f|j dSr) r@rrr&r[rrOrrrr)rrprofrrrr|Q s  zWorker.cycle_profilec s|js dSt}|j|j}W5QRXtfdd|Di}|jr^dd|D}D]`\}}|dk rft||}| |}t j |d|j dd}t |d|t j |d|j|ddqft} |d| |dS) zz Get a frame from all actively computing threads Merge these frames into existing profile counts Ncsi|]}||qSrrrNidentframesrrrPe sz*Worker.trigger_profile..cSsi|]}|t|qSr)r&Z ll_get_stackrrrrrPh sTzdistributed/worker.py)rzprofile-duration)rr@rrrk_current_framesrrzr rYr&processrZ llprocessrr) rrrZllframesrframerZllframerrrrrr{Y s:  zWorker.trigger_profile)rc s6t|j}|r|jjn$dkr*|jnfdd|jD|dkrLd}nt|f}|dkrhd}n"t|fd}|t krd}|dkr|dkrt n,|dkrt n|}fddt ||Dtj t d} stS|dkr2|dks ||kr2dkr|j} n |j} t | | } | S)Ncs$g|]\}}|kr||fqSrrrNtd)rrrr sz&Worker.get_profile..rrcsg|] }|qSrrrNi)historyrrr s)r@rrgr&rrbisect bisect_left bisect_rightrbrrangemergerr[rr) rrrrrristartistopZiistoprrr)rrrroy s6       zWorker.get_profile)rrrcsdk}t|j}p|fdd|jDfdd|jDd}|r|d||jdf|d|dd |jDf|S) Ncs4g|],\}}|kr krnq||dfqSrcrrrrrrr s  z/Worker.get_profile_metadata..cs>g|]6\}}|kr krnq|dd|DfqS)cSsi|]\}}||dqSrr)rNrrrrrrP sz:Worker.get_profile_metadata...)rz)rNrrrrrr s  )countsr7rrcr7cSsi|]\}}||dqSrrrrrrrP sz/Worker.get_profile_metadata..)r@rrrrOrrrz)rrrZ add_recentrrrrrrp s     zWorker.get_profile_metadatazCollection[str] | Nonec s^|j&tfdd|jDW5QRX|dk rLfdd|DddDS)Ncsi|]\}}||qSrr)rNtidr) sys_framesrrrP sz)Worker.get_call_stack..csi|]}|kr||qSrrrrrrrP scSsi|]\}}|t|qSr)r&r5)rNrrrrrrP s)rrkrrrz)rr7r)rrrrn s "zWorker.get_call_stackcs|j|jt|jIdHSr)rrrr<rrKrrrr< s zWorker.benchmark_diskcs|j|jtIdHSr)rrrr=rKrrrr= szWorker.benchmark_memory)r$rcst|j|dIdHS)N)r9r$)r>r9)rr$rrrr> szWorker.benchmark_networkrc CsB|j2|jr |jW5QRS|W5QRSW5QRXdSr)rr _get_clientrKrrrclient sz Worker.client)r#rc Cs|dkrtjd}t|d}zddlm}|}Wntk rHYn`Xddlm}|j rl|j j |j j kst |j t r|j |j j kst |j |r|j j|j j kr||_|js ddlm}t|j}||j |j|jd|dd |d |_tj|j|s |jjd ks t|jS) zGet local client attached to this worker If no such client exists, create one See Also -------- get_client N!distributed.comm.timeouts.connectsr)default_client)ClusterrTr()rrZset_as_defaultr5Zdirect_to_workersrr#r)rWrXrYr"distributed.clientrrQZdistributed.deploy.clusterrrr$reZ _start_argrZscheduler_addressrrrPrrrrrrr\)rr#rrrrr5rrrr sN              zWorker._get_clientcCs|jtS)aGet the key of the task we are currently running This only makes sense to run within a task Examples -------- >>> from dask.distributed import get_worker >>> def f(): ... return get_worker().get_current_task() >>> future = client.submit(f) # doctest: +SKIP >>> future.result() # doctest: +SKIP 'f-1234' See Also -------- get_worker )rrUrrKrrrget_current_task szWorker.get_current_taskc Csz|jWnrtk r}zTtjd|dt|trLddl}|t |drn| \}}| ||W5d}~XYnXdS)NzValidate state failedrrra) rvalidate_staterrrrrzr{r|rrar)rrr{rrrrrr0 s    zWorker.validate_statecCstjdtdd|jS)Nz_The `Worker.incoming_transfer_log` attribute has been renamed to `Worker.transfer_incoming_log`rr)rRrSrTrrKrrrincoming_transfer_logA s zWorker.incoming_transfer_logcCstjdtdd|jS)Nz`The `Worker.outgoing_count` attribute has been renamed to `Worker.transfer_outgoing_count_total`rr)rRrSrTrrKrrrrK s zWorker.outgoing_countcCstjdtdd|jS)NzbThe `Worker.outgoing_current_count` attribute has been renamed to `Worker.transfer_outgoing_count`rr)rRrSrTrrKrrroutgoing_current_countU s zWorker.outgoing_current_countcCstjdtdd|jS)Nz_The `Worker.outgoing_transfer_log` attribute has been renamed to `Worker.transfer_outgoing_log`rr)rRrSrTrrKrrroutgoing_transfer_log_ s zWorker.outgoing_transfer_logcCstjdtdd|jS)NzfThe `Worker.total_in_connections` attribute has been renamed to `Worker.transfer_outgoing_count_limit`rr)rRrSrTrrKrrrtotal_in_connectionsi s zWorker.total_in_connections)NN)T)Fr)rTTr$)Nr:)NNNN)TN)NT)rTN)rNT)NNrN)NN)NNNF)rN)N)N)~r __module__ __qualname____doc__rWeakSetr__annotations__rrrrrrtpropertyr rar rrrrbZmemory_monitorrirKrSZ busy_workersZ comm_nbytesZcomm_threshold_bytesrdZdata_needed_per_workerZexecuted_countrZ generationZhas_whatrZin_flight_tasksZin_flight_workersrxZ long_runningrZ stimulus_logZstimulus_storyrhrrZtarget_message_sizeZtotal_out_connectionsrZtransition_counterr rZ validate_taskrrrrrrrrArsetterrrrrrrJrrrGrrr4r7r.r;rrTrr~r<rAr1r2rTrqrrrsrmr_rrer?rlrsr}r~rryr/r0r9r:rrrr|r{rorprnr<r=r>rrrrrrrrr __classcell__rrrrrs"    d          3  S/ w5 r#    0 Z  # "0 <    rrc Cs^z tjdWStk rXztddtjDWYStk rRtdYnXYnXdS)aGet the worker currently running this task Examples -------- >>> def f(): ... worker = get_worker() # The worker on which this task is running ... return worker.address >>> future = client.submit(f) # doctest: +SKIP >>> future.result() # doctest: +SKIP 'tcp://127.0.0.1:47373' See Also -------- get_client worker_client r(css|]}|jtkr|VqdSrr'r)rrrr s zget_worker..zNo workers foundN)rYrrrrr StopIterationrQrrrr get_workert s rTrcCs|dkrtjd}t|d}|r.|r.t|}z t}Wntk rLYnX|r^|jj|krj|j |dSddl m }z | }Wntk rd}YnX|r|r|jj|kr|S|r|||dStddS)aGet a client while within a task. This client connects to the same scheduler to which the worker is connected Parameters ---------- address : str, optional The address of the scheduler to connect to. Defaults to the scheduler the worker is connected to. timeout : int or str Timeout (in seconds) for getting the Client. Defaults to the ``distributed.comm.timeouts.connect`` configuration value. resolve_address : bool, default True Whether to resolve `address` to its canonical form. Returns ------- Client Examples -------- >>> def f(): ... client = get_client(timeout="10s") ... futures = client.map(lambda x: x + 1, range(10)) # spawn many tasks ... results = client.gather(futures) ... return sum(results) >>> future = client.submit(f) # doctest: +SKIP >>> future.result() # doctest: +SKIP 55 See Also -------- get_worker worker_client secede Nrrr"rrz.No global client found and no address provided) rWrXrYr"comm_resolve_addressrrQrr$rrrrf)r$r#r.r(rrrrr get_client s*'        rcCs@t}tttj}|j|jttj |dtddS)a Have this task secede from the worker's thread pool This opens up a new scheduling slot and a new thread for a new task. This enables the client to schedule tasks on this node, which is especially useful while waiting for other jobs to finish (e.g., with ``client.gather``). Examples -------- >>> def mytask(x): ... # do some work ... client = get_client() ... futures = client.map(...) # do some remote work ... secede() # while that work happens, remove ourself from the pool ... return client.gather(futures) # return gathered results See Also -------- get_client get_worker zsecede-)rZcompute_durationrN) r tpe_seceder@rYrrrrrwr)r(rIrrrrJ s  rJc@seZdZdZdS)raReschedule this task Raising this exception will stop the current execution of the task and ask the scheduler to reschedule this task, possibly on a different machine. This does not guarantee that the task will move onto a different machine. The scheduler will proceed through its normal heuristics to determine the optimal machine to accept this task. The machine will likely change if the load across the cluster has significantly changed since first scheduling the task. N)rrrrrrrrr srcsFdkrjdkrjfdd}t|ddIdHS)aGet keys from worker The worker has a two step handshake to acknowledge when data has been fully delivered. This function implements that handshake. See Also -------- Worker.get_data Worker.gather_dep utils_comm.gather_data_from_workers Nc sIdH}d|_zht|ddIdH}z |d}Wntk r^td|YnX|dkrx|dIdH|WS|XdS)Nz#Ephemeral Worker->Worker for gatherr1)rrrHr7rrPrzUnexpected responser)r+rZreuser:rrQr)rrrrr7rPr9rrr(rr _get_data s(  z'get_data_from_worker.._get_datary)Z operation)rrr])r9r7r(rrPrrrrrrry s ryd)maxsizecCsNt|dkrDz t|}Wn&tk r>t|}|t|<YnX|St|S)z'Load a function from bytes, cache bytesr)rbr rrCrW)Z bytes_objectrrrrloads_function? s   rcCst|dk rt|}|r(t|tr(t|}|r@t|tr@t|}|tk rb|sT|sT|rXtt}|f}||pjd|ppifS)z>> inc = lambda x: x + 1 >>> execute_task((inc, 1)) 2 >>> execute_task((sum, [1, 2, (inc, 3)])) 7 rrN)rrqrrer)r^r rrrrr\ s  rrVc Csztt|}W5QRXWn^tk r^t|}t|dkrZt|t|<W5QRXYntk rzt|}YnX|S)z)Dump a function to bytes, cache functionsr) _cache_lock cache_dumpsrrCdumpsrb TypeError)r rrrrdumps_functions s  rcCst|r|dtkrdttt|ddsdt|dt|dd}t|dkr`t|d|d<|Sttt|ddst|dt|dddSt|S) uKSerialize a dask task Returns a dict of bytestrings that can each be loaded with ``loads`` Examples -------- Either returns a task as a function, args, kwargs dict >>> from operator import add >>> dumps_task((add, 1)) # doctest: +SKIP {'function': b'€•Œ _operator”Œadd”“”.' 'args': b'€•KK†”.'} Or as a single task blob if it can't easily decompose the result. This happens either if the task is highly nested, or if it isn't a task at all >>> dumps_task(1) # doctest: +SKIP {'task': b'€•K.'} rrNr)rrr) rrr4rqrLr warn_dumpsrbrD)r^rrrr dumps_task s" rFg.AcCsr||}tdsnt||krndtd<t|}t|dkrT|ddd|dd}tdtt||f|S) z6Dump an object to bytes, warn if those bytes are largerTFN2z ... iaNLarge object of size %s detected in task graph: %s Consider scattering large objects ahead of time with client.scatter to reduce scheduler burden and keep data on workers future = client.submit(func, big_data) # bad big_future = client.scatter(big_data) # good future = client.submit(func, big_future) # good)_warn_dumps_warnedrbrrRrSr)objrlimitbrrrrr s  rc Cs\t}||||<W5QRXtt_|t_|t_t||||} | ||=W5QRX| S)Run a function, collect information Returns ------- msg: dictionary with status, result/error, timings, etc.. )rUrr@rYrrrr) rrrrrrr time_delayrrrrrr src Cst}t}zz|||}WnRttfk r8Yn^tk rp}zt|} d| d<|| d<W5d}~XYn&Xdd|t||dk rt|ndd} W5t}X||| d<||| d <|| d <| S) r task-erredrHrNrrrHrrrrrrr rUrr@rrrr7rRr rrrrrrrrrrrrrr s*    rc st}t}zz|||IdH}WnRttfk r>Yn^tk rv}zt|} d| d<|| d<W5d}~XYn&Xdd|t||dk rt|ndd} W5t}X||| d<||| d <|| d <| S) rNrrHrrrrrrrrrrrrr s*    rc CsVt}||||<W5QRX|t_|t_dt_|||}| ||=W5QRX|S)rT)rUrrYrrr*) rrrrrrrrrrrrr8 s  rcCsNGddd}|}d|kr0|t|d|d<d|krJ|t|d|d<|S)zMake a worker msg, which contains args and kwargs, safe to cast to str: allowing for some arguments to raise exceptions during conversion and ignoring them. c@seZdZddZddZdS)zget_msg_safe_str..ReprcSs||_||_dSr_fZ_val)rrvalrrrrtY sz'get_msg_safe_str..Repr.__init__cSs ||jSrrrKrrrr] sz'get_msg_safe_str..Repr.__repr__N)rrrrtrrrrrReprX srrr)rrr)rrrrrget_msg_safe_strR srr)rrc Csd}ddtt|D}t|D]~\}}z t|}Wntk rNd}YnX|||<|t|d7}|dk r"||kr"dd|d|d d|Sq"d d|S) zwConvert args to a string, allowing for some arguments to raise exceptions during conversion and ignoring them. rcSsg|]}dqSrrrrrrrm sz'convert_args_to_str.. < could not convert arg to str >rNz({}, rz({}))rrb enumeraterrr!r`)rrlengthstrsrargsargrrrrh s  *rr)rrrc Csd}ddtt|D}t|D]\}\}}z t|}Wntk rVd}YnXt|d|}|||<|t|d7}|dk r&||kr&dd |d|d d|Sq&d d |S) zyConvert kwargs to a string, allowing for some arguments to raise exceptions during conversion and ignoring them. rcSsg|]}dqSrrrrrrr sz)convert_kwargs_to_str..rz: rNz{{{}rrz{{{}}})rrbrrzrrr!r`) rrrrrargnamerrZskwargrrrr{ s  *rrc s0|pi}t|}t|}|s*|s*td|r8t|}|rFt|}t|drX||d<t|drj||d<tdt|zB|s|||}n.|r|||IdH}n|jj |f||d}Wn`t k r}z@tj dt t|ddt |ddt|dddd t|} W5d}~XYnXd t|d } | S) NzCombination not supportedZ dask_workerZdask_schedulerzRun out-of-band function %rz2Run Failed Function: %s args: %s kwargs: %s rFrTrrr)rCrWrRr\rNrrrr?r@rrrrrr7rD) rrrrrr,Zis_cororrrrrrr/ s>        r/cCs$dd}|td<dd}|td<dS)NcsttjIdH}|Sr)rUr;Z real_time)r(rrrr gpu_metric sz#add_gpu_metrics..gpu_metricr+cSstSr)r;Zone_timerFrrr gpu_startup sz$add_gpu_metrics..gpu_startup)rr)rrrrradd_gpu_metrics srr seprfilerrz TextIO | Nonerr)rrrrrcGsz t}Wntk rYndXttt||||d}|tjkrLd|d<n*|tjkr`d|d<n|dk rvtd|| d|t j |||||ddS) a A drop-in replacement of the built-in ``print`` function for remote printing from workers to clients. If called from outside a dask worker, its arguments are passed directly to ``builtins.print()``. If called by code running on a worker, then in addition to printing locally, any clients connected (possibly remotely) to the scheduler managing this worker will receive an event instructing them to print the same output to their own standard output or standard error streams. For example, the user can perform simple debugging of remote computations by including calls to this ``print`` function in the submitted code and inspecting the output in a local Jupyter notebook or interpreter session. All arguments behave the same as those of ``builtins.print()``, with the exception that the ``file`` keyword argument, if specified, must either be ``sys.stdout`` or ``sys.stderr``; arbitrary file-like objects are not allowed. All non-keyword arguments are converted to strings using ``str()`` and written to the stream, separated by ``sep`` and followed by ``end``. Both ``sep`` and ``end`` must be strings; they can also be ``None``, which means to use the default values. If no objects are given, ``print()`` will just write ``end``. Parameters ---------- sep : str, optional String inserted between values, default a space. end : str, optional String appended after the last value, default a newline. file : ``sys.stdout`` or ``sys.stderr``, optional Defaults to the current sys.stdout. flush : bool, default False Whether to forcibly flush the stream. Examples -------- >>> from dask.distributed import Client, print >>> client = distributed.Client(...) >>> def worker_function(): ... print("Hello from worker!") >>> client.submit(worker_function) Hello from worker! )rrrrrrrNz|Remote printing to arbitrary file objects is not supported. file kwarg must be one of None, sys.stdout, or sys.stderr; got: printr) rrQrrqrrkstdoutstderrrrbuiltinsr)rrrrrr(rrrrr s&3       rrz str | Warningztype[Warning] | Nonerr)rcategoryrrnrcCsVz t}Wntk rYn X|dt|t|dt|||d|dS)a A drop-in replacement of the built-in ``warnings.warn()`` function for issuing warnings remotely from workers to clients. If called from outside a dask worker, its arguments are passed directly to ``warnings.warn()``. If called by code running on a worker, then in addition to emitting a warning locally, any clients connected (possibly remotely) to the scheduler managing this worker will receive an event instructing them to emit the same warning (subject to their own local filters, etc.). When implementing computations that may run on a worker, the user can call this ``warn`` function to ensure that any remote client sessions will see their warnings, for example in a Jupyter output cell. While all of the arguments are respected by the locally emitted warning (with same meanings as in ``warnings.warn()``), ``stacklevel`` and ``source`` are ignored by clients because they would not be meaningful in the client's thread. Examples -------- >>> from dask.distributed import Client, warn >>> client = Client() >>> def do_warn(): ... warn("A warning from a worker.") >>> client.submit(do_warn).result() /path/to/distributed/client.py:678: UserWarning: A warning from a worker. rS)rr rN)rrQrrCrrRrS)rr rrnr(rrrrS s! rS1 kiB100 kiB1 MiB10 MiBz100 MiB1 srfr)rootdirsizesrc Cst|}i}|D]}t|d}t|}ttttd}t|}t |}t } d} t | |krt |t |dd&} | || t| W5QRX| |7} qV| t | ||<W5QRXq|S)z~ Benchmark disk bandwidth Returns ------- out: dict Maps sizes of outputs to measured bandwidths )dirrrab)mode)r"r#pathlibPathrrqrrr!r2r@rr}choicerrrrr) rrrIrsize_strrnamessizer rrrrrrr<Q s$     r<z2 kiB10 kiBrrr200 ms)rrc Cslt|}i}|D]V}t|}t|}t}d}t||krT|dd}~||7}q.|t|||<q|S)z Benchmark memory bandwidth Returns ------- out: dict Maps sizes of outputs to measured bandwidths rN)r"r!r2r@) rrIrrrr rrrrrrr=v s   r=r rrrrz50 MiBz)ConnectionPool | Callable[[str], RPCType])r$r9rrc st|}i}||4IdHn}|D]b}t|}tt|}t} d} t| |krr|j|dIdH| |d7} qD| t| ||<q"W5QIdHRX|S)z Benchmark network communications to another worker Returns ------- out: dict Maps sizes of outputs to measured bandwidths Nrrr)r"r!rDr2r@Zecho) r$r9rrIrr8rrr rrrrrr> s $r>)NNT)NNNN)N)N)rNT)Nr r)rr)r!r) __future__rrrr rrloggingrvrrr}rkr]rUrRr collectionsrrcollections.abcrrrrr r concurrent.futuresr contextlibr datetimer inspectrtypingrrrrrrrZtlzrrrZtornado.iolooprrWZ dask.corerZ dask.systemrZ dask.utilsrrrr r!r"r#r$r"r%r&r'Zdistributed.batchedr(Zdistributed.collectionsr)Zdistributed.commr*r+r,r-r.rZdistributed.comm.addressingr/Zdistributed.comm.utilsr0Zdistributed.compatibilityr1r2r3Zdistributed.corer4r5r6r7r8r9ZRPCTyper:Zdistributed.diagnosticsr;Zdistributed.diagnostics.pluginr<Zdistributed.diskutilsr=r>Zdistributed.httpr?Zdistributed.metricsr@Zdistributed.noderAZdistributed.proctitlerBZdistributed.protocolrCrDZdistributed.pubsubrEZdistributed.securityrFZdistributed.shufflerGZdistributed.sizeofrHrRZdistributed.threadpoolexecutorrIrJrZdistributed.utilsrKrLrMrNrOrPrQrRrSrTrUrVrWrXrYrZZdistributed.utils_commr[r\r]Zdistributed.utils_perfr^r_Zdistributed.versionsr`Zdistributed.worker_memoryrarbrcZ distributed.worker_state_machinerdrerfrgrhrirjrkrlrmrnrorprqrrrsrtrurvrwrxryrzr{r|r}Ztyping_extensionsr~rrrZdistributed.nannyrrr getLoggerrrrXrYrzrrrrrrrr(rrrrrrrryZ job_counterr rrrrrVrrrrrrrrrrrrrr/rZ_global_workersrr UserWarningrSr<r=r>rrrrs*      $   (                    H p        " D$ 4   /1 'R<&