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 run(self) -> SuggestionGenerator:
      """"""
      ws = self.manager.scheduler.workers.get(self.address)
      if ws is None:
          logger.debug(
              "Removing policy %s: Worker no longer in cluster", self
          )
          self.manager.policies.remove(self)
          return
  
      if ws.actors:
          logger.warning(
              f"Tried retiring worker {self.address}, but it holds actor(s) "
              f"{set(ws.actors)}, which can't be moved."
              "The worker will not be retired."
          )
          self.no_recipients = True
          self.manager.policies.remove(self)
          return
  
      nrepl = 0
      nno_rec = 0
  
      logger.debug("Retiring %s", ws)
      for ts in ws.has_what:
          if ts.actor:
              # This is just a proxy Actor object; if there were any originals we
              # would have stopped earlier
              continue
  
          if ts.who_has and len(ts.who_has) > 1:
              # There are already replicas of this key on other workers.
              # Suggest dropping the replica from this worker.
              # Use cases:
              # 1. The suggestion is accepted by the AMM and by the Worker.
              #    The replica on this worker is dropped.
              # 2. The suggestion is accepted by the AMM, but rejected by the Worker.
              #    We'll try again at the next AMM iteration.
              # 3. The suggestion is rejected by the AMM, because another policy
              #    (e.g. ReduceReplicas) already suggested the same for this worker
              # 4. The suggestion is rejected by the AMM, because the task has
              #    dependents queued or running on the same worker.
              #    We'll try again at the next AMM iteration.
              # 5. The suggestion is rejected by the AMM, because all replicas of the
              #    key are on workers being retired and the other RetireWorker
              #    instances already made the same suggestion. We need to deal with
              #    this case and create a replica elsewhere.
              drop_ws = yield Suggestion("drop", ts, {ws})
              if drop_ws:
                  continue  # Use case 1 or 2
              if (ts.who_has or set()) & self.manager.scheduler.running:
                  continue  # Use case 3 or 4
              # Use case 5
  
          # Either the worker holds the only replica or all replicas are being held
          # by workers that are being retired
          nrepl += 1
          # Don't create an unnecessary additional replica if another policy already
          # asked for one
          try:
              has_pending_repl = bool(self.manager.pending[ts][0])
          except KeyError:
              has_pending_repl = False
  
          if not has_pending_repl:
              rec_ws = yield Suggestion("replicate", ts)
              if not rec_ws:
                  # replication was rejected by the AMM (see _find_recipient)
                  nno_rec += 1
  
      if nno_rec:
          # All workers are paused or closing_gracefully.
          # Scheduler.retire_workers will read this flag and exit immediately.
          # TODO after we implement the automatic transition of workers from paused
          #      to closing_gracefully after a timeout expires, we should revisit this
          #      code to wait for paused workers and only exit immediately if all
          #      workers are in closing_gracefully status.
          self.no_recipients = True
          logger.warning(
              f"Tried retiring worker {self.address}, but {nno_rec} tasks could not "
              "be moved as there are no suitable workers to receive them. "
              "The worker will not be retired."
          )
          self.manager.policies.remove(self)
      elif nrepl:
          logger.info(
              f"Retiring worker {self.address}; {nrepl} keys are being moved away.",
          )
      else:
          logger.info(
              f"Retiring worker {self.address}; no unique keys need to be moved away."
          )
          self.manager.policies.remove(self)
  
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 3 for calling the RetireWorker obj's underlying member method code with example usage: obj.run() without return types