ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-12-06 17:27:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve names returned by server_info counters

Browse Source
This commit is contained in:
Scott Schurr
2021-11-15 19:07:22 -08:00
committed by manojsdoshi
parent fc04336caa
commit d54f6278bb
8 changed files with 100 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/ripple/app/misc/NetworkOPs.cpp
View File

@@ -2943,7 +2943,7 @@ NetworkOPsImp::reportFeeChange()
if (f != mLastFeeSummary)
{
m_job_queue.addJob(
jtCLIENT, "reportFeeChange->pubServer", [this](Job&) {
jtCLIENT_FEE_CHANGE, "reportFeeChange->pubServer", [this](Job&) {
pubServer();
});
}
@@ -2953,7 +2953,7 @@ void
NetworkOPsImp::reportConsensusStateChange(ConsensusPhase phase)
{
m_job_queue.addJob(
jtCLIENT,
jtCLIENT_CONSENSUS,
"reportConsensusStateChange->pubConsensus",
[this, phase](Job&) { pubConsensus(phase); });
}
@@ -3346,7 +3346,7 @@ NetworkOPsImp::addAccountHistoryJob(SubAccountHistoryInfoWeak subInfo)
}
app_.getJobQueue().addJob(
jtCLIENT,
jtCLIENT_ACCT_HIST,
"AccountHistoryTxStream",
[this, dbType = databaseType, subInfo](Job&) {
auto const& accountId = subInfo.index_->accountId_;

63
src/ripple/core/Job.h
View File

@@ -39,34 +39,41 @@ enum JobType {
// earlier jobs having lower priority than later jobs. If you wish to
// insert a job at a specific priority, simply add it at the right location.
jtPACK, // Make a fetch pack for a peer
jtPUBOLDLEDGER, // An old ledger has been accepted
jtCLIENT, // A websocket command from the client
jtRPC, // A websocket command from the client
jtVALIDATION_ut, // A validation from an untrusted source
jtUPDATE_PF, // Update pathfinding requests
jtTRANSACTION_l, // A local transaction
jtREPLAY_REQ, // Peer request a ledger delta or a skip list
jtLEDGER_REQ, // Peer request ledger/txnset data
jtPROPOSAL_ut, // A proposal from an untrusted source
jtREPLAY_TASK, // A Ledger replay task/subtask
jtLEDGER_DATA, // Received data for a ledger we're acquiring
jtSWEEP, // Sweep for stale structures
jtTRANSACTION, // A transaction received from the network
jtMISSING_TXN, // Request missing transactions
jtREQUESTED_TXN, // Reply with requested transactions
jtBATCH, // Apply batched transactions
jtADVANCE, // Advance validated/acquired ledgers
jtPUBLEDGER, // Publish a fully-accepted ledger
jtTXN_DATA, // Fetch a proposed set
jtWAL, // Write-ahead logging
jtVALIDATION_t, // A validation from a trusted source
jtWRITE, // Write out hashed objects
jtACCEPT, // Accept a consensus ledger
jtPROPOSAL_t, // A proposal from a trusted source
jtNETOP_CLUSTER, // NetworkOPs cluster peer report
jtNETOP_TIMER, // NetworkOPs net timer processing
jtADMIN, // An administrative operation
jtPACK, // Make a fetch pack for a peer
jtPUBOLDLEDGER, // An old ledger has been accepted
jtCLIENT, // A placeholder for the priority of all jtCLIENT jobs
jtCLIENT_SUBSCRIBE, // A websocket subscription by a client
jtCLIENT_FEE_CHANGE, // Subscription for fee change by a client
jtCLIENT_CONSENSUS, // Subscription for consensus state change by a client
jtCLIENT_ACCT_HIST, // Subscription for account history by a client
jtCLIENT_SHARD, // Client request for shard archiving
jtCLIENT_RPC, // Client RPC request
jtCLIENT_WEBSOCKET, // Client websocket request
jtRPC, // A websocket command from the client
jtSWEEP, // Sweep for stale structures
jtVALIDATION_ut, // A validation from an untrusted source
jtUPDATE_PF, // Update pathfinding requests
jtTRANSACTION_l, // A local transaction
jtREPLAY_REQ, // Peer request a ledger delta or a skip list
jtLEDGER_REQ, // Peer request ledger/txnset data
jtPROPOSAL_ut, // A proposal from an untrusted source
jtREPLAY_TASK, // A Ledger replay task/subtask
jtLEDGER_DATA, // Received data for a ledger we're acquiring
jtTRANSACTION, // A transaction received from the network
jtMISSING_TXN, // Request missing transactions
jtREQUESTED_TXN, // Reply with requested transactions
jtBATCH, // Apply batched transactions
jtADVANCE, // Advance validated/acquired ledgers
jtPUBLEDGER, // Publish a fully-accepted ledger
jtTXN_DATA, // Fetch a proposed set
jtWAL, // Write-ahead logging
jtVALIDATION_t, // A validation from a trusted source
jtWRITE, // Write out hashed objects
jtACCEPT, // Accept a consensus ledger
jtPROPOSAL_t, // A proposal from a trusted source
jtNETOP_CLUSTER, // NetworkOPs cluster peer report
jtNETOP_TIMER, // NetworkOPs net timer processing
jtADMIN, // An administrative operation
// Special job types which are not dispatched by the job pool
jtPEER,

87
src/ripple/core/JobTypes.h
View File

@@ -67,46 +67,55 @@ private:
};
// clang-format off
add(jtPACK, "makeFetchPack", 1, 0ms, 0ms);
add(jtPUBOLDLEDGER, "publishAcqLedger", 2, 10000ms, 15000ms);
add(jtVALIDATION_ut, "untrustedValidation", maxLimit, 2000ms, 5000ms);
add(jtTRANSACTION_l, "localTransaction", maxLimit, 100ms, 500ms);
add(jtREPLAY_REQ, "ledgerReplayRequest", 10, 250ms, 1000ms);
add(jtLEDGER_REQ, "ledgerRequest", 4, 0ms, 0ms);
add(jtPROPOSAL_ut, "untrustedProposal", maxLimit, 500ms, 1250ms);
add(jtREPLAY_TASK, "ledgerReplayTask", maxLimit, 0ms, 0ms);
add(jtLEDGER_DATA, "ledgerData", 4, 0ms, 0ms);
add(jtCLIENT, "clientCommand", maxLimit, 2000ms, 5000ms);
add(jtRPC, "RPC", maxLimit, 0ms, 0ms);
add(jtUPDATE_PF, "updatePaths", 1, 0ms, 0ms);
add(jtTRANSACTION, "transaction", maxLimit, 250ms, 1000ms);
add(jtBATCH, "batch", maxLimit, 250ms, 1000ms);
add(jtADVANCE, "advanceLedger", maxLimit, 0ms, 0ms);
add(jtPUBLEDGER, "publishNewLedger", maxLimit, 3000ms, 4500ms);
add(jtTXN_DATA, "fetchTxnData", 5, 0ms, 0ms);
add(jtWAL, "writeAhead", maxLimit, 1000ms, 2500ms);
add(jtVALIDATION_t, "trustedValidation", maxLimit, 500ms, 1500ms);
add(jtWRITE, "writeObjects", maxLimit, 1750ms, 2500ms);
add(jtACCEPT, "acceptLedger", maxLimit, 0ms, 0ms);
add(jtPROPOSAL_t, "trustedProposal", maxLimit, 100ms, 500ms);
add(jtSWEEP, "sweep", 1, 0ms, 0ms);
add(jtNETOP_CLUSTER, "clusterReport", 1, 9999ms, 9999ms);
add(jtNETOP_TIMER, "heartbeat", 1, 999ms, 999ms);
add(jtADMIN, "administration", maxLimit, 0ms, 0ms);
add(jtMISSING_TXN, "handleHaveTransactions", 1200, 0ms, 0ms);
add(jtREQUESTED_TXN, "doTransactions", 1200, 0ms, 0ms);
// avg peak
// JobType name limit latency latency
add(jtPACK, "makeFetchPack", 1, 0ms, 0ms);
add(jtPUBOLDLEDGER, "publishAcqLedger", 2, 10000ms, 15000ms);
add(jtVALIDATION_ut, "untrustedValidation", maxLimit, 2000ms, 5000ms);
add(jtTRANSACTION_l, "localTransaction", maxLimit, 100ms, 500ms);
add(jtREPLAY_REQ, "ledgerReplayRequest", 10, 250ms, 1000ms);
add(jtLEDGER_REQ, "ledgerRequest", 4, 0ms, 0ms);
add(jtPROPOSAL_ut, "untrustedProposal", maxLimit, 500ms, 1250ms);
add(jtREPLAY_TASK, "ledgerReplayTask", maxLimit, 0ms, 0ms);
add(jtLEDGER_DATA, "ledgerData", 4, 0ms, 0ms);
add(jtCLIENT, "clientCommand", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_SUBSCRIBE, "clientSubscribe", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_FEE_CHANGE, "clientFeeChange", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_CONSENSUS, "clientConsensus", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_ACCT_HIST, "clientAccountHistory", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_SHARD, "clientShardArchive", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_RPC, "clientRPC", maxLimit, 2000ms, 5000ms);
add(jtCLIENT_WEBSOCKET, "clientWebsocket", maxLimit, 2000ms, 5000ms);
add(jtRPC, "RPC", maxLimit, 0ms, 0ms);
add(jtUPDATE_PF, "updatePaths", 1, 0ms, 0ms);
add(jtTRANSACTION, "transaction", maxLimit, 250ms, 1000ms);
add(jtBATCH, "batch", maxLimit, 250ms, 1000ms);
add(jtADVANCE, "advanceLedger", maxLimit, 0ms, 0ms);
add(jtPUBLEDGER, "publishNewLedger", maxLimit, 3000ms, 4500ms);
add(jtTXN_DATA, "fetchTxnData", 5, 0ms, 0ms);
add(jtWAL, "writeAhead", maxLimit, 1000ms, 2500ms);
add(jtVALIDATION_t, "trustedValidation", maxLimit, 500ms, 1500ms);
add(jtWRITE, "writeObjects", maxLimit, 1750ms, 2500ms);
add(jtACCEPT, "acceptLedger", maxLimit, 0ms, 0ms);
add(jtPROPOSAL_t, "trustedProposal", maxLimit, 100ms, 500ms);
add(jtSWEEP, "sweep", 1, 0ms, 0ms);
add(jtNETOP_CLUSTER, "clusterReport", 1, 9999ms, 9999ms);
add(jtNETOP_TIMER, "heartbeat", 1, 999ms, 999ms);
add(jtADMIN, "administration", maxLimit, 0ms, 0ms);
add(jtMISSING_TXN, "handleHaveTransactions", 1200, 0ms, 0ms);
add(jtREQUESTED_TXN, "doTransactions", 1200, 0ms, 0ms);
add(jtPEER, "peerCommand", 0, 200ms, 2500ms);
add(jtDISK, "diskAccess", 0, 500ms, 1000ms);
add(jtTXN_PROC, "processTransaction", 0, 0ms, 0ms);
add(jtOB_SETUP, "orderBookSetup", 0, 0ms, 0ms);
add(jtPATH_FIND, "pathFind", 0, 0ms, 0ms);
add(jtHO_READ, "nodeRead", 0, 0ms, 0ms);
add(jtHO_WRITE, "nodeWrite", 0, 0ms, 0ms);
add(jtGENERIC, "generic", 0, 0ms, 0ms);
add(jtNS_SYNC_READ, "SyncReadNode", 0, 0ms, 0ms);
add(jtNS_ASYNC_READ, "AsyncReadNode", 0, 0ms, 0ms);
add(jtNS_WRITE, "WriteNode", 0, 0ms, 0ms);
add(jtPEER, "peerCommand", 0, 200ms, 2500ms);
add(jtDISK, "diskAccess", 0, 500ms, 1000ms);
add(jtTXN_PROC, "processTransaction", 0, 0ms, 0ms);
add(jtOB_SETUP, "orderBookSetup", 0, 0ms, 0ms);
add(jtPATH_FIND, "pathFind", 0, 0ms, 0ms);
add(jtHO_READ, "nodeRead", 0, 0ms, 0ms);
add(jtHO_WRITE, "nodeWrite", 0, 0ms, 0ms);
add(jtGENERIC, "generic", 0, 0ms, 0ms);
add(jtNS_SYNC_READ, "SyncReadNode", 0, 0ms, 0ms);
add(jtNS_ASYNC_READ, "AsyncReadNode", 0, 0ms, 0ms);
add(jtNS_WRITE, "WriteNode", 0, 0ms, 0ms);
// clang-format on
}

4
src/ripple/core/impl/JobQueue.cpp
View File

@@ -94,7 +94,9 @@ JobQueue::addRefCountedJob(
// FIXME: Workaround incorrect client shutdown ordering
// do not add jobs to a queue with no threads
assert(type == jtCLIENT || m_workers.getNumberOfThreads() > 0);
assert(
(type >= jtCLIENT && type <= jtCLIENT_WEBSOCKET) ||
m_workers.getNumberOfThreads() > 0);
{
std::lock_guard lock(m_mutex);

4
src/ripple/net/impl/RPCSub.cpp
View File

@@ -96,7 +96,9 @@ public:
JLOG(j_.info()) << "RPCCall::fromNetwork start";
mSending = m_jobQueue.addJob(
jtCLIENT, "RPCSub::sendThread", [this](Job&) { sendThread(); });
jtCLIENT_SUBSCRIBE, "RPCSub::sendThread", [this](Job&) {
sendThread();
});
}
}

9
src/ripple/rpc/impl/RPCHandler.cpp
View File

@@ -129,12 +129,11 @@ fillHandler(JsonContext& context, Handler const*& result)
{
if (!isUnlimited(context.role))
{
// VFALCO NOTE Should we also add up the jtRPC jobs?
//
int jc = context.app.getJobQueue().getJobCountGE(jtCLIENT);
if (jc > Tuning::maxJobQueueClients)
// Count all jobs at jtCLIENT priority or higher.
int const jobCount = context.app.getJobQueue().getJobCountGE(jtCLIENT);
if (jobCount > Tuning::maxJobQueueClients)
{
JLOG(context.j.debug()) << "Too busy for command: " << jc;
JLOG(context.j.debug()) << "Too busy for command: " << jobCount;
return rpcTOO_BUSY;
}
}

4
src/ripple/rpc/impl/ServerHandlerImp.cpp
View File

@@ -292,7 +292,7 @@ ServerHandlerImp::onRequest(Session& session)
std::shared_ptr<Session> detachedSession = session.detach();
auto const postResult = m_jobQueue.postCoro(
jtCLIENT,
jtCLIENT_RPC,
"RPC-Client",
[this, detachedSession](std::shared_ptr<JobQueue::Coro> coro) {
processSession(detachedSession, coro);
@@ -339,7 +339,7 @@ ServerHandlerImp::onWSMessage(
JLOG(m_journal.trace()) << "Websocket received '" << jv << "'";
auto const postResult = m_jobQueue.postCoro(
jtCLIENT,
jtCLIENT_WEBSOCKET,
"WS-Client",
[this, session, jv = std::move(jv)](
std::shared_ptr<JobQueue::Coro> const& coro) {

4
src/ripple/rpc/impl/ShardArchiveHandler.cpp
View File

@@ -382,7 +382,7 @@ ShardArchiveHandler::next(std::lock_guard<std::mutex> const& l)
return onClosureFailed(
"failed to wrap closure for starting download", l);
app_.getJobQueue().addJob(jtCLIENT, "ShardArchiveHandler", *wrapper);
app_.getJobQueue().addJob(jtCLIENT_SHARD, "ShardArchiveHandler", *wrapper);
return true;
}
@@ -465,7 +465,7 @@ ShardArchiveHandler::complete(path dstPath)
}
// Process in another thread to not hold up the IO service
app_.getJobQueue().addJob(jtCLIENT, "ShardArchiveHandler", *wrapper);
app_.getJobQueue().addJob(jtCLIENT_SHARD, "ShardArchiveHandler", *wrapper);
}
void