class_docstr: Inanely suggest to drop every single key in the cluster
code: def run(self):
      for ts in self.manager.scheduler.tasks.values():
          # Instead of yielding ("drop", ts, None) for each worker, which would result
          # in semi-predictable output about which replica survives, randomly choose a
          # different survivor at each AMM run.
          candidates = list(ts.who_has)
          random.shuffle(candidates)
          for ws in candidates:
              yield "drop", ts, {ws}
  
      # Stop running after ~2s
      self.i += 1
      if self.i == 20:
          self.manager.policies.remove(self)
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import random
  import warnings
  from collections.abc import Iterator
  from contextlib import contextmanager
  from typing import Any, Literal
  import pytest
  import dask.config
  from distributed import Event, Lock, Scheduler, wait
  from distributed.active_memory_manager import ActiveMemoryManagerExtension, ActiveMemoryManagerPolicy, RetireWorker
  from distributed.core import Status
  from distributed.utils_test import NO_AMM, BlockedGatherDep, assert_story, async_poll_for, captured_logger, gen_cluster, gen_test, inc, lock_inc, slowinc, wait_for_state
  from distributed.worker_state_machine import AcquireReplicasEvent
label_desc: Use image node_id 2 for calling the DropEverything obj's underlying member method code with example usage: obj.run() without return types