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

Update type of coordinators hash

Browse Source 
This fixes some serialization issues due to `BinaryReader` not being
able to deserialize types of size_t.
This commit is contained in:
Lukas Joswiak 2022-09-13 15:43:56 -07:00
parent 9fd22546ea
commit 8c50f98c00
8 changed files with 46 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
fdbclient/PaxosConfigTransaction.actor.cpp
View File

@ -35,7 +35,8 @@ class CommitQuorum {
Standalone<VectorRef<ConfigMutationRef>> mutations;
ConfigCommitAnnotation annotation;
ConfigTransactionCommitRequest getCommitRequest(ConfigGeneration generation, size_t coordinatorsHash) const {
ConfigTransactionCommitRequest getCommitRequest(ConfigGeneration generation,
CoordinatorsHash coordinatorsHash) const {
return ConfigTransactionCommitRequest(coordinatorsHash, generation, mutations, annotation);
}
@ -63,7 +64,7 @@ class CommitQuorum {
ACTOR static Future<Void> addRequestActor(CommitQuorum* self,
ConfigGeneration generation,
size_t coordinatorsHash,
CoordinatorsHash coordinatorsHash,
ConfigTransactionInterface cti) {
try {
if (cti.hostname.present()) {
@ -112,7 +113,7 @@ public:
}
void setTimestamp() { annotation.timestamp = now(); }
size_t expectedSize() const { return annotation.expectedSize() + mutations.expectedSize(); }
Future<Void> commit(ConfigGeneration generation, size_t coordinatorsHash) {
Future<Void> commit(ConfigGeneration generation, CoordinatorsHash coordinatorsHash) {
// Send commit message to all replicas, even those that did not return the used replica.
// This way, slow replicas are kept up date.
for (const auto& cti : ctis) {
@ -125,7 +126,7 @@ public:
class GetGenerationQuorum {
ActorCollection actors{ false };
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
std::vector<ConfigTransactionInterface> ctis;
std::map<ConfigGeneration, std::vector<ConfigTransactionInterface>> seenGenerations;
Promise<ConfigGeneration> result;
@ -241,7 +242,7 @@ class GetGenerationQuorum {
public:
GetGenerationQuorum() = default;
explicit GetGenerationQuorum(size_t coordinatorsHash,
explicit GetGenerationQuorum(CoordinatorsHash coordinatorsHash,
std::vector<ConfigTransactionInterface> const& ctis,
Future<Void> coordinatorsChangedFuture,
Optional<Version> const& lastSeenLiveVersion = {})
@ -267,7 +268,7 @@ public:
};
class PaxosConfigTransactionImpl {
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
std::vector<ConfigTransactionInterface> ctis;
GetGenerationQuorum getGenerationQuorum;
CommitQuorum commitQuorum;

2
fdbclient/include/fdbclient/ConfigKnobs.h
View File

@ -25,6 +25,8 @@
#include "fdbclient/FDBTypes.h"
typedef uint64_t CoordinatorsHash;
/*
* KnobValueRefs are stored in the configuration database, and in local configuration files. They are created from
* ParsedKnobValue objects, so it is assumed that the value type is correct for the corresponding knob name

20
fdbclient/include/fdbclient/ConfigTransactionInterface.h
View File

@ -65,12 +65,12 @@ struct ConfigTransactionGetGenerationReply {
struct ConfigTransactionGetGenerationRequest {
static constexpr FileIdentifier file_identifier = 138941;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
// A hint to catch up lagging nodes:
Optional<Version> lastSeenLiveVersion;
ReplyPromise<ConfigTransactionGetGenerationReply> reply;
ConfigTransactionGetGenerationRequest() = default;
explicit ConfigTransactionGetGenerationRequest(size_t coordinatorsHash,
explicit ConfigTransactionGetGenerationRequest(CoordinatorsHash coordinatorsHash,
Optional<Version> const& lastSeenLiveVersion)
: coordinatorsHash(coordinatorsHash), lastSeenLiveVersion(lastSeenLiveVersion) {}
@ -94,13 +94,13 @@ struct ConfigTransactionGetReply {
struct ConfigTransactionGetRequest {
static constexpr FileIdentifier file_identifier = 923040;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
ConfigGeneration generation;
ConfigKey key;
ReplyPromise<ConfigTransactionGetReply> reply;
ConfigTransactionGetRequest() = default;
explicit ConfigTransactionGetRequest(size_t coordinatorsHash, ConfigGeneration generation, ConfigKey key)
explicit ConfigTransactionGetRequest(CoordinatorsHash coordinatorsHash, ConfigGeneration generation, ConfigKey key)
: coordinatorsHash(coordinatorsHash), generation(generation), key(key) {}
template <class Ar>
@ -112,14 +112,14 @@ struct ConfigTransactionGetRequest {
struct ConfigTransactionCommitRequest {
static constexpr FileIdentifier file_identifier = 103841;
Arena arena;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
ConfigGeneration generation{ ::invalidVersion, ::invalidVersion };
VectorRef<ConfigMutationRef> mutations;
ConfigCommitAnnotationRef annotation;
ReplyPromise<Void> reply;
ConfigTransactionCommitRequest() = default;
explicit ConfigTransactionCommitRequest(size_t coordinatorsHash,
explicit ConfigTransactionCommitRequest(CoordinatorsHash coordinatorsHash,
ConfigGeneration generation,
VectorRef<ConfigMutationRef> mutations,
ConfigCommitAnnotationRef annotation)
@ -150,12 +150,12 @@ struct ConfigTransactionGetConfigClassesReply {
struct ConfigTransactionGetConfigClassesRequest {
static constexpr FileIdentifier file_identifier = 7163400;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
ConfigGeneration generation;
ReplyPromise<ConfigTransactionGetConfigClassesReply> reply;
ConfigTransactionGetConfigClassesRequest() = default;
explicit ConfigTransactionGetConfigClassesRequest(size_t coordinatorsHash, ConfigGeneration generation)
explicit ConfigTransactionGetConfigClassesRequest(CoordinatorsHash coordinatorsHash, ConfigGeneration generation)
: coordinatorsHash(coordinatorsHash), generation(generation) {}
template <class Ar>
@ -179,13 +179,13 @@ struct ConfigTransactionGetKnobsReply {
struct ConfigTransactionGetKnobsRequest {
static constexpr FileIdentifier file_identifier = 987410;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
ConfigGeneration generation;
Optional<Key> configClass;
ReplyPromise<ConfigTransactionGetKnobsReply> reply;
ConfigTransactionGetKnobsRequest() = default;
explicit ConfigTransactionGetKnobsRequest(size_t coordinatorsHash,
explicit ConfigTransactionGetKnobsRequest(CoordinatorsHash coordinatorsHash,
ConfigGeneration generation,
Optional<Key> configClass)
: coordinatorsHash(coordinatorsHash), generation(generation), configClass(configClass) {}

1
fdbclient/include/fdbclient/FDBTypes.h
View File

@ -41,6 +41,7 @@ typedef StringRef KeyRef;
typedef StringRef ValueRef;
typedef int64_t Generation;
typedef UID SpanID;
typedef uint64_t CoordinatorsHash;
enum {
tagLocalitySpecial = -1, // tag with this locality means it is invalidTag (id=0), txsTag (id=1), or cacheTag (id=2)

4
fdbserver/ConfigBroadcaster.actor.cpp
View File

@ -80,7 +80,7 @@ class ConfigBroadcasterImpl {
Version lastCompactedVersion;
Version largestLiveVersion;
Version mostRecentVersion;
size_t coordinatorsHash;
CoordinatorsHash coordinatorsHash;
std::unique_ptr<IConfigConsumer> consumer;
Future<Void> consumerFuture;
ActorCollection actors{ false };
@ -717,7 +717,7 @@ ACTOR static Future<Void> lockConfigNodesImpl(ServerCoordinators coordinators) {
return Void();
}
size_t coordinatorsHash = std::hash<std::string>()(coordinators.ccr->getConnectionString().toString());
CoordinatorsHash coordinatorsHash = std::hash<std::string>()(coordinators.ccr->getConnectionString().toString());
std::vector<Future<Void>> lockRequests;
lockRequests.reserve(coordinators.configServers.size());

42
fdbserver/ConfigNode.actor.cpp
View File

@ -135,11 +135,11 @@ class ConfigNodeImpl {
Counter getGenerationRequests;
Future<Void> logger;
ACTOR static Future<size_t> getCoordinatorsHash(ConfigNodeImpl* self) {
state size_t coordinatorsHash = 0;
ACTOR static Future<CoordinatorsHash> getCoordinatorsHash(ConfigNodeImpl* self) {
state CoordinatorsHash coordinatorsHash = 0;
Optional<Value> value = wait(self->kvStore->readValue(coordinatorsHashKey));
if (value.present()) {
coordinatorsHash = BinaryReader::fromStringRef<size_t>(value.get(), IncludeVersion());
coordinatorsHash = BinaryReader::fromStringRef<CoordinatorsHash>(value.get(), IncludeVersion());
} else {
self->kvStore->set(
KeyValueRef(coordinatorsHashKey, BinaryWriter::toValue(coordinatorsHash, IncludeVersion())));
@ -148,14 +148,12 @@ class ConfigNodeImpl {
return coordinatorsHash;
}
// The returned value is a hash of the nodes current idea of the
// coordinators.
ACTOR static Future<Optional<size_t>> getLocked(ConfigNodeImpl* self) {
ACTOR static Future<Optional<CoordinatorsHash>> getLocked(ConfigNodeImpl* self) {
Optional<Value> value = wait(self->kvStore->readValue(lockedKey));
if (!value.present()) {
return Optional<size_t>();
return Optional<CoordinatorsHash>();
}
return BinaryReader::fromStringRef<Optional<size_t>>(value.get(), IncludeVersion());
return BinaryReader::fromStringRef<Optional<CoordinatorsHash>>(value.get(), IncludeVersion());
}
ACTOR static Future<ConfigGeneration> getGeneration(ConfigNodeImpl* self) {
@ -254,7 +252,7 @@ class ConfigNodeImpl {
// New transactions increment the database's current live version. This effectively serves as a lock, providing
// serializability
ACTOR static Future<Void> getNewGeneration(ConfigNodeImpl* self, ConfigTransactionGetGenerationRequest req) {
state size_t coordinatorsHash = wait(getCoordinatorsHash(self));
state CoordinatorsHash coordinatorsHash = wait(getCoordinatorsHash(self));
if (req.coordinatorsHash != coordinatorsHash) {
req.reply.sendError(coordinators_changed());
return Void();
@ -273,13 +271,13 @@ class ConfigNodeImpl {
}
ACTOR static Future<Void> get(ConfigNodeImpl* self, ConfigTransactionGetRequest req) {
state Optional<size_t> locked = wait(getLocked(self));
state Optional<CoordinatorsHash> locked = wait(getLocked(self));
if (locked.present()) {
CODE_PROBE(true, "attempting to read from a locked ConfigNode");
req.reply.sendError(coordinators_changed());
return Void();
}
state size_t coordinatorsHash = wait(getCoordinatorsHash(self));
state CoordinatorsHash coordinatorsHash = wait(getCoordinatorsHash(self));
if (req.coordinatorsHash != coordinatorsHash) {
req.reply.sendError(coordinators_changed());
return Void();
@ -316,13 +314,13 @@ class ConfigNodeImpl {
// TODO: Currently it is possible that extra configuration classes may be returned, we
// may want to fix this to clean up the contract
ACTOR static Future<Void> getConfigClasses(ConfigNodeImpl* self, ConfigTransactionGetConfigClassesRequest req) {
state Optional<size_t> locked = wait(getLocked(self));
state Optional<CoordinatorsHash> locked = wait(getLocked(self));
if (locked.present()) {
CODE_PROBE(true, "attempting to read config classes from locked ConfigNode");
req.reply.sendError(coordinators_changed());
return Void();
}
state size_t coordinatorsHash = wait(getCoordinatorsHash(self));
state CoordinatorsHash coordinatorsHash = wait(getCoordinatorsHash(self));
if (req.coordinatorsHash != coordinatorsHash) {
req.reply.sendError(coordinators_changed());
return Void();
@ -360,13 +358,13 @@ class ConfigNodeImpl {
// Retrieve all knobs explicitly defined for the specified configuration class
ACTOR static Future<Void> getKnobs(ConfigNodeImpl* self, ConfigTransactionGetKnobsRequest req) {
state Optional<size_t> locked = wait(getLocked(self));
state Optional<CoordinatorsHash> locked = wait(getLocked(self));
if (locked.present()) {
CODE_PROBE(true, "attempting to read knobs from locked ConfigNode");
req.reply.sendError(coordinators_changed());
return Void();
}
state size_t coordinatorsHash = wait(getCoordinatorsHash(self));
state CoordinatorsHash coordinatorsHash = wait(getCoordinatorsHash(self));
if (req.coordinatorsHash != coordinatorsHash) {
req.reply.sendError(coordinators_changed());
return Void();
@ -453,13 +451,13 @@ class ConfigNodeImpl {
}
ACTOR static Future<Void> commit(ConfigNodeImpl* self, ConfigTransactionCommitRequest req) {
state Optional<size_t> locked = wait(getLocked(self));
state Optional<CoordinatorsHash> locked = wait(getLocked(self));
if (locked.present()) {
CODE_PROBE(true, "attempting to write to locked ConfigNode");
req.reply.sendError(coordinators_changed());
return Void();
}
state size_t coordinatorsHash = wait(getCoordinatorsHash(self));
state CoordinatorsHash coordinatorsHash = wait(getCoordinatorsHash(self));
if (req.coordinatorsHash != coordinatorsHash) {
req.reply.sendError(coordinators_changed());
return Void();
@ -675,7 +673,7 @@ class ConfigNodeImpl {
req.reply.send(ConfigBroadcastRegisteredReply{ isRegistered, generation.committedVersion });
}
when(state ConfigBroadcastReadyRequest readyReq = waitNext(cbi->ready.getFuture())) {
state Optional<size_t> locked = wait(getLocked(self));
state Optional<CoordinatorsHash> locked = wait(getLocked(self));
// New ConfigNodes with no previous state should always
// apply snapshots from the ConfigBroadcaster. Otherwise,
@ -715,8 +713,8 @@ class ConfigNodeImpl {
// Make sure freshly up to date ConfigNode isn't
// locked! This is possible if it was a coordinator in
// a previous generation.
self->kvStore->set(
KeyValueRef(lockedKey, BinaryWriter::toValue(Optional<size_t>(), IncludeVersion())));
self->kvStore->set(KeyValueRef(
lockedKey, BinaryWriter::toValue(Optional<CoordinatorsHash>(), IncludeVersion())));
}
self->kvStore->set(KeyValueRef(coordinatorsHashKey,
BinaryWriter::toValue(readyReq.coordinatorsHash, IncludeVersion())));
@ -768,13 +766,13 @@ class ConfigNodeImpl {
}
when(state ConfigFollowerLockRequest req = waitNext(cfi->lock.getFuture())) {
++self->lockRequests;
size_t coordinatorsHash = wait(getCoordinatorsHash(self));
CoordinatorsHash coordinatorsHash = wait(getCoordinatorsHash(self));
if (coordinatorsHash == 0 || coordinatorsHash == req.coordinatorsHash) {
TraceEvent("ConfigNodeLocking", self->id).log();
self->kvStore->set(KeyValueRef(registeredKey, BinaryWriter::toValue(false, IncludeVersion())));
self->kvStore->set(KeyValueRef(
lockedKey,
BinaryWriter::toValue(Optional<size_t>(req.coordinatorsHash), IncludeVersion())));
BinaryWriter::toValue(Optional<CoordinatorsHash>(req.coordinatorsHash), IncludeVersion())));
wait(self->kvStore->commit());
}
req.reply.send(Void());

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

@ -156,14 +156,14 @@ struct ConfigBroadcastReadyReply {
struct ConfigBroadcastReadyRequest {
static constexpr FileIdentifier file_identifier = 7402862;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
std::map<ConfigKey, KnobValue> snapshot;
Version snapshotVersion{ 0 };
Version liveVersion{ 0 };
ReplyPromise<ConfigBroadcastReadyReply> reply;
ConfigBroadcastReadyRequest() = default;
ConfigBroadcastReadyRequest(size_t coordinatorsHash,
ConfigBroadcastReadyRequest(CoordinatorsHash coordinatorsHash,
std::map<ConfigKey, KnobValue> const& snapshot,
Version snapshotVersion,
Version liveVersion)

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

@ -211,11 +211,11 @@ struct ConfigFollowerGetCommittedVersionRequest {
struct ConfigFollowerLockRequest {
static constexpr FileIdentifier file_identifier = 1867800;
size_t coordinatorsHash{ 0 };
CoordinatorsHash coordinatorsHash{ 0 };
ReplyPromise<Void> reply;
ConfigFollowerLockRequest() = default;
explicit ConfigFollowerLockRequest(size_t coordinatorsHash) : coordinatorsHash(coordinatorsHash) {}
explicit ConfigFollowerLockRequest(CoordinatorsHash coordinatorsHash) : coordinatorsHash(coordinatorsHash) {}
template <class Ar>
void serialize(Ar& ar) {