class_docstr: Make sure that in-memory tasks are not replicated on more workers than desired;
  drop the excess replicas.
class_docstr_tok: ["Make","sure","that","in-memory","tasks","are","not","replicated","on","more","workers","than","desired",";","drop","the","excess","replicas","."]
code: def run(self) -> SuggestionGenerator:
      nkeys = 0
      ndrop = 0
  
      for ts in self.manager.scheduler.replicated_tasks:
          desired_replicas = 1  # TODO have a marker on TaskState
          assert ts.who_has
  
          nwaiters = len(ts.waiters or ())
          if desired_replicas < nwaiters < 20:
              # If a dependent task has not been assigned to a worker yet, err on the
              # side of caution and preserve an additional replica for it.
              # However, if two dependent tasks have been already assigned to the same
              # worker, don't double count them.
              # This calculation is quite CPU-intensive, so it's disabled for tasks
              # with lots of waiters.
              nwaiters = len(
                  {
                      waiter.processing_on or waiter
                      for waiter in ts.waiters or ()
                  }
              )
  
          ndrop_key = len(ts.who_has) - max(desired_replicas, nwaiters)
          if ts in self.manager.pending:
              pending_repl, pending_drop = self.manager.pending[ts]
              ndrop_key += len(pending_repl) - len(pending_drop)
  
          if ndrop_key > 0:
              nkeys += 1
              ndrop += ndrop_key
              for _ in range(ndrop_key):
                  yield Suggestion("drop", ts)
  
      if ndrop:
          logger.debug(
              "ReduceReplicas: Dropping %d superfluous replicas of %d tasks",
              ndrop,
              nkeys,
          )
  
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 for calling the ReduceReplicas obj's underlying member method code with example usage: obj.run() without return types