merge release 5.2 into master
This commit is contained in:
Evan Tschannen
commit
520aaf731d
|
|
@ -12,7 +12,7 @@ Features
|
|||
* Added a TLS plugin implementation. `(PR #343) <https://github.com/apple/foundationdb/pull/343>`_
|
||||
* Backup supports HTTPS for blobstore connections. `(PR #343) <https://github.com/apple/foundationdb/pull/343>`_
|
||||
* Added the APPEND_IF_FITS atomic operation. `(PR #22) <https://github.com/apple/foundationdb/pull/22>`_
|
||||
* Updated the SET_VERSIONSTAMPED_VALUE atomic operation to place the version stamp at a specified offset in a value. `(Issue #148) <https://github.com/apple/foundationdb/issues/148>`_
|
||||
* Updated the SET_VERSIONSTAMPED_VALUE atomic operation to place the versionstamp at a specified offset in a value. `(Issue #148) <https://github.com/apple/foundationdb/issues/148>`_
|
||||
|
||||
Performance
|
||||
-----------
|
||||
|
|
|
|||
|
|
@ -1022,57 +1022,8 @@ ACTOR Future<Optional<vector<StorageServerInterface>>> transactionalGetServerInt
|
|||
return serverInterfaces;
|
||||
}
|
||||
|
||||
template <class F>
|
||||
pair<KeyRange, Reference<LocationInfo>> getCachedKeyLocation( Database cx, Key key, F StorageServerInterface::*member, bool isBackward = false ) {
|
||||
auto ssi = cx->getCachedLocation( key, isBackward );
|
||||
if (!ssi.second) {
|
||||
return ssi;
|
||||
}
|
||||
|
||||
for(int i = 0; i < ssi.second->size(); i++) {
|
||||
if( IFailureMonitor::failureMonitor().onlyEndpointFailed(ssi.second->get(i, member).getEndpoint()) ) {
|
||||
cx->invalidateCache( key );
|
||||
ssi.second.clear();
|
||||
return ssi;
|
||||
}
|
||||
}
|
||||
|
||||
return ssi;
|
||||
}
|
||||
|
||||
template <class F>
|
||||
vector< pair<KeyRange,Reference<LocationInfo>> > getCachedKeyRangeLocations( Database cx, KeyRange keys, int limit, bool reverse, F StorageServerInterface::*member ) {
|
||||
ASSERT (!keys.empty());
|
||||
|
||||
vector< pair<KeyRange,Reference<LocationInfo>> > locations;
|
||||
if (!cx->getCachedLocations(keys, locations, limit, reverse))
|
||||
return vector< pair<KeyRange,Reference<LocationInfo>> >();
|
||||
|
||||
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, member).getEndpoint()) ) {
|
||||
onlyEndpointFailed = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if( onlyEndpointFailed ) {
|
||||
cx->invalidateCache( it.first.begin );
|
||||
foundFailed = true;
|
||||
}
|
||||
}
|
||||
|
||||
if(foundFailed) {
|
||||
return vector< pair<KeyRange,Reference<LocationInfo>> >();
|
||||
}
|
||||
|
||||
return locations;
|
||||
}
|
||||
|
||||
//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_internal( Database cx, Key key, TransactionInfo info, bool isBackward = false ) {
|
||||
if (isBackward) {
|
||||
ASSERT( key != allKeys.begin && key <= allKeys.end );
|
||||
} else {
|
||||
|
|
@ -1097,7 +1048,25 @@ ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation( Database
|
|||
}
|
||||
}
|
||||
|
||||
ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations( Database cx, KeyRange keys, int limit, bool reverse, TransactionInfo info ) {
|
||||
template <class F>
|
||||
Future<pair<KeyRange, Reference<LocationInfo>>> getKeyLocation( Database const& cx, Key const& key, F StorageServerInterface::*member, TransactionInfo const& info, bool isBackward = false ) {
|
||||
auto ssi = cx->getCachedLocation( key, isBackward );
|
||||
if (!ssi.second) {
|
||||
return getKeyLocation_internal( cx, key, info, isBackward );
|
||||
}
|
||||
|
||||
for(int i = 0; i < ssi.second->size(); i++) {
|
||||
if( IFailureMonitor::failureMonitor().onlyEndpointFailed(ssi.second->get(i, member).getEndpoint()) ) {
|
||||
cx->invalidateCache( key );
|
||||
ssi.second.clear();
|
||||
return getKeyLocation_internal( cx, key, info, isBackward );
|
||||
}
|
||||
}
|
||||
|
||||
return ssi;
|
||||
}
|
||||
|
||||
ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations_internal( Database cx, KeyRange keys, int limit, bool reverse, TransactionInfo info ) {
|
||||
if( info.debugID.present() )
|
||||
g_traceBatch.addEvent("TransactionDebug", info.debugID.get().first(), "NativeAPI.getKeyLocations.Before");
|
||||
|
||||
|
|
@ -1124,11 +1093,43 @@ ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLoca
|
|||
}
|
||||
}
|
||||
|
||||
template <class F>
|
||||
Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLocations( Database const& cx, KeyRange const& keys, int limit, bool reverse, F StorageServerInterface::*member, TransactionInfo const& info ) {
|
||||
ASSERT (!keys.empty());
|
||||
|
||||
vector< pair<KeyRange,Reference<LocationInfo>> > locations;
|
||||
if (!cx->getCachedLocations(keys, locations, limit, reverse)) {
|
||||
return getKeyRangeLocations_internal( cx, keys, limit, reverse, info );
|
||||
}
|
||||
|
||||
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, member).getEndpoint()) ) {
|
||||
onlyEndpointFailed = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if( onlyEndpointFailed ) {
|
||||
cx->invalidateCache( it.first.begin );
|
||||
foundFailed = true;
|
||||
}
|
||||
}
|
||||
|
||||
if(foundFailed) {
|
||||
return getKeyRangeLocations_internal( cx, keys, limit, reverse, info );
|
||||
}
|
||||
|
||||
return locations;
|
||||
}
|
||||
|
||||
ACTOR Future<Void> warmRange_impl( Transaction *self, Database cx, KeyRange keys ) {
|
||||
state int totalRanges = 0;
|
||||
state int totalRequests = 0;
|
||||
loop {
|
||||
vector<pair<KeyRange, Reference<LocationInfo>>> locations = wait(getKeyRangeLocations(cx, keys, CLIENT_KNOBS->WARM_RANGE_SHARD_LIMIT, false, self->info));
|
||||
vector<pair<KeyRange, Reference<LocationInfo>>> locations = wait(getKeyRangeLocations_internal(cx, keys, CLIENT_KNOBS->WARM_RANGE_SHARD_LIMIT, false, self->info));
|
||||
totalRanges += CLIENT_KNOBS->WARM_RANGE_SHARD_LIMIT;
|
||||
totalRequests++;
|
||||
if(locations.size() == 0 || totalRanges >= cx->locationCacheSize || locations[locations.size()-1].first.end >= keys.end)
|
||||
|
|
@ -1165,12 +1166,7 @@ ACTOR Future<Optional<Value>> getValue( Future<Version> version, Key key, Databa
|
|||
validateVersion(ver);
|
||||
|
||||
loop {
|
||||
state pair<KeyRange, Reference<LocationInfo>> ssi = getCachedKeyLocation(cx, key, &StorageServerInterface::getValue);
|
||||
if (!ssi.second) {
|
||||
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info ) );
|
||||
ssi = std::move(ssi2);
|
||||
}
|
||||
|
||||
state pair<KeyRange, Reference<LocationInfo>> ssi = wait( getKeyLocation(cx, key, &StorageServerInterface::getValue, info) );
|
||||
state Optional<UID> getValueID = Optional<UID>();
|
||||
state uint64_t startTime;
|
||||
state double startTimeD;
|
||||
|
|
@ -1249,11 +1245,7 @@ ACTOR Future<Key> getKey( Database cx, KeySelector k, Future<Version> version, T
|
|||
}
|
||||
|
||||
Key locationKey(k.getKey(), k.arena());
|
||||
state pair<KeyRange, Reference<LocationInfo>> ssi = getCachedKeyLocation(cx, locationKey, &StorageServerInterface::getKey, k.isBackward());
|
||||
if(!ssi.second) {
|
||||
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation(cx, locationKey, info, k.isBackward()) );
|
||||
ssi = std::move(ssi2);
|
||||
}
|
||||
state pair<KeyRange, Reference<LocationInfo>> ssi = wait( getKeyLocation(cx, locationKey, &StorageServerInterface::getKey, info, k.isBackward()) );
|
||||
|
||||
try {
|
||||
if( info.debugID.present() )
|
||||
|
|
@ -1312,11 +1304,7 @@ ACTOR Future< Void > watchValue( Future<Version> version, Key key, Optional<Valu
|
|||
ASSERT(ver != latestVersion);
|
||||
|
||||
loop {
|
||||
state pair<KeyRange, Reference<LocationInfo>> ssi = getCachedKeyLocation(cx, key, &StorageServerInterface::watchValue );
|
||||
if (!ssi.second) {
|
||||
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, key, info ) );
|
||||
ssi = std::move(ssi2);
|
||||
}
|
||||
state pair<KeyRange, Reference<LocationInfo>> ssi = wait( getKeyLocation(cx, key, &StorageServerInterface::watchValue, info ) );
|
||||
|
||||
try {
|
||||
state Optional<UID> watchValueID = Optional<UID>();
|
||||
|
|
@ -1387,12 +1375,7 @@ 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 = getCachedKeyRangeLocations( cx, keys, CLIENT_KNOBS->GET_RANGE_SHARD_LIMIT, reverse, &StorageServerInterface::getKeyValues);
|
||||
if (!locations.size()) {
|
||||
vector< pair<KeyRange, Reference<LocationInfo>> > _locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->GET_RANGE_SHARD_LIMIT, reverse, info ) );
|
||||
locations = std::move(_locations);
|
||||
}
|
||||
|
||||
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->GET_RANGE_SHARD_LIMIT, reverse, &StorageServerInterface::getKeyValues, info ) );
|
||||
ASSERT( locations.size() );
|
||||
state int shard = 0;
|
||||
loop {
|
||||
|
|
@ -1649,12 +1632,7 @@ ACTOR Future<Standalone<RangeResultRef>> getRange( Database cx, Reference<Transa
|
|||
|
||||
Key locationKey = reverse ? Key(end.getKey(), end.arena()) : Key(begin.getKey(), begin.arena());
|
||||
bool locationBackward = reverse ? (end-1).isBackward() : begin.isBackward();
|
||||
state pair<KeyRange, Reference<LocationInfo>> beginServer = getCachedKeyLocation( cx, locationKey, &StorageServerInterface::getKeyValues, locationBackward );
|
||||
if (!beginServer.second) {
|
||||
pair<KeyRange, Reference<LocationInfo>> ssi2 = wait( getKeyLocation( cx, locationKey, info, locationBackward ) );
|
||||
beginServer = std::move(ssi2);
|
||||
}
|
||||
|
||||
state pair<KeyRange, Reference<LocationInfo>> beginServer = wait( getKeyLocation( cx, locationKey, &StorageServerInterface::getKeyValues, info, locationBackward ) );
|
||||
state KeyRange shard = beginServer.first;
|
||||
state bool modifiedSelectors = false;
|
||||
state GetKeyValuesRequest req;
|
||||
|
|
@ -2919,11 +2897,7 @@ ACTOR Future< StorageMetrics > waitStorageMetrics(
|
|||
{
|
||||
state int tooManyShardsCount = 0;
|
||||
loop {
|
||||
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = getCachedKeyRangeLocations( cx, keys, shardLimit, false, &StorageServerInterface::waitMetrics);
|
||||
if (!locations.size()) {
|
||||
vector< pair<KeyRange, Reference<LocationInfo>> > _locations = wait( getKeyRangeLocations( cx, keys, shardLimit, false, TransactionInfo(TaskDataDistribution) ) );
|
||||
locations = std::move(_locations);
|
||||
}
|
||||
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = wait( getKeyRangeLocations( cx, keys, shardLimit, false, &StorageServerInterface::waitMetrics, TransactionInfo(TaskDataDistribution) ) );
|
||||
|
||||
if( locations.size() == shardLimit ) {
|
||||
TraceEvent(!g_network->isSimulated() && ++tooManyShardsCount >= 15 ? SevWarnAlways : SevWarn, "WaitStorageMetricsPenalty")
|
||||
|
|
@ -2977,11 +2951,7 @@ Future< StorageMetrics > Transaction::getStorageMetrics( KeyRange const& keys, i
|
|||
ACTOR Future< Standalone<VectorRef<KeyRef>> > splitStorageMetrics( Database cx, KeyRange keys, StorageMetrics limit, StorageMetrics estimated )
|
||||
{
|
||||
loop {
|
||||
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = getCachedKeyRangeLocations( cx, keys, CLIENT_KNOBS->STORAGE_METRICS_SHARD_LIMIT, false, &StorageServerInterface::splitMetrics);
|
||||
if (!locations.size()) {
|
||||
vector< pair<KeyRange, Reference<LocationInfo>> > _locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->STORAGE_METRICS_SHARD_LIMIT, false, TransactionInfo(TaskDataDistribution) ) );
|
||||
locations = std::move(_locations);
|
||||
}
|
||||
state vector< pair<KeyRange, Reference<LocationInfo>> > locations = wait( getKeyRangeLocations( cx, keys, CLIENT_KNOBS->STORAGE_METRICS_SHARD_LIMIT, false, &StorageServerInterface::splitMetrics, TransactionInfo(TaskDataDistribution) ) );
|
||||
state StorageMetrics used;
|
||||
state Standalone<VectorRef<KeyRef>> results;
|
||||
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ BlobStoreEndpoint::Stats BlobStoreEndpoint::Stats::operator-(const Stats &rhs) {
|
|||
BlobStoreEndpoint::Stats BlobStoreEndpoint::s_stats;
|
||||
|
||||
BlobStoreEndpoint::BlobKnobs::BlobKnobs() {
|
||||
secure_connection = 1;
|
||||
connect_tries = CLIENT_KNOBS->BLOBSTORE_CONNECT_TRIES;
|
||||
connect_timeout = CLIENT_KNOBS->BLOBSTORE_CONNECT_TIMEOUT;
|
||||
max_connection_life = CLIENT_KNOBS->BLOBSTORE_MAX_CONNECTION_LIFE;
|
||||
|
|
@ -71,6 +72,7 @@ BlobStoreEndpoint::BlobKnobs::BlobKnobs() {
|
|||
|
||||
bool BlobStoreEndpoint::BlobKnobs::set(StringRef name, int value) {
|
||||
#define TRY_PARAM(n, sn) if(name == LiteralStringRef(#n) || name == LiteralStringRef(#sn)) { n = value; return true; }
|
||||
TRY_PARAM(secure_connection, sc)
|
||||
TRY_PARAM(connect_tries, ct);
|
||||
TRY_PARAM(connect_timeout, cto);
|
||||
TRY_PARAM(max_connection_life, mcl);
|
||||
|
|
@ -98,6 +100,7 @@ std::string BlobStoreEndpoint::BlobKnobs::getURLParameters() const {
|
|||
static BlobKnobs defaults;
|
||||
std::string r;
|
||||
#define _CHECK_PARAM(n, sn) if(n != defaults. n) { r += format("%s%s=%d", r.empty() ? "" : "&", #sn, n); }
|
||||
_CHECK_PARAM(secure_connection, sc);
|
||||
_CHECK_PARAM(connect_tries, ct);
|
||||
_CHECK_PARAM(connect_timeout, cto);
|
||||
_CHECK_PARAM(max_connection_life, mcl);
|
||||
|
|
@ -149,7 +152,7 @@ Reference<BlobStoreEndpoint> BlobStoreEndpoint::fromString(std::string const &ur
|
|||
StringRef value = t.eat("&");
|
||||
char *valueEnd;
|
||||
int ivalue = strtol(value.toString().c_str(), &valueEnd, 10);
|
||||
if(*valueEnd || ivalue == 0)
|
||||
if(*valueEnd || (ivalue == 0 && value.toString() != "0"))
|
||||
throw format("%s is not a valid value for %s", value.toString().c_str(), name.toString().c_str());
|
||||
if(!knobs.set(name, ivalue))
|
||||
throw format("%s is not a valid parameter name", name.toString().c_str());
|
||||
|
|
@ -393,8 +396,10 @@ ACTOR Future<BlobStoreEndpoint::ReusableConnection> connect_impl(Reference<BlobS
|
|||
return rconn;
|
||||
}
|
||||
}
|
||||
|
||||
state Reference<IConnection> conn = wait(INetworkConnections::net()->connect(b->host, b->service.empty() ? "https" : b->service, true));
|
||||
std::string service = b->service;
|
||||
if (service.empty())
|
||||
service = b->knobs.secure_connection ? "https" : "http";
|
||||
state Reference<IConnection> conn = wait(INetworkConnections::net()->connect(b->host, service, b->knobs.secure_connection ? true : false));
|
||||
|
||||
TraceEvent("BlobStoreEndpointNewConnection")
|
||||
.detail("RemoteEndpoint", conn->getPeerAddress())
|
||||
|
|
|
|||
|
|
@ -48,7 +48,8 @@ public:
|
|||
|
||||
struct BlobKnobs {
|
||||
BlobKnobs();
|
||||
int connect_tries,
|
||||
int secure_connection,
|
||||
connect_tries,
|
||||
connect_timeout,
|
||||
max_connection_life,
|
||||
request_tries,
|
||||
|
|
@ -70,6 +71,7 @@ public:
|
|||
std::string getURLParameters() const;
|
||||
static std::vector<std::string> getKnobDescriptions() {
|
||||
return {
|
||||
"secure_connection (or sc) Set 1 for secure connection and 0 for insecure connection.",
|
||||
"connect_tries (or ct) Number of times to try to connect for each request.",
|
||||
"connect_timeout (or cto) Number of seconds to wait for a connect request to succeed.",
|
||||
"max_connection_life (or mcl) Maximum number of seconds to use a single TCP connection.",
|
||||
|
|
|
|||
Loading…
Reference in New Issue