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

made a large number of fixes to make fearless DR correctness clean.

This commit is contained in:
Evan Tschannen 2017-10-19 15:36:32 -07:00
parent ff1b49be2e
commit e2c1e87df6
16 changed files with 354 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
fdbclient/Knobs.cpp
View File

@ -75,6 +75,9 @@ ClientKnobs::ClientKnobs(bool randomize) {
init( DEFAULT_MAX_OUTSTANDING_WATCHES, 1e4 );
init( ABSOLUTE_MAX_WATCHES, 1e6 );
init( WATCH_POLLING_TIME, 1.0 ); if( randomize && BUGGIFY ) WATCH_POLLING_TIME = 5.0;
init( NO_RECENT_UPDATES_DURATION, 20.0 ); if( randomize && BUGGIFY ) NO_RECENT_UPDATES_DURATION = 0.1;
init( FAST_WATCH_TIMEOUT, 20.0 ); if( randomize && BUGGIFY ) FAST_WATCH_TIMEOUT = 1.0;
init( WATCH_TIMEOUT, 900.0 ); if( randomize ) WATCH_TIMEOUT = 20.0;
// Core
init( CORE_VERSIONSPERSECOND, 1e6 );

3
fdbclient/Knobs.h
View File

@ -74,6 +74,9 @@ public:
int DEFAULT_MAX_OUTSTANDING_WATCHES;
int ABSOLUTE_MAX_WATCHES; //The client cannot set the max outstanding watches higher than this
double WATCH_POLLING_TIME;
double NO_RECENT_UPDATES_DURATION;
double FAST_WATCH_TIMEOUT;
double WATCH_TIMEOUT;
double IS_ACCEPTABLE_DELAY;

113
fdbclient/NativeAPI.actor.cpp
View File

@ -1071,6 +1071,8 @@ ACTOR Future<Optional<vector<StorageServerInterface>>> getServerInterfaces(
return serverInterfaces;
}
//FIXME: we are now using a version for getting server interfaces instead of latest version, because latest version is wrong if the storage server we are talking to is behind.
// A better solution would be for latest version to be a flag, and the storage server gives you data at the latest version if that is greater than the supplied read verison.
ACTOR Future<Optional<vector<StorageServerInterface>>> transactionalGetServerInterfaces( Future<Version> ver, Database cx, TransactionInfo info, vector<UID> ids ) {
state vector< Future< Optional<StorageServerInterface> > > serverListEntries;
for( int s = 0; s < ids.size(); s++ ) {
@ -1090,7 +1092,7 @@ ACTOR Future<Optional<vector<StorageServerInterface>>> transactionalGetServerInt
}
//If isBackward == true, returns the shard containing the key before 'key' (an infinitely long, inexpressible key). Otherwise returns the shard containing key
ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation( Database cx, Key key, TransactionInfo info, bool isBackward = false ) {
ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation( Database cx, Key key, TransactionInfo info, Version version, bool isBackward = false ) {
if (isBackward)
ASSERT( key != allKeys.begin && key <= allKeys.end );
else
@ -1138,14 +1140,14 @@ ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation( Database
state Standalone<RangeResultRef> results;
if( isBackward ) {
Standalone<RangeResultRef> _results = wait(
getRange( cx, latestVersion,
getRange( cx, version,
lastLessThan( keyServersK ),
firstGreaterOrEqual( keyServersK ) + 1,
GetRangeLimits( 2 ), false, info ) );
results = _results;
} else {
Standalone<RangeResultRef> _results = wait(
getRange( cx, latestVersion,
getRange( cx, version,
lastLessOrEqual( keyServersK ),
firstGreaterThan( keyServersK ) + 1,
GetRangeLimits( 2 ), false, info ) );
@ -1177,7 +1179,7 @@ ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation( Database
ASSERT(false);
}
Optional<vector<StorageServerInterface>> interfs = wait( getServerInterfaces(cx, info, src) );
Optional<vector<StorageServerInterface>> interfs = wait( transactionalGetServerInterfaces(version, cx, info, src) );
if( interfs.present() ) {
if( info.debugID.present() )
g_traceBatch.addEvent("TransactionDebug", info.debugID.get().first(), "NativeAPI.getKeyLocation.Interfs.present");
@ -1198,7 +1200,7 @@ ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation( Database
return make_pair(range, cx->setCachedLocation( range, serverInterfaces ));
}
ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations_internal( Database cx, KeyRange keys, int limit, bool reverse, TransactionInfo info ) {
ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations_internal( Database cx, KeyRange keys, int limit, bool reverse, TransactionInfo info, Version version = latestVersion ) {
//printf("getKeyRangeLocations: getting '%s'-'%s'\n", keyServersKey(keys.begin).toString().c_str(), keyServersKey(keys.end).toString().c_str());
loop {
state vector< pair<KeyRange,Reference<LocationInfo>> > result;
@ -1207,7 +1209,7 @@ ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLoca
state bool ok = true;
state Standalone<RangeResultRef> keyServersEntries =
wait(
getRange( cx, latestVersion,
getRange( cx, version,
lastLessOrEqual( keyServersKey(keys.begin) ),
firstGreaterOrEqual( keyServersKey(keys.end) ) + 1,
GetRangeLimits( limit + 1 ), reverse, info ) );
@ -1231,7 +1233,7 @@ ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLoca
ASSERT( servers.size() );
serverListReads.push_back( make_pair( range, getServerInterfaces(cx, info, servers) ) );
serverListReads.push_back( make_pair( range, transactionalGetServerInterfaces(version, cx, info, servers) ) );
serverListServers.push_back( servers );
}
@ -1264,14 +1266,14 @@ ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLoca
}
}
Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations( Database cx, KeyRange keys, int limit, bool reverse, TransactionInfo info ) {
Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations( Database cx, KeyRange keys, int limit, bool reverse, TransactionInfo info, Version version = latestVersion ) {
ASSERT (!keys.empty());
vector< pair<KeyRange,Reference<LocationInfo>> > result;
if (cx->getCachedLocations(keys, result, limit, reverse))
return result;
return getKeyRangeLocations_internal( cx, keys, limit, reverse, info );
return getKeyRangeLocations_internal( cx, keys, limit, reverse, info, version );
}
ACTOR Future<Void> warmRange_impl( Transaction *self, Database cx, KeyRange keys ) {
@ -1301,7 +1303,7 @@ ACTOR Future<Optional<Value>> getValue( Future<Version> version, Key key, Databa
loop {
ssi = cx->getCachedLocation( key );
if (!ssi.second) {
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info ) );
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info, ver ) );
ssi = std::move(ssi2);
} else {
bool onlyEndpointFailed = false;
@ -1315,11 +1317,11 @@ ACTOR Future<Optional<Value>> getValue( Future<Version> version, Key key, Databa
if( onlyEndpointFailed ) {
cx->invalidateCache( key );
cx->invalidateCache( ssi.second );
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info ) );
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info, ver ) );
ssi = std::move(ssi2);
}
}
//Reference<LocationInfo> ssi = wait( getKeyLocation( cx, key ) );
state Optional<UID> getValueID = Optional<UID>();
state uint64_t startTime;
state double startTimeD;
@ -1383,7 +1385,7 @@ ACTOR Future<Optional<Value>> getValue( Future<Version> version, Key key, Databa
}
ACTOR Future<Key> getKey( Database cx, KeySelector k, Future<Version> version, TransactionInfo info ) {
Version _ = wait(version);
state Version ver = wait(version);
if( info.debugID.present() )
g_traceBatch.addEvent("TransactionDebug", info.debugID.get().first(), "NativeAPI.getKey.AfterVersion");
@ -1397,7 +1399,27 @@ ACTOR Future<Key> getKey( Database cx, KeySelector k, Future<Version> version, T
return Key();
}
state pair<KeyRange, Reference<LocationInfo>> ssi = wait( getKeyLocation(cx, k.getKey(), info, k.isBackward()) );
state pair<KeyRange, Reference<LocationInfo>> ssi = cx->getCachedLocation( k.getKey(), k.isBackward() );
if (!ssi.second) {
pair<KeyRange, Reference<LocationInfo>> _ssi = wait( getKeyLocation(cx, k.getKey(), info, ver, k.isBackward()) );
ssi = std::move(_ssi);
} else {
bool onlyEndpointFailed = false;
for(int i = 0; i < ssi.second->size(); i++) {
if( IFailureMonitor::failureMonitor().onlyEndpointFailed(ssi.second->get(i, &StorageServerInterface::getKey).getEndpoint()) ) {
onlyEndpointFailed = true;
break;
}
}
if( onlyEndpointFailed ) {
cx->invalidateCache( k.getKey() );
cx->invalidateCache( ssi.second );
pair<KeyRange, Reference<LocationInfo>> _ssi = wait( getKeyLocation(cx, k.getKey(), info, ver, k.isBackward()) );
ssi = std::move(_ssi);
}
}
try {
if( info.debugID.present() )
g_traceBatch.addEvent("TransactionDebug", info.debugID.get().first(), "NativeAPI.getKey.Before"); //.detail("StartKey", printable(k.getKey())).detail("offset",k.offset).detail("orEqual",k.orEqual);
@ -1457,8 +1479,23 @@ ACTOR Future< Void > watchValue( Future<Version> version, Key key, Optional<Valu
loop {
state pair<KeyRange, Reference<LocationInfo>> ssi = cx->getCachedLocation( key );
if (!ssi.second) {
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info ) );
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info, ver ) );
ssi = std::move(ssi2);
} else {
bool onlyEndpointFailed = false;
for(int i = 0; i < ssi.second->size(); i++) {
if( IFailureMonitor::failureMonitor().onlyEndpointFailed(ssi.second->get(i, &StorageServerInterface::watchValue).getEndpoint()) ) {
onlyEndpointFailed = true;
break;
}
}
if( onlyEndpointFailed ) {
cx->invalidateCache( key );
cx->invalidateCache( ssi.second );
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info, ver ) );
ssi = std::move(ssi2);
}
}
try {
@ -1531,7 +1568,29 @@ ACTOR Future<Standalone<RangeResultRef>> getExactRange( Database cx, Version ver
//printf("getExactRange( '%s', '%s' )\n", keys.begin.toString().c_str(), keys.end.toString().c_str());
loop {
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->GET_RANGE_SHARD_LIMIT, reverse, info ) );
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->GET_RANGE_SHARD_LIMIT, reverse, info, version ) );
bool foundFailed = false;
for(auto& it : locations) {
bool onlyEndpointFailed = false;
for(int i = 0; i < it.second->size(); i++) {
if( IFailureMonitor::failureMonitor().onlyEndpointFailed(it.second->get(i, &StorageServerInterface::getKeyValues).getEndpoint()) ) {
onlyEndpointFailed = true;
break;
}
}
if( onlyEndpointFailed ) {
cx->invalidateCache( it.first.begin );
cx->invalidateCache( it.second );
foundFailed = true;
}
}
if(foundFailed) {
vector< pair<KeyRange, Reference<LocationInfo>> > _locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->GET_RANGE_SHARD_LIMIT, reverse, info, version ) );
locations = std::move(_locations);
}
ASSERT( locations.size() );
state int shard = 0;
@ -1736,7 +1795,27 @@ ACTOR Future<Standalone<RangeResultRef>> getRange( Database cx, Future<Version>
return output;
}
state pair<KeyRange, Reference<LocationInfo>> beginServer = wait( getKeyLocation( cx, reverse ? end.getKey() : begin.getKey(), info, reverse ? (end-1).isBackward() : begin.isBackward() ) );
state pair<KeyRange, Reference<LocationInfo>> beginServer = cx->getCachedLocation( reverse ? end.getKey() : begin.getKey(), reverse ? (end-1).isBackward() : begin.isBackward() );
if (!beginServer.second) {
pair<KeyRange, Reference<LocationInfo>> ssi = wait( getKeyLocation( cx, reverse ? end.getKey() : begin.getKey(), info, version, reverse ? (end-1).isBackward() : begin.isBackward() ) );
beginServer = std::move(ssi);
} else {
bool onlyEndpointFailed = false;
for(int i = 0; i < beginServer.second->size(); i++) {
if( IFailureMonitor::failureMonitor().onlyEndpointFailed(beginServer.second->get(i, &StorageServerInterface::getKeyValues).getEndpoint()) ) {
onlyEndpointFailed = true;
break;
}
}
if( onlyEndpointFailed ) {
cx->invalidateCache( reverse ? end.getKey() : begin.getKey() );
cx->invalidateCache( beginServer.second );
pair<KeyRange, Reference<LocationInfo>> ssi = wait( getKeyLocation( cx, reverse ? end.getKey() : begin.getKey(), info, version, reverse ? (end-1).isBackward() : begin.isBackward() ) );
beginServer = std::move(ssi);
}
}
state KeyRange shard = beginServer.first;
state bool modifiedSelectors = false;
state GetKeyValuesRequest req;

78
fdbrpc/LoadBalance.actor.h
View File

@ -51,16 +51,16 @@ struct ModelHolder : NonCopyable, public ReferenceCounted<ModelHolder> {
}
}
void release(bool clean, double penalty, bool measureLatency = true) {
void release(bool clean, bool futureVersion, double penalty, bool measureLatency = true) {
if(model && !released) {
released = true;
double latency = (clean || measureLatency) ? now() - startTime : 0.0;
model->endRequest(token, latency, penalty, delta, clean);
model->endRequest(token, latency, penalty, delta, clean, futureVersion);
}
}
~ModelHolder() {
release(false, -1.0, false);
release(false, false, -1.0, false);
}
};
@ -81,18 +81,19 @@ Optional<LoadBalancedReply> getLoadBalancedReply(void*);
// Throws an error if the request returned an error that should bubble out
// Returns false if we got an error that should result in reissuing the request
template <class T>
bool checkAndProcessResult(ErrorOr<T> result, Reference<ModelHolder> holder, bool atMostOnce) {
bool checkAndProcessResult(ErrorOr<T> result, Reference<ModelHolder> holder, bool atMostOnce, bool triedAllOptions) {
int errCode = result.isError() ? result.getError().code() : error_code_success;
bool maybeDelivered = errCode == error_code_broken_promise || errCode == error_code_request_maybe_delivered;
bool receivedResponse = result.present() || (!maybeDelivered && errCode != error_code_process_behind);
bool futureVersion = errCode == error_code_future_version || errCode == error_code_process_behind;
Optional<LoadBalancedReply> loadBalancedReply;
if(!result.isError()) {
loadBalancedReply = getLoadBalancedReply(&result.get());
}
holder->release(receivedResponse, loadBalancedReply.present() ? loadBalancedReply.get().penalty : -1.0);
holder->release(receivedResponse, futureVersion, loadBalancedReply.present() ? loadBalancedReply.get().penalty : -1.0);
if(result.present()) {
return true;
}
@ -105,11 +106,15 @@ bool checkAndProcessResult(ErrorOr<T> result, Reference<ModelHolder> holder, boo
throw request_maybe_delivered();
}
if(triedAllOptions && errCode == error_code_process_behind) {
throw future_version();
}
return false;
}
ACTOR template <class Request>
Future<Optional<REPLY_TYPE(Request)>> makeRequest(RequestStream<Request> const* stream, Request request, double backoff, Future<Void> requestUnneeded, QueueModel *model, bool isFirstRequest, bool atMostOnce) {
Future<Optional<REPLY_TYPE(Request)>> makeRequest(RequestStream<Request> const* stream, Request request, double backoff, Future<Void> requestUnneeded, QueueModel *model, bool isFirstRequest, bool atMostOnce, bool triedAllOptions) {
if(backoff > 0.0) {
Void _ = wait(delay(backoff) || requestUnneeded);
}
@ -121,7 +126,7 @@ Future<Optional<REPLY_TYPE(Request)>> makeRequest(RequestStream<Request> const*
state Reference<ModelHolder> holder(new ModelHolder(model, stream->getEndpoint().token.first()));
ErrorOr<REPLY_TYPE(Request)> result = wait(stream->tryGetReply(request));
if(checkAndProcessResult(result, holder, atMostOnce)) {
if(checkAndProcessResult(result, holder, atMostOnce, triedAllOptions)) {
return result.get();
}
else {
@ -181,26 +186,32 @@ Future< REPLY_TYPE(Request) > loadBalance(
double nextMetric = 1e9;
double bestTime = 1e9;
double nextTime = 1e9;
for(int i=0; i<alternatives->countBest(); i++) {
for(int i=0; i<alternatives->size(); i++) {
if(bestMetric < 1e8 && i == alternatives->countBest()) {
break;
}
RequestStream<Request> const* thisStream = &alternatives->get( i, channel );
if (!IFailureMonitor::failureMonitor().getState( thisStream->getEndpoint() ).failed) {
auto& qd = model->getMeasurement(thisStream->getEndpoint().token.first());
double thisMetric = qd.smoothOutstanding.smoothTotal();
double thisTime = qd.latency;
if(now() > qd.failedUntil) {
double thisMetric = qd.smoothOutstanding.smoothTotal();
double thisTime = qd.latency;
if(thisMetric < bestMetric) {
if(i != bestAlt) {
nextAlt = bestAlt;
nextMetric = bestMetric;
nextTime = bestTime;
if(thisMetric < bestMetric) {
if(i != bestAlt) {
nextAlt = bestAlt;
nextMetric = bestMetric;
nextTime = bestTime;
}
bestAlt = i;
bestMetric = thisMetric;
bestTime = thisTime;
} else if( thisMetric < nextMetric ) {
nextAlt = i;
nextMetric = thisMetric;
nextTime = thisTime;
}
bestAlt = i;
bestMetric = thisMetric;
bestTime = thisTime;
} else if( thisMetric < nextMetric ) {
nextAlt = i;
nextMetric = thisMetric;
nextTime = thisTime;
}
}
}
@ -209,13 +220,15 @@ Future< REPLY_TYPE(Request) > loadBalance(
RequestStream<Request> const* thisStream = &alternatives->get( i, channel );
if (!IFailureMonitor::failureMonitor().getState( thisStream->getEndpoint() ).failed) {
auto& qd = model->getMeasurement(thisStream->getEndpoint().token.first());
double thisMetric = qd.smoothOutstanding.smoothTotal();
double thisTime = qd.latency;
if(now() > qd.failedUntil) {
double thisMetric = qd.smoothOutstanding.smoothTotal();
double thisTime = qd.latency;
if( thisMetric < nextMetric ) {
nextAlt = i;
nextMetric = thisMetric;
nextTime = thisTime;
if( thisMetric < nextMetric ) {
nextAlt = i;
nextMetric = thisMetric;
nextTime = thisTime;
}
}
}
}
@ -238,6 +251,7 @@ Future< REPLY_TYPE(Request) > loadBalance(
state int numAttempts = 0;
state double backoff = 0;
state bool triedAllOptions = false;
loop {
// Find an alternative, if any, that is not failed, starting with nextAlt
state RequestStream<Request> const* stream = NULL;
@ -252,6 +266,7 @@ Future< REPLY_TYPE(Request) > loadBalance(
if (!IFailureMonitor::failureMonitor().getState( stream->getEndpoint() ).failed && (!firstRequestEndpoint.present() || stream->getEndpoint().token.first() != firstRequestEndpoint.get()))
break;
nextAlt = (nextAlt+1) % alternatives->size();
if(nextAlt == startAlt) triedAllOptions = true;
stream=NULL;
}
@ -304,7 +319,7 @@ Future< REPLY_TYPE(Request) > loadBalance(
firstRequestEndpoint = Optional<uint64_t>();
} else if( firstRequest.isValid() ) {
//Issue a second request, the first one is taking a long time.
secondRequest = makeRequest(stream, request, backoff, requestFinished.getFuture(), model, false, atMostOnce);
secondRequest = makeRequest(stream, request, backoff, requestFinished.getFuture(), model, false, atMostOnce, triedAllOptions);
state bool firstFinished = false;
loop {
@ -347,7 +362,7 @@ Future< REPLY_TYPE(Request) > loadBalance(
}
} else {
//Issue a request, if it takes too long to get a reply, go around the loop
firstRequest = makeRequest(stream, request, backoff, requestFinished.getFuture(), model, true, atMostOnce);
firstRequest = makeRequest(stream, request, backoff, requestFinished.getFuture(), model, true, atMostOnce, triedAllOptions);
firstRequestEndpoint = stream->getEndpoint().token.first();
loop {
@ -387,6 +402,7 @@ Future< REPLY_TYPE(Request) > loadBalance(
}
nextAlt = (nextAlt+1) % alternatives->size();
if(nextAlt == startAlt) triedAllOptions = true;
resetReply(request, taskID);
secondDelay = Never();
}

13
fdbrpc/QueueModel.cpp
View File

@ -21,7 +21,7 @@
#include "QueueModel.h"
#include "LoadBalance.h"
void QueueModel::endRequest( uint64_t id, double latency, double penalty, double delta, bool clean ) {
void QueueModel::endRequest( uint64_t id, double latency, double penalty, double delta, bool clean, bool futureVersion ) {
auto& d = data[id];
d.smoothOutstanding.addDelta(-delta);
@ -31,6 +31,17 @@ void QueueModel::endRequest( uint64_t id, double latency, double penalty, double
d.latency = std::max(d.latency, latency);
}
if(futureVersion) {
if(now() > d.increaseBackoffTime) {
d.futureVersionBackoff = std::min( d.futureVersionBackoff * FLOW_KNOBS->FUTURE_VERSION_BACKOFF_GROWTH, FLOW_KNOBS->FUTURE_VERSION_MAX_BACKOFF );
d.increaseBackoffTime = now() + d.futureVersionBackoff;
}
d.failedUntil = now() + d.futureVersionBackoff;
} else if(clean) {
d.futureVersionBackoff = FLOW_KNOBS->FUTURE_VERSION_INITIAL_BACKOFF;
d.increaseBackoffTime = 0.0;
}
if(penalty > 0) {
d.penalty = penalty;
}

7
fdbrpc/QueueModel.h
View File

@ -32,14 +32,17 @@ struct QueueData {
Smoother smoothOutstanding;
double latency;
double penalty;
QueueData() : latency(0.001), penalty(1.0), smoothOutstanding(FLOW_KNOBS->QUEUE_MODEL_SMOOTHING_AMOUNT) {}
double failedUntil;
double futureVersionBackoff;
double increaseBackoffTime;
QueueData() : latency(0.001), penalty(1.0), smoothOutstanding(FLOW_KNOBS->QUEUE_MODEL_SMOOTHING_AMOUNT), failedUntil(0), futureVersionBackoff(FLOW_KNOBS->FUTURE_VERSION_INITIAL_BACKOFF), increaseBackoffTime(0) {}
};
typedef double TimeEstimate;
class QueueModel {
public:
void endRequest( uint64_t id, double latency, double penalty, double delta, bool clean );
void endRequest( uint64_t id, double latency, double penalty, double delta, bool clean, bool futureVersion );
QueueData& getMeasurement( uint64_t id );
double addRequest( uint64_t id );
double secondMultiplier;

126
fdbrpc/sim2.actor.cpp
View File

@ -1023,62 +1023,112 @@ public:
// The following function will determine if the specified configuration of available and dead processes can allow the cluster to survive
virtual bool canKillProcesses(std::vector<ProcessInfo*> const& availableProcesses, std::vector<ProcessInfo*> const& deadProcesses, KillType kt, KillType* newKillType) const
{
bool canSurvive = true;
int nQuorum = ((desiredCoordinators+1)/2)*2-1;
bool canSurvive = true;
int nQuorum = ((desiredCoordinators+1)/2)*2-1;
KillType newKt = kt;
KillType newKt = kt;
if ((kt == KillInstantly) || (kt == InjectFaults) || (kt == RebootAndDelete) || (kt == RebootProcessAndDelete))
{
LocalityGroup processesLeft, processesDead;
std::vector<LocalityData> localitiesDead, localitiesLeft, badCombo;
std::set<Optional<Standalone<StringRef>>> uniqueMachines;
ASSERT(storagePolicy);
ASSERT(tLogPolicy);
for (auto processInfo : availableProcesses) {
processesLeft.add(processInfo->locality);
localitiesLeft.push_back(processInfo->locality);
uniqueMachines.insert(processInfo->locality.machineId());
LocalityGroup primaryProcessesLeft, primaryProcessesDead;
LocalityGroup primarySatelliteProcessesLeft, primarySatelliteProcessesDead;
LocalityGroup remoteProcessesLeft, remoteProcessesDead;
LocalityGroup remoteSatelliteProcessesLeft, remoteSatelliteProcessesDead;
std::vector<LocalityData> primaryLocalitiesDead, primaryLocalitiesLeft;
std::vector<LocalityData> primarySatelliteLocalitiesDead, primarySatelliteLocalitiesLeft;
std::vector<LocalityData> remoteLocalitiesDead, remoteLocalitiesLeft;
std::vector<LocalityData> remoteSatelliteLocalitiesDead, remoteSatelliteLocalitiesLeft;
std::vector<LocalityData> badCombo;
std::set<Optional<Standalone<StringRef>>> uniqueMachines;
if(!hasRemoteReplication) {
for (auto processInfo : availableProcesses) {
primaryProcessesLeft.add(processInfo->locality);
primaryLocalitiesLeft.push_back(processInfo->locality);
uniqueMachines.insert(processInfo->locality.machineId());
}
for (auto processInfo : deadProcesses) {
primaryProcessesDead.add(processInfo->locality);
primaryLocalitiesDead.push_back(processInfo->locality);
}
} else {
for (auto processInfo : availableProcesses) {
uniqueMachines.insert(processInfo->locality.machineId());
if(processInfo->locality.dcId() == primaryDcId) {
primaryProcessesLeft.add(processInfo->locality);
primaryLocalitiesLeft.push_back(processInfo->locality);
} else if(processInfo->locality.dcId() == remoteDcId) {
remoteProcessesLeft.add(processInfo->locality);
remoteLocalitiesLeft.push_back(processInfo->locality);
} else if(std::find(primarySatelliteDcIds.begin(), primarySatelliteDcIds.end(), processInfo->locality.dcId()) != primarySatelliteDcIds.end()) {
primarySatelliteProcessesLeft.add(processInfo->locality);
primarySatelliteLocalitiesLeft.push_back(processInfo->locality);
} else if(std::find(remoteSatelliteDcIds.begin(), remoteSatelliteDcIds.end(), processInfo->locality.dcId()) != remoteSatelliteDcIds.end()) {
remoteSatelliteProcessesLeft.add(processInfo->locality);
remoteSatelliteLocalitiesLeft.push_back(processInfo->locality);
}
}
for (auto processInfo : deadProcesses) {
if(processInfo->locality.dcId() == primaryDcId) {
primaryProcessesDead.add(processInfo->locality);
primaryLocalitiesDead.push_back(processInfo->locality);
} else if(processInfo->locality.dcId() == remoteDcId) {
remoteProcessesDead.add(processInfo->locality);
remoteLocalitiesDead.push_back(processInfo->locality);
} else if(std::find(primarySatelliteDcIds.begin(), primarySatelliteDcIds.end(), processInfo->locality.dcId()) != primarySatelliteDcIds.end()) {
primarySatelliteProcessesDead.add(processInfo->locality);
primarySatelliteLocalitiesDead.push_back(processInfo->locality);
} else if(std::find(remoteSatelliteDcIds.begin(), remoteSatelliteDcIds.end(), processInfo->locality.dcId()) != remoteSatelliteDcIds.end()) {
remoteSatelliteProcessesDead.add(processInfo->locality);
remoteSatelliteLocalitiesDead.push_back(processInfo->locality);
}
}
}
for (auto processInfo : deadProcesses) {
processesDead.add(processInfo->locality);
localitiesDead.push_back(processInfo->locality);
bool tooManyDead = false;
bool notEnoughLeft = false;
bool primaryTLogsDead = tLogWriteAntiQuorum ? !validateAllCombinations(badCombo, primaryProcessesDead, tLogPolicy, primaryLocalitiesLeft, tLogWriteAntiQuorum, false) : primaryProcessesDead.validate(tLogPolicy);
if(!hasRemoteReplication) {
tooManyDead = primaryTLogsDead || primaryProcessesDead.validate(storagePolicy);
notEnoughLeft = !primaryProcessesLeft.validate(tLogPolicy) || !primaryProcessesLeft.validate(storagePolicy);
} else {
bool remoteTLogsDead = tLogWriteAntiQuorum ? !validateAllCombinations(badCombo, remoteProcessesDead, tLogPolicy, remoteLocalitiesLeft, tLogWriteAntiQuorum, false) : remoteProcessesDead.validate(tLogPolicy);
if(!hasSatelliteReplication) {
tooManyDead = primaryTLogsDead || remoteTLogsDead ||
( ( primaryProcessesDead.validate(storagePolicy) || primaryProcessesDead.validate(remoteStoragePolicy) ) && ( remoteProcessesDead.validate(storagePolicy) || remoteProcessesDead.validate(remoteStoragePolicy) ) );
notEnoughLeft = ( !primaryProcessesLeft.validate(tLogPolicy) || !primaryProcessesLeft.validate(storagePolicy) ) && ( !remoteProcessesLeft.validate(tLogPolicy) || !remoteProcessesLeft.validate(storagePolicy) );
} else {
bool primarySatelliteTLogsDead = satelliteTLogWriteAntiQuorum ? !validateAllCombinations(badCombo, primarySatelliteProcessesDead, satelliteTLogPolicy, primarySatelliteLocalitiesLeft, satelliteTLogWriteAntiQuorum, false) : primarySatelliteProcessesDead.validate(satelliteTLogPolicy);
bool remoteSatelliteTLogsDead = satelliteTLogWriteAntiQuorum ? !validateAllCombinations(badCombo, remoteSatelliteProcessesDead, satelliteTLogPolicy, remoteSatelliteLocalitiesLeft, satelliteTLogWriteAntiQuorum, false) : remoteSatelliteProcessesDead.validate(satelliteTLogPolicy);
tooManyDead = ( primaryTLogsDead && primarySatelliteTLogsDead ) || ( remoteTLogsDead && remoteSatelliteTLogsDead ) ||
( ( primaryProcessesDead.validate(storagePolicy) || primaryProcessesDead.validate(remoteStoragePolicy) ) && ( remoteProcessesDead.validate(storagePolicy) || remoteProcessesDead.validate(remoteStoragePolicy) ) );
notEnoughLeft = ( !primaryProcessesLeft.validate(tLogPolicy) || !primaryProcessesLeft.validate(storagePolicy) || !primarySatelliteProcessesLeft.validate(satelliteTLogPolicy) ) && ( !remoteProcessesLeft.validate(tLogPolicy) || !remoteProcessesLeft.validate(storagePolicy) || !remoteSatelliteProcessesLeft.validate(satelliteTLogPolicy) );
}
}
// Reboot if dead machines do fulfill policies
if (processesDead.validate(tLogPolicy)) {
if (tooManyDead) {
newKt = Reboot;
canSurvive = false;
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("tLogPolicy", tLogPolicy->info()).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("DeadZones", ::describeZones(localitiesDead)).detail("DeadDataHalls", ::describeDataHalls(localitiesDead)).detail("Reason", "tLogPolicy validates against dead processes.");
}
else if (processesDead.validate(storagePolicy)) {
newKt = Reboot;
canSurvive = false;
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("storagePolicy", storagePolicy->info()).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("DeadZones", ::describeZones(localitiesDead)).detail("DeadDataHalls", ::describeDataHalls(localitiesDead)).detail("Reason", "storagePolicy validates against dead processes.");
}
// Check all combinations of the AntiQuorum within the failed
else if ((tLogWriteAntiQuorum) && (!validateAllCombinations(badCombo, processesDead, tLogPolicy, localitiesLeft, tLogWriteAntiQuorum, false)))
{
newKt = Reboot;
canSurvive = false;
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("storagePolicy", storagePolicy->info()).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("BadZones", ::describeZones(badCombo)).detail("BadDataHalls", ::describeDataHalls(badCombo)).detail("Reason", "tLog AntiQuorum does not validates against dead processes.");
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("tLogPolicy", tLogPolicy->info()).detail("Reason", "tLogPolicy validates against dead processes.");
}
// Reboot and Delete if remaining machines do NOT fulfill policies
else if ((kt != RebootAndDelete) && (kt != RebootProcessAndDelete) && (!processesLeft.validate(tLogPolicy))) {
else if ((kt != RebootAndDelete) && (kt != RebootProcessAndDelete) && notEnoughLeft) {
newKt = (g_random->random01() < 0.33) ? RebootAndDelete : Reboot;
canSurvive = false;
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("tLogPolicy", tLogPolicy->info()).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("RemainingZones", ::describeZones(localitiesLeft)).detail("RemainingDataHalls", ::describeDataHalls(localitiesLeft)).detail("Reason", "tLogPolicy does not validates against remaining processes.");
}
else if ((kt != RebootAndDelete) && (kt != RebootProcessAndDelete) && (!processesLeft.validate(storagePolicy))) {
newKt = (g_random->random01() < 0.33) ? RebootAndDelete : Reboot;
canSurvive = false;
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("storagePolicy", storagePolicy->info()).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("RemainingZones", ::describeZones(localitiesLeft)).detail("RemainingDataHalls", ::describeDataHalls(localitiesLeft)).detail("Reason", "storagePolicy does not validates against remaining processes.");
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("tLogPolicy", tLogPolicy->info()).detail("Reason", "tLogPolicy does not validates against remaining processes.");
}
else if ((kt != RebootAndDelete) && (kt != RebootProcessAndDelete) && (nQuorum > uniqueMachines.size())) {
newKt = (g_random->random01() < 0.33) ? RebootAndDelete : Reboot;
canSurvive = false;
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("storagePolicy", storagePolicy->info()).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("RemainingZones", ::describeZones(localitiesLeft)).detail("RemainingDataHalls", ::describeDataHalls(localitiesLeft)).detail("Quorum", nQuorum).detail("Machines", uniqueMachines.size()).detail("Reason", "Not enough unique machines to perform auto configuration of coordinators.");
TraceEvent("KillChanged").detail("KillType", kt).detail("NewKillType", newKt).detail("storagePolicy", storagePolicy->info()).detail("Quorum", nQuorum).detail("Machines", uniqueMachines.size()).detail("Reason", "Not enough unique machines to perform auto configuration of coordinators.");
}
else {
TraceEvent("CanSurviveKills").detail("KillType", kt).detail("ProcessesLeft", processesLeft.size()).detail("ProcessesDead", processesDead.size()).detail("DeadZones", ::describeZones(localitiesDead)).detail("DeadDataHalls", ::describeDataHalls(localitiesDead)).detail("tLogPolicy", tLogPolicy->info()).detail("storagePolicy", storagePolicy->info()).detail("Quorum", nQuorum).detail("Machines", uniqueMachines.size()).detail("ZonesLeft", ::describeZones(localitiesLeft)).detail("DataHallsLeft", ::describeDataHalls(localitiesLeft)).detail("ValidateRemaining", processesLeft.validate(tLogPolicy));
TraceEvent("CanSurviveKills").detail("KillType", kt).detail("tLogPolicy", tLogPolicy->info()).detail("storagePolicy", storagePolicy->info()).detail("Quorum", nQuorum).detail("Machines", uniqueMachines.size());
}
}
if (newKillType) *newKillType = newKt;

12
fdbrpc/simulator.h
View File

@ -286,7 +286,17 @@ public:
class ClusterConnectionString* extraDB;
IRepPolicyRef storagePolicy;
IRepPolicyRef tLogPolicy;
int tLogWriteAntiQuorum;
int32_t tLogWriteAntiQuorum;
Optional<Standalone<StringRef>> primaryDcId;
bool hasRemoteReplication;
IRepPolicyRef remoteTLogPolicy;
Optional<Standalone<StringRef>> remoteDcId;
IRepPolicyRef remoteStoragePolicy;
bool hasSatelliteReplication;
IRepPolicyRef satelliteTLogPolicy;
int32_t satelliteTLogWriteAntiQuorum;
std::vector<Optional<Standalone<StringRef>>> primarySatelliteDcIds;
std::vector<Optional<Standalone<StringRef>>> remoteSatelliteDcIds;
//Used by workloads that perform reconfigurations
int testerCount;

5
fdbserver/ClusterController.actor.cpp
View File

@ -652,6 +652,7 @@ public:
} else {
satelliteDCs.insert( req.configuration.remoteSatelliteDcIds.begin(), req.configuration.remoteSatelliteDcIds.end() );
}
//FIXME: recruitment does not respect usable_dcs, a.k.a if usable_dcs is 1 we should recruit all tlogs in one data center
auto satelliteLogs = getWorkersForTlogs( req.configuration, req.configuration.satelliteTLogReplicationFactor, req.configuration.getDesiredSatelliteLogs(), req.configuration.satelliteTLogPolicy, id_used, false, satelliteDCs );
for(int i = 0; i < satelliteLogs.size(); i++) {
@ -1512,7 +1513,7 @@ ACTOR Future<Void> timeKeeperSetVersion(ClusterControllerData *self) {
}
}
} catch (Error & e) {
TraceEvent(SevWarnAlways, "TimeKeeperSetupVersionFailed").detail("cause", e.what());
TraceEvent(SevWarnAlways, "TimeKeeperSetupVersionFailed").error(e);
}
return Void();
@ -1558,7 +1559,7 @@ ACTOR Future<Void> timeKeeper(ClusterControllerData *self) {
}
} catch (Error &e) {
// Failed to update time-version map even after retries, just ignore this iteration
TraceEvent(SevWarn, "TimeKeeperFailed").detail("cause", e.what());
TraceEvent(SevWarn, "TimeKeeperFailed").error(e);
}
Void _ = wait(delay(SERVER_KNOBS->TIME_KEEPER_DELAY));

25
fdbserver/LogSystemPeekCursor.actor.cpp
View File

@ -464,7 +464,7 @@ ArenaReader* ILogSystem::SetPeekCursor::reader() { return serverCursors[currentS
void ILogSystem::SetPeekCursor::calcHasMessage() {
if(bestSet >= 0 && bestServer >= 0) {
if(nextVersion.present()) {
//TraceEvent("LPC_calcNext").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage).detail("nextVersion", nextVersion.get().toString());
//TraceEvent("LPC_calcNext").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage).detail("nextVersion", nextVersion.get().toString());
serverCursors[bestSet][bestServer]->advanceTo( nextVersion.get() );
}
if( serverCursors[bestSet][bestServer]->hasMessage() ) {
@ -473,7 +473,7 @@ void ILogSystem::SetPeekCursor::calcHasMessage() {
currentCursor = bestServer;
hasNextMessage = true;
//TraceEvent("LPC_calc1").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage);
//TraceEvent("LPC_calc1").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage);
for (auto& cursors : serverCursors) {
for(auto& c : cursors) {
@ -496,10 +496,10 @@ void ILogSystem::SetPeekCursor::calcHasMessage() {
if(useBestSet) {
updateMessage(bestSet, false); // Use Quorum logic
//TraceEvent("LPC_calc2").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage);
//TraceEvent("LPC_calc2").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage);
if(!hasNextMessage) {
updateMessage(bestSet, true);
//TraceEvent("LPC_calc3").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage);
//TraceEvent("LPC_calc3").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage);
}
} else {
for(int i = 0; i < logSets.size() && !hasNextMessage; i++) {
@ -507,13 +507,13 @@ void ILogSystem::SetPeekCursor::calcHasMessage() {
updateMessage(i, false); // Use Quorum logic
}
}
//TraceEvent("LPC_calc4").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage);
//TraceEvent("LPC_calc4").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage);
for(int i = 0; i < logSets.size() && !hasNextMessage; i++) {
if(i != bestSet) {
updateMessage(i, true);
}
}
//TraceEvent("LPC_calc5").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage);
//TraceEvent("LPC_calc5").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage);
}
}
@ -525,7 +525,7 @@ void ILogSystem::SetPeekCursor::updateMessage(int logIdx, bool usePolicy) {
auto& serverCursor = serverCursors[logIdx][i];
if (nextVersion.present()) serverCursor->advanceTo(nextVersion.get());
sortedVersions.push_back(std::pair<LogMessageVersion, int>(serverCursor->version(), i));
//TraceEvent("LPC_update1").detail("ver", messageVersion.toString()).detail("tag", tag).detail("hasNextMessage", hasNextMessage).detail("serverVer", serverCursor->version().toString()).detail("i", i);
//TraceEvent("LPC_update1").detail("ver", messageVersion.toString()).detail("tag", tag.toString()).detail("hasNextMessage", hasNextMessage).detail("serverVer", serverCursor->version().toString()).detail("i", i);
}
if(usePolicy) {
@ -622,15 +622,22 @@ ACTOR Future<Void> setPeekGetMore(ILogSystem::SetPeekCursor* self, LogMessageVer
bestSetValid = self->localityGroup.size() < self->logSets[self->bestSet]->tLogReplicationFactor || !self->localityGroup.validate(self->logSets[self->bestSet]->tLogPolicy);
}
if(bestSetValid) {
//TraceEvent("LPC_getMore3", self->randomID).detail("start", startVersion.toString()).detail("t", self->tag);
if(!self->useBestSet) {
self->useBestSet = true;
self->calcHasMessage();
if (self->hasMessage() || self->version() > startVersion)
return Void();
}
//TraceEvent("LPC_getMore3", self->randomID).detail("start", startVersion.toString()).detail("t", self->tag.toString()).detail("bestSetSize", self->serverCursors[self->bestSet].size());
vector<Future<Void>> q;
for (auto& c : self->serverCursors[self->bestSet]) {
if (!c->hasMessage()) {
q.push_back(c->getMore());
q.push_back(c->onFailed());
}
}
Void _ = wait(quorum(q, 1));
self->useBestSet = true;
} else {
//FIXME: this will peeking way too many cursors when satellites exist, and does not need to peek bestSet cursors since we cannot get anymore data from them
vector<Future<Void>> q;

33
fdbserver/masterserver.actor.cpp
View File

@ -80,15 +80,18 @@ public:
Promise<Void> fullyRecovered;
DBCoreState prevDBState;
DBCoreState myDBState;
bool finalWriteStarted;
Future<Void> previousWrite;
ReusableCoordinatedState( ServerCoordinators const& coordinators, PromiseStream<Future<Void>> const& addActor, UID const& dbgid ) : coordinators(coordinators), cstate(coordinators), addActor(addActor), dbgid(dbgid) {}
ReusableCoordinatedState( ServerCoordinators const& coordinators, PromiseStream<Future<Void>> const& addActor, UID const& dbgid ) : coordinators(coordinators), cstate(coordinators), addActor(addActor), dbgid(dbgid), finalWriteStarted(false), previousWrite(Void()) {}
Future<Void> read() {
return _read(this);
}
Future<Void> write(DBCoreState newState, bool finalWrite = false) {
return _write(this, newState, finalWrite);
previousWrite = _write(this, newState, finalWrite);
return previousWrite;
}
Future<Void> move( ClusterConnectionString const& nc ) {
@ -104,7 +107,11 @@ private:
ACTOR Future<Void> _read(ReusableCoordinatedState* self) {
Value prevDBStateRaw = wait( self->cstate.read() );
self->addActor.send( masterTerminateOnConflict( self->dbgid, self->fullyRecovered, self->cstate.onConflict(), self->switchedState.getFuture() ) );
Future<Void> onConflict = masterTerminateOnConflict( self->dbgid, self->fullyRecovered, self->cstate.onConflict(), self->switchedState.getFuture() );
if(onConflict.isReady() && onConflict.isError()) {
throw onConflict.getError();
}
self->addActor.send( onConflict );
if( prevDBStateRaw.size() ) {
self->prevDBState = BinaryReader::fromStringRef<DBCoreState>(prevDBStateRaw, IncludeVersion());
@ -115,6 +122,14 @@ private:
}
ACTOR Future<Void> _write(ReusableCoordinatedState* self, DBCoreState newState, bool finalWrite) {
if(self->finalWriteStarted) {
Void _ = wait( Future<Void>(Never()) );
}
if(finalWrite) {
self->finalWriteStarted = true;
}
try {
Void _ = wait( self->cstate.setExclusive( BinaryWriter::toValue(newState, IncludeVersion()) ) );
} catch (Error& e) {
@ -931,11 +946,19 @@ static std::set<int> const& normalMasterErrors() {
}
ACTOR Future<Void> changeCoordinators( Reference<MasterData> self ) {
Void _ = wait( self->cstate.fullyRecovered.getFuture() );
loop {
ChangeCoordinatorsRequest req = waitNext( self->myInterface.changeCoordinators.getFuture() );
state ChangeCoordinatorsRequest changeCoordinatorsRequest = req;
while( !self->cstate.previousWrite.isReady() ) {
Void _ = wait( self->cstate.previousWrite );
Void _ = wait( delay(0) ); //if a new core state is ready to be written, have that take priority over our finalizing write;
}
if(!self->cstate.fullyRecovered.isSet()) {
Void _ = wait( self->cstate.write(self->cstate.myDBState, true) );
}
try {
Void _ = wait( self->cstate.move( ClusterConnectionString( changeCoordinatorsRequest.newConnectionString.toString() ) ) );
}

39
fdbserver/storageserver.actor.cpp
View File

@ -336,6 +336,8 @@ public:
AsyncMap<Key,bool> watches;
int64_t watchBytes;
AsyncVar<bool> noRecentUpdates;
double lastUpdate;
Int64MetricHandle readQueueSizeMetric;
@ -424,7 +426,7 @@ public:
debug_inApplyUpdate(false), debug_lastValidateTime(0), watchBytes(0),
logProtocol(0), counters(this), tag(invalidTag), maxQueryQueue(0), thisServerID(ssi.id()),
readQueueSizeMetric(LiteralStringRef("StorageServer.ReadQueueSize")),
behind(false), byteSampleClears(false, LiteralStringRef("\xff\xff\xff"))
behind(false), byteSampleClears(false, LiteralStringRef("\xff\xff\xff")), noRecentUpdates(false), lastUpdate(now())
{
version.initMetric(LiteralStringRef("StorageServer.Version"), counters.cc.id);
oldestVersion.initMetric(LiteralStringRef("StorageServer.OldestVersion"), counters.cc.id);
@ -784,13 +786,27 @@ ACTOR Future<Void> watchValue_impl( StorageServer* data, WatchValueRequest req )
}
ACTOR Future<Void> watchValueQ( StorageServer* data, WatchValueRequest req ) {
choose {
when( Void _ = wait( watchValue_impl( data, req ) ) ) {}
when( Void _ = wait( BUGGIFY ? Never() : delay( g_network->isSimulated() ? 20 : 900 ) ) ) {
req.reply.sendError( timed_out() );
state Future<Void> watch = watchValue_impl( data, req );
state double startTime = now();
loop {
double timeoutDelay = -1;
if(data->noRecentUpdates.get()) {
timeoutDelay = std::max(CLIENT_KNOBS->FAST_WATCH_TIMEOUT - (now() - startTime), 0.0);
} else if(!BUGGIFY) {
timeoutDelay = std::max(CLIENT_KNOBS->WATCH_TIMEOUT - (now() - startTime), 0.0);
}
choose {
when( Void _ = wait( watch ) ) {
return Void();
}
when( Void _ = wait( timeoutDelay < 0 ? Never() : delay(timeoutDelay) ) ) {
req.reply.sendError( timed_out() );
return Void();
}
when( Void _ = wait( data->noRecentUpdates.onChange()) ) {}
}
}
return Void();
}
ACTOR Future<Void> getShardState_impl( StorageServer* data, GetShardStateRequest req ) {
@ -2445,6 +2461,8 @@ ACTOR Future<Void> update( StorageServer* data, bool* pReceivedUpdate )
data->mutableData().createNewVersion(ver);
if (data->otherError.getFuture().isReady()) data->otherError.getFuture().get();
data->noRecentUpdates.set(false);
data->lastUpdate = now();
data->version.set( ver ); // Triggers replies to waiting gets for new version(s)
if (data->otherError.getFuture().isReady()) data->otherError.getFuture().get();
@ -3045,6 +3063,7 @@ ACTOR Future<Void> storageServerCore( StorageServer* self, StorageServerInterfac
state ActorCollection actors(false);
state double lastLoopTopTime = now();
state Future<Void> dbInfoChange = Void();
state Future<Void> checkLastUpdate = Void();
actors.add(updateStorage(self));
actors.add(waitFailureServer(ssi.waitFailure.getFuture()));
@ -3066,6 +3085,14 @@ ACTOR Future<Void> storageServerCore( StorageServer* self, StorageServerInterfac
lastLoopTopTime = loopTopTime;
choose {
when( Void _ = wait( checkLastUpdate ) ) {
if(now() - self->lastUpdate >= CLIENT_KNOBS->NO_RECENT_UPDATES_DURATION) {
self->noRecentUpdates.set(true);
checkLastUpdate = delay(CLIENT_KNOBS->NO_RECENT_UPDATES_DURATION);
} else {
checkLastUpdate = delay( std::max(CLIENT_KNOBS->NO_RECENT_UPDATES_DURATION-(now()-self->lastUpdate), 0.1) );
}
}
when( Void _ = wait( dbInfoChange ) ) {
TEST( self->logSystem ); // shardServer dbInfo changed
dbInfoChange = self->db->onChange();

2
fdbserver/workloads/BackupCorrectness.actor.cpp
View File

@ -447,7 +447,7 @@ struct BackupAndRestoreCorrectnessWorkload : TestWorkload {
TraceEvent("BARW_NonzeroTaskWait", randomID).detail("backupTag", printable(self->backupTag)).detail("taskCount", taskCount).detail("waitCycles", waitCycles);
printf("%.6f %-10s Wait #%4d for %lld tasks to end\n", now(), randomID.toString().c_str(), waitCycles, (long long) taskCount);
Void _ = wait(delay(20.0));
Void _ = wait(delay(5.0));
tr->commit();
tr = Reference<ReadYourWritesTransaction>(new ReadYourWritesTransaction(cx));
int64_t _taskCount = wait( backupAgent.getTaskCount(tr) );

2
fdbserver/workloads/BackupToDBCorrectness.actor.cpp
View File

@ -344,7 +344,7 @@ struct BackupToDBCorrectnessWorkload : TestWorkload {
TraceEvent("BARW_NonzeroTaskWait", randomID).detail("backupTag", printable(tag)).detail("taskCount", taskCount).detail("waitCycles", waitCycles);
printf("%.6f %-10s Wait #%4d for %lld tasks to end\n", now(), randomID.toString().c_str(), waitCycles, (long long) taskCount);
Void _ = wait(delay(20.0));
Void _ = wait(delay(5.0));
tr->commit();
tr = Reference<ReadYourWritesTransaction>(new ReadYourWritesTransaction(cx));
int64_t _taskCount = wait( backupAgent->getTaskCount(tr) );

5
flow/Knobs.cpp
View File

@ -138,7 +138,10 @@ FlowKnobs::FlowKnobs(bool randomize, bool isSimulated) {
init( ALTERNATIVES_FAILURE_RESET_TIME, 5.0 );
init( ALTERNATIVES_FAILURE_MAX_DELAY, 1.0 );
init( ALTERNATIVES_FAILURE_MIN_DELAY, 0.05 );
init( ALTERNATIVES_FAILURE_DELAY_RATIO, 0.2 );
init( ALTERNATIVES_FAILURE_DELAY_RATIO, 0.2 );
init( FUTURE_VERSION_INITIAL_BACKOFF, 1.0 );
init( FUTURE_VERSION_MAX_BACKOFF, 8.0 );
init( FUTURE_VERSION_BACKOFF_GROWTH, 2.0 );
}
bool Knobs::setKnob( std::string const& knob, std::string const& value ) {

3
flow/Knobs.h
View File

@ -159,6 +159,9 @@ public:
double ALTERNATIVES_FAILURE_MAX_DELAY;
double ALTERNATIVES_FAILURE_MIN_DELAY;
double ALTERNATIVES_FAILURE_DELAY_RATIO;
double FUTURE_VERSION_INITIAL_BACKOFF;
double FUTURE_VERSION_MAX_BACKOFF;
double FUTURE_VERSION_BACKOFF_GROWTH;
FlowKnobs(bool randomize = false, bool isSimulated = false);
};