Use workers for file download

audbackend/core/backend/artifactory.py

@@ -209,6 +209,7 @@ def _copy_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Copy file on backend."""
@@ -263,6 +264,7 @@ def _get_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Get file from backend."""
@@ -287,6 +289,7 @@ def _move_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Move file on backend."""

audbackend/core/backend/base.py

@@ -176,6 +176,7 @@ def _copy_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Copy file on backend.
@@ -189,14 +190,20 @@ def _copy_file(
         """
         with tempfile.TemporaryDirectory() as tmp:
             tmp_path = audeer.path(tmp, "~")
-            tmp_path = self.get_file(src_path, tmp_path, verbose=verbose)
+            tmp_path = self.get_file(
+                src_path,
+                tmp_path,
+                num_workers=num_workers,
+                verbose=verbose,
+            )
             self.put_file(tmp_path, dst_path, verbose=verbose)
 
     def copy_file(
         self,
         src_path: str,
         dst_path: str,
         *,
+        num_workers: int = 1,
         validate: bool = False,
         verbose: bool = False,
     ):
@@ -219,6 +226,7 @@ def copy_file(
         Args:
             src_path: source path to file on backend
             dst_path: destination path to file on backend
+            num_workers: number of parallel jobs
             validate: verify file was successfully copied
             verbose: show debug messages
 
@@ -246,6 +254,7 @@ def copy_file(
                 self._copy_file,
                 src_path,
                 dst_path,
+                num_workers,
                 verbose,
             )
 
@@ -435,6 +444,7 @@ def get_archive(
         dst_root: str,
         *,
         tmp_root: str = None,
+        num_workers: int = 1,
         validate: bool = False,
         verbose: bool = False,
     ) -> list[str]:
@@ -459,6 +469,7 @@ def get_archive(
             dst_root: local destination directory
             tmp_root: directory under which archive is temporarily extracted.
                 Defaults to temporary directory of system
+            num_workers: number of parallel jobs
             validate: verify archive was successfully
                 retrieved from the backend
             verbose: show debug messages
@@ -496,6 +507,7 @@ def get_archive(
             self.get_file(
                 src_path,
                 local_archive,
+                num_workers=num_workers,
                 validate=validate,
                 verbose=verbose,
             )
@@ -510,6 +522,7 @@ def _get_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):  # pragma: no cover
         r"""Get file from backend."""
@@ -520,6 +533,7 @@ def get_file(
         src_path: str,
         dst_path: str,
         *,
+        num_workers: int = 1,
         validate: bool = False,
         verbose: bool = False,
     ) -> str:
@@ -546,6 +560,7 @@ def get_file(
         Args:
             src_path: path to file on backend
             dst_path: destination path to local file
+            num_workers: number of parallel jobs
             validate: verify file was successfully
                 retrieved from the backend
             verbose: show debug messages
@@ -594,6 +609,7 @@ def get_file(
                     self._get_file,
                     src_path,
                     tmp_path,
+                    num_workers,
                     verbose,
                 )
                 audeer.move_file(tmp_path, dst_path)
@@ -738,6 +754,7 @@ def _move_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Move file on backend.
@@ -749,14 +766,15 @@ def _move_file(
         if backend supports a native way to move files.
 
         """
-        self.copy_file(src_path, dst_path, verbose=verbose)
+        self.copy_file(src_path, dst_path, num_workers=num_workers, verbose=verbose)
         self.remove_file(src_path)
 
     def move_file(
         self,
         src_path: str,
         dst_path: str,
         *,
+        num_workers: int = 1,
         validate: bool = False,
         verbose: bool = False,
     ):
@@ -783,6 +801,7 @@ def move_file(
         Args:
             src_path: source path to file on backend
             dst_path: destination path to file on backend
+            num_workers: number of parallel jobs
             validate: verify file was successfully moved
             verbose: show debug messages
 
@@ -812,6 +831,7 @@ def move_file(
                 self.copy_file(
                     src_path,
                     dst_path,
+                    num_workers=num_workers,
                     validate=True,
                     verbose=verbose,
                 )
@@ -821,6 +841,7 @@ def move_file(
                     self._move_file,
                     src_path,
                     dst_path,
+                    num_workers,
                     verbose,
                 )
         else:

audbackend/core/backend/filesystem.py

@@ -53,6 +53,7 @@ def _copy_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Copy file on backend."""
@@ -114,6 +115,7 @@ def _get_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Get file from backend."""
@@ -142,6 +144,7 @@ def _move_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Move file on backend."""

audbackend/core/backend/minio.py

@@ -2,7 +2,9 @@
 import getpass
 import mimetypes
 import os
+import signal
 import tempfile
+import threading
 
 import minio
 
@@ -201,6 +203,7 @@ def _copy_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Copy file on backend."""
@@ -212,7 +215,7 @@ def _copy_file(
         if self._size(src_path) / 1024 / 1024 / 1024 >= 4.9:
             with tempfile.TemporaryDirectory() as tmp_dir:
                 tmp_path = audeer.path(tmp_dir, os.path.basename(src_path))
-                self._get_file(src_path, tmp_path, verbose)
+                self._get_file(src_path, tmp_path, num_workers, verbose)
                 self._put_file(tmp_path, dst_path, checksum, verbose)
         else:
             self._client.copy_object(
@@ -276,6 +279,7 @@ def _get_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Get file from backend."""
@@ -284,33 +288,94 @@ def _get_file(
             bucket_name=self.repository,
             object_name=src_path,
         ).size
-        chunk = 4 * 1024
-        with audeer.progress_bar(total=src_size, disable=not verbose) as pbar:
-            desc = audeer.format_display_message(
-                f"Download {os.path.basename(str(src_path))}", pbar=True
-            )
-            pbar.set_description_str(desc)
-            pbar.refresh()
 
-            dst_size = 0
-            try:
-                response = self._client.get_object(
-                    bucket_name=self.repository,
-                    object_name=src_path,
-                )
-                with open(dst_path, "wb") as dst_fp:
-                    while src_size > dst_size:
-                        data = response.read(chunk)
-                        n_data = len(data)
-                        if n_data > 0:
-                            dst_fp.write(data)
-                            dst_size += n_data
-                            pbar.update(n_data)
-            except Exception as e:  # pragma: no cover
-                raise RuntimeError(f"Error downloading file: {e}")
-            finally:
-                response.close()
-                response.release_conn()
+        # Create cancellation event for handling interrupts
+        cancel_event = threading.Event()
+
+        # Install signal handler to set cancel_event on Ctrl+C
+        def signal_handler(signum, frame):
+            cancel_event.set()  # pragma: no cover
+
+        original_handler = signal.signal(signal.SIGINT, signal_handler)
+
+        # Setup progress bar
+        desc = audeer.format_display_message(
+            f"Download {os.path.basename(str(src_path))}",
+            pbar=verbose,
+        )
+        pbar = audeer.progress_bar(total=src_size, desc=desc, disable=not verbose)
+
+        try:
+            if num_workers == 1:
+                # Simple single-threaded download
+                with pbar:
+                    self._download_file(src_path, dst_path, pbar, cancel_event)
+            else:
+                # Multi-threaded download with pre-allocated file
+                with open(dst_path, "wb") as f:
+                    f.truncate(src_size)
+
+                # Create and run download tasks
+                tasks = []
+                # Ensure num_workers does not exceed src_size
+                num_workers = min(num_workers, src_size) if src_size > 0 else 1
+                chunk_size = src_size // num_workers
+                for i in range(num_workers):
+                    offset = i * chunk_size
+                    length = chunk_size if i < num_workers - 1 else src_size - offset
+                    tasks.append(
+                        ([src_path, dst_path, pbar, cancel_event, offset, length], {})
+                    )
+
+                with pbar:
+                    audeer.run_tasks(
+                        self._download_file, tasks, num_workers=num_workers
+                    )
+        except KeyboardInterrupt:
+            # Clean up partial file
+            if os.path.exists(dst_path):
+                os.remove(dst_path)
+            raise
+        finally:
+            # Restore original signal handler
+            signal.signal(signal.SIGINT, original_handler)
+
+    def _download_file(
+        self,
+        src_path: str,
+        dst_path: str,
+        pbar,
+        cancel_event: threading.Event = None,
+        offset: int = 0,
+        length: int | None = None,
+    ):
+        """Download file or part of file."""
+        chunk_size = 4 * 1024  # 4 KB
+
+        # Get the data stream
+        kwargs = {"offset": offset}
+        if length is not None:
+            kwargs["length"] = length
+        response = self._client.get_object(
+            bucket_name=self.repository,
+            object_name=src_path,
+            **kwargs,
+        )
+
+        try:
+            with open(dst_path, "r+b" if offset else "wb") as f:
+                if offset:
+                    f.seek(offset)
+
+                while data := response.read(chunk_size):
+                    # Check if cancellation was requested
+                    if cancel_event and cancel_event.is_set():
+                        raise KeyboardInterrupt("Download cancelled by user")
+                    f.write(data)
+                    pbar.update(len(data))
+        finally:
+            response.close()
+            response.release_conn()
 
     def _ls(
         self,
@@ -329,10 +394,11 @@ def _move_file(
         self,
         src_path: str,
         dst_path: str,
+        num_workers: int,
         verbose: bool,
     ):
         r"""Move file on backend."""
-        self._copy_file(src_path, dst_path, verbose)
+        self._copy_file(src_path, dst_path, num_workers, verbose)
         self._remove_file(src_path)
 
     def _open(