Support multipart_threshold for downloads using CRT

.changes/next-release/enhancement-crt-69912.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "crt",
+  "description": "Configure CRT client to download with single GET when object size is below ``multipart_threshold``"
+}

awscli/customizations/s3/factory.py

@@ -20,6 +20,7 @@
 from s3transfer.crt import (
     BotocoreCRTCredentialsWrapper,
     BotocoreCRTRequestSerializer,
+    CRTTransferConfig,
     CRTTransferManager,
     acquire_crt_s3_process_lock,
     create_s3_crt_client,
@@ -241,12 +242,22 @@ def _acquire_crt_s3_process_lock(self):
 
     def _create_crt_transfer_manager(self, params, runtime_config):
         self._acquire_crt_s3_process_lock()
+        config_kwargs = self._resolve_crt_client_config_kwargs(runtime_config)
         return CRTTransferManager(
-            self._create_crt_client(params, runtime_config),
+            self._create_crt_client(params, runtime_config, config_kwargs),
             self._create_crt_request_serializer(params),
+            transfer_config=self._create_crt_transfer_config(config_kwargs),
         )
 
-    def _create_crt_client(self, params, runtime_config):
+    def _create_crt_transfer_config(self, config_kwargs):
+        # The crt client only applies its multipart threshold to uploads, so
+        # downloads rely on the transfer config to match it. Leaving the
+        # threshold unset keeps the client's own download behavior.
+        return CRTTransferConfig(
+            multipart_threshold=config_kwargs.get('multipart_upload_threshold')
+        )
+
+    def _create_crt_client(self, params, runtime_config, config_kwargs):
         create_crt_client_kwargs = {
             'region': self._resolve_region(params),
             'verify': self._resolve_verify(params),
@@ -257,9 +268,7 @@ def _create_crt_client(self, params, runtime_config):
         target_throughput = runtime_config.get('target_bandwidth', None)
         if target_throughput:
             create_crt_client_kwargs['target_throughput'] = target_throughput
-        create_crt_client_kwargs.update(
-            self._resolve_crt_client_config_kwargs(runtime_config)
-        )
+        create_crt_client_kwargs.update(config_kwargs)
         if params.get('sign_request', True):
             crt_credentials_provider = self._get_crt_credentials_provider()
             create_crt_client_kwargs['crt_credentials_provider'] = (

awscli/s3transfer/crt.py

@@ -222,8 +222,31 @@ def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None):
     return target_gbps
 
 
+class CRTTransferConfig:
+    def __init__(self, multipart_threshold=None):
+        """Configuration the CRT transfer manager applies itself
+
+        This only covers configuration that the CRT s3 client cannot apply
+        on its own. Everything else is configured on the client, so it
+        deliberately has no equivalent here.
+
+        :type multipart_threshold: Optional[int]
+        :param multipart_threshold: The size, in bytes, that a download must
+            exceed to be split into ranged requests. The CRT s3 client only
+            applies its own threshold to uploads. If not set, the client
+            decides how to split every download.
+        """
+        self.multipart_threshold = multipart_threshold
+
+
 class CRTTransferManager:
-    def __init__(self, crt_s3_client, crt_request_serializer, osutil=None):
+    def __init__(
+        self,
+        crt_s3_client,
+        crt_request_serializer,
+        osutil=None,
+        transfer_config=None,
+    ):
         """A transfer manager interface for Amazon S3 on CRT s3 client.
 
         :type crt_s3_client: awscrt.s3.S3Client
@@ -237,12 +260,19 @@ def __init__(self, crt_s3_client, crt_request_serializer, osutil=None):
         :type osutil: s3transfer.utils.OSUtils
         :param osutil: OSUtils object to use for os-related behavior when
             using with transfer manager.
+
+        :type transfer_config: s3transfer.crt.CRTTransferConfig
+        :param transfer_config: The transfer configuration to apply. If not
+            provided, the CRT s3 client's own configuration applies to every
+            transfer.
         """
         if osutil is None:
             self._osutil = OSUtils()
         self._crt_s3_client = crt_s3_client
         self._s3_args_creator = S3ClientArgsCreator(
-            crt_request_serializer, self._osutil
+            crt_request_serializer,
+            self._osutil,
+            transfer_config=transfer_config,
         )
         self._crt_exception_translator = (
             crt_request_serializer.translate_crt_exception
@@ -398,6 +428,7 @@ def __init__(self, transfer_id=None, call_args=None):
         self._transfer_id = transfer_id
         self._call_args = call_args
         self._user_context = {}
+        self._size = None
 
     @property
     def call_args(self):
@@ -411,6 +442,13 @@ def transfer_id(self):
     def user_context(self):
         return self._user_context
 
+    @property
+    def size(self):
+        return self._size
+
+    def provide_transfer_size(self, size):
+        self._size = size
+
 
 class CRTTransferFuture(BaseTransferFuture):
     def __init__(self, meta=None, coordinator=None):
@@ -760,9 +798,10 @@ def set_s3_request(self, s3_request):
 
 
 class S3ClientArgsCreator:
-    def __init__(self, crt_request_serializer, os_utils):
+    def __init__(self, crt_request_serializer, os_utils, transfer_config=None):
         self._request_serializer = crt_request_serializer
         self._os_utils = os_utils
+        self._transfer_config = transfer_config
         self._client_config = crt_request_serializer.client_config
         self._service_model = crt_request_serializer.service_model
 
@@ -892,8 +931,19 @@ def _get_make_request_args_get_object(
         make_request_args['recv_filepath'] = recv_filepath
         make_request_args['on_body'] = on_body
         make_request_args['checksum_config'] = checksum_config
+        if self._should_download_in_single_request(future.meta.size):
+            make_request_args['type'] = S3RequestType.DEFAULT
+            make_request_args['operation_name'] = _get_operation_name(
+                request_type
+            )
         return make_request_args
 
+    def _should_download_in_single_request(self, size):
+        if self._transfer_config is None or size is None:
+            return False
+        threshold = self._transfer_config.multipart_threshold
+        return threshold is not None and size <= threshold
+
     def _should_calculate_upload_checksum(self, request_type, extra_args):
         if any(
             checksum_arg in extra_args

tests/functional/s3transfer/test_crt.py

@@ -16,12 +16,14 @@
 import time
 from concurrent.futures import Future
 
+import pytest
 from botocore.session import Session
 from s3transfer.subscribers import BaseSubscriber
 
 from tests import (
     HAS_CRT,
     FileCreator,
+    FileSizeProvider,
     NonSeekableReader,
     NonSeekableWriter,
     mock,
@@ -831,3 +833,88 @@ def test_crt_s3_client_error_handling(self):
         )
         with self.assertRaises(awscrt.exceptions.AwsCrtError):
             future.result()
+
+
+MULTIPART_THRESHOLD = 8 * 1024 * 1024
+DOWNLOADED_CONTENT = 'content'
+
+
+@pytest.fixture
+def files():
+    file_creator = FileCreator()
+    yield file_creator
+    file_creator.remove_all()
+
+
+@pytest.fixture
+def filename(files):
+    return files.full_path('myfile')
+
+
+@pytest.fixture
+def crt_client(files):
+    client = mock.Mock(awscrt.s3.S3Client)
+
+    def simulate_make_request(**kwargs):
+        files.create_file(
+            kwargs['recv_filepath'], DOWNLOADED_CONTENT, mode='w'
+        )
+        kwargs['on_done'](error=None)
+        return mock.Mock(awscrt.s3.S3Request)
+
+    client.make_request.side_effect = simulate_make_request
+    return client
+
+
+@pytest.fixture
+def request_serializer():
+    session = Session()
+    session.set_config_variable('region', 'us-west-2')
+    return s3transfer.crt.BotocoreCRTRequestSerializer(session)
+
+
+@pytest.fixture
+def download(crt_client, request_serializer, filename):
+    """Downloads an object of a given size and returns the crt request args"""
+
+    def _download(size, multipart_threshold):
+        transfer_manager = s3transfer.crt.CRTTransferManager(
+            crt_s3_client=crt_client,
+            crt_request_serializer=request_serializer,
+            transfer_config=s3transfer.crt.CRTTransferConfig(
+                multipart_threshold=multipart_threshold
+            ),
+        )
+        subscribers = [FileSizeProvider(size)]
+        transfer_manager.download(
+            'test_bucket', 'test_key', filename, {}, subscribers
+        ).result()
+        return crt_client.make_request.call_args[1]
+
+    return _download
+
+
+class TestDownloadMultipartThreshold:
+    @pytest.mark.parametrize(
+        'size', [MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD]
+    )
+    def test_within_threshold_downloads_in_single_request(
+        self, download, size
+    ):
+        request_args = download(size, MULTIPART_THRESHOLD)
+        assert request_args['type'] == awscrt.s3.S3RequestType.DEFAULT
+        assert request_args['operation_name'] == 'GetObject'
+
+    def test_above_threshold_splits_download(self, download):
+        request_args = download(MULTIPART_THRESHOLD + 1, MULTIPART_THRESHOLD)
+        assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT
+        assert 'operation_name' not in request_args
+
+    def test_unset_threshold_splits_download(self, download):
+        request_args = download(1, None)
+        assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT
+
+    def test_single_request_download_writes_file(self, download, filename):
+        download(MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD)
+        with open(filename) as f:
+            assert f.read() == DOWNLOADED_CONTENT