350 lines
12 KiB
C++
350 lines
12 KiB
C++
/*
|
|
* SimpleConfigTransaction.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 <algorithm>
|
|
|
|
#include "fdbclient/CommitTransaction.h"
|
|
#include "fdbclient/DatabaseContext.h"
|
|
#include "fdbclient/IKnobCollection.h"
|
|
#include "fdbclient/SimpleConfigTransaction.h"
|
|
#include "flow/Arena.h"
|
|
#include "flow/actorcompiler.h" // This must be the last #include.
|
|
|
|
class SimpleConfigTransactionImpl {
|
|
ConfigTransactionCommitRequest toCommit;
|
|
Future<ConfigGeneration> getGenerationFuture;
|
|
ConfigTransactionInterface cti;
|
|
int numRetries{ 0 };
|
|
bool committed{ false };
|
|
Optional<UID> dID;
|
|
Database cx;
|
|
|
|
ACTOR static Future<ConfigGeneration> getGeneration(SimpleConfigTransactionImpl* self) {
|
|
if (self->dID.present()) {
|
|
TraceEvent("SimpleConfigTransactionGettingReadVersion", self->dID.get());
|
|
}
|
|
state ConfigTransactionGetGenerationReply reply;
|
|
if (self->cti.hostname.present()) {
|
|
wait(store(reply,
|
|
retryGetReplyFromHostname(ConfigTransactionGetGenerationRequest{ 0, Optional<Version>() },
|
|
self->cti.hostname.get(),
|
|
WLTOKEN_CONFIGTXN_GETGENERATION)));
|
|
} else {
|
|
wait(store(reply,
|
|
retryBrokenPromise(self->cti.getGeneration,
|
|
ConfigTransactionGetGenerationRequest{ 0, Optional<Version>() })));
|
|
}
|
|
if (self->dID.present()) {
|
|
TraceEvent("SimpleConfigTransactionGotReadVersion", self->dID.get())
|
|
.detail("Version", reply.generation.liveVersion);
|
|
}
|
|
return reply.generation;
|
|
}
|
|
|
|
ACTOR static Future<Optional<Value>> get(SimpleConfigTransactionImpl* self, KeyRef key) {
|
|
if (!self->getGenerationFuture.isValid()) {
|
|
self->getGenerationFuture = getGeneration(self);
|
|
}
|
|
state ConfigKey configKey = ConfigKey::decodeKey(key);
|
|
ConfigGeneration generation = wait(self->getGenerationFuture);
|
|
if (self->dID.present()) {
|
|
TraceEvent("SimpleConfigTransactionGettingValue", self->dID.get())
|
|
.detail("ConfigClass", configKey.configClass)
|
|
.detail("KnobName", configKey.knobName);
|
|
}
|
|
state ConfigTransactionGetReply reply;
|
|
if (self->cti.hostname.present()) {
|
|
wait(store(reply,
|
|
retryGetReplyFromHostname(ConfigTransactionGetRequest{ 0, generation, configKey },
|
|
self->cti.hostname.get(),
|
|
WLTOKEN_CONFIGTXN_GET)));
|
|
} else {
|
|
wait(store(reply,
|
|
retryBrokenPromise(self->cti.get, ConfigTransactionGetRequest{ 0, generation, configKey })));
|
|
}
|
|
if (self->dID.present()) {
|
|
TraceEvent("SimpleConfigTransactionGotValue", self->dID.get())
|
|
.detail("Value", reply.value.get().toString());
|
|
}
|
|
if (reply.value.present()) {
|
|
return reply.value.get().toValue();
|
|
} else {
|
|
return Optional<Value>{};
|
|
}
|
|
}
|
|
|
|
ACTOR static Future<RangeResult> getConfigClasses(SimpleConfigTransactionImpl* self) {
|
|
if (!self->getGenerationFuture.isValid()) {
|
|
self->getGenerationFuture = getGeneration(self);
|
|
}
|
|
ConfigGeneration generation = wait(self->getGenerationFuture);
|
|
state ConfigTransactionGetConfigClassesReply reply;
|
|
if (self->cti.hostname.present()) {
|
|
wait(store(reply,
|
|
retryGetReplyFromHostname(ConfigTransactionGetConfigClassesRequest{ 0, generation },
|
|
self->cti.hostname.get(),
|
|
WLTOKEN_CONFIGTXN_GETCLASSES)));
|
|
} else {
|
|
wait(store(
|
|
reply,
|
|
retryBrokenPromise(self->cti.getClasses, ConfigTransactionGetConfigClassesRequest{ 0, generation })));
|
|
}
|
|
RangeResult result;
|
|
for (const auto& configClass : reply.configClasses) {
|
|
result.push_back_deep(result.arena(), KeyValueRef(configClass, ""_sr));
|
|
}
|
|
return result;
|
|
}
|
|
|
|
ACTOR static Future<RangeResult> getKnobs(SimpleConfigTransactionImpl* self, Optional<Key> configClass) {
|
|
if (!self->getGenerationFuture.isValid()) {
|
|
self->getGenerationFuture = getGeneration(self);
|
|
}
|
|
ConfigGeneration generation = wait(self->getGenerationFuture);
|
|
state ConfigTransactionGetKnobsReply reply;
|
|
if (self->cti.hostname.present()) {
|
|
wait(store(reply,
|
|
retryGetReplyFromHostname(ConfigTransactionGetKnobsRequest{ 0, generation, configClass },
|
|
self->cti.hostname.get(),
|
|
WLTOKEN_CONFIGTXN_GETKNOBS)));
|
|
} else {
|
|
wait(store(reply,
|
|
retryBrokenPromise(self->cti.getKnobs,
|
|
ConfigTransactionGetKnobsRequest{ 0, generation, configClass })));
|
|
}
|
|
RangeResult result;
|
|
for (const auto& knobName : reply.knobNames) {
|
|
result.push_back_deep(result.arena(), KeyValueRef(knobName, ""_sr));
|
|
}
|
|
return result;
|
|
}
|
|
|
|
ACTOR static Future<Void> commit(SimpleConfigTransactionImpl* self) {
|
|
if (!self->getGenerationFuture.isValid()) {
|
|
self->getGenerationFuture = getGeneration(self);
|
|
}
|
|
self->toCommit.coordinatorsHash = 0;
|
|
wait(store(self->toCommit.generation, self->getGenerationFuture));
|
|
self->toCommit.annotation.timestamp = now();
|
|
if (self->cti.hostname.present()) {
|
|
wait(retryGetReplyFromHostname(self->toCommit, self->cti.hostname.get(), WLTOKEN_CONFIGTXN_COMMIT));
|
|
} else {
|
|
wait(retryBrokenPromise(self->cti.commit, self->toCommit));
|
|
}
|
|
self->committed = true;
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> onError(SimpleConfigTransactionImpl* self, Error e) {
|
|
// TODO: Improve this:
|
|
if (e.code() == error_code_transaction_too_old || e.code() == error_code_not_committed) {
|
|
wait(delay((1 << self->numRetries++) * 0.01 * deterministicRandom()->random01()));
|
|
self->reset();
|
|
return Void();
|
|
}
|
|
throw e;
|
|
}
|
|
|
|
public:
|
|
SimpleConfigTransactionImpl(Database const& cx) : cx(cx) {
|
|
const ClusterConnectionString& cs = cx->getConnectionRecord()->getConnectionString();
|
|
if (cs.coords.size()) {
|
|
std::vector<NetworkAddress> coordinators = cs.coords;
|
|
std::sort(coordinators.begin(), coordinators.end());
|
|
cti = ConfigTransactionInterface(coordinators[0]);
|
|
} else {
|
|
cti = ConfigTransactionInterface(cs.hostnames[0]);
|
|
}
|
|
}
|
|
|
|
SimpleConfigTransactionImpl(ConfigTransactionInterface const& cti) : cti(cti) {}
|
|
|
|
void set(KeyRef key, ValueRef value) {
|
|
toCommit.mutations.push_back_deep(toCommit.arena,
|
|
IKnobCollection::createSetMutation(toCommit.arena, key, value));
|
|
}
|
|
|
|
void clear(KeyRef key) {
|
|
toCommit.mutations.push_back_deep(toCommit.arena, IKnobCollection::createClearMutation(toCommit.arena, key));
|
|
}
|
|
|
|
Future<Optional<Value>> get(KeyRef key) { return get(this, key); }
|
|
|
|
Future<RangeResult> getRange(KeyRangeRef keys) {
|
|
if (keys == configClassKeys) {
|
|
return getConfigClasses(this);
|
|
} else if (keys == globalConfigKnobKeys) {
|
|
return getKnobs(this, {});
|
|
} else if (configKnobKeys.contains(keys) && keys.singleKeyRange()) {
|
|
const auto configClass = keys.begin.removePrefix(configKnobKeys.begin);
|
|
return getKnobs(this, configClass);
|
|
} else {
|
|
throw invalid_config_db_range_read();
|
|
}
|
|
}
|
|
|
|
Future<Void> commit() { return commit(this); }
|
|
|
|
Future<Void> onError(Error const& e) { return onError(this, e); }
|
|
|
|
Future<Version> getReadVersion() {
|
|
if (!getGenerationFuture.isValid())
|
|
getGenerationFuture = getGeneration(this);
|
|
return map(getGenerationFuture, [](auto const& gen) { return gen.committedVersion; });
|
|
}
|
|
|
|
Optional<Version> getCachedReadVersion() const {
|
|
if (getGenerationFuture.isValid() && getGenerationFuture.isReady() && !getGenerationFuture.isError()) {
|
|
return getGenerationFuture.get().committedVersion;
|
|
} else {
|
|
return {};
|
|
}
|
|
}
|
|
|
|
Version getCommittedVersion() const { return committed ? getGenerationFuture.get().liveVersion : ::invalidVersion; }
|
|
|
|
void reset() {
|
|
getGenerationFuture = Future<ConfigGeneration>{};
|
|
toCommit = {};
|
|
committed = false;
|
|
}
|
|
|
|
void fullReset() {
|
|
numRetries = 0;
|
|
dID = {};
|
|
reset();
|
|
}
|
|
|
|
size_t getApproximateSize() const { return toCommit.expectedSize(); }
|
|
|
|
void debugTransaction(UID dID) { this->dID = dID; }
|
|
|
|
void checkDeferredError(Error const& deferredError) const {
|
|
if (deferredError.code() != invalid_error_code) {
|
|
throw deferredError;
|
|
}
|
|
if (cx.getPtr()) {
|
|
cx->checkDeferredError();
|
|
}
|
|
}
|
|
}; // SimpleConfigTransactionImpl
|
|
|
|
Future<Version> SimpleConfigTransaction::getReadVersion() {
|
|
return impl->getReadVersion();
|
|
}
|
|
|
|
Optional<Version> SimpleConfigTransaction::getCachedReadVersion() const {
|
|
return impl->getCachedReadVersion();
|
|
}
|
|
|
|
Future<Optional<Value>> SimpleConfigTransaction::get(Key const& key, Snapshot snapshot) {
|
|
return impl->get(key);
|
|
}
|
|
|
|
Future<RangeResult> SimpleConfigTransaction::getRange(KeySelector const& begin,
|
|
KeySelector const& end,
|
|
int limit,
|
|
Snapshot snapshot,
|
|
Reverse reverse) {
|
|
if (reverse) {
|
|
throw client_invalid_operation();
|
|
}
|
|
return impl->getRange(KeyRangeRef(begin.getKey(), end.getKey()));
|
|
}
|
|
|
|
Future<RangeResult> SimpleConfigTransaction::getRange(KeySelector begin,
|
|
KeySelector end,
|
|
GetRangeLimits limits,
|
|
Snapshot snapshot,
|
|
Reverse reverse) {
|
|
if (reverse) {
|
|
throw client_invalid_operation();
|
|
}
|
|
return impl->getRange(KeyRangeRef(begin.getKey(), end.getKey()));
|
|
}
|
|
|
|
void SimpleConfigTransaction::set(KeyRef const& key, ValueRef const& value) {
|
|
impl->set(key, value);
|
|
}
|
|
|
|
void SimpleConfigTransaction::clear(KeyRef const& key) {
|
|
impl->clear(key);
|
|
}
|
|
|
|
Future<Void> SimpleConfigTransaction::commit() {
|
|
return impl->commit();
|
|
}
|
|
|
|
Version SimpleConfigTransaction::getCommittedVersion() const {
|
|
return impl->getCommittedVersion();
|
|
}
|
|
|
|
double SimpleConfigTransaction::getTagThrottledDuration() const {
|
|
return 0.0;
|
|
}
|
|
|
|
int64_t SimpleConfigTransaction::getTotalCost() const {
|
|
return 0;
|
|
}
|
|
|
|
int64_t SimpleConfigTransaction::getApproximateSize() const {
|
|
return impl->getApproximateSize();
|
|
}
|
|
|
|
void SimpleConfigTransaction::setOption(FDBTransactionOptions::Option option, Optional<StringRef> value) {
|
|
// TODO: Support using this option to determine atomicity
|
|
}
|
|
|
|
Future<Void> SimpleConfigTransaction::onError(Error const& e) {
|
|
return impl->onError(e);
|
|
}
|
|
|
|
void SimpleConfigTransaction::cancel() {
|
|
// TODO: Implement someday
|
|
throw client_invalid_operation();
|
|
}
|
|
|
|
void SimpleConfigTransaction::reset() {
|
|
return impl->reset();
|
|
}
|
|
|
|
void SimpleConfigTransaction::fullReset() {
|
|
return impl->fullReset();
|
|
}
|
|
|
|
void SimpleConfigTransaction::debugTransaction(UID dID) {
|
|
impl->debugTransaction(dID);
|
|
}
|
|
|
|
void SimpleConfigTransaction::checkDeferredError() const {
|
|
impl->checkDeferredError(deferredError);
|
|
}
|
|
|
|
void SimpleConfigTransaction::construct(Database const& cx) {
|
|
impl = PImpl<SimpleConfigTransactionImpl>::create(cx);
|
|
}
|
|
|
|
SimpleConfigTransaction::SimpleConfigTransaction(ConfigTransactionInterface const& cti)
|
|
: impl(PImpl<SimpleConfigTransactionImpl>::create(cti)) {}
|
|
|
|
SimpleConfigTransaction::SimpleConfigTransaction() = default;
|
|
|
|
SimpleConfigTransaction::~SimpleConfigTransaction() = default;
|