Skip to content

Commit 003945f

Browse files
kai-ionkai-ion
committed
Honor requestTimeoutMs for CRT event stream writes
When requestTimeoutMs > 0, compute a deadline at stream creation and use wait_until in SendBuffer(). If the deadline passes, the write fails immediately instead of blocking indefinitely when network drops. When requestTimeoutMs = 0, behavior is unchanged (infinite wait).
1 parent 531fc62 commit 003945f

5 files changed

Lines changed: 33 additions & 9 deletions

File tree

‎generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -261,7 +261,8 @@ void TranscribeStreamingServiceClient::StartCallAnalyticsStreamTranscriptionAsyn
261261

262262
#if AWS_SDK_USE_CRT_HTTP
263263
// Push-based WriteData path (CRT HTTP client only)
264-
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient());
264+
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient(), 8 * 1024,
265+
m_clientConfiguration.requestTimeoutMs);
265266
auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER);
266267

267268
auto eventEncoderStream = Aws::MakeShared<Model::AudioStream>(ALLOCATION_TAG, writeDataStreamBuf);
@@ -361,7 +362,8 @@ void TranscribeStreamingServiceClient::StartMedicalScribeStreamAsync(
361362

362363
#if AWS_SDK_USE_CRT_HTTP
363364
// Push-based WriteData path (CRT HTTP client only)
364-
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient());
365+
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient(), 8 * 1024,
366+
m_clientConfiguration.requestTimeoutMs);
365367
auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER);
366368

367369
auto eventEncoderStream = Aws::MakeShared<Model::MedicalScribeInputStream>(ALLOCATION_TAG, writeDataStreamBuf);
@@ -477,7 +479,8 @@ void TranscribeStreamingServiceClient::StartMedicalStreamTranscriptionAsync(
477479

478480
#if AWS_SDK_USE_CRT_HTTP
479481
// Push-based WriteData path (CRT HTTP client only)
480-
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient());
482+
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient(), 8 * 1024,
483+
m_clientConfiguration.requestTimeoutMs);
481484
auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER);
482485

483486
auto eventEncoderStream = Aws::MakeShared<Model::AudioStream>(ALLOCATION_TAG, writeDataStreamBuf);
@@ -569,7 +572,8 @@ void TranscribeStreamingServiceClient::StartStreamTranscriptionAsync(
569572

570573
#if AWS_SDK_USE_CRT_HTTP
571574
// Push-based WriteData path (CRT HTTP client only)
572-
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient());
575+
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient(), 8 * 1024,
576+
m_clientConfiguration.requestTimeoutMs);
573577
auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER);
574578

575579
auto eventEncoderStream = Aws::MakeShared<Model::AudioStream>(ALLOCATION_TAG, writeDataStreamBuf);

‎src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
#include <aws/core/utils/Array.h>
1111
#include <aws/crt/Types.h>
1212

13+
#include <chrono>
1314
#include <condition_variable>
1415
#include <memory>
1516
#include <mutex>
@@ -35,7 +36,7 @@ namespace Stream {
3536
*/
3637
class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf {
3738
public:
38-
explicit HttpWriteDataStreamBuf(const std::shared_ptr<Aws::Http::HttpClient>& client, size_t bufferLength = 8 * 1024);
39+
explicit HttpWriteDataStreamBuf(const std::shared_ptr<Aws::Http::HttpClient>& client, size_t bufferLength = 8 * 1024, size_t requestTimeoutMs = 0);
3940
HttpWriteDataStreamBuf(const HttpWriteDataStreamBuf& other) = delete;
4041
HttpWriteDataStreamBuf(HttpWriteDataStreamBuf&& other) noexcept = delete;
4142
HttpWriteDataStreamBuf& operator=(const HttpWriteDataStreamBuf& other) = delete;
@@ -102,6 +103,8 @@ class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf {
102103
std::condition_variable m_writeComplete;
103104
bool m_writeInProgress{false};
104105
bool m_writeError{false};
106+
bool m_hasDeadline{false};
107+
std::chrono::steady_clock::time_point m_deadline;
105108

106109
// State management
107110
enum class STATE {

‎src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,15 +5,21 @@
55
#include <aws/core/http/HttpClient.h>
66
#include <aws/core/utils/stream/HttpWriteDataStreamBuf.h>
77

8+
#include <chrono>
89
#include <utility>
910

1011
namespace {
1112
const char* WRITE_DATA_BUF_LOG_NAME = "HttpWriteDataStreamBuf";
1213
}
1314

1415
Aws::Utils::Stream::HttpWriteDataStreamBuf::HttpWriteDataStreamBuf(const std::shared_ptr<Aws::Http::HttpClient>& client,
15-
size_t bufferLength)
16+
size_t bufferLength,
17+
size_t requestTimeoutMs)
1618
: m_client{client}, m_buffer{bufferLength} {
19+
if (requestTimeoutMs > 0) {
20+
m_hasDeadline = true;
21+
m_deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(requestTimeoutMs);
22+
}
1723
ResetPutArea();
1824
}
1925

@@ -163,7 +169,18 @@ bool Aws::Utils::Stream::HttpWriteDataStreamBuf::SendBuffer(bool endStream) {
163169
endStream);
164170

165171
std::unique_lock<std::mutex> lock{m_writeMutex};
166-
m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; });
172+
if (m_hasDeadline) {
173+
bool completed = m_writeComplete.wait_until(lock, m_deadline,
174+
[this]() -> bool { return !m_writeInProgress; });
175+
if (!completed) {
176+
m_writeError = true;
177+
m_writeInProgress = false;
178+
ResetPutArea();
179+
return false;
180+
}
181+
} else {
182+
m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; });
183+
}
167184

168185
ResetPutArea();
169186

‎tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name}
2727
#set($streamModelNameWithFirstLetterCapitalized = $CppViewHelper.capitalizeFirstChar($streamModelName))
2828
\#if AWS_SDK_USE_CRT_HTTP
2929
// Push-based WriteData path (CRT HTTP client only)
30-
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient());
30+
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, GetHttpClient(), 8 * 1024, m_clientConfiguration.requestTimeoutMs);
3131
auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER);
3232

3333
auto eventEncoderStream = Aws::MakeShared<Model::${streamModelType}>(ALLOCATION_TAG, writeDataStreamBuf);

‎tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name}
2929
#set($streamModelNameWithFirstLetterCapitalized = $CppViewHelper.capitalizeFirstChar($streamModelName))
3030
\#if AWS_SDK_USE_CRT_HTTP
3131
// Push-based WriteData path (CRT HTTP client only)
32-
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, m_httpClient);
32+
auto writeDataStreamBuf = Aws::MakeShared<Aws::Utils::Stream::HttpWriteDataStreamBuf>(ALLOCATION_TAG, m_httpClient, 8 * 1024, m_clientConfig->requestTimeoutMs);
3333
auto eventEncoderStream = Aws::MakeShared<Model::${streamModelType}>(ALLOCATION_TAG, writeDataStreamBuf);
3434
request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream);
3535

0 commit comments

Comments
 (0)