Skip to content

Commit 5eb6cb6

Browse files
latobaritalatobarita
authored
Merge pull request #4196 from marta-lokhova/tx_queue_simplify
Simplify tx queue, remove transaction chains support Reviewed-by: dmkozh
2 parents 4c0c913 + c06ba8e commit 5eb6cb6

13 files changed

+207
-465
lines changed

‎src/herder/HerderImpl.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -504,7 +504,7 @@ HerderImpl::broadcast(SCPEnvelope const& e)
504504
e.statement.slotIndex);
505505

506506
mSCPMetrics.mEnvelopeEmit.Mark();
507-
mApp.getOverlayManager().broadcastMessage(m, false);
507+
mApp.getOverlayManager().broadcastMessage(m);
508508
}
509509
}
510510

‎src/herder/TransactionQueue.cpp

+150-402
Large diffs are not rendered by default.

‎src/herder/TransactionQueue.h

+8-28
Original file line numberDiff line numberDiff line change
@@ -72,23 +72,6 @@ class TransactionQueue
7272
ADD_STATUS_COUNT
7373
};
7474

75-
/*
76-
* Information about queue of transaction for given account. mAge and
77-
* mTotalFees are stored in queue, but mMaxSeq must be computed each
78-
* time (its O(1) anyway).
79-
*/
80-
struct AccountTxQueueInfo
81-
{
82-
SequenceNumber mMaxSeq{0};
83-
int64_t mTotalFees{0};
84-
size_t mQueueSizeOps{0};
85-
size_t mBroadcastQueueOps{0};
86-
uint32_t mAge{0};
87-
88-
friend bool operator==(AccountTxQueueInfo const& x,
89-
AccountTxQueueInfo const& y);
90-
};
91-
9275
/**
9376
* AccountState stores the following information:
9477
* - mTotalFees: the sum of feeBid() over every transaction for which this
@@ -108,15 +91,12 @@ class TransactionQueue
10891
VirtualClock::time_point mInsertionTime;
10992
bool mSubmittedFromSelf;
11093
};
111-
using TimestampedTransactions = std::vector<TimestampedTx>;
11294
using Transactions = std::vector<TransactionFrameBasePtr>;
11395
struct AccountState
11496
{
11597
int64_t mTotalFees{0};
116-
size_t mQueueSizeOps{0};
117-
size_t mBroadcastQueueOps{0};
11898
uint32_t mAge{0};
119-
TimestampedTransactions mTransactions;
99+
std::optional<TimestampedTx> mTransaction;
120100
};
121101

122102
explicit TransactionQueue(Application& app, uint32 pendingDepth,
@@ -129,6 +109,9 @@ class TransactionQueue
129109

130110
AddResult tryAdd(TransactionFrameBasePtr tx, bool submittedFromSelf);
131111
void removeApplied(Transactions const& txs);
112+
// Ban transactions that are no longer valid or have insufficient fee;
113+
// transaction per account limit applies here, so `txs` should have no
114+
// duplicate source accounts
132115
void ban(Transactions const& txs);
133116

134117
/**
@@ -148,7 +131,7 @@ class TransactionQueue
148131
virtual size_t getMaxQueueSizeOps() const = 0;
149132

150133
#ifdef BUILD_TESTS
151-
AccountTxQueueInfo
134+
AccountState
152135
getAccountTransactionQueueInfo(AccountID const& accountID) const;
153136
size_t countBanned(int index) const;
154137
#endif
@@ -202,18 +185,15 @@ class TransactionQueue
202185
BROADCAST_STATUS_SUCCESS,
203186
BROADCAST_STATUS_SKIPPED
204187
};
205-
BroadcastStatus broadcastTx(AccountState& state, TimestampedTx& tx);
188+
BroadcastStatus broadcastTx(TimestampedTx& tx);
206189
AddResult canAdd(TransactionFrameBasePtr tx,
207190
AccountStates::iterator& stateIter,
208-
TimestampedTransactions::iterator& oldTxIter,
209191
std::vector<std::pair<TxStackPtr, bool>>& txsToEvict);
210192

211193
void releaseFeeMaybeEraseAccountState(TransactionFrameBasePtr tx);
212194

213-
void prepareDropTransaction(AccountState& as, TimestampedTx& tstx);
214-
void dropTransactions(AccountStates::iterator stateIter,
215-
TimestampedTransactions::iterator begin,
216-
TimestampedTransactions::iterator end);
195+
void prepareDropTransaction(AccountState& as);
196+
void dropTransaction(AccountStates::iterator stateIter);
217197

218198
void clearAll();
219199

‎src/herder/test/TransactionQueueTests.cpp

+27-14
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,14 @@ class TransactionQueueTest
125125
REQUIRE(size - toRemove.size() >=
126126
mTransactionQueue.getTransactions({}).size());
127127
}
128+
129+
// Everything that got removed should have age=0
130+
for (auto const& tx : toRemove)
131+
{
132+
auto txInfo = mTransactionQueue.getAccountTransactionQueueInfo(
133+
tx->getSourceID());
134+
REQUIRE(txInfo.mAge == 0);
135+
}
128136
}
129137

130138
void
@@ -201,11 +209,16 @@ class TransactionQueueTest
201209
accountState.mAccountID);
202210
REQUIRE(accountTransactionQueueInfo.mTotalFees ==
203211
expectedFees[accountState.mAccountID]);
204-
REQUIRE(accountTransactionQueueInfo.mMaxSeq == seqNum);
212+
auto queueSeqNum =
213+
accountTransactionQueueInfo.mTransaction
214+
? accountTransactionQueueInfo.mTransaction->mTx->getSeqNum()
215+
: 0;
216+
totOps += accountTransactionQueueInfo.mTransaction
217+
? accountTransactionQueueInfo.mTransaction->mTx
218+
->getNumOperations()
219+
: 0;
220+
REQUIRE(queueSeqNum == seqNum);
205221
REQUIRE(accountTransactionQueueInfo.mAge == accountState.mAge);
206-
REQUIRE(accountTransactionQueueInfo.mBroadcastQueueOps ==
207-
accountTransactionQueueInfo.mQueueSizeOps);
208-
totOps += accountTransactionQueueInfo.mQueueSizeOps;
209222

210223
expectedTxs.insert(expectedTxs.end(),
211224
accountState.mAccountTransactions.begin(),
@@ -543,19 +556,21 @@ testTransactionQueueBasicScenarios()
543556
test.add(txSeqA1T1, TransactionQueue::AddResult::ADD_STATUS_DUPLICATE);
544557
test.check(state);
545558

546-
auto status = TransactionQueue::AddResult::ADD_STATUS_TRY_AGAIN_LATER;
547-
test.add(txSeqA1T2, status);
559+
test.add(txSeqA1T2,
560+
TransactionQueue::AddResult::ADD_STATUS_TRY_AGAIN_LATER);
548561
test.check(state);
549562

550-
// Regardless of seqnum validity, tx is rejected due to limit
563+
// Tx is rejected due to limit or bad seqnum
551564
// too low
552-
test.add(txSeqA1T0, status);
565+
test.add(txSeqA1T0, TransactionQueue::AddResult::ADD_STATUS_ERROR);
553566
test.check(state);
554567
// too high
555-
test.add(txSeqA1T4, status);
568+
test.add(txSeqA1T4,
569+
TransactionQueue::AddResult::ADD_STATUS_TRY_AGAIN_LATER);
556570
test.check(state);
557571
// just right
558-
test.add(txSeqA1T3, status);
572+
test.add(txSeqA1T3,
573+
TransactionQueue::AddResult::ADD_STATUS_TRY_AGAIN_LATER);
559574
test.check(state);
560575
}
561576

@@ -990,14 +1005,12 @@ TEST_CASE_VERSIONS("TransactionQueue with PreconditionsV2",
9901005
test.add(txSeqA1S5MinSeqNum,
9911006
TransactionQueue::AddResult::ADD_STATUS_DUPLICATE);
9921007

993-
// try fill gap (invalid behavior), but account limit kicks in first
9941008
// try to fill in gap with a tx
995-
test.add(txSeqA1S2,
996-
TransactionQueue::AddResult::ADD_STATUS_TRY_AGAIN_LATER);
1009+
test.add(txSeqA1S2, TransactionQueue::AddResult::ADD_STATUS_ERROR);
9971010

9981011
// try to fill in gap with a minSeqNum tx
9991012
test.add(txSeqA1S4MinSeqNum,
1000-
TransactionQueue::AddResult::ADD_STATUS_TRY_AGAIN_LATER);
1013+
TransactionQueue::AddResult::ADD_STATUS_ERROR);
10011014

10021015
test.check({{{account1, 0, {txSeqA1S5MinSeqNum}}, {account2}}, {}});
10031016

‎src/overlay/Floodgate.cpp

+2-3
Original file line numberDiff line numberDiff line change
@@ -85,8 +85,7 @@ Floodgate::addRecord(StellarMessage const& msg, Peer::pointer peer, Hash& index)
8585

8686
// send message to anyone you haven't gotten it from
8787
bool
88-
Floodgate::broadcast(StellarMessage const& msg, bool force,
89-
std::optional<Hash> const& hash)
88+
Floodgate::broadcast(StellarMessage const& msg, std::optional<Hash> const& hash)
9089
{
9190
ZoneScoped;
9291
if (mShuttingDown)
@@ -102,7 +101,7 @@ Floodgate::broadcast(StellarMessage const& msg, bool force,
102101

103102
FloodRecord::pointer fr;
104103
auto result = mFloodMap.find(index);
105-
if (result == mFloodMap.end() || force)
104+
if (result == mFloodMap.end())
106105
{ // no one has sent us this message / start from scratch
107106
fr = std::make_shared<FloodRecord>(
108107
mApp.getHerder().trackingConsensusLedgerIndex(), Peer::pointer());

‎src/overlay/Floodgate.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ class Floodgate
6060

6161
// returns true if msg was sent to at least one peer
6262
// The hash required for transactions
63-
bool broadcast(StellarMessage const& msg, bool force,
63+
bool broadcast(StellarMessage const& msg,
6464
std::optional<Hash> const& hash = std::nullopt);
6565

6666
// returns the list of peers that sent us the item with hash `msgID`

‎src/overlay/OverlayManager.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ class OverlayManager
7777
// When passing a transaction message,
7878
// the hash of TransactionEnvelope must be passed also for pull mode.
7979
virtual bool
80-
broadcastMessage(StellarMessage const& msg, bool force = false,
80+
broadcastMessage(StellarMessage const& msg,
8181
std::optional<Hash> const hash = std::nullopt) = 0;
8282

8383
// Make a note in the FloodGate that a given peer has provided us with a

‎src/overlay/OverlayManagerImpl.cpp

+2-2
Original file line numberDiff line numberDiff line change
@@ -1151,11 +1151,11 @@ OverlayManagerImpl::forgetFloodedMsg(Hash const& msgID)
11511151
}
11521152

11531153
bool
1154-
OverlayManagerImpl::broadcastMessage(StellarMessage const& msg, bool force,
1154+
OverlayManagerImpl::broadcastMessage(StellarMessage const& msg,
11551155
std::optional<Hash> const hash)
11561156
{
11571157
ZoneScoped;
1158-
auto res = mFloodGate.broadcast(msg, force, hash);
1158+
auto res = mFloodGate.broadcast(msg, hash);
11591159
if (res)
11601160
{
11611161
mOverlayMetrics.mMessagesBroadcast.Mark();

‎src/overlay/OverlayManagerImpl.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,7 @@ class OverlayManagerImpl : public OverlayManager
139139
Hash& msgID) override;
140140
void forgetFloodedMsg(Hash const& msgID) override;
141141
bool
142-
broadcastMessage(StellarMessage const& msg, bool force = false,
142+
broadcastMessage(StellarMessage const& msg,
143143
std::optional<Hash> const hash = std::nullopt) override;
144144
void connectTo(PeerBareAddress const& address) override;
145145

‎src/overlay/SurveyManager.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -345,7 +345,7 @@ SurveyManager::processTopologyRequest(SurveyRequestMessage const& request) const
345345
void
346346
SurveyManager::broadcast(StellarMessage const& msg) const
347347
{
348-
mApp.getOverlayManager().broadcastMessage(msg, false);
348+
mApp.getOverlayManager().broadcastMessage(msg);
349349
}
350350

351351
void

‎src/overlay/test/FloodTests.cpp

+11-8
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ TEST_CASE("Flooding", "[flood][overlay][acceptance]")
160160
auto msg = tx1->toStellarMessage();
161161
auto res = inApp->getHerder().recvTransaction(tx1, false);
162162
REQUIRE(res == TransactionQueue::AddResult::ADD_STATUS_PENDING);
163-
inApp->getOverlayManager().broadcastMessage(msg, false,
163+
inApp->getOverlayManager().broadcastMessage(msg,
164164
tx1->getFullHash());
165165
};
166166

@@ -171,11 +171,14 @@ TEST_CASE("Flooding", "[flood][overlay][acceptance]")
171171

172172
for (auto const& s : sources)
173173
{
174-
okCount += (herder.getTransactionQueue()
175-
.getAccountTransactionQueueInfo(s)
176-
.mMaxSeq == expectedSeq)
177-
? 1
178-
: 0;
174+
auto accState =
175+
herder.getTransactionQueue().getAccountTransactionQueueInfo(
176+
s);
177+
auto seqNum = accState.mTransaction
178+
? accState.mTransaction->mTx->getSeqNum()
179+
: 0;
180+
181+
okCount += !!(seqNum == expectedSeq);
179182
}
180183
bool res = okCount == sources.size();
181184
LOG_DEBUG(DEFAULT_LOG, "{}{}{} / {} authenticated peers: {}",
@@ -277,7 +280,7 @@ TEST_CASE("Flooding", "[flood][overlay][acceptance]")
277280
auto before =
278281
overlaytestutils::getAdvertisedHashCount(node);
279282
node->getOverlayManager().broadcastMessage(
280-
testTransaction->toStellarMessage(), false,
283+
testTransaction->toStellarMessage(),
281284
testTransaction->getFullHash());
282285
REQUIRE(before ==
283286
overlaytestutils::getAdvertisedHashCount(node));
@@ -301,7 +304,7 @@ TEST_CASE("Flooding", "[flood][overlay][acceptance]")
301304
auto before =
302305
overlaytestutils::getAdvertisedHashCount(node);
303306
node->getOverlayManager().broadcastMessage(
304-
testTransaction->toStellarMessage(), false,
307+
testTransaction->toStellarMessage(),
305308
testTransaction->getFullHash());
306309

307310
REQUIRE(before + 1 ==

‎src/overlay/test/OverlayManagerTests.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -295,7 +295,7 @@ class OverlayManagerTests
295295
}
296296
}
297297
auto broadcastTxnMsg = [&](auto msg) {
298-
pm.broadcastMessage(msg, false, xdrSha256(msg.transaction()));
298+
pm.broadcastMessage(msg, xdrSha256(msg.transaction()));
299299
};
300300
broadcastTxnMsg(AtoB);
301301
crank(10);

‎src/simulation/LoadGenerator.cpp

+1-2
Original file line numberDiff line numberDiff line change
@@ -2027,8 +2027,7 @@ LoadGenerator::execute(TransactionFramePtr& txf, LoadGenMode mode,
20272027
}
20282028
else
20292029
{
2030-
mApp.getOverlayManager().broadcastMessage(msg, false,
2031-
txf->getFullHash());
2030+
mApp.getOverlayManager().broadcastMessage(msg, txf->getFullHash());
20322031
}
20332032

20342033
return status;

0 commit comments

Comments
 (0)