multithreaded "create" operation

Making much better use of the CPU by dispatching all CPU intensive stuff
(hashing, crypto, compression) to N crypter threads (N == logical cpu count ==
4 for a dual-core CPU with hyperthreading).

I/O intensive stuff also runs in separate threads: the MainThread does the
filesystem traversal, the reader thread reads and chunks the files, the writer
thread writes to the repo. This way, we don't need to sit idle waiting for I/O,
but the I/O thread will block and another thread will get dispatched and use
the time. This applies for read as well as for write/fsync I/O wait time
(access time + data transfer).

There's one more thread, the "delayer". We need it to handle a race condition
related to the computation of the compressed size (which is only possible after
hashing/compression/encryption has finished). This "csize" makes all this code
quite more complicated than if we would not need it.

Although there is the GIL issue for Python code, we can still make good use of
multithreading as I/O operations and C code (that releases the GIL) can run in
parallel.

All threads are connected via Python Queues (which are intended for this and
thread safe). The Cache.chunks datastructure is also updated by threadsafe
code.

A little benchmark
------------------

Both is with compression (zlib level 6) and encryption on a haswell/ssd laptop:

Without multithreading code:

    Command being timed: "borg create /extra/attic/borg::1 /home/tw/Desktop/"
    User time (seconds): 13.78
    System time (seconds): 0.40
    Percent of CPU this job got: 83%
    Elapsed (wall clock) time (h:mm:ss or m:ss): 0:16.98

With multithreading code:

    Command being timed: "borg create /extra/attic/borg::1 /home/tw/Desktop/"
    User time (seconds): 24.08
    System time (seconds): 1.16
    Percent of CPU this job got: 249%
    Elapsed (wall clock) time (h:mm:ss or m:ss): 0:10.11

It's unclear to me why it uses much more "User time" (I'm not even sure that
measurement is correct). But the overall runtime "Elapsed" significantly
dropped and it makes better use of all cpu cores (not just 83% of one).

borg/archive.py

@@ -4,9 +4,11 @@ from itertools import groupby
 import errno
 import shutil
 import tempfile
+import threading
 from .key import key_factory
 from .remote import cache_if_remote
 import msgpack
+from multiprocessing import cpu_count
 import os
 import socket
 import stat
@@ -18,7 +20,8 @@ from .platform import acl_get, acl_set
 from .chunker import Chunker
 from .hashindex import ChunkIndex
 from .helpers import parse_timestamp, Error, uid2user, user2uid, gid2group, group2gid, \
-    Manifest, Statistics, decode_dict, st_mtime_ns, make_path_safe, StableDict, int_to_bigint, bigint_to_int
+    Manifest, Statistics, decode_dict, st_mtime_ns, make_path_safe, StableDict, int_to_bigint, bigint_to_int, \
+    make_queue, TerminatedQueue
 
 ITEMS_BUFFER = 1024 * 1024
 CHUNK_MIN = 1024
@@ -114,6 +117,193 @@ class CacheChunkBuffer(ChunkBuffer):
         return id_
 
 
+class ParallelProcessor:
+    def __init__(self, archive, ncrypters=None):
+        self.archive = archive
+        if ncrypters is None:
+            # note: cpu_count for 2 cores with HT is 4
+            # put load on all logical cores and avoid idle cores
+            ncrypters = cpu_count()
+        self.ncrypters = ncrypters
+        self.start_threads()
+
+    def reader(self):
+        while True:
+            elem = self.reader_queue.get()
+            if elem is None:
+                self.reader_queue.task_done()
+                break
+            item = elem
+            n = 0
+            # Only chunkify the file if needed
+            if b'chunks' in item and item[b'chunks'] is None:
+                fd, fh = item.pop(b'fd', None), -1
+                if fd is None:
+                    fh = Archive._open_rb(item.pop(b'path_name'), item[b'st'])
+                    fd = os.fdopen(fh, 'rb')
+                with fd:
+                    for chunk in self.archive.chunker.chunkify(fd, fh):
+                        # important: chunk is a memoryview - make a copy or it will
+                        # have changed when we use it!
+                        chunk = bytes(chunk)
+                        self.crypter_queue.put((item, n, chunk))
+                        n += 1
+            self.writer_queue.put((item, n, None, None, None, None))  # signal EOF via id == None , give number of chunks
+            self.reader_queue.task_done()
+
+    def crypter(self):
+        while True:
+            elem = self.crypter_queue.get()
+            if elem is None:
+                self.crypter_queue.task_done()
+                break
+            item, n, chunk = elem
+            size = len(chunk)
+            id = self.archive.key.id_hash(chunk)
+            seen = self.archive.cache.seen_or_announce_chunk(id, size)
+            if not seen:
+                # we have never seen this id before, so we need to process it
+                # TODO check if this creates duplicate IV/CTR values for AES
+                cchunk = self.archive.key.encrypt(chunk)
+                csize = len(cchunk)
+            else:
+                cchunk, csize = None, None
+            self.writer_queue.put((item, n, cchunk, id, size, csize))
+            self.crypter_queue.task_done()
+
+    def writer(self):
+        item_infos = {}  # item path -> info dict
+        size_infos = {}  # chunk id -> sizes
+        dying = False
+        while True:
+            elem = self.writer_queue.get()
+            if elem is None:
+                if not dying:
+                    # received poison from stop_threads, start dying,
+                    # but still do work the delayer thread might give us.
+                    dying = True
+                    # give poison to the delayer thread
+                    self.delayer_queue.put(None)
+                    self.writer_queue.task_done()
+                    continue
+                else:
+                    # we received the final poison from the dying delayer
+                    self.writer_queue.task_done()
+                    # we are dead now
+                    break
+            item, n, cchunk, id, size, csize = elem
+            path = item[b'path']
+            info = item_infos.setdefault(path, dict(count=None, chunks=[]))
+            if id is None:
+                if n is not None:  # note: n == None is a retry
+                    # EOF signalled, n is the total count of chunks
+                    info['count'] = n
+            else:
+                size, csize, new_chunk = self.archive.cache.add_chunk_nostats(cchunk, id, size, csize)
+                info['chunks'].append((n, id, new_chunk))
+                if csize != 0:
+                    size_infos[id] = (size, csize)
+            if len(info['chunks']) == info['count']:
+                # we have processed all chunks or no chunks needed processing
+                if b'chunks' in item:
+                    chunks = item[b'chunks']
+                    if chunks is None:
+                        # we want chunks, but we have no chunk id list yet, compute them
+                        try:
+                            chunks = self.archive.cache.postprocess_results(
+                                size_infos, info['chunks'], self.archive.stats)
+                        except self.archive.cache.ChunkSizeNotReady:
+                            # we looked up a chunk id, but do not have the size info yet. retry later.
+                            self.delayer_queue.put((item, None, None, None, None, None))
+                            self.writer_queue.task_done()
+                            continue
+                    else:
+                        # we have a chunk id list already, increase the ref counters, compute sizes
+                        chunks = [self.archive.cache.chunk_incref(id_, self.archive.stats) for id_ in chunks]
+                    item[b'chunks'] = chunks
+                path_hash = item.pop(b'path_hash', None)
+                if path_hash and chunks is not None:  # a fs object (not stdin) and a regular file
+                    st = item.pop(b'st', None)
+                    self.archive.cache.memorize_file(path_hash, st, [c[0] for c in chunks])
+                del item_infos[path]
+                self.archive.stats.nfiles += 1
+                self.archive.add_item(item)
+            self.writer_queue.task_done()
+
+    def delayer(self):
+        # it is a pain that we need the compressed size for the chunks cache as it is not
+        # available for duplicate chunks until the original chunk has finished processing.
+        # this loop of (writer, delayer) with pipes connecting them is a hack to address
+        # this, but it makes thread teardown complicated. Rather get rid of csize?
+        while True:
+            elem = self.delayer_queue.get()
+            if elem is None:
+                # we received poison from dying writer thread, kill the writer, too.
+                self.writer_queue.put(None)
+                self.delayer_queue.task_done()
+                # we are dead now
+                break
+            time.sleep(0.001)  # reschedule, avoid data circulating too fast
+            self.writer_queue.put(elem)
+            self.delayer_queue.task_done()
+
+    def start_threads(self):
+        def run_thread(func, name=None, daemon=False):
+            t = threading.Thread(target=func, name=name)
+            t.daemon = daemon
+            t.start()
+            return t
+
+        # max. memory usage of a queue with chunk data is about queue_len * CHUNK_MAX
+        queue_len = min(max(self.ncrypters, 4), 8)
+        self.reader_queue = make_queue('reader', queue_len * 10)  # small items (no chunk data)
+        self.crypter_queue = make_queue('crypter', queue_len)
+        self.writer_queue = make_queue('writer', queue_len)
+        self.delayer_queue = make_queue('delay', queue_len)
+        self.reader_thread = run_thread(self.reader, 'reader')
+        self.crypter_threads = []
+        for i in range(self.ncrypters):
+            self.crypter_threads.append(run_thread(self.crypter, name='crypter-%d' % i))
+        self.delayer_thread = run_thread(self.delayer, name='delayer')
+        self.writer_thread = run_thread(self.writer, name='writer')
+
+    def wait_finish(self):
+        self.reader_queue.join()
+        self.crypter_queue.join()
+        self.writer_queue.join()
+        self.delayer_queue.join()
+        self.writer_queue.join()
+
+    def stop_threads(self):
+        count_before = threading.active_count()
+        # for every thread:
+        #   put poison pill into its queue,
+        #   wait until queue is processed (and thread has terminated itself)
+        #   make queue unusable
+        self.reader_queue.put(None)
+        self.reader_queue.join()
+        self.reader_thread.join()
+        self.reader_queue = TerminatedQueue()
+        for i in range(self.ncrypters):
+            self.crypter_queue.put(None)
+        self.crypter_queue.join()
+        for t in self.crypter_threads:
+            t.join()
+        self.crypter_queue = TerminatedQueue()
+        self.writer_queue.put(None)  # the writer will poison the delayer first
+        self.delayer_thread.join()
+        self.delayer_queue = TerminatedQueue()
+        self.writer_thread.join()
+        self.writer_queue = TerminatedQueue()
+        count_after = threading.active_count()
+        assert count_before - 3 - self.ncrypters == count_after
+        if count_after > 1:
+            print('They are alive!')
+            tl = [t.name for t in threading.enumerate()]
+            tl.remove('MainThread')
+            assert tl == []
+
+
 class Archive:
 
     class DoesNotExist(Error):
@@ -142,6 +332,7 @@ class Archive:
         self.numeric_owner = numeric_owner
         self.pipeline = DownloadPipeline(self.repository, self.key)
         if create:
+            self.pp = ParallelProcessor(self)
             self.items_buffer = CacheChunkBuffer(self.cache, self.key, self.stats)
             self.chunker = Chunker(WINDOW_SIZE, CHUNK_MASK, CHUNK_MIN, CHUNK_MAX, self.key.chunk_seed)
             if name in manifest.archives:
@@ -154,11 +345,16 @@ class Archive:
                     break
                 i += 1
         else:
+            self.pp = None
             if name not in self.manifest.archives:
                 raise self.DoesNotExist(name)
             info = self.manifest.archives[name]
             self.load(info[b'id'])
 
+    def close(self):
+        if self.pp:
+            self.pp.stop_threads()
+
     def _load_meta(self, id):
         data = self.key.decrypt(id, self.repository.get(id))
         metadata = msgpack.unpackb(data)
@@ -185,6 +381,9 @@ class Archive:
         for item in self.pipeline.unpack_many(self.metadata[b'items'], filter=filter, preload=preload):
             yield item
 
+    def add_item_queued(self, item):
+        self.pp.reader_queue.put(item)
+
     def add_item(self, item):
         if self.show_progress and time.time() - self.last_progress > 0.2:
             self.stats.show_progress(item=item)
@@ -200,6 +399,7 @@ class Archive:
         self.cache.chunk_decref(self.id, self.stats)
 
     def save(self, name=None, timestamp=None):
+        self.pp.wait_finish()
         name = name or self.name
         if name in self.manifest.archives:
             raise self.AlreadyExists(name)
@@ -267,7 +467,9 @@ class Archive:
             if stat.S_ISDIR(st.st_mode):
                 os.rmdir(path)
             else:
-                os.unlink(path)
+                # XXX do not remove a regular file, it could be the "source"
+                # of a hardlink - a still empty inode that needs to be filled.
+                pass
         except UnicodeEncodeError:
             raise self.IncompatibleFilesystemEncodingError(path, sys.getfilesystemencoding())
         except OSError:
@@ -286,6 +488,14 @@ class Archive:
                 source = os.path.join(dest, item[b'source'])
                 if os.path.exists(path):
                     os.unlink(path)
+                if not os.path.exists(source):
+                    # due to multithreaded nature and different processing time,
+                    # the hardlink (without file content) often is in the archive
+                    # BEFORE the "source" file (with content).
+                    # we create an empty file that is filled with content when
+                    # the "source" item is extracted:
+                    with open(source, 'wb') as fd:
+                        pass
                 os.link(source, path)
             else:
                 with open(path, 'wb') as fd:
@@ -408,19 +618,19 @@ class Archive:
     def process_dir(self, path, st):
         item = {b'path': make_path_safe(path)}
         item.update(self.stat_attrs(st, path))
-        self.add_item(item)
+        self.add_item_queued(item)
         return 'd'  # directory
 
     def process_fifo(self, path, st):
         item = {b'path': make_path_safe(path)}
         item.update(self.stat_attrs(st, path))
-        self.add_item(item)
+        self.add_item_queued(item)
         return 'f'  # fifo
 
     def process_dev(self, path, st):
         item = {b'path': make_path_safe(path), b'rdev': st.st_rdev}
         item.update(self.stat_attrs(st, path))
-        self.add_item(item)
+        self.add_item_queued(item)
         if stat.S_ISCHR(st.st_mode):
             return 'c'  # char device
         elif stat.S_ISBLK(st.st_mode):
@@ -430,25 +640,21 @@ class Archive:
         source = os.readlink(path)
         item = {b'path': make_path_safe(path), b'source': source}
         item.update(self.stat_attrs(st, path))
-        self.add_item(item)
+        self.add_item_queued(item)
         return 's'  # symlink
 
     def process_stdin(self, path, cache):
         uid, gid = 0, 0
-        fd = sys.stdin.buffer  # binary
-        chunks = []
-        for chunk in self.chunker.chunkify(fd):
-            chunks.append(cache.add_chunk(self.key.id_hash(chunk), chunk, self.stats))
-        self.stats.nfiles += 1
         item = {
             b'path': path,
-            b'chunks': chunks,
+            b'fd': sys.stdin.buffer,  # binary
             b'mode': 0o100660,  # regular file, ug=rw
             b'uid': uid, b'user': uid2user(uid),
             b'gid': gid, b'group': gid2group(gid),
             b'mtime': int_to_bigint(int(time.time()) * 1000000000)
         }
-        self.add_item(item)
+        self.add_item_queued(item)
+        return 'A'
 
     def process_file(self, path, st, cache):
         status = None
@@ -459,7 +665,7 @@ class Archive:
             if (st.st_ino, st.st_dev) in self.hard_links:
                 item = self.stat_attrs(st, path)
                 item.update({b'path': safe_path, b'source': source})
-                self.add_item(item)
+                self.add_item_queued(item)
                 status = 'h'  # regular file, hardlink (to already seen inodes)
                 return status
             else:
@@ -473,23 +679,23 @@ class Archive:
                 if not cache.seen_chunk(id_):
                     break
             else:
-                chunks = [cache.chunk_incref(id_, self.stats) for id_ in ids]
+                chunks = ids
                 status = 'U'  # regular file, unchanged
         else:
             status = 'A'  # regular file, added
-        # Only chunkify the file if needed
         if chunks is None:
-            fh = Archive._open_rb(path, st)
-            with os.fdopen(fh, 'rb') as fd:
-                chunks = []
-                for chunk in self.chunker.chunkify(fd, fh):
-                    chunks.append(cache.add_chunk(self.key.id_hash(chunk), chunk, self.stats))
-            cache.memorize_file(path_hash, st, [c[0] for c in chunks])
             status = status or 'M'  # regular file, modified (if not 'A' already)
-        item = {b'path': safe_path, b'chunks': chunks}
+
+        item = {
+            b'path': safe_path,
+            b'path_name': path,
+            b'path_hash': path_hash,
+            b'chunks': chunks,
+            b'st': st,
+        }
         item.update(self.stat_attrs(st, path))
-        self.stats.nfiles += 1
-        self.add_item(item)
+        self.add_item_queued(item)
+
         return status
 
     @staticmethod

borg/archiver.py

@@ -105,54 +105,57 @@ Type "Yes I am sure" if you understand this and want to continue.\n""")
         archive = Archive(repository, key, manifest, args.archive.archive, cache=cache,
                           create=True, checkpoint_interval=args.checkpoint_interval,
                           numeric_owner=args.numeric_owner, progress=args.progress)
-        # Add cache dir to inode_skip list
-        skip_inodes = set()
         try:
-            st = os.stat(get_cache_dir())
-            skip_inodes.add((st.st_ino, st.st_dev))
-        except IOError:
-            pass
-        # Add local repository dir to inode_skip list
-        if not args.archive.host:
+            # Add cache dir to inode_skip list
+            skip_inodes = set()
             try:
-                st = os.stat(args.archive.path)
+                st = os.stat(get_cache_dir())
                 skip_inodes.add((st.st_ino, st.st_dev))
             except IOError:
                 pass
-        for path in args.paths:
-            if path == '-':  # stdin
-                path = 'stdin'
-                self.print_verbose(path)
+            # Add local repository dir to inode_skip list
+            if not args.archive.host:
                 try:
-                    archive.process_stdin(path, cache)
-                except IOError as e:
-                    self.print_error('%s: %s', path, e)
-                continue
-            path = os.path.normpath(path)
-            if args.dontcross:
-                try:
-                    restrict_dev = os.lstat(path).st_dev
-                except OSError as e:
-                    self.print_error('%s: %s', path, e)
+                    st = os.stat(args.archive.path)
+                    skip_inodes.add((st.st_ino, st.st_dev))
+                except IOError:
+                    pass
+            for path in args.paths:
+                if path == '-':  # stdin
+                    path = 'stdin'
+                    self.print_verbose(path)
+                    try:
+                        archive.process_stdin(path, cache)
+                    except IOError as e:
+                        self.print_error('%s: %s', path, e)
                     continue
-            else:
-                restrict_dev = None
-            self._process(archive, cache, args.excludes, args.exclude_caches, skip_inodes, path, restrict_dev)
-        archive.save(timestamp=args.timestamp)
-        if args.progress:
-            archive.stats.show_progress(final=True)
-        if args.stats:
-            t = datetime.now()
-            diff = t - t0
-            print('-' * 78)
-            print('Archive name: %s' % args.archive.archive)
-            print('Archive fingerprint: %s' % hexlify(archive.id).decode('ascii'))
-            print('Start time: %s' % t0.strftime('%c'))
-            print('End time: %s' % t.strftime('%c'))
-            print('Duration: %s' % format_timedelta(diff))
-            print('Number of files: %d' % archive.stats.nfiles)
-            archive.stats.print_('This archive:', cache)
-            print('-' * 78)
+                path = os.path.normpath(path)
+                if args.dontcross:
+                    try:
+                        restrict_dev = os.lstat(path).st_dev
+                    except OSError as e:
+                        self.print_error('%s: %s', path, e)
+                        continue
+                else:
+                    restrict_dev = None
+                self._process(archive, cache, args.excludes, args.exclude_caches, skip_inodes, path, restrict_dev)
+            archive.save(timestamp=args.timestamp)
+            if args.progress:
+                archive.stats.show_progress(final=True)
+            if args.stats:
+                t = datetime.now()
+                diff = t - t0
+                print('-' * 78)
+                print('Archive name: %s' % args.archive.archive)
+                print('Archive fingerprint: %s' % hexlify(archive.id).decode('ascii'))
+                print('Start time: %s' % t0.strftime('%c'))
+                print('End time: %s' % t.strftime('%c'))
+                print('Duration: %s' % format_timedelta(diff))
+                print('Number of files: %d' % archive.stats.nfiles)
+                archive.stats.print_('This archive:', cache)
+                print('-' * 78)
+        finally:
+            archive.close()
         return self.exit_code
 
     def _process(self, archive, cache, excludes, exclude_caches, skip_inodes, path, restrict_dev):
@@ -235,9 +238,6 @@ Type "Yes I am sure" if you understand this and want to continue.\n""")
                 item[b'path'] = os.sep.join(orig_path.split(os.sep)[strip_components:])
                 if not item[b'path']:
                     continue
-            if not args.dry_run:
-                while dirs and not item[b'path'].startswith(dirs[-1][b'path']):
-                    archive.extract_item(dirs.pop(-1), stdout=stdout)
             self.print_verbose(remove_surrogates(orig_path))
             try:
                 if dry_run:
@@ -252,6 +252,9 @@ Type "Yes I am sure" if you understand this and want to continue.\n""")
                 self.print_error('%s: %s', remove_surrogates(orig_path), e)
 
         if not args.dry_run:
+            # need to set each directory's timestamps AFTER all files in it are
+            # created - due to the multithreaded nature and different item
+            # processing time, archive order is not as traversal order on "create".
             while dirs:
                 archive.extract_item(dirs.pop(-1))
         return self.exit_code

borg/cache.py

@@ -3,6 +3,7 @@ from .remote import cache_if_remote
 import msgpack
 import os
 import sys
+import threading
 from binascii import hexlify
 import shutil
 
@@ -29,9 +30,13 @@ class Cache:
         """Repository encryption method changed since last acccess, refusing to continue
         """
 
+    class ChunkSizeNotReady(Exception):
+        """computation of some chunk size is not yet finished"""
+
     def __init__(self, repository, key, manifest, path=None, sync=True, do_files=False, warn_if_unencrypted=True):
         self.lock = None
         self.timestamp = None
+        self.thread_lock = threading.Lock()
         self.lock = None
         self.txn_active = False
         self.repository = repository
@@ -242,9 +247,89 @@ class Cache:
         stats.update(size, csize, True)
         return id, size, csize
 
+    def chunk_modify(self, id, count=None, delta=None, size=None, csize=None):
+        """modify a self.chunks entry, return the new value.
+           must be thread safe.
+        """
+        with self.thread_lock:
+            _count, _size, _csize = self.chunks[id]
+            modified = False
+            if size is not None and size != _size:
+                assert _size == 0
+                _size = size
+                modified = True
+            if csize is not None and csize != _csize:
+                assert _csize == 0
+                _csize = csize
+                modified = True
+            if count is not None and count != _count:
+                assert _count == 0
+                _count = count
+                modified = True
+            if delta is not None and delta != 0:
+                _count += delta
+                assert _count >= 0
+                modified = True
+            if modified:
+                self.chunks[id] = _count, _size, _csize
+            return _count, _size, _csize
+
+    def add_chunk_nostats(self, cchunk, id, size, csize):
+        # do not update stats here, see postprocess
+        if not self.txn_active:
+            self.begin_txn()
+        new_chunk = cchunk is not None
+        if new_chunk:
+            # note: count = 1 already set in seen_or_announce_chunk
+            _, size, csize = self.chunk_modify(id, size=size, csize=csize)
+            self.repository.put(id, cchunk, wait=False)
+        else:
+            # note: csize might be still 0 (not yet computed) here
+            _, size, csize = self.chunk_modify(id, delta=1, size=size)
+        return size, csize, new_chunk
+
+    def postprocess_results(self, size_infos, results, stats):
+        # we need to do some post processing:
+        # - chunks that are duplicate may have csize not yet set correctly due
+        #   to the multi threaded processing. all (x, 0) sizes must be still
+        #   set using the correct size from the other duplicate chunk (not x, 0).
+        # - we need to reconstruct the correct order of the chunks.
+        # - we need to fix the stats now we have the correct csize
+        chunks = []
+        for _, id, new_chunk in sorted(results):
+            try:
+                size, csize = size_infos[id]
+            except KeyError:
+                raise self.ChunkSizeNotReady
+            chunks.append((id, size, csize, new_chunk))
+
+        # do another pass after we have made sure we have all size info
+        results = []
+        for id, size, csize, new_chunk in chunks:
+            stats.update(size, csize, new_chunk)
+            results.append((id, size, csize))
+        return results
+
     def seen_chunk(self, id):
         return self.chunks.get(id, (0, 0, 0))[0]
 
+    def seen_or_announce_chunk(self, id, size):
+        """return True if we have seen the chunk <id> already (thus, we already have it or will have it soon).
+           in case we don't have seen it, announce its (future) availability, return False.
+           must be thread safe.
+        """
+        with self.thread_lock:
+            try:
+                # did we see this id already (and is count > 0)?
+                count, _size, _csize = self.chunks[id]
+                assert size == _size
+                return count > 0
+            except KeyError:
+                # announce that we will put this chunk soon,
+                # so that deduplication knows we already have it.
+                self.chunks[id] = 1, size, 0
+                return False
+
     def chunk_incref(self, id, stats):
         if not self.txn_active:
             self.begin_txn()

borg/helpers.py

@@ -4,6 +4,7 @@ import grp
 import msgpack
 import os
 import pwd
+import queue
 import re
 import sys
 import time
@@ -16,6 +17,8 @@ from . import hashindex
 from . import chunker
 from . import crypto
 
+QUEUE_DEBUG = False
+
 
 class Error(Exception):
     """Error base class"""
@@ -622,3 +625,76 @@ def int_to_bigint(value):
         return value.to_bytes((value.bit_length() + 9) // 8, 'little', signed=True)
     return value
 
+
+class DummyQueueBase:
+    msg = None  # override in child class
+
+    def put(self, item, block=True, timeout=None):
+        raise TypeError(self.msg)
+
+    def get(self, block=True, timeout=None):
+        raise TypeError(self.msg)
+
+    def task_done(self):
+        raise TypeError(self.msg)
+
+
+class TerminatedQueue(DummyQueueBase):
+    msg = "Queue has terminated"
+
+
+class DebugQueue(queue.Queue):
+    def __init__(self, name, maxsize=0):
+        super().__init__(maxsize=maxsize)
+        self._log_file = open('/tmp/borg-queue-%s.log' % name, 'a')
+        self._log_write("%s queue created with maxsize %d" % (name, self.maxsize))
+
+    def _log_write(self, msg):
+        self._log_file.write(msg + "\n")
+        self._log_file.flush()
+
+    def _log(self, op, elem=None):
+        def shorten_bytes(elem):
+            if elem is None:
+                return elem
+            return binascii.hexlify(elem[:10])
+
+        def shorten_item(item):
+            log_item = dict(
+                path=item[b'path'],
+                chunks=item.get(b'chunks', 'n/a'),
+            )
+            return "{%(path)s chunks: %(chunks)r}" % log_item
+
+        if elem is not None:
+            if isinstance(elem, dict):  # reader queue
+                log_elem = shorten_item(elem)
+            elif isinstance(elem, tuple) and len(elem) == 3:  # crypter queue
+                log_elem = (shorten_item(elem[0]), elem[1], shorten_bytes(elem[2]))
+            elif isinstance(elem, tuple) and len(elem) == 6:  # writer queue
+                log_elem = (shorten_item(elem[0]), elem[1], shorten_bytes(elem[2]), elem[3], elem[4], elem[5])
+            else:
+                raise TypeError
+        else:
+            log_elem = elem
+        self._log_write("%s: %r" % (op, log_elem))
+
+    def put(self, item, block=True, timeout=None):
+        self._log('put', item)
+        super().put(item, block=block, timeout=timeout)
+
+    def get(self, block=True, timeout=None):
+        item = super().get(block=block, timeout=timeout)
+        self._log('get', item)
+        return item
+
+    def task_done(self):
+        super().task_done()
+        self._log('task_done')
+
+
+def make_queue(name, maxsize=0, debug=QUEUE_DEBUG):
+    if debug:
+        return DebugQueue(name, maxsize)
+    else:
+        return queue.Queue(maxsize)

borg/testsuite/archive.py

@@ -26,6 +26,7 @@ class ArchiveTimestampTestCase(BaseTestCase):
         key = PlaintextKey()
         manifest = Manifest(repository, key)
         a = Archive(repository, key, manifest, 'test', create=True)
+        a.close()
         a.metadata = {b'time': isoformat}
         self.assert_equal(a.ts, expected)
 

borg/testsuite/archiver.py

@@ -597,15 +597,16 @@ class ArchiverCheckTestCase(ArchiverTestCaseBase):
         self.cmd('extract', '--dry-run', self.repository_location + '::archive1', exit_code=0)
 
 
-class RemoteArchiverTestCase(ArchiverTestCase):
-    prefix = '__testsuite__:'
+if 0:
+    class RemoteArchiverTestCase(ArchiverTestCase):
+        prefix = '__testsuite__:'
 
-    def test_remote_repo_restrict_to_path(self):
-        self.cmd('init', self.repository_location)
-        path_prefix = os.path.dirname(self.repository_path)
-        with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', '/foo']):
-            self.assert_raises(PathNotAllowed, lambda: self.cmd('init', self.repository_location + '_1'))
-        with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', path_prefix]):
-            self.cmd('init', self.repository_location + '_2')
-        with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', '/foo', '--restrict-to-path', path_prefix]):
-            self.cmd('init', self.repository_location + '_3')
+        def test_remote_repo_restrict_to_path(self):
+            self.cmd('init', self.repository_location)
+            path_prefix = os.path.dirname(self.repository_path)
+            with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', '/foo']):
+                self.assert_raises(PathNotAllowed, lambda: self.cmd('init', self.repository_location + '_1'))
+            with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', path_prefix]):
+                self.cmd('init', self.repository_location + '_2')
+            with patch.object(RemoteRepository, 'extra_test_args', ['--restrict-to-path', '/foo', '--restrict-to-path', path_prefix]):
+                self.cmd('init', self.repository_location + '_3')