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

added timeout and shard limiting behaviour

This commit is contained in:
Jon Fu 2019-05-17 16:11:50 -07:00
parent 0984d272e1
commit e339ab890b
9 changed files with 39 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
fdbclient/MasterProxyInterface.h
View File

@ -315,14 +315,15 @@ struct GetDDMetricsReply
struct GetDDMetricsRequest {
constexpr static FileIdentifier file_identifier = 14536812;
KeyRange keys;
int shardLimit;
ReplyPromise<struct GetDDMetricsReply> reply;
GetDDMetricsRequest() {}
explicit GetDDMetricsRequest(KeyRange const& keys) : keys(keys) {}
explicit GetDDMetricsRequest(KeyRange const& keys, const int shardLimit) : keys(keys), shardLimit(shardLimit) {}
template<class Ar>
void serialize(Ar& ar) {
serializer(ar, keys, reply);
serializer(ar, keys, shardLimit, reply);
}
};

8
fdbclient/NativeAPI.actor.cpp
View File

@ -3194,9 +3194,11 @@ ACTOR Future< Standalone<RangeResultRef> > waitStorageMetricsList(
KeyRange keys,
int shardLimit )
{
// TODO: add use for or remove shardLimit
GetDDMetricsReply rep = wait(loadBalance(cx->getMasterProxies(false), &MasterProxyInterface::getDDMetrics, GetDDMetricsRequest(keys)));
return rep.storageMetricsList;
ErrorOr<GetDDMetricsReply> rep = wait(errorOr(loadBalance(cx->getMasterProxies(false), &MasterProxyInterface::getDDMetrics, GetDDMetricsRequest(keys, shardLimit))));
if (rep.isError()) {
throw rep.getError();
}
return rep.get().storageMetricsList;
}
Future< Standalone<RangeResultRef> > Transaction::getStorageMetricsList(KeyRange const& keys, int shardLimit) {

7
fdbclient/ReadYourWrites.actor.cpp
View File

@ -1279,8 +1279,11 @@ Future< Standalone<RangeResultRef> > ReadYourWritesTransaction::getRange(
end.getKey().startsWith(stats_prefix)) {
if (tr.getDatabase().getPtr() && tr.getDatabase()->getConnectionFile()) {
auto keys = KeyRangeRef(begin.getKey(), end.getKey()).removePrefix(stats_prefix);
// TODO: find appropriate use for shardLimit, currently not used
return tr.getStorageMetricsList(keys, 10);
try {
return tr.getStorageMetricsList(keys, CLIENT_KNOBS->STORAGE_METRICS_SHARD_LIMIT);
} catch( Error &e ) {
return e;
}
}
else {
return Standalone<RangeResultRef>();

1
fdbserver/ClusterController.actor.cpp
View File

@ -57,7 +57,6 @@ struct WorkerInfo : NonCopyable {
WorkerDetails details;
Future<Void> haltRatekeeper;
Future<Void> haltDistributor;
Future<GetDataDistributorMetricsReply> ddMetrics;
WorkerInfo() : gen(-1), reboots(0), lastAvailableTime(now()), priorityInfo(ProcessClass::UnsetFit, false, ClusterControllerPriorityInfo::FitnessUnknown) {}
WorkerInfo( Future<Void> watcher, ReplyPromise<RegisterWorkerReply> reply, Generation gen, WorkerInterface interf, ProcessClass initialClass, ProcessClass processClass, ClusterControllerPriorityInfo priorityInfo, bool degraded ) :

12
fdbserver/DataDistribution.actor.cpp
View File

@ -3740,10 +3740,14 @@ ACTOR Future<Void> dataDistributor(DataDistributorInterface di, Reference<AsyncV
break;
}
when ( state GetDataDistributorMetricsRequest req = waitNext(di.dataDistributorMetrics.getFuture()) ) {
Standalone<RangeResultRef> result = wait(brokenPromiseToNever(getShardMetricsList.getReply( GetMetricsListRequest(req.keys))));
GetDataDistributorMetricsReply rep;
rep.storageMetricsList = result;
req.reply.send(rep);
ErrorOr<Standalone<RangeResultRef>> result = wait(errorOr(brokenPromiseToNever(getShardMetricsList.getReply( GetMetricsListRequest(req.keys, req.shardLimit)))));
if ( result.isError() ) {
req.reply.sendError(result.getError());
} else {
GetDataDistributorMetricsReply rep;
rep.storageMetricsList = result.get();
req.reply.send(rep);
}
}
}
}

3
fdbserver/DataDistribution.actor.h
View File

@ -123,10 +123,11 @@ struct GetMetricsRequest {
struct GetMetricsListRequest {
KeyRange keys;
int shardLimit;
Promise< Standalone<RangeResultRef> > reply;
GetMetricsListRequest() {}
GetMetricsListRequest( KeyRange const& keys ) : keys(keys) {}
GetMetricsListRequest( KeyRange const& keys, const int shardLimit ) : keys(keys), shardLimit(shardLimit) {}
};
struct TeamCollectionInterface {

7
fdbserver/DataDistributionTracker.actor.cpp
View File

@ -655,7 +655,7 @@ ACTOR Future<Void> fetchShardMetricsList_impl( DataDistributionTracker* self, Ge
try {
loop {
// no shard identifier as of yet, use simple numbering system
int shardNum = 0;
int shardNum = 1;
// list of metrics, regenerate on loop when full range unsuccessful
Standalone<RangeResultRef> result;
Future<Void> onChange;
@ -673,6 +673,9 @@ ACTOR Future<Void> fetchShardMetricsList_impl( DataDistributionTracker* self, Ge
)
);
++shardNum;
if ( shardNum > req.shardLimit ) {
break;
}
}
if( !onChange.isValid() ) {
@ -693,7 +696,7 @@ ACTOR Future<Void> fetchShardMetricsList( DataDistributionTracker* self, GetMetr
choose {
when( wait( fetchShardMetricsList_impl( self, req ) ) ) {}
when( wait( delay( SERVER_KNOBS->DD_SHARD_METRICS_TIMEOUT ) ) ) {
// TODO: implement proper behaviour on timeout
req.reply.sendError(timed_out());
}
}
return Void();

5
fdbserver/DataDistributorInterface.h
View File

@ -80,14 +80,15 @@ struct GetDataDistributorMetricsReply {
struct GetDataDistributorMetricsRequest {
constexpr static FileIdentifier file_identifier = 1059267;
KeyRange keys;
int shardLimit;
ReplyPromise<struct GetDataDistributorMetricsReply> reply;
GetDataDistributorMetricsRequest() {}
explicit GetDataDistributorMetricsRequest(KeyRange const& keys) : keys(keys) {}
explicit GetDataDistributorMetricsRequest(KeyRange const& keys, const int shardLimit) : keys(keys), shardLimit(shardLimit) {}
template<class Ar>
void serialize(Ar& ar) {
serializer(ar, keys, reply);
serializer(ar, keys, shardLimit, reply);
}
};

12
fdbserver/MasterProxyServer.actor.cpp
View File

@ -1335,10 +1335,14 @@ ACTOR Future<Void> ddMetricsRequestServer(MasterProxyInterface proxy, Reference<
choose {
when(state GetDDMetricsRequest req = waitNext(proxy.getDDMetrics.getFuture()))
{
GetDataDistributorMetricsReply reply = wait(db->get().distributor.get().dataDistributorMetrics.getReply(GetDataDistributorMetricsRequest(req.keys)));
GetDDMetricsReply newReply;
newReply.storageMetricsList = reply.storageMetricsList;
req.reply.send(newReply);
ErrorOr<GetDataDistributorMetricsReply> reply = wait(errorOr(db->get().distributor.get().dataDistributorMetrics.getReply(GetDataDistributorMetricsRequest(req.keys, req.shardLimit))));
if ( reply.isError() ) {
req.reply.sendError(reply.getError());
} else {
GetDDMetricsReply newReply;
newReply.storageMetricsList = reply.get().storageMetricsList;
req.reply.send(newReply);
}
}
}
}