Method get_stats

salt/utils/mmap_cache.py:1562–1612

Return statistics about the cache. ``occupied`` and ``deleted`` are read from the header in O(1). ``heap_live_bytes`` is computed by iterating the roster — O(occupied). Keys: ``occupied``, ``deleted``, ``empty``, ``total``, ``load_factor``, ``heap_size_byte

(self)

            return ret

    def get_stats(self):
        """
        Return statistics about the cache.

        ``occupied`` and ``deleted`` are read from the header in O(1).
        ``heap_live_bytes`` is computed by iterating the roster — O(occupied).

        Keys: ``occupied``, ``deleted``, ``empty``, ``total``,
        ``load_factor``, ``heap_size_bytes``, ``heap_live_bytes``.
        """
        with self._thread_lock:
            if not self.open(write=False):
                return {
                    "occupied": 0,
                    "deleted": 0,
                    "empty": 0,
                    "total": self.size - 1,
                    "load_factor": 0.0,
                    "heap_size_bytes": 0,
                    "heap_live_bytes": 0,
                }

            occupied, deleted, _ = self._read_header()
            data_size = self.size - 1
            empty = data_size - occupied - deleted

            heap_live = 0
            for slot in self._roster_read():
                if slot == 0 or slot >= self.size:
                    continue
                offset = slot * self.slot_size
                if self._mm[offset] == OCCUPIED:
                    _, length, _ = self._read_slot_pointer(offset)
                    heap_live += length

            try:
                heap_size = os.path.getsize(self.heap_path)
            except OSError:
                heap_size = 0

            return {
                "occupied": occupied,
                "deleted": deleted,
                "empty": empty,
                "total": data_size,
                "load_factor": (
                    (occupied + deleted) / data_size if data_size > 0 else 0.0
                ),
                "heap_size_bytes": heap_size,
                "heap_live_bytes": heap_live,
            }

    def atomic_rebuild(self, iterator):
        """