Skip to content

Commit 8416718

Browse files
kai-ionkai-ion
committed
Implement Retry Behavior 2.1 core logic gated behind AWS_NEW_RETRIES_2026
1 parent e53cb43 commit 8416718

6 files changed

Lines changed: 635 additions & 42 deletions

File tree

‎src/aws-cpp-sdk-core/include/aws/core/client/AdaptiveRetryStrategy.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,8 @@ class AWS_CORE_API AdaptiveRetryStrategy : public StandardRetryStrategy
141141

142142
const char* GetStrategyName() const override { return "adaptive";}
143143

144+
AdaptiveRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts, double transientBackoffBaseSec);
145+
144146
protected:
145147
RetryTokenBucket m_retryTokenBucket;
146148
bool m_fastFail = false;

‎src/aws-cpp-sdk-core/include/aws/core/client/RetryStrategy.h‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
#pragma once
77

88
#include <aws/core/Core_EXPORTS.h>
9+
#include <aws/core/utils/memory/AWSMemory.h>
910
#include <aws/core/utils/threading/ReaderWriterLock.h>
1011
#include <memory>
1112

@@ -102,10 +103,17 @@ namespace Aws
102103
virtual int GetRetryQuota() const = 0;
103104
};
104105

106+
enum class RetryCostClassification
107+
{
108+
REQUEST_TIMEOUT_BASED,
109+
THROTTLE_BASED
110+
};
111+
105112
class AWS_CORE_API DefaultRetryQuotaContainer : public RetryQuotaContainer
106113
{
107114
public:
108115
DefaultRetryQuotaContainer();
116+
DefaultRetryQuotaContainer(int retryCost, int throttlingRetryCost, RetryCostClassification classification);
109117
virtual ~DefaultRetryQuotaContainer() = default;
110118
virtual bool AcquireRetryQuota(int capacityAmount) override;
111119
virtual bool AcquireRetryQuota(const AWSError<CoreErrors>& error) override;
@@ -116,13 +124,18 @@ namespace Aws
116124
protected:
117125
mutable Aws::Utils::Threading::ReaderWriterLock m_retryQuotaLock;
118126
int m_retryQuota;
127+
int m_retryCost;
128+
int m_throttlingRetryCost;
129+
RetryCostClassification m_classification;
119130
};
120131

121132
class AWS_CORE_API StandardRetryStrategy : public RetryStrategy
122133
{
123134
public:
124135
StandardRetryStrategy(long maxAttempts = 3);
125136
StandardRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts = 3);
137+
StandardRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts, double transientBackoffBaseSec);
138+
virtual ~StandardRetryStrategy();
126139

127140
virtual void RequestBookkeeping(const HttpResponseOutcome& httpResponseOutcome) override;
128141
virtual void RequestBookkeeping(const HttpResponseOutcome& httpResponseOutcome, const AWSError<CoreErrors>& lastError) override;
@@ -138,6 +151,10 @@ namespace Aws
138151
protected:
139152
std::shared_ptr<RetryQuotaContainer> m_retryQuotaContainer;
140153
long m_maxAttempts;
154+
155+
private:
156+
struct RetryImpl;
157+
Aws::UniquePtr<RetryImpl> m_impl;
141158
};
142159
} // namespace Client
143160
} // namespace Aws

‎src/aws-cpp-sdk-core/source/client/AdaptiveRetryStrategy.cpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,10 @@ namespace Aws
169169
StandardRetryStrategy(retryQuotaContainer, maxAttempts)
170170
{}
171171

172+
AdaptiveRetryStrategy::AdaptiveRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts, double transientBackoffBaseSec) :
173+
StandardRetryStrategy(retryQuotaContainer, maxAttempts, transientBackoffBaseSec)
174+
{}
175+
172176
bool AdaptiveRetryStrategy::HasSendToken()
173177
{
174178
return m_retryTokenBucket.Acquire(1, m_fastFail);

‎src/aws-cpp-sdk-core/source/client/ClientConfiguration.cpp‎

Lines changed: 21 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -544,6 +544,8 @@ ClientConfiguration::ClientConfiguration(bool /*useSmartDefaults*/, const char*
544544
}
545545

546546
std::shared_ptr<RetryStrategy> InitRetryStrategy(int maxAttempts, Aws::String retryMode) {
547+
const bool newRetriesEnabled = Aws::Environment::GetEnv("AWS_NEW_RETRIES_2026") == "true";
548+
547549
if (retryMode.empty())
548550
{
549551
retryMode = Aws::Environment::GetEnv("AWS_RETRY_MODE");
@@ -552,37 +554,48 @@ std::shared_ptr<RetryStrategy> InitRetryStrategy(int maxAttempts, Aws::String re
552554
{
553555
retryMode = Aws::Config::GetCachedConfigValue("retry_mode");
554556
}
557+
if (newRetriesEnabled && retryMode.empty())
558+
{
559+
retryMode = "standard";
560+
}
555561

556562
std::shared_ptr<RetryStrategy> retryStrategy;
557563
if (retryMode == "standard")
558564
{
559-
if (maxAttempts < 0)
565+
long attempts = (maxAttempts < 0) ? 3 : maxAttempts;
566+
if (newRetriesEnabled)
560567
{
561-
// negative value set above force usage of default max attempts
562-
retryStrategy = Aws::MakeShared<StandardRetryStrategy>(CLIENT_CONFIG_TAG);
568+
auto quota = Aws::MakeShared<DefaultRetryQuotaContainer>(CLIENT_CONFIG_TAG, 14, 5, RetryCostClassification::THROTTLE_BASED);
569+
retryStrategy = Aws::MakeShared<StandardRetryStrategy>(CLIENT_CONFIG_TAG, quota, attempts, 0.05);
563570
}
564571
else
565572
{
566-
retryStrategy = Aws::MakeShared<StandardRetryStrategy>(CLIENT_CONFIG_TAG, maxAttempts);
573+
retryStrategy = Aws::MakeShared<StandardRetryStrategy>(CLIENT_CONFIG_TAG, attempts);
567574
}
568575
}
569576
else if (retryMode == "adaptive")
570577
{
571-
if (maxAttempts < 0)
578+
long attempts = (maxAttempts < 0) ? 3 : maxAttempts;
579+
if (newRetriesEnabled)
572580
{
573-
// negative value set above force usage of default max attempts
574-
retryStrategy = Aws::MakeShared<AdaptiveRetryStrategy>(CLIENT_CONFIG_TAG);
581+
auto quota = Aws::MakeShared<DefaultRetryQuotaContainer>(CLIENT_CONFIG_TAG, 14, 5, RetryCostClassification::THROTTLE_BASED);
582+
retryStrategy = Aws::MakeShared<AdaptiveRetryStrategy>(CLIENT_CONFIG_TAG, quota, attempts, 0.05);
575583
}
576584
else
577585
{
578-
retryStrategy = Aws::MakeShared<AdaptiveRetryStrategy>(CLIENT_CONFIG_TAG, maxAttempts);
586+
retryStrategy = Aws::MakeShared<AdaptiveRetryStrategy>(CLIENT_CONFIG_TAG, attempts);
579587
}
580588
}
581589
else
582590
{
583591
retryStrategy = Aws::MakeShared<DefaultRetryStrategy>(CLIENT_CONFIG_TAG);
584592
}
585593

594+
if (newRetriesEnabled)
595+
{
596+
AWS_LOGSTREAM_INFO(CLIENT_CONFIG_TAG, "Retry Behavior 2.1 active (AWS_NEW_RETRIES_2026=true), mode=" << retryMode);
597+
}
598+
586599
return retryStrategy;
587600
}
588601

‎src/aws-cpp-sdk-core/source/client/RetryStrategy.cpp‎

Lines changed: 107 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
#include <aws/core/client/CoreErrors.h>
88
#include <aws/core/client/RetryStrategy.h>
99
#include <aws/core/utils/Outcome.h>
10+
#include <aws/core/utils/StringUtils.h>
1011
#include <aws/core/utils/local/Random.h>
1112

1213
using namespace Aws::Utils::Threading;
@@ -19,11 +20,98 @@ namespace Aws
1920
static const int RETRY_COST = 5;
2021
static const int TIMEOUT_RETRY_COST = 10;
2122

23+
// ---- DefaultRetryQuotaContainer ----
24+
25+
DefaultRetryQuotaContainer::DefaultRetryQuotaContainer()
26+
: m_retryQuota(INITIAL_RETRY_TOKENS),
27+
m_retryCost(RETRY_COST),
28+
m_throttlingRetryCost(TIMEOUT_RETRY_COST),
29+
m_classification(RetryCostClassification::REQUEST_TIMEOUT_BASED)
30+
{}
31+
32+
DefaultRetryQuotaContainer::DefaultRetryQuotaContainer(int retryCost, int throttlingRetryCost, RetryCostClassification classification)
33+
: m_retryQuota(INITIAL_RETRY_TOKENS),
34+
m_retryCost(retryCost),
35+
m_throttlingRetryCost(throttlingRetryCost),
36+
m_classification(classification)
37+
{}
38+
39+
bool DefaultRetryQuotaContainer::AcquireRetryQuota(int capacityAmount)
40+
{
41+
WriterLockGuard guard(m_retryQuotaLock);
42+
43+
if (capacityAmount > m_retryQuota)
44+
{
45+
return false;
46+
}
47+
m_retryQuota -= capacityAmount;
48+
return true;
49+
}
50+
51+
bool DefaultRetryQuotaContainer::AcquireRetryQuota(const AWSError<CoreErrors>& error)
52+
{
53+
int capacityAmount;
54+
if (m_classification == RetryCostClassification::THROTTLE_BASED)
55+
{
56+
capacityAmount = error.ShouldThrottle() ? m_throttlingRetryCost : m_retryCost;
57+
}
58+
else
59+
{
60+
capacityAmount = error.GetErrorType() == CoreErrors::REQUEST_TIMEOUT ? TIMEOUT_RETRY_COST : RETRY_COST;
61+
}
62+
return AcquireRetryQuota(capacityAmount);
63+
}
64+
65+
void DefaultRetryQuotaContainer::ReleaseRetryQuota(int capacityAmount)
66+
{
67+
WriterLockGuard guard(m_retryQuotaLock);
68+
m_retryQuota = (std::min)(m_retryQuota + capacityAmount, INITIAL_RETRY_TOKENS);
69+
}
70+
71+
void DefaultRetryQuotaContainer::ReleaseRetryQuota(const AWSError<CoreErrors>& error)
72+
{
73+
int capacityAmount;
74+
if (m_classification == RetryCostClassification::THROTTLE_BASED)
75+
{
76+
capacityAmount = error.ShouldThrottle() ? m_throttlingRetryCost : m_retryCost;
77+
}
78+
else
79+
{
80+
capacityAmount = error.GetErrorType() == CoreErrors::REQUEST_TIMEOUT ? TIMEOUT_RETRY_COST : RETRY_COST;
81+
}
82+
ReleaseRetryQuota(capacityAmount);
83+
}
84+
85+
// ---- StandardRetryStrategy pimpl ----
86+
87+
struct StandardRetryStrategy::RetryImpl
88+
{
89+
bool newRetriesEnabled = false;
90+
double transientBackoffBaseSec = 0.05;
91+
};
92+
2293
StandardRetryStrategy::StandardRetryStrategy(long maxAttempts)
23-
: m_retryQuotaContainer(Aws::MakeShared<DefaultRetryQuotaContainer>("StandardRetryStrategy")), m_maxAttempts(maxAttempts) {}
94+
: m_retryQuotaContainer(Aws::MakeShared<DefaultRetryQuotaContainer>("StandardRetryStrategy")),
95+
m_maxAttempts(maxAttempts),
96+
m_impl(Aws::MakeUnique<RetryImpl>("StandardRetryStrategy"))
97+
{}
2498

2599
StandardRetryStrategy::StandardRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts)
26-
: m_retryQuotaContainer(retryQuotaContainer), m_maxAttempts(maxAttempts) {}
100+
: m_retryQuotaContainer(retryQuotaContainer),
101+
m_maxAttempts(maxAttempts),
102+
m_impl(Aws::MakeUnique<RetryImpl>("StandardRetryStrategy"))
103+
{}
104+
105+
StandardRetryStrategy::StandardRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts, double transientBackoffBaseSec)
106+
: m_retryQuotaContainer(retryQuotaContainer),
107+
m_maxAttempts(maxAttempts),
108+
m_impl(Aws::MakeUnique<RetryImpl>("StandardRetryStrategy"))
109+
{
110+
m_impl->newRetriesEnabled = true;
111+
m_impl->transientBackoffBaseSec = transientBackoffBaseSec;
112+
}
113+
114+
StandardRetryStrategy::~StandardRetryStrategy() = default;
27115

28116
void StandardRetryStrategy::RequestBookkeeping(const HttpResponseOutcome& httpResponseOutcome)
29117
{
@@ -54,45 +142,30 @@ namespace Aws
54142

55143
long StandardRetryStrategy::CalculateDelayBeforeNextRetry(const AWSError<CoreErrors>& error, long attemptedRetries) const
56144
{
57-
AWS_UNREFERENCED_PARAM(error);
58-
// Maximum left shift factor is capped by ceil(log2(max_delay)), to avoid wrap-around and overflow into negative values:
59-
return std::min(static_cast<int>(Aws::Utils::GetRandomValue() % 1000) * (1 << std::min(attemptedRetries, 15L)), 20000);
60-
}
145+
if (!m_impl->newRetriesEnabled)
146+
{
147+
AWS_UNREFERENCED_PARAM(error);
148+
return (std::min)(static_cast<int>(Aws::Utils::GetRandomValue() % 1000) * (1 << (std::min)(attemptedRetries, 15L)), 20000);
149+
}
61150

62-
DefaultRetryQuotaContainer::DefaultRetryQuotaContainer() : m_retryQuota(INITIAL_RETRY_TOKENS)
63-
{}
151+
double x = error.ShouldThrottle() ? 1.0 : m_impl->transientBackoffBaseSec;
152+
double exponentialPart = x * static_cast<double>(1L << (std::min)(attemptedRetries, 30L));
153+
double cappedPart = (std::min)(exponentialPart, 20.0);
64154

65-
bool DefaultRetryQuotaContainer::AcquireRetryQuota(int capacityAmount)
66-
{
67-
WriterLockGuard guard(m_retryQuotaLock);
155+
double b = static_cast<double>(Aws::Utils::GetRandomValue() % 10000) / 10000.0;
156+
double t_i = b * cappedPart;
68157

69-
if (capacityAmount > m_retryQuota)
70-
{
71-
return false;
72-
}
73-
else
158+
const auto& headers = error.GetResponseHeaders();
159+
auto it = headers.find("x-amz-retry-after");
160+
if (it != headers.end())
74161
{
75-
m_retryQuota -= capacityAmount;
76-
return true;
162+
double headerSec = static_cast<double>(Aws::Utils::StringUtils::ConvertToInt64(it->second.c_str())) / 1000.0;
163+
double clamped = (std::max)(t_i, (std::min)(headerSec, 5.0 + t_i));
164+
return static_cast<long>(clamped * 1000.0);
77165
}
78-
}
79166

80-
bool DefaultRetryQuotaContainer::AcquireRetryQuota(const AWSError<CoreErrors>& error)
81-
{
82-
int capacityAmount = error.GetErrorType() == CoreErrors::REQUEST_TIMEOUT ? TIMEOUT_RETRY_COST : RETRY_COST;
83-
return AcquireRetryQuota(capacityAmount);
167+
return static_cast<long>(t_i * 1000.0);
84168
}
85169

86-
void DefaultRetryQuotaContainer::ReleaseRetryQuota(int capacityAmount)
87-
{
88-
WriterLockGuard guard(m_retryQuotaLock);
89-
m_retryQuota = (std::min)(m_retryQuota + capacityAmount, INITIAL_RETRY_TOKENS);
90-
}
91-
92-
void DefaultRetryQuotaContainer::ReleaseRetryQuota(const AWSError<CoreErrors>& error)
93-
{
94-
int capacityAmount = error.GetErrorType() == CoreErrors::REQUEST_TIMEOUT ? TIMEOUT_RETRY_COST : RETRY_COST;
95-
ReleaseRetryQuota(capacityAmount);
96-
}
97170
}
98171
}

0 commit comments

Comments
 (0)