apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-05-14 01:42:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add support for returning whether a key-backed type range read has more results or not.

Browse Source
This commit is contained in:
A.J. Beamon 2022-07-13 15:46:10 -07:00
parent b85fbaef52
commit 91949439d2
7 changed files with 109 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
fdbclient/BackupContainer.actor.cpp
View File

@ -393,21 +393,21 @@ ACTOR Future<Version> timeKeeperVersionFromDatetime(std::string datetime, Databa
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
state std::vector<std::pair<int64_t, Version>> results =
state KeyBackedRangeResult<std::pair<int64_t, Version>> rangeResult =
wait(versionMap.getRange(tr, 0, time, 1, Snapshot::False, Reverse::True));
if (results.size() != 1) {
if (rangeResult.results.size() != 1) {
// No key less than time was found in the database
// Look for a key >= time.
wait(store(results, versionMap.getRange(tr, time, std::numeric_limits<int64_t>::max(), 1)));
wait(store(rangeResult, versionMap.getRange(tr, time, std::numeric_limits<int64_t>::max(), 1)));
if (results.size() != 1) {
if (rangeResult.results.size() != 1) {
fprintf(stderr, "ERROR: Unable to calculate a version for given date/time.\n");
throw backup_error();
}
}
// Adjust version found by the delta between time and the time found and min with 0.
auto& result = results[0];
auto& result = rangeResult.results[0];
return std::max<Version>(0, result.second + (time - result.first) * CLIENT_KNOBS->CORE_VERSIONSPERSECOND);
} catch (Error& e) {
@ -432,21 +432,21 @@ ACTOR Future<Optional<int64_t>> timeKeeperEpochsFromVersion(Version v, Reference
mid = (min + max + 1) / 2; // ceiling
// Find the highest time < mid
state std::vector<std::pair<int64_t, Version>> results =
state KeyBackedRangeResult<std::pair<int64_t, Version>> rangeResult =
wait(versionMap.getRange(tr, min, mid, 1, Snapshot::False, Reverse::True));
if (results.size() != 1) {
if (rangeResult.results.size() != 1) {
if (mid == min) {
// There aren't any records having a version < v, so just look for any record having a time < now
// and base a result on it
wait(store(results, versionMap.getRange(tr, 0, (int64_t)now(), 1)));
wait(store(rangeResult, versionMap.getRange(tr, 0, (int64_t)now(), 1)));
if (results.size() != 1) {
if (rangeResult.results.size() != 1) {
// There aren't any timekeeper records to base a result on so return nothing
return Optional<int64_t>();
}
found = results[0];
found = rangeResult.results[0];
break;
}
@ -454,7 +454,7 @@ ACTOR Future<Optional<int64_t>> timeKeeperEpochsFromVersion(Version v, Reference
continue;
}
found = results[0];
found = rangeResult.results[0];
if (v < found.second) {
max = found.first;

48
fdbclient/FileBackupAgent.actor.cpp
View File

@ -123,9 +123,9 @@ ACTOR Future<std::vector<KeyBackedTag>> TagUidMap::getAll_impl(TagUidMap* tagsMa
Reference<ReadYourWritesTransaction> tr,
Snapshot snapshot) {
state Key prefix = tagsMap->prefix; // Copying it here as tagsMap lifetime is not tied to this actor
TagMap::PairsType tagPairs = wait(tagsMap->getRange(tr, std::string(), {}, 1e6, snapshot));
TagMap::RangeResultType tagPairs = wait(tagsMap->getRange(tr, std::string(), {}, 1e6, snapshot));
std::vector<KeyBackedTag> results;
for (auto& p : tagPairs)
for (auto& p : tagPairs.results)
results.push_back(KeyBackedTag(p.first, prefix));
return results;
}
@ -1564,18 +1564,22 @@ struct BackupSnapshotDispatchTask : BackupTaskFuncBase {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
state Future<std::vector<std::pair<Key, bool>>> bounds = config.snapshotRangeDispatchMap().getRange(
tr, beginKey, keyAfter(normalKeys.end), CLIENT_KNOBS->TOO_MANY);
state Future<BackupConfig::RangeDispatchMapT::RangeResultType> bounds =
config.snapshotRangeDispatchMap().getRange(
tr, beginKey, keyAfter(normalKeys.end), CLIENT_KNOBS->TOO_MANY);
wait(success(bounds) && taskBucket->keepRunning(tr, task) &&
store(recentReadVersion, tr->getReadVersion()));
if (bounds.get().empty())
if (!bounds.get().results.empty()) {
dispatchBoundaries.reserve(dispatchBoundaries.size() + bounds.get().results.size());
dispatchBoundaries.insert(
dispatchBoundaries.end(), bounds.get().results.begin(), bounds.get().results.end());
}
if (!bounds.get().more)
break;
dispatchBoundaries.reserve(dispatchBoundaries.size() + bounds.get().size());
dispatchBoundaries.insert(dispatchBoundaries.end(), bounds.get().begin(), bounds.get().end());
beginKey = keyAfter(bounds.get().back().first);
beginKey = keyAfter(bounds.get().results.back().first);
tr->reset();
} catch (Error& e) {
wait(tr->onError(e));
@ -2543,17 +2547,17 @@ struct BackupSnapshotManifest : BackupTaskFuncBase {
wait(store(bc, config.backupContainer().getOrThrow(tr)));
}
BackupConfig::RangeFileMapT::PairsType rangeresults =
BackupConfig::RangeFileMapT::RangeResultType rangeresults =
wait(config.snapshotRangeFileMap().getRange(tr, startKey, {}, batchSize));
for (auto& p : rangeresults) {
for (auto& p : rangeresults.results) {
localmap.insert(p);
}
if (rangeresults.size() < batchSize)
if (!rangeresults.more)
break;
startKey = keyAfter(rangeresults.back().first);
startKey = keyAfter(rangeresults.results.back().first);
tr->reset();
} catch (Error& e) {
wait(tr->onError(e));
@ -3617,7 +3621,7 @@ struct RestoreDispatchTaskFunc : RestoreTaskFuncBase {
// Get a batch of files. We're targeting batchSize blocks being dispatched so query for batchSize files (each
// of which is 0 or more blocks).
state int taskBatchSize = BUGGIFY ? 1 : CLIENT_KNOBS->RESTORE_DISPATCH_ADDTASK_SIZE;
state RestoreConfig::FileSetT::Values files = wait(restore.fileSet().getRange(
state RestoreConfig::FileSetT::RangeResultType files = wait(restore.fileSet().getRange(
tr, Optional<RestoreConfig::RestoreFile>({ beginVersion, beginFile }), {}, taskBatchSize));
// allPartsDone will be set once all block tasks in the current batch are finished.
@ -3636,7 +3640,7 @@ struct RestoreDispatchTaskFunc : RestoreTaskFuncBase {
}
// If there were no files to load then this batch is done and restore is almost done.
if (files.size() == 0) {
if (files.results.size() == 0) {
// If adding to existing batch then blocks could be in progress so create a new Dispatch task that waits for
// them to finish
if (addingToExistingBatch) {
@ -3714,13 +3718,13 @@ struct RestoreDispatchTaskFunc : RestoreTaskFuncBase {
// blocks per Dispatch task and target batchSize total per batch but a batch must end on a complete version
// boundary so exceed the limit if necessary to reach the end of a version of files.
state std::vector<Future<Key>> addTaskFutures;
state Version endVersion = files[0].version;
state Version endVersion = files.results[0].version;
state int blocksDispatched = 0;
state int64_t beginBlock = Params.beginBlock().getOrDefault(task);
state int i = 0;
for (; i < files.size(); ++i) {
RestoreConfig::RestoreFile& f = files[i];
for (; i < files.results.size(); ++i) {
RestoreConfig::RestoreFile& f = files.results[i];
// Here we are "between versions" (prior to adding the first block of the first file of a new version) so
// this is an opportunity to end the current dispatch batch (which must end on a version boundary) if the
@ -5057,11 +5061,11 @@ public:
doc.setKey("CurrentSnapshot", snapshot);
}
KeyBackedMap<int64_t, std::pair<std::string, Version>>::PairsType errors =
KeyBackedMap<int64_t, std::pair<std::string, Version>>::RangeResultType errors =
wait(config.lastErrorPerType().getRange(
tr, 0, std::numeric_limits<int>::max(), CLIENT_KNOBS->TOO_MANY));
JsonBuilderArray errorList;
for (auto& e : errors) {
for (auto& e : errors.results) {
std::string msg = e.second.first;
Version ver = e.second.second;
@ -5209,13 +5213,13 @@ public:
// Append the errors, if requested
if (showErrors) {
KeyBackedMap<int64_t, std::pair<std::string, Version>>::PairsType errors =
KeyBackedMap<int64_t, std::pair<std::string, Version>>::RangeResultType errors =
wait(config.lastErrorPerType().getRange(
tr, 0, std::numeric_limits<int>::max(), CLIENT_KNOBS->TOO_MANY));
std::string recentErrors;
std::string pastErrors;
for (auto& e : errors) {
for (auto& e : errors.results) {
Version v = e.second.second;
std::string msg = format(
"%s ago : %s\n",

5
fdbclient/NativeAPI.actor.cpp
View File

@ -7948,7 +7948,8 @@ ACTOR Future<std::vector<std::pair<UID, StorageWiggleValue>>> readStorageWiggleV
state KeyBackedObjectMap<UID, StorageWiggleValue, decltype(IncludeVersion())> metadataMap(readKey,
IncludeVersion());
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
state std::vector<std::pair<UID, StorageWiggleValue>> res;
state KeyBackedRangeResult<std::pair<UID, StorageWiggleValue>> res;
// read the wiggling pairs
loop {
try {
@ -7964,7 +7965,7 @@ ACTOR Future<std::vector<std::pair<UID, StorageWiggleValue>>> readStorageWiggleV
wait(tr->onError(e));
}
}
return res;
return res.results;
}
ACTOR Future<Void> splitStorageMetricsStream(PromiseStream<Key> resultStream,

97
fdbclient/include/fdbclient/KeyBackedTypes.h
View File

@ -151,6 +151,12 @@ inline KeyRange TupleCodec<KeyRange>::unpack(Standalone<StringRef> const& val) {
return KeyRangeRef(t.getString(0), t.getString(1));
}
template <typename ResultType>
struct KeyBackedRangeResult {
std::vector<ResultType> results;
bool more;
};
// Convenient read/write access to a single value of type T stored at key
// Even though 'this' is not actually mutated, methods that change the db key are not const.
template <typename T, typename Codec = TupleCodec<T>>
@ -308,16 +314,16 @@ public:
typedef _KeyType KeyType;
typedef _ValueType ValueType;
typedef std::pair<KeyType, ValueType> PairType;
typedef std::vector<PairType> PairsType;
typedef KeyBackedRangeResult<PairType> RangeResultType;
// If end is not present one key past the end of the map is used.
template <class Transaction>
Future<PairsType> getRange(Transaction tr,
Optional<KeyType> const& begin,
Optional<KeyType> const& end,
int limit,
Snapshot snapshot = Snapshot::False,
Reverse reverse = Reverse::False) const {
Future<RangeResultType> getRange(Transaction tr,
Optional<KeyType> const& begin,
Optional<KeyType> const& end,
int limit,
Snapshot snapshot = Snapshot::False,
Reverse reverse = Reverse::False) const {
Key prefix = subspace.begin; // 'this' could be invalid inside lambda
Key beginKey = begin.present() ? prefix.withSuffix(KeyCodec::pack(begin.get())) : subspace.begin;
@ -326,16 +332,18 @@ public:
typename transaction_future_type<Transaction, RangeResult>::type getRangeFuture =
tr->getRange(KeyRangeRef(beginKey, endKey), GetRangeLimits(limit), snapshot, reverse);
return holdWhile(getRangeFuture,
map(safeThreadFutureToFuture(getRangeFuture), [prefix](RangeResult const& kvs) -> PairsType {
PairsType results;
for (int i = 0; i < kvs.size(); ++i) {
KeyType key = KeyCodec::unpack(kvs[i].key.removePrefix(prefix));
ValueType val = ValueCodec::unpack(kvs[i].value);
results.push_back(PairType(key, val));
}
return results;
}));
return holdWhile(
getRangeFuture,
map(safeThreadFutureToFuture(getRangeFuture), [prefix](RangeResult const& kvs) -> RangeResultType {
RangeResultType rangeResult;
for (int i = 0; i < kvs.size(); ++i) {
KeyType key = KeyCodec::unpack(kvs[i].key.removePrefix(prefix));
ValueType val = ValueCodec::unpack(kvs[i].value);
rangeResult.results.push_back(PairType(key, val));
}
rangeResult.more = kvs.more;
return rangeResult;
}));
}
template <class Transaction>
@ -497,15 +505,15 @@ public:
typedef _KeyType KeyType;
typedef _ValueType ValueType;
typedef std::pair<KeyType, ValueType> PairType;
typedef std::vector<PairType> PairsType;
typedef KeyBackedRangeResult<PairType> RangeResultType;
template <class Transaction>
Future<PairsType> getRange(Transaction tr,
Optional<KeyType> const& begin,
Optional<KeyType> const& end,
int limit,
Snapshot snapshot = Snapshot::False,
Reverse reverse = Reverse::False) const {
Future<RangeResultType> getRange(Transaction tr,
Optional<KeyType> const& begin,
Optional<KeyType> const& end,
int limit,
Snapshot snapshot = Snapshot::False,
Reverse reverse = Reverse::False) const {
Key beginKey = begin.present() ? subspace.begin.withSuffix(KeyCodec::pack(begin.get())) : subspace.begin;
Key endKey = end.present() ? subspace.begin.withSuffix(KeyCodec::pack(end.get())) : subspace.end;
@ -514,14 +522,15 @@ public:
return holdWhile(
getRangeFuture,
map(safeThreadFutureToFuture(getRangeFuture), [self = *this](RangeResult const& kvs) -> PairsType {
PairsType results;
map(safeThreadFutureToFuture(getRangeFuture), [self = *this](RangeResult const& kvs) -> RangeResultType {
RangeResultType rangeResult;
for (int i = 0; i < kvs.size(); ++i) {
KeyType key = KeyCodec::unpack(kvs[i].key.removePrefix(self.subspace.begin));
ValueType val = ObjectReader::fromStringRef<ValueType>(kvs[i].value, self.versionOptions);
results.push_back(PairType(key, val));
rangeResult.results.push_back(PairType(key, val));
}
return results;
rangeResult.more = kvs.more;
return rangeResult;
}));
}
@ -584,15 +593,15 @@ public:
KeyBackedSet(KeyRef key) : subspace(prefixRange(key)) {}
typedef _ValueType ValueType;
typedef std::vector<ValueType> Values;
typedef KeyBackedRangeResult<ValueType> RangeResultType;
template <class Transaction>
Future<Values> getRange(Transaction tr,
Optional<ValueType> const& begin,
Optional<ValueType> const& end,
int limit,
Snapshot snapshot = Snapshot::False,
Reverse reverse = Reverse::False) const {
Future<RangeResultType> getRange(Transaction tr,
Optional<ValueType> const& begin,
Optional<ValueType> const& end,
int limit,
Snapshot snapshot = Snapshot::False,
Reverse reverse = Reverse::False) const {
Key prefix = subspace.begin; // 'this' could be invalid inside lambda
Key beginKey = begin.present() ? prefix.withSuffix(Codec::pack(begin.get())) : subspace.begin;
Key endKey = end.present() ? prefix.withSuffix(Codec::pack(end.get())) : subspace.end;
@ -600,14 +609,16 @@ public:
typename transaction_future_type<Transaction, RangeResult>::type getRangeFuture =
tr->getRange(KeyRangeRef(beginKey, endKey), GetRangeLimits(limit), snapshot, reverse);
return holdWhile(getRangeFuture,
map(safeThreadFutureToFuture(getRangeFuture), [prefix](RangeResult const& kvs) -> Values {
Values results;
for (int i = 0; i < kvs.size(); ++i) {
results.push_back(Codec::unpack(kvs[i].key.removePrefix(prefix)));
}
return results;
}));
return holdWhile(
getRangeFuture,
map(safeThreadFutureToFuture(getRangeFuture), [prefix](RangeResult const& kvs) -> RangeResultType {
RangeResultType rangeResult;
for (int i = 0; i < kvs.size(); ++i) {
rangeResult.results.push_back(Codec::unpack(kvs[i].key.removePrefix(prefix)));
}
rangeResult.more = kvs.more;
return rangeResult;
}));
}
template <class Transaction>

6
fdbserver/MetricLogger.actor.cpp
View File

@ -106,7 +106,7 @@ struct MetricsConfig {
typedef KeyBackedMap<int64_t, MetricsRule> RuleMapT;
RuleMapT ruleMap;
RuleMapT::PairsType rules;
RuleMapT::RangeResultType rules;
KeyBackedMap<Key, int64_t> addressMap;
KeyBackedMap<std::pair<Key, Key>, int64_t> nameAndTypeMap;
@ -152,11 +152,11 @@ ACTOR Future<Void> metricRuleUpdater(Database cx, MetricsConfig* config, TDMetri
state Future<Void> newMetric = collection->metricAdded.onTrigger();
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
MetricsConfig::RuleMapT::PairsType rules = wait(config->ruleMap.getRange(tr, 0, {}, 1e6));
MetricsConfig::RuleMapT::RangeResultType rules = wait(config->ruleMap.getRange(tr, 0, {}, 1e6));
for (auto& it : collection->metricMap) {
it.value->setConfig(false);
for (auto i = rules.rbegin(); !(i == rules.rend()); ++i)
for (auto i = rules.results.rbegin(); !(i == rules.results.rend()); ++i)
if (i->second.applyTo(it.value.getPtr(), collection->address))
break;
}

11
fdbserver/TSSMappingUtil.actor.cpp
View File

@ -26,14 +26,15 @@
ACTOR Future<Void> readTSSMappingRYW(Reference<ReadYourWritesTransaction> tr,
std::map<UID, StorageServerInterface>* tssMapping) {
KeyBackedMap<UID, UID> tssMapDB = KeyBackedMap<UID, UID>(tssMappingKeys.begin);
state std::vector<std::pair<UID, UID>> uidMapping =
state KeyBackedMap<UID, UID>::RangeResultType uidMapping =
wait(tssMapDB.getRange(tr, UID(), Optional<UID>(), CLIENT_KNOBS->TOO_MANY));
ASSERT(uidMapping.size() < CLIENT_KNOBS->TOO_MANY);
ASSERT(uidMapping.results.size() < CLIENT_KNOBS->TOO_MANY);
state std::map<UID, StorageServerInterface> mapping;
for (auto& it : uidMapping) {
state UID ssId = it.first;
Optional<Value> v = wait(tr->get(serverListKeyFor(it.second)));
state std::vector<std::pair<UID, UID>>::iterator mapItr;
for (mapItr = uidMapping.results.begin(); mapItr != uidMapping.results.end(); ++mapItr) {
state UID ssId = mapItr->first;
Optional<Value> v = wait(tr->get(serverListKeyFor(mapItr->second)));
(*tssMapping)[ssId] = decodeServerListValue(v.get());
}
return Void();

12
fdbserver/workloads/TimeKeeperCorrectness.actor.cpp
View File

@ -79,23 +79,23 @@ struct TimeKeeperCorrectnessWorkload : TestWorkload {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
std::vector<std::pair<int64_t, Version>> allItems =
KeyBackedRangeResult<std::pair<int64_t, Version>> allItems =
wait(dbTimeKeeper.getRange(tr, 0, Optional<int64_t>(), self->inMemTimeKeeper.size() + 2));
if (allItems.size() > SERVER_KNOBS->TIME_KEEPER_MAX_ENTRIES + 1) {
if (allItems.results.size() > SERVER_KNOBS->TIME_KEEPER_MAX_ENTRIES + 1) {
TraceEvent(SevError, "TKCorrectness_TooManyEntries")
.detail("Expected", SERVER_KNOBS->TIME_KEEPER_MAX_ENTRIES + 1)
.detail("Found", allItems.size());
.detail("Found", allItems.results.size());
return false;
}
if (allItems.size() < self->testDuration / SERVER_KNOBS->TIME_KEEPER_DELAY) {
if (allItems.results.size() < self->testDuration / SERVER_KNOBS->TIME_KEEPER_DELAY) {
TraceEvent(SevWarnAlways, "TKCorrectness_TooFewEntries")
.detail("Expected", self->testDuration / SERVER_KNOBS->TIME_KEEPER_DELAY)
.detail("Found", allItems.size());
.detail("Found", allItems.results.size());
}
for (auto item : allItems) {
for (auto item : allItems.results) {
auto it = self->inMemTimeKeeper.lower_bound(item.first);
if (it == self->inMemTimeKeeper.end()) {
continue;