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

changed to using a vector of logSets instead of a duplicate set of logs for remote servers 

finished porting changes to the tlog
everything but peeking is finished in the TagPartitionedLogSystem
This commit is contained in:
Evan Tschannen 2017-07-09 14:46:16 -07:00
parent 0906250e78
commit 979ebcef6c
13 changed files with 1012 additions and 680 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
fdbserver/DBCoreState.h
View File

@ -34,137 +34,129 @@
// transaction log replicas and the write quorum that was used to commit to them. The read quorum required to
// ensure durability of locking and recovery is therefore tLogWriteAntiQuorum + 1.
struct OldTLogCoreData {
vector< UID > tLogs;
int32_t tLogWriteAntiQuorum;
int32_t tLogReplicationFactor;
struct CoreTLogSet {
std::vector< UID > tLogs;
int32_t tLogWriteAntiQuorum; // The write anti quorum previously used to write to tLogs, which might be different from the anti quorum suggested by the current configuration going forward!
int32_t tLogReplicationFactor; // The replication factor previously used to write to tLogs, which might be different from the current configuration
std::vector< LocalityData > tLogLocalities; // Stores the localities of the log servers
vector< UID > remoteTLogs;
int32_t remoteTLogReplicationFactor;
std::vector< LocalityData > remoteTLogLocalities;
IRepPolicyRef tLogPolicy;
Version epochEnd;
IRepPolicyRef tLogPolicy;
bool isLocal;
bool hasBest;
OldTLogCoreData() : tLogWriteAntiQuorum(0), tLogReplicationFactor(0), epochEnd(0), remoteTLogReplicationFactor(0) {}
CoreTLogSet() : tLogWriteAntiQuorum(0), tLogReplicationFactor(0), isLocal(true), hasBest(true) {}
bool operator == (OldTLogCoreData const& rhs) const { return tLogs == rhs.tLogs && tLogWriteAntiQuorum == rhs.tLogWriteAntiQuorum && tLogReplicationFactor == rhs.tLogReplicationFactor &&
epochEnd == rhs.epochEnd && remoteTLogs == rhs.remoteTLogs && remoteTLogReplicationFactor == rhs.remoteTLogReplicationFactor &&
((!tLogPolicy && !rhs.tLogPolicy) || (tLogPolicy && rhs.tLogPolicy && (tLogPolicy->info() == rhs.tLogPolicy->info()))); }
bool operator == (CoreTLogSet const& rhs) const {
return tLogs == rhs.tLogs && tLogWriteAntiQuorum == rhs.tLogWriteAntiQuorum && tLogReplicationFactor == rhs.tLogReplicationFactor && isLocal == rhs.isLocal &&
hasBest == rhs.hasBest && ((!tLogPolicy && !rhs.tLogPolicy) || (tLogPolicy && rhs.tLogPolicy && (tLogPolicy->info() == rhs.tLogPolicy->info())));
}
template <class Archive>
void serialize(Archive& ar) {
ar & tLogs & tLogWriteAntiQuorum & tLogReplicationFactor & tLogPolicy & tLogLocalities & isLocal & hasBest;
}
};
struct OldTLogCoreData {
std::vector<CoreTLogSet> tLogs;
Version epochEnd;
OldTLogCoreData() : epochEnd(0) {}
bool operator == (OldTLogCoreData const& rhs) const {
return tLogs == rhs.tLogs && epochEnd == rhs.epochEnd;
}
template <class Archive>
void serialize(Archive& ar) {
ar & tLogs & tLogWriteAntiQuorum & tLogReplicationFactor & tLogPolicy & epochEnd & tLogLocalities;
if( ar.protocolVersion() >= 0x0FDB00A560010001LL) {
ar & remoteTLogs & remoteTLogReplicationFactor & remoteTLogLocalities;
ar & tLogs & epochEnd;
}
else if(ar.isDeserializing) {
remoteTLogReplicationFactor = 0;
tLogs.push_back(CoreTLogSet());
ar & tLogs[0].tLogs & tLogs[0].tLogWriteAntiQuorum & tLogs[0].tLogReplicationFactor & tLogs[0].tLogPolicy & epochEnd & tLogs[0].tLogLocalities;
}
}
};
struct DBCoreState {
vector< UID > tLogs;
int32_t tLogWriteAntiQuorum; // The write anti quorum previously used to write to tLogs, which might be different from the anti quorum suggested by the current configuration going forward!
int32_t tLogReplicationFactor; // The replication factor previously used to write to tLogs, which might be different from the current configuration
IRepPolicyRef tLogPolicy;
std::vector<CoreTLogSet> tLogs;
std::vector<OldTLogCoreData> oldTLogData;
DBRecoveryCount recoveryCount; // Increases with sequential successful recoveries.
int logSystemType;
std::vector< LocalityData > tLogLocalities; // Stores the localities of the log servers
std::vector<OldTLogCoreData> oldTLogData;
DBCoreState() : recoveryCount(0), logSystemType(0) {}
vector< UID > remoteTLogs;
int32_t remoteTLogReplicationFactor;
std::vector< LocalityData > remoteTLogLocalities;
bool remoteTLogsRecovered;
DBCoreState() : recoveryCount(0), tLogWriteAntiQuorum(0), tLogReplicationFactor(0), logSystemType(0), remoteTLogReplicationFactor(0), remoteTLogsRecovered(false) {}
//FIXME: should this include old remote logs
vector<UID> getPriorCommittedLogServers() {
vector<UID> priorCommittedLogServers;
for(int i = 0; i < oldTLogData.size(); i++) {
for(auto it : oldTLogData[i].tLogs) {
priorCommittedLogServers.push_back(it);
for(auto& it : oldTLogData[i].tLogs) {
for(auto& log : it.tLogs) {
priorCommittedLogServers.push_back(log);
}
}
}
return priorCommittedLogServers;
}
bool isEqual(DBCoreState const& r) const {
if (logSystemType != r.logSystemType || recoveryCount != r.recoveryCount || tLogWriteAntiQuorum != r.tLogWriteAntiQuorum || tLogReplicationFactor != r.tLogReplicationFactor || oldTLogData.size() != r.oldTLogData.size()
|| tLogLocalities != r.tLogLocalities || tLogs != r.tLogs || remoteTLogs != r.remoteTLogs || remoteTLogReplicationFactor != r.remoteTLogReplicationFactor || remoteTLogLocalities != r.remoteTLogLocalities || remoteTLogsRecovered != r.remoteTLogsRecovered)
return false;
for(int i = 0; i < oldTLogData.size(); i++ ) {
if (oldTLogData[i] != r.oldTLogData[i])
return false;
}
return true;
return logSystemType == r.logSystemType && recoveryCount == r.recoveryCount && tLogs == r.tLogs && oldTLogData == r.oldTLogData;
}
bool operator == ( const DBCoreState& rhs ) const { return isEqual(rhs); }
template <class Archive>
void serialize(Archive& ar) {
UID functionId = g_nondeterministic_random->randomUniqueID();
ASSERT( ar.protocolVersion() >= 0x0FDB00A320050001LL );
ar & tLogs & tLogWriteAntiQuorum & recoveryCount & tLogReplicationFactor & logSystemType;
if( ar.protocolVersion() >= 0x0FDB00A460010001LL) {
uint64_t tLocalitySize = (uint64_t)tLogLocalities.size();
ar & oldTLogData & tLogPolicy & tLocalitySize;
if (ar.isDeserializing) {
tLogLocalities.reserve(tLocalitySize);
for (size_t i = 0; i < tLocalitySize; i++) {
LocalityData locality;
ar & locality;
tLogLocalities.push_back(locality);
}
}
else {
for (auto& locality : tLogLocalities) {
ar & locality;
}
}
}
else if(ar.isDeserializing) {
oldTLogData.clear();
oldTLogData.push_back(OldTLogCoreData());
ar & oldTLogData[0].tLogs & oldTLogData[0].epochEnd & oldTLogData[0].tLogWriteAntiQuorum & oldTLogData[0].tLogReplicationFactor;
tLogPolicy = IRepPolicyRef(new PolicyAcross(tLogReplicationFactor, "zoneid", IRepPolicyRef(new PolicyOne())));
if(!oldTLogData[0].tLogs.size()) {
oldTLogData.pop_back();
}
else {
for(int i = 0; i < oldTLogData.size(); i++ ) {
oldTLogData[i].tLogPolicy = IRepPolicyRef(new PolicyAcross(oldTLogData[i].tLogReplicationFactor, "zoneid", IRepPolicyRef(new PolicyOne())));
if (oldTLogData[i].tLogs.size())
{
oldTLogData[i].tLogLocalities.reserve(oldTLogData[i].tLogs.size());
for (auto& tLog : oldTLogData[i].tLogs) {
LocalityData locality;
locality.set(LocalityData::keyZoneId, g_random->randomUniqueID().toString());
locality.set(LocalityData::keyDataHallId, LiteralStringRef("0"));
oldTLogData[i].tLogLocalities.push_back(locality);
}
if( ar.protocolVersion() >= 0x0FDB00A560010001LL) {
ar & tLogs & oldTLogData & recoveryCount & logSystemType;
} else if(ar.isDeserializing) {
tLogs.push_back(CoreTLogSet());
ar & tLogs[0].tLogs & tLogs[0].tLogWriteAntiQuorum & recoveryCount & tLogs[0].tLogReplicationFactor & logSystemType;
if( ar.protocolVersion() >= 0x0FDB00A460010001LL) {
uint64_t tLocalitySize = (uint64_t)tLogLocalities.size();
ar & oldTLogData & tLogs[0].tLogPolicy & tLocalitySize;
if (ar.isDeserializing) {
tLogs[0].tLogLocalities.reserve(tLocalitySize);
for (size_t i = 0; i < tLocalitySize; i++) {
LocalityData locality;
ar & locality;
tLogs[0].tLogLocalities.push_back(locality);
}
}
}
tLogLocalities.reserve(tLogs.size());
for (auto& tLog : tLogs) {
LocalityData locality;
locality.set(LocalityData::keyZoneId, g_random->randomUniqueID().toString());
locality.set(LocalityData::keyDataHallId, LiteralStringRef("0"));
tLogLocalities.push_back(locality);
else {
oldTLogData.clear();
oldTLogData.push_back(OldTLogCoreData());
oldTLogData.tLogs.push_back(CoreTLogSet());
ar & oldTLogData[0].tLogs[0].tLogs & oldTLogData[0].tLogs[0].epochEnd & oldTLogData[0].tLogs[0].tLogWriteAntiQuorum & oldTLogData[0].tLogs[0].tLogReplicationFactor;
tLogs[0].tLogPolicy = IRepPolicyRef(new PolicyAcross(tLogs[0].tLogReplicationFactor, "zoneid", IRepPolicyRef(new PolicyOne())));
if(!oldTLogData[0].tLogs[0].tLogs.size()) {
oldTLogData.pop_back();
}
else {
for(int i = 0; i < oldTLogData.tLogs[0].size(); i++ ) {
oldTLogData[i].tLogs[0].tLogPolicy = IRepPolicyRef(new PolicyAcross(oldTLogData[i].tLogs[0].tLogReplicationFactor, "zoneid", IRepPolicyRef(new PolicyOne())));
if (oldTLogData[i].tLogs[0].tLogs.size())
{
oldTLogData[i].tLogs[0].tLogLocalities.reserve(oldTLogData[i].tLogs[0].tLogs.size());
for (auto& tLog : oldTLogData[i].tLogs[0].tLogs) {
LocalityData locality;
locality.set(LocalityData::keyZoneId, g_random->randomUniqueID().toString());
locality.set(LocalityData::keyDataHallId, LiteralStringRef("0"));
oldTLogData[i].tLogs[0].tLogLocalities.push_back(locality);
}
}
}
}
tLogs[0].tLogLocalities.reserve(tLogs[0].tLogs.size());
for (auto& tLog : tLogs[0].tLogs) {
LocalityData locality;
locality.set(LocalityData::keyZoneId, g_random->randomUniqueID().toString());
locality.set(LocalityData::keyDataHallId, LiteralStringRef("0"));
tLogs[0].tLogLocalities.push_back(locality);
}
}
}
if( ar.protocolVersion() >= 0x0FDB00A560010001LL) {
ar & remoteTLogs & remoteTLogReplicationFactor & remoteTLogLocalities & remoteTLogsRecovered;
}
else if(ar.isDeserializing) {
remoteTLogReplicationFactor = 0;
remoteTLogsRecovered = false;
}
TraceEvent("CoreStateSerialize").detail("AntiQuorum", tLogWriteAntiQuorum)
.detail("logSystemType", logSystemType).detail("recoveryCount", recoveryCount)
@ -174,9 +166,6 @@ struct DBCoreState {
.detail("oldTLogData", oldTLogData.size())
.detail("deserializing", ar.isDeserializing);
}
std::string toString() const { return format("type: %d anti: %d replication: %d policy: %s tLogs: %s oldGenerations: %d tlocalities: %s",
logSystemType, tLogWriteAntiQuorum, tLogReplicationFactor, (tLogPolicy ? tLogPolicy->info().c_str() : "[unset]"), describe(tLogs).c_str(), oldTLogData.size(), describe(tLogLocalities).c_str()); }
};
#endif

6
fdbserver/DatabaseConfiguration.cpp
View File

@ -38,6 +38,7 @@ void DatabaseConfiguration::resetInternal() {
tLogPolicy = IRepPolicyRef();
remoteTLogCount = 0;
remoteTLogReplicationFactor = 0;
remoteTLogPolicy = IRepPolicyRef();
}
void parse( int* i, ValueRef const& v ) {
@ -53,6 +54,7 @@ void parseReplicationPolicy(IRepPolicyRef* policy, ValueRef const& v) {
void DatabaseConfiguration::setDefaultReplicationPolicy() {
storagePolicy = IRepPolicyRef(new PolicyAcross(storageTeamSize, "zoneid", IRepPolicyRef(new PolicyOne())));
tLogPolicy = IRepPolicyRef(new PolicyAcross(tLogReplicationFactor, "zoneid", IRepPolicyRef(new PolicyOne())));
remoteTLogPolicy = IRepPolicyRef(new PolicyAcross(remoteTLogReplicationFactor, "zoneid", IRepPolicyRef(new PolicyOne())));
}
bool DatabaseConfiguration::isValid() const {
@ -73,7 +75,8 @@ bool DatabaseConfiguration::isValid() const {
storagePolicy &&
tLogPolicy &&
remoteTLogCount >= 0 &&
remoteTLogReplicationFactor >=0;
remoteTLogReplicationFactor >=0 &&
remoteTLogPolicy;
}
std::map<std::string, std::string> DatabaseConfiguration::toMap() const {
@ -143,6 +146,7 @@ bool DatabaseConfiguration::setInternal(KeyRef key, ValueRef value) {
else if (ck == LiteralStringRef("log_replication_policy")) parseReplicationPolicy(&tLogPolicy, value);
else if (ck == LiteralStringRef("remote_logs")) parse(&remoteTLogCount, value);
else if (ck == LiteralStringRef("remote_log_replication")) parse(&remoteTLogReplicationFactor, value);
else if (ck == LiteralStringRef("remote_replication_policy")) parseReplicationPolicy(&remoteTLogPolicy, value);
else return false;
return true; // All of the above options currently require recovery to take effect
}

3
fdbserver/DatabaseConfiguration.h
View File

@ -63,10 +63,11 @@ struct DatabaseConfiguration {
int32_t autoDesiredTLogCount;
int32_t tLogWriteAntiQuorum;
int32_t tLogReplicationFactor;
int32_t remoteTLogReplicationFactor;
KeyValueStoreType tLogDataStoreType;
IRepPolicyRef tLogPolicy;
int32_t remoteTLogCount;
int32_t remoteTLogReplicationFactor;
IRepPolicyRef remoteTLogPolicy;
// Storage servers
int32_t durableStorageQuorum;

32
fdbserver/LogRouter.actor.cpp
View File

@ -78,8 +78,9 @@ struct LogRouterData {
Deque<std::pair<Version, Standalone<VectorRef<uint8_t>>>> messageBlocks;
Map< Tag, TagData > tag_data;
Tag routerTag;
int logSet;
LogRouterData(UID dbgid, Tag routerTag) : dbgid(dbgid), routerTag(routerTag), logSystem(new AsyncVar<Reference<ILogSystem>>()) {}
LogRouterData(UID dbgid, Tag routerTag, int logSet) : dbgid(dbgid), routerTag(routerTag), logSet(logSet), logSystem(new AsyncVar<Reference<ILogSystem>>()) {}
};
void commitMessages( LogRouterData* self, Version version, Arena arena, StringRef messages, VectorRef< TagMessagesRef > tags) {
@ -160,7 +161,7 @@ ACTOR Future<Void> pullAsyncData( LogRouterData *self, Tag tag ) {
state Version tagAt = self->version.get()+1;
state Version tagPopped = 0;
state Version lastVer = 0;
state std::vector<Tag> tags;
state std::vector<int> tags;
loop {
loop {
@ -214,18 +215,13 @@ ACTOR Future<Void> pullAsyncData( LogRouterData *self, Tag tag ) {
StringRef msg = r->getMessage();
auto originalTags = r->getTags();
tags.clear();
for(auto tag : originalTags) {
if(tag >= 0 && tag < SERVER_KNOBS->MAX_TAG) {
tags.push_back(self->logSystem->get()->getRemoteLogTag(tag));
}
}
//FIXME: do we add txsTags?
self->logSystem->get()->addRemoteTags(tags);
self->logSystem->get()->addRemoteTags(self->logSet, originalTags, tags);
for(auto tag : tags) {
auto it = tag_offsets.find(tag);
if (it == tag_offsets.end()) {
it = tag_offsets.insert(mapPair( tag, TagMessagesRef() ));
it = tag_offsets.insert(mapPair( Tag(tag), TagMessagesRef() ));
it->value.tag = it->key;
}
it->value.messageOffsets.push_back( arena, wr.getLength() );
@ -353,9 +349,10 @@ ACTOR Future<Void> logRouterPop( LogRouterData* self, TLogPopRequest req ) {
ACTOR Future<Void> logRouterCore(
TLogInterface interf,
Tag tag,
int logSet,
Reference<AsyncVar<ServerDBInfo>> db)
{
state LogRouterData logRouterData(interf.id(), tag);
state LogRouterData logRouterData(interf.id(), tag, logSet);
state PromiseStream<Future<Void>> addActor;
state Future<Void> error = actorCollection( addActor.getFuture() );
state Future<Void> dbInfoChange = Void();
@ -365,7 +362,8 @@ ACTOR Future<Void> logRouterCore(
loop choose {
when( Void _ = wait( dbInfoChange ) ) {
dbInfoChange = db->onChange();
if( db->get().recoveryState >= RecoveryState::FULLY_RECOVERED && std::count( db->get().logSystemConfig.logRouters.begin(), db->get().logSystemConfig.logRouters.end(), interf.id() ) ) {
if( db->get().recoveryState >= RecoveryState::FULLY_RECOVERED && logSet < db->get().logSystemConfig.tLogs.size() &&
std::count( db->get().logSystemConfig.tLogs[logSet].logRouters.begin(), db->get().logSystemConfig.tLogs[logSet].logRouters.end(), interf.id() ) ) {
logRouterData.logSystem->set(ILogSystem::fromServerDBInfo( logRouterData.dbgid, db->get() ));
} else {
logRouterData.logSystem->set(Reference<ILogSystem>());
@ -381,10 +379,12 @@ ACTOR Future<Void> logRouterCore(
}
}
ACTOR Future<Void> checkRemoved(Reference<AsyncVar<ServerDBInfo>> db, uint64_t recoveryCount, TLogInterface myInterface) {
ACTOR Future<Void> checkRemoved(Reference<AsyncVar<ServerDBInfo>> db, uint64_t recoveryCount, TLogInterface myInterface, int logSet) {
loop{
if (db->get().recoveryCount >= recoveryCount && !std::count(db->get().logSystemConfig.logRouters.begin(), db->get().logSystemConfig.logRouters.end(), myInterface.id()))
throw worker_removed();
if (db->get().recoveryCount >= recoveryCount && logSet < db->get().logSystemConfig.tLogs.size() &&
!std::count(db->get().logSystemConfig.tLogs[logSet].logRouters.begin(), db->get().logSystemConfig.tLogs[logSet].logRouters.end(), myInterface.id())) {
throw worker_removed();
}
Void _ = wait(db->onChange());
}
}
@ -395,10 +395,10 @@ ACTOR Future<Void> logRouter(
Reference<AsyncVar<ServerDBInfo>> db)
{
try {
state Future<Void> core = logRouterCore(interf, req.routerTag, db);
state Future<Void> core = logRouterCore(interf, req.routerTag, req.logSet, db);
loop choose{
when(Void _ = wait(core)) { return Void(); }
when(Void _ = wait(checkRemoved(db, req.recoveryCount, interf))) {}
when(Void _ = wait(checkRemoved(db, req.recoveryCount, interf, req.logSet))) {}
}
}
catch (Error& e) {

9
fdbserver/LogSystem.h
View File

@ -339,16 +339,9 @@ struct ILogSystem {
virtual Future<Void> onLogSystemConfigChange() = 0;
// Returns when the log system configuration has changed due to a tlog rejoin.
virtual int getLogServerCount() = 0;
// Used by LogPushData; returns the number of log servers
virtual void getPushLocations( std::vector<Tag> const& tags, vector<int>& locations ) = 0;
virtual void addRemoteTags( std::vector<Tag>& tags ) = 0;
virtual void getRemotePushLocations( std::vector<Tag> const& tags, vector<int>& locations ) = 0;
virtual Tag getRemoteLogTag( Tag serverTag ) = 0;
virtual void addRemoteTags( int logSet, std::vector<Tag> originalTags, std::vector<int>& tags ) = 0;
virtual Tag getRandomRouterTag() = 0;

225
fdbserver/LogSystemConfig.h
View File

@ -55,197 +55,150 @@ protected:
Optional<Interface> iface;
};
struct OldTLogConf {
vector<OptionalInterface<TLogInterface>> tLogs;
struct TLogSet {
std::vector<OptionalInterface<TLogInterface>> tLogs;
std::vector<TLogInterface> logRouters;
int32_t tLogWriteAntiQuorum, tLogReplicationFactor;
std::vector< LocalityData > tLogLocalities; // Stores the localities of the log servers
vector<OptionalInterface<TLogInterface>> remoteTLogs;
std::vector< LocalityData > remoteTLogLocalities;
int32_t remoteTLogReplicationFactor;
IRepPolicyRef tLogPolicy;
Version epochEnd;
bool isLocal;
bool hasBest;
OldTLogConf() : tLogWriteAntiQuorum(0), tLogReplicationFactor(0), remoteTLogReplicationFactor(0), epochEnd(0) {}
TLogSet() : tLogWriteAntiQuorum(0), tLogReplicationFactor(0), isLocal(true), hasBest(true) {}
std::string toString() const {
return format("anti: %d replication: %d remote: %d end: %d tLogs: %s remoteTLogs: %s tlocalities: %s remoteLocalities: %s",
tLogWriteAntiQuorum, tLogReplicationFactor, remoteTLogReplicationFactor, epochEnd, describe(tLogs).c_str(), describe(remoteTLogs).c_str(), describe(tLogLocalities).c_str(), describe(remoteTLogLocalities).c_str()); }
return format("anti: %d replication: %d local: %d best: %d routers: %d tLogs: %s", tLogWriteAntiQuorum, tLogReplicationFactor, isLocal, hasBest, logRouters.size(), describe(tLogs).c_str());
}
bool operator == ( const OldTLogConf& rhs ) const {
bool bIsEqual = true;
if (tLogWriteAntiQuorum != rhs.tLogWriteAntiQuorum) {
bIsEqual = false;
bool operator == ( const TLogSet& rhs ) const {
if (tLogWriteAntiQuorum != rhs.tLogWriteAntiQuorum || tLogReplicationFactor != rhs.tLogReplicationFactor || isLocal != rhs.isLocal || hasBest != rhs.hasBest || tLogs.size() != rhs.tLogs.size()) {
return false;
}
else if (tLogReplicationFactor != rhs.tLogReplicationFactor) {
bIsEqual = false;
if ((tLogPolicy && !rhs.tLogPolicy) || (!tLogPolicy && rhs.tLogPolicy) || (tLogPolicy && (tLogPolicy->info() != rhs.tLogPolicy->info()))) {
return false;
}
else if (remoteTLogReplicationFactor != rhs.remoteTLogReplicationFactor) {
bIsEqual = false;
}
else if (tLogs.size() != rhs.tLogs.size()) {
bIsEqual = false;
}
else if (remoteTLogs.size() != rhs.remoteTLogs.size()) {
bIsEqual = false;
}
if (bIsEqual) {
for(int j = 0; j < tLogs.size(); j++ ) {
if (tLogs[j].id() != rhs.tLogs[j].id()) {
bIsEqual = false;
break;
}
else if (tLogs[j].present() != rhs.tLogs[j].present()) {
bIsEqual = false;
break;
}
else if (tLogs[j].present() && tLogs[j].interf().commit.getEndpoint().token != rhs.tLogs[j].interf().commit.getEndpoint().token ) {
bIsEqual = false;
break;
}
}
for(int j = 0; j < remoteTLogs.size(); j++ ) {
if (remoteTLogs[j].id() != rhs.remoteTLogs[j].id()) {
bIsEqual = false;
break;
}
else if (remoteTLogs[j].present() != rhs.remoteTLogs[j].present()) {
bIsEqual = false;
break;
}
else if (remoteTLogs[j].present() && remoteTLogs[j].interf().commit.getEndpoint().token != rhs.remoteTLogs[j].interf().commit.getEndpoint().token ) {
bIsEqual = false;
break;
}
for(int j = 0; j < tLogs.size(); j++ ) {
if (tLogs[j].id() != rhs.tLogs[j].id() || tLogs[j].present() != rhs.tLogs[j].present() || ( tLogs[j].present() && tLogs[j].interf().commit.getEndpoint().token != rhs.tLogs[j].interf().commit.getEndpoint().token ) ) {
return false;
}
}
return bIsEqual;
return true;
}
bool isEqualIds(TLogSet const& r) const {
if (tLogWriteAntiQuorum != r.tLogWriteAntiQuorum || tLogReplicationFactor != r.tLogReplicationFactor || isLocal != r.isLocal || hasBest != r.hasBest || tLogs.size() != r.tLogs.size()) {
return false;
}
if ((tLogPolicy && !r.tLogPolicy) || (!tLogPolicy && r.tLogPolicy) || (tLogPolicy && (tLogPolicy->info() != r.tLogPolicy->info()))) {
return false;
}
for(int i = 0; i < tLogs.size(); i++) {
if( tLogs[i].id() != r.tLogs[i].id() ) {
return false;
}
}
return true;
}
template <class Ar>
void serialize( Ar& ar ) {
ar & tLogs & tLogWriteAntiQuorum & tLogReplicationFactor & tLogPolicy & tLogLocalities & epochEnd & remoteTLogs & remoteTLogLocalities & remoteTLogReplicationFactor;
ar & tLogs & logRouters & tLogWriteAntiQuorum & tLogReplicationFactor & tLogPolicy & tLogLocalities & isLocal & hasBest;
}
};
struct OldTLogConf {
std::vector<TLogSet> tLogs;
Version epochEnd;
OldTLogConf() : epochEnd(0) {}
std::string toString() const {
return format("end: %d %s", epochEnd, describe(tLogs).c_str());
}
bool operator == ( const OldTLogConf& rhs ) const {
return tLogs == rhs.tLogs && epochEnd == rhs.epochEnd;
}
bool isEqualIds(OldTLogConf const& r) const {
if(tLogs.size() != r.tLogs.size()) {
return false;
}
for(int i = 0; i < tLogs.size(); i++) {
if(!tLogs[i].isEqualIds(r.tLogs[i])) {
return false;
}
}
return true;
}
template <class Ar>
void serialize( Ar& ar ) {
ar & tLogs & epochEnd;
}
};
struct LogSystemConfig {
int logSystemType;
std::vector<OptionalInterface<TLogInterface>> tLogs;
std::vector< LocalityData > tLogLocalities;
std::vector<OptionalInterface<TLogInterface>> remoteTLogs;
std::vector< LocalityData > remoteTLogLocalities;
std::vector<OptionalInterface<TLogInterface>> logRouters;
std::vector<TLogSet> tLogs;
std::vector<OldTLogConf> oldTLogs;
int32_t tLogWriteAntiQuorum, tLogReplicationFactor;
int32_t remoteTLogReplicationFactor;
bool remoteTLogsRecovered;
IRepPolicyRef tLogPolicy;
//LogEpoch epoch;
LogSystemConfig() : tLogWriteAntiQuorum(0), tLogReplicationFactor(0), remoteTLogReplicationFactor(0), logSystemType(0) {}
LogSystemConfig() : logSystemType(0) {}
std::string toString() const { return format("type: %d anti: %d replication: %d remote: %d, recovered: %d tLogs: %s oldGenerations: %d logRouters: %s remoteTLogs: %s tlocalities: %s remoteLocalities: %s",
logSystemType, tLogWriteAntiQuorum, tLogReplicationFactor, remoteTLogReplicationFactor, remoteTLogsRecovered, describe(tLogs).c_str(), oldTLogs.size(), describe(logRouters).c_str(), describe(remoteTLogs).c_str(), describe(tLogLocalities).c_str(), describe(remoteTLogLocalities).c_str()); }
std::string toString() const {
return format("type: %d oldGenerations: %d %s", logSystemType, oldTLogs.size(), describe(tLogs).c_str());
}
std::vector<TLogInterface> allPresentLogs() const {
std::vector<TLogInterface> results;
for( int i = 0; i < tLogs.size(); i++ )
if( tLogs[i].present() )
results.push_back(tLogs[i].interf());
for( int i = 0; i < remoteTLogs.size(); i++ )
if( remoteTLogs[i].present() )
results.push_back(remoteTLogs[i].interf());
for( int i = 0; i < tLogs.size(); i++ ) {
for( int j = 0; j < tLogs[i].tLogs.size(); j++ ) {
if( tLogs[i].tLogs[j].present() ) {
results.push_back(tLogs[i].tLogs[j].interf());
}
}
}
return results;
}
bool operator == ( const LogSystemConfig& rhs ) const { return isEqual(rhs); }
bool isEqual(LogSystemConfig const& r) const {
if (logSystemType != r.logSystemType || tLogWriteAntiQuorum != r.tLogWriteAntiQuorum || tLogReplicationFactor != r.tLogReplicationFactor || tLogs.size() != r.tLogs.size() || oldTLogs.size() != r.oldTLogs.size()
|| remoteTLogs.size() != r.remoteTLogs.size() || logRouters.size() != r.logRouters.size() || remoteTLogReplicationFactor != r.remoteTLogReplicationFactor || remoteTLogsRecovered != r.remoteTLogsRecovered )
return false;
else if ((tLogPolicy && !r.tLogPolicy) || (!tLogPolicy && r.tLogPolicy) || (tLogPolicy && (tLogPolicy->info() != r.tLogPolicy->info())))
return false;
for(int i = 0; i < tLogs.size(); i++ ) {
if( tLogs[i].id() != r.tLogs[i].id() || tLogs[i].present() != r.tLogs[i].present() )
return false;
if( tLogs[i].present() && tLogs[i].interf().commit.getEndpoint().token != r.tLogs[i].interf().commit.getEndpoint().token )
return false;
}
for(int i = 0; i < remoteTLogs.size(); i++ ) {
if( remoteTLogs[i].id() != r.remoteTLogs[i].id() || remoteTLogs[i].present() != r.remoteTLogs[i].present() )
return false;
if( remoteTLogs[i].present() && remoteTLogs[i].interf().commit.getEndpoint().token != r.remoteTLogs[i].interf().commit.getEndpoint().token )
return false;
}
for(int i = 0; i < logRouters.size(); i++ ) {
if( logRouters[i].id() != r.logRouters[i].id() || logRouters[i].present() != r.logRouters[i].present() )
return false;
if( logRouters[i].present() && logRouters[i].interf().commit.getEndpoint().token != r.logRouters[i].interf().commit.getEndpoint().token )
return false;
}
for(int i = 0; i < oldTLogs.size(); i++ ) {
if (oldTLogs[i] != r.oldTLogs[i])
return false;
}
return true;
return logSystemType == r.logSystemType && tLogs == r.tLogs && oldTLogs == r.oldTLogs;
}
bool isEqualIds(LogSystemConfig const& r) const {
if(logSystemType!=r.logSystemType || tLogWriteAntiQuorum!=r.tLogWriteAntiQuorum || tLogReplicationFactor!=r.tLogReplicationFactor || tLogs.size() != r.tLogs.size() || oldTLogs.size() != r.oldTLogs.size()
|| remoteTLogs.size() != r.remoteTLogs.size() || logRouters.size() != r.logRouters.size() || remoteTLogReplicationFactor != r.remoteTLogReplicationFactor || remoteTLogsRecovered != r.remoteTLogsRecovered )
return false;
else if ((tLogPolicy && !r.tLogPolicy) || (!tLogPolicy && r.tLogPolicy) || (tLogPolicy && (tLogPolicy->info() != r.tLogPolicy->info()))) {
if( logSystemType!=r.logSystemType || tLogs.size() != r.tLogs.size() || oldTLogs.size() != r.oldTLogs.size() ) {
return false;
}
for(int i = 0; i < tLogs.size(); i++ ) {
if( tLogs[i].id() != r.tLogs[i].id() )
if( !tLogs[i].isEqualIds(r.tLogs[i]) ) {
return false;
}
for(int i = 0; i < remoteTLogs.size(); i++ ) {
if( remoteTLogs[i].id() != r.remoteTLogs[i].id() )
return false;
}
for(int i = 0; i < logRouters.size(); i++ ) {
if( logRouters[i].id() != r.logRouters[i].id() )
return false;
}
for(int i = 0; i < oldTLogs.size(); i++ ) {
if (oldTLogs[i].tLogWriteAntiQuorum != r.oldTLogs[i].tLogWriteAntiQuorum || oldTLogs[i].tLogReplicationFactor != r.oldTLogs[i].tLogReplicationFactor || oldTLogs[i].tLogs.size() != r.oldTLogs[i].tLogs.size()
|| oldTLogs[i].remoteTLogs.size() != r.oldTLogs[i].remoteTLogs.size() || oldTLogs[i].remoteTLogReplicationFactor != r.oldTLogs[i].remoteTLogReplicationFactor )
return false;
for(int j = 0; j < oldTLogs[i].tLogs.size(); j++ ) {
if( oldTLogs[i].tLogs[j].id() != r.oldTLogs[i].tLogs[j].id() )
return false;
}
for(int j = 0; j < oldTLogs[i].remoteTLogs.size(); j++ ) {
if( oldTLogs[i].remoteTLogs[j].id() != r.oldTLogs[i].remoteTLogs[j].id() )
return false;
}
for(int i = 0; i < oldTLogs.size(); i++ ) {
if( !oldTLogs[i].isEqualIds(r.oldTLogs[i]) ) {
return false;
}
}
return true;
}
bool isNextGenerationOf(LogSystemConfig const& r) const {
if( !oldTLogs.size() || oldTLogs[0].tLogWriteAntiQuorum!=r.tLogWriteAntiQuorum || oldTLogs[0].tLogReplicationFactor!=r.tLogReplicationFactor || oldTLogs[0].tLogs.size() != r.tLogs.size()
|| oldTLogs[0].remoteTLogReplicationFactor != r.remoteTLogReplicationFactor || oldTLogs[0].remoteTLogs.size() != r.remoteTLogs.size() )
if( !oldTLogs.size() || tLogs.size() != oldTLogs[0].tLogs.size() ) {
return false;
for(int i = 0; i < oldTLogs[0].tLogs.size(); i++ ) {
if( oldTLogs[0].tLogs[i].id() != r.tLogs[i].id() )
return false;
}
for(int i = 0; i < oldTLogs[0].remoteTLogs.size(); i++ ) {
if( oldTLogs[0].remoteTLogs[i].id() != r.remoteTLogs[i].id() )
for( int i = 0; i < tLogs.size(); i++ ) {
if( !tLogs[i].isEqualIds(oldTLogs[0].tLogs[i]) ) {
return false;
}
}
return true;
}
template <class Ar>
void serialize( Ar& ar ) {
ar & logSystemType & tLogs & oldTLogs & tLogWriteAntiQuorum & tLogReplicationFactor & tLogPolicy & tLogLocalities & remoteTLogs & remoteTLogLocalities & logRouters & remoteTLogReplicationFactor & remoteTLogsRecovered;
ar & logSystemType & tLogs & oldTLogs;
}
};

186
fdbserver/TLogServer.actor.cpp
View File

@ -37,6 +37,7 @@
#include "ServerDBInfo.h"
#include "LogSystem.h"
#include "WaitFailure.h"
#include "RecoveryState.h"
using std::pair;
using std::make_pair;
@ -316,13 +317,14 @@ struct LogData : NonCopyable, public ReferenceCounted<LogData> {
PromiseStream<Future<Void>> addActor;
TLogData* tLogData;
Reference<AsyncVar<Reference<ILogSystem>>> logSystem;
Optional<Tag> remoteTag;
int persistentDataFormat;
explicit LogData(TLogData* tLogData, TLogInterface interf, int persistentDataFormat = 1) : tLogData(tLogData), knownCommittedVersion(0), tli(interf), logId(interf.id()),
cc("TLog", interf.id().toString()),
bytesInput("bytesInput", cc),
bytesDurable("bytesDurable", cc),
explicit LogData(TLogData* tLogData, TLogInterface interf, Optional<Tag> remoteTag, int persistentDataFormat = 1) : tLogData(tLogData), knownCommittedVersion(0), tli(interf), logId(interf.id()),
cc("TLog", interf.id().toString()), bytesInput("bytesInput", cc), bytesDurable("bytesDurable", cc), remoteTag(remoteTag), persistentDataFormat(persistentDataFormat), logSystem(new AsyncVar<Reference<ILogSystem>>()),
// These are initialized differently on init() or recovery
recoveryCount(), stopped(false), initialized(false), queueCommittingVersion(0), newPersistentDataVersion(invalidVersion), persistentDataFormat(persistentDataFormat)
recoveryCount(), stopped(false), initialized(false), queueCommittingVersion(0), newPersistentDataVersion(invalidVersion)
{
startRole(interf.id(), UID(), "TLog");
@ -968,6 +970,9 @@ ACTOR Future<Void> doQueueCommit( TLogData* self, Reference<LogData> logData ) {
logData->queueCommittedVersion.set(ver);
self->queueCommitEnd.set(commitNumber);
if(logData->remoteTag.present() && logData->logSystem->get())
logData->logSystem->get()->pop(ver+1, logData->remoteTag.get());
TraceEvent("TLogCommitDurable", self->dbgid).detail("Version", ver);
return Void();
@ -1095,11 +1100,24 @@ ACTOR Future<Void> rejoinMasters( TLogData* self, TLogInterface tli, DBRecoveryC
state UID lastMasterID(0,0);
loop {
auto const& inf = self->dbInfo->get();
bool isDisplaced = inf.recoveryCount >= recoveryCount && inf.recoveryState != 0 &&
!std::count( inf.logSystemConfig.tLogs.begin(), inf.logSystemConfig.tLogs.end(), tli.id() ) &&
!std::count( inf.priorCommittedLogServers.begin(), inf.priorCommittedLogServers.end(), tli.id() );
for(int i = 0; i < inf.logSystemConfig.oldTLogs.size() && isDisplaced; i++) {
isDisplaced = !std::count( inf.logSystemConfig.oldTLogs[i].tLogs.begin(), inf.logSystemConfig.oldTLogs[i].tLogs.end(), tli.id() );
bool isDisplaced = inf.recoveryCount >= recoveryCount && inf.recoveryState != 0 && !std::count( inf.priorCommittedLogServers.begin(), inf.priorCommittedLogServers.end(), tli.id() );
if(isDisplaced) {
for(auto& log : inf.logSystemConfig.tLogs) {
if( std::count( log.tLogs.begin(), log.tLogs.end(), tli.id() ) ) {
isDisplaced = false;
break;
}
}
}
if(isDisplaced) {
for(auto& old : inf.logSystemConfig.oldTLogs) {
for(auto& log : old.tLogs) {
if( std::count( log.tLogs.begin(), log.tLogs.end(), tli.id() ) ) {
isDisplaced = false;
break;
}
}
}
}
if ( isDisplaced )
{
@ -1170,7 +1188,26 @@ void getQueuingMetrics( TLogData* self, TLogQueuingMetricsRequest const& req ) {
}
ACTOR Future<Void> serveTLogInterface( TLogData* self, TLogInterface tli, Reference<LogData> logData, PromiseStream<Void> warningCollectorInput ) {
state Future<Void> dbInfoChange = Void();
loop choose {
when( Void _ = wait( dbInfoChange ) ) {
dbInfoChange = self->dbInfo->onChange();
bool found = false;
if(self->dbInfo->get().recoveryState >= RecoveryState::FULLY_RECOVERED) {
for(auto& log : self->dbInfo->get().logSystemConfig.tLogs) {
if( std::count( self->dbInfo->get().logSystemConfig.tLogs.begin(), self->dbInfo->get().logSystemConfig.tLogs.end(), logData->logId ) ) {
found = true;
break;
}
}
}
if(found) {
logData->logSystem->set(ILogSystem::fromServerDBInfo( self->dbgid, self->dbInfo->get() ));
} else {
logData->logSystem->set(Reference<ILogSystem>());
}
}
when( TLogPeekRequest req = waitNext( tli.peekMessages.getFuture() ) ) {
logData->addActor.send( tLogPeekMessages( self, req, logData ) );
}
@ -1178,6 +1215,7 @@ ACTOR Future<Void> serveTLogInterface( TLogData* self, TLogInterface tli, Refere
logData->addActor.send( tLogPop( self, req, logData ) );
}
when( TLogCommitRequest req = waitNext( tli.commit.getFuture() ) ) {
ASSERT(!logData->remoteTag.present());
TEST(logData->stopped); // TLogCommitRequest while stopped
if (!logData->stopped)
logData->addActor.send( tLogCommit( self, req, logData, warningCollectorInput ) );
@ -1218,6 +1256,107 @@ void removeLog( TLogData* self, Reference<LogData> logData ) {
}
}
ACTOR Future<Void> pullAsyncData( TLogData* self, Reference<LogData> logData, Tag tag ) {
state Future<Void> dbInfoChange = Void();
state Reference<ILogSystem::IPeekCursor> r;
state Version tagAt = logData->version.get()+1;
state Version tagPopped = 0;
state Version lastVer = 0;
loop {
loop {
choose {
when(Void _ = wait( r ? r->getMore() : Never() ) ) {
break;
}
when( Void _ = wait( dbInfoChange ) ) {
if(r) tagPopped = std::max(tagPopped, r->popped());
if( logData->logSystem->get() )
r = logData->logSystem->get()->peek( tagAt, tag );
else
r = Reference<ILogSystem::IPeekCursor>();
dbInfoChange = logData->logSystem->onChange();
}
}
}
Version ver = 0;
Arena arena;
BinaryWriter wr(Unversioned());
Map<Tag, TagMessagesRef> tag_offsets;
while (true) {
bool foundMessage = r->hasMessage();
if (!foundMessage || r->version().version != ver) {
ASSERT(r->version().version > lastVer);
if (ver) {
VectorRef<TagMessagesRef> r;
for(auto& t : tag_offsets)
r.push_back( arena, t.value );
commitMessages(logData, ver, arena, wr.toStringRef(), r, self->bytesInput);
// Log the changes to the persistent queue, to be committed by commitQueue()
TLogQueueEntryRef qe;
qe.version = ver;
qe.messages = wr.toStringRef();
qe.tags = r;
self->persistentQueue->push( qe );
// Notifies the commitQueue actor to commit persistentQueue, and also unblocks tLogPeekMessages actors
//FIXME: could we just use the ver and lastVer variables, or replace them with this?
self->prevVersion = logData->version.get();
logData->version.set( ver );
}
lastVer = ver;
ver = r->version().version;
tag_offsets = Map<Tag, TagMessagesRef>();
wr = BinaryWriter(Unversioned());
arena = Arena();
if (!foundMessage) {
ver--;
if(ver > logData->version.get()) {
commitMessages(logData, ver, arena, StringRef(), VectorRef<TagMessagesRef>(), self->bytesInput);
// Log the changes to the persistent queue, to be committed by commitQueue()
TLogQueueEntryRef qe;
qe.version = ver;
qe.messages = StringRef();
qe.tags = VectorRef<TagMessagesRef>();
self->persistentQueue->push( qe );
// Notifies the commitQueue actor to commit persistentQueue, and also unblocks tLogPeekMessages actors
//FIXME: could we just use the ver and lastVer variables, or replace them with this?
self->prevVersion = logData->version.get();
logData->version.set( ver );
}
break;
}
}
StringRef msg = r->getMessage();
auto tags = r->getTags();
for(auto tag : tags) {
auto it = tag_offsets.find(tag);
if (it == tag_offsets.end()) {
it = tag_offsets.insert(mapPair( tag, TagMessagesRef() ));
it->value.tag = it->key;
}
it->value.messageOffsets.push_back( arena, wr.getLength() );
}
//FIXME: do not reserialize tag data
wr << uint32_t( msg.size() + sizeof(uint32_t) + sizeof(uint16_t) + tags.size()*sizeof(Tag) ) << r->version().sub << uint16_t(tags.size());
for(auto t : tags) {
wr << t;
}
wr.serializeBytes( msg );
r->nextMessage();
}
tagAt = r->version().version;
}
}
ACTOR Future<Void> tLogCore( TLogData* self, Reference<LogData> logData, Future<Void> recovery ) {
if(logData->removed.isReady()) {
Void _ = wait(delay(0)); //to avoid iterator invalidation in restorePersistentState when removed is already ready
@ -1250,6 +1389,10 @@ ACTOR Future<Void> tLogCore( TLogData* self, Reference<LogData> logData, Future<
logData->addActor.send( traceCounters("TLogMetrics", logData->logId, SERVER_KNOBS->STORAGE_LOGGING_DELAY, &logData->cc, self->dbgid.toString() + "/TLogMetrics"));
logData->addActor.send( serveTLogInterface(self, logData->tli, logData, warningCollectorInput) );
if(logData->remoteTag.present()) {
logData->addActor.send( pullAsyncData(self, logData, logData->remoteTag.get()) );
}
try {
Void _ = wait( error );
throw internal_error();
@ -1358,7 +1501,8 @@ ACTOR Future<Void> restorePersistentState( TLogData* self, LocalityData locality
DUMPTOKEN( recruited.getQueuingMetrics );
DUMPTOKEN( recruited.confirmRunning );
logData = Reference<LogData>( new LogData(self, recruited, persistentDataFormat) );
//We do not need the remoteTag, because we will not be loading any additional data
logData = Reference<LogData>( new LogData(self, recruited, Optional<Tag>(), persistentDataFormat) );
logData->stopped = true;
self->id_data[id1] = logData;
@ -1526,10 +1670,9 @@ ACTOR Future<Void> recoverTagFromLogSystem( TLogData* self, Reference<LogData> l
StringRef msg = r->getMessage();
auto tags = r->getTags();
wr << uint32_t( msg.size() + sizeof(uint32_t) + sizeof(uint16_t) + tags.size()*sizeof(Tag) ) << r->version().sub << uint16_t(tags.size());
for(auto& t : tags) {
for(auto t : tags) {
wr << t;
}
wr.serializeBytes( msg );
r->nextMessage();
}
@ -1559,11 +1702,20 @@ ACTOR Future<Void> recoverTagFromLogSystem( TLogData* self, Reference<LogData> l
ACTOR Future<Void> updateLogSystem(TLogData* self, Reference<LogData> logData, LogSystemConfig recoverFrom, Reference<AsyncVar<Reference<ILogSystem>>> logSystem) {
loop {
TraceEvent("TLogUpdate", self->dbgid).detail("logId", logData->logId).detail("recoverFrom", recoverFrom.toString()).detail("dbInfo", self->dbInfo->get().logSystemConfig.toString());
bool found = false;
if( self->dbInfo->get().logSystemConfig.isEqualIds(recoverFrom) ) {
logSystem->set(ILogSystem::fromLogSystemConfig( logData->logId, self->dbInfo->get().myLocality, self->dbInfo->get().logSystemConfig ));
} else if( self->dbInfo->get().logSystemConfig.isNextGenerationOf(recoverFrom) && std::count( self->dbInfo->get().logSystemConfig.tLogs.begin(), self->dbInfo->get().logSystemConfig.tLogs.end(), logData->logId ) ) {
logSystem->set(ILogSystem::fromOldLogSystemConfig( logData->logId, self->dbInfo->get().myLocality, self->dbInfo->get().logSystemConfig ));
} else {
found = true;
} else if( self->dbInfo->get().logSystemConfig.isNextGenerationOf(recoverFrom) ) {
for( auto& it : self->dbInfo->get().logSystemConfig.tLogs ) {
if( std::count(it.tLogs.begin(), it.tLogs.end(), logData->logId ) ) {
logSystem->set(ILogSystem::fromOldLogSystemConfig( logData->logId, self->dbInfo->get().myLocality, self->dbInfo->get().logSystemConfig ));
found = true;
break;
}
}
}
if( !found ) {
logSystem->set(Reference<ILogSystem>());
}
Void _ = wait( self->dbInfo->onChange() );
@ -1650,7 +1802,7 @@ ACTOR Future<Void> tLogStart( TLogData* self, InitializeTLogRequest req, Localit
it.second->stopped = true;
}
state Reference<LogData> logData = Reference<LogData>( new LogData(self, recruited) );
state Reference<LogData> logData = Reference<LogData>( new LogData(self, recruited, req.remoteTag) );
self->id_data[recruited.id()] = logData;
logData->recoveryCount = req.epoch;
logData->removed = rejoinMasters(self, recruited, req.epoch);

1021
fdbserver/TagPartitionedLogSystem.actor.cpp
View File

File diff suppressed because it is too large Load Diff

6
fdbserver/WorkerInterface.h
View File

@ -79,25 +79,25 @@ struct InitializeTLogRequest {
std::vector<Tag> recoverTags;
KeyValueStoreType storeType;
Optional<Tag> remoteTag;
Version minRemoteVersion;
ReplyPromise< struct TLogInterface > reply;
InitializeTLogRequest() {}
template <class Ar>
void serialize( Ar& ar ) {
ar & recruitmentID & recoverFrom & recoverAt & knownCommittedVersion & epoch & recoverTags & storeType & remoteTag & minRemoteVersion & reply;
ar & recruitmentID & recoverFrom & recoverAt & knownCommittedVersion & epoch & recoverTags & storeType & remoteTag & reply;
}
};
struct InitializeLogRouterRequest {
uint64_t recoveryCount;
int logSet;
Tag routerTag;
ReplyPromise<struct TLogInterface> reply;
template <class Ar>
void serialize(Ar& ar) {
ar & recoveryCount & routerTag & reply;
ar & recoveryCount & routerTag & logSet & reply;
}
};

1
fdbserver/fdbserver.vcxproj
View File

@ -54,6 +54,7 @@
<ActorCompiler Include="LogSystemDiskQueueAdapter.actor.cpp" />
<ActorCompiler Include="LogSystemPeekCursor.actor.cpp" />
<ActorCompiler Include="OldTLogServer.actor.cpp" />
<ActorCompiler Include="LogRouter.actor.cpp" />
<ClCompile Include="SkipList.cpp" />
<ActorCompiler Include="WaitFailure.actor.cpp" />
<ActorCompiler Include="tester.actor.cpp" />

7
fdbserver/fdbserver.vcxproj.filters
View File

@ -247,6 +247,13 @@
<Filter>workloads</Filter>
</ActorCompiler>
<ActorCompiler Include="OldTLogServer.actor.cpp" />
<ActorCompiler Include="LogRouter.actor.cpp" />
<ActorCompiler Include="workloads\SlowTaskWorkload.actor.cpp">
<Filter>workloads</Filter>
</ActorCompiler>
<ActorCompiler Include="workloads\DiskDurability.actor.cpp">
<Filter>workloads</Filter>
</ActorCompiler>
</ItemGroup>
<ItemGroup>
<ClCompile Include="SkipList.cpp" />

5
fdbserver/masterserver.actor.cpp
View File

@ -948,7 +948,7 @@ ACTOR Future<Void> trackTlogRecovery( Reference<MasterData> self, Reference<Asyn
loop {
DBCoreState coreState;
self->logSystem->toCoreState( coreState );
if( !self->fullyRecovered.isSet() && coreState.remoteTLogsRecovered ) { //FIXME: !coreState.oldTLogData.size()
if( !self->fullyRecovered.isSet() && !coreState.oldTLogData.size() ) {
if( !skipTransition ) {
Void _ = wait( writeRecoveredMasterState(self) );
self->registrationTrigger.trigger();
@ -1108,7 +1108,6 @@ ACTOR Future<Void> masterCore( Reference<MasterData> self, PromiseStream<Future<
TraceEvent("MasterRecoveryState", self->dbgid)
.detail("StatusCode", RecoveryStatus::writing_coordinated_state)
.detail("Status", RecoveryStatus::names[RecoveryStatus::writing_coordinated_state])
.detail("TLogs", self->logSystem->getLogServerCount())
.detail("TLogList", self->logSystem->describe())
.trackLatest(format("%s/MasterRecoveryState", printable(self->dbName).c_str() ).c_str());
@ -1125,7 +1124,7 @@ ACTOR Future<Void> masterCore( Reference<MasterData> self, PromiseStream<Future<
DBCoreState coreState;
self->logSystem->toCoreState( coreState );
state bool skipTransition = false; //FIXME: !coreState.oldTLogData.size();
state bool skipTransition = !coreState.oldTLogData.size();
debug_advanceMaxCommittedVersion(UID(), self->recoveryTransactionVersion);
Void _ = wait( writeTransitionMasterState( self, skipTransition ) );