Merge pull request #715 from charlesrocket/backports-wGPXUHk

LLMQ backports

src/llmq/quorums_chainlocks.cpp

@@ -108,7 +108,7 @@ void CChainLocksHandler::ProcessMessage(CNode* pfrom, const std::string& strComm
 
 void CChainLocksHandler::ProcessNewChainLock(NodeId from, const llmq::CChainLockSig& clsig, const uint256& hash)
 {
- {
+ if (from != -1) {
  LOCK(cs_main);
  EraseObjectRequest(from, CInv(MSG_CLSIG, hash));
  }

src/llmq/quorums_dkgsession.cpp

@@ -260,6 +260,8 @@ bool CDKGSession::PreVerifyMessage(const uint256& hash, const CDKGContribution&
 
 void CDKGSession::ReceiveMessage(const uint256& hash, const CDKGContribution& qc, bool& retBan)
 {
+ LOCK(cs_pending);
+
  CDKGLogger logger(*this, __func__);
 
  retBan = false;
@@ -359,6 +361,8 @@ void CDKGSession::ReceiveMessage(const uint256& hash, const CDKGContribution& qc
 // See CBLSWorker::VerifyContributionShares for more details.
 void CDKGSession::VerifyPendingContributions()
 {
+ AssertLockHeld(cs_pending);
+
  CDKGLogger logger(*this, __func__);
 
  cxxtimer::Timer t1(true);
@@ -412,7 +416,10 @@ void CDKGSession::VerifyAndComplain(CDKGPendingMessages& pendingMessages)
  return;
  }
 
- VerifyPendingContributions();
+ {
+ LOCK(cs_pending);
+ VerifyPendingContributions();
+ }
 
  CDKGLogger logger(*this, __func__);
 

src/llmq/quorums_dkgsession.h

@@ -277,6 +277,7 @@ class CDKGSession
  std::map<uint256, CDKGJustification> justifications;
  std::map<uint256, CDKGPrematureCommitment> prematureCommitments;
 
+ mutable CCriticalSection cs_pending;
  std::vector<size_t> pendingContributionVerifications;
 
  // filled by ReceivePrematureCommitment and used by FinalizeCommitments

src/llmq/quorums_dkgsessionhandler.cpp

@@ -29,30 +29,29 @@ void CDKGPendingMessages::PushPendingMessage(NodeId from, CDataStream& vRecv)
  // this will also consume the data, even if we bail out early
  auto pm = std::make_shared<CDataStream>(std::move(vRecv));
 
- {
- LOCK(cs);
-
- if (messagesPerNode[from] >= maxMessagesPerNode) {
- // TODO ban?
- LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from);
- return;
- }
- messagesPerNode[from]++;
- }
-
  CHashWriter hw(SER_GETHASH, 0);
  hw.write(pm->data(), pm->size());
  uint256 hash = hw.GetHash();
 
- LOCK2(cs_main, cs);
+ if (from != -1) {
+ LOCK(cs_main);
+ EraseObjectRequest(from, CInv(invType, hash));
+ }
+
+ LOCK(cs);
+
+ if (messagesPerNode[from] >= maxMessagesPerNode) {
+ // TODO ban?
+ LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from);
+ return;
+ }
+ messagesPerNode[from]++;
 
  if (!seenMessages.emplace(hash).second) {
  LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from);
  return;
  }
 
- EraseObjectRequest(from, CInv(invType, hash));
-
  pendingMessages.emplace_back(std::make_pair(from, std::move(pm)));
 }
 
@@ -454,10 +453,6 @@ bool ProcessPendingMessageBatch(CDKGSession& session, CDKGPendingMessages& pendi
  const auto& msg = *p.second;
 
  auto hash = ::SerializeHash(msg);
- {
- LOCK(cs_main);
- EraseObjectRequest(p.first, CInv(MessageType, hash));
- }
 
  bool ban = false;
  if (!session.PreVerifyMessage(hash, msg, ban)) {

src/llmq/quorums_instantsend.cpp

@@ -680,6 +680,13 @@ void CInstantSendManager::ProcessMessage(CNode* pfrom, const std::string& strCom
 
 void CInstantSendManager::ProcessMessageInstantSendLock(CNode* pfrom, const llmq::CInstantSendLock& islock, CConnman& connman)
 {
+ auto hash = ::SerializeHash(islock);
+
+ {
+ LOCK(cs_main);
+ EraseObjectRequest(pfrom->GetId(), CInv(MSG_ISLOCK, hash));
+ }
+
  bool ban = false;
  if (!PreVerifyInstantSendLock(pfrom->GetId(), islock, ban)) {
  if (ban) {
@@ -689,8 +696,6 @@ void CInstantSendManager::ProcessMessageInstantSendLock(CNode* pfrom, const llmq
  return;
  }
 
- auto hash = ::SerializeHash(islock);
-
  LOCK(cs);
  if (db.GetInstantSendLockByHash(hash) != nullptr) {
  return;
@@ -883,11 +888,6 @@ std::unordered_set<uint256> CInstantSendManager::ProcessPendingInstantSendLocks(
 
 void CInstantSendManager::ProcessInstantSendLock(NodeId from, const uint256& hash, const CInstantSendLock& islock)
 {
- {
- LOCK(cs_main);
- EraseObjectRequest(from, CInv(MSG_ISLOCK, hash));
- }
-
  CTransactionRef tx;
  uint256 hashBlock;
  const CBlockIndex* pindexMined = nullptr;

src/llmq/quorums_signing.cpp

@@ -482,6 +482,11 @@ void CSigningManager::ProcessMessage(CNode* pfrom, const std::string& strCommand
 
 void CSigningManager::ProcessMessageRecoveredSig(CNode* pfrom, const CRecoveredSig& recoveredSig, CConnman& connman)
 {
+ {
+ LOCK(cs_main);
+ EraseObjectRequest(pfrom->GetId(), CInv(MSG_QUORUM_RECOVERED_SIG, recoveredSig.GetHash()));
+ }
+
  bool ban = false;
  if (!PreVerifyRecoveredSig(pfrom->GetId(), recoveredSig, ban)) {
  if (ban) {
@@ -682,11 +687,6 @@ void CSigningManager::ProcessRecoveredSig(NodeId nodeId, const CRecoveredSig& re
 {
  auto llmqType = (Consensus::LLMQType)recoveredSig.llmqType;
 
- {
- LOCK(cs_main);
- EraseObjectRequest(nodeId, CInv(MSG_QUORUM_RECOVERED_SIG, recoveredSig.GetHash()));
- }
-
  if (db.HasRecoveredSigForHash(recoveredSig.GetHash())) {
  return;
  }

src/net_processing.cpp

@@ -2734,6 +2734,11 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
  CInv inv(nInvType, tx.GetHash());
  pfrom->AddInventoryKnown(inv);
 
+ {
+ LOCK(cs_main);
+ EraseObjectRequest(pfrom->GetId(), inv);
+ }
+
  // Process custom logic, no matter if tx will be accepted to mempool later or not
  if (nInvType == MSG_DSTX) {
  uint256 hashTx = tx.GetHash();
@@ -2786,8 +2791,6 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
  bool fMissingInputs = false;
  CValidationState state;
 
- EraseObjectRequest(pfrom->GetId(), inv);
-
  if (!AlreadyHave(inv) && AcceptToMemoryPool(mempool, state, ptx, &fMissingInputs /* pfMissingInputs */,
  false /* bypass_limits */, 0 /* nAbsurdFee */)) {
  // Process custom txes, this changes AlreadyHave to "true"