B
    836\q                @   s  d dl mZmZmZ d dlZd dlmZmZ d dlm	Z	 d dl
Z
d dlZd dlZd dlmZ d dlZd dlZd dlZd dlZd dlZd dlZd dlmZ d dlmZ yd dlmZmZmZ W n( ek
r   d dlmZmZmZ Y nX d d	lm Z  d d
l!m"Z" d dl#m$Z$ d dl%m&Z& ddl'm(Z(m)Z) ddl*m+Z+ ddl)m,Z,m-Z-m.Z. ddl/m0Z0 ddl1m2Z2m3Z3m4Z4 ddl5m6Z6m7Z7m8Z8m9Z9m:Z: ddl;m<Z< ddl=m>Z> ddl?m@Z@ ddlAmBZB ddlCmDZD ddlEmZmFZFmGZGmHZH ddlImJZJ ddlKmLZL ddlMmNZM ddlOmPZPmQZR ddlSmTZTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[m\Z\m]Z]m^Z^m_Z_m`Z`maZambZbmcZcmdZdmeZe ddlfmgZgmhZh dd limjZjmkZkmlZl eZm ZneoepZqejrsd!Ztd"Zuyd dlvZvevw jxZyW n( ek
r   eqzd# d$ZydZvY nX d%Z{d&Z|d'Z}d(Z~eJgZg ZG d)d* d*e@Zd+d, ZdTd/d0Zd1d2 ZQG d3d4 d4eZd5d6 Ze"jdUd7d8Zd gZdVd9d:Zd;d< Ze Zd=d> Zd?d@ ZdAgZejdBfdCdDZdEdF ZdGdH ZdIdJ ZdWdKdLZdXdMdNZdOdP Ze"jdQi dd.fdRdSZdS )Y    )print_functiondivisionabsolute_importN)defaultdictdeque)	timedelta)PicklingError)istask)apply)pluckpartialmerge)Return)gen)IOLoop)Event   )profilecomm)BatchedSend)get_address_hostget_local_address_forconnect)offload)unicodeget_thread_identityfinalize)error_messageCommClosedError	send_recvpingpongcoerce_to_address)	WorkSpace)time)
ServerNode)preload_modules)setproctitle)pickleto_serializedeserialize_bytesserialize_bytelist)PubSubWorkerExtension)Security)safe_sizeof)ThreadPoolExecutorsecede)funcnameget_iphas_arg_maybe_complex
log_errorsignoring
mp_contextimport_filesilence_loggingthread_statejson_load_robust	key_splitformat_bytesDequeHandlerPeriodicCallbackparse_bytesparse_timedeltaiscoroutinefunction)	pack_datagather_from_workers)ThrottledGCenable_gc_diagnosisdisable_gc_diagnosiszdistributed.admin.pdb-on-errz--no-value-sentinel--z3Please install psutil to estimate worker memory useg    eA)waitingready	executingzlong-running)rG   rH   constrained)rG   rH   rJ   rI   zlong-running)rH   rJ   c                   s  e Zd ZdZd fdd	Zd	d
 Zdd Zedd Zdd Z	dddZ
ejdd Zejdd Zejdd Zdd ZejdddZdddZejddd Zdd!d"Zdd$d%Zejdd'd(Zdd)d*Zejdd,d-Zd.d/ Zd0d1 Zejdd2d3Zejd4d5 Zd6d7 Zejdd8d9Zdd:d;Zejdd<d=Z ejd>d? Z!ddAdBZ"dCdD Z#ddEdFZ$ddGdHZ%ddIdJZ&ddKdLZ'dMdN Z(dOdP Z)ddQdRZ*dSdT Z+ddUdVZ,dWdX Z-e.dfdYdZZ/dd[d\Z0dd]d^Z1d_d` Z2dadb Z3dcdd Z4dedf Z5ddgdhZ6didj Z7ejddkdlZ8dmdn Z9ejdodp Z:ejdqdr Z;dsdt Z<dudv Z=ddwdxZ>ddydzZ?d{d| Z@ejdd~dZAdddZBdddZCejdddd}i fddZDdddZEdd ZFdd ZGejdddZHejdd ZIdd ZJdd ZKdddZLdddZMdddZNdd ZOdd ZPdd ZQdd ZRdd ZSdd ZTdd ZUdd ZVdd ZWdd ZXedd ZYdddZZdd Z[  Z\S )Workera   Worker 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 the the worker uses
    to manage and track internal computations.

    **State**

    **Informational State**

    These attributes don't change significantly during execution.

    * **ncores:** ``int``:
        Number of cores used by this worker process
    * **executor:** ``concurrent.futures.ThreadPoolExecutor``:
        Executor used to perform computation
    * **local_dir:** ``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}``:
    * **total_out_connections**: ``int``
        The maximum number of concurrent outgoing requests for data
    * **total_in_connections**: ``int``
        The maximum number of concurrent incoming requests for data
    * **total_comm_nbytes**: ``int``
    * **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**

    This 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.

    * **data:** ``{key: object}``:
        Dictionary mapping keys to actual values
    * **task_state**: ``{key: string}``:
        The state of all tasks that the scheduler has asked us to compute.
        Valid states include waiting, constrained, executing, memory, erred
    * **tasks**: ``{key: dict}``
        The function, args, kwargs of a task.  We run this when appropriate
    * **dependencies**: ``{key: {deps}}``
        The data needed by this key to run
    * **dependents**: ``{dep: {keys}}``
        The keys that use this dependency
    * **data_needed**: deque(keys)
        The keys whose data we still lack, arranged in a deque
    * **waiting_for_data**: ``{kep: {deps}}``
        A dynamic verion of dependencies.  All dependencies that we still don't
        have for a particular key.
    * **ready**: [keys]
        Keys that are ready to run.  Stored in a LIFO stack
    * **constrained**: [keys]
        Keys for which we have the data to run, but are waiting on abstract
        resources like GPUs.  Stored in a FIFO deque
    * **executing**: {keys}
        Keys that are currently executing
    * **executed_count**: int
        A number of tasks that this worker has run in its lifetime
    * **long_running**: {keys}
        A set of keys of tasks that are running and have started their own
        long-running clients.

    * **dep_state**: ``{dep: string}``:
        The state of all dependencies required by our tasks
        Valid states include waiting, flight, and memory
    * **who_has**: ``{dep: {worker}}``
        Workers that we believe have this data
    * **has_what**: ``{worker: {deps}}``
        The data that we care about that we think a worker has
    * **pending_data_per_worker**: ``{worker: [dep]}``
        The data on each worker that we still want, prioritized as a deque
    * **in_flight_tasks**: ``{task: worker}``
        All dependencies that are coming to us in current peer-to-peer
        connections and the workers from which they are coming.
    * **in_flight_workers**: ``{worker: {task}}``
        The workers from which we are currently gathering data and the
        dependencies we expect from those connections
    * **comm_bytes**: ``int``
        The total number of bytes in flight
    * **suspicious_deps**: ``{dep: int}``
        The number of times a dependency has not been where we expected it

    * **nbytes**: ``{key: int}``
        The size of a particular piece of data
    * **types**: ``{key: type}``
        The type of a particular piece of data
    * **threads**: ``{key: int}``
        The ID of the thread on which the task ran
    * **active_threads**: ``{int: key}``
        The keys currently running on active threads
    * **exceptions**: ``{key: exception}``
        The exception caused by running a task if it erred
    * **tracebacks**: ``{key: traceback}``
        The exception caused by running a task if it erred
    * **startstops**: ``{key: [(str, float, float)]}``
        Log of transfer, load, and compute times for a task

    * **priorities**: ``{key: tuple}``
        The priority of a key given by the scheduler.  Determines run order.
    * **durations**: ``{key: float}``
        Expected duration of a task
    * **resource_restrictions**: ``{key: {str: number}}``
        Abstract resources required to run a task

    Parameters
    ----------
    scheduler_ip: str
    scheduler_port: int
    ip: str, optional
    ncores: int, optional
    loop: tornado.ioloop.IOLoop
    local_dir: 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' for 60% of memory use.
        Use strings or numbers like 5GB or 5e9
    memory_target_fraction: float
        Fraction of memory to try to stay beneath
    memory_spill_fraction: float
        Fraction of memory at which we start spilling to disk
    memory_pause_fraction: float
        Fraction of memory at which we stop running new tasks
    executor: concurrent.futures.Executor
    resources: dict
        Resources that thiw worker has like ``{'GPU': 2}``

    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
    Ndask-worker-spaceTauto200msc       %         sB  t  | _t  | _t  | _t  | _t  | _t  | _t  | _tt	| _
tt| _t | _t | _t  | _t  | _tjd| _tjd| _d| _d| _tdd | _t	 | _t  | _t  | _t  | _t  | _t  | _ t | _!t  | _"tt#j$| _%tdd| _&t#$ | _'tdd| _(t  | _)d| _*t  | _+tt,| _-t  | _.t, | _/t | _0t	 | _1d| _2t	 | _3d | _4ttjd	d| _5d
| _6tdd| _7|8dd| _9| j:| j;| j;| j<| j=| j>| j?| j?| j?| j@| j?| j?| j?d| _A| jB| jC| jD| jEd| _Ftdd| _Gd| _Htdd| _Id| _Jd| _Kd| _Ld | _M|8dtjd}tN|dd}| O  |rftP|}|d }nH|d krtjdd rtjd}n |d krtQ|}ntQ||f}d| _R|ptS| _T|pi | _U|pi V | _W|| _X|| _Y| jYd krtjd| _Y|| _Z| jZd kr"tjd| _Z|| _[tN|dd| _\t  | _]|rNt^|d t_t`jab|| _c| jcjddd| _e| jejf| _g|pth | _itj| jithstk| jild| _m| jind| _otp|| jT| _qd| _rd|kr|8d| _sntjd| _sd|kr|8d| _tntjd| _td |kr.|8d | _untjd!| _u| jqr| jssT| jtrydd"lvmw}mx}my} W n tzk
r   tzd#Y nX t`ja{| jgd$}|t|t}d%d&t~||}tt| jq| js } |i || t| _nt  | _i | _|pt | _d | _t | _|
| _|pt| jTd'd(| _td)d*d(| _|	| _d| _t  | _d| _d | _| jgtjakrntjad| jg i | _|p|i | _|pi | _|rt |ni | _| j| j| j| j| j| j| jt| j| j| j| j| j| j| j| j| j| jd+}!| j| jt|| jdd,| j| jd-}"tt| jf |!|"| j| jmd.| | || _| jj| j| d/| _t| jd0| jd1}#|#| jd2< || _| jqrd| _t| j| j\d0 | jd1}#|#| jd3< |d krt}x|D ]}$|$|  qW ttd4| _td5 t| jtNtjd6ddd0 | jd1}#|#| jd7< t| j|d0 | jd1}#|#| jd8< tĠtƠ|  d S )9Nz'distributed.worker.connections.outgoingz'distributed.worker.connections.incomingg    cAr   c               S   s   dS )Nr    rO   rO   rO   1lib/python3.7/site-packages/distributed/worker.py<lambda>  s    z!Worker.__init__.<locals>.<lambda>i  )maxlenz+distributed.comm.recent-messages-log-lengthg    ׇAi validateF))rG   rH   )rG   memory)rG   error)rH   rI   )rH   rT   )rJ   rI   )rI   rT   )rI   rU   )rI   rescheduled)rI   zlong-running)zlong-runningrU   )zlong-runningrT   )zlong-runningrV   ))rG   flight)rG   rT   )rW   rG   )rW   rT   profile_cycle_intervalz distributed.worker.profile.cycleZms)defaultaddresszscheduler-addresszdistributed.worker.preloadzdistributed.worker.preload-argv)levelzworker-)prefixworkermemory_target_fractionz distributed.worker.memory.targetmemory_spill_fractionzdistributed.worker.memory.spillmemory_pause_fractionzdistributed.worker.memory.pause)BufferFileFuncz3Please `pip install zict` for spill-to-disk workersstorageraise)Zon_errorzDask-Worker-Threads')Zthread_name_prefixr   zDask-Actor-Threads)gatherrunrun_coroutineget_dataupdate_datadelete_data	terminateZpingupload_filestart_ipython
call_stackr   Zprofile_metadataget_logskeysversionsactor_executeactor_attribute)report)closezcompute-taskzrelease-taskzdelete-datazsteal-request)handlersstream_handlersio_loopconnection_args)	schedulerZioloopr]   i  )ry   	heartbeatrT   )loggerzdask-worker [not started]z#distributed.worker.profile.intervalr   zprofile-cycle)dicttasks
task_state	dep_statedependencies
dependentswaiting_for_datawho_hasr   sethas_whatr   pending_data_per_worker	threadingZLock_lockdata_neededin_flight_tasksin_flight_workersdaskconfiggettotal_out_connectionstotal_in_connectionstotal_comm_nbytescomm_nbytessuspicious_deps_missing_dep_flightnbytestypesthreads
exceptions
tracebacksactive_threads_lockactive_threadsr   createprofile_keysprofile_keys_historyprofile_recentprofile_history
priorities
generation	durationslist
startstopsresource_restrictionsrH   rJ   rI   executed_countlong_runningbatched_streamZrecent_messages_logtarget_message_sizelogpoprS   transition_waiting_readytransition_waiting_donetransition_ready_executingtransition_ready_memory transition_constrained_executingtransition_executing_done!transition_executing_long_running_transitionstransition_dep_waiting_flighttransition_dep_waiting_memorytransition_dep_flight_waitingtransition_dep_flight_memory_dep_transitionsincoming_transfer_logincoming_countoutgoing_transfer_logoutgoing_countoutgoing_current_countrepetitively_busy_clientr@   _setup_loggingr:   r!   Z_port_ncoresncorestotal_resourcescopyavailable_resourcesdeath_timeoutpreloadpreload_argvcontact_addressmemory_monitor_interval
extensionsr8   r"   ospathabspathZ
_workspaceZnew_work_dir_workdirZdir_path	local_dirr,   security
isinstanceAssertionErrorZget_connection_argsrz   Zget_listen_argslisten_argsparse_memory_limitmemory_limitpausedr^   r_   r`   Zzictra   rb   rc   ImportErrorjoinr   r*   r)   intfloatweightdataactorsr   currentloopstatusr   _closed	reconnectr.   executoractor_executornamescheduler_delaystream_commsheartbeat_active_ipython_kernelsysinsertservicesservice_portsservice_specsmetricsrf   rg   rh   ri   rj   rk   rl   r    rm   rn   get_call_stackget_profileget_profile_metadatarp   rq   rr   rs   rt   _closeadd_taskrelease_keysteal_requestsuperrK   __init__rpcr{   rZ   execution_stater>   r|   ry   periodic_callbacksZ_address_memory_monitoringmemory_monitorDEFAULT_EXTENSIONSrD   r}   _throttled_gcr&   trigger_profilecycle_profile_global_workersappendweakrefref)%selfZscheduler_ipZscheduler_portZscheduler_filer   r   r   r   r   r   r   r   r   	resourcesZsilence_logsr   r   r   r   r   r   r   r   kwargsrX   ZcfgZscheduler_addrra   rb   rc   r   rd   targetrw   rx   pcZext)	__class__rO   rP   r      s   


























zWorker.__init__c             C   sB   d| j j| j| jt| jt| j| jt| jt| j	t| j
f	 S )NzJ<%s: %s, %s, stored: %d, running: %d/%d, ready: %d, comm: %d, waiting: %d>)r  __name__rZ   r   lenr   rI   r   rH   r   r   )r
  rO   rO   rP   __repr__
  s
    zWorker.__repr__c             C   sN   t tjdd| _| jttjd t	| j t
| tj| j d S )Nzdistributed.admin.log-length)nzdistributed.admin.log-format)r=   r   r   r   _deque_handlerZsetFormatterloggingZ	Formatterr}   Z
addHandlerr   ZremoveHandler)r
  rO   rO   rP   r     s    zWorker._setup_loggingc             C   s   | j S )z" For API compatibility with Nanny )rZ   )r
  rO   rO   rP   worker_address  s    zWorker.worker_addressc                sR   t t jt jt jt jd} fdd j D }t| j	
 |S )N)rI   Z	in_memoryrH   	in_flightc                s   i | ]\}}| |qS rO   rO   ).0kZmetric)r
  rO   rP   
<dictcomp>!  s    z&Worker.get_metrics.<locals>.<dictcomp>)r~   r  rI   r   rH   r   r   itemsr   monitorrecent)r
  coreZcustomrO   )r
  rP   get_metrics  s    
zWorker.get_metricsc             C   s    t | j| j| jj| j| jdS )N)typeidr{   r   r   )r   r  r!  r{   rZ   r   r   )r
  r   rO   rO   rP   identity%  s
    zWorker.identityc             c   sj  | j d   t }| jd kr&| j| _td xr| jr\t || j kr\| jddV  d S | j	dkrlt
jyt }t| jj| jdV }|jtdd| jt| j| j| j| jt | j| j| j| jt |  d	d
gdV  |jd
gd}| jr| jt |  }|dk rw4t
t|d|}|V }t }|| d }|d | | _ d| _	P W q4 t!k
r   td| jj t
"dV  Y q4 t
j#k
r   td Y q4X q4W |d dkrt$d|f nhxL|d D ]@}	t%&|	}
t'|
dr|
| d}n|
 }td|
|f  qW td| jj td t(d| j)d| _*| j*+| | j d +  | j),| j-| d S )Nr|   z1-------------------------------------------------r   )timeout)closedclosing)rz   zregister-workerF)opZreplyrZ   rq   r   r   r   nowr  r   Zlocal_directoryr   pidr   Zmsgpack)serializers)deserializersr   )seconds   r#   runningzWaiting to connect to: %26sg?z&Timed out when connecting to schedulerr   OKz%Unexpected response from register: %rzworker-setupsdask_worker)r/  zInit function %s ran: output=%sz        Registered to: %26sZ2ms)intervalr   ).r   stopr#   r   rZ   r}   infor   r   r   r   r   r   r{   rz   writer~   r   r   r   r   r   r   r   r   r   r   getpidr  readwith_timeoutr   r   EnvironmentErrorsleepTimeoutError
ValueErrorr'   loadsr2   r   r   r   startadd_callbackhandle_scheduler)r
  r<  _startr   futureZdiffresponseZ_endmiddleZfunction_bytesZsetup_functionresultrO   rO   rP   _register_with_scheduler0  s|    






zWorker._register_with_schedulerc             c   s   | j sd| _ td| j  zytt }| jj| jt |  dV }t }|| d }|d dkrn| 	 V  d S |d | | _
|d d	 | jd
 _W n tk
r   td Y nX W d d| _ X n
td d S )NTzHeartbeat: %s)rZ   r'  r   r,  r   Zmissingr#   zheartbeat-intervali  r|   zHeartbeat to scheduler failedFzHeartbeat skipped: channel busy)r   r}   debugrZ   r#   r{   Zheartbeat_workerr   r  rD  r   r   Zcallback_timer   warning)r
  r<  rA  endrB  rO   rO   rP   r|   v  s(    

zWorker.heartbeatc          
   c   s   zPy| j || j| jgdV  W n. tk
rL } zt|  W d d }~X Y nX W d | jrrtd | j	| j
 n| jddV  X d S )N)Zevery_cyclez0Connection to scheduler broken.  Reconnecting...F)ru   )Zhandle_streamensure_communicatingensure_computing	Exceptionr}   	exceptionr   r2  r   r=  rD  r   )r
  r   erO   rO   rP   r>    s    


zWorker.handle_schedulerc             C   s6   ddl m} | jdkr,|| jd| itd| _| j S )zUStart an IPython kernel

        Returns Jupyter connection info dictionary.
        r   )rn   Nr]   )ipnsr   )Z_ipython_utilsrn   r   rM  r}   Zget_connection_info)r
  r   rn   rO   rO   rP   rn     s    

zWorker.start_ipythonc          
   #   s   t j| j|  fdd}t|dk r2||}nt||V }|ryt  W n@ tk
r } z"t	| t
dt|dW d d }~X Y nX t
dt|dd S )Nc          	      s>   t | tr|  } t d}||  |  W d Q R X | S )Nwb)r   r   encodeopenr3  flush)r   f)out_filenamerO   rP   func  s    

z Worker.upload_file.<locals>.funci'  rU   )r   rK  r.  )r   r   )r   r   r   r   r  r   r7   rJ  r}   rK  r   r   r(   )r
  r   filenamer   loadrU  rL  rO   )rT  rP   rm     s    

 zWorker.upload_filec             C   s
   t | jS )N)r   r   )r
  r   rO   rO   rP   rq     s    zWorker.keysc             #   sr    fdd|  D }t| j jdV \}}}|rTtd||| td|dn j|dd td	d
id S )Nc                s*   i | ]"\}}| j krd d |D |qS )c             S   s   g | ]}t |qS rO   )r!   )r  ZaddrrO   rO   rP   
<listcomp>  s    z,Worker.gather.<locals>.<dictcomp>.<listcomp>)r   )r  r  v)r
  rO   rP   r    s   z!Worker.gather.<locals>.<dictcomp>)r   whoz4Could not find data: %s on workers: %s (who_has: %s)zmissing-data)r   rq   F)r   ru   r   r.  )r  rC   r   rZ   r}   rF  r   rj   )r
  r   r   rC  Zmissing_keysZmissing_workersrO   )r
  rP   rf     s    

zWorker.gatherc                sR   | j |d krtj n&j  fddtt|t D  fdd D S )Nc                s   g | ]} |  qS rO   rO   )r  i)LrO   rP   rX    s    z#Worker.get_logs.<locals>.<listcomp>c                s   g | ]}|j  |fqS rO   )Z	levelnameformat)r  msg)deque_handlerrO   rP   rX    s    )r  r   r   rangeminr  )r
  r   r  rO   )r\  r_  rP   rp     s     zWorker.get_logs c             C   s   x| j  D ]\}}t|tr(|\}}nd}t|tr@|\}}n
|i  }}|| fd| ji|| j|< | j| ||f | j| j| j|< qW d S )Nr   ry   )	r   r  r   tupler   r   listenportr   )r
  Z	listen_ipr  rY  re  r  rO   rO   rP   start_services  s    




zWorker.start_servicesr   c             c   s  | j d kstt  dt_|sPt| jj}| jt	| jj| j
d t| j| _nxt|trtt| jj }| _| j||f| j
d nB| j|| j
d t| j| _yt|}W n tk
r   |}Y nX d|kr|d\}}| jp| j| _t| j| | j| jd | | yd| jj|| jf }W n& tk
rJ   d| jj|f }Y nX td| j td| x6| j D ](\}}td	||d
 t| f  qrW td| jj td td| j  | j!rtdt"| j! td| j t#d| j  | $ V  | %  d S )NT)r   z://)Z	parameterZfile_dirargvz%s%s:%dz%s%sz      Start worker at: %26sz         Listening to: %26sz  %16s at: %26s:zWaiting to connect to: %26sz1-------------------------------------------------z              Threads: %26dz               Memory: %26sz      Local Directory: %26szdask-worker [%s])&r   r   rE   r9   Zon_event_loop_threadr   r{   rZ   rd  r   r   rM  r   r   r1   r:  splitr   r%   r   r   r   rf  Zlistenerr\   re  rJ  r}   r2  r   r  strr   r   r<   r&   rD  Zstart_periodic_callbacks)r
  Zaddr_or_portZlisten_hostprotocolZlistening_addressr  rY  rO   rO   rP   r?    sR    




$

zWorker._startc             C   s   | j | j| d S )N)r   r=  r?  )r
  re  rO   rO   rP   r<  5  s    zWorker.start
   c       	   
   c   s  t   | jdkrd S t  td| j d| _td |   x| j	 D ]}|  qNW t
ttj* |rtt|d| jj| jdV  W d Q R X | j  | jjj  t| jtr| jjj  | jj||d n| jjdd	 | jj||d | j  x | j D ]\}}|  qW d
| _|rld| j krl| !| j"| j d f}|# V  W d Q R X | j$r| j$j%& s| j$'ddi | j$r| j$(  | j!(  | j)*  | +  | ( V  td W d Q R X d S )N)r$  r%  zStopping worker at %sr%  zdask-worker [closing])r+  )rZ   )waitr#  F)rm  r$  nannyr&  zclose-streamzdask-worker [closed]),r4   r   rF   r}   r2  rZ   r&   r1  r   valuesr5   r7  r   r9  r6  r   r{   Z
unregisterr   Z	close_rpcr   Z_work_queueZqueueclearr   r   r.   Zshutdownr   releaser   r  r   r   rM  rl   r   r   r$  sendrv   r   r   _remove_from_global_workers)	r
  ru   r#  rn  Zexecutor_waitr  r  rY  rrO   rO   rP   r   8  sJ    







zWorker._closec             C   s   |    d S )N)rs  )r
  rO   rO   rP   __del__k  s    zWorker.__del__c             C   s>   x8t tD ],}| | kr"t| | d kr
t| q
W d S )N)r   r  remove)r
  r	  rO   rO   rP   rs  n  s
    


z"Worker._remove_from_global_workersc             c   s   | j |dV  tdd S )N)ru   r.  )r   r   )r
  r   ru   rO   rO   rP   rl   u  s    zWorker.terminatec             c   s   | j  V  | jdkstd S )Nr$  )r   rm  r   r   )r
  rO   rO   rP   wait_until_closedz  s    zWorker.wait_until_closedc                sX    j krDtdjdj  < tj fdd}j| j   | d S )NZ1ms)r0  r   c              3   s.   t  jdV } | ddiV  |  d S )N)rz   r&  Zconnection_stream)r   rz   r3  r<  )r   )rZ   bcommr
  rO   rP   batched_send_connect  s    z3Worker.send_to_worker.<locals>.batched_send_connect)r   r   r   r   	coroutiner=  rr  )r
  rZ   r^  ry  rO   )rZ   rx  r
  rP   send_to_worker  s    

zWorker.send_to_workerc             #   sD  t  }|d kr j}|r8|r8t|jt jkr8|d }|dk	rX j|krXtddi  jd7  _ fdd|D }t|t|k rxJt	|t	| D ]6}| j
krddlm}	 |	t j
|  j|||< qW d	d
d | D d}
 fdd|D }t  } jd k	r$ jd ||  t  }zny4|j|
|dV }|j|dV }|d	ks^t|W n4 tk
r   tjd j|dd |   Y nX W d   jd8  _X t  } jd k	rЈ jd ||  ttd | }  jd7  _|| pd} j| j | j || d ||||||| d	 tdd S )Nr,  Fr   busyr   c                s"   i | ]}| j kr j | |qS rO   )r   )r  r  )r
  rO   rP   r    s    z#Worker.get_data.<locals>.<dictcomp>)Actorr.  c             S   s   i | ]\}}t ||qS rO   )r(   )r  r  rY  rO   rO   rP   r    s    )r   r   c                s   i | ]} j ||qS rO   )r   r   )r  r  )r
  rO   rP   r    s    zget-data-load-duration)r)  )r*  z$failed during get data with %s -> %sT)exc_infozget-data-send-durationg      ?)	r<  r1  rB  durationrZ  rq   total
compressed	bandwidthz
dont-reply)r#   r   r   Zpeer_addressrZ   r   r   r   r  r   r   actorr}  r   r  digestsaddr3  r5  r   r7  r}   rK  abortsumfilterro  r   r   r  r   )r
  r   rq   rZ  r)  max_connectionsr<  r   r  r}  r^  r   r1  r  rA  total_bytesr  rO   )r
  rP   ri     s`    
 
zWorker.get_datac             C   s   x|  D ]\}}|| jkr.| j|d|d n@| || d| j|< d | j|< d | j|< d | j|< t | j|< || j	kr| j
|d|d | j|df q
W |r| jdt|d dd |  D dd	}|S )
NrT   )valuezreceive-from-scatterzadd-keys)r&  rq   c             S   s   i | ]\}}t ||qS rO   )sizeof)r  r  rY  rO   rO   rP   r    s    z&Worker.update_data.<locals>.<dictcomp>r.  )r   r   )r  r   
transitionput_key_in_memoryr   r   r   r   r   r   transition_depr   r  r   rr  r   )r
  r   r   ru   r)  keyr  r2  rO   rO   rP   rj     s$    





zWorker.update_datac             c   s   |rxHt |D ]<}| j|df || jkr6| | || jkr| | qW tdt	| |rtd | j
j| jt |dV  tdd S )NdeletezDeleted %d keysz#Reporting loss of keys to scheduler)rZ   rq   r.  )r   r   r  r   r   r   release_depr}   rE  r  r{   Zremove_keysr   r   )r
  r   rq   ru   r  rO   rO   rP   rk     s    




zWorker.delete_datac             k   sl   xP|  D ]D\}}|| jkr:| j|  || j|  7  < n
|| j|< || j|< q
W | jj| j| jdV  d S )N)r  r]   )r  r   r   r{   set_resourcesr   )r
  r  rt  quantityrO   rO   rP   r    s    


zWorker.set_resourcesFc          
      s2  y| j krx j| }|dkrT| jks8| jks8ttd||  | d S |tkr`d S |dkrx j	|=  j
|= |d k	rt| jf }  jd8  _ j|dkrd j|<  | d  j |<  j|df | j|< |	 j|< d S  j|df yXt }t|||| j |< |r0d  j|< t }|| dkrZ j| d||f W nf tk
r } zFtjd	d
d t|}||d< d|d<  j|  j|df d S d }~X Y nX | j|< |	 j|< |
r|
 j|< d j|< |d k	r j| |pi }t| j|< t  j |< x|D ]}| j!krNt  j!|<  j!| "| | jkr j|dkrd}nd}| j|<  j|d|f  j| dkr2 j | "| q2W x|# D ]~\}}|st| j$krt| j$|<  j$| | x>|D ]6} j%| "|  j| dkr j&| | qW qW  j | rn j'| n (|d  j)r|rt* fdd|D stt* fdd|D stx|D ]} +| qW  ,| W nF tk
r, } z&t-| t.rdd l/}|0   W d d }~X Y nX d S )NrT   z,Asked to compute pre-existing result: %s: %sZerredr   znew-task-already-in-memorynewg{Gz?ZdeserializezCould not deserialize taskT)r~  r  z
task-erredr&  zdeserialize-errorrG   znew-deprH   c             3   s   | ]}| j kV  qd S )N)r   )r  dep)r
  rO   rP   	<genexpr>t  s    z"Worker.add_task.<locals>.<genexpr>c             3   s   | ]}| j kV  qd S )N)r   )r  r  )r
  rO   rP   r  u  s    r   )1r   r   r   r   r   r}   rE  send_task_state_to_schedulerIN_PLAYr   r   rc  r   r   r   r   r  r   r   r#   _deserializer   rJ  rF  r   r   rr  r   r   updater   r   r   r   r  r  r   r   r   r   r  rS   allvalidate_depvalidate_keyrK  LOG_PDBpdb	set_trace)r
  r  functionargsr  taskr   r   Zpriorityr  r   r  kwargs2stater<  r1  rL  Zemsgr  workersr]   r  rO   )r
  rP   r     s    




















zWorker.add_taskc             K   s   y| j | }W n tk
r"   d S X ||kr0d S | j||f }||f|}| jd|||p\|f || j kr|pr|| j |< | jr| | d S )Nr  )r   KeyErrorr   r   r  rS   r  )r
  r  finishr  r<  rU  r  rO   rO   rP   r    s    
zWorker.transition_depc          
   C   sx   y0| j r$|| jkst| j| s$t|| j|< W nB tk
rr } z$t| tr`dd l}|	   W d d }~X Y nX d S )Nr   )
rS   r   r   r   rJ  r}   rK  r  r  r  )r
  r  r]   rL  r  rO   rO   rP   r     s    
z$Worker.transition_dep_waiting_flightc          
   C   sR  y| j r|| jkst| j|= |rxy| j| | W n tk
rL   Y nX y| j| | W n tk
rv   Y nX | j|s|| jkr| j	| | j
| j| xB| j|dD ]0}| j| dkr|r| j| q| j| qW | j| s| | W nF tk
rL } z&t| tr:dd l}|   W d d }~X Y nX d S )NrO   rG   r   )rS   r   r   r   rv  r  r   r   r   r  r   r=  handle_missing_depr   r   r   
appendleftr  r  rJ  r}   rK  r  r  r  )r
  r  r]   rv  r  rL  r  rO   rO   rP   r     s<    

z$Worker.transition_dep_flight_waitingc          
   C   s   y`| j r|| jkst| j|= | j| rTd| j|< | || | jd|gd n
| | W nB t	k
r } z$t
| trdd l}|   W d d }~X Y nX d S )NrT   zadd-keys)r&  rq   r   )rS   r   r   r   r   r  r   rr  r  rJ  r}   rK  r  r  r  )r
  r  r  rL  r  rO   rO   rP   r     s     


z#Worker.transition_dep_flight_memoryc          
   C   s   yF| j rD|| jkst|| jks$t|| jks2t| j| dksDtW nB tk
r } z$t| t	rvdd l
}|   W d d }~X Y nX |tk	r|| jkr| j||dd d S )NrT   r   F)r  )rS   r   r   r   r   r   rJ  r}   rK  r  r  r  no_valuer  )r
  r  r  rL  r  rO   rO   rP   r     s    
z$Worker.transition_dep_waiting_memoryc             K   sh   | j | }||krd S | j||f }||f|}| j|||p@|f |pL|| j |< | jrd| | d S )N)r   r   r   r  rS   r  )r
  r  r  r  r<  rU  r  rO   rO   rP   r    s    
zWorker.transitionc          
      s   y j rr j| dkst| jks(t j| r6tt fdd j| D sVt| jksdt| jksrt j|d  | j	kr j
| dS t j j| |f W nB tk
r } z$t| trdd l}|   W d d }~X Y nX d S )NrG   c             3   s"   | ]}| j kp| jkV  qd S )N)r   r   )r  r  )r
  rO   rP   r    s    z2Worker.transition_waiting_ready.<locals>.<genexpr>rJ   r   )rS   r   r   r   r  r   rI   rH   r   r   rJ   r  heapqheappushr   rJ  r}   rK  r  r  r  )r
  r  rL  r  rO   )r
  rP   r     s&     

zWorker.transition_waiting_readyc          
   C   s   yX| j rD| j| dkst|| jks(t|| jks6t|| jksDt| j|= | | W nB tk
r } z$t	| t
rdd l}|   W d d }~X Y nX d S )NrG   r   )rS   r   r   r   rI   rH   r  rJ  r}   rK  r  r  r  )r
  r  r  rL  r  rO   rO   rP   r   	  s    
zWorker.transition_waiting_donec          
      s   yt j rV| jkst j| tks(t| jks6tt fdd j| D sVt j	|  j
 j| W nB tk
r } z$t| trdd l}|   W d d }~X Y nX d S )Nc             3   s"   | ]}| j kp| jkV  qd S )N)r   r   )r  r  )r
  rO   rP   r  !  s    z4Worker.transition_ready_executing.<locals>.<genexpr>r   )rS   r   r   r   READYrH   r  r   rI   r  r   r=  executerJ  r}   rK  r  r  r  )r
  r  rL  r  rO   )r
  rP   r     s     
z!Worker.transition_ready_executingc             C   s   |  | d S )N)r  )r
  r  r  rO   rO   rP   r   ,  s    zWorker.transition_ready_memoryc             C   s^   |  | x,| j|  D ]\}}| j|  |8  < qW | jrZtdd | j D sZtd S )Nc             s   s   | ]}|d kV  qdS )r   NrO   )r  rY  rO   rO   rP   r  5  s    z:Worker.transition_constrained_executing.<locals>.<genexpr>)r   r   r  r   rS   r  ro  r   )r
  r  resourcer  rO   rO   rP   r   /  s
    
z'Worker.transition_constrained_executingc       
   
   C   s  y| j r>|| jks"|| jks"t|| jks0t|| jks>td }|| jkrzx,| j|  D ]\}}| j|  |7  < q\W | j	| dkr| j
| |  jd7  _n| j	| dkr| j
| |tk	rdyd| j	|< | j||dd W nd tk
rJ } zDtjddd	 t|}|d
 | j|< |d | j|< d| j	|< d}W d d }~X Y nX || jkrd| |d |r~| jr~| | nt|S  tk
r   td Y nF tk
r } z&t| trdd l}	|	   W d d }~X Y nX d S )NrI   r   zlong-runningrT   F)r  zFailed to put key in memoryT)r~  rK  	tracebackrU   zComm closedr   )rS   rI   r   r   r   rH   r   r  r   r   rv  r   r  r  rJ  r}   r2  r   r   r   r   r  r   r  r   r7  rK  r  r  r  )
r
  r  r  ru   outr  r  rL  r^  r  rO   rO   rP   r   7  sN    




z Worker.transition_executing_donec          
   C   s   yL| j r|| jkst| j| | j| | jd||d |   W nB t	k
r } z$t
| tr|dd l}|   W d d }~X Y nX d S )Nzlong-running)r&  r  compute_durationr   )rS   rI   r   rv  r   r  r   rr  rI  rJ  r}   rK  r  r  r  )r
  r  r  rL  r  rO   rO   rP   r   h  s    

z(Worker.transition_executing_long_runningc             C   s$   | j |dkr | j|d|d d S )NrI   zlong-running)r  )r   r   r  )r
  r  r  rO   rO   rP   maybe_transition_long_running{  s    z$Worker.maybe_transition_long_runningc             C   s,   || j k|| jk|td| jk|| jkdS )Nr   )rI   r   heapr   )rI   r   r   rH   r   )r
  r  rO   rO   rP   stateof  s    zWorker.stateofc                s    fdd| j D S )Nc                s<   g | ]4 t  fd dD s4t  fddD r qS )c             3   s   | ]}| kV  qd S )NrO   )r  r  )r^  rO   rP   r    s    z*Worker.story.<locals>.<listcomp>.<genexpr>c             3   s0   | ](} D ]}t |tttfr
||kV  q
qd S )N)r   rc  r   r   )r  r  c)r^  rO   rP   r    s   )any)r  )rq   )r^  rP   rX    s    z Worker.story.<locals>.<listcomp>)r   )r
  rq   rO   )rq   rP   story  s    zWorker.storyc          
      s  d}yx|rj rtjjk rd}tdtj tjj j d }|jkrrj   d}qj	|dkrj
|df j   d}qj| }jrtfdd|D stfd	d
|D }fdd|D rVtd| fddD }x|D ]}j| qW jjjf|  fdd
|D }j
d||f d}x |rtjjk sjjk r| }j| dkrqp|jkrqpfdd
j| D }|sd}qptj  fdd
|D }|rt|}	ntt|}	 |	|\}
} j|7  _|
j|	< x|
D ]}j!|d|	d qRW jjj"|	||
||d d}qpW |s|sj   qW W nF t#k
r } z&t$| t%rdd l&}|'   W d d }~X Y nX d S )NTFz7Ensure communicating.  Pending: %d.  Connections: %d/%dr   rG   zcommunication passc             3   s   | ]}| j kV  qd S )N)r   )r  r  )r
  rO   rP   r    s    z.Worker.ensure_communicating.<locals>.<genexpr>c                s   g | ]} j | d kr|qS )rG   )r   )r  r  )r
  rO   rP   rX    s    z/Worker.ensure_communicating.<locals>.<listcomp>c                s   h | ]} j |s|qS rO   )r   r   )r  r  )r
  rO   rP   	<setcomp>  s    z.Worker.ensure_communicating.<locals>.<setcomp>z"Can't find dependencies for key %sc                s   h | ]}| j kr|qS rO   )r   )r  r  )r
  rO   rP   r    s    c                s   g | ]}| kr|qS rO   rO   )r  r  )missing_depsrO   rP   rX    s    zgather-dependenciesc                s   g | ]}| j kr|qS rO   )r   )r  w)r
  rO   rP   rX    s    c                s   g | ]}t | kr|qS rO   )r   )r  r  )hostrO   rP   rX    s    rW   )r]   )cause)(r   r  r   r   r}   rE  r   popleftr   r   r   r  r   rS   r  r   r2  r   r  r   r=  r  r   r   r   r   r   r   rZ   randomZchoicer   select_keys_for_gatherr  
gather_deprJ  rK  r  r  r  )r
  Zchangedr  depsZmissing_deps2r  r  r  Zlocalr]   Z	to_gathertotal_nbytesdrL  r  rO   )r  r  r
  rP   rH    s    $










zWorker.ensure_communicatingc             C   s*  || j ks| j|ry| j | }W n tk
rB   | j| }Y nX | j| pTt|}| j|pht|}~yt|}W n  t	k
r   t
|j}Y nX dd||| j||d}nJ|| jkrdd|| j|| j| | j| d}ntd|| j|  d S || jkr| j| |d< | j| d S )	Nztask-finishedr.  )r&  r   r  r   threadr   z
task-erredrU   )r&  r   r  r  rK  r  z'Key not ready to send to worker, %s: %sr   )r   r   r   r  r   r  r   r   dumps_functionr   r'   dumpsr  r   r   r   r}   rU   r   r   r   rr  )r
  r  r  r   typr  rO   rO   rP   r    s>    



z#Worker.send_task_state_to_schedulerc             C   s  || j krd S || jkr$|| j|< n8t }|| j |< t }|| dkr\| j| d||f || jkrtt|| j|< t|| j|< xT| j	
|dD ]B}|| jkr|| j| kr| j| | | j| s| |d qW |r|| jkr| |d | j|df d S )Ng{Gz?z
disk-writerO   rH   rT   zput-in-memory)r   r   r#   r   r  r   r  r   r   r   r   r   rv  r  r   r   )r
  r  r  r  r<  r1  r  rO   rO   rP   r    s*    





zWorker.put_key_in_memoryc             C   st   |h}| j | }| j| }xP|rj| }| j|dkr:q|| j |  | jkrPP || || j | 7 }qW ||fS )NrG   )r   r   r  r   r   r   r  )r
  r]   r  r  r  r\  r  rO   rO   rP   r  "  s    


zWorker.select_keys_for_gatherc             #   s"   j dkrd S t  i }zy jr2   t fdd|D } jd|||f tdt	| t
 }t j|| jdV }t
 }|d dkr܈ jd	||f x*|D ]"} j|d d
kr |d qW d S |r j| d| j | j f t fdd|d D }	|| p*d}
 j| j | j || d  j |
 fdd|d D |	|	|
 |d  jd k	r jd |	|
   jd |
  jd t	|d    jd7  _ jd|t|d f |d r jdt|d d W n tk
r } z\td|  jd|f x: j|D ]*} j | !|  j | sN j |= qNW W d d }~X Y nN t"k
r } z.t|  jrt#rdd l$}|%   W d d }~X Y nX W d   j&|8  _&|dddk}|di }x j'|D ]}|s@||kr@ j|d|| d  n& j|dkrf j|d|| d! |s||kr| j(kr jd"|f  jd#||d$ qW  jr    )  |sd _* +  n8  j*d7  _*t,-d%d& j*  V   .|V   +  X W d Q R X d S )'Nr-  c             3   s"   | ]} j |d kr|V  qdS )rW   N)r   r   )r  r  )r
  rO   rP   r  ?  s    z$Worker.gather_dep.<locals>.<genexpr>zrequest-depzRequest %d keys)rZ  r   r|  zbusy-gatherrW   rG   Ztransferc             3   s   | ]} j |d V  qdS )r   N)r   r   )r  r  )r
  rO   rP   r  X  s    r   g      ?g       @c                s   i | ]} j |d |qS )N)r   r   )r  r  )r
  rO   rP   r  _  s    z%Worker.gather_dep.<locals>.<dictcomp>)r<  r1  rB  r  rq   r  r  rZ  ztransfer-bandwidthztransfer-durationztransfer-countr   zreceive-depzadd-keys)r&  rq   z+Worker stream died during communication: %szreceive-dep-failedr   rb  rT   )r  )r]   rv  zmissing-depzmissing-data)r&  Zerrant_workerr  g?g      ?)/r   r4   rS   validate_staterc  r   r  r}   rE  r  r#   get_data_from_workerr   rZ   r   r   r  r   r   r  r   r  r  Zcountersr   r   r   rr  r7  rK  r   r   r   rv  rJ  r  r  r  r   r   r   rI  r   rH  r   r8  query_who_has)r
  r]   r  r  r  r  rA  r<  r1  r  r  rL  r  r  r|  r   rO   )r
  rP   r  3  s    










zWorker.gather_depc             C   sb   t dt| }xB| j| D ]4}t|}|d | j|< |d | j|< | |d qW | | d S )Nz/Could not find dependent %s.  Check worker logsrK  r  rU   )r:  rj  r   r   r   r   r  r  )r
  r  excr  r^  rO   rO   rP   bad_dep  s    zWorker.bad_depc       	      /   s  t |} jd|f zyD fdd|D }|s:d S x6t |D ]*} j| }|dkrD||  | qDW |szd S x |D ]}td| j|  qW  jj	t |dV }dd |
 D } | x|D ]} j|  d	7  < ||s j|d
 j|f  | q҈ j|df x0 j|dD ]}| jkr: j| q:W qW W nl tk
r   tjddd |dd} jd||f |dkrȈ j f|d|d	 iV  n Y nX W d y x|D ]} j| qW W n tk
r
   Y nX    X d S )Nzhandle-missingc                s   h | ]}| j kr|qS rO   )r   )r  r  )r
  rO   rP   r    s    z,Worker.handle_missing_dep.<locals>.<setcomp>   z.Dependent not found: %s %s .  Asking scheduler)rq   c             S   s   i | ]\}}|r||qS rO   rO   )r  r  rY  rO   rO   rP   r    s    z-Worker.handle_missing_dep.<locals>.<dictcomp>r   zno workers foundznew workers foundrO   z#Handle missing dep failed, retryingT)r~  retrieszhandle-missing-failedr   )r   r   r  r   rv  r  r}   r2  r{   r   r  update_who_hasr   r   r  r   r   rJ  rU   r  r   r  rH  )	r
  r  r  Zoriginal_depsr  Z
suspiciousr   r  r  rO   )r
  rP   r    sT    







zWorker.handle_missing_depc          	   g   s:   t  * | jj|dV }| | t|W d Q R X d S )N)rq   )r4   r{   r   r  r   r   )r
  r  rA  rO   rO   rP   r    s    
zWorker.query_who_hasc          
   C   s   yhxb|  D ]V\}}|sq|| jkr6| j| | nt|| j|< x|D ]}| j| | qJW qW W nB tk
r } z$t| t	rdd l
}|   W d d }~X Y nX d S )Nr   )r  r   r  r   r   r  rJ  r}   rK  r  r  r  )r
  r   r  r  r]   rL  r  rO   rO   rP   r    s    


zWorker.update_who_hasc             C   s<   | j |d }d||d}| j| |dkr8| | d S )Nzsteal-response)r&  r  r  )rH   rG   )r   r   r   rr  r   )r
  r  r  rA  rO   rO   rP   r     s    zWorker.steal_requestc          
   C   s  y"|| j krd S | j |}|r:| j|dd|if n| j|df | j|= || jkr|| jkry| j|= W n" tk
r   tj	ddd Y nX | j
|= | j|= || jkr|| jkr| j|= | j
|= | j|= || jkr| j|= xR| j|dD ]@}|| jkr| j| | | j| s| j| dkr| | qW || jkrL| j|= | j|= | j|= || jkrp| j|= || jkr| j|= || jkr| j|= || jkr| j| || jkr|dkrx.| j|  D ]\}}| j|  |7  < qW | j|= |r"|tkr"| jd	||d
 W nZ t k
r:   Y nF t!k
r~ }	 z&t"|	 t#rldd l$}
|
%   W d d }	~	X Y nX d S )Nzrelease-keyr  z$Tried to delete %s but no file foundT)r~  rO   )rG   rW   rI   rq  )r&  r  r  r   )&r   r   r   r  r   r   r   FileNotFoundErrorr}   rU   r   r   r   r   r   r   discardr  r   r   r   r   r   r   rI   rv  r   r  r   
PROCESSINGr   rr  r   rJ  rK  r  r  r  )r
  r  r  reasonru   r  r  r  r  rL  r  rO   rO   rP   r     sp    




zWorker.release_keyc          
   C   s  y4|| j krd S | j|df | j |}|| jkr@| j|= || jkrpx$| j|D ]}| j| | qXW || jkr|| j	kr| j	|= | j
|= || jkr| j|= | j
|= | j|= || jkr| j|}| j| | x2| j|dD ] }| j| dkr| j||d qW |r4|dkr4| jd|gd W nF tk
r| } z&t| trjdd l}|   W d d }~X Y nX d S )Nzrelease-deprO   rT   )r  zrelease-worker-data)r&  rq   r   )r   r   r  r   r   r   r   rv  r   r   r   r   r   r   r   r   r   r   rr  rJ  r}   rK  r  r  r  )r
  r  ru   r  r]   r  rL  r  rO   rO   rP   r  H  sB    







zWorker.release_depc          
   C   s   y| j |tkrd S | j |= | j|= || jkr8| j|= x8| j|dD ]&}| j| | | j| sH| j|= qHW || jkr|| j	kr| j	|= || j
kr| j
|= W nB tk
r } z$t| trdd l}|   W d d }~X Y nX d S )NrO   r   )r   r   PENDINGr   r   r   r   r   rv  r   r   rJ  r}   rK  r  r  r  )r
  r  r  rL  r  rO   rO   rP   rescind_keyq  s,    





zWorker.rescind_keyrO   c             #   s|   |p| j }td  d7  < |p i }|j|f|| t fddd}|  z
 V  W d|  X   }t|dS )aC   Safely run function in thread pool executor

        We've run into issues running concurrent.future futures within
        tornado.  Apparently it's advantageous to use timeouts and periodic
        callbacks to ensure things run smoothly.  This can get tricky, so we
        pull it off into an separate method.
        r   r   c                  s   t d jS )Nzfuture state: %s - %s)r}   rE  Z_staterO   )r@  r  rO   rP   rQ     s   z(Worker.executor_submit.<locals>.<lambda>i  N)	r   job_counterZsubmitr>   r<  r1  rC  r   r   )r
  r  r  r  r  r   r  rC  rO   )r@  r  rP   executor_submit  s    



zWorker.executor_submitc             C   s   |pi }t | |||||dS )N)r  r  r  rm  )rg   )r
  r   r  r  rm  r  rO   rO   rP   rg     s    z
Worker.runc             C   s   t | |||||dS )N)r  r  r  rm  )rg   )r
  r   r  r  r  rm  rO   rO   rP   rh     s    zWorker.run_coroutinec          
   c   s   | dd}|}| j| }t||}t|d | }	t|rJ|||V }
n:|rz| j|	t|||| j|	| j| j	f| j
dV }
n
|||}
tdt|
dd S )Nseparate_threadT.)r  r   r.  )r   rC  )r   r   getattrr;   rA   r  apply_function_actorr   r   r   r   r   r   r(   )r
  r   r  r  r  r  r  r  rU  r   rC  rO   rO   rP   rs     s$    


zWorker.actor_executec             C   s   t | j| |}dt|dS )Nr.  )r   rC  )r  r   r(   )r
  r   r  Z	attributer  rO   rO   rP   rt     s    zWorker.actor_attributec             C   s@   || j krdS x,| j |  D ]\}}| j| |k rdS qW dS )NTF)r   r  r   )r
  r  r  ZneededrO   rO   rP   meets_resource_constraints  s    
z!Worker.meets_resource_constraintsc          
   C   s  | j r
d S yxd| jrpt| j| jk rp| jd }| j|dkrJ| j  q| |rl| j  | 	|d qP qW xF| j
rt| j| jk rt| j
\}}| j|tkrt| 	|d qtW W nB tk
r } z$t| trdd l}|   W d d }~X Y nX d S )Nr   rJ   rI   )r   rJ   r  rI   r   r   r   r  r  r  rH   r  heappopr  rJ  r}   rK  r  r  r  )r
  r  _rL  r  rO   rO   rP   rI    s,    




zWorker.ensure_computingc             c   s  d }| j dkrd S zyB|| jks.|| jkr2d S | jrX|| jksFt| j| dksXt| j| \}}}t }i }xd| j| D ]V}	y| j	|	 ||	< W q~ t
k
r   ddlm}
 |
t| j|	 | j|	| ||	< Y q~X q~W t||ttfd}t||ttfd}t }|| dkrF| j| d||f | jd k	rF| jd ||  td	|| j y.| j|t|||| j|| j| j| jfd
V }W n* tk
r } z
|} W d d }~X Y nX | j |dkrd S ||d< |!dd }| j| d|d |d f |d | j"|< |d dkrj|d | j#|< |d | j$|< | j%|d|d | jd k	r| jd |d |d   nt&|!dt'r| j()d|d | j%|ddd | j*|dd nd|d | j+|< |d  | j,|< t-d!t.t/|d d" t0|d"d#t1|d"d#t2|d j	 | %|d$ td%|| | jrH|| jks8t|| jksHt| 3  | 4  W n^ t5k
r } z>||krt6d&| n"t7| t8rd'd l9}|:   W d d }~X Y nX W d || jkr| j;| X d S )(N)r%  r$  rI   r   )r}  )Z	key_typesg{Gzt?z	disk-readzdisk-load-durationzExecute key: %s worker: %s)r  )rI   zlong-runningr  rC  Zcomputer<  r1  r  r&  ztask-finishedr   r   rT   )r  ztask-durationzactual-exceptionZ
reschedule)r&  r  rV   F)ru   rK  r  zH Compute Failed
Function:  %s
args:      %s
kwargs:    %s
Exception: %s
i  )max_lenrU   z*Send compute response to scheduler: %s, %szThread Pool Executor error: %sr   )<r   rI   r   rS   r   r   r   r#   r   r   r  r  r}  r   r   rZ   rB   bytesr   r   r  r  r  r}   rE  r  apply_functionr   r   r   r   RuntimeErrorr   r   r   r   r   r  r   
Rescheduler   rr  r   r   r   rF  rj  r0   convert_args_to_strconvert_kwargs_to_strreprrI  rH  rJ  rU   rK  r  r  r  rv  )r
  r  ru   Zexecutor_errorr  r  r  r<  r   r  r}  Zargs2r  r1  rC  rL  r  r  rO   rO   rP   r    s    
(



zWorker.executec             c   s  | j r
dS d| _ d}| jj}| j}|| j }| jr|| jkr| j  | j	st
dt|d t| jt| j d| _	n<| j	rt
dt|d t| jt| j d| _	|   | jr|| jkr| j| j }d}|| }x||kr|| jjst
dt| jt| j P | jj \}}	}
~~	||
7 }|d	7 }tjV  | j}||kr||kr| j  | j}qW |rt
d
|t| d| _ t|dS )z Track this process's memory usage and act accordingly

        If we rise above 70% memory use, start dumping data to disk.

        If we rise above 80% memory use, stop execution of new tasks
        NTr   z^Worker is at %d%% memory usage. Pausing worker.  Process memory: %s -- Worker memory limit: %sd   z^Worker is at %d%% memory usage. Resuming worker. Process memory: %s -- Worker memory limit: %sFzMemory use is high but worker has no data to store to disk.  Perhaps some other process is leaking memory?  Process memory: %s -- Worker memory limit: %sr   z+Moved %d pieces of data data and %s to disk)r   r  procZmemory_infoZrssr   r`   r  Zcollectr   r}   rF  r   r<   rI  r_   r^   r   ZfastZevictr   ZmomentrE  r   )r
  r  r  rT   Zfracr  countZneedr  rY  r   rO   rO   rP   r  Z  s\    







zWorker.memory_monitorc             C   sR   t  | j }| jt  }| _| j||f | j|t| j	f | j	
  d S )N)r#   r   r   r   r   r   r  r   r~   r   rp  )r
  r'  profrO   rO   rP   r    s
    zWorker.cycle_profilec          	      s   | j s
dS t }| j | j  }W dQ R X t   fdd|D  xT  D ]H\}}|dk	rPt|| }tj	|d| j
dd tj	|d| j| dd qPW t }| jdk	r| jd ||  dS )zz
        Get a frame from all actively computing threads

        Merge these frames into existing profile counts
        Nc                s   i | ]} | |qS rO   rO   )r  ident)framesrO   rP   r    s    z*Worker.trigger_profile.<locals>.<dictcomp>zdistributed/worker.py)r1  zprofile-duration)r   r#   r   r   r   _current_framesr  r;   r   Zprocessr   r   r  r  )r
  r<  r   r  framer  r1  rO   )r  rP   r    s"    
zWorker.trigger_profilec                s&  t  | j }d kr| j nfdd| jD  |d kr>d}nt |f}|d krZd }n"t |fd }|t kr|d }|dkr|d krt  n,|d krt n|} fddt	||D  t
jtd  }	 st
 S |d kr"|d ks||k r"d kr| j}
n
| j }
t
|	|
}	|	S )Nc                s$   g | ]\}} |kr||  fqS rO   rO   )r  tr  )r  rO   rP   rX    s    z&Worker.get_profile.<locals>.<listcomp>r   r   c                s   g | ]} | qS rO   rO   )r  r[  )historyrO   rP   rX    s    )r#   r   r   r   bisectZbisect_leftZbisect_rightr  r   r`  r   r   r   r   r   r   )r
  r   r<  r1  r  r'  ZistartZistopZiistopr  r  rO   )r  r  rP   r     s2    


zWorker.get_profilec                s   d krd}t  | j }p| p&d  fdd| jD  fdd| jD d}|r|d || jd f |d	 |d
d | j D f |S )NTr   c                s4   g | ],\}} |  k r k rn q||d  fqS )r  rO   )r  r  r  )r<  r1  rO   rP   rX    s    z/Worker.get_profile_metadata.<locals>.<listcomp>c                s>   g | ]6\}} |  k r k rn q|d d |  D fqS )c             S   s   i | ]\}}|d  |qS )r  rO   )r  r  r  rO   rO   rP   r    s    z:Worker.get_profile_metadata.<locals>.<listcomp>.<dictcomp>)r  )r  r  rY  )r<  r1  rO   rP   rX    s   )countsrq   r  r  rq   c             S   s   i | ]\}}|d  |qS )r  rO   )r  r  rY  rO   rO   rP   r    s   z/Worker.get_profile_metadata.<locals>.<dictcomp>)r#   r   r   r   r  r   r   r  )r
  r   r<  r1  Z
add_recentr'  rC  rO   )r<  r1  rP   r     s    zWorker.get_profile_metadatac          	      sn   | j . t  | j } fdd| D  W d Q R X d k	rXfdd  D  dd   D }|S )Nc                s   i | ]\}} | |qS rO   rO   )r  r  r  )r  rO   rP   r    s   z)Worker.get_call_stack.<locals>.<dictcomp>c                s   i | ]\}}| kr||qS rO   rO   )r  r  r  )rq   rO   rP   r    s    c             S   s   i | ]\}}t ||qS rO   )r   ro   )r  r  r  rO   rO   rP   r  	  s    )r   r   r  r   r   r  )r
  r   rq   r   rC  rO   )r  rq   rP   r     s    

zWorker.get_call_stackc             C   sp   || j ks|| jkst|| jks&t|| jks4t|| jksBt|| jksPt|| jkrl| j| dksltd S )NrT   )r   r   r   r   r   rI   rH   r   )r
  r  rO   rO   rP   validate_key_memory	  s    
zWorker.validate_key_memoryc                sN   | j kst| jkst| jks*tt fdd j| D sJtd S )Nc             3   s"   | ]}| j kp| jkV  qd S )N)r   r   )r  r  )r
  rO   rP   r  	  s    z0Worker.validate_key_executing.<locals>.<genexpr>)rI   r   r   r   r  r   )r
  r  rO   )r
  rP   validate_key_executing	  s    zWorker.validate_key_executingc                sb   |t d jkst| jks"t| jks0t| jks>tt fdd j| D s^td S )Nr   c             3   s"   | ]}| j kp| jkV  qd S )N)r   r   )r  r  )r
  rO   rP   r  	  s    z,Worker.validate_key_ready.<locals>.<genexpr>)r   rH   r   r   rI   r   r  r   )r
  r  rO   )r
  rP   validate_key_ready	  s
    zWorker.validate_key_readyc                s2   | j kstt fdd j| D r.td S )Nc             3   s   | ]}| j kV  qd S )N)r   )r  r  )r
  rO   rP   r   	  s    z.Worker.validate_key_waiting.<locals>.<genexpr>)r   r   r  r   )r
  r  rO   )r
  rP   validate_key_waiting	  s    zWorker.validate_key_waitingc          
   C   s   y\| j | }|dkr | | n:|dkr4| | n&|dkrH| | n|dkrZ| | W nB tk
r } z$t| trdd l	}|
   W d d }~X Y nX d S )NrT   rG   rH   rI   r   )r   r  r  r  r  rJ  r}   rK  r  r  r  )r
  r  r  rL  r  rO   rO   rP   r  "	  s     

zWorker.validate_keyc                sN   | j kst| jkst j| s*tt fdd j| D rJtd S )Nc             3   s   | ]}| j kV  qd S )N)rH   )r  r  )r
  rO   rP   r  8	  s    z.Worker.validate_dep_waiting.<locals>.<genexpr>)r   r   r   r   r  )r
  r  rO   )r
  rP   validate_dep_waiting4	  s    zWorker.validate_dep_waitingc                s\   | j kst| jkstt fdd j| D r<t j| }| j| ksXtd S )Nc             3   s   | ]}| j kV  qd S )N)rH   )r  r  )r
  rO   rP   r  =	  s    z-Worker.validate_dep_flight.<locals>.<genexpr>)r   r   r   r  r   r   r   )r
  r  ZpeerrO   )r
  rP   validate_dep_flight:	  s
     
zWorker.validate_dep_flightc             C   sT   || j ks|| jkst|| jks&t|| jks4t|| jkrP| j| dksPtd S )NrT   )r   r   r   r   r   r   )r
  r  rO   rO   rP   validate_dep_memoryA	  s
    
zWorker.validate_dep_memoryc          
   C   s   yT| j | }|dkr | | n2|dkr4| | n|dkrH| | n
td|W nB tk
r } z$t| trdd l	}|
   W d d }~X Y nX d S )NrG   rW   rT   zUnknown dependent stater   )r   r  r  r  r:  rJ  r}   rK  r  r  r  )r
  r  r  rL  r  rO   rO   rP   r  H	  s    

zWorker.validate_depc          
   C   s  | j dkrd S yfx6| j D ](\}}x|D ]}|| j| ks,tq,W qW x6| j D ](\}}x|D ]}|| j| ksdtqdW qVW x| jD ]}| | qW x| jD ]}| | qW xX| j	 D ]J\}}|| j
krx6|D ].}|| jks|| jks| j| | jstqW qW xb| jD ]X}| j| dkrt| j| tsBt|| j	ksRt|| jks|| jkstqW W nF tk
r }	 z&t|	 trdd l}
|
   W d d }	~	X Y nX d S )Nr-  rT   r   )r   r   r  r   r   r   r  r   r  r   r   r   r   issubsetr   r   r   r   r   r   r   rJ  r}   rK  r  r  r  )r
  r  r  r  r]   rq   r  r  r  rL  r  rO   rO   rP   r  Z	  s<    






&
zWorker.validate_statec          	   C   s*   | j  | jr| jS |  S W d Q R X d S )N)r   r   _get_client)r
  rO   rO   rP   client	  s    zWorker.client   c          
   C   s   yddl m} | }W n tk
r*   Y n,X |jrB|jj| jjksP|j| jjkrV|| _| jsddl m} | jt	
 k}|| j| j| jd|dd|d| _|s| jjdkst| jS )z Get local client attached to this worker

        If no such client exists, create one

        See Also
        --------
        get_client
        r   )default_client)ClientTr]   )r   r   Zset_as_defaultasynchronousZdirect_to_workersr   r#  r-  )r  r  r:  r{   rZ   Z
_start_argr   r  r   r   r   r   r   r   )r
  r#  r  r  r  r  rO   rO   rP   r  	  s*    	


zWorker._get_clientc             C   s   | j t  S )a   Get 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
        )r   r   )r
  rO   rO   rP   get_current_task	  s    zWorker.get_current_task)NNNNNrL   NNNTrM   NNNNNNNNrN   NN)N)NNT)N)NN)NN)rb  )r   )r   )Trl  TT)T)NNNN)NNTN)NNT)
NNNNNNNNNF)N)NT)N)N)N)N)N)N)T)N)NNT)F)rO   NN)rO   TN)rO   NT)NNN)F)NNNN)Nr   N)NN)r   )]r  
__module____qualname____doc__r   r  r   propertyr  r  r"  r   rz  rD  r|   r>  rn   rm   rq   rf   rp   rf  r?  r<  r   ru  rs  rl   rw  r{  ri   rj   rk   r  r   r  r   r   r   r   r  r   r   r   r   r   r  r   r   r  r  r  rH  r  r  r  r  r  r  r  r  r   r   r  r  r  rg   rh   rs   rt   r  rI  r  r  r  r  r   r   r   r  r  r  r  r  r  r  r  r  r  r  r  r  __classcell__rO   rO   )r  rP   rK   T   s    *         	
F


<
2	 C
  
j

$



1

Q%
m	6
C
)  


	fF
'

	*
!rK   c              C   sR   y
t jd S  tk
rL   x$tddd D ]} |  }|r*|S q*W tdY nX dS )a   Get 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]   NzNo workers found)r9   r   AttributeErrorr  r:  )r	  r]   rO   rO   rP   
get_worker	  s    
r  r   Tc             C   s   | r|rt | } y
t }W n tk
r0   Y nX | rB|jj| krN|j|dS ddlm} | }|rx| rt|jj| krx|S | rddlm	} || |dS tddS )a  Get 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, default 3
        Timeout (in seconds) for getting the Client
    resolve_address : bool, default True
        Whether to resolve `address` to its canonical form.

    Returns
    -------
    Client

    Examples
    --------
    >>> def f():
    ...     client = get_client()
    ...     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
    )r#  r   )_get_global_client)r  z.No global client found and no address providedN)
r   resolve_addressr  r:  r{   rZ   r  r  r  r  )rZ   r#  r  r]   r  r  r  rO   rO   rP   
get_client	  s     %

r  c              C   s2   t  } t  t tj }| jj| jtj|d dS )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
    )r  N)	r  
tpe_seceder#   r9   
start_timer   r=  r  r  )r]   r  rO   rO   rP   r/   
  s
    
r/   c               @   s   e Zd ZdZdS )r  a   Reschedule 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)r  r  r  r  rO   rO   rO   rP   r  7
  s   r  c          	   C   s   | d krd S | dkr*t ttd|t  } ttt, t| }t|tr\|dkr\t |t S W d Q R X t| t	t
fr|t| S t | S d S )NrM   r   )r   TOTAL_MEMORYra  r   r5   r:  	TypeErrorr   r   r   rj  r?   )r   r   xrO   rO   rP   r   F
  s    r   c       
   
   c   s   |dkr| j }|dkr| j}| |V }z\t|||d|||dV }y|d }	W n tk
rl   td|Y nX |	dkr|dV  W d| || X t	|dS )a   Get 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_deps
    utils_comm.gather_data_from_workers
    Nri   )r)  r*  r&  rq   rZ  r  r   zUnexpected responser.  )
r)  r*  r   r   r  r:  r3  Zreuser   r   )
r   rq   r]   rZ  r  r)  r*  r   rA  r   rO   rO   rP   r  V
  s&    
r  c             C   sb   | dk	rt | } |r t |}|r.t |}|dk	rP| sB|sB|rFtt} |f}| |pXd|p^i fS )z> Deserialize task inputs and regularize to func, args, kwargs NrO   )r'   r;  r   execute_task)r  r  r  r  rO   rO   rP   r  
  s    


r  c             C   sL   t | r,| d | dd  }}|tt| S t| trDttt| S | S dS )z Evaluate a nested task

    >>> inc = lambda x: x + 1
    >>> execute_task((inc, 1))
    2
    >>> execute_task((sum, [1, 2, (inc, 3)]))
    7
    r   r   N)r	   mapr  r   r   )r  rU  r  rO   rO   rP   r  
  s    	
r  c             C   s`   yt |  }W nN tk
r>   t| }t|dk r:|t | < Y n tk
rZ   t| }Y nX |S )z+ Dump a function to bytes, cache functions i )cacher  r'   r  r  r  )rU  rC  rO   rO   rP   r  
  s    
r  c             C   s   t | r| d tkrdttt| dd sdt| d t| d d}t| dkr`t| d |d< |S ttt| dd st| d t| dd dS t| S )	uL   Serialize 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'   KK.'}

    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.'}
    r   r,  Nr   )r  r     r   r  )	r	   r
   r  r  r3   r  
warn_dumpsr  r(   )r  r  rO   rO   rP   
dumps_task
  s    "

r  Fg    .Ac             C   sr   || }t d snt||krndt d< t| }t|dkrT|dd d |dd  }tdtt||f  |S )	z8 Dump an object to bytes, warn if those bytes are large r   TF   N2   z ... iaN  Large 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_warnedr  rj  warningswarnr<   )objr  limitbsrO   rO   rP   r  
  s    r  c          
   C   s   t  }| |||< W dQ R X t t_|t_|t_t }	zry| ||}
W n: tk
r } zt|}d|d< ||d< W dd}~X Y n&X dd|
t|
|
dk	rt	|
ndd}W dt }X |	| |d< || |d	< ||d
< | ||= W dQ R X |S )z Run a function, collect information

    Returns
    -------
    msg: dictionary with status, result/error, timings, etc..
    Nz
task-erredr&  zactual-exceptionztask-finishedr.  )r&  r   rC  r   r   r<  r1  r  )
r   r#   r9   r  r   r  rJ  r   r  r   )r  r  r  r   r  r   r   Z
time_delayr  r<  rC  rL  r^  rG  rO   rO   rP   r  
  s2    r  c       	   	   C   sN   t  }| |||< W dQ R X |t_|t_| ||}| ||= W dQ R X |S )z Run a function, collect information

    Returns
    -------
    msg: dictionary with status, result/error, timings, etc..
    N)r   r9   r   r  )	r  r  r  r   r  r   r   r  rC  rO   rO   rP   r    s    
r  c             C   sP   G dd dt }|  } d| kr2|t| d | d< d| krL|t| d | d< | S )z Make 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               @   s   e Zd Zdd Zdd ZdS )zget_msg_safe_str.<locals>.Reprc             S   s   || _ || _d S )N)_f_val)r
  rS  valrO   rO   rP   r   .  s    z'get_msg_safe_str.<locals>.Repr.__init__c             S   s   |  | jS )N)r$  r%  )r
  rO   rO   rP   r  2  s    z'get_msg_safe_str.<locals>.Repr.__repr__N)r  r  r  r   r  rO   rO   rO   rP   Repr-  s   r'  r  r  )objectr   r  r  )r^  r'  rO   rO   rP   get_msg_safe_str(  s    r)  c          	   C   s   d}dd t t| D }xt| D ]z\}}yt|}W n tk
rP   d}Y nX |||< |t|d 7 }|dk	r$||kr$dd|d|d	  d| S q$W d
d|S dS )zx Convert args to a string, allowing for some arguments to raise
    exceptions during conversion and ignoring them.
    r   c             S   s   g | ]}d qS )rb  rO   )r  r[  rO   rO   rP   rX  B  s    z'convert_args_to_str.<locals>.<listcomp>z < could not convert arg to str >r,  Nz({}z, r   z({}))r`  r  	enumerater  rJ  r]  r   )r  r  lengthstrsr[  argsargrO   rO   rP   r  =  s    
(r  c       	   	   C   s   d}dd t t| D }xt|  D ]\}\}}yt|}W n tk
rX   d}Y nX t|d | }|||< |t|d 7 }|dk	r(||kr(dd	|d|d
  d| S q(W dd	|S dS )zz Convert kwargs to a string, allowing for some arguments to raise
    exceptions during conversion and ignoring them.
    r   c             S   s   g | ]}d qS )rb  rO   )r  r[  rO   rO   rP   rX  U  s    z)convert_kwargs_to_str.<locals>.<listcomp>z < could not convert arg to str >z: r,  Nz{{{}z, r   z{{{}}})r`  r  r*  r  r  rJ  r]  r   )	r  r  r+  r,  r[  argnamer-  r.  ZskwargrO   rO   rP   r  P  s    
(r  c             C   s   t |S )N)r  )r  rY  rO   rO   rP   r   d  s    r   rO   c       
      c   s@  t |}|d krt|}n
td |s6|s6td|rDt |}|rRt |}t|drd| |d< t|drv| |d< tdt	| y>|s|||}n*|r|||V }n| j
j|f|| d }W n` tk
r$ } z@tjdtt	|d d t|ddt|ddd	d
 t|}	W d d }~X Y nX dt|d}	t|	d S )Nz\The is_coro= parameter is deprecated. We now automatically detect coroutines/async functionszCombination not supportedr/  Zdask_schedulerzRun out-of-band function %rz3 Run Failed
Function: %s
args:     %s
kwargs:   %s
i  )r  T)r~  r.  )r   rC  )r'   r;  rA   r  r  r   r2   r}   r2  r0   r   r=  rJ  rF  rj  r  r  r   r(   r   )
Zserverr   r  r  r  Zis_cororm  rC  rL  rA  rO   rO   rP   rg   h  s<    







rg   )Nr   T)NNNN)NNNN)N)N)Z
__future__r   r   r   r  collectionsr   r   Zdatetimer   r  r  r   r'   r   r  r   r   r  r  r   Z	dask.corer	   Zdask.compatibilityr
   Zcytoolzr   r   r   r   ZtoolzZtornado.genr   Ztornador   Ztornado.ioloopr   Ztornado.locksr   rb  r   r   Zbatchedr   r   r   r   Z
comm.utilsr   Zcompatibilityr   r   r   r  r   r   r   r    r!   Z	diskutilsr"   r   r#   Znoder$   Z
preloadingr%   Z	proctitler&   rk  r(   r)   r*   Zpubsubr+   r   r,   r  r-   Zthreadpoolexecutorr.   r/   r  Zutilsr0   r1   r2   r3   r4   r5   r6   r7   r8   r9   r:   r;   r<   r=   r>   r?   r@   rA   Z
utils_commrB   rC   Z
utils_perfrD   rE   rF   	cpu_countr   Z	getLoggerr  r}   r   r   r  r  ZpsutilZvirtual_memoryr  r  rF  r  r  r  r  r  r  rK   r  r  rJ  r  r   rz  r  r  r  r  r~   r  r  r  r  r  r  r  r  r)  r  r  r   rg   rO   rO   rO   rP   <module>   s   P


                   
: %
!%

