class_docstr: Accept, buffer, and write many small objects to many files
  
  This takes in lots of small objects, writes them to a local directory, and
  then reads them back when all writes are complete.  It buffers these
  objects in memory so that it can optimize disk access for larger writes.
  
  **State**
  
  -   shards: dict[str, list[bytes]]
  
      This is our in-memory buffer of data waiting to be written to files.
  
  -   sizes: dict[str, int]
  
      The size of each list of shards.  We find the largest and write data from that buffer
  
  Parameters
  ----------
  directory : str or pathlib.Path
      Where to write and read data.  Ideally points to fast disk.
  memory_limiter : ResourceLimiter
      Limiter for in-memory buffering (at most this much data)
      before writes to disk occur. If the incoming data that has yet
      to be processed exceeds this limit, then the buffer will block
      until below the threshold. See :meth:`.write` for the
      implementation of this scheme.
class_docstr_tok: ["Accept",",","buffer",",","and","write","many","small","objects","to","many","files","This","takes","in","lots","of","small","objects",",","writes","them","to","a","local","directory",",","and","then","reads","them","back","when","all","writes","are","complete",".","It","buffers","these","objects","in","memory","so","that","it","can","optimize","disk","access","for","larger","writes",".","*","*","State","*","*","-","shards",":","dict","[","str",",","list","[","bytes","]","]","This","is","our","in-memory","buffer","of","data","waiting","to","be","written","to","files",".","-","sizes",":","dict","[","str",",","int","]","The","size","of","each","list","of","shards",".","We","find","the","largest","and","write","data","from","that","buffer","Parameters","--","--","--","--","--","directory",":","str","or","pathlib.Path","Where","to","write","and","read","data",".","Ideally","points","to","fast","disk",".","memory_limiter",":","ResourceLimiter","Limiter","for","in-memory","buffering","(","at","most","this","much","data",")","before","writes","to","disk","occur",".","If","the","incoming","data","that","has","yet","to","be","processed","exceeds","this","limit",",","then","the","buffer","will","block","until","below","the","threshold",".","See",":","meth",":","`",".write","`","for","the","implementation","of","this","scheme","."]
docstr: Read a complete file back into memory
docstr_tok: ["Read","a","complete","file","back","into","memory"]
code: def read(self, id: str) -> Any:
      """Read a complete file back into memory"""
      self.raise_on_exception()
      if not self._inputs_done:
          raise RuntimeError("Tried to read from file before done.")
  
      try:
          with self._directory_lock.read():
              if self._closed:
                  raise RuntimeError("Already closed")
              fname = (self.directory / str(id)).resolve()
              # Note: don't add `with context_meter.meter("p2p-disk-read"):` to
              # measure seconds here, as it would shadow "p2p-get-output-cpu" and
              # "p2p-get-output-noncpu". Also, for rechunk it would not measure
              # the whole disk access, as _read returns memory-mapped buffers.
              data, size = self._read(fname)
              context_meter.digest_metric("p2p-disk-read", 1, "count")
              context_meter.digest_metric(
                  "p2p-disk-read", size, "bytes"
              )
      except FileNotFoundError:
          raise KeyError(id)
  
      if data:
          self.bytes_read += size
          return data
      else:
          raise KeyError(id)
  
imports: from __future__ import annotations
  import contextlib
  import pathlib
  import shutil
  import threading
  from collections.abc import Callable, Generator, Iterable
  from contextlib import contextmanager
  from typing import Any
  from toolz import concat
  from distributed.metrics import context_meter, thread_time
  from distributed.shuffle._buffer import ShardsBuffer
  from distributed.shuffle._limiter import ResourceLimiter
  from distributed.shuffle._pickle import pickle_bytelist
  from distributed.utils import Deadline, empty_context, log_errors, nbytes
label_desc: Use image node_id 2 for calling the DiskShardsBuffer obj's underlying member method code with example usage: obj.read(id) and returns: data