class_docstr: Replicate somewhere else all unique in-memory tasks on a worker, preparing for
  its shutdown.
  
  At any given time, the AMM may have registered multiple instances of this policy,
  one for each worker currently being retired - meaning that most of the time no
  instances will be registered at all. For this reason, this policy doesn't figure in
  the dask config (:file:`distributed.yaml`). Instances are added by
  :meth:`distributed.Scheduler.retire_workers` and automatically remove themselves
  once the worker has been retired. If the AMM is disabled in the dask config,
  :meth:`~distributed.Scheduler.retire_workers` will start a temporary ad-hoc one.
  
  **Failure condition**
  
  There may not be any suitable workers to receive the tasks from the retiring worker.
  This happens in two use cases:
  
  1. This is the only worker in the cluster, or
  2. All workers are either paused or being retired at the same time
  
  In either case, this policy will fail to move out all keys and set the
  ``no_recipients`` boolean to True. :meth:`~distributed.Scheduler.retire_workers`
  will abort the retirement.
  
  There is a third use case, where a task fails to be replicated away for whatever
  reason, e.g. because its recipient is unresponsive but the Scheduler doesn't know
  yet. In this case we'll just wait for the next AMM iteration and try again (possibly
  with a different receiving worker, e.g. if the receiving worker was hung but not yet
  declared dead).
  
  **Retiring a worker with spilled tasks**
  
  On its very first iteration, this policy suggests that other workers should fetch
  all unique in-memory tasks of the retiring worker. Frequently, this means that in
  the next few moments the retiring worker will be bombarded by
  :meth:`distributed.worker.Worker.get_data` calls from the rest of the cluster. This
  can be a problem if most of the managed memory of the worker has been spilled out,
  as it could send the worker above its terminate threshold. Two measures are in place
  in order to prevent this:
  
  - At every iteration, this policy drops all tasks on the retiring worker that have
    already been replicated somewhere else. This makes room for further tasks to be
    moved out of the spill file in order to be replicated onto another worker.
  - Once the worker passes the ``pause`` threshold,
    :meth:`~distributed.worker.Worker.get_data` throttles the number of outgoing
    connections to 1.
  
  Parameters
  ==========
  address: str
      URI of the worker to be retired
class_docstr_tok: ["Replicate","somewhere","else","all","unique","in-memory","tasks","on","a","worker",",","preparing","for","its","shutdown",".","At","any","given","time",",","the","AMM","may","have","registered","multiple","instances","of","this","policy",",","one","for","each","worker","currently","being","retired","-","meaning","that","most","of","the","time","no","instances","will","be","registered","at","all",".","For","this","reason",",","this","policy","does","n't","figure","in","the","dask","config","(",":","file",":","`","distributed.yaml","`",")",".","Instances","are","added","by",":","meth",":","`","distributed.Scheduler.retire_workers","`","and","automatically","remove","themselves","once","the","worker","has","been","retired",".","If","the","AMM","is","disabled","in","the","dask","config",",",":","meth",":","`","~distributed.Scheduler.retire_workers","`","will","start","a","temporary","ad-hoc","one",".","*","*","Failure","condition","*","*","There","may","not","be","any","suitable","workers","to","receive","the","tasks","from","the","retiring","worker",".","This","happens","in","two","use","cases",":","1",".","This","is","the","only","worker","in","the","cluster",",","or","2",".","All","workers","are","either","paused","or","being","retired","at","the","same","time","In","either","case",",","this","policy","will","fail","to","move","out","all","keys","and","set","the","``","no_recipients","``","boolean","to","True",".",":","meth",":","`","~distributed.Scheduler.retire_workers","`","will","abort","the","retirement",".","There","is","a","third","use","case",",","where","a","task","fails","to","be","replicated","away","for","whatever","reason",",","e.g",".","because","its","recipient","is","unresponsive","but","the","Scheduler","does","n't","know","yet",".","In","this","case","we","'ll","just","wait","for","the","next","AMM","iteration","and","try","again","(","possibly","with","a","different","receiving","worker",",","e.g",".","if","the","receiving","worker","was","hung","but","not","yet","declared","dead",")",".","*","*","Retiring","a","worker","with","spilled","tasks","*","*","On","its","very","first","iteration",",","this","policy","suggests","that","other","workers","should","fetch","all","unique","in-memory","tasks","of","the","retiring","worker",".","Frequently",",","this","means","that","in","the","next","few","moments","the","retiring","worker","will","be","bombarded","by",":","meth",":","`","distributed.worker.Worker.get_data","`","calls","from","the","rest","of","the","cluster",".","This","can","be","a","problem","if","most","of","the","managed","memory","of","the","worker","has","been","spilled","out",",","as","it","could","send","the","worker","above","its","terminate","threshold",".","Two","measures","are","in","place","in","order","to","prevent","this",":","-","At","every","iteration",",","this","policy","drops","all","tasks","on","the","retiring","worker","that","have","already","been","replicated","somewhere","else",".","This","makes","room","for","further","tasks","to","be","moved","out","of","the","spill","file","in","order","to","be","replicated","onto","another","worker",".","-","Once","the","worker","passes","the","``","pause","``","threshold",",",":","meth",":","`","~distributed.worker.Worker.get_data","`","throttles","the","number","of","outgoing","connections","to","1",".","Parameters","==========","address",":","str","URI","of","the","worker","to","be","retired"]
code: def __init__(self, address: str):
      self.address = address
      self.no_recipients = False
  
imports: from __future__ import annotations
  import abc
  import logging
  from collections import defaultdict
  from collections.abc import Generator
  from typing import TYPE_CHECKING, Any, Literal, NamedTuple, Union
  import dask
  from dask.typing import Key
  from dask.utils import parse_timedelta
  from distributed import client
  from distributed import scheduler
  from distributed.compatibility import PeriodicCallback
  from distributed.core import Status
  from distributed.metrics import time
  from distributed.utils import import_term, log_errors
label_desc: Use image node_id 1 to create a new RetireWorker object from inherited base classes: ActiveMemoryManagerPolicy with example: obj = RetireWorker(address)