class_docstr: MutableMapping that automatically spills out dask key/value pairs to disk when
  the total size of the stored data exceeds the target. If max_spill is provided the
  key/value pairs won't be spilled once this threshold has been reached.
  
  Parameters
  ----------
  spill_directory: str
      Location on disk to write the spill files to
  target: int
      Managed memory, in bytes, to start spilling at
  max_spill: int | False, optional
      Limit of number of bytes to be spilled on disk. Set to False to disable.
class_docstr_tok: ["MutableMapping","that","automatically","spills","out","dask","key\/value","pairs","to","disk","when","the","total","size","of","the","stored","data","exceeds","the","target",".","If","max_spill","is","provided","the","key\/value","pairs","wo","n't","be","spilled","once","this","threshold","has","been","reached",".","Parameters","--","--","--","--","--","spill_directory",":","str","Location","on","disk","to","write","the","spill","files","to","target",":","int","Managed","memory",",","in","bytes",",","to","start","spilling","at","max_spill",":","int","|","False",",","optional","Limit","of","number","of","bytes","to","be","spilled","on","disk",".","Set","to","False","to","disable","."]
docstr: Capture metrics re. disk read/write, serialize/deserialize, and
  compress/decompress.
  
  Note that this duplicates capturing from gather_dep, get_data, and execute. It
  is repeated here to make it possible to split serialize/deserialize and
  compress/decompress triggered by spill/unspill from those triggered by network
  comms.
code: def _capture_metrics(self) -> Iterator[None]:
      """Capture metrics re. disk read/write, serialize/deserialize, and
      compress/decompress.
  
      Note that this duplicates capturing from gather_dep, get_data, and execute. It
      is repeated here to make it possible to split serialize/deserialize and
      compress/decompress triggered by spill/unspill from those triggered by network
      comms.
      """
  
      def metrics_callback(
          label: Hashable, value: float, unit: str
      ) -> None:
          assert isinstance(label, str)
          self.cumulative_metrics[label, unit] += value
  
      with context_meter.add_callback(metrics_callback):
          yield
  
imports: from __future__ import annotations
  import logging
  from collections import defaultdict
  from collections.abc import Callable, Hashable, Iterator, Mapping, MutableMapping, Sized
  from contextlib import contextmanager
  from functools import partial
  from typing import Literal, NamedTuple, Protocol, cast
  import zict
  from dask.typing import Key
  from distributed.metrics import context_meter
  from distributed.protocol import deserialize_bytes, serialize_bytelist
  from distributed.protocol.compression import get_compression_settings
  from distributed.sizeof import safe_sizeof
  from distributed.utils import RateLimiterFilter, nbytes
label_desc: Use image node_id 2 for calling the SpillBuffer obj's underlying member method code with example usage: obj._capture_metrics() without return types