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

FastRestore:Revise trace events to be descriptive 

Revert changes that send mutations to appliers out of order
This commit is contained in:
Meng Xu 2020-04-22 12:12:48 -07:00
parent 38193a3866
commit f073049865
4 changed files with 26 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
fdbserver/RestoreApplier.actor.cpp
View File

@ -89,8 +89,8 @@ ACTOR Future<Void> restoreApplierCore(RestoreApplierInterface applierInterf, int
} }
} }
} catch (Error& e) { } catch (Error& e) {
TraceEvent(SevWarn, "FastRestore") TraceEvent(SevWarn, "FastRestoreApplierError", self->id())
.detail("RestoreLoaderError", e.what()) .detail("Error", e.what())
.detail("RequestType", requestTypeStr); .detail("RequestType", requestTypeStr);
break; break;
} }

8
fdbserver/RestoreCommon.actor.h
View File

@ -296,8 +296,9 @@ Future<Void> getBatchReplies(RequestStream<Request> Interface::*channel, std::ma
if (ongoingReplies.empty()) { if (ongoingReplies.empty()) {
break; break;
} else { } else {
wait(quorum(ongoingReplies, std::min((int)SERVER_KNOBS->FASTRESTORE_REQBATCH_PARALLEL, wait(waitForAny(ongoingReplies));
(int)ongoingReplies.size()))); // wait(quorum(ongoingReplies, std::min((int)SERVER_KNOBS->FASTRESTORE_REQBATCH_PARALLEL,
// (int)ongoingReplies.size())));
} }
// At least one reply is received; Calculate the reply duration // At least one reply is received; Calculate the reply duration
for (int j = 0; j < ongoingReplies.size(); ++j) { for (int j = 0; j < ongoingReplies.size(); ++j) {
@ -354,6 +355,9 @@ Future<Void> getBatchReplies(RequestStream<Request> Interface::*channel, std::ma
} catch (Error& e) { } catch (Error& e) {
if (e.code() == error_code_operation_cancelled) break; if (e.code() == error_code_operation_cancelled) break;
fprintf(stdout, "sendBatchRequests Error code:%d, error message:%s\n", e.code(), e.what()); fprintf(stdout, "sendBatchRequests Error code:%d, error message:%s\n", e.code(), e.what());
TraceEvent(SevWarn, "FastRestoreSendBatchRequests")
.detail("ErrorCode", e.code())
.detail("ErrorInfo", e.what());
for (auto& request : requests) { for (auto& request : requests) {
TraceEvent(SevWarn, "FastRestore") TraceEvent(SevWarn, "FastRestore")
.detail("SendBatchRequests", requests.size()) .detail("SendBatchRequests", requests.size())

24
fdbserver/RestoreLoader.actor.cpp
View File

@ -540,8 +540,8 @@ ACTOR Future<Void> sendMutationsToApplier(VersionedMutationsMap* pkvOps, int bat
.detail("Version", commitVersion.toString()) .detail("Version", commitVersion.toString())
.detail("Mutation", kvm.toString()); .detail("Mutation", kvm.toString());
} }
applierVersionedMutationsBuffer[applierID].push_back(applierVersionedMutationsBuffer[applierID].arena(), applierVersionedMutationsBuffer[applierID].push_back_deep(
VersionedMutation(kvm, commitVersion)); applierVersionedMutationsBuffer[applierID].arena(), VersionedMutation(kvm, commitVersion));
msgSize += kvm.expectedSize(); msgSize += kvm.expectedSize();
} }
@ -558,8 +558,10 @@ ACTOR Future<Void> sendMutationsToApplier(VersionedMutationsMap* pkvOps, int bat
.detail("MessageIndex", msgIndex) .detail("MessageIndex", msgIndex)
.detail("RestoreAsset", asset.toString()) .detail("RestoreAsset", asset.toString())
.detail("Requests", requests.size()); .detail("Requests", requests.size());
fSends.push_back(sendBatchRequests(&RestoreApplierInterface::sendMutationVector, *pApplierInterfaces, // fSends.push_back(sendBatchRequests(&RestoreApplierInterface::sendMutationVector, *pApplierInterfaces,
requests, TaskPriority::RestoreLoaderSendMutations)); // requests, TaskPriority::RestoreLoaderSendMutations));
wait(sendBatchRequests(&RestoreApplierInterface::sendMutationVector, *pApplierInterfaces, requests,
TaskPriority::RestoreLoaderSendMutations));
msgIndex++; msgIndex++;
msgSize = 0; msgSize = 0;
for (auto& applierID : applierIDs) { for (auto& applierID : applierIDs) {
@ -582,12 +584,18 @@ ACTOR Future<Void> sendMutationsToApplier(VersionedMutationsMap* pkvOps, int bat
.detail("MessageIndex", msgIndex) .detail("MessageIndex", msgIndex)
.detail("RestoreAsset", asset.toString()) .detail("RestoreAsset", asset.toString())
.detail("Requests", requests.size()); .detail("Requests", requests.size());
fSends.push_back(sendBatchRequests(&RestoreApplierInterface::sendMutationVector, *pApplierInterfaces, requests, // fSends.push_back(sendBatchRequests(&RestoreApplierInterface::sendMutationVector, *pApplierInterfaces,
TaskPriority::RestoreLoaderSendMutations)); // requests,
// TaskPriority::RestoreLoaderSendMutations));
wait(sendBatchRequests(&RestoreApplierInterface::sendMutationVector, *pApplierInterfaces, requests,
TaskPriority::RestoreLoaderSendMutations));
} }
wait(waitForAll(fSends)); // wait(waitForAll(fSends));
TraceEvent("FastRestore").detail("LoaderSendMutationOnAppliers", kvCount); TraceEvent("FastRestoreLoaderSendMutationToAppliers")
.detail("BatchIndex", batchIndex)
.detail("RestoreAsset", asset.toString())
.detail("Mutations", kvCount);
return Void(); return Void();
} }

2
fdbserver/RestoreUtil.h
View File

@ -45,6 +45,8 @@ struct VersionedMutation {
VersionedMutation() = default; VersionedMutation() = default;
explicit VersionedMutation(MutationRef mutation, LogMessageVersion version) explicit VersionedMutation(MutationRef mutation, LogMessageVersion version)
: mutation(mutation), version(version) {} : mutation(mutation), version(version) {}
explicit VersionedMutation(Arena& to, const VersionedMutation& from)
: mutation(to, from.mutation), version(from.version) {}
template <class Ar> template <class Ar>
void serialize(Ar& ar) { void serialize(Ar& ar) {