Method atomic_rebuild

salt/utils/mmap_cache.py:1614–1850

Rebuild the cache from an iterator of ``(key, value)`` or ``(key,)`` tuples. Writes fresh temporary index, heap, and roster files, then atomically swaps all three. Roster is swapped last so readers never see a new roster pointing into an old index.

(self, iterator)

            }

    def atomic_rebuild(self, iterator):
        """
        Rebuild the cache from an iterator of ``(key, value)`` or ``(key,)``
        tuples.

        Writes fresh temporary index, heap, and roster files, then atomically
        swaps all three.  Roster is swapped last so readers never see a new
        roster pointing into an old index.
        """
        os.makedirs(os.path.dirname(self.path), exist_ok=True)

        tmp_dir = os.path.dirname(self.path)
        tmp_idx_fd, tmp_idx_path = tempfile.mkstemp(dir=tmp_dir, prefix=".mmcache_idx_")
        # Rebuild always starts with a single segment; further rolls happen
        # naturally if max_segment_bytes is exceeded during the write loop.
        tmp_heap_base = tmp_idx_path + ".heap"
        tmp_roster_path = tmp_idx_path + ".roster"
        # Used in ``except`` cleanup; must exist even if we fail before the
        # inner loop assigns the working segment counter.
        tmp_seg_id = 0

        try:
            with self._thread_lock:
                with self._lock():
                    total_idx = self.size * self.slot_size
                    chunk = 1024 * 1024
                    zeros = b"\x00" * min(chunk, total_idx)
                    with os.fdopen(tmp_idx_fd, "wb") as f:
                        written = 0
                        while written < total_idx:
                            to_write = min(chunk, total_idx - written)
                            f.write(zeros[:to_write])
                            written += to_write
                        f.flush()
                        os.fsync(f.fileno())
                    tmp_idx_fd = -1

                    # seg_id and seg_pos track the current temporary segment.
                    tmp_seg_id = 0
                    tmp_seg_pos = 0  # bytes written to current tmp segment
                    tmp_heap_segs = [tmp_heap_base]  # paths of tmp heap segments
                    occupied_count = 0
                    hwm = 0
                    roster_entries = []

                    heap_f = salt.utils.files.fopen(  # pylint: disable=resource-leakage
                        tmp_heap_base, "wb"
                    )
                    try:
                        with salt.utils.files.fopen(tmp_idx_path, "r+b") as idx_f:
                            mm = mmap.mmap(idx_f.fileno(), 0, access=mmap.ACCESS_WRITE)
                            try:
                                mm[0] = _HEADER_MAGIC

                                for item in iterator:
                                    if (
                                        isinstance(item, (list, tuple))
                                        and len(item) > 1