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

Merge remote-tracking branch 'fork/ddsketch' into ddsketch

This commit is contained in:
Sam Gwydir 2022-11-15 12:18:11 -08:00
parent 4254429e73 47a8cebfb3
commit ac729d130d
4 changed files with 34 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
fdbclient/NativeAPI.actor.cpp
View File

@ -1016,9 +1016,7 @@ ACTOR static Future<Void> monitorClientDBInfoChange(DatabaseContext* cx,
proxiesChangeTrigger->trigger();
}
}
when(wait(actors.getResult())) {
UNSTOPPABLE_ASSERT(false);
}
when(wait(actors.getResult())) { UNSTOPPABLE_ASSERT(false); }
}
}
}
@ -1839,13 +1837,13 @@ DatabaseContext::DatabaseContext(const Error& err)
ccBG("BlobGranuleReadMetrics"), bgReadInputBytes("BGReadInputBytes", ccBG),
bgReadOutputBytes("BGReadOutputBytes", ccBG), bgReadSnapshotRows("BGReadSnapshotRows", ccBG),
bgReadRowsCleared("BGReadRowsCleared", ccBG), bgReadRowsInserted("BGReadRowsInserted", ccBG),
bgReadRowsUpdated("BGReadRowsUpdated", ccBG), bgLatencies(1000), bgGranulesPerRequest(1000),
usedAnyChangeFeeds(false), ccFeed("ChangeFeedClientMetrics"), feedStreamStarts("FeedStreamStarts", ccFeed),
bgReadRowsUpdated("BGReadRowsUpdated", ccBG), bgLatencies(), bgGranulesPerRequest(), usedAnyChangeFeeds(false),
ccFeed("ChangeFeedClientMetrics"), feedStreamStarts("FeedStreamStarts", ccFeed),
feedMergeStreamStarts("FeedMergeStreamStarts", ccFeed), feedErrors("FeedErrors", ccFeed),
feedNonRetriableErrors("FeedNonRetriableErrors", ccFeed), feedPops("FeedPops", ccFeed),
feedPopsFallback("FeedPopsFallback", ccFeed), latencies(1000), readLatencies(1000), commitLatencies(1000),
GRVLatencies(1000), mutationsPerCommit(1000), bytesPerCommit(1000), sharedStatePtr(nullptr),
transactionTracingSample(false), smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
feedPopsFallback("FeedPopsFallback", ccFeed), latencies(), readLatencies(), commitLatencies(), GRVLatencies(),
mutationsPerCommit(), bytesPerCommit(), sharedStatePtr(nullptr), transactionTracingSample(false),
smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
connectToDatabaseEventCacheHolder(format("ConnectToDatabase/%s", dbId.toString().c_str())) {}
// Static constructor used by server processes to create a DatabaseContext
@ -3423,9 +3421,7 @@ ACTOR Future<Optional<Value>> getValue(Reference<TransactionState> trState,
std::vector<Error>{ transaction_too_old(), future_version() });
}
choose {
when(wait(trState->cx->connectionFileChanged())) {
throw transaction_too_old();
}
when(wait(trState->cx->connectionFileChanged())) { throw transaction_too_old(); }
when(GetValueReply _reply = wait(loadBalance(
trState->cx.getPtr(),
locationInfo.locations,
@ -3572,9 +3568,7 @@ ACTOR Future<Key> getKey(Reference<TransactionState> trState,
state GetKeyReply reply;
try {
choose {
when(wait(trState->cx->connectionFileChanged())) {
throw transaction_too_old();
}
when(wait(trState->cx->connectionFileChanged())) { throw transaction_too_old(); }
when(GetKeyReply _reply = wait(loadBalance(
trState->cx.getPtr(),
locationInfo.locations,
@ -3738,9 +3732,7 @@ ACTOR Future<Version> watchValue(Database cx, Reference<const WatchParameters> p
TaskPriority::DefaultPromiseEndpoint))) {
resp = r;
}
when(wait(cx->connectionRecord ? cx->connectionRecord->onChange() : Never())) {
wait(Never());
}
when(wait(cx->connectionRecord ? cx->connectionRecord->onChange() : Never())) { wait(Never()); }
}
if (watchValueID.present()) {
g_traceBatch.addEvent("WatchValueDebug", watchValueID.get().first(), "NativeAPI.watchValue.After");
@ -4059,9 +4051,7 @@ Future<RangeResultFamily> getExactRange(Reference<TransactionState> trState,
state GetKeyValuesFamilyReply rep;
try {
choose {
when(wait(trState->cx->connectionFileChanged())) {
throw transaction_too_old();
}
when(wait(trState->cx->connectionFileChanged())) { throw transaction_too_old(); }
when(GetKeyValuesFamilyReply _rep = wait(loadBalance(
trState->cx.getPtr(),
locations[shard].locations,
@ -4960,9 +4950,7 @@ ACTOR Future<Void> getRangeStreamFragment(Reference<TransactionState> trState,
return Void();
}
when(GetKeyValuesStreamReply _rep = waitNext(replyStream.getFuture())) {
rep = _rep;
}
when(GetKeyValuesStreamReply _rep = waitNext(replyStream.getFuture())) { rep = _rep; }
}
++trState->cx->transactionPhysicalReadsCompleted;
} catch (Error& e) {
@ -5455,9 +5443,7 @@ ACTOR Future<Void> watch(Reference<Watch> watch,
loop {
choose {
// NativeAPI watchValue future finishes or errors
when(wait(watch->watchFuture)) {
break;
}
when(wait(watch->watchFuture)) { break; }
when(wait(cx->connectionFileChanged())) {
CODE_PROBE(true, "Recreated a watch after switch");
@ -7042,9 +7028,7 @@ ACTOR Future<GetReadVersionReply> getConsistentReadVersion(SpanContext parentSpa
state Future<Void> onProxiesChanged = cx->onProxiesChanged();
choose {
when(wait(onProxiesChanged)) {
onProxiesChanged = cx->onProxiesChanged();
}
when(wait(onProxiesChanged)) { onProxiesChanged = cx->onProxiesChanged(); }
when(GetReadVersionReply v =
wait(basicLoadBalance(cx->getGrvProxies(UseProvisionalProxies(
flags & GetReadVersionRequest::FLAG_USE_PROVISIONAL_PROXIES)),
@ -7470,9 +7454,7 @@ ACTOR Future<ProtocolVersion> getClusterProtocolImpl(
needToConnect = false;
}
choose {
when(wait(coordinator->onChange())) {
needToConnect = true;
}
when(wait(coordinator->onChange())) { needToConnect = true; }
when(ProtocolVersion pv = wait(protocolVersion)) {
if (!expectedVersion.present() || expectedVersion.get() != pv) {
@ -9032,12 +9014,8 @@ ACTOR Future<std::vector<CheckpointMetaData>> getCheckpointMetaData(Database cx,
}
choose {
when(wait(cx->connectionFileChanged())) {
cx->invalidateCache(KeyRef(), keys);
}
when(wait(waitForAll(futures))) {
break;
}
when(wait(cx->connectionFileChanged())) { cx->invalidateCache(KeyRef(), keys); }
when(wait(waitForAll(futures))) { break; }
when(wait(delay(timeout))) {
TraceEvent("GetCheckpointTimeout").detail("Range", keys).detail("Version", version);
}
@ -9684,12 +9662,8 @@ ACTOR Future<Void> changeFeedWhenAtLatest(Reference<ChangeFeedData> self, Versio
// only allowed to use empty versions if you're caught up
Future<Void> waitEmptyVersion = (self->notAtLatest.get() == 0) ? changeFeedWaitLatest(self, version) : Never();
choose {
when(wait(waitEmptyVersion)) {
break;
}
when(wait(lastReturned)) {
break;
}
when(wait(waitEmptyVersion)) { break; }
when(wait(lastReturned)) { break; }
when(wait(self->refresh.getFuture())) {}
when(wait(self->notAtLatest.onChange())) {}
}

11
fdbrpc/include/fdbrpc/DDSketch.h
View File

@ -96,7 +96,12 @@ public:
try {
buckets.at(index)++;
} catch (std::out_of_range const& e) {
fmt::print(stderr, "ERROR: Invalid DDSketch bucket index ({}) at {}/{} for sample: {}\n", e.what(), index, buckets.size(), sample);
fmt::print(stderr,
"ERROR: Invalid DDSketch bucket index ({}) at {}/{} for sample: {}\n",
e.what(),
index,
buckets.size(),
sample);
}
}
@ -158,7 +163,8 @@ public:
}
}
ASSERT(found);
if (!found) return -1;
if (!found)
return -1;
return static_cast<Impl*>(this)->getValue(index);
}
@ -213,6 +219,7 @@ public:
multiplier(fastLogger::correctingFactor * log(2) / log(gamma)) {
ASSERT(errorGuarantee > 0);
offset = getIndex(1.0 / DDSketchBase<DDSketch<T>, T>::EPS);
ASSERT(offset > 0);
this->setBucketSize(2 * offset);
}

6
fdbrpc/include/fdbrpc/TSSComparison.h
View File

@ -105,9 +105,9 @@ struct TSSMetrics : ReferenceCounted<TSSMetrics>, NonCopyable {
TSSMetrics()
: cc("TSSClientMetrics"), requests("Requests", cc), streamComparisons("StreamComparisons", cc),
ssErrors("SSErrors", cc), tssErrors("TSSErrors", cc), tssTimeouts("TSSTimeouts", cc),
mismatches("Mismatches", cc), SSgetValueLatency(1000), SSgetKeyLatency(1000), SSgetKeyValuesLatency(1000),
SSgetMappedKeyValuesLatency(1000), TSSgetValueLatency(1000), TSSgetKeyLatency(1000),
TSSgetKeyValuesLatency(1000), TSSgetMappedKeyValuesLatency(1000) {}
mismatches("Mismatches", cc), SSgetValueLatency(), SSgetKeyLatency(), SSgetKeyValuesLatency(),
SSgetMappedKeyValuesLatency(), TSSgetValueLatency(), TSSgetKeyLatency(), TSSgetKeyValuesLatency(),
TSSgetMappedKeyValuesLatency() {}
};
template <class Rep>

8
fdbserver/include/fdbserver/workloads/ReadWriteWorkload.actor.h
View File

@ -47,7 +47,7 @@ DESCR struct ReadMetric {
// Common ReadWrite test settings
struct ReadWriteCommon : KVWorkload {
static constexpr int sampleSize = 10000;
static constexpr double sampleError = 0.01;
friend struct ReadWriteCommonImpl;
// general test setting
@ -88,9 +88,9 @@ struct ReadWriteCommon : KVWorkload {
explicit ReadWriteCommon(WorkloadContext const& wcx)
: KVWorkload(wcx), totalReadsMetric("ReadWrite.TotalReads"_sr), totalRetriesMetric("ReadWrite.TotalRetries"_sr),
aTransactions("A Transactions"), bTransactions("B Transactions"), retries("Retries"), latencies(sampleSize),
readLatencies(sampleSize), commitLatencies(sampleSize), GRVLatencies(sampleSize), fullReadLatencies(sampleSize),
readLatencyTotal(0), readLatencyCount(0), loadTime(0.0), clientBegin(0) {
aTransactions("A Transactions"), bTransactions("B Transactions"), retries("Retries"), latencies(sampleError),
readLatencies(sampleError), commitLatencies(sampleError), GRVLatencies(sampleError),
fullReadLatencies(sampleError), readLatencyTotal(0), readLatencyCount(0), loadTime(0.0), clientBegin(0) {
transactionSuccessMetric.init("ReadWrite.SuccessfulTransaction"_sr);
transactionFailureMetric.init("ReadWrite.FailedTransaction"_sr);