Add some simple implementations to PaxosConfigTransaction methods
This commit is contained in:
parent
54fb0cbe85
commit
de871da75f
|
@ -20,6 +20,7 @@
|
|||
|
||||
#include "fdbclient/ConfigTransactionInterface.h"
|
||||
#include "fdbclient/CoordinationInterface.h"
|
||||
#include "fdbclient/SystemData.h"
|
||||
#include "flow/IRandom.h"
|
||||
|
||||
ConfigTransactionInterface::ConfigTransactionInterface() : _id(deterministicRandom()->randomUniqueID()) {}
|
||||
|
@ -53,3 +54,22 @@ bool ConfigGeneration::operator==(ConfigGeneration const& rhs) const {
|
|||
bool ConfigGeneration::operator!=(ConfigGeneration const& rhs) const {
|
||||
return !(*this == rhs);
|
||||
}
|
||||
|
||||
void ConfigTransactionCommitRequest::set(KeyRef key, ValueRef value) {
|
||||
if (key == configTransactionDescriptionKey) {
|
||||
annotation.description = KeyRef(arena, value);
|
||||
} else {
|
||||
ConfigKey configKey = ConfigKeyRef::decodeKey(key);
|
||||
auto knobValue = IKnobCollection::parseKnobValue(
|
||||
configKey.knobName.toString(), value.toString(), IKnobCollection::Type::TEST);
|
||||
mutations.emplace_back_deep(arena, configKey, knobValue.contents());
|
||||
}
|
||||
}
|
||||
|
||||
void ConfigTransactionCommitRequest::clear(KeyRef key) {
|
||||
if (key == configTransactionDescriptionKey) {
|
||||
annotation.description = ""_sr;
|
||||
} else {
|
||||
mutations.emplace_back_deep(arena, ConfigKeyRef::decodeKey(key), Optional<KnobValueRef>{});
|
||||
}
|
||||
}
|
||||
|
|
|
@ -101,6 +101,9 @@ struct ConfigTransactionCommitRequest {
|
|||
|
||||
size_t expectedSize() const { return mutations.expectedSize() + annotation.expectedSize(); }
|
||||
|
||||
void set(KeyRef key, ValueRef value);
|
||||
void clear(KeyRef key);
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, arena, generation, mutations, annotation, reply);
|
||||
|
|
|
@ -18,24 +18,125 @@
|
|||
* limitations under the License.
|
||||
*/
|
||||
|
||||
#include "fdbclient/DatabaseContext.h"
|
||||
#include "fdbclient/PaxosConfigTransaction.h"
|
||||
#include "flow/actorcompiler.h" // must be last include
|
||||
|
||||
class PaxosConfigTransactionImpl {};
|
||||
class PaxosConfigTransactionImpl {
|
||||
ConfigTransactionCommitRequest toCommit;
|
||||
Future<ConfigGeneration> getGenerationFuture;
|
||||
std::vector<ConfigTransactionInterface> ctis;
|
||||
int numRetries{ 0 };
|
||||
bool committed{ false };
|
||||
Optional<UID> dID;
|
||||
Database cx;
|
||||
|
||||
ACTOR static Future<ConfigGeneration> getGeneration(PaxosConfigTransactionImpl* self) {
|
||||
state std::vector<Future<ConfigTransactionGetGenerationReply>> getGenerationFutures;
|
||||
getGenerationFutures.reserve(self->ctis.size());
|
||||
for (auto const& cti : self->ctis) {
|
||||
getGenerationFutures.push_back(cti.getGeneration.getReply(ConfigTransactionGetGenerationRequest{}));
|
||||
}
|
||||
// FIXME: Must tolerate failures and disagreement
|
||||
wait(waitForAll(getGenerationFutures));
|
||||
return getGenerationFutures[0].get().generation;
|
||||
}
|
||||
|
||||
ACTOR static Future<Optional<Value>> get(PaxosConfigTransactionImpl* self, Key key) {
|
||||
if (!self->getGenerationFuture.isValid()) {
|
||||
self->getGenerationFuture = getGeneration(self);
|
||||
}
|
||||
state ConfigKey configKey = ConfigKey::decodeKey(key);
|
||||
ConfigGeneration generation = wait(self->getGenerationFuture);
|
||||
// TODO: Load balance
|
||||
ConfigTransactionGetReply reply =
|
||||
wait(self->ctis[0].get.getReply(ConfigTransactionGetRequest{ generation, configKey }));
|
||||
if (reply.value.present()) {
|
||||
return reply.value.get().toValue();
|
||||
} else {
|
||||
return Optional<Value>{};
|
||||
}
|
||||
}
|
||||
|
||||
public:
|
||||
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().liveVersion;
|
||||
} else {
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
Version getCommittedVersion() const { return committed ? getGenerationFuture.get().liveVersion : ::invalidVersion; }
|
||||
|
||||
int64_t getApproximateSize() const { return toCommit.expectedSize(); }
|
||||
|
||||
void set(KeyRef key, ValueRef value) { toCommit.set(key, value); }
|
||||
|
||||
void clear(KeyRef key) { toCommit.clear(key); }
|
||||
|
||||
Future<Optional<Value>> get(Key const& key) { return get(this, key); }
|
||||
|
||||
Future<Void> onError(Error const& e) {
|
||||
// TODO: Improve this:
|
||||
if (e.code() == error_code_transaction_too_old) {
|
||||
reset();
|
||||
return delay((1 << numRetries++) * 0.01 * deterministicRandom()->random01());
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
|
||||
void debugTransaction(UID dID) { this->dID = dID; }
|
||||
|
||||
void reset() {
|
||||
getGenerationFuture = Future<ConfigGeneration>{};
|
||||
toCommit = {};
|
||||
committed = false;
|
||||
}
|
||||
|
||||
void fullReset() {
|
||||
numRetries = 0;
|
||||
dID = {};
|
||||
reset();
|
||||
}
|
||||
|
||||
void checkDeferredError(Error const& deferredError) const {
|
||||
if (deferredError.code() != invalid_error_code) {
|
||||
throw deferredError;
|
||||
}
|
||||
if (cx.getPtr()) {
|
||||
cx->checkDeferredError();
|
||||
}
|
||||
}
|
||||
|
||||
PaxosConfigTransactionImpl(Database const& cx) : cx(cx) {
|
||||
auto coordinators = cx->getConnectionFile()->getConnectionString().coordinators();
|
||||
ctis.reserve(coordinators.size());
|
||||
for (const auto& coordinator : coordinators) {
|
||||
ctis.emplace_back(coordinator);
|
||||
}
|
||||
}
|
||||
|
||||
PaxosConfigTransactionImpl(std::vector<ConfigTransactionInterface> const& ctis) : ctis(ctis) {}
|
||||
};
|
||||
|
||||
Future<Version> PaxosConfigTransaction::getReadVersion() {
|
||||
// TODO: Implement
|
||||
return ::invalidVersion;
|
||||
return impl().getReadVersion();
|
||||
}
|
||||
|
||||
Optional<Version> PaxosConfigTransaction::getCachedReadVersion() const {
|
||||
// TODO: Implement
|
||||
return ::invalidVersion;
|
||||
return impl().getCachedReadVersion();
|
||||
}
|
||||
|
||||
Future<Optional<Value>> PaxosConfigTransaction::get(Key const& key, Snapshot snapshot) {
|
||||
// TODO: Implement
|
||||
return Optional<Value>{};
|
||||
Future<Optional<Value>> PaxosConfigTransaction::get(Key const& key, Snapshot) {
|
||||
return impl().get(key);
|
||||
}
|
||||
|
||||
Future<Standalone<RangeResultRef>> PaxosConfigTransaction::getRange(KeySelector const& begin,
|
||||
|
@ -59,13 +160,11 @@ Future<Standalone<RangeResultRef>> PaxosConfigTransaction::getRange(KeySelector
|
|||
}
|
||||
|
||||
void PaxosConfigTransaction::set(KeyRef const& key, ValueRef const& value) {
|
||||
// TODO: Implememnt
|
||||
ASSERT(false);
|
||||
return impl().set(key, value);
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::clear(KeyRef const& key) {
|
||||
// TODO: Implememnt
|
||||
ASSERT(false);
|
||||
return impl().clear(key);
|
||||
}
|
||||
|
||||
Future<Void> PaxosConfigTransaction::commit() {
|
||||
|
@ -75,61 +174,49 @@ Future<Void> PaxosConfigTransaction::commit() {
|
|||
}
|
||||
|
||||
Version PaxosConfigTransaction::getCommittedVersion() const {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
return ::invalidVersion;
|
||||
return impl().getCommittedVersion();
|
||||
}
|
||||
|
||||
int64_t PaxosConfigTransaction::getApproximateSize() const {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
return 0;
|
||||
return impl().getApproximateSize();
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::setOption(FDBTransactionOptions::Option option, Optional<StringRef> value) {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
// TODO: Support using this option to determine atomicity
|
||||
}
|
||||
|
||||
Future<Void> PaxosConfigTransaction::onError(Error const& e) {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
return Void();
|
||||
return impl().onError(e);
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::cancel() {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
// TODO: Implement someday
|
||||
throw client_invalid_operation();
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::reset() {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
impl().reset();
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::fullReset() {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
impl().fullReset();
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::debugTransaction(UID dID) {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
impl().debugTransaction(dID);
|
||||
}
|
||||
|
||||
void PaxosConfigTransaction::checkDeferredError() const {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
impl().checkDeferredError(deferredError);
|
||||
}
|
||||
|
||||
PaxosConfigTransaction::PaxosConfigTransaction() {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
}
|
||||
PaxosConfigTransaction::PaxosConfigTransaction(std::vector<ConfigTransactionInterface> const& ctis)
|
||||
: _impl(std::make_unique<PaxosConfigTransactionImpl>(ctis)) {}
|
||||
|
||||
PaxosConfigTransaction::PaxosConfigTransaction() = default;
|
||||
|
||||
PaxosConfigTransaction::~PaxosConfigTransaction() = default;
|
||||
|
||||
void PaxosConfigTransaction::setDatabase(Database const& cx) {
|
||||
// TODO: Implement
|
||||
ASSERT(false);
|
||||
_impl = std::make_unique<PaxosConfigTransactionImpl>(cx);
|
||||
}
|
||||
|
|
|
@ -33,6 +33,7 @@ class PaxosConfigTransaction final : public IConfigTransaction, public FastAlloc
|
|||
PaxosConfigTransactionImpl& impl() { return *_impl; }
|
||||
|
||||
public:
|
||||
PaxosConfigTransaction(std::vector<ConfigTransactionInterface> const&);
|
||||
PaxosConfigTransaction();
|
||||
~PaxosConfigTransaction();
|
||||
void setDatabase(Database const&) override;
|
||||
|
|
|
@ -124,25 +124,9 @@ public:
|
|||
|
||||
SimpleConfigTransactionImpl(ConfigTransactionInterface const& cti) : cti(cti) {}
|
||||
|
||||
void set(KeyRef key, ValueRef value) {
|
||||
if (key == configTransactionDescriptionKey) {
|
||||
toCommit.annotation.description = KeyRef(toCommit.arena, value);
|
||||
} else {
|
||||
ConfigKey configKey = ConfigKeyRef::decodeKey(key);
|
||||
auto knobValue = IKnobCollection::parseKnobValue(
|
||||
configKey.knobName.toString(), value.toString(), IKnobCollection::Type::TEST);
|
||||
toCommit.mutations.emplace_back_deep(toCommit.arena, configKey, knobValue.contents());
|
||||
}
|
||||
}
|
||||
void set(KeyRef key, ValueRef value) { toCommit.set(key, value); }
|
||||
|
||||
void clear(KeyRef key) {
|
||||
if (key == configTransactionDescriptionKey) {
|
||||
toCommit.annotation.description = ""_sr;
|
||||
} else {
|
||||
toCommit.mutations.emplace_back_deep(
|
||||
toCommit.arena, ConfigKeyRef::decodeKey(key), Optional<KnobValueRef>{});
|
||||
}
|
||||
}
|
||||
void clear(KeyRef key) { toCommit.clear(key); }
|
||||
|
||||
Future<Optional<Value>> get(KeyRef key) { return get(this, key); }
|
||||
|
||||
|
|
Loading…
Reference in New Issue