Make S3 transfer operations cancellable

rfi_file_monitor/operations/s3_uploader.py

@@ -4,6 +4,7 @@
 gi.require_version("Gtk", "3.0")
 from gi.repository import Gtk
 import boto3
+import boto3.s3.transfer
 import botocore
 from munch import Munch
 from tenacity import retry, stop_after_attempt, wait_exponential, \
@@ -274,6 +275,7 @@ def _attach_metadata(cls, file: File, endpoint_url: str, key: str, params: Munch
     def _run(cls, file: File, preflight_check_metadata: Dict[int, Dict[str, Any]], params: Munch, operation_index:int):
         client_options = cls._get_client_options(params)
         s3_client = boto3.client('s3', **client_options)
+        transfer_manager = boto3.s3.transfer.create_transfer_manager(client=s3_client, config=TransferConfig)
 
         # object creation options
         object_acl_options = cls._get_dict_acl_options(preflight_check_metadata, 'object_acl_options', ALLOWED_OBJECT_ACL_OPTIONS)
@@ -303,12 +305,12 @@ def _run(cls, file: File, preflight_check_metadata: Dict[int, Dict[str, Any]], p
                     raise SkippedOperation('File has been uploaded already')
 
         try:
-            s3_client.upload_file(
-                Filename=file.filename,
-                Bucket=params.bucket_name,
-                Key=key,
-                Config=TransferConfig,
-                Callback=S3ProgressPercentage(file, file.filename, operation_index),
+            with boto3.s3.transfer.S3Transfer(manager=transfer_manager) as s3_transfer:
+                s3_transfer.upload_file(
+                    filename=file.filename,
+                    bucket=params.bucket_name,
+                    key=key,
+                    callback=S3ProgressPercentage(file, file.filename, operation_index, manager=transfer_manager),
                 )
             if object_tags:
                 s3_client.put_object_tagging(
@@ -323,7 +325,7 @@ def _run(cls, file: File, preflight_check_metadata: Dict[int, Dict[str, Any]], p
                     **object_acl_options,
                 )
         except Exception as e:
-            logger.exception(f'S3UploaderOperation.run exception')
+            logger.debug(f'S3UploaderOperation.run exception: {e}')
             del s3_client
             del client_options
             return str(e)

rfi_file_monitor/utils/s3.py

@@ -1,14 +1,21 @@
+from __future__ import annotations
+
 import boto3.s3.transfer
 
 from ..file import File
+from . import ExitableThread
 
 import os
 import hashlib
 from typing import Optional, Union
+import threading
+import logging
+
+logger = logging.getLogger(__name__)
 
 KB = 1024
 MB = KB * KB
-TransferConfig = boto3.s3.transfer.TransferConfig(max_concurrency=1, multipart_chunksize=8*MB, multipart_threshold=8*MB)
+TransferConfig = boto3.s3.transfer.TransferConfig(max_concurrency=1, multipart_chunksize=8*MB, multipart_threshold=8*MB, use_threads=False)
 
 def calculate_etag(file: Union[str, os.PathLike]):
     # taken from https://stackoverflow.com/a/52300584
@@ -37,7 +44,14 @@ def calculate_etag(file: Union[str, os.PathLike]):
 # taken from https://boto3.amazonaws.com/v1/documentation/api/latest/guide/s3-uploading-files.html
 class S3ProgressPercentage(object):
 
-    def __init__(self, file: File, filename: Union[str, os.PathLike], operation_index: int, size: Optional[float] = None):
+    def __init__(self,
+        file: File,
+        filename: Union[str, os.PathLike],
+        operation_index: int,
+        size: Optional[float] = None,
+        manager: Optional[boto3.s3.transfer.TransferManager] = None,
+        ):
+
         self._file = file
         if size:
             self._size = size
@@ -46,8 +60,22 @@ def __init__(self, file: File, filename: Union[str, os.PathLike], operation_inde
         self._seen_so_far = 0
         self._last_percentage = 0
         self._operation_index = operation_index
+        self._manager = manager
 
     def __call__(self, bytes_amount):
+        current_thread = threading.current_thread()
+
+        if isinstance(current_thread, ExitableThread) and current_thread.should_exit and self._manager is not None:
+            # see https://github.com/boto/s3transfer/pull/144
+            # use shutdown when fixed
+            self._manager._shutdown(cancel=True, cancel_msg='Job cancelled')
+            return
+        elif not isinstance(current_thread, ExitableThread):
+            logger.warning('S3ProgressPercentage not running in ExitableThread!!!!!')
+        elif current_thread is threading.main_thread():
+            logger.warning('S3ProgressPercentage __call__ running in main thread!')
+
+
         # To simplify, assume this is hooked up to a single filename
         self._seen_so_far += bytes_amount
         percentage = (self._seen_so_far / self._size) * 100