Skip to content

Commit c8c42ac

Browse files
SergeyRyabininSergeyRyabinin
committed
Cancel S3 CRT requests if should not continue
1 parent 948982c commit c8c42ac

File tree

6 files changed

+518
-55
lines changed

6 files changed

+518
-55
lines changed

‎generated/src/aws-cpp-sdk-s3-crt/include/aws/s3-crt/S3CrtClient.h‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6976,14 +6976,12 @@ namespace Aws
69766976
std::shared_ptr<Aws::Utils::Threading::Semaphore> clientShutdownSem;
69776977
};
69786978

6979-
static void CrtClientShutdownCallback(void *data) {
6980-
auto *wrappedData = static_cast<CrtClientShutdownCallbackDataWrapper*>(data);
6981-
if (wrappedData->fn)
6982-
{
6983-
wrappedData->fn(wrappedData->data);
6984-
}
6985-
wrappedData->clientShutdownSem->Release();
6986-
}
6979+
static void CrtClientShutdownCallback(void *data);
6980+
void CancelCrtRequestAsync(aws_s3_meta_request *meta_request) const;
6981+
static int S3CrtRequestHeadersCallback(aws_s3_meta_request *meta_request, const struct aws_http_headers *headers, int response_status, void *user_data);
6982+
static int S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request, const struct aws_byte_cursor *body, uint64_t range_start, void *user_data);
6983+
static void S3CrtRequestProgressCallback(struct aws_s3_meta_request *meta_request, const struct aws_s3_meta_request_progress *progress, void *user_data);
6984+
static void S3CrtRequestFinishCallback(struct aws_s3_meta_request *meta_request, const struct aws_s3_meta_request_result *meta_request_result, void *user_data);
69876985

69886986
void InitCrtEndpointFromUri(aws_uri &endpoint_uri, const Aws::Http::URI &uri) const;
69896987

‎generated/src/aws-cpp-sdk-s3-crt/source/S3CrtClient.cpp‎

Lines changed: 121 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -517,8 +517,24 @@ void S3CrtClient::OverrideEndpoint(const Aws::String& endpoint)
517517
m_endpointProvider->OverrideEndpoint(endpoint);
518518
}
519519

520+
void S3CrtClient::CrtClientShutdownCallback(void *data)
521+
{
522+
auto *wrappedData = static_cast<CrtClientShutdownCallbackDataWrapper*>(data);
523+
if (wrappedData->fn)
524+
{
525+
wrappedData->fn(wrappedData->data);
526+
}
527+
wrappedData->clientShutdownSem->Release();
528+
}
529+
530+
void S3CrtClient::CancelCrtRequestAsync(aws_s3_meta_request *meta_request) const {
531+
assert(meta_request);
532+
m_clientConfiguration.executor->Submit([meta_request]() {
533+
aws_s3_meta_request_cancel(meta_request);
534+
});
535+
}
520536

521-
static int S3CrtRequestHeadersCallback(struct aws_s3_meta_request *meta_request, const struct aws_http_headers *headers,
537+
int S3CrtClient::S3CrtRequestHeadersCallback(struct aws_s3_meta_request *meta_request, const struct aws_http_headers *headers,
522538
int response_status, void *user_data)
523539
{
524540
AWS_UNREFERENCED_PARAM(meta_request);
@@ -531,10 +547,17 @@ static int S3CrtRequestHeadersCallback(struct aws_s3_meta_request *meta_request,
531547
userData->response->AddHeader(StringUtils::FromByteCursor(header.name), StringUtils::FromByteCursor(header.value));
532548
}
533549
userData->response->SetResponseCode(static_cast<HttpResponseCode>(response_status));
550+
551+
auto& shouldContinueFn = userData->originalRequest->GetContinueRequestHandler();
552+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
553+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
554+
userData->s3CrtClient->CancelCrtRequestAsync(meta_request);
555+
}
556+
534557
return AWS_OP_SUCCESS;
535558
}
536559

537-
static int S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request, const struct aws_byte_cursor *body, uint64_t range_start, void *user_data)
560+
int S3CrtClient::S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request, const struct aws_byte_cursor *body, uint64_t range_start, void *user_data)
538561
{
539562
AWS_UNREFERENCED_PARAM(range_start);
540563

@@ -557,26 +580,88 @@ static int S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request,
557580
receivedHandler(userData->request.get(), userData->response.get(), static_cast<long long>(body->len));
558581
}
559582
AWS_LOGSTREAM_TRACE(ALLOCATION_TAG, body->len << " bytes written to response.");
583+
auto& shouldContinueFn = userData->originalRequest->GetContinueRequestHandler();
584+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
585+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
586+
userData->s3CrtClient->CancelCrtRequestAsync(meta_request);
587+
}
560588

561589
return AWS_OP_SUCCESS;
562590
}
563591

564-
static void S3CrtRequestProgressCallback(struct aws_s3_meta_request *meta_request, const struct aws_s3_meta_request_progress *progress, void *user_data)
592+
void S3CrtClient::S3CrtRequestProgressCallback(struct aws_s3_meta_request *meta_request, const struct aws_s3_meta_request_progress *progress, void *user_data)
565593
{
566594
AWS_UNREFERENCED_PARAM(meta_request);
567595
auto *userData = static_cast<S3CrtClient::CrtRequestCallbackUserData*>(user_data);
568596

569597
auto& progressHandler = userData->request->GetDataSentEventHandler();
570598
if (progressHandler)
571599
{
572-
progressHandler(userData->request.get(), static_cast<long long>(progress->bytes_transferred));
600+
progressHandler(userData->request.get(), static_cast<long long>(progress->bytes_transferred));
573601
}
574602
AWS_LOGSTREAM_TRACE(ALLOCATION_TAG, progress->bytes_transferred << " bytes transferred.");
603+
auto& shouldContinueFn = userData->originalRequest->GetContinueRequestHandler();
604+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
605+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
606+
userData->s3CrtClient->CancelCrtRequestAsync(meta_request);
607+
}
575608

576609
return;
577610
}
578611

579-
static void S3CrtRequestFinishCallback(struct aws_s3_meta_request *meta_request,
612+
CoreErrors MapCrtError(const int crtErrorCode) {
613+
switch (crtErrorCode) {
614+
case aws_s3_errors::AWS_ERROR_S3_REQUEST_HAS_COMPLETED:
615+
return CoreErrors::OK;
616+
case aws_s3_errors::AWS_ERROR_S3_MISSING_CONTENT_RANGE_HEADER:
617+
case aws_s3_errors::AWS_ERROR_S3_MISSING_CONTENT_LENGTH_HEADER:
618+
case aws_s3_errors::AWS_ERROR_S3_MISSING_ETAG:
619+
case aws_s3_errors::AWS_ERROR_S3_MISSING_UPLOAD_ID:
620+
return CoreErrors::MISSING_PARAMETER;
621+
case aws_s3_errors::AWS_ERROR_S3_INVALID_CONTENT_RANGE_HEADER:
622+
case aws_s3_errors::AWS_ERROR_S3_INVALID_CONTENT_LENGTH_HEADER:
623+
case aws_s3_errors::AWS_ERROR_S3_INVALID_RANGE_HEADER:
624+
case aws_s3_errors::AWS_ERROR_S3_MULTIRANGE_HEADER_UNSUPPORTED:
625+
case aws_s3_errors::AWS_ERROR_S3_INCORRECT_CONTENT_LENGTH:
626+
case aws_s3_errors::AWS_ERROR_S3_INVALID_MEMORY_LIMIT_CONFIG:
627+
return CoreErrors::INVALID_PARAMETER_VALUE;
628+
case aws_s3_errors::AWS_ERROR_S3_INTERNAL_ERROR:
629+
case aws_s3_errors::AWS_ERROR_S3_PROXY_PARSE_FAILED:
630+
case aws_s3_errors::AWS_ERROR_S3_UNSUPPORTED_PROXY_SCHEME:
631+
case aws_s3_errors::AWS_ERROR_S3_NON_RECOVERABLE_ASYNC_ERROR:
632+
case aws_s3_errors::AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE:
633+
case aws_s3_errors::AWS_ERROR_S3_EXCEEDS_MEMORY_LIMIT:
634+
return CoreErrors::INTERNAL_FAILURE;
635+
case aws_s3_errors::AWS_ERROR_S3_SLOW_DOWN:
636+
return CoreErrors::SLOW_DOWN;
637+
case aws_s3_errors::AWS_ERROR_S3_INVALID_RESPONSE_STATUS:
638+
case aws_s3_errors::AWS_ERROR_S3_RESPONSE_CHECKSUM_MISMATCH:
639+
case aws_s3_errors::AWS_ERROR_S3_CHECKSUM_CALCULATION_FAILED:
640+
case aws_s3_errors::AWS_ERROR_S3_LIST_PARTS_PARSE_FAILED:
641+
case aws_s3_errors::AWS_ERROR_S3_RESUMED_PART_CHECKSUM_MISMATCH:
642+
case aws_s3_errors::AWS_ERROR_S3_FILE_MODIFIED:
643+
case aws_s3_errors::AWS_ERROR_S3_INTERNAL_PART_SIZE_MISMATCH_RETRYING_WITH_RANGE:
644+
case aws_s3_errors::AWS_ERROR_S3_RECV_FILE_ALREADY_EXISTS:
645+
case aws_s3_errors::AWS_ERROR_S3_RECV_FILE_NOT_FOUND:
646+
return CoreErrors::VALIDATION;
647+
case aws_s3_errors::AWS_ERROR_S3_CANCELED:
648+
return CoreErrors::USER_CANCELLED;
649+
case aws_s3_errors::AWS_ERROR_S3_PAUSED:
650+
case aws_s3_errors::AWS_ERROR_S3_RESUME_FAILED:
651+
case aws_s3_errors::AWS_ERROR_S3_OBJECT_MODIFIED:
652+
return CoreErrors::UNKNOWN;
653+
case aws_s3_errors::AWS_ERROR_S3_REQUEST_TIME_TOO_SKEWED:
654+
return CoreErrors::REQUEST_TIME_TOO_SKEWED;
655+
case aws_s3_errors::AWS_ERROR_S3EXPRESS_CREATE_SESSION_FAILED:
656+
return CoreErrors::CLIENT_SIGNING_FAILURE;
657+
case aws_s3_errors::AWS_ERROR_S3_REQUEST_TIMEOUT:
658+
return CoreErrors::REQUEST_TIMEOUT;
659+
default:
660+
return CoreErrors::INTERNAL_FAILURE;
661+
}
662+
}
663+
664+
void S3CrtClient::S3CrtRequestFinishCallback(struct aws_s3_meta_request *meta_request,
580665
const struct aws_s3_meta_request_result *meta_request_result, void *user_data)
581666
{
582667
AWS_UNREFERENCED_PARAM(meta_request);
@@ -623,7 +708,7 @@ static void S3CrtRequestFinishCallback(struct aws_s3_meta_request *meta_request,
623708
<< " (" << aws_error_lib_name(meta_request_result->error_code) << ": " << aws_error_name(meta_request_result->error_code) << ")";
624709

625710
userData->response->SetClientErrorMessage(ss.str());
626-
userData->response->SetClientErrorType(CoreErrors::INTERNAL_FAILURE);
711+
userData->response->SetClientErrorType(MapCrtError(meta_request_result->error_code));
627712
}
628713

629714
aws_s3_meta_request_release(meta_request);
@@ -876,9 +961,16 @@ void S3CrtClient::CopyObjectAsync(const CopyObjectRequest& request, const CopyOb
876961
options.message= crtHttpRequest->GetUnderlyingMessage();
877962
userData->crtHttpRequest = crtHttpRequest;
878963

879-
if (aws_s3_client_make_meta_request(m_s3CrtClient, &options) == nullptr)
880-
{
881-
return handler(this, request, CopyObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE", "Unable to create s3 meta request", false)), handlerContext);
964+
aws_s3_meta_request* meta_request = aws_s3_client_make_meta_request(m_s3CrtClient, &options);
965+
if (meta_request == nullptr) {
966+
return handler(this, request,
967+
CopyObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE",
968+
"Unable to create s3 meta request", false)), handlerContext);
969+
}
970+
auto& shouldContinueFn = request.GetContinueRequestHandler();
971+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
972+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
973+
aws_s3_meta_request_cancel(meta_request);
882974
}
883975
}
884976

@@ -1021,9 +1113,16 @@ void S3CrtClient::GetObjectAsync(const GetObjectRequest& request, const GetObjec
10211113
options.message= crtHttpRequest->GetUnderlyingMessage();
10221114
userData->crtHttpRequest = crtHttpRequest;
10231115

1024-
if (aws_s3_client_make_meta_request(m_s3CrtClient, &options) == nullptr)
1025-
{
1026-
return handler(this, request, GetObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE", "Unable to create s3 meta request", false)), handlerContext);
1116+
aws_s3_meta_request* meta_request = aws_s3_client_make_meta_request(m_s3CrtClient, &options);
1117+
if (meta_request == nullptr) {
1118+
return handler(this, request,
1119+
GetObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE",
1120+
"Unable to create s3 meta request", false)), handlerContext);
1121+
}
1122+
auto& shouldContinueFn = request.GetContinueRequestHandler();
1123+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
1124+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
1125+
aws_s3_meta_request_cancel(meta_request);
10271126
}
10281127
}
10291128

@@ -1201,9 +1300,16 @@ void S3CrtClient::PutObjectAsync(const PutObjectRequest& request, const PutObjec
12011300
options.message= crtHttpRequest->GetUnderlyingMessage();
12021301
userData->crtHttpRequest = crtHttpRequest;
12031302

1204-
if (aws_s3_client_make_meta_request(m_s3CrtClient, &options) == nullptr)
1205-
{
1206-
return handler(this, request, PutObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE", "Unable to create s3 meta request", false)), handlerContext);
1303+
aws_s3_meta_request* meta_request = aws_s3_client_make_meta_request(m_s3CrtClient, &options);
1304+
if (meta_request == nullptr) {
1305+
return handler(this, request,
1306+
PutObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE",
1307+
"Unable to create s3 meta request", false)), handlerContext);
1308+
}
1309+
auto& shouldContinueFn = request.GetContinueRequestHandler();
1310+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
1311+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
1312+
aws_s3_meta_request_cancel(meta_request);
12071313
}
12081314
}
12091315

‎tests/aws-cpp-sdk-s3-crt-integration-tests/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ foreach(TEST IN LISTS TEST_LIST)
2020
"${TEST_MAIN_FILE}"
2121
"BucketAndObjectOperationTest.cpp"
2222
"S3ExpressTest.cpp"
23+
"CancelCrtRequestTest.cpp"
2324
)
2425

2526
file(GLOB AWS_S3_CRT_INTEGRATION_TESTS_SRC

0 commit comments

Comments
 (0)