Merge pull request #6913 from ThomasWaldmann/prune-checkpointing-master

prune/delete --checkpoint-interval=1800 and ctrl-c/SIGINT support

src/borg/archiver/__init__.py

@@ -14,6 +14,7 @@ try:
     import os
     import shlex
     import signal
+    import time
     from datetime import datetime
 
     from ..logger import create_logger, setup_logging
@@ -119,6 +120,7 @@ class Archiver(
         self.exit_code = EXIT_SUCCESS
         self.lock_wait = lock_wait
         self.prog = prog
+        self.last_checkpoint = time.monotonic()
 
     def print_error(self, msg, *args):
         msg = args and msg % args or msg
@@ -433,6 +435,20 @@ class Archiver(
             logger.debug("Enabling debug topic %s", topic)
             logging.getLogger(topic).setLevel("DEBUG")
 
+    def maybe_checkpoint(self, *, checkpoint_func, checkpoint_interval):
+        checkpointed = False
+        sig_int_triggered = sig_int and sig_int.action_triggered()
+        if sig_int_triggered or checkpoint_interval and time.monotonic() - self.last_checkpoint > checkpoint_interval:
+            if sig_int_triggered:
+                logger.info("checkpoint requested: starting checkpoint creation...")
+            checkpoint_func()
+            checkpointed = True
+            self.last_checkpoint = time.monotonic()
+            if sig_int_triggered:
+                sig_int.action_completed()
+                logger.info("checkpoint requested: finished checkpoint creation!")
+        return checkpointed
+
     def run(self, args):
         os.umask(args.umask)  # early, before opening files
         self.lock_wait = args.lock_wait

src/borg/archiver/delete.py

@@ -5,7 +5,7 @@ from .common import with_repository
 from ..archive import Archive, Statistics
 from ..cache import Cache
 from ..constants import *  # NOQA
-from ..helpers import Manifest
+from ..helpers import Manifest, sig_int
 from ..helpers import log_multi, format_archive
 
 from ..logger import create_logger
@@ -57,11 +57,19 @@ class DeleteMixIn:
 
         stats = Statistics(iec=args.iec)
         with Cache(repository, key, manifest, progress=args.progress, lock_wait=self.lock_wait, iec=args.iec) as cache:
+
+            def checkpoint_func():
+                manifest.write()
+                repository.commit(compact=False, save_space=args.save_space)
+                cache.commit()
+
             msg_delete = "Would delete archive: {} ({}/{})" if dry_run else "Deleting archive: {} ({}/{})"
             msg_not_found = "Archive {} not found ({}/{})."
             logger_list = logging.getLogger("borg.output.list")
-            delete_count = 0
+            uncommitted_deletes = 0
             for i, archive_name in enumerate(archive_names, 1):
+                if sig_int and sig_int.action_done():
+                    break
                 try:
                     archive_info = manifest.archives[archive_name]
                 except KeyError:
@@ -80,12 +88,15 @@ class DeleteMixIn:
                             consider_part_files=args.consider_part_files,
                         )
                         archive.delete(stats, progress=args.progress, forced=args.forced)
-                        delete_count += 1
-            if delete_count > 0:
-                # only write/commit if we actually changed something, see #6060.
-                manifest.write()
-                repository.commit(compact=False, save_space=args.save_space)
-                cache.commit()
+                        checkpointed = self.maybe_checkpoint(
+                            checkpoint_func=checkpoint_func, checkpoint_interval=args.checkpoint_interval
+                        )
+                        uncommitted_deletes = 0 if checkpointed else (uncommitted_deletes + 1)
+            if sig_int:
+                # Ctrl-C / SIGINT: do not checkpoint (commit) again, we already have a checkpoint in this case.
+                self.print_error("Got Ctrl-C / SIGINT.")
+            elif uncommitted_deletes > 0:
+                checkpoint_func()
             if args.stats:
                 log_multi(str(stats), logger=logging.getLogger("borg.output.stats"))
 
@@ -160,4 +171,13 @@ class DeleteMixIn:
         subparser.add_argument(
             "--save-space", dest="save_space", action="store_true", help="work slower, but using less space"
         )
+        subparser.add_argument(
+            "-c",
+            "--checkpoint-interval",
+            metavar="SECONDS",
+            dest="checkpoint_interval",
+            type=int,
+            default=1800,
+            help="write checkpoint every SECONDS seconds (Default: 1800)",
+        )
         define_archive_filters_group(subparser)

src/borg/archiver/prune.py

@@ -8,7 +8,7 @@ from ..cache import Cache
 from ..constants import *  # NOQA
 from ..helpers import format_archive
 from ..helpers import interval, prune_within, prune_split, PRUNING_PATTERNS
-from ..helpers import Manifest
+from ..helpers import Manifest, sig_int
 from ..helpers import log_multi
 from ..helpers import ProgressIndicatorPercent
 
@@ -68,12 +68,21 @@ class PruneMixIn:
         to_delete = (set(archives) | checkpoints) - (set(keep) | set(keep_checkpoints))
         stats = Statistics(iec=args.iec)
         with Cache(repository, key, manifest, lock_wait=self.lock_wait, iec=args.iec) as cache:
+
+            def checkpoint_func():
+                manifest.write()
+                repository.commit(compact=False, save_space=args.save_space)
+                cache.commit()
+
             list_logger = logging.getLogger("borg.output.list")
             # set up counters for the progress display
             to_delete_len = len(to_delete)
             archives_deleted = 0
+            uncommitted_deletes = 0
             pi = ProgressIndicatorPercent(total=len(to_delete), msg="Pruning archives %3.0f%%", msgid="prune")
             for archive in archives_checkpoints:
+                if sig_int and sig_int.action_done():
+                    break
                 if archive in to_delete:
                     pi.show()
                     if args.dry_run:
@@ -85,6 +94,10 @@ class PruneMixIn:
                             repository, key, manifest, archive.name, cache, consider_part_files=args.consider_part_files
                         )
                         archive.delete(stats, forced=args.forced)
+                        checkpointed = self.maybe_checkpoint(
+                            checkpoint_func=checkpoint_func, checkpoint_interval=args.checkpoint_interval
+                        )
+                        uncommitted_deletes = 0 if checkpointed else (uncommitted_deletes + 1)
                 else:
                     if is_checkpoint(archive.name):
                         log_message = "Keeping checkpoint archive:"
@@ -97,10 +110,11 @@ class PruneMixIn:
                         "{message:<40} {archive}".format(message=log_message, archive=format_archive(archive))
                     )
             pi.finish()
-            if to_delete and not args.dry_run:
-                manifest.write()
-                repository.commit(compact=False, save_space=args.save_space)
-                cache.commit()
+            if sig_int:
+                # Ctrl-C / SIGINT: do not checkpoint (commit) again, we already have a checkpoint in this case.
+                self.print_error("Got Ctrl-C / SIGINT.")
+            elif uncommitted_deletes > 0:
+                checkpoint_func()
             if args.stats:
                 log_multi(str(stats), logger=logging.getLogger("borg.output.stats"))
         return self.exit_code
@@ -227,3 +241,12 @@ class PruneMixIn:
         subparser.add_argument(
             "--save-space", dest="save_space", action="store_true", help="work slower, but using less space"
         )
+        subparser.add_argument(
+            "-c",
+            "--checkpoint-interval",
+            metavar="SECONDS",
+            dest="checkpoint_interval",
+            type=int,
+            default=1800,
+            help="write checkpoint every SECONDS seconds (Default: 1800)",
+        )