Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fail loadgen when load is too high, remove Simulation dependency #2220

Merged
merged 1 commit into from
Aug 14, 2019
Merged

Fail loadgen when load is too high, remove Simulation dependency #2220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions src/simulation/CoreTests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,8 @@ TEST_CASE(
auto nodes = simulation->getNodes();
auto& app = *nodes[0]; // pick a node to generate load

app.getLoadGenerator().generateLoad(true, 3, 0, 0, 10, 100);
auto& lg = app.getLoadGenerator();
lg.generateLoad(true, 3, 0, 0, 10, 100);
try
{
simulation->crankUntil(
Expand All @@ -346,21 +347,21 @@ TEST_CASE(
// to the second node in time and the second node gets the
// nomination
return simulation->haveAllExternalized(5, 2) &&
simulation->accountsOutOfSyncWithDb(app).empty();
lg.checkAccountSynced(app, true).empty();
},
3 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, false);

app.getLoadGenerator().generateLoad(false, 3, 0, 10, 10, 100);
lg.generateLoad(false, 3, 0, 10, 10, 100);
simulation->crankUntil(
[&]() {
return simulation->haveAllExternalized(8, 2) &&
simulation->accountsOutOfSyncWithDb(app).empty();
lg.checkAccountSynced(app, false).empty();
},
2 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, true);
}
catch (...)
{
auto problems = simulation->accountsOutOfSyncWithDb(app);
auto problems = lg.checkAccountSynced(app, false);
REQUIRE(problems.empty());
}

Expand Down Expand Up @@ -510,14 +511,15 @@ netTopologyTest(std::string const& name,
assert(!nodes.empty());
auto& app = *nodes[0];

app.getLoadGenerator().generateLoad(true, 50, 0, 0, 10, 100);
auto& lg = app.getLoadGenerator();
lg.generateLoad(true, 50, 0, 0, 10, 100);
auto& complete =
app.getMetrics().NewMeter({"loadgen", "run", "complete"}, "run");

sim->crankUntil(
[&]() {
return sim->haveAllExternalized(8, 2) &&
sim->accountsOutOfSyncWithDb(app).empty() &&
lg.checkAccountSynced(app, true).empty() &&
complete.count() == 1;
},
2 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, true);
Expand Down
148 changes: 103 additions & 45 deletions src/simulation/LoadGenerator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,22 +36,30 @@ namespace stellar
using namespace std;
using namespace txtest;

// Units of load are is scheduled at 100ms intervals.
// Units of load are scheduled at 100ms intervals.
const uint32_t LoadGenerator::STEP_MSECS = 100;
//
const uint32_t LoadGenerator::TX_SUBMIT_MAX_TRIES = 1000;

// If submission fails with txBAD_SEQ, attempt refreshing the account or
// re-submitting a new payment
const uint32_t LoadGenerator::TX_SUBMIT_MAX_TRIES = 10;

// After successfully submitting desired load, wait a bit to let it get into the
// ledger.
const uint32_t LoadGenerator::TIMEOUT_NUM_LEDGERS = 20;

LoadGenerator::LoadGenerator(Application& app)
: mMinBalance(0), mLastSecond(0), mApp(app), mTotalSubmitted(0)
: mMinBalance(0)
, mLastSecond(0)
, mApp(app)
, mTotalSubmitted(0)
, mLoadgenComplete(
mApp.getMetrics().NewMeter({"loadgen", "run", "complete"}, "run"))
, mLoadgenFail(
mApp.getMetrics().NewMeter({"loadgen", "run", "failed"}, "run"))
{
createRootAccount();
}

LoadGenerator::~LoadGenerator()
{
clear();
}

void
LoadGenerator::createRootAccount()
{
Expand Down Expand Up @@ -92,12 +100,14 @@ LoadGenerator::getTxPerStep(uint32_t txRate)
}

void
LoadGenerator::clear()
LoadGenerator::reset()
{
mAccounts.clear();
mRoot.reset();
mStartTime.reset();
mTotalSubmitted = 0;
mWaitTillCompleteForLedgers = 0;
mFailed = false;
}

// Schedule a callback to generateLoad() STEP_MSECS miliseconds from now.
Expand All @@ -106,6 +116,18 @@ LoadGenerator::scheduleLoadGeneration(bool isCreate, uint32_t nAccounts,
uint32_t offset, uint32_t nTxs,
uint32_t txRate, uint32_t batchSize)
{
// If previously scheduled step of load did not succeed, fail this loadgen
// run.
if (mFailed)
{
CLOG(ERROR, "LoadGen") << "Load generation failed, ensure correct "
"number parameters are set and accounts are "
"created, or retry with smaller tx rate.";
mLoadgenFail.Mark();
reset();
return;
}

if (!mLoadTimer)
{
mLoadTimer = std::make_unique<VirtualTimer>(mApp.getClock());
Expand Down Expand Up @@ -156,7 +178,7 @@ LoadGenerator::generateLoad(bool isCreate, uint32_t nAccounts, uint32_t offset,
if ((isCreate && nAccounts == 0) || (!isCreate && nTxs == 0))
{
// Done submitting the load, now ensure it propagates to the DB.
waitTillComplete();
waitTillComplete(isCreate);
return;
}

Expand Down Expand Up @@ -229,18 +251,23 @@ LoadGenerator::submitCreationTx(uint32_t nAccounts, uint32_t offset,
while ((status = tx.execute(mApp, true, code, batchSize)) !=
TransactionQueue::AddResult::ADD_STATUS_PENDING)
{
handleFailedSubmission(tx.mFrom, status, code); // Update seq num
// Ignore duplicate transactions, simply continue generating load
if (status == TransactionQueue::AddResult::ADD_STATUS_DUPLICATE)
{
createDuplicate = true;
break;
}
if (++numTries >= TX_SUBMIT_MAX_TRIES)

if (++numTries >= TX_SUBMIT_MAX_TRIES ||
status != TransactionQueue::AddResult::ADD_STATUS_ERROR)
{
CLOG(ERROR, "LoadGen") << "Error creating account!";
clear();
// Failed to submit the step of load
mFailed = true;
return 0;
}

// In case of bad seqnum, attempt refreshing it from the DB
maybeHandleFailedTx(tx.mFrom, status, code);
}

if (!createDuplicate)
Expand All @@ -267,16 +294,18 @@ LoadGenerator::submitPaymentTx(uint32_t nAccounts, uint32_t offset,
while ((status = tx.execute(mApp, false, code, batchSize)) !=
TransactionQueue::AddResult::ADD_STATUS_PENDING)
{
handleFailedSubmission(tx.mFrom, status, code); // Update seq num
tx = paymentTransaction(nAccounts, offset, ledgerNum,
sourceAccountId); // re-generate the tx
if (++numTries >= TX_SUBMIT_MAX_TRIES)
if (++numTries >= TX_SUBMIT_MAX_TRIES ||
status != TransactionQueue::AddResult::ADD_STATUS_ERROR)
{
CLOG(ERROR, "LoadGen") << "Error submitting tx: did you specify "
"correct number of accounts and offset?";
clear();
mFailed = true;
return 0;
}

// In case of bad seqnum, attempt refreshing it from the DB
maybeHandleFailedTx(tx.mFrom, status, code); // Update seq num

// Regenerate a new payment tx
tx = paymentTransaction(nAccounts, offset, ledgerNum, sourceAccountId);
}

nTxs -= 1;
Expand Down Expand Up @@ -409,13 +438,13 @@ LoadGenerator::findAccount(uint64_t accountId, uint32_t ledgerNum)
if (res == mAccounts.end())
{
SequenceNumber sn = static_cast<SequenceNumber>(ledgerNum) << 32;
auto name = "TestAccount-" + to_string(accountId);
auto account = TestAccount{mApp, txtest::getAccount(name.c_str()), sn};
newAccountPtr = make_shared<TestAccount>(account);
auto name = "TestAccount-" + std::to_string(accountId);
newAccountPtr =
std::make_shared<TestAccount>(mApp, txtest::getAccount(name), sn);

if (!loadAccount(newAccountPtr, mApp))
{
std::runtime_error(
throw std::runtime_error(
marta-lokhova marked this conversation as resolved.
Show resolved Hide resolved
fmt::format("Account {0} must exist in the DB.", accountId));
}
mAccounts.insert(
Expand Down Expand Up @@ -445,9 +474,9 @@ LoadGenerator::paymentTransaction(uint32_t numAccounts, uint32_t offset,
}

void
LoadGenerator::handleFailedSubmission(TestAccountPtr sourceAccount,
TransactionQueue::AddResult status,
TransactionResultCode code)
LoadGenerator::maybeHandleFailedTx(TestAccountPtr sourceAccount,
TransactionQueue::AddResult status,
TransactionResultCode code)
{
// Note that if transaction is a DUPLICATE, its sequence number is
// incremented on the next call to execute.
Expand All @@ -463,52 +492,81 @@ LoadGenerator::handleFailedSubmission(TestAccountPtr sourceAccount,
}

std::vector<LoadGenerator::TestAccountPtr>
LoadGenerator::checkAccountSynced(Application& app)
LoadGenerator::checkAccountSynced(Application& app, bool isCreate)
{
std::vector<TestAccountPtr> result;
for (auto const& acc : mAccounts)
{
TestAccountPtr account = acc.second;
auto currentSeqNum = account->getLastSequenceNumber();
auto reloadRes = loadAccount(account, app);
// reload the account
if (!reloadRes || currentSeqNum != account->getLastSequenceNumber())
auto accountFromDB = *account;

auto reloadRes = loadAccount(accountFromDB, app);
// For account creation, reload accounts from the DB
// For payments, ensure that the sequence number matches expected
// seqnum. Timeout after 20 ledgers.
if (isCreate)
{
if (!reloadRes)
{
CLOG(TRACE, "LoadGen") << "Account " << account->getAccountId()
<< " is not created yet!";
result.push_back(account);
}
}
else if (!reloadRes)
{
auto msg =
fmt::format("Account {} used to submit payment tx could not "
"load, DB might be in a corrupted state",
account->getAccountId());
throw std::runtime_error(msg);
}
else if (account->getLastSequenceNumber() !=
accountFromDB.getLastSequenceNumber())
{
CLOG(DEBUG, "LoadGen")
CLOG(TRACE, "LoadGen")
<< "Account " << account->getAccountId()
<< " is at sequence num " << currentSeqNum
<< ", but the DB is at " << account->getLastSequenceNumber();
<< " is at sequence num " << account->getLastSequenceNumber()
<< ", but the DB is at "
<< accountFromDB.getLastSequenceNumber();
result.push_back(account);
}
}
return result;
}

void
LoadGenerator::waitTillComplete()
LoadGenerator::waitTillComplete(bool isCreate)
{
if (!mLoadTimer)
{
mLoadTimer = std::make_unique<VirtualTimer>(mApp.getClock());
}
vector<TestAccountPtr> inconsistencies;
inconsistencies = checkAccountSynced(mApp);
inconsistencies = checkAccountSynced(mApp, isCreate);
marta-lokhova marked this conversation as resolved.
Show resolved Hide resolved

if (inconsistencies.empty())
{
CLOG(INFO, "LoadGen") << "Load generation complete.";
mApp.getMetrics()
.NewMeter({"loadgen", "run", "complete"}, "run")
.Mark();
clear();
mLoadgenComplete.Mark();
reset();
return;
}
else
{
if (++mWaitTillCompleteForLedgers >= TIMEOUT_NUM_LEDGERS)
{
CLOG(INFO, "LoadGen") << "Load generation failed.";
mLoadgenFail.Mark();
reset();
return;
}

mLoadTimer->expires_from_now(
mApp.getConfig().getExpectedLedgerCloseTime());
mLoadTimer->async_wait([this]() { this->waitTillComplete(); },
&VirtualTimer::onFailureNoop);
mLoadTimer->async_wait(
[this, isCreate]() { this->waitTillComplete(isCreate); },
&VirtualTimer::onFailureNoop);
}
}

Expand Down
Loading