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

Improve performance of TransactionTagCounter

This commit is contained in:
sfc-gh-tclinkenbeard 2023-05-25 23:26:57 -07:00
parent e724c90ffe
commit f741a584c0
6 changed files with 210 additions and 248 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
fdbclient/include/fdbclient/StorageServerInterface.h
View File

@ -1168,23 +1168,27 @@ struct GetStorageMetricsRequest {
}
};
// Tracks the busyness of tags on individual storage servers.
struct BusyTagInfo {
constexpr static FileIdentifier file_identifier = 4528694;
TransactionTag tag;
double rate{ 0.0 };
double fractionalBusyness{ 0.0 };
BusyTagInfo() = default;
BusyTagInfo(TransactionTag const& tag, double rate, double fractionalBusyness)
: tag(tag), rate(rate), fractionalBusyness(fractionalBusyness) {}
bool operator<(BusyTagInfo const& rhs) const { return rate < rhs.rate; }
bool operator>(BusyTagInfo const& rhs) const { return rate > rhs.rate; }
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, tag, rate, fractionalBusyness);
}
};
struct StorageQueuingMetricsReply {
struct TagInfo {
constexpr static FileIdentifier file_identifier = 4528694;
TransactionTag tag;
double rate{ 0.0 };
double fractionalBusyness{ 0.0 };
TagInfo() = default;
TagInfo(TransactionTag const& tag, double rate, double fractionalBusyness)
: tag(tag), rate(rate), fractionalBusyness(fractionalBusyness) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, tag, rate, fractionalBusyness);
}
};
constexpr static FileIdentifier file_identifier = 7633366;
double localTime;
int64_t instanceID; // changes if bytesDurable and bytesInput reset
@ -1195,7 +1199,7 @@ struct StorageQueuingMetricsReply {
double cpuUsage{ 0.0 };
double diskUsage{ 0.0 };
double localRateLimit;
std::vector<TagInfo> busiestTags;
std::vector<BusyTagInfo> busiestTags;
template <class Ar>
void serialize(Ar& ar) {

181
fdbserver/TransactionTagCounter.actor.cpp Normal file
View File

@ -0,0 +1,181 @@
/*
* TransactionTagCounter.actor.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2022 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "fdbclient/NativeAPI.actor.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/TransactionTagCounter.h"
#include "flow/Trace.h"
#include "flow/actorcompiler.h"
class TransactionTagCounterImpl {
UID thisServerID;
TransactionTagMap<double> intervalCosts;
double intervalTotalCost = 0;
double intervalStart = 0;
int maxTagsTracked;
double minRateTracked;
std::vector<BusyTagInfo> previousBusiestTags;
Reference<EventCacheHolder> busiestReadTagEventHolder;
std::vector<BusyTagInfo> getBusiestTagsFromLastInterval(double elapsed) const {
std::priority_queue<BusyTagInfo, std::vector<BusyTagInfo>, std::greater<BusyTagInfo>> topKTags;
for (auto const& [tag, cost] : intervalCosts) {
auto const rate = cost / elapsed;
auto const fractionalBusyness = std::min(1.0, cost / intervalTotalCost);
if (rate < minRateTracked) {
continue;
} else if (topKTags.size() < maxTagsTracked) {
topKTags.emplace(tag, rate, fractionalBusyness);
} else if (topKTags.top().rate < rate) {
topKTags.pop();
topKTags.emplace(tag, rate, fractionalBusyness);
}
}
std::vector<BusyTagInfo> result;
while (!topKTags.empty()) {
result.push_back(std::move(topKTags.top()));
topKTags.pop();
}
return result;
}
public:
TransactionTagCounterImpl(UID thisServerID, int maxTagsTracked, double minRateTracked)
: thisServerID(thisServerID), maxTagsTracked(maxTagsTracked), minRateTracked(minRateTracked),
busiestReadTagEventHolder(makeReference<EventCacheHolder>(thisServerID.toString() + "/BusiestReadTag")) {}
void addRequest(Optional<TagSet> const& tags, int64_t bytes) {
auto const cost = getReadOperationCost(bytes);
intervalTotalCost += cost;
if (tags.present()) {
for (auto const& tag : tags.get()) {
CODE_PROBE(true, "Tracking transaction tag in TransactionTagCounter");
intervalCosts[TransactionTag(tag, tags.get().getArena())] += cost / CLIENT_KNOBS->READ_TAG_SAMPLE_RATE;
}
}
}
void startNewInterval() {
double elapsed = now() - intervalStart;
previousBusiestTags.clear();
if (intervalStart > 0 && CLIENT_KNOBS->READ_TAG_SAMPLE_RATE > 0 && elapsed > 0) {
previousBusiestTags = getBusiestTagsFromLastInterval(elapsed);
// For status, report the busiest tag:
if (previousBusiestTags.empty()) {
TraceEvent("BusiestReadTag", thisServerID).detail("TagCost", 0);
} else {
auto busiestTagInfo = previousBusiestTags[0];
for (int i = 1; i < previousBusiestTags.size(); ++i) {
auto const& tagInfo = previousBusiestTags[i];
if (tagInfo.rate > busiestTagInfo.rate) {
busiestTagInfo = tagInfo;
}
}
TraceEvent("BusiestReadTag", thisServerID)
.detail("Tag", printable(busiestTagInfo.tag))
.detail("TagCost", busiestTagInfo.rate)
.detail("FractionalBusyness", busiestTagInfo.fractionalBusyness);
}
for (const auto& tagInfo : previousBusiestTags) {
TraceEvent("BusyReadTag", thisServerID)
.detail("Tag", printable(tagInfo.tag))
.detail("TagCost", tagInfo.rate)
.detail("FractionalBusyness", tagInfo.fractionalBusyness);
}
}
intervalCosts.clear();
intervalTotalCost = 0;
intervalStart = now();
}
std::vector<BusyTagInfo> const& getBusiestTags() const { return previousBusiestTags; }
};
TransactionTagCounter::TransactionTagCounter(UID thisServerID, int maxTagsTracked, double minRateTracked)
: impl(PImpl<TransactionTagCounterImpl>::create(thisServerID, maxTagsTracked, minRateTracked)) {}
TransactionTagCounter::~TransactionTagCounter() = default;
void TransactionTagCounter::addRequest(Optional<TagSet> const& tags, int64_t bytes) {
return impl->addRequest(tags, bytes);
}
void TransactionTagCounter::startNewInterval() {
return impl->startNewInterval();
}
std::vector<BusyTagInfo> const& TransactionTagCounter::getBusiestTags() const {
return impl->getBusiestTags();
}
namespace {
bool containsTag(std::vector<BusyTagInfo> const& busyTags, TransactionTagRef tag) {
return std::count_if(busyTags.begin(), busyTags.end(), [tag](auto const& tagInfo) { return tagInfo.tag == tag; }) ==
1;
}
TagSet getTagSet(TransactionTagRef tag) {
TagSet result;
result.addTag(tag);
return result;
}
} // namespace
TEST_CASE("/fdbserver/TransactionTagCounter/IgnoreBeyondMaxTags") {
state TransactionTagCounter counter(
UID(), /*maxTagsTracked=*/2, /*minRateTracked=*/10.0 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.startNewInterval();
ASSERT_EQ(counter.getBusiestTags().size(), 0);
{
wait(delay(1.0));
counter.addRequest(getTagSet("tagA"_sr), 10 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.addRequest(getTagSet("tagA"_sr), 10 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.addRequest(getTagSet("tagB"_sr), 15 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.addRequest(getTagSet("tagC"_sr), 20 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.startNewInterval();
auto const busiestTags = counter.getBusiestTags();
ASSERT_EQ(busiestTags.size(), 2);
ASSERT(containsTag(busiestTags, "tagA"_sr));
ASSERT(!containsTag(busiestTags, "tagB"_sr));
ASSERT(containsTag(busiestTags, "tagC"_sr));
}
return Void();
}
TEST_CASE("/fdbserver/TransactionTagCounter/IgnoreBelowMinRate") {
state TransactionTagCounter counter(
UID(), /*maxTagsTracked=*/2, /*minRateTracked=*/10.0 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.startNewInterval();
ASSERT_EQ(counter.getBusiestTags().size(), 0);
{
wait(delay(1.0));
counter.addRequest(getTagSet("tagA"_sr), 5 * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE);
counter.startNewInterval();
auto const busiestTags = counter.getBusiestTags();
ASSERT_EQ(busiestTags.size(), 0);
}
return Void();
}

227
fdbserver/TransactionTagCounter.cpp
View File

@ -1,227 +0,0 @@
/*
* TransactionTagCounter.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2022 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "fdbclient/NativeAPI.actor.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/TransactionTagCounter.h"
#include "flow/Trace.h"
namespace {
class TopKTags {
public:
struct TagAndCount {
TransactionTag tag;
int64_t count;
bool operator<(TagAndCount const& other) const { return count < other.count; }
explicit TagAndCount(TransactionTag tag, int64_t count) : tag(tag), count(count) {}
};
private:
// Because the number of tracked is expected to be small, they can be tracked
// in a simple vector. If the number of tracked tags increases, a more sophisticated
// data structure will be required.
std::vector<TagAndCount> topTags;
int limit;
public:
explicit TopKTags(int limit) : limit(limit) {
ASSERT_GT(limit, 0);
topTags.reserve(limit);
}
void incrementCount(TransactionTag tag, int previousCount, int increase) {
auto iter = std::find_if(topTags.begin(), topTags.end(), [tag](const auto& tc) { return tc.tag == tag; });
if (iter != topTags.end()) {
ASSERT_EQ(previousCount, iter->count);
iter->count += increase;
} else if (topTags.size() < limit) {
ASSERT_EQ(previousCount, 0);
topTags.emplace_back(tag, increase);
} else {
auto toReplace = std::min_element(topTags.begin(), topTags.end());
ASSERT_GE(toReplace->count, previousCount);
if (toReplace->count < previousCount + increase) {
toReplace->tag = tag;
toReplace->count = previousCount + increase;
}
}
}
std::vector<StorageQueuingMetricsReply::TagInfo> getBusiestTags(double elapsed, double totalSampleCount) const {
std::vector<StorageQueuingMetricsReply::TagInfo> result;
for (auto const& tagAndCounter : topTags) {
auto rate = (tagAndCounter.count / CLIENT_KNOBS->READ_TAG_SAMPLE_RATE) / elapsed;
if (rate > SERVER_KNOBS->MIN_TAG_READ_PAGES_RATE * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE) {
result.emplace_back(tagAndCounter.tag, rate, tagAndCounter.count / totalSampleCount);
}
}
return result;
}
void clear() { topTags.clear(); }
};
} // namespace
class TransactionTagCounterImpl {
UID thisServerID;
TransactionTagMap<int64_t> intervalCounts;
int64_t intervalTotalSampledCount = 0;
TopKTags topTags;
double intervalStart = 0;
std::vector<StorageQueuingMetricsReply::TagInfo> previousBusiestTags;
Reference<EventCacheHolder> busiestReadTagEventHolder;
public:
TransactionTagCounterImpl(UID thisServerID)
: thisServerID(thisServerID), topTags(SERVER_KNOBS->SS_THROTTLE_TAGS_TRACKED),
busiestReadTagEventHolder(makeReference<EventCacheHolder>(thisServerID.toString() + "/BusiestReadTag")) {}
void addRequest(Optional<TagSet> const& tags, int64_t bytes) {
if (tags.present()) {
CODE_PROBE(true, "Tracking transaction tag in counter");
auto const cost = getReadOperationCost(bytes);
for (auto& tag : tags.get()) {
int64_t& count = intervalCounts[TransactionTag(tag, tags.get().getArena())];
topTags.incrementCount(tag, count, cost);
count += cost;
}
intervalTotalSampledCount += cost;
}
}
void startNewInterval() {
double elapsed = now() - intervalStart;
previousBusiestTags.clear();
if (intervalStart > 0 && CLIENT_KNOBS->READ_TAG_SAMPLE_RATE > 0 && elapsed > 0) {
previousBusiestTags = topTags.getBusiestTags(elapsed, intervalTotalSampledCount);
// For status, report the busiest tag:
if (previousBusiestTags.empty()) {
TraceEvent("BusiestReadTag", thisServerID).detail("TagCost", 0.0);
} else {
auto busiestTagInfo = previousBusiestTags[0];
for (int i = 1; i < previousBusiestTags.size(); ++i) {
auto const& tagInfo = previousBusiestTags[i];
if (tagInfo.rate > busiestTagInfo.rate) {
busiestTagInfo = tagInfo;
}
}
TraceEvent("BusiestReadTag", thisServerID)
.detail("Tag", printable(busiestTagInfo.tag))
.detail("TagCost", busiestTagInfo.rate)
.detail("FractionalBusyness", busiestTagInfo.fractionalBusyness);
}
for (const auto& tagInfo : previousBusiestTags) {
TraceEvent("BusyReadTag", thisServerID)
.detail("Tag", printable(tagInfo.tag))
.detail("TagCost", tagInfo.rate)
.detail("FractionalBusyness", tagInfo.fractionalBusyness);
}
}
intervalCounts.clear();
intervalTotalSampledCount = 0;
topTags.clear();
intervalStart = now();
}
std::vector<StorageQueuingMetricsReply::TagInfo> const& getBusiestTags() const { return previousBusiestTags; }
};
TransactionTagCounter::TransactionTagCounter(UID thisServerID)
: impl(PImpl<TransactionTagCounterImpl>::create(thisServerID)) {}
TransactionTagCounter::~TransactionTagCounter() = default;
void TransactionTagCounter::addRequest(Optional<TagSet> const& tags, int64_t bytes) {
return impl->addRequest(tags, bytes);
}
void TransactionTagCounter::startNewInterval() {
return impl->startNewInterval();
}
std::vector<StorageQueuingMetricsReply::TagInfo> const& TransactionTagCounter::getBusiestTags() const {
return impl->getBusiestTags();
}
TEST_CASE("/TransactionTagCounter/TopKTags") {
TopKTags topTags(2);
// Ensure that costs are larger enough to show up
auto const costMultiplier =
std::max<double>(1.0,
2 * SERVER_KNOBS->MIN_TAG_READ_PAGES_RATE * CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE *
CLIENT_KNOBS->READ_TAG_SAMPLE_RATE);
ASSERT_EQ(topTags.getBusiestTags(1.0, 0).size(), 0);
topTags.incrementCount("a"_sr, 0, 1 * costMultiplier);
{
auto const busiestTags = topTags.getBusiestTags(1.0, 1 * costMultiplier);
ASSERT_EQ(busiestTags.size(), 1);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "a"_sr; }),
1);
}
topTags.incrementCount("b"_sr, 0, 2 * costMultiplier);
topTags.incrementCount("c"_sr, 0, 3 * costMultiplier);
{
auto busiestTags = topTags.getBusiestTags(1.0, 6 * costMultiplier);
ASSERT_EQ(busiestTags.size(), 2);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "a"_sr; }),
0);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "b"_sr; }),
1);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "c"_sr; }),
1);
}
topTags.incrementCount("a"_sr, 1 * costMultiplier, 3 * costMultiplier);
{
auto busiestTags = topTags.getBusiestTags(1.0, 9 * costMultiplier);
ASSERT_EQ(busiestTags.size(), 2);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "a"_sr; }),
1);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "b"_sr; }),
0);
ASSERT_EQ(std::count_if(busiestTags.begin(),
busiestTags.end(),
[](auto const& tagInfo) { return tagInfo.tag == "c"_sr; }),
1);
}
topTags.clear();
ASSERT_EQ(topTags.getBusiestTags(1.0, 0).size(), 0);
return Void();
}

2
fdbserver/include/fdbserver/Ratekeeper.h
View File

@ -72,7 +72,7 @@ public:
StorageQueuingMetricsReply lastReply;
bool acceptingRequests;
limitReason_t limitReason;
std::vector<StorageQueuingMetricsReply::TagInfo> busiestReadTags, busiestWriteTags;
std::vector<BusyTagInfo> busiestReadTags, busiestWriteTags;
StorageQueueInfo(const UID& id, const LocalityData& locality);
StorageQueueInfo(const UID& rateKeeperID, const UID& id, const LocalityData& locality);

4
fdbserver/include/fdbserver/TransactionTagCounter.h
View File

@ -28,7 +28,7 @@ class TransactionTagCounter {
PImpl<class TransactionTagCounterImpl> impl;
public:
TransactionTagCounter(UID thisServerID);
TransactionTagCounter(UID thisServerID, int maxTagsTracked, double minRateTracked);
~TransactionTagCounter();
// Update counters tracking the busyness of each tag in the current interval
@ -38,5 +38,5 @@ public:
void startNewInterval();
// Returns the set of busiest tags as of the end of the last interval
std::vector<StorageQueuingMetricsReply::TagInfo> const& getBusiestTags() const;
std::vector<BusyTagInfo> const& getBusiestTags() const;
};

6
fdbserver/storageserver.actor.cpp
View File

@ -1641,7 +1641,11 @@ public:
serveAuditStorageParallelismLock(SERVER_KNOBS->SERVE_AUDIT_STORAGE_PARALLELISM),
instanceID(deterministicRandom()->randomUniqueID().first()), shuttingDown(false), behind(false),
versionBehind(false), debug_inApplyUpdate(false), debug_lastValidateTime(0), lastBytesInputEBrake(0),
lastDurableVersionEBrake(0), maxQueryQueue(0), transactionTagCounter(ssi.id()),
lastDurableVersionEBrake(0), maxQueryQueue(0),
transactionTagCounter(ssi.id(),
/*maxTagsTracked=*/SERVER_KNOBS->SS_THROTTLE_TAGS_TRACKED,
/*minRateTracked=*/SERVER_KNOBS->MIN_TAG_READ_PAGES_RATE *
CLIENT_KNOBS->TAG_THROTTLING_PAGE_SIZE),
busiestWriteTagContext(ssi.id()), counters(this),
storageServerSourceTLogIDEventHolder(
makeReference<EventCacheHolder>(ssi.id().toString() + "/StorageServerSourceTLogID")),