From 733693ad31c094d1ff5fe44cebeeb7aea65555be Mon Sep 17 00:00:00 2001 From: sbaluja Date: Wed, 16 Sep 2026 16:57:22 -0400 Subject: [PATCH 1/4] Bound bidirectional streaming by requestTimeoutMs (per-write + time-to-first-byte) (#XXXX) HttpWriteDataStreamBuf bounds each write by requestTimeoutMs and bounds the wait for the first response byte; once a response arrives the wait is unbounded so healthy long-lived streams are not cut. On timeout it closes the connection to drive the real CRT completion callback (never fabricates completion). Both the smithy task and the legacy path signal NotifyResponseStarted() on headers. --- .../client/AWSClientBidirectionalStreaming.h | 6 +- .../utils/stream/HttpWriteDataStreamBuf.h | 13 ++- ...mithyBidirectionalStreamingWriteDataTask.h | 6 +- .../utils/stream/HttpWriteDataStreamBuf.cpp | 48 +++++++- .../stream/HttpWriteDataStreamBufTest.cpp | 106 ++++++++++++++++++ .../aws/testing/mocks/http/MockConnection.h | 46 +++++++- .../aws/testing/mocks/http/MockHttpClient.h | 4 + 7 files changed, 218 insertions(+), 11 deletions(-) diff --git a/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h b/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h index d02aa78debdf..72965722dfb0 100644 --- a/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h +++ b/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h @@ -70,8 +70,12 @@ void SubmitBidirectionalStreamingRequest( requestCopy->SetEventStreamHandler(requestCopy->GetEventStreamHandler()); // Wire initial response handler on httpRequest (CRT reads it from there) + std::weak_ptr wBuf = writeDataStreamBuf; httpRequest->SetHeadersReceivedEventHandler( - [wReq](const Aws::Http::HttpRequest*, Aws::Http::HttpResponse* response) { + [wReq, wBuf](const Aws::Http::HttpRequest*, Aws::Http::HttpResponse* response) { + if (auto buf = wBuf.lock()) { + buf->NotifyResponseStarted(); + } auto req = wReq.lock(); if (!req || !response) return; auto& cb = req->GetEventStreamHandler().GetInitialResponseCallbackEx(); diff --git a/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h b/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h index d76c463db6d8..86c98888a92d 100644 --- a/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h +++ b/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h @@ -10,7 +10,9 @@ #include #include +#include #include +#include #include #include @@ -35,7 +37,9 @@ namespace Stream { */ class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf { public: - explicit HttpWriteDataStreamBuf(const std::shared_ptr& client, size_t bufferLength = 8 * 1024); + explicit HttpWriteDataStreamBuf(const std::shared_ptr& client, + size_t bufferLength = 8 * 1024, + uint64_t requestTimeoutMs = 0); HttpWriteDataStreamBuf(const HttpWriteDataStreamBuf& other) = delete; HttpWriteDataStreamBuf(HttpWriteDataStreamBuf&& other) noexcept = delete; HttpWriteDataStreamBuf& operator=(const HttpWriteDataStreamBuf& other) = delete; @@ -65,6 +69,9 @@ class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf { */ void WaitForStreamComplete(); + /** Marks that the first response byte/headers arrived; lifts WaitForStreamComplete()'s first-byte timeout. */ + void NotifyResponseStarted(); + protected: // Write support int_type overflow(int_type c) override; @@ -88,6 +95,7 @@ class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf { */ bool SendBuffer(bool endStream = false); void ResetPutArea(); + void CloseConnection(); // Client state std::shared_ptr m_client; @@ -112,6 +120,9 @@ class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf { std::condition_variable m_shutdownCondition; std::mutex m_shutdownMutex; bool m_streamComplete{false}; + bool m_responseStarted{false}; + + std::chrono::milliseconds m_writeTimeout{0}; }; } // namespace Stream } // namespace Utils diff --git a/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h b/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h index 76f886f1cb08..dee6ad1104e7 100644 --- a/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h +++ b/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h @@ -118,8 +118,12 @@ class AWS_CORE_LOCAL SmithyBidirectionalStreamingWriteDataTask final { // Wire initial response handler on httpRequest (CRT reads it from here) std::weak_ptr wReq = m_request; + std::weak_ptr wBuf = m_writeDataStreamBuf; httpRequest->SetHeadersReceivedEventHandler( - [wReq](const Aws::Http::HttpRequest*, Aws::Http::HttpResponse* response) { + [wReq, wBuf](const Aws::Http::HttpRequest*, Aws::Http::HttpResponse* response) { + if (auto buf = wBuf.lock()) { + buf->NotifyResponseStarted(); + } auto req = wReq.lock(); if (!req || !response) return; auto& cb = req->GetEventStreamHandler().GetInitialResponseCallbackEx(); diff --git a/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp b/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp index 1734e1c506ab..bca0e0cfa41a 100644 --- a/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp +++ b/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp @@ -5,6 +5,7 @@ #include #include +#include #include namespace { @@ -12,8 +13,9 @@ const char* WRITE_DATA_BUF_LOG_NAME = "HttpWriteDataStreamBuf"; } Aws::Utils::Stream::HttpWriteDataStreamBuf::HttpWriteDataStreamBuf(const std::shared_ptr& client, - size_t bufferLength) - : m_client{client}, m_buffer{bufferLength} { + size_t bufferLength, + uint64_t requestTimeoutMs) + : m_client{client}, m_buffer{bufferLength}, m_writeTimeout{requestTimeoutMs} { ResetPutArea(); } @@ -102,12 +104,41 @@ void Aws::Utils::Stream::HttpWriteDataStreamBuf::WaitForStreamComplete() { if (m_state == STATE::UNINITIALIZED) { return; } + if (m_writeTimeout.count() > 0 && !m_responseStarted) { + const auto deadline = std::chrono::steady_clock::now() + m_writeTimeout; + if (!m_shutdownCondition.wait_until(lock, deadline, + [this]() -> bool { return m_streamComplete || m_responseStarted; })) { + m_writeError = true; + lock.unlock(); + CloseConnection(); + lock.lock(); + } + } m_shutdownCondition.wait(lock, [this]() -> bool { return m_streamComplete; }); m_stream.reset(); m_connection.reset(); } +void Aws::Utils::Stream::HttpWriteDataStreamBuf::NotifyResponseStarted() { + { + std::unique_lock const lock{m_shutdownMutex}; + m_responseStarted = true; + } + m_shutdownCondition.notify_all(); +} + +void Aws::Utils::Stream::HttpWriteDataStreamBuf::CloseConnection() { + std::shared_ptr connection; + { + std::unique_lock const lock{m_shutdownMutex}; + connection = m_connection; + } + if (connection) { + connection->Close(); + } +} + std::streambuf::int_type Aws::Utils::Stream::HttpWriteDataStreamBuf::overflow(std::streambuf::int_type c) { if (traits_type::eq_int_type(c, traits_type::eof())) { return traits_type::not_eof(c); @@ -173,7 +204,18 @@ bool Aws::Utils::Stream::HttpWriteDataStreamBuf::SendBuffer(bool endStream) { endStream); std::unique_lock lock{m_writeMutex}; - m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; }); + if (m_writeTimeout.count() > 0) { + const auto deadline = std::chrono::steady_clock::now() + m_writeTimeout; + if (!m_writeComplete.wait_until(lock, deadline, [this]() -> bool { return !m_writeInProgress; })) { + m_writeError = true; + lock.unlock(); + CloseConnection(); + lock.lock(); + m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; }); + } + } else { + m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; }); + } ResetPutArea(); diff --git a/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp b/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp index ef6d054a0ae4..01e349aefc59 100644 --- a/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp +++ b/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp @@ -6,6 +6,9 @@ #include #include +#include +#include + namespace { const char* TEST_ALLOCATION_LOG_TAG = "HttpWriteDataStreamBufTest"; } @@ -162,6 +165,55 @@ TEST_F(HttpWriteDataStreamBufTest, TestGetResponse) { EXPECT_EQ(resp->GetResponseCode(), Aws::Http::HttpResponseCode::OK); } +TEST_F(HttpWriteDataStreamBufTest, TestTimeoutSetPromptWriteUnaffected) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + auto closeCount = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, 0); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + testCase.closeCount = closeCount; + client_->SetConnectionTestCase(testCase); + { + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10, 1000}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + stream << "float on okay"; + } + EXPECT_STREQ(output->str().c_str(), "float on okay"); + EXPECT_EQ(*closeCount, 0); +} + +TEST_F(HttpWriteDataStreamBufTest, TestPerWriteTimeoutClosesConnection) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + auto closeCount = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, 0); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + testCase.withholdWriteComplete = true; + testCase.closeCount = closeCount; + client_->SetConnectionTestCase(testCase); + { + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10, 50}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + const auto start = std::chrono::steady_clock::now(); + stream << "a stalled write that never completes on its own"; + stream.flush(); + const auto elapsedMs = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start).count(); + EXPECT_GE(elapsedMs, 40); + EXPECT_LT(elapsedMs, 5000); + EXPECT_TRUE(stream.fail()); + } + EXPECT_EQ(*closeCount, 1); +} + TEST_F(HttpWriteDataStreamBufTest, TestExactBufferBoundary) { auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, @@ -180,3 +232,57 @@ TEST_F(HttpWriteDataStreamBufTest, TestExactBufferBoundary) { EXPECT_STREQ(output->str().c_str(), "well we are"); } + +TEST_F(HttpWriteDataStreamBufTest, TestFirstByteTimeoutClosesConnection) { + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto closeCount = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, 0); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + testCase.withholdStreamComplete = true; + testCase.closeCount = closeCount; + client_->SetConnectionTestCase(testCase); + + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10, 50}; + data_stream_buf.Initialize(request); + const auto start = std::chrono::steady_clock::now(); + data_stream_buf.WaitForStreamComplete(); + const auto elapsedMs = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start).count(); + EXPECT_GE(elapsedMs, 40); + EXPECT_LT(elapsedMs, 5000); + EXPECT_EQ(*closeCount, 1); +} + +TEST_F(HttpWriteDataStreamBufTest, TestFirstByteReceivedThenUnbounded) { + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto closeCount = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, 0); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + testCase.withholdStreamComplete = true; + testCase.closeCount = closeCount; + client_->SetConnectionTestCase(testCase); + + auto buf = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, client_, 10, 50); + buf->Initialize(request); + buf->NotifyResponseStarted(); + + std::atomic returned{false}; + std::thread waiter([&]() { buf->WaitForStreamComplete(); returned = true; }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + EXPECT_FALSE(returned.load()); + EXPECT_EQ(*closeCount, 0); + + client_->GetLastConnection()->FireStreamComplete(0); + waiter.join(); + EXPECT_TRUE(returned.load()); + EXPECT_EQ(*closeCount, 0); +} diff --git a/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h b/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h index 87e08c7fb805..1cd23285995b 100644 --- a/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h +++ b/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h @@ -7,6 +7,7 @@ #include #include #include +#include #include #include @@ -19,6 +20,9 @@ struct ConnectionTestCase { int writeDataCompleteErrorCode{0}; std::shared_ptr writeDataStream; std::shared_ptr response; + bool withholdWriteComplete{false}; + bool withholdStreamComplete{false}; + std::shared_ptr closeCount; }; class MockClientStream : public Aws::Http::ClientStream { @@ -31,8 +35,12 @@ class MockClientStream : public Aws::Http::ClientStream { int WriteData(std::shared_ptr stream, const std::function& onComplete, bool endStream) override { *m_testCase.writeDataStream << stream->rdbuf(); - onComplete(m_testCase.writeDataCompleteErrorCode); - if (endStream) { + if (m_testCase.withholdWriteComplete) { + m_pendingWriteComplete = onComplete; + } else { + onComplete(m_testCase.writeDataCompleteErrorCode); + } + if (endStream && !m_testCase.withholdStreamComplete) { m_onStreamComplete(m_testCase.streamCompleteErrorCode); } return m_testCase.writeDataErrorCode; @@ -40,9 +48,21 @@ class MockClientStream : public Aws::Http::ClientStream { std::shared_ptr GetResponse() const override { return m_testCase.response; } + void FireCompletionOnClose() { + if (m_pendingWriteComplete) { + auto cb = m_pendingWriteComplete; + m_pendingWriteComplete = nullptr; + cb(1); + } + m_onStreamComplete(1); + } + + void FireStreamComplete(int errorCode) { m_onStreamComplete(errorCode); } + private: ConnectionTestCase m_testCase; std::function m_onStreamComplete; + std::function m_pendingWriteComplete; }; class MockConnection : public Aws::Http::Connection { @@ -52,11 +72,27 @@ class MockConnection : public Aws::Http::Connection { std::shared_ptr NewClientStream(const std::shared_ptr& request, std::function onStreamComplete) override { AWS_UNREFERENCED_PARAM(request); - return Aws::MakeShared("MockConnection", m_testCase, std::move(onStreamComplete)); + auto stream = Aws::MakeShared("MockConnection", m_testCase, std::move(onStreamComplete)); + m_stream = stream; + return stream; + } + + void Close() override { + if (m_testCase.closeCount) { + ++(*m_testCase.closeCount); + } + if (auto stream = m_stream.lock()) { + static_cast(stream.get())->FireCompletionOnClose(); + } } - void Close() override {} + void FireStreamComplete(int errorCode) { + if (auto stream = m_stream.lock()) { + static_cast(stream.get())->FireStreamComplete(errorCode); + } + } private: ConnectionTestCase m_testCase; -}; \ No newline at end of file + std::weak_ptr m_stream; +}; diff --git a/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h b/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h index 4192d1377b9f..64c324ed6a12 100644 --- a/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h +++ b/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h @@ -94,15 +94,19 @@ class MockHttpClient : public Aws::Http::HttpClient const std::function, int)>& onClientConnectionAvailable) override { AWS_UNREFERENCED_PARAM(request); auto connection = Aws::MakeShared(MockHttpAllocationTag, m_connectionTestCase); + m_lastConnection = connection; onClientConnectionAvailable(connection, m_connectionTestCase.connectionErrorCode); return m_connectionTestCase.connectionError; } + std::shared_ptr GetLastConnection() const { return m_lastConnection; } + private: mutable ConnectionTestCase m_connectionTestCase; mutable Aws::Vector m_requestsMade; mutable Aws::Queue m_responsesToUse; mutable Aws::Queue m_responseAndRequestsCallback; + mutable std::shared_ptr m_lastConnection; }; class MockHttpClientFactory : public Aws::Http::HttpClientFactory From 549654f64e7d06d15882d14ddf5355fde22b5651 Mon Sep 17 00:00:00 2001 From: sbaluja Date: Wed, 16 Sep 2026 16:57:22 -0400 Subject: [PATCH 2/4] Pass requestTimeoutMs into bidirectional write buffer in event-stream templates (#XXXX) --- .../velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm | 2 +- .../cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm index a71775efd9eb..e9272d24db28 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm @@ -27,7 +27,7 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} #set($streamModelNameWithFirstLetterCapitalized = $CppViewHelper.capitalizeFirstChar($streamModelName)) \#if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm index dbfc4949f72f..6036ec6a8298 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm @@ -29,7 +29,7 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} #set($streamModelNameWithFirstLetterCapitalized = $CppViewHelper.capitalizeFirstChar($streamModelName)) \#if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, m_httpClient); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, m_httpClient, 8 * 1024, m_clientConfiguration.requestTimeoutMs); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); From a2f40bb728e84463e65333254f5578adaffc5fbc Mon Sep 17 00:00:00 2001 From: sbaluja Date: Wed, 16 Sep 2026 16:57:23 -0400 Subject: [PATCH 3/4] Regenerate bidirectional clients for write timeout (#XXXX) --- .../source/BedrockRuntimeClient.cpp | 3 ++- .../source/ConnectHealthClient.cpp | 3 ++- .../source/LexRuntimeV2Client.cpp | 3 ++- .../src/aws-cpp-sdk-polly/source/PollyClient.cpp | 3 ++- .../aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp | 3 ++- .../source/SageMakerRuntimeHTTP2Client.cpp | 3 ++- .../source/TranscribeStreamingServiceClient.cpp | 12 ++++++++---- 7 files changed, 20 insertions(+), 10 deletions(-) diff --git a/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp b/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp index 6a3a0e6abc74..2e8bb3bbf1c1 100644 --- a/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp +++ b/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp @@ -400,7 +400,8 @@ void BedrockRuntimeClient::InvokeModelWithBidirectionalStreamAsync( #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, m_httpClient); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, m_httpClient, 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); request.SetBody(eventEncoderStream); diff --git a/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp b/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp index de1f0790a917..7039f151681e 100644 --- a/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp +++ b/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp @@ -522,7 +522,8 @@ void ConnectHealthClient::StartMedicalScribeListeningSessionAsync( #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); diff --git a/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp b/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp index 0d3fafbc61e7..52729e1932ee 100644 --- a/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp +++ b/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp @@ -494,7 +494,8 @@ void LexRuntimeV2Client::StartConversationAsync(Model::StartConversationRequest& #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); diff --git a/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp b/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp index c950a69309a6..eaf81d478374 100644 --- a/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp +++ b/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp @@ -345,7 +345,8 @@ void PollyClient::StartSpeechSynthesisStreamAsync(Model::StartSpeechSynthesisStr #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); diff --git a/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp b/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp index 506f8157add8..a06045498ee2 100644 --- a/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp +++ b/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp @@ -398,7 +398,8 @@ void QBusinessClient::ChatAsync(Model::ChatRequest& request, const ChatStreamRea #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); diff --git a/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp b/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp index c85754599b95..b97cce0cbd5a 100644 --- a/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp +++ b/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp @@ -232,7 +232,8 @@ void SageMakerRuntimeHTTP2Client::InvokeEndpointWithBidirectionalStreamAsync( #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); diff --git a/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp b/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp index 8917477057cf..7ade3c14b225 100644 --- a/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp +++ b/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp @@ -261,7 +261,8 @@ void TranscribeStreamingServiceClient::StartCallAnalyticsStreamTranscriptionAsyn #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); @@ -361,7 +362,8 @@ void TranscribeStreamingServiceClient::StartMedicalScribeStreamAsync( #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); @@ -477,7 +479,8 @@ void TranscribeStreamingServiceClient::StartMedicalStreamTranscriptionAsync( #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); @@ -569,7 +572,8 @@ void TranscribeStreamingServiceClient::StartStreamTranscriptionAsync( #if AWS_SDK_USE_CRT_HTTP // Push-based WriteData path (CRT HTTP client only) - auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, + m_clientConfiguration.requestTimeoutMs); auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); From 7d1167307a5a4340cd67ea3566457c7b76f636ec Mon Sep 17 00:00:00 2001 From: sbaluja Date: Thu, 17 Sep 2026 12:31:23 -0400 Subject: [PATCH 4/4] use RAII locking --- .../utils/stream/HttpWriteDataStreamBuf.cpp | 56 +++++++++++-------- 1 file changed, 34 insertions(+), 22 deletions(-) diff --git a/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp b/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp index bca0e0cfa41a..6833ac214b19 100644 --- a/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp +++ b/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp @@ -100,22 +100,28 @@ std::shared_ptr Aws::Utils::Stream::HttpWriteDataStream } void Aws::Utils::Stream::HttpWriteDataStreamBuf::WaitForStreamComplete() { - std::unique_lock lock{m_shutdownMutex}; - if (m_state == STATE::UNINITIALIZED) { - return; - } - if (m_writeTimeout.count() > 0 && !m_responseStarted) { - const auto deadline = std::chrono::steady_clock::now() + m_writeTimeout; - if (!m_shutdownCondition.wait_until(lock, deadline, - [this]() -> bool { return m_streamComplete || m_responseStarted; })) { - m_writeError = true; - lock.unlock(); - CloseConnection(); - lock.lock(); + bool timedOut = false; + { + std::unique_lock lock{m_shutdownMutex}; + if (m_state == STATE::UNINITIALIZED) { + return; + } + if (m_writeTimeout.count() > 0 && !m_responseStarted) { + const auto deadline = std::chrono::steady_clock::now() + m_writeTimeout; + if (!m_shutdownCondition.wait_until(lock, deadline, + [this]() -> bool { return m_streamComplete || m_responseStarted; })) { + m_writeError = true; + timedOut = true; + } } } - m_shutdownCondition.wait(lock, [this]() -> bool { return m_streamComplete; }); + if (timedOut) { + CloseConnection(); + } + + std::unique_lock lock{m_shutdownMutex}; + m_shutdownCondition.wait(lock, [this]() -> bool { return m_streamComplete; }); m_stream.reset(); m_connection.reset(); } @@ -203,17 +209,23 @@ bool Aws::Utils::Stream::HttpWriteDataStreamBuf::SendBuffer(bool endStream) { }, endStream); - std::unique_lock lock{m_writeMutex}; - if (m_writeTimeout.count() > 0) { - const auto deadline = std::chrono::steady_clock::now() + m_writeTimeout; - if (!m_writeComplete.wait_until(lock, deadline, [this]() -> bool { return !m_writeInProgress; })) { - m_writeError = true; - lock.unlock(); - CloseConnection(); - lock.lock(); + bool timedOut = false; + { + std::unique_lock lock{m_writeMutex}; + if (m_writeTimeout.count() > 0) { + const auto deadline = std::chrono::steady_clock::now() + m_writeTimeout; + if (!m_writeComplete.wait_until(lock, deadline, [this]() -> bool { return !m_writeInProgress; })) { + m_writeError = true; + timedOut = true; + } + } else { m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; }); } - } else { + } + + if (timedOut) { + CloseConnection(); + std::unique_lock lock{m_writeMutex}; m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; }); }