Skip to content

Commit a65477e

Browse files
kai-ionkai-ion
authored
[Feature] New Retry Behavior (#3836)
* Implement Retry Behavior 2.1 core logic gated behind AWS_NEW_RETRIES_2026 * removed transient back off * moved implementation to pimple * added a new function to support dynamodb transient backoff * added a new function to support dynamodb transient backoff * changed from feature bool to interface * update test to use environment var * white space change * adding logging and dry * updated testing and namespace * updated testing and namespace * updated struct to class * fixed static int to structs * adding readerlock wrapper
1 parent 4e370d8 commit a65477e

5 files changed

Lines changed: 346 additions & 5 deletions

File tree

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

Lines changed: 7 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

@@ -123,6 +124,7 @@ namespace Aws
123124
public:
124125
StandardRetryStrategy(long maxAttempts = 3);
125126
StandardRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts = 3);
127+
virtual ~StandardRetryStrategy();
126128

127129
virtual void RequestBookkeeping(const HttpResponseOutcome& httpResponseOutcome) override;
128130
virtual void RequestBookkeeping(const HttpResponseOutcome& httpResponseOutcome, const AWSError<CoreErrors>& lastError) override;
@@ -135,9 +137,14 @@ namespace Aws
135137

136138
const char* GetStrategyName() const override { return "standard";}
137139

140+
class RetryImpl;
141+
138142
protected:
139143
std::shared_ptr<RetryQuotaContainer> m_retryQuotaContainer;
140144
long m_maxAttempts;
145+
146+
private:
147+
Aws::UniquePtr<RetryImpl> m_impl;
141148
};
142149
} // namespace Client
143150
} // namespace Aws

‎src/aws-cpp-sdk-core/include/aws/core/internal/RetryStrategyImpl.h‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/**
2+
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
* SPDX-License-Identifier: Apache-2.0.
4+
*/
5+
6+
#pragma once
7+
8+
#include <aws/core/Core_EXPORTS.h>
9+
#include <aws/core/client/AWSError.h>
10+
#include <aws/core/client/CoreErrors.h>
11+
#include <aws/core/client/RetryStrategy.h>
12+
#include <aws/core/utils/threading/ReaderWriterLock.h>
13+
14+
namespace Aws
15+
{
16+
namespace Client
17+
{
18+
struct QuotaConfig
19+
{
20+
int retryCost = 14;
21+
int throttlingCost = 5;
22+
int initialTokens = 500;
23+
};
24+
25+
class AWS_CORE_LOCAL ThrottleBasedRetryQuotaContainer : public RetryQuotaContainer
26+
{
27+
public:
28+
ThrottleBasedRetryQuotaContainer(const QuotaConfig& config = QuotaConfig{})
29+
: m_config(config), m_retryQuota(config.initialTokens) {}
30+
31+
virtual ~ThrottleBasedRetryQuotaContainer() = default;
32+
33+
bool AcquireRetryQuota(int capacityAmount) override
34+
{
35+
Aws::Utils::Threading::WriterLockGuard guard(m_retryQuotaLock);
36+
if (capacityAmount > m_retryQuota)
37+
{
38+
return false;
39+
}
40+
else
41+
{
42+
m_retryQuota -= capacityAmount;
43+
return true;
44+
}
45+
}
46+
47+
bool AcquireRetryQuota(const AWSError<CoreErrors>& error) override
48+
{
49+
int capacityAmount = error.ShouldThrottle() ? m_config.throttlingCost : m_config.retryCost;
50+
return AcquireRetryQuota(capacityAmount);
51+
}
52+
53+
void ReleaseRetryQuota(int capacityAmount) override
54+
{
55+
Aws::Utils::Threading::WriterLockGuard guard(m_retryQuotaLock);
56+
m_retryQuota = (std::min)(m_retryQuota + capacityAmount, m_config.initialTokens);
57+
}
58+
59+
void ReleaseRetryQuota(const AWSError<CoreErrors>& error) override
60+
{
61+
int capacityAmount = error.ShouldThrottle() ? m_config.throttlingCost : m_config.retryCost;
62+
ReleaseRetryQuota(capacityAmount);
63+
}
64+
65+
int GetRetryQuota() const override
66+
{
67+
Aws::Utils::Threading::ReaderLockGuard guard(m_retryQuotaLock);
68+
return m_retryQuota;
69+
}
70+
71+
private:
72+
QuotaConfig m_config;
73+
mutable Aws::Utils::Threading::ReaderWriterLock m_retryQuotaLock;
74+
int m_retryQuota;
75+
};
76+
} // namespace Client
77+
} // namespace Aws

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -552,6 +552,10 @@ std::shared_ptr<RetryStrategy> InitRetryStrategy(int maxAttempts, Aws::String re
552552
{
553553
retryMode = Aws::Config::GetCachedConfigValue("retry_mode");
554554
}
555+
if (Aws::Utils::StringUtils::ToLower(Aws::Environment::GetEnv("AWS_NEW_RETRIES_2026").c_str()) == "true" && retryMode.empty())
556+
{
557+
retryMode = "standard";
558+
}
555559

556560
std::shared_ptr<RetryStrategy> retryStrategy;
557561
if (retryMode == "standard")

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

Lines changed: 93 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,96 @@
66
#include <aws/core/client/AWSError.h>
77
#include <aws/core/client/CoreErrors.h>
88
#include <aws/core/client/RetryStrategy.h>
9+
#include <aws/core/internal/RetryStrategyImpl.h>
10+
#include <aws/core/platform/Environment.h>
911
#include <aws/core/utils/Outcome.h>
12+
#include <aws/core/utils/StringUtils.h>
1013
#include <aws/core/utils/local/Random.h>
14+
#include <aws/core/utils/logging/LogMacros.h>
1115

1216
using namespace Aws::Utils::Threading;
17+
using namespace Aws::Client;
18+
19+
namespace Aws
20+
{
21+
namespace Client
22+
{
23+
class StandardRetryStrategy::RetryImpl
24+
{
25+
public:
26+
virtual ~RetryImpl() = default;
27+
virtual long CalculateDelay(const AWSError<CoreErrors>& error, long attemptedRetries) const = 0;
28+
};
29+
}
30+
}
31+
32+
namespace {
33+
const char RETRY_STRATEGY_TAG[] = "StandardRetryStrategy";
34+
35+
bool IsNewRetriesEnabled()
36+
{
37+
return Aws::Utils::StringUtils::ToLower(Aws::Environment::GetEnv("AWS_NEW_RETRIES_2026").c_str()) == "true";
38+
}
39+
40+
class LegacyRetryImpl : public StandardRetryStrategy::RetryImpl
41+
{
42+
public:
43+
long CalculateDelay(const AWSError<CoreErrors>& error, long attemptedRetries) const override
44+
{
45+
AWS_UNREFERENCED_PARAM(error);
46+
// Maximum left shift factor is capped by ceil(log2(max_delay)), to avoid wrap-around and overflow into negative values:
47+
return (std::min)(static_cast<int>(Aws::Utils::GetRandomValue() % 1000) * (1 << (std::min)(attemptedRetries, 15L)), 20000);
48+
}
49+
};
50+
51+
class NewRetriesImpl : public StandardRetryStrategy::RetryImpl
52+
{
53+
public:
54+
long CalculateDelay(const AWSError<CoreErrors>& error, long attemptedRetries) const override
55+
{
56+
double x = error.ShouldThrottle() ? 1.0 : 0.05;
57+
double exponentialPart = x * static_cast<double>(1L << (std::min)(attemptedRetries, 30L));
58+
double cappedPart = (std::min)(exponentialPart, 20.0);
59+
60+
double b = static_cast<double>(Aws::Utils::GetRandomValue() % 10000) / 10000.0;
61+
double t_i = b * cappedPart;
62+
63+
const auto& headers = error.GetResponseHeaders();
64+
auto it = headers.find("x-amz-retry-after");
65+
if (it != headers.end())
66+
{
67+
long long headerMs = Aws::Utils::StringUtils::ConvertToInt64(it->second.c_str());
68+
if (headerMs >= 0)
69+
{
70+
double headerSec = static_cast<double>(headerMs) / 1000.0;
71+
double clamped = (std::max)(t_i, (std::min)(headerSec, 5.0 + t_i));
72+
return static_cast<long>(clamped * 1000.0);
73+
}
74+
AWS_LOGSTREAM_DEBUG(RETRY_STRATEGY_TAG, "Ignoring invalid x-amz-retry-after value: " << it->second);
75+
}
76+
77+
return static_cast<long>(t_i * 1000.0);
78+
}
79+
};
80+
81+
Aws::UniquePtr<StandardRetryStrategy::RetryImpl> CreateRetryImpl()
82+
{
83+
if (IsNewRetriesEnabled())
84+
{
85+
return Aws::MakeUnique<NewRetriesImpl>("StandardRetryStrategy");
86+
}
87+
return Aws::MakeUnique<LegacyRetryImpl>("StandardRetryStrategy");
88+
}
89+
90+
std::shared_ptr<RetryQuotaContainer> CreateQuotaContainer()
91+
{
92+
if (IsNewRetriesEnabled())
93+
{
94+
return Aws::MakeShared<ThrottleBasedRetryQuotaContainer>("StandardRetryStrategy");
95+
}
96+
return Aws::MakeShared<DefaultRetryQuotaContainer>("StandardRetryStrategy");
97+
}
98+
} // anonymous namespace
1399

14100
namespace Aws
15101
{
@@ -20,10 +106,14 @@ namespace Aws
20106
static const int TIMEOUT_RETRY_COST = 10;
21107

22108
StandardRetryStrategy::StandardRetryStrategy(long maxAttempts)
23-
: m_retryQuotaContainer(Aws::MakeShared<DefaultRetryQuotaContainer>("StandardRetryStrategy")), m_maxAttempts(maxAttempts) {}
109+
: m_retryQuotaContainer(CreateQuotaContainer()), m_maxAttempts(maxAttempts),
110+
m_impl(CreateRetryImpl()) {}
24111

25112
StandardRetryStrategy::StandardRetryStrategy(std::shared_ptr<RetryQuotaContainer> retryQuotaContainer, long maxAttempts)
26-
: m_retryQuotaContainer(retryQuotaContainer), m_maxAttempts(maxAttempts) {}
113+
: m_retryQuotaContainer(retryQuotaContainer), m_maxAttempts(maxAttempts),
114+
m_impl(CreateRetryImpl()) {}
115+
116+
StandardRetryStrategy::~StandardRetryStrategy() = default;
27117

28118
void StandardRetryStrategy::RequestBookkeeping(const HttpResponseOutcome& httpResponseOutcome)
29119
{
@@ -54,9 +144,7 @@ namespace Aws
54144

55145
long StandardRetryStrategy::CalculateDelayBeforeNextRetry(const AWSError<CoreErrors>& error, long attemptedRetries) const
56146
{
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);
147+
return m_impl->CalculateDelay(error, attemptedRetries);
60148
}
61149

62150
DefaultRetryQuotaContainer::DefaultRetryQuotaContainer() : m_retryQuota(INITIAL_RETRY_TOKENS)

0 commit comments

Comments
 (0)