Merge branch 'master' into object_lock

* master:
  Apply code review feedback
  Do not run a the test with extra parameters on old apiver
  Add test of B2Api/Services initialization
  Rename parameter for consistency
  remove useless line
  More detailed download performance logs
  Add logging performance summary of parallel download threads
  Refactor stats collector to be usable for other threads
  Fix CopySizeTooBig exception
  Add download stats collector for parallel download writer thread
  Make streams loggable
  Add `max_download_streams_per_file` parameter to B2Api and underlying classes

Author: ppolewicz

CHANGELOG.md

@@ -7,11 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+* Logging performance summary of parallel download threads
+* Add `max_download_streams_per_file` parameter to B2Api class and underlying structures
 * Add `is_file_lock_enabled` parameter to `Bucket.update()` and related methods
 
 ### Fixed
 * Replace `ReplicationScanResult.source_has_sse_c_enabled` with `source_encryption_mode`
 * Fix `B2Api.get_key()` and `RawSimulator.delete_key()`
+* Fix calling `CopySizeTooBig` exception
 
 ### Infrastructure
 * Fix nox's deprecated `session.install()` calls

b2sdk/api.py

@@ -64,6 +64,7 @@ def __init__(
         max_download_workers: Optional[int] = None,
         save_to_buffer_size: Optional[int] = None,
         check_download_hash: bool = True,
+        max_download_streams_per_file: Optional[int] = None,
     ):
         """
         Initialize Services object using given session.
@@ -74,6 +75,7 @@ def __init__(
         :param max_download_workers: maximum number of download threads
         :param save_to_buffer_size: buffer size to use when writing files using DownloadedFile.save_to
         :param check_download_hash: whether to check hash of downloaded files. Can be disabled for files with internal checksums, for example, or to forcefully retrieve objects with corrupted payload or hash value
+        :param max_download_streams_per_file: how many streams to use for parallel downloader
         """
         self.api = api
         self.session = api.session
@@ -82,11 +84,13 @@ def __init__(
             services=self, max_workers=max_upload_workers
         )
         self.copy_manager = self.COPY_MANAGER_CLASS(services=self, max_workers=max_copy_workers)
+        assert max_download_streams_per_file is None or max_download_streams_per_file >= 1
         self.download_manager = self.DOWNLOAD_MANAGER_CLASS(
             services=self,
             max_workers=max_download_workers,
             write_buffer_size=save_to_buffer_size,
             check_hash=check_download_hash,
+            max_download_streams_per_file=max_download_streams_per_file,
         )
         self.emerger = Emerger(self)
 
@@ -129,6 +133,7 @@ def __init__(
         max_download_workers: Optional[int] = None,
         save_to_buffer_size: Optional[int] = None,
         check_download_hash: bool = True,
+        max_download_streams_per_file: Optional[int] = None,
     ):
         """
         Initialize the API using the given account info.
@@ -145,6 +150,7 @@ def __init__(
         :param max_download_workers: maximum number of download threads
         :param save_to_buffer_size: buffer size to use when writing files using DownloadedFile.save_to
         :param check_download_hash: whether to check hash of downloaded files. Can be disabled for files with internal checksums, for example, or to forcefully retrieve objects with corrupted payload or hash value
+        :param max_download_streams_per_file: number of streams for parallel download manager
         """
         self.session = self.SESSION_CLASS(
             account_info=account_info, cache=cache, api_config=api_config
@@ -159,6 +165,7 @@ def __init__(
             max_download_workers=max_download_workers,
             save_to_buffer_size=save_to_buffer_size,
             check_download_hash=check_download_hash,
+            max_download_streams_per_file=max_download_streams_per_file,
         )
 
     @property

b2sdk/stream/progress.py

@@ -36,6 +36,9 @@ def _progress_update(self, delta):
         self.bytes_completed += delta
         self.progress_listener.bytes_completed(self.bytes_completed + self.offset)
 
+    def __str__(self):
+        return str(self.stream)
+
 
 class ReadingStreamWithProgress(AbstractStreamWithProgress):
     """

b2sdk/transfer/inbound/download_manager.py

@@ -44,7 +44,13 @@ class DownloadManager(TransferManager, ThreadPoolMixin, metaclass=B2TraceMetaAbs
     PARALLEL_DOWNLOADER_CLASS = staticmethod(ParallelDownloader)
     SIMPLE_DOWNLOADER_CLASS = staticmethod(SimpleDownloader)
 
-    def __init__(self, write_buffer_size: Optional[int] = None, check_hash: bool = True, **kwargs):
+    def __init__(
+        self,
+        write_buffer_size: Optional[int] = None,
+        check_hash: bool = True,
+        max_download_streams_per_file: Optional[int] = None,
+        **kwargs
+    ):
         """
         Initialize the DownloadManager using the given services object.
         """
@@ -58,6 +64,7 @@ def __init__(self, write_buffer_size: Optional[int] = None, check_hash: bool = T
                 align_factor=write_buffer_size,
                 thread_pool=self._thread_pool,
                 check_hash=check_hash,
+                max_streams=max_download_streams_per_file,
             ),
             self.SIMPLE_DOWNLOADER_CLASS(
                 min_chunk_size=self.MIN_CHUNK_SIZE,

b2sdk/transfer/inbound/downloaded_file.py

@@ -79,6 +79,9 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         self.file.close()
         set_file_mtime(self.path_, self.mod_time_to_set)
 
+    def __str__(self):
+        return str(self.path_)
+
 
 class DownloadedFile:
     """

b2sdk/transfer/inbound/downloader/parallel.py

@@ -10,6 +10,7 @@
 
 from concurrent import futures
 from io import IOBase
+from time import perf_counter_ns
 from typing import Optional
 import logging
 import queue
@@ -18,6 +19,7 @@
 from requests.models import Response
 
 from .abstract import AbstractDownloader
+from .stats_collector import StatsCollector
 from b2sdk.encryption.setting import EncryptionSetting
 from b2sdk.file_version import DownloadVersion
 from b2sdk.session import B2Session
@@ -118,7 +120,15 @@ def download(
         if self._check_hash:
             # we skip hashing if we would not check it - hasher object is actually a EmptyHasher instance
             # but we avoid here reading whole file (except for the first part) from disk again
+            before_hash = perf_counter_ns()
             self._finish_hashing(first_part, file, hasher, download_version.content_length)
+            after_hash = perf_counter_ns()
+            logger.info(
+                'download stats | %s | %s total: %.3f ms',
+                file,
+                'finish_hash',
+                (after_hash - before_hash) / 1000000,
+            )
 
         return bytes_written, hasher.hexdigest()
 
@@ -203,18 +213,33 @@ def __init__(self, file, max_queue_depth):
         self.file = file
         self.queue = queue.Queue(max_queue_depth)
         self.total = 0
+        self.stats_collector = StatsCollector(str(self.file), 'writer', 'seek')
         super(WriterThread, self).__init__()
 
     def run(self):
         file = self.file
         queue_get = self.queue.get
+        stats_collector_read_append = self.stats_collector.read.append
+        stats_collector_other_append = self.stats_collector.other.append
+        stats_collector_write_append = self.stats_collector.write.append
+        start = perf_counter_ns()
         while 1:
+
+            before_read = perf_counter_ns()
             shutdown, offset, data = queue_get()
+            stats_collector_read_append(perf_counter_ns() - before_read)
+
             if shutdown:
                 break
+            before_seek = perf_counter_ns()
             file.seek(offset)
+            after_seek = perf_counter_ns()
             file.write(data)
+            after_write = perf_counter_ns()
+            stats_collector_other_append(after_seek - before_seek)
+            stats_collector_write_append(after_write - after_seek)
             self.total += len(data)
+        self.stats_collector.total = perf_counter_ns() - start
 
     def __enter__(self):
         self.start()
@@ -223,6 +248,7 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.queue.put((True, None, None))
         self.join()
+        self.stats_collector.report()
 
 
 def download_first_part(
@@ -243,6 +269,19 @@ def download_first_part(
     :param chunk_size: size (in bytes) of read data chunks
     :param encryption: encryption mode, algorithm and key
     """
+    # This function contains a loop that has heavy impact on performance.
+    # It has not been broken down to several small functions due to fear of
+    # performance overhead of calling a python function. Advanced performance optimization
+    # techniques are in use here, for example avoiding internal python getattr calls by
+    # caching function signatures in local variables. Most of this code was written in
+    # times where python 2.7 (or maybe even 2.6) had to be supported, so maybe some
+    # of those optimizations could be removed without affecting performance.
+    #
+    # Due to reports of hard to debug performance issues, this code has also been riddled
+    # with performance measurements. A known issue is GCP VMs which have more network speed
+    # than storage speed, but end users have different issues with network and storage.
+    # Basic tools to figure out where the time is being spent is a must for long-term
+    # maintainability.
 
     writer_queue_put = writer.queue.put
     hasher_update = hasher.update
@@ -253,14 +292,29 @@ def download_first_part(
 
     bytes_read = 0
     stop = False
+
+    stats_collector = StatsCollector(response.url, f'{first_offset}:{last_offset}', 'hash')
+    stats_collector_read_append = stats_collector.read.append
+    stats_collector_other_append = stats_collector.other.append
+    stats_collector_write_append = stats_collector.write.append
+    start = before_read = perf_counter_ns()
     for data in response.iter_content(chunk_size=chunk_size):
+        stats_collector_read_append(perf_counter_ns() - before_read)
         if first_offset + bytes_read + len(data) >= last_offset:
             to_write = data[:last_offset - bytes_read]
             stop = True
         else:
             to_write = data
+        before_put = perf_counter_ns()
         writer_queue_put((False, first_offset + bytes_read, to_write))
+
+        before_hash = perf_counter_ns()
         hasher_update(to_write)
+        after_hash = perf_counter_ns()
+
+        stats_collector_write_append(before_hash - before_put)
+        stats_collector_other_append(after_hash - before_hash)
+
         bytes_read += len(to_write)
         if stop:
             break
@@ -284,11 +338,24 @@ def download_first_part(
             cloud_range.as_tuple(),
             encryption=encryption,
         ) as response:
+            before_read = perf_counter_ns()
             for to_write in response.iter_content(chunk_size=chunk_size):
+                stats_collector_read_append(perf_counter_ns() - before_read)
+
+                before_put = perf_counter_ns()
                 writer_queue_put((False, first_offset + bytes_read, to_write))
+                before_hash = perf_counter_ns()
                 hasher_update(to_write)
+                after_hash = perf_counter_ns()
+
+                stats_collector_write_append(before_hash - before_put)
+                stats_collector_other_append(after_hash - before_hash)
+
                 bytes_read += len(to_write)
+                before_read = perf_counter_ns()
         tries_left -= 1
+    stats_collector.total = perf_counter_ns() - start
+    stats_collector.report()
 
 
 def download_non_first_part(
@@ -321,15 +388,27 @@ def download_non_first_part(
             'download attempts remaining: %i, bytes read already: %i. Getting range %s now.',
             retries_left, bytes_read, cloud_range
         )
+        stats_collector = StatsCollector(url, f'{cloud_range.start}:{cloud_range.end}', 'none')
+        stats_collector_read_append = stats_collector.read.append
+        stats_collector_write_append = stats_collector.write.append
+        start = before_read = perf_counter_ns()
         with session.download_file_from_url(
             url,
             cloud_range.as_tuple(),
             encryption=encryption,
         ) as response:
+            before_read = perf_counter_ns()
             for to_write in response.iter_content(chunk_size=chunk_size):
+                after_read = perf_counter_ns()
                 writer_queue_put((False, start_range + bytes_read, to_write))
+                after_write = perf_counter_ns()
+                stats_collector_read_append(after_read - before_read)
+                stats_collector_write_append(after_write - after_read)
                 bytes_read += len(to_write)
+                before_read = perf_counter_ns()
         retries_left -= 1
+        stats_collector.total = perf_counter_ns() - start
+        stats_collector.report()
 
 
 class PartToDownload:

b2sdk/transfer/inbound/downloader/stats_collector.py

@@ -0,0 +1,51 @@
+######################################################################
+#
+# File: b2sdk/transfer/inbound/downloader/stats_collector.py
+#
+# Copyright 2020 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+
+import logging
+from dataclasses import dataclass, field
+from typing import List  # 3.7 doesn't understand `list` vs `List`
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class StatsCollector:
+    name: str  #: file name or object url
+    detail: str  #: description of the thread, ex. "10000000:20000000" or "writer"
+    other_name: str  #: other statistic, typically "seek" or "hash"
+    total: Optional[int] = None
+    other: List[int] = field(default_factory=list)
+    write: List[int] = field(default_factory=list)
+    read: List[int] = field(default_factory=list)
+
+    def report(self):
+        if self.read:
+            logger.info('download stats | %s | TTFB: %.3f ms', self, self.read[0] / 1000000)
+            logger.info(
+                'download stats | %s | read() without TTFB: %.3f ms', self,
+                sum(self.read[1:]) / 1000000
+            )
+        if self.other:
+            logger.info(
+                'download stats | %s | %s total: %.3f ms', self, self.other_name,
+                sum(self.other) / 1000000
+            )
+        if self.write:
+            logger.info(
+                'download stats | %s | write() total: %.3f ms', self,
+                sum(self.write) / 1000000
+            )
+        if self.total is not None:
+            overhead = self.total - sum(self.write) - sum(self.other) - sum(self.read)
+            logger.info('download stats | %s | overhead: %.3f ms', self, overhead / 1000000)
+
+    def __str__(self):
+        return f'{self.name}[{self.detail}]'