wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add /GlobalTagThrottler/WriteThrottling unit test

This commit is contained in:
sfc-gh-tclinkenbeard 2022-06-15 19:28:25 -07:00
parent a423fc77db
commit d692516f64
1 changed files with 40 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
fdbserver/GlobalTagThrottler.actor.cpp
View File

@ -258,22 +258,23 @@ Optional<double> testGetTPSLimit(GlobalTagThrottler& globalTagThrottler, Transac
} }
class TestStorageServers { class TestStorageServers {
class ReadCost { class Cost {
Smoother smoother; Smoother smoother;
public: public:
ReadCost() : smoother(5.0) {} Cost() : smoother(5.0) {}
ReadCost& operator+=(double delta) { Cost& operator+=(double delta) {
smoother.addDelta(delta); smoother.addDelta(delta);
return *this; return *this;
} }
double smoothRate() const { return smoother.smoothRate(); } double smoothRate() const { return smoother.smoothRate(); }
}; };
std::vector<std::map<TransactionTag, ReadCost>> readCosts; std::vector<std::map<TransactionTag, Cost>> readCosts;
std::vector<std::map<TransactionTag, Cost>> writeCosts;
public: public:
TestStorageServers(size_t size) : readCosts(size) { ASSERT_GT(size, 0); } TestStorageServers(size_t size) : readCosts(size), writeCosts(size) { ASSERT_GT(size, 0); }
void addReadCost(TransactionTag tag, double cost) { void addReadCost(TransactionTag tag, double cost) {
auto const costPerSS = cost / readCosts.size(); auto const costPerSS = cost / readCosts.size();
@ -282,6 +283,13 @@ public:
} }
} }
void addWriteCost(TransactionTag tag, double cost) {
auto const costPerSS = cost / writeCosts.size();
for (auto& writeCost : writeCosts) {
writeCost[tag] += costPerSS;
}
}
std::vector<StorageQueueInfo> getStorageQueueInfos() const { std::vector<StorageQueueInfo> getStorageQueueInfos() const {
std::vector<StorageQueueInfo> result; std::vector<StorageQueueInfo> result;
result.reserve(readCosts.size()); result.reserve(readCosts.size());
@ -291,6 +299,10 @@ public:
double fractionalBusyness{ 0.0 }; // unused for global tag throttling double fractionalBusyness{ 0.0 }; // unused for global tag throttling
sqInfo.busiestReadTags.emplace_back(tag, readCost.smoothRate(), fractionalBusyness); sqInfo.busiestReadTags.emplace_back(tag, readCost.smoothRate(), fractionalBusyness);
} }
for (const auto& [tag, writeCost] : writeCosts[i]) {
double fractionalBusyness{ 0.0 }; // unused for global tag throttling
sqInfo.busiestWriteTags.emplace_back(tag, writeCost.smoothRate(), fractionalBusyness);
}
result.push_back(sqInfo); result.push_back(sqInfo);
} }
return result; return result;
@ -301,12 +313,17 @@ ACTOR static Future<Void> testClient(GlobalTagThrottler* globalTagThrottler,
TestStorageServers* testStorageServers, TestStorageServers* testStorageServers,
TransactionTag tag, TransactionTag tag,
double desiredTpsRate, double desiredTpsRate,
double costPerTransaction) { double costPerTransaction,
bool write) {
loop { loop {
auto tpsLimit = testGetTPSLimit(*globalTagThrottler, tag); auto tpsLimit = testGetTPSLimit(*globalTagThrottler, tag);
state double tpsRate = tpsLimit.present() ? std::min<double>(desiredTpsRate, tpsLimit.get()) : desiredTpsRate; state double tpsRate = tpsLimit.present() ? std::min<double>(desiredTpsRate, tpsLimit.get()) : desiredTpsRate;
wait(delay(1 / tpsRate)); wait(delay(1 / tpsRate));
if (write) {
testStorageServers->addWriteCost(tag, costPerTransaction);
} else {
testStorageServers->addReadCost(tag, costPerTransaction); testStorageServers->addReadCost(tag, costPerTransaction);
}
globalTagThrottler->addRequests(tag, 1); globalTagThrottler->addRequests(tag, 1);
} }
} }
@ -354,7 +371,21 @@ TEST_CASE("/GlobalTagThrottler/NoActiveThrottling") {
TransactionTag testTag = "sampleTag1"_sr; TransactionTag testTag = "sampleTag1"_sr;
tagQuotaValue.totalReadQuota = 100.0; tagQuotaValue.totalReadQuota = 100.0;
globalTagThrottler.setQuota(testTag, tagQuotaValue); globalTagThrottler.setQuota(testTag, tagQuotaValue);
state Future<Void> client = testClient(&globalTagThrottler, &testStorageServers, testTag, 5.0, 6.0); state Future<Void> client = testClient(&globalTagThrottler, &testStorageServers, testTag, 5.0, 6.0, false);
state Future<Void> monitor = monitorClientRates(&globalTagThrottler, testTag, 100.0 / 6.0);
state Future<Void> updater = updateGlobalTagThrottler(&globalTagThrottler, &testStorageServers);
wait(timeoutError(monitor || client || updater, 300.0));
return Void();
}
TEST_CASE("/GlobalTagThrottler/WriteThrottling") {
state GlobalTagThrottler globalTagThrottler(Database{}, UID{});
state TestStorageServers testStorageServers(10);
ThrottleApi::TagQuotaValue tagQuotaValue;
TransactionTag testTag = "sampleTag1"_sr;
tagQuotaValue.totalWriteQuota = 100.0;
globalTagThrottler.setQuota(testTag, tagQuotaValue);
state Future<Void> client = testClient(&globalTagThrottler, &testStorageServers, testTag, 5.0, 6.0, true);
state Future<Void> monitor = monitorClientRates(&globalTagThrottler, testTag, 100.0 / 6.0); state Future<Void> monitor = monitorClientRates(&globalTagThrottler, testTag, 100.0 / 6.0);
state Future<Void> updater = updateGlobalTagThrottler(&globalTagThrottler, &testStorageServers); state Future<Void> updater = updateGlobalTagThrottler(&globalTagThrottler, &testStorageServers);
wait(timeoutError(monitor || client || updater, 300.0)); wait(timeoutError(monitor || client || updater, 300.0));
@ -368,7 +399,7 @@ TEST_CASE("/GlobalTagThrottler/ActiveThrottling") {
TransactionTag testTag = "sampleTag1"_sr; TransactionTag testTag = "sampleTag1"_sr;
tagQuotaValue.totalReadQuota = 100.0; tagQuotaValue.totalReadQuota = 100.0;
globalTagThrottler.setQuota(testTag, tagQuotaValue); globalTagThrottler.setQuota(testTag, tagQuotaValue);
state Future<Void> client = testClient(&globalTagThrottler, &testStorageServers, testTag, 20.0, 10.0); state Future<Void> client = testClient(&globalTagThrottler, &testStorageServers, testTag, 20.0, 10.0, false);
state Future<Void> monitor = monitorClientRates(&globalTagThrottler, testTag, 10.0); state Future<Void> monitor = monitorClientRates(&globalTagThrottler, testTag, 10.0);
state Future<Void> updater = updateGlobalTagThrottler(&globalTagThrottler, &testStorageServers); state Future<Void> updater = updateGlobalTagThrottler(&globalTagThrottler, &testStorageServers);
wait(timeoutError(monitor || client || updater, 300.0)); wait(timeoutError(monitor || client || updater, 300.0));