diff --git a/src/borg/repository.py b/src/borg/repository.py index 254c094589..44a453f562 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -1,6 +1,7 @@ import io import os import sys +import threading import time from collections import defaultdict, namedtuple from pathlib import Path @@ -126,18 +127,38 @@ class PackWriter: """Buffers chunks into a pack file and writes it to the store when full. add() buffers a (chunk_id, cdata) pair and marks the chunk pending (F_PENDING); - flush() writes the pack and sets each entry's pack_id, obj_offset and obj_size, - clearing F_PENDING. + when the pack is full, it is built, hashed and stored, and each entry's pack_id, + obj_offset and obj_size are set, clearing F_PENDING. + + With async_store (the default), a full pack is handed to a background store-thread + (at most one in flight), so the caller can assemble the next pack while the previous + one is hashed and stored (#9988). The ChunkIndex is only ever touched by the calling + thread: the store-thread's results (or error) are applied when it is joined, at the + next pack boundary or flush(). Consequently, add() returns the *previous* pack's + results while the current pack's store is in flight, and a store error surfaces one + pack later, from whichever add()/flush() call joins the store-thread. + + flush() is a barrier: it joins an in-flight store and writes the current buffer + synchronously, so afterwards nothing is buffered or in flight and no chunk written + through this writer is F_PENDING anymore. The ChunkIndex comes from the repository, or from an explicit chunks index when there is no repository (see the chunks property). max_count bounds how many chunks a pack holds; max_size bounds its byte size. - flush() fires when either limit is reached. Set a limit to None to disable it; + A pack is written when either limit is reached. Set a limit to None to disable it; at least one must be set, otherwise the pack buffer is unbounded. """ - def __init__(self, store, *, max_count=None, max_size=None, chunks=None, repository=None): + class _Outcome: + """What one pack store produced: filled in by _store_pieces, applied by _apply_outcome.""" + + def __init__(self, pending_ids): + self.pending_ids = pending_ids # chunk ids to drop from the index if the store fails + self.results = None # list of (chunk_id, pack_id, obj_offset, obj_size) on success + self.error = None # the store exception on failure + + def __init__(self, store, *, max_count=None, max_size=None, chunks=None, repository=None, async_store=True): if repository is None and chunks is None: raise ValueError("PackWriter requires either a repository or an explicit chunks index") if max_count is None and max_size is None: @@ -146,9 +167,13 @@ def __init__(self, store, *, max_count=None, max_size=None, chunks=None, reposit self.max_count = max_count # None = no count limit self.max_size = max_size # None = no size limit self.repository = repository + self.async_store = async_store + # BORG_PACK_TRACE=yes prints store-thread lifecycle markers to stderr, see _trace. + self.trace_store = os.environ.get("BORG_PACK_TRACE", "no") == "yes" self._chunks = chunks # used when there is no repository self._pieces = [] # list of (chunk_id, cdata) self._size = 0 # byte size of buffered pieces + self._inflight = None # (thread, outcome) of the pack store-thread, at most one in flight @property def chunks(self): @@ -159,57 +184,159 @@ def chunks(self): return self._chunks def add(self, chunk_id, cdata): - """Buffer a chunk. Returns flush results if the pack is now full, else None.""" + """Buffer a chunk. + + When the chunk fills the pack, the pack is written and the results of the + *previously* written pack (with async_store) or of this pack (without) are + returned as a list of (chunk_id, pack_id, obj_offset, obj_size) tuples. + Returns None when there is nothing to report. + """ self.chunks.add(chunk_id, 0) # size: plaintext chunk size, set by the cache layer self._pieces.append((chunk_id, cdata)) self._size += len(cdata) if (self.max_count is not None and len(self._pieces) >= self.max_count) or ( self.max_size is not None and self._size >= self.max_size ): + if self.async_store: + results = self.join_inflight() # apply the previous pack's store, or raise its error + self._handoff() # current pack -> background store-thread + return results return self.flush() return None - def flush(self): - """Write the current pack to the store. + def _take_pieces(self): + """Take the buffered pieces, leaving an empty buffer.""" + pieces, self._pieces, self._size = self._pieces, [], 0 + return pieces + + @staticmethod + def _trace(char, trace): + """Emit one lifecycle marker of the background store-thread to stderr: + < thread started, H hashing starts, S storing starts, > thread finished. + Only active with BORG_PACK_TRACE=yes (debugging aid: visualizes how pack + stores overlap with the assembly of the next pack, #9988).""" + if trace: + sys.stderr.write(char) + sys.stderr.flush() + + def _store_pieces(self, pieces, outcome, trace=False): + """Build, hash and store one pack; record the results or the error in *outcome*. + + Runs in the store-thread (async, trace=True) or inline in the calling thread + (sync/flush). Touches only the store (borgstore >= 0.6 serializes all Store + operations internally, see borgstore #206), never the ChunkIndex. + """ + self._trace("<", trace) + try: + # Build the pack bytes once by joining all pieces (avoids O(n^2) copies + # that incremental string concatenation would cause in Python). + pack_data = b"".join(cdata for _, cdata in pieces) + + # Name the pack by the SHA-256 of its bytes: the name commits to the stored content, + # so borgstore can verify and cache the file. + self._trace("H", trace) + pack_id = sha256(pack_data).digest() + + # Record (chunk_id, pack_id, obj_offset, obj_size) for every piece. + results = [] + offset = 0 + for chunk_id, cdata in pieces: + obj_size = len(cdata) + results.append((chunk_id, pack_id, offset, obj_size)) + offset += obj_size + + self._trace("S", trace) + self.store.store("packs/" + bin_to_hex(pack_id), pack_data) + except BaseException as exc: # incl. KeyboardInterrupt: it must not vanish with the thread + outcome.error = exc + else: + outcome.results = results + finally: + self._trace(">", trace) - Returns a list of (chunk_id, pack_id, obj_offset, obj_size) tuples -- - one entry per chunk that was written. Returns None if there was nothing - to flush. Always updates the ChunkIndex with the real pack_id and obj_offset. + def _apply_outcome(self, outcome): + """Apply one finished pack store to the ChunkIndex (calling thread only). + + On success, set the real pack locations (clearing F_PENDING) and return the results; + on failure, drop the failed pack's index entries and raise the store error. """ - if not self._pieces: - return None + if outcome.error is not None: + # the pack was not stored: drop the index entries for its chunks. + for chunk_id in outcome.pending_ids: + if chunk_id in self.chunks: # a chunk_id may appear more than once in this pack + del self.chunks[chunk_id] + raise outcome.error + self.chunks.update_pack_info(outcome.results) # set the real location and clear F_PENDING + return outcome.results + + def _handoff(self): + """Hand the buffered pieces to a background store-thread (at most one in flight).""" + assert self._inflight is None, "join_inflight() must run before handing off another pack" + pieces = self._take_pieces() + outcome = self._Outcome([chunk_id for chunk_id, _ in pieces]) + # daemon: normally irrelevant, because flush() and close() always join the thread + # (also while unwinding a Ctrl-C), so it is never still running at interpreter + # shutdown. it is a safety net for the pathological case of a store that hangs + # (e.g. a dead sftp/rest connection without timeout) on a path that never joins: + # exiting then beats hanging forever in threading._shutdown. losing an unjoined + # store costs nothing: its index entries are only applied at the join, and all + # backends write to a temp name + rename (or have the server verify a content + # hash), so an aborted store can leave garbage, but never a corrupt pack. + thread = threading.Thread( + target=self._store_pieces, + args=(pieces, outcome), + kwargs=dict(trace=self.trace_store), + name="borg-pack-store", + daemon=True, + ) + self._inflight = (thread, outcome) + thread.start() - # Build the pack bytes once by joining all pieces (avoids O(n^2) copies - # that incremental string concatenation would cause in Python). - pack_data = b"".join(cdata for _, cdata in self._pieces) + def _drop_buffered(self): + """Drop the buffered pieces and their (still pending) index entries. - # Name the pack by the SHA-256 of its bytes: the name commits to the stored content, - # so borgstore can verify and cache the file. - pack_id = sha256(pack_data).digest() + Called when a pack store failed: the caller is aborting, so chunks not yet handed + to the store die with it. Dropping their entries keeps the index free of F_PENDING + leftovers, like the sync store path does, so the close()-time index persist works. + """ + pieces = self._take_pieces() + for chunk_id, _ in pieces: + if chunk_id in self.chunks: # a chunk_id may appear more than once in the buffer + del self.chunks[chunk_id] - # Record (chunk_id, pack_id, obj_offset, obj_size) for every piece. - results = [] - offset = 0 - for chunk_id, cdata in self._pieces: - obj_size = len(cdata) - results.append((chunk_id, pack_id, offset, obj_size)) - offset += obj_size + def join_inflight(self): + """Wait for an in-flight pack store and apply it to the index. - key = "packs/" + bin_to_hex(pack_id) - pending_ids = [chunk_id for chunk_id, _ in self._pieces] + Returns its results, None when nothing was in flight. If the store failed, the + writer is emptied (see _drop_buffered) and the store error is raised. + """ + if self._inflight is None: + return None + thread, outcome = self._inflight + thread.join() + self._inflight = None try: - self.store.store(key, pack_data) - except Exception: - # the pack was not stored: drop the index entries for its chunks. - for chunk_id in pending_ids: - if chunk_id in self.chunks: # a chunk_id may appear more than once in this pack - del self.chunks[chunk_id] + return self._apply_outcome(outcome) + except BaseException: + self._drop_buffered() raise - finally: - self._pieces = [] # cleared on success and on failure - self._size = 0 - self.chunks.update_pack_info(results) # set the real location and clear F_PENDING - return results + + def flush(self): + """Write the current pack to the store. This is a barrier: any in-flight store + is joined first and the current buffer is written synchronously, so afterwards + no chunk written through this writer is F_PENDING anymore. + + Returns a list of (chunk_id, pack_id, obj_offset, obj_size) tuples covering + every chunk written by this flush (including a joined in-flight pack), or + None if there was nothing to do. + """ + results = self.join_inflight() or [] + if self._pieces: + pieces = self._take_pieces() + outcome = self._Outcome([chunk_id for chunk_id, _ in pieces]) + self._store_pieces(pieces, outcome) + results += self._apply_outcome(outcome) + return results or None class PackReader: @@ -550,6 +677,8 @@ def __init__( # permissions are not given to the (remote) backend here; they are enforced on the # server side by "borg serve --rest --permissions ...". backend = build_rest_backend(location) + # note: borgstore >= 0.6 Store serializes all its operations internally, so the + # PackWriter store-thread and the main thread can share it (borgstore #206). self.store = Store(backend=backend, config=ns_config, cache_url=cache_url) else: self.store = Store(url, config=ns_config, permissions=permissions, cache_url=cache_url) @@ -716,7 +845,11 @@ def open(self, *, exclusive, lock_wait=None, lock=True): max_size = int(max_size_env) else: max_size = None if max_count is not None else DEFAULT_PACK_MAX_SIZE - self._pack_writer = PackWriter(self.store, repository=self, max_count=max_count, max_size=max_size) + # BORG_PACK_ASYNC=no disables the background store-thread (debugging aid, see PackWriter). + async_store = os.environ.get("BORG_PACK_ASYNC", "yes") != "no" + self._pack_writer = PackWriter( + self.store, repository=self, max_count=max_count, max_size=max_size, async_store=async_store + ) self.opened = True @property @@ -774,6 +907,15 @@ def flush(self): def close(self): if self._pack_writer is not None: + try: + # normally a no-op: flush() is a barrier and runs before close(). when close() runs + # while unwinding an error, a pack store may still be in flight: join it, so a stored + # pack gets recorded in the index and a failed one gets its index entries dropped. + self._pack_writer.join_inflight() + except Exception as exc: + # do not raise: we are closing, probably unwinding an error already; raising here + # would just mask that original error. + logger.warning("pack store failed during close: %s", exc) assert not self._pack_writer._pieces, "PackWriter has unflushed chunks; call flush() before close()" # close() may run again after the store was already closed (idempotent close), so we can # only persist while the store is open. Persisting is also a no-op unless chunks were added @@ -934,7 +1076,7 @@ def list(self, limit=None, marker=None): result = [] for chunk_id, entry in self.chunks.iteritems(): if self.chunks.is_pending(chunk_id): - continue # buffered in PackWriter, not flushed to a pack yet + continue # buffered in PackWriter (or its store still in flight), not read-able yet if collect: result.append((chunk_id, entry.obj_size)) if len(result) == limit: @@ -951,8 +1093,14 @@ def get(self, id, read_data=True, raise_missing=True): raise self.ObjectNotFound(id, str(self._location)) return None if self.chunks.is_pending(id): - # buffered but not flushed; a chunk must be flushed before any read, so this is a code - # bug (wrong flush/index ordering), not a missing object: raise regardless of raise_missing. + # the chunk may be in a pack whose background store is still in flight: + # join it, which resolves the chunk's pack location (read barrier). + self._pack_writer.join_inflight() + entry = self.chunks.get(id) # re-fetch, the join updated the entry + if entry is None or self.chunks.is_pending(id): + # still pending: buffered but not flushed; a chunk must be flushed before any read, so this + # is a code bug (wrong flush/index ordering), not a missing object: raise regardless of + # raise_missing. entry None: the join failed sometime earlier and dropped the entry. raise self.PackLocationUnknown(id, str(self._location)) pack_id, obj_offset, obj_size = entry.pack_id, entry.obj_offset, entry.obj_size id_hex = bin_to_hex(id) @@ -1033,8 +1181,10 @@ def put(self, id, data): """put a repo object Buffers the chunk in the pack writer. When the chunk fills the pack and - triggers a flush, returns a list of (chunk_id, pack_id, obj_offset, obj_size) - tuples, one per chunk written to disk by that flush; otherwise returns None. + triggers a pack write, returns a list of (chunk_id, pack_id, obj_offset, obj_size) + tuples, one per written chunk; otherwise returns None. With the background + store-thread (see PackWriter), the returned tuples are those of the *previous* + pack, whose store was joined before handing off the current one. """ self._lock_refresh() data_size = len(data) diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 1e6d7cfd1e..ba9bc7e06c 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -211,7 +211,9 @@ def test_multi_object_pack_roundtrip(repo_fixtures, request): repository._pack_writer.max_count = 2 # this test is written for exactly two objects per pack repository.put(H(0), chunk0) assert repository.chunks.is_pending(H(0)) # buffered: the pack is not full yet - repository.put(H(1), chunk1) # fills the pack, flushing both objects at once + repository.put(H(1), chunk1) # fills the pack, writing both objects at once + # the full pack went to a background store-thread; join it so the entries are resolved + repository._pack_writer.join_inflight() # both objects share one pack, written exactly once, laid out in put() order pack_id = repository.chunks[H(0)].pack_id assert not repository.chunks.is_pending(H(0)) @@ -761,7 +763,7 @@ def test_pack_writer_n1_flush(): store = MockStore() chunk_id = b"c" * 32 cdata = b"payload" - pw = PackWriter(store, max_count=1, chunks=ChunkIndex()) + pw = PackWriter(store, max_count=1, chunks=ChunkIndex(), async_store=False) results = pw.add(chunk_id, cdata) assert results is not None assert len(results) == 1 @@ -776,7 +778,7 @@ def test_pack_writer_n2_flush(): store = MockStore() id1, id2 = b"a" * 32, b"b" * 32 data1, data2 = b"first", b"second" - pw = PackWriter(store, max_count=2, chunks=ChunkIndex()) + pw = PackWriter(store, max_count=2, chunks=ChunkIndex(), async_store=False) assert pw.add(id1, data1) is None results = pw.add(id2, data2) assert results is not None @@ -790,7 +792,7 @@ def test_pack_writer_n2_flush(): def test_pack_writer_flushes_on_max_size(): # max_count is high, so the flush is driven by max_size alone. store = MockStore() - pw = PackWriter(store, max_count=100, max_size=10, chunks=ChunkIndex()) + pw = PackWriter(store, max_count=100, max_size=10, chunks=ChunkIndex(), async_store=False) assert pw.add(b"a" * 32, b"12345") is None results = pw.add(b"b" * 32, b"67890") assert results is not None @@ -799,14 +801,14 @@ def test_pack_writer_flushes_on_max_size(): def test_pack_writer_max_size_none_is_count_only(): store = MockStore() - pw = PackWriter(store, max_count=2, max_size=None, chunks=ChunkIndex()) + pw = PackWriter(store, max_count=2, max_size=None, chunks=ChunkIndex(), async_store=False) assert pw.add(b"a" * 32, b"x" * 10_000) is None assert pw.add(b"b" * 32, b"y" * 10_000) is not None def test_pack_writer_max_count_none_is_size_only(): store = MockStore() - pw = PackWriter(store, max_count=None, max_size=10, chunks=ChunkIndex()) + pw = PackWriter(store, max_count=None, max_size=10, chunks=ChunkIndex(), async_store=False) assert pw.add(b"a" * 32, b"12345") is None assert pw.add(b"b" * 32, b"67890") is not None @@ -822,7 +824,7 @@ def test_pack_writer_rolls_back_index_on_failed_store(): # later identical chunk would dedup against -- silent data loss (#9744 review). chunks = ChunkIndex() chunk_id = b"e" * 32 - pw = PackWriter(FailingPackStore(MockStore()), max_count=1, chunks=chunks) + pw = PackWriter(FailingPackStore(MockStore()), max_count=1, chunks=chunks, async_store=False) with pytest.raises(OSError): pw.add(chunk_id, b"payload") # max_count=1 -> add() flushes immediately and fails assert chunks.get(chunk_id) is None # rolled back: no phantom entry left behind @@ -839,7 +841,7 @@ def test_failed_store_phantom_not_persisted(tmp_path): # so one store models "just the pack write broke" (PackWriter and the index share a # store in production). the failing store is thus load-bearing for every assertion below. repository.store = FailingPackStore(repository.store) - pw = PackWriter(repository.store, max_count=1, repository=repository) + pw = PackWriter(repository.store, max_count=1, repository=repository, async_store=False) with pytest.raises(OSError): pw.add(chunk_id, fchunk(b"DATA")) assert repository.chunks.get(chunk_id) is None # rolled back from the in-memory index ... @@ -849,6 +851,63 @@ def test_failed_store_phantom_not_persisted(tmp_path): assert reloaded.get(chunk_id) is None +def test_pack_writer_async_defers_results(): + # With async_store (the default), a full pack goes to a background store-thread and add() + # returns the *previous* pack's results; flush() is a barrier returning whatever is left (#9988). + store = MockStore() + id1, id2, id3 = b"a" * 32, b"b" * 32, b"c" * 32 + data1, data2, data3 = b"first", b"second", b"third" + chunks = ChunkIndex() + pw = PackWriter(store, max_count=1, chunks=chunks) + assert pw.add(id1, data1) is None # pack 1 handed to the store-thread, nothing to report yet + results = pw.add(id2, data2) # joins pack 1's store, hands off pack 2 + assert results == [(id1, sha256(data1).digest(), 0, len(data1))] + assert not chunks.is_pending(id1) # the join resolved pack 1's entries + assert pw.add(id3, data3) == [(id2, sha256(data2).digest(), 0, len(data2))] + assert pw.flush() == [(id3, sha256(data3).digest(), 0, len(data3))] # barrier: joins pack 3 + for chunk_id in (id1, id2, id3): + assert not chunks.is_pending(chunk_id) + + +def test_pack_writer_async_flush_combines_inflight_and_buffered(): + # flush() with a store in flight *and* buffered pieces returns the results of both packs. + store = MockStore() + id1, id2 = b"a" * 32, b"b" * 32 + pw = PackWriter(store, max_count=1, chunks=ChunkIndex()) + assert pw.add(id1, b"first") is None # in flight + pw.max_count = 2 # keep the next piece buffered + assert pw.add(id2, b"second") is None # buffered + results = pw.flush() + assert [chunk_id for chunk_id, _, _, _ in results] == [id1, id2] + + +def test_pack_writer_async_error_surfaces_at_join(): + # A background store failure surfaces at the next join (here: the add() that fills the next + # pack), and the writer rolls back the failed pack's index entries *and* drops the buffered + # pieces, so no phantom/pending entries stay behind for the close()-time index persist (#9744). + chunks = ChunkIndex() + id1, id2 = b"e" * 32, b"f" * 32 + pw = PackWriter(FailingPackStore(MockStore()), max_count=1, chunks=chunks) + assert pw.add(id1, b"payload") is None # handed off; the failure is not visible yet + with pytest.raises(OSError): + pw.add(id2, b"moredata") # joins the failed store of pack 1 + assert chunks.get(id1) is None # rolled back: no phantom entry left behind + assert chunks.get(id2) is None # buffered piece dropped along with the aborting command + assert not pw._pieces # writer is empty, close() will not trip over leftovers + + +def test_pack_writer_async_get_waits_for_inflight_store(tmp_path): + # get() of a chunk whose pack store is still in flight must join the store-thread and + # then read normally (read barrier), instead of raising PackLocationUnknown. + with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository: + repository._pack_writer.max_count = 2 + chunk0, chunk1 = fchunk(b"foo", chunk_id=H(0)), fchunk(b"bar", chunk_id=H(1)) + repository.put(H(0), chunk0) + repository.put(H(1), chunk1) # fills the pack -> handed to the store-thread + assert repository.get(H(0)) == chunk0 # waits for the store-thread, then reads + assert repository.get(H(1)) == chunk1 + + def test_get_read_data_false_with_range(tmp_path): # read_data=False with ChunkIndex entries limits the load to each object's boundary. hdr_size = RepoObj.obj_header.size