diff --git a/src/avalanche/peermanager.cpp b/src/avalanche/peermanager.cpp index a9f049939..dbbda5a07 100644 --- a/src/avalanche/peermanager.cpp +++ b/src/avalanche/peermanager.cpp @@ -1,565 +1,565 @@ // Copyright (c) 2020 The Bitcoin developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #include #include #include #include #include // For ChainstateActive() #include #include namespace avalanche { bool PeerManager::addNode(NodeId nodeid, const ProofId &proofid) { - auto &pview = peers.get(); + auto &pview = peers.get(); auto it = pview.find(proofid); if (it == pview.end()) { // If the node exists, it is actually updating its proof to an unknown // one. In this case we need to remove it so it is not both active and // pending at the same time. removeNode(nodeid); pendingNodes.emplace(proofid, nodeid); return false; } return addOrUpdateNode(peers.project<0>(it), nodeid); } bool PeerManager::addOrUpdateNode(const PeerSet::iterator &it, NodeId nodeid) { assert(it != peers.end()); const PeerId peerid = it->peerid; auto nit = nodes.find(nodeid); if (nit == nodes.end()) { if (!nodes.emplace(nodeid, peerid).second) { return false; } } else { const PeerId oldpeerid = nit->peerid; if (!nodes.modify(nit, [&](Node &n) { n.peerid = peerid; })) { return false; } // We actually have this node already, we need to update it. bool success = removeNodeFromPeer(peers.find(oldpeerid)); assert(success); } bool success = addNodeToPeer(it); assert(success); // If the added node was in the pending set, remove it pendingNodes.get().erase(nodeid); return true; } bool PeerManager::addNodeToPeer(const PeerSet::iterator &it) { assert(it != peers.end()); return peers.modify(it, [&](Peer &p) { if (p.node_count++ > 0) { // We are done. return; } // We need to allocate this peer. p.index = uint32_t(slots.size()); const uint32_t score = p.getScore(); const uint64_t start = slotCount; slots.emplace_back(start, score, it->peerid); slotCount = start + score; }); } bool PeerManager::removeNode(NodeId nodeid) { auto it = nodes.find(nodeid); if (it == nodes.end()) { return false; } const PeerId peerid = it->peerid; nodes.erase(it); pendingNodes.get().erase(nodeid); // Keep the track of the reference count. bool success = removeNodeFromPeer(peers.find(peerid)); assert(success); return true; } bool PeerManager::removeNodeFromPeer(const PeerSet::iterator &it, uint32_t count) { // It is possible for nodes to be dangling. If there was an inflight query // when the peer gets removed, the node was not erased. In this case there // is nothing to do. if (it == peers.end()) { return true; } assert(count <= it->node_count); if (count == 0) { // This is a NOOP. return false; } const uint32_t new_count = it->node_count - count; if (!peers.modify(it, [&](Peer &p) { p.node_count = new_count; })) { return false; } if (new_count > 0) { // We are done. return true; } // There are no more node left, we need to cleanup. const size_t i = it->index; assert(i < slots.size()); if (i + 1 == slots.size()) { slots.pop_back(); slotCount = slots.empty() ? 0 : slots.back().getStop(); } else { fragmentation += slots[i].getScore(); slots[i] = slots[i].withPeerId(NO_PEER); } return true; } bool PeerManager::updateNextRequestTime(NodeId nodeid, TimePoint timeout) { auto it = nodes.find(nodeid); if (it == nodes.end()) { return false; } return nodes.modify(it, [&](Node &n) { n.nextRequestTime = timeout; }); } static bool isOrphanState(const ProofValidationState &state) { return state.GetResult() == ProofValidationResult::MISSING_UTXO || state.GetResult() == ProofValidationResult::HEIGHT_MISMATCH; } bool PeerManager::registerProof(const ProofRef &proof) { if (exists(proof->getId())) { // The proof is already registered, or orphaned. return false; } // Check the proof's validity. ProofValidationState state; bool valid = [&](ProofValidationState &state) { LOCK(cs_main); const CCoinsViewCache &coins = ::ChainstateActive().CoinsTip(); return proof->verify(state, coins); }(state); if (!valid) { if (isOrphanState(state)) { orphanProofs.addProof(proof); } // Reject invalid proof. return false; } return createPeer(proof); } NodeId PeerManager::selectNode() { for (int retry = 0; retry < SELECT_NODE_MAX_RETRY; retry++) { const PeerId p = selectPeer(); // If we cannot find a peer, it may be due to the fact that it is // unlikely due to high fragmentation, so compact and retry. if (p == NO_PEER) { compact(); continue; } // See if that peer has an available node. auto &nview = nodes.get(); auto it = nview.lower_bound(boost::make_tuple(p, TimePoint())); if (it != nview.end() && it->peerid == p && it->nextRequestTime <= std::chrono::steady_clock::now()) { return it->nodeid; } } return NO_NODE; } void PeerManager::updatedBlockTip() { std::vector invalidPeers; std::vector newOrphans; { LOCK(cs_main); const CCoinsViewCache &coins = ::ChainstateActive().CoinsTip(); for (const auto &p : peers) { ProofValidationState state; if (!p.proof->verify(state, coins)) { if (isOrphanState(state)) { newOrphans.push_back(p.proof); } invalidPeers.push_back(p.peerid); } } } // Remove the invalid peers before the orphans rescan. This makes it // possible to pull back proofs with utxos that conflicted with these // invalid peers. for (const auto &pid : invalidPeers) { removePeer(pid); } orphanProofs.rescan(*this); for (auto &p : newOrphans) { orphanProofs.addProof(p); } } ProofRef PeerManager::getProof(const ProofId &proofid) const { ProofRef proof = nullptr; forPeer(proofid, [&](const Peer &p) { proof = p.proof; return true; }); if (!proof) { proof = orphanProofs.getProof(proofid); } return proof; } bool PeerManager::isValid(const ProofId &proofid) const { - auto &pview = peers.get(); + auto &pview = peers.get(); return pview.find(proofid) != pview.end(); } bool PeerManager::isOrphan(const ProofId &proofid) const { return orphanProofs.getProof(proofid) != nullptr; } bool PeerManager::createPeer(const ProofRef &proof) { assert(proof); const ProofId &proofid = proof->getId(); if (isValid(proofid)) { return false; } // New peer means new peerid! const PeerId peerid = nextPeerId++; // Attach UTXOs to this proof. std::unordered_set conflicting_proofs; for (const auto &s : proof->getStakes()) { auto p = utxos.emplace(s.getStake().getUTXO(), proof); if (!p.second) { // We have a collision with an existing proof. conflicting_proofs.insert(p.first->second); } } // For now, if there is a conflict, just cleanup the mess. if (conflicting_proofs.size() > 0) { for (const auto &s : proof->getStakes()) { auto it = utxos.find(s.getStake().getUTXO()); assert(it != utxos.end()); // We need to delete that one. if (it->second->getId() == proofid) { utxos.erase(it); } } // Orphan the proof so it can be pulled back if the conflicting ones are // invalidated. orphanProofs.addProof(proof); return false; } // We have no peer for this proof, time to create it. auto inserted = peers.emplace(peerid, proof); assert(inserted.second); // If there are nodes waiting for this proof, add them auto &pendingNodesView = pendingNodes.get(); auto range = pendingNodesView.equal_range(proofid); // We want to update the nodes then remove them from the pending set. That // will invalidate the range iterators, so we need to save the node ids // first before we can loop over them. std::vector nodeids; nodeids.reserve(std::distance(range.first, range.second)); std::transform(range.first, range.second, std::back_inserter(nodeids), [](const PendingNode &n) { return n.nodeid; }); for (const NodeId &nodeid : nodeids) { addOrUpdateNode(inserted.first, nodeid); } return true; } bool PeerManager::removePeer(const PeerId peerid) { auto it = peers.find(peerid); if (it == peers.end()) { return false; } // Remove all nodes from this peer. removeNodeFromPeer(it, it->node_count); auto &nview = nodes.get(); // Add the nodes to the pending set auto range = nview.equal_range(peerid); for (auto &nit = range.first; nit != range.second; ++nit) { pendingNodes.emplace(it->getProofId(), nit->nodeid); }; // Remove nodes associated with this peer, unless their timeout is still // active. This ensure that we don't overquery them in case they are // subsequently added to another peer. nview.erase(nview.lower_bound(boost::make_tuple(peerid, TimePoint())), nview.upper_bound(boost::make_tuple( peerid, std::chrono::steady_clock::now()))); // Release UTXOs attached to this proof. for (const auto &s : it->proof->getStakes()) { bool deleted = utxos.erase(s.getStake().getUTXO()) > 0; assert(deleted); } m_unbroadcast_proofids.erase(it->proof->getId()); peers.erase(it); return true; } PeerId PeerManager::selectPeer() const { if (slots.empty() || slotCount == 0) { return NO_PEER; } const uint64_t max = slotCount; for (int retry = 0; retry < SELECT_PEER_MAX_RETRY; retry++) { size_t i = selectPeerImpl(slots, GetRand(max), max); if (i != NO_PEER) { return i; } } return NO_PEER; } uint64_t PeerManager::compact() { // There is nothing to compact. if (fragmentation == 0) { return 0; } std::vector newslots; newslots.reserve(peers.size()); uint64_t prevStop = 0; uint32_t i = 0; for (auto it = peers.begin(); it != peers.end(); it++) { if (it->node_count == 0) { continue; } newslots.emplace_back(prevStop, it->getScore(), it->peerid); prevStop = slots[i].getStop(); if (!peers.modify(it, [&](Peer &p) { p.index = i++; })) { return 0; } } slots = std::move(newslots); const uint64_t saved = slotCount - prevStop; slotCount = prevStop; fragmentation = 0; return saved; } bool PeerManager::verify() const { uint64_t prevStop = 0; for (size_t i = 0; i < slots.size(); i++) { const Slot &s = slots[i]; // Slots must be in correct order. if (s.getStart() < prevStop) { return false; } prevStop = s.getStop(); // If this is a dead slot, then nothing more needs to be checked. if (s.getPeerId() == NO_PEER) { continue; } // We have a live slot, verify index. auto it = peers.find(s.getPeerId()); if (it == peers.end() || it->index != i) { return false; } } std::unordered_set peersUtxos; for (const auto &p : peers) { // A peer should have a proof attached if (!p.proof) { return false; } // Check utxos consistency for (const auto &ss : p.proof->getStakes()) { auto it = utxos.find(ss.getStake().getUTXO()); if (it == utxos.end()) { return false; } if (it->second->getId() != p.getProofId()) { return false; } if (!peersUtxos.emplace(it->first).second) { return false; } } // Count node attached to this peer. const auto count_nodes = [&]() { size_t count = 0; auto &nview = nodes.get(); auto begin = nview.lower_bound(boost::make_tuple(p.peerid, TimePoint())); auto end = nview.upper_bound(boost::make_tuple(p.peerid + 1, TimePoint())); for (auto it = begin; it != end; ++it) { count++; } return count; }; if (p.node_count != count_nodes()) { return false; } // If there are no nodes attached to this peer, then we are done. if (p.node_count == 0) { continue; } // The index must point to a slot refering to this peer. if (p.index >= slots.size() || slots[p.index].getPeerId() != p.peerid) { return false; } // If the score do not match, same thing. if (slots[p.index].getScore() != p.getScore()) { return false; } } // Check there is no dangling utxo for (const auto &[outpoint, proof] : utxos) { if (!peersUtxos.count(outpoint)) { return false; } } return true; } PeerId selectPeerImpl(const std::vector &slots, const uint64_t slot, const uint64_t max) { assert(slot <= max); size_t begin = 0, end = slots.size(); uint64_t bottom = 0, top = max; // Try to find the slot using dichotomic search. while ((end - begin) > 8) { // The slot we picked in not allocated. if (slot < bottom || slot >= top) { return NO_PEER; } // Guesstimate the position of the slot. size_t i = begin + ((slot - bottom) * (end - begin) / (top - bottom)); assert(begin <= i && i < end); // We have a match. if (slots[i].contains(slot)) { return slots[i].getPeerId(); } // We undershooted. if (slots[i].precedes(slot)) { begin = i + 1; if (begin >= end) { return NO_PEER; } bottom = slots[begin].getStart(); continue; } // We overshooted. if (slots[i].follows(slot)) { end = i; top = slots[end].getStart(); continue; } // We have an unalocated slot. return NO_PEER; } // Enough of that nonsense, let fallback to linear search. for (size_t i = begin; i < end; i++) { // We have a match. if (slots[i].contains(slot)) { return slots[i].getPeerId(); } } // We failed to find a slot, retry. return NO_PEER; } void PeerManager::addUnbroadcastProof(const ProofId &proofid) { // The proof should be valid if (isValid(proofid)) { m_unbroadcast_proofids.insert(proofid); } } void PeerManager::removeUnbroadcastProof(const ProofId &proofid) { m_unbroadcast_proofids.erase(proofid); } } // namespace avalanche diff --git a/src/avalanche/peermanager.h b/src/avalanche/peermanager.h index 8a044787f..4d09f3556 100644 --- a/src/avalanche/peermanager.h +++ b/src/avalanche/peermanager.h @@ -1,287 +1,287 @@ // Copyright (c) 2020 The Bitcoin developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #ifndef BITCOIN_AVALANCHE_PEERMANAGER_H #define BITCOIN_AVALANCHE_PEERMANAGER_H #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include namespace avalanche { /** * Maximum number of stakes in the orphanProofs. * Benchmarking on a consumer grade computer shows that 10000 stakes can be * verified in less than 1 second. */ static constexpr size_t AVALANCHE_ORPHANPROOFPOOL_SIZE = 10000; class Delegation; namespace { struct TestPeerManager; } struct Slot { private: uint64_t start; uint32_t score; PeerId peerid; public: Slot(uint64_t startIn, uint32_t scoreIn, PeerId peeridIn) : start(startIn), score(scoreIn), peerid(peeridIn) {} Slot withStart(uint64_t startIn) const { return Slot(startIn, score, peerid); } Slot withScore(uint64_t scoreIn) const { return Slot(start, scoreIn, peerid); } Slot withPeerId(PeerId peeridIn) const { return Slot(start, score, peeridIn); } uint64_t getStart() const { return start; } uint64_t getStop() const { return start + score; } uint32_t getScore() const { return score; } PeerId getPeerId() const { return peerid; } bool contains(uint64_t slot) const { return getStart() <= slot && slot < getStop(); } bool precedes(uint64_t slot) const { return slot >= getStop(); } bool follows(uint64_t slot) const { return getStart() > slot; } }; struct Peer { PeerId peerid; uint32_t index = -1; uint32_t node_count = 0; ProofRef proof; // The network stack uses timestamp in seconds, so we oblige. std::chrono::seconds registration_time; Peer(PeerId peerid_, ProofRef proof_) : peerid(peerid_), proof(std::move(proof_)), registration_time(GetTime()) {} const ProofId &getProofId() const { return proof->getId(); } uint32_t getScore() const { return proof->getScore(); } }; struct proof_index { using result_type = ProofId; result_type operator()(const Peer &p) const { return p.proof->getId(); } }; struct next_request_time {}; struct PendingNode { ProofId proofid; NodeId nodeid; PendingNode(ProofId proofid_, NodeId nodeid_) : proofid(proofid_), nodeid(nodeid_){}; }; struct by_proofid; struct by_nodeid; namespace bmi = boost::multi_index; class PeerManager { std::vector slots; uint64_t slotCount = 0; uint64_t fragmentation = 0; /** * Several nodes can make an avalanche peer. In this case, all nodes are * considered interchangeable parts of the same peer. */ using PeerSet = boost::multi_index_container< Peer, bmi::indexed_by< // index by peerid bmi::hashed_unique>, // index by proof - bmi::hashed_unique, proof_index, + bmi::hashed_unique, proof_index, SaltedProofIdHasher>>>; PeerId nextPeerId = 0; PeerSet peers; std::unordered_map utxos; using NodeSet = boost::multi_index_container< Node, bmi::indexed_by< // index by nodeid bmi::hashed_unique>, // sorted by peerid/nextRequestTime bmi::ordered_non_unique< bmi::tag, bmi::composite_key< Node, bmi::member, bmi::member>>>>; NodeSet nodes; using PendingNodeSet = boost::multi_index_container< PendingNode, bmi::indexed_by< // index by proofid bmi::hashed_non_unique< bmi::tag, bmi::member, SaltedProofIdHasher>, // index by nodeid bmi::hashed_unique< bmi::tag, bmi::member>>>; PendingNodeSet pendingNodes; static constexpr int SELECT_PEER_MAX_RETRY = 3; static constexpr int SELECT_NODE_MAX_RETRY = 3; /** * Tracks proof which for which the UTXO are unavailable. */ OrphanProofPool orphanProofs{AVALANCHE_ORPHANPROOFPOOL_SIZE}; /** * Track proof ids to broadcast */ std::unordered_set m_unbroadcast_proofids; public: /** * Node API. */ bool addNode(NodeId nodeid, const ProofId &proofid); bool removeNode(NodeId nodeid); // Update when a node is to be polled next. bool updateNextRequestTime(NodeId nodeid, TimePoint timeout); // Randomly select a node to poll. NodeId selectNode(); template bool forNode(NodeId nodeid, Callable &&func) const { auto it = nodes.find(nodeid); return it != nodes.end() && func(*it); } template void forEachNode(const Peer &peer, Callable &&func) const { auto &nview = nodes.get(); auto range = nview.equal_range(peer.peerid); for (auto it = range.first; it != range.second; ++it) { func(*it); } } /** * Proof and Peer related API. */ bool registerProof(const ProofRef &proof); bool exists(const ProofId &proofid) const { return getProof(proofid) != nullptr; } template bool forPeer(const ProofId &proofid, Callable &&func) const { - auto &pview = peers.get(); + auto &pview = peers.get(); auto it = pview.find(proofid); return it != pview.end() && func(*it); } template void forEachPeer(Callable &&func) const { for (const auto &p : peers) { func(p); } } /** * Update the peer set when a new block is connected. */ void updatedBlockTip(); /** * Proof broadcast API. */ void addUnbroadcastProof(const ProofId &proofid); void removeUnbroadcastProof(const ProofId &proofid); auto getUnbroadcastProofs() const { return m_unbroadcast_proofids; } /**************************************************** * Functions which are public for testing purposes. * ****************************************************/ /** * Remove an existing peer. */ bool removePeer(const PeerId peerid); /** * Randomly select a peer to poll. */ PeerId selectPeer() const; /** * Trigger maintenance of internal data structures. * Returns how much slot space was saved after compaction. */ uint64_t compact(); /** * Perform consistency check on internal data structures. */ bool verify() const; // Accessors. uint64_t getSlotCount() const { return slotCount; } uint64_t getFragmentation() const { return fragmentation; } ProofRef getProof(const ProofId &proofid) const; bool isValid(const ProofId &proofid) const; bool isOrphan(const ProofId &proofid) const; private: bool createPeer(const ProofRef &proof); bool addOrUpdateNode(const PeerSet::iterator &it, NodeId nodeid); bool addNodeToPeer(const PeerSet::iterator &it); bool removeNodeFromPeer(const PeerSet::iterator &it, uint32_t count = 1); friend struct ::avalanche::TestPeerManager; }; /** * Internal methods that are exposed for testing purposes. */ PeerId selectPeerImpl(const std::vector &slots, const uint64_t slot, const uint64_t max); } // namespace avalanche #endif // BITCOIN_AVALANCHE_PEERMANAGER_H diff --git a/src/avalanche/test/peermanager_tests.cpp b/src/avalanche/test/peermanager_tests.cpp index 92dbb8ae0..44e7f500a 100644 --- a/src/avalanche/test/peermanager_tests.cpp +++ b/src/avalanche/test/peermanager_tests.cpp @@ -1,945 +1,945 @@ // Copyright (c) 2020 The Bitcoin developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #include #include #include #include #include #include