Add support for max_request_concurrency

Author: hssyoo

.changes/next-release/enhancement-awscrt-49742.json

@@ -1,5 +1,5 @@
 {
   "type": "enhancement",
   "category": "``awscrt``",
-  "description": "``CRTTransferManager`` now supports the following ``TransferConfig`` options - ``multipart_threshold``, ``multipart_chunksize``"
+  "description": "``CRTTransferManager`` now supports the following ``TransferConfig`` options - ``multipart_threshold``, ``multipart_chunksize``, ``max_request_concurrency``"
 }

s3transfer/crt.py

@@ -753,8 +753,15 @@ def _get_crt_transfer_config_options(self):
         ):
             # Let CRT dynamically calculate part size.
             part_size = None
+        max_connections = self._config.max_request_concurrency
+        if (
+            self._config.get_deep_attr('max_request_concurrency')
+            is self._config.UNSET_DEFAULT
+        ):
+            max_connections = None
         return {
             'part_size': part_size,
+            'max_active_connections_override': max_connections,
         }
 
     def get_make_request_args(

tests/functional/test_crt.py

@@ -776,6 +776,7 @@ def test_transfer_config_used_in_upload_request(self):
         config = TransferConfig(
             multipart_threshold=4 * MB,
             multipart_chunksize=2 * MB,
+            max_request_concurrency=100,
         )
         transfer_manager = s3transfer.crt.CRTTransferManager(
             crt_s3_client=self.s3_crt_client,
@@ -790,11 +791,13 @@ def test_transfer_config_used_in_upload_request(self):
         callargs_kwargs = self.s3_crt_client.make_request.call_args[1]
         assert callargs_kwargs['multipart_upload_threshold'] == 4 * MB
         assert callargs_kwargs['part_size'] == 2 * MB
+        assert callargs_kwargs['max_active_connections_override'] == 100
 
     def test_transfer_config_used_in_download_request(self):
         config = TransferConfig(
             multipart_threshold=4 * MB,
             multipart_chunksize=2 * MB,
+            max_request_concurrency=100,
         )
         transfer_manager = s3transfer.crt.CRTTransferManager(
             crt_s3_client=self.s3_crt_client,
@@ -808,6 +811,7 @@ def test_transfer_config_used_in_download_request(self):
 
         callargs_kwargs = self.s3_crt_client.make_request.call_args[1]
         assert callargs_kwargs['part_size'] == 2 * MB
+        assert callargs_kwargs['max_active_connections_override'] == 100
         # Config option only used for PUT requests.
         assert 'multipart_upload_threshold' not in callargs_kwargs
 
@@ -827,3 +831,20 @@ def test_unset_part_size_defaults_to_none_in_upload_request(self):
 
         callargs_kwargs = self.s3_crt_client.make_request.call_args[1]
         assert callargs_kwargs['part_size'] is None
+
+    def test_unset_max_concurrency_defaults_to_none(self):
+        config = TransferConfig(
+            max_request_concurrency=TransferConfig.UNSET_DEFAULT,
+        )
+        transfer_manager = s3transfer.crt.CRTTransferManager(
+            crt_s3_client=self.s3_crt_client,
+            crt_request_serializer=self.request_serializer,
+            config=config,
+        )
+        future = transfer_manager.upload(
+            self.filename, self.bucket, self.key, {}, []
+        )
+        future.result()
+
+        callargs_kwargs = self.s3_crt_client.make_request.call_args[1]
+        assert callargs_kwargs['max_active_connections_override'] is None