Skip to content

Commit a94edca

Browse files
SergeyRyabininSergeyRyabinin
authored
Cancel S3 CRT requests if should not continue (#3231)
1 parent b5ec369 commit a94edca

File tree

6 files changed

+546
-63
lines changed

6 files changed

+546
-63
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: 135 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -517,28 +517,59 @@ void S3CrtClient::OverrideEndpoint(const Aws::String& endpoint)
517517
m_endpointProvider->OverrideEndpoint(endpoint);
518518
}
519519

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

521-
static int S3CrtRequestHeadersCallback(struct aws_s3_meta_request *meta_request, const struct aws_http_headers *headers,
539+
int S3CrtClient::S3CrtRequestHeadersCallback(struct aws_s3_meta_request *meta_request, const struct aws_http_headers *headers,
522540
int response_status, void *user_data)
523541
{
524542
AWS_UNREFERENCED_PARAM(meta_request);
525543
auto *userData = static_cast<S3CrtClient::CrtRequestCallbackUserData*>(user_data);
544+
if (!userData || !userData->response || !userData->originalRequest) {
545+
return AWS_OP_ERR;
546+
}
547+
526548
size_t headersCount = aws_http_headers_count(headers);
527-
for (size_t i = 0; i < headersCount; i++)
528-
{
549+
for (size_t i = 0; i < headersCount; i++) {
529550
struct aws_http_header header;
530551
aws_http_headers_get_index(headers, i, &header);
531552
userData->response->AddHeader(StringUtils::FromByteCursor(header.name), StringUtils::FromByteCursor(header.value));
532553
}
533554
userData->response->SetResponseCode(static_cast<HttpResponseCode>(response_status));
555+
556+
auto& shouldContinueFn = userData->originalRequest->GetContinueRequestHandler();
557+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
558+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
559+
userData->s3CrtClient->CancelCrtRequestAsync(meta_request);
560+
}
561+
534562
return AWS_OP_SUCCESS;
535563
}
536564

537-
static int S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request, const struct aws_byte_cursor *body, uint64_t range_start, void *user_data)
565+
int S3CrtClient::S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request, const struct aws_byte_cursor *body, uint64_t range_start, void *user_data)
538566
{
539567
AWS_UNREFERENCED_PARAM(range_start);
540568

541569
auto *userData = static_cast<S3CrtClient::CrtRequestCallbackUserData*>(user_data);
570+
if (!userData || !userData->response || !userData->request) {
571+
return AWS_OP_ERR;
572+
}
542573
auto& bodyStream = userData->response->GetResponseBody();
543574

544575
bodyStream.write(reinterpret_cast<char*>(body->ptr), static_cast<std::streamsize>(body->len));
@@ -557,26 +588,90 @@ static int S3CrtRequestGetBodyCallback(struct aws_s3_meta_request *meta_request,
557588
receivedHandler(userData->request.get(), userData->response.get(), static_cast<long long>(body->len));
558589
}
559590
AWS_LOGSTREAM_TRACE(ALLOCATION_TAG, body->len << " bytes written to response.");
591+
auto& shouldContinueFn = userData->originalRequest->GetContinueRequestHandler();
592+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
593+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
594+
userData->s3CrtClient->CancelCrtRequestAsync(meta_request);
595+
}
560596

561597
return AWS_OP_SUCCESS;
562598
}
563599

564-
static void S3CrtRequestProgressCallback(struct aws_s3_meta_request *meta_request, const struct aws_s3_meta_request_progress *progress, void *user_data)
600+
void S3CrtClient::S3CrtRequestProgressCallback(struct aws_s3_meta_request *meta_request, const struct aws_s3_meta_request_progress *progress, void *user_data)
565601
{
566602
AWS_UNREFERENCED_PARAM(meta_request);
603+
AWS_CHECK_PTR(SERVICE_NAME, user_data);
567604
auto *userData = static_cast<S3CrtClient::CrtRequestCallbackUserData*>(user_data);
568605

606+
AWS_CHECK_PTR(SERVICE_NAME, userData->request);
569607
auto& progressHandler = userData->request->GetDataSentEventHandler();
570-
if (progressHandler)
571-
{
572-
progressHandler(userData->request.get(), static_cast<long long>(progress->bytes_transferred));
608+
if (progressHandler) {
609+
progressHandler(userData->request.get(), static_cast<long long>(progress->bytes_transferred));
573610
}
574611
AWS_LOGSTREAM_TRACE(ALLOCATION_TAG, progress->bytes_transferred << " bytes transferred.");
612+
AWS_CHECK_PTR(SERVICE_NAME, userData->originalRequest);
613+
auto& shouldContinueFn = userData->originalRequest->GetContinueRequestHandler();
614+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
615+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
616+
userData->s3CrtClient->CancelCrtRequestAsync(meta_request);
617+
}
575618

576619
return;
577620
}
578621

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

625720
userData->response->SetClientErrorMessage(ss.str());
626-
userData->response->SetClientErrorType(CoreErrors::INTERNAL_FAILURE);
721+
userData->response->SetClientErrorType(MapCrtError(meta_request_result->error_code));
627722
}
628723

629724
aws_s3_meta_request_release(meta_request);
@@ -876,9 +971,16 @@ void S3CrtClient::CopyObjectAsync(const CopyObjectRequest& request, const CopyOb
876971
options.message= crtHttpRequest->GetUnderlyingMessage();
877972
userData->crtHttpRequest = crtHttpRequest;
878973

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);
974+
aws_s3_meta_request* meta_request = aws_s3_client_make_meta_request(m_s3CrtClient, &options);
975+
if (meta_request == nullptr) {
976+
return handler(this, request,
977+
CopyObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE",
978+
"Unable to create s3 meta request", false)), handlerContext);
979+
}
980+
auto& shouldContinueFn = request.GetContinueRequestHandler();
981+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
982+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
983+
aws_s3_meta_request_cancel(meta_request);
882984
}
883985
}
884986

@@ -1021,9 +1123,16 @@ void S3CrtClient::GetObjectAsync(const GetObjectRequest& request, const GetObjec
10211123
options.message= crtHttpRequest->GetUnderlyingMessage();
10221124
userData->crtHttpRequest = crtHttpRequest;
10231125

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);
1126+
aws_s3_meta_request* meta_request = aws_s3_client_make_meta_request(m_s3CrtClient, &options);
1127+
if (meta_request == nullptr) {
1128+
return handler(this, request,
1129+
GetObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE",
1130+
"Unable to create s3 meta request", false)), handlerContext);
1131+
}
1132+
auto& shouldContinueFn = request.GetContinueRequestHandler();
1133+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
1134+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
1135+
aws_s3_meta_request_cancel(meta_request);
10271136
}
10281137
}
10291138

@@ -1201,9 +1310,16 @@ void S3CrtClient::PutObjectAsync(const PutObjectRequest& request, const PutObjec
12011310
options.message= crtHttpRequest->GetUnderlyingMessage();
12021311
userData->crtHttpRequest = crtHttpRequest;
12031312

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);
1313+
aws_s3_meta_request* meta_request = aws_s3_client_make_meta_request(m_s3CrtClient, &options);
1314+
if (meta_request == nullptr) {
1315+
return handler(this, request,
1316+
PutObjectOutcome(Aws::Client::AWSError<S3CrtErrors>(S3CrtErrors::INTERNAL_FAILURE, "INTERNAL_FAILURE",
1317+
"Unable to create s3 meta request", false)), handlerContext);
1318+
}
1319+
auto& shouldContinueFn = request.GetContinueRequestHandler();
1320+
const HttpRequest* httpRequest = userData->request ? userData->request.get() : nullptr;
1321+
if (shouldContinueFn && !shouldContinueFn(httpRequest)) {
1322+
aws_s3_meta_request_cancel(meta_request);
12071323
}
12081324
}
12091325

‎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)