diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -599,7 +599,6 @@ torcontrol.cpp txdb.cpp txmempool.cpp - txrequest.cpp validation.cpp validationinterface.cpp versionbits.cpp diff --git a/src/txrequest.h b/src/txrequest.h --- a/src/txrequest.h +++ b/src/txrequest.h @@ -5,13 +5,19 @@ #ifndef BITCOIN_TXREQUEST_H #define BITCOIN_TXREQUEST_H +#include #include // For NodeId #include +#include +#include -#include -#include +#include +#include +#include #include +#include +#include /** * Data structure to keep track of, and schedule, transaction downloads from @@ -118,14 +124,575 @@ * (amortized O(1) per announcement). */ class TxRequestTracker { - // Avoid littering this header file with implementation details. - class Impl; - const std::unique_ptr m_impl; + /** + * The various states a (txid, peer) pair can be in. + * + * Note that CANDIDATE is split up into 3 substates (DELAYED, BEST, READY), + * allowing more efficient implementation. Also note that the sorting order + * of ByTxIdView relies on the specific order of values in this enum. + * + * Expected behaviour is: + * - When first announced by a peer, the state is CANDIDATE_DELAYED until + * reqtime is reached. + * - Announcements that have reached their reqtime but not been requested + * will be either CANDIDATE_READY or CANDIDATE_BEST. Neither of those has an + * expiration time; they remain in that state until they're requested or + * no longer needed. CANDIDATE_READY announcements are promoted to + * CANDIDATE_BEST when they're the best one left. + * - When requested, an announcement will be in state REQUESTED until + * expiry is reached. + * - If expiry is reached, or the peer replies to the request (either with + * NOTFOUND or the tx), the state becomes COMPLETED. + */ + enum class State : uint8_t { + /** A CANDIDATE announcement whose reqtime is in the future. */ + CANDIDATE_DELAYED, + /** + * A CANDIDATE announcement that's not CANDIDATE_DELAYED or + * CANDIDATE_BEST. + */ + CANDIDATE_READY, + /** + * The best CANDIDATE for a given txid; only if there is no REQUESTED + * announcement already for that txid. The CANDIDATE_BEST is the + * highest-priority announcement among all CANDIDATE_READY (and _BEST) + * ones for that txid. + */ + CANDIDATE_BEST, + /** A REQUESTED announcement. */ + REQUESTED, + /** A COMPLETED announcement. */ + COMPLETED, + }; + + //! Type alias for sequence numbers. + using SequenceNumber = uint64_t; + + /** + * An announcement. This is the data we track for each txid that is + * announced to us by each peer. + */ + struct Announcement { + /** TxId that was announced. */ + const TxId m_txid; + /** + * For CANDIDATE_{DELAYED,BEST,READY} the reqtime; for REQUESTED the + * expiry. + */ + std::chrono::microseconds m_time; + /** What peer the request was from. */ + const NodeId m_peer; + /** What sequence number this announcement has. */ + const SequenceNumber m_sequence : 60; + /** Whether the request is preferred. */ + const bool m_preferred : 1; + + /** + * What state this announcement is in. + * This is a uint8_t instead of a State to silence a GCC warning in + * versions prior to 8.4 and 9.3. See + * https://gcc.gnu.org/bugzilla/show_bug.cgi?id=61414 + */ + uint8_t m_state : 3; + + /** Convert m_state to a State enum. */ + State GetState() const { return static_cast(m_state); } + + /** Convert a State enum to a uint8_t and store it in m_state. */ + void SetState(State state) { m_state = static_cast(state); } + + /** + * Whether this announcement is selected. There can be at most 1 + * selected peer per txid. + */ + bool IsSelected() const { + return GetState() == State::CANDIDATE_BEST || + GetState() == State::REQUESTED; + } + + /** Whether this announcement is waiting for a certain time to pass. */ + bool IsWaiting() const { + return GetState() == State::REQUESTED || + GetState() == State::CANDIDATE_DELAYED; + } + + /** + * Whether this announcement can feasibly be selected if the current + * IsSelected() one disappears. + */ + bool IsSelectable() const { + return GetState() == State::CANDIDATE_READY || + GetState() == State::CANDIDATE_BEST; + } + + /** + * Construct a new announcement from scratch, initially in + * CANDIDATE_DELAYED state. + */ + Announcement(const TxId &txid, NodeId peer, bool preferred, + std::chrono::microseconds reqtime, SequenceNumber sequence) + : m_txid(txid), m_time(reqtime), m_peer(peer), m_sequence(sequence), + m_preferred(preferred), + m_state(static_cast(State::CANDIDATE_DELAYED)) {} + }; + + //! Type alias for priorities. + using Priority = uint64_t; + + /** + * A functor with embedded salt that computes priority of an announcement. + * + * Higher priorities are selected first. + */ + class PriorityComputer { + const uint64_t m_k0, m_k1; + + public: + explicit PriorityComputer(bool deterministic) + : m_k0{deterministic ? 0 : GetRand(0xFFFFFFFFFFFFFFFF)}, + m_k1{deterministic ? 0 : GetRand(0xFFFFFFFFFFFFFFFF)} {} + + Priority operator()(const TxId &txid, NodeId peer, + bool preferred) const { + uint64_t low_bits = CSipHasher(m_k0, m_k1) + .Write(txid.begin(), txid.size()) + .Write(peer) + .Finalize() >> + 1; + return low_bits | uint64_t{preferred} << 63; + } + + Priority operator()(const Announcement &ann) const { + return operator()(ann.m_txid, ann.m_peer, ann.m_preferred); + } + }; + + // Definitions for the 3 indexes used in the main data structure. + // + // Each index has a By* type to identify it, a By*View data type to + // represent the view of announcement it is sorted by, and an + // By*ViewExtractor type to convert an announcement into the By*View type. + // See + // https://www.boost.org/doc/libs/1_58_0/libs/multi_index/doc/reference/key_extraction.html#key_extractors + // for more information about the key extraction concept. + + // The ByPeer index is sorted by (peer, state == CANDIDATE_BEST, txid) + // + // Uses: + // * Looking up existing announcements by peer/txid, by checking both (peer, + // false, txid) and (peer, true, txid). + // * Finding all CANDIDATE_BEST announcements for a given peer in + // GetRequestable. + struct ByPeer {}; + using ByPeerView = std::tuple; + struct ByPeerViewExtractor { + using result_type = ByPeerView; + result_type operator()(const Announcement &ann) const { + return ByPeerView{ann.m_peer, + ann.GetState() == State::CANDIDATE_BEST, + ann.m_txid}; + } + }; + + // The ByTxId index is sorted by (txid, state, priority). + // + // Note: priority == 0 whenever state != CANDIDATE_READY. + // + // Uses: + // * Deleting all announcements with a given txid in ForgetTxId. + // * Finding the best CANDIDATE_READY to convert to CANDIDATE_BEST, when no + // other CANDIDATE_READY or REQUESTED announcement exists for that txid. + // * Determining when no more non-COMPLETED announcements for a given txid + // exist, so the COMPLETED ones can be deleted. + struct ByTxId {}; + using ByTxIdView = std::tuple; + class ByTxIdViewExtractor { + const PriorityComputer &m_computer; + + public: + ByTxIdViewExtractor(const PriorityComputer &computer) + : m_computer(computer) {} + using result_type = ByTxIdView; + result_type operator()(const Announcement &ann) const { + const Priority prio = (ann.GetState() == State::CANDIDATE_READY) + ? m_computer(ann) + : 0; + return ByTxIdView{ann.m_txid, ann.GetState(), prio}; + } + }; + + enum class WaitState { + //! Used for announcements that need efficient testing of "is their + //! timestamp in the future?". + FUTURE_EVENT, + //! Used for announcements whose timestamp is not relevant. + NO_EVENT, + //! Used for announcements that need efficient testing of "is their + //! timestamp in the past?". + PAST_EVENT, + }; + + static WaitState GetWaitState(const Announcement &ann) { + if (ann.IsWaiting()) { + return WaitState::FUTURE_EVENT; + } + if (ann.IsSelectable()) { + return WaitState::PAST_EVENT; + } + return WaitState::NO_EVENT; + } + + // The ByTime index is sorted by (wait_state, time). + // + // All announcements with a timestamp in the future can be found by + // iterating the index forward from the beginning. All announcements with a + // timestamp in the past can be found by iterating the index backwards from + // the end. + // + // Uses: + // * Finding CANDIDATE_DELAYED announcements whose reqtime has passed, and + // REQUESTED announcements whose expiry has passed. + // * Finding CANDIDATE_READY/BEST announcements whose reqtime is in the + // future + // (when the clock time went backwards). + struct ByTime {}; + using ByTimeView = std::pair; + struct ByTimeViewExtractor { + using result_type = ByTimeView; + result_type operator()(const Announcement &ann) const { + return ByTimeView{GetWaitState(ann), ann.m_time}; + } + }; + + /** + * Data type for the main data structure (Announcement objects with + * ByPeer/ByTxId/ByTime indexes). + */ + using Index = boost::multi_index_container< + Announcement, + boost::multi_index::indexed_by< + boost::multi_index::ordered_unique, + ByPeerViewExtractor>, + boost::multi_index::ordered_non_unique< + boost::multi_index::tag, ByTxIdViewExtractor>, + boost::multi_index::ordered_non_unique< + boost::multi_index::tag, ByTimeViewExtractor>>>; + + /** Helper type to simplify syntax of iterator types. */ + template + using Iter = typename Index::index::type::iterator; + + /** Per-peer statistics object. */ + struct PeerInfo { + //! Total number of announcements for this peer. + size_t m_total = 0; + //! Number of COMPLETED announcements for this peer. + size_t m_completed = 0; + //! Number of REQUESTED announcements for this peer. + size_t m_requested = 0; + + /** Compare two PeerInfo objects. Only used for sanity checking. */ + bool operator==(const PeerInfo &other) const { + return std::tie(m_total, m_completed, m_requested) == + std::tie(other.m_total, other.m_completed, + other.m_requested); + }; + }; + + /** Per-txid statistics object. Only used for sanity checking. */ + struct TxIdInfo { + //! Number of CANDIDATE_DELAYED announcements for this txid. + size_t m_candidate_delayed = 0; + //! Number of CANDIDATE_READY announcements for this txid. + size_t m_candidate_ready = 0; + //! Number of CANDIDATE_BEST announcements for this txid (at most one). + size_t m_candidate_best = 0; + //! Number of REQUESTED announcements for this txid (at most one; + //! mutually exclusive with CANDIDATE_BEST). + size_t m_requested = 0; + //! The priority of the CANDIDATE_BEST announcement if one exists, or + //! max() otherwise. + Priority m_priority_candidate_best = + std::numeric_limits::max(); + //! The highest priority of all CANDIDATE_READY announcements (or min() + //! if none exist). + Priority m_priority_best_candidate_ready = + std::numeric_limits::min(); + //! All peers we have an announcement for this txid for. + std::vector m_peers; + }; + + /** + * (Re)compute the PeerInfo map from the index. Only used for sanity + * checking. + */ + static std::unordered_map + RecomputePeerInfo(const Index &index) { + std::unordered_map ret; + for (const Announcement &ann : index) { + PeerInfo &info = ret[ann.m_peer]; + ++info.m_total; + info.m_requested += (ann.GetState() == State::REQUESTED); + info.m_completed += (ann.GetState() == State::COMPLETED); + } + return ret; + } + + /** Compute the TxIdInfo map. Only used for sanity checking. */ + static std::map + ComputeTxIdInfo(const Index &index, const PriorityComputer &computer) { + std::map ret; + for (const Announcement &ann : index) { + TxIdInfo &info = ret[ann.m_txid]; + // Classify how many announcements of each state we have for this + // txid. + info.m_candidate_delayed += + (ann.GetState() == State::CANDIDATE_DELAYED); + info.m_candidate_ready += + (ann.GetState() == State::CANDIDATE_READY); + info.m_candidate_best += (ann.GetState() == State::CANDIDATE_BEST); + info.m_requested += (ann.GetState() == State::REQUESTED); + // And track the priority of the best CANDIDATE_READY/CANDIDATE_BEST + // announcements. + if (ann.GetState() == State::CANDIDATE_BEST) { + info.m_priority_candidate_best = computer(ann); + } + if (ann.GetState() == State::CANDIDATE_READY) { + info.m_priority_best_candidate_ready = std::max( + info.m_priority_best_candidate_ready, computer(ann)); + } + // Also keep track of which peers this txid has an announcement for + // (so we can detect duplicates). + info.m_peers.push_back(ann.m_peer); + } + return ret; + } + + //! Wrapper around Index::...::erase that keeps m_peerinfo up to date. + template Iter Erase(Iter it) { + auto peerit = m_peerinfo.find(it->m_peer); + peerit->second.m_completed -= it->GetState() == State::COMPLETED; + peerit->second.m_requested -= it->GetState() == State::REQUESTED; + if (--peerit->second.m_total == 0) { + m_peerinfo.erase(peerit); + } + return m_index.get().erase(it); + } + + //! Wrapper around Index::...::modify that keeps m_peerinfo up to date. + template + void Modify(Iter it, Modifier modifier) { + auto peerit = m_peerinfo.find(it->m_peer); + peerit->second.m_completed -= it->GetState() == State::COMPLETED; + peerit->second.m_requested -= it->GetState() == State::REQUESTED; + m_index.get().modify(it, std::move(modifier)); + peerit->second.m_completed += it->GetState() == State::COMPLETED; + peerit->second.m_requested += it->GetState() == State::REQUESTED; + } + + //! Convert a CANDIDATE_DELAYED announcement into a CANDIDATE_READY. If this + //! makes it the new best CANDIDATE_READY (and no REQUESTED exists) and + //! better than the CANDIDATE_BEST (if any), it becomes the new + //! CANDIDATE_BEST. + void PromoteCandidateReady(Iter it) { + assert(it != m_index.get().end()); + assert(it->GetState() == State::CANDIDATE_DELAYED); + // Convert CANDIDATE_DELAYED to CANDIDATE_READY first. + Modify(it, [](Announcement &ann) { + ann.SetState(State::CANDIDATE_READY); + }); + // The following code relies on the fact that the ByTxId is sorted by + // txid, and then by state (first _DELAYED, then _READY, then + // _BEST/REQUESTED). Within the _READY announcements, the best one + // (highest priority) comes last. Thus, if an existing _BEST exists for + // the same txid that this announcement may be preferred over, it must + // immediately follow the newly created _READY. + auto it_next = std::next(it); + if (it_next == m_index.get().end() || + it_next->m_txid != it->m_txid || + it_next->GetState() == State::COMPLETED) { + // This is the new best CANDIDATE_READY, and there is no + // IsSelected() announcement for this txid already. + Modify(it, [](Announcement &ann) { + ann.SetState(State::CANDIDATE_BEST); + }); + } else if (it_next->GetState() == State::CANDIDATE_BEST) { + Priority priority_old = m_computer(*it_next); + Priority priority_new = m_computer(*it); + if (priority_new > priority_old) { + // There is a CANDIDATE_BEST announcement already, but this one + // is better. + Modify(it_next, [](Announcement &ann) { + ann.SetState(State::CANDIDATE_READY); + }); + Modify(it, [](Announcement &ann) { + ann.SetState(State::CANDIDATE_BEST); + }); + } + } + } + + //! Change the state of an announcement to something non-IsSelected(). If it + //! was IsSelected(), the next best announcement will be marked + //! CANDIDATE_BEST. + void ChangeAndReselect(Iter it, State new_state) { + assert(new_state == State::COMPLETED || + new_state == State::CANDIDATE_DELAYED); + assert(it != m_index.get().end()); + if (it->IsSelected() && it != m_index.get().begin()) { + auto it_prev = std::prev(it); + // The next best CANDIDATE_READY, if any, immediately precedes the + // REQUESTED or CANDIDATE_BEST announcement in the ByTxId index. + if (it_prev->m_txid == it->m_txid && + it_prev->GetState() == State::CANDIDATE_READY) { + // If one such CANDIDATE_READY exists (for this txid), convert + // it to CANDIDATE_BEST. + Modify(it_prev, [](Announcement &ann) { + ann.SetState(State::CANDIDATE_BEST); + }); + } + } + Modify( + it, [new_state](Announcement &ann) { ann.SetState(new_state); }); + } + + //! Check if 'it' is the only announcement for a given txid that isn't + //! COMPLETED. + bool IsOnlyNonCompleted(Iter it) { + assert(it != m_index.get().end()); + // Not allowed to call this on COMPLETED announcements. + assert(it->GetState() != State::COMPLETED); + + // This announcement has a predecessor that belongs to the same txid. + // Due to ordering, and the fact that 'it' is not COMPLETED, its + // predecessor cannot be COMPLETED here. + if (it != m_index.get().begin() && + std::prev(it)->m_txid == it->m_txid) { + return false; + } + + // This announcement has a successor that belongs to the same txid, + // and is not COMPLETED. + if (std::next(it) != m_index.get().end() && + std::next(it)->m_txid == it->m_txid && + std::next(it)->GetState() != State::COMPLETED) { + return false; + } + + return true; + } + + /** + * Convert any announcement to a COMPLETED one. If there are no + * non-COMPLETED announcements left for this txid, they are deleted. If + * this was a REQUESTED announcement, and there are other CANDIDATEs left, + * the best one is made CANDIDATE_BEST. Returns whether the announcement + * still exists. + */ + bool MakeCompleted(Iter it) { + assert(it != m_index.get().end()); + + // Nothing to be done if it's already COMPLETED. + if (it->GetState() == State::COMPLETED) { + return true; + } + + if (IsOnlyNonCompleted(it)) { + // This is the last non-COMPLETED announcement for this txid. + // Delete all. + TxId txid = it->m_txid; + do { + it = Erase(it); + } while (it != m_index.get().end() && it->m_txid == txid); + return false; + } + + // Mark the announcement COMPLETED, and select the next best + // announcement (the first CANDIDATE_READY) if needed. + ChangeAndReselect(it, State::COMPLETED); + + return true; + } + + //! Make the data structure consistent with a given point in time: + //! - REQUESTED annoucements with expiry <= now are turned into COMPLETED. + //! - CANDIDATE_DELAYED announcements with reqtime <= now are turned into + //! CANDIDATE_{READY,BEST}. + //! - CANDIDATE_{READY,BEST} announcements with reqtime > now are turned + //! into CANDIDATE_DELAYED. + void SetTimePoint(std::chrono::microseconds now, + std::vector> *expired) { + if (expired) { + expired->clear(); + } + // Iterate over all CANDIDATE_DELAYED and REQUESTED from old to new, as + // long as they're in the past, and convert them to CANDIDATE_READY and + // COMPLETED respectively. + while (!m_index.empty()) { + auto it = m_index.get().begin(); + if (it->GetState() == State::CANDIDATE_DELAYED && + it->m_time <= now) { + PromoteCandidateReady(m_index.project(it)); + } else if (it->GetState() == State::REQUESTED && + it->m_time <= now) { + if (expired) { + expired->emplace_back(it->m_peer, it->m_txid); + } + MakeCompleted(m_index.project(it)); + } else { + break; + } + } + + while (!m_index.empty()) { + // If time went backwards, we may need to demote CANDIDATE_BEST and + // CANDIDATE_READY announcements back to CANDIDATE_DELAYED. This is + // an unusual edge case, and unlikely to matter in production. + // However, it makes it much easier to specify and test + // TxRequestTracker::Impl's behaviour. + auto it = std::prev(m_index.get().end()); + if (it->IsSelectable() && it->m_time > now) { + ChangeAndReselect(m_index.project(it), + State::CANDIDATE_DELAYED); + } else { + break; + } + } + } + + //! The current sequence number. Increases for every announcement. This is + //! used to sort txid returned by GetRequestable in announcement order. + SequenceNumber m_current_sequence{0}; + + //! This tracker's priority computer. + const PriorityComputer m_computer; + + //! This tracker's main data structure. See SanityCheck() for the invariants + //! that apply to it. + Index m_index; + + //! Map with this tracker's per-peer statistics. + std::unordered_map m_peerinfo; public: //! Construct a TxRequestTracker. - explicit TxRequestTracker(bool deterministic = false); - ~TxRequestTracker(); + explicit TxRequestTracker(bool deterministic = false) + : m_computer(deterministic), + // Explicitly initialize m_index as we need to pass a reference to + // m_computer to ByTxIdViewExtractor. + m_index(boost::make_tuple( + boost::make_tuple(ByPeerViewExtractor(), std::less()), + boost::make_tuple(ByTxIdViewExtractor(m_computer), + std::less()), + boost::make_tuple(ByTimeViewExtractor(), + std::less()))) {} + + // Disable copying and assigning (a default copy won't work due the stateful + // ByTxIdViewExtractor). + TxRequestTracker(const TxRequestTracker &) = delete; + TxRequestTracker &operator=(const TxRequestTracker &) = delete; + + ~TxRequestTracker() = default; // Conceptually, the data structure consists of a collection of // "announcements", one for each peer/txid combination: @@ -154,14 +721,77 @@ * (whether it's CANDIDATE, REQUESTED, or COMPLETED). */ void ReceivedInv(NodeId peer, const TxId &txid, bool preferred, - std::chrono::microseconds reqtime); + std::chrono::microseconds reqtime) { + // Bail out if we already have a CANDIDATE_BEST announcement for this + // (txid, peer) combination. The case where there is a + // non-CANDIDATE_BEST announcement already will be caught by the + // uniqueness property of the ByPeer index when we try to emplace the + // new object below. + if (m_index.get().count(ByPeerView{peer, true, txid})) { + return; + } + + // Try creating the announcement with CANDIDATE_DELAYED state (which + // will fail due to the uniqueness of the ByPeer index if a + // non-CANDIDATE_BEST announcement already exists with the same txid + // and peer). Bail out in that case. + auto ret = m_index.get().emplace(txid, peer, preferred, reqtime, + m_current_sequence); + if (!ret.second) { + return; + } + + // Update accounting metadata. + ++m_peerinfo[peer].m_total; + ++m_current_sequence; + } /** * Deletes all announcements for a given peer. * * It should be called when a peer goes offline. */ - void DisconnectedPeer(NodeId peer); + void DisconnectedPeer(NodeId peer) { + auto &index = m_index.get(); + auto it = + index.lower_bound(ByPeerView{peer, false, TxId(uint256::ZERO)}); + while (it != index.end() && it->m_peer == peer) { + // Check what to continue with after this iteration. 'it' will be + // deleted in what follows, so we need to decide what to continue + // with afterwards. There are a number of cases to consider: + // - std::next(it) is end() or belongs to a different peer. In that + // case, this is the last iteration of the loop (denote this by + // setting it_next to end()). + // - 'it' is not the only non-COMPLETED announcement for its txid. + // This means it will be deleted, but no other Announcement + // objects will be modified. Continue with std::next(it) if it + // belongs to the same peer, but decide this ahead of time (as + // 'it' may change position in what follows). + // - 'it' is the only non-COMPLETED announcement for its txid. This + // means it will be deleted along with all other announcements for + // the same txid - which may include std::next(it). However, other + // than 'it', no announcements for the same peer can be affected + // (due to (peer, txid) uniqueness). In other words, the situation + // where std::next(it) is deleted can only occur if std::next(it) + // belongs to a different peer but the same txid as 'it'. This is + // covered by the first bulletpoint already, and we'll have set + // it_next to end(). + auto it_next = + (std::next(it) == index.end() || std::next(it)->m_peer != peer) + ? index.end() + : std::next(it); + // If the announcement isn't already COMPLETED, first make it + // COMPLETED (which will mark other CANDIDATEs as CANDIDATE_BEST, or + // delete all of a txid's announcements if no non-COMPLETED ones are + // left). + if (MakeCompleted(m_index.project(it))) { + // Then actually delete the announcement (unless it was already + // deleted by MakeCompleted). + Erase(it); + } + it = it_next; + } + } /** * Deletes all announcements for a given txid. @@ -170,7 +800,13 @@ * should ensure that new announcements for the same txid will not trigger * new ReceivedInv calls, at least in the short term after this call. */ - void ForgetTxId(const TxId &txid); + void ForgetTxId(const TxId &txid) { + auto it = m_index.get().lower_bound( + ByTxIdView{txid, State::CANDIDATE_DELAYED, 0}); + while (it != m_index.get().end() && it->m_txid == txid) { + it = Erase(it); + } + } /** * Find the txids to request now from peer. @@ -196,7 +832,35 @@ */ std::vector GetRequestable(NodeId peer, std::chrono::microseconds now, - std::vector> *expired = nullptr); + std::vector> *expired) { + // Move time. + SetTimePoint(now, expired); + + // Find all CANDIDATE_BEST announcements for this peer. + std::vector selected; + auto it_peer = m_index.get().lower_bound( + ByPeerView{peer, true, TxId(uint256::ZERO)}); + while (it_peer != m_index.get().end() && + it_peer->m_peer == peer && + it_peer->GetState() == State::CANDIDATE_BEST) { + selected.emplace_back(&*it_peer); + ++it_peer; + } + + // Sort by sequence number. + std::sort(selected.begin(), selected.end(), + [](const Announcement *a, const Announcement *b) { + return a->m_sequence < b->m_sequence; + }); + + // Convert to TxId and return. + std::vector ret; + ret.reserve(selected.size()); + std::transform(selected.begin(), selected.end(), + std::back_inserter(ret), + [](const Announcement *ann) { return ann->m_txid; }); + return ret; + } /** * Marks a transaction as requested, with a specified expiry. @@ -210,7 +874,67 @@ * we're no longer waiting for a response to it. */ void RequestedTx(NodeId peer, const TxId &txid, - std::chrono::microseconds expiry); + std::chrono::microseconds expiry) { + auto it = m_index.get().find(ByPeerView{peer, true, txid}); + if (it == m_index.get().end()) { + // There is no CANDIDATE_BEST announcement, look for a _READY or + // _DELAYED instead. If the caller only ever invokes RequestedTx + // with the values returned by GetRequestable, and no other + // non-const functions other than ForgetTxHash and GetRequestable in + // between, this branch will never execute (as txids returned by + // GetRequestable always correspond to CANDIDATE_BEST + // announcements). + + it = m_index.get().find(ByPeerView{peer, false, txid}); + if (it == m_index.get().end() || + (it->GetState() != State::CANDIDATE_DELAYED && + it->GetState() != State::CANDIDATE_READY)) { + // There is no CANDIDATE announcement tracked for this peer, so + // we have nothing to do. Either this txid wasn't tracked at + // all (and the caller should have called ReceivedInv), or it + // was already requested and/or completed for other reasons and + // this is just a superfluous RequestedTx call. + return; + } + + // Look for an existing CANDIDATE_BEST or REQUESTED with the same + // txid. We only need to do this if the found announcement had a + // different state than CANDIDATE_BEST. If it did, invariants + // guarantee that no other CANDIDATE_BEST or REQUESTED can exist. + auto it_old = m_index.get().lower_bound( + ByTxIdView{txid, State::CANDIDATE_BEST, 0}); + if (it_old != m_index.get().end() && + it_old->m_txid == txid) { + if (it_old->GetState() == State::CANDIDATE_BEST) { + // The data structure's invariants require that there can be + // at most one CANDIDATE_BEST or one REQUESTED announcement + // per txid (but not both simultaneously), so we have to + // convert any existing CANDIDATE_BEST to another + // CANDIDATE_* when constructing another REQUESTED. It + // doesn't matter whether we pick CANDIDATE_READY or + // _DELAYED here, as SetTimePoint() will correct it at + // GetRequestable() time. If time only goes forward, it will + // always be _READY, so pick that to avoid extra work in + // SetTimePoint(). + Modify(it_old, [](Announcement &ann) { + ann.SetState(State::CANDIDATE_READY); + }); + } else if (it_old->GetState() == State::REQUESTED) { + // As we're no longer waiting for a response to the previous + // REQUESTED announcement, convert it to COMPLETED. This + // also helps guaranteeing progress. + Modify(it_old, [](Announcement &ann) { + ann.SetState(State::COMPLETED); + }); + } + } + } + + Modify(it, [expiry](Announcement &ann) { + ann.SetState(State::REQUESTED); + ann.m_time = expiry; + }); + } /** * Converts a CANDIDATE or REQUESTED announcement to a COMPLETED one. If no @@ -220,34 +944,106 @@ * a peer. When the transaction is not needed entirely anymore, ForgetTxId * should be called instead of, or in addition to, this call. */ - void ReceivedResponse(NodeId peer, const TxId &txid); + void ReceivedResponse(NodeId peer, const TxId &txid) { + // We need to search the ByPeer index for both (peer, false, txid) and + // (peer, true, txid). + auto it = m_index.get().find(ByPeerView{peer, false, txid}); + if (it == m_index.get().end()) { + it = m_index.get().find(ByPeerView{peer, true, txid}); + } + if (it != m_index.get().end()) { + MakeCompleted(m_index.project(it)); + } + } // The operations below inspect the data structure. /** Count how many REQUESTED announcements a peer has. */ - size_t CountInFlight(NodeId peer) const; + size_t CountInFlight(NodeId peer) const { + auto it = m_peerinfo.find(peer); + if (it != m_peerinfo.end()) { + return it->second.m_requested; + } + return 0; + } /** Count how many CANDIDATE announcements a peer has. */ - size_t CountCandidates(NodeId peer) const; + size_t CountCandidates(NodeId peer) const { + auto it = m_peerinfo.find(peer); + if (it != m_peerinfo.end()) { + return it->second.m_total - it->second.m_requested - + it->second.m_completed; + } + return 0; + } /** * Count how many announcements a peer has (REQUESTED, CANDIDATE, and * COMPLETED combined). */ - size_t Count(NodeId peer) const; + size_t Count(NodeId peer) const { + auto it = m_peerinfo.find(peer); + if (it != m_peerinfo.end()) { + return it->second.m_total; + } + return 0; + } /** * Count how many announcements are being tracked in total across all peers * and transaction hashes. */ - size_t Size() const; + size_t Size() const { return m_index.size(); } /** Access to the internal priority computation (testing only) */ uint64_t ComputePriority(const TxId &txid, NodeId peer, - bool preferred) const; + bool preferred) const { + // Return Priority as a uint64_t as Priority is internal. + return uint64_t{m_computer(txid, peer, preferred)}; + } /** Run internal consistency check (testing only). */ - void SanityCheck() const; + void SanityCheck() const { + // Recompute m_peerdata from m_index. This verifies the data in it as it + // should just be caching statistics on m_index. It also verifies the + // invariant that no PeerInfo announcements with m_total==0 exist. + assert(m_peerinfo == RecomputePeerInfo(m_index)); + + // Calculate per-txid statistics from m_index, and validate + // invariants. + for (auto &item : ComputeTxIdInfo(m_index, m_computer)) { + TxIdInfo &info = item.second; + + // Cannot have only COMPLETED peer (txid should have been forgotten + // already) + assert(info.m_candidate_delayed + info.m_candidate_ready + + info.m_candidate_best + info.m_requested > + 0); + + // Can have at most 1 CANDIDATE_BEST/REQUESTED peer + assert(info.m_candidate_best + info.m_requested <= 1); + + // If there are any CANDIDATE_READY announcements, there must be + // exactly one CANDIDATE_BEST or REQUESTED announcement. + if (info.m_candidate_ready > 0) { + assert(info.m_candidate_best + info.m_requested == 1); + } + + // If there is both a CANDIDATE_READY and a CANDIDATE_BEST + // announcement, the CANDIDATE_BEST one must be at least as good + // (equal or higher priority) as the best CANDIDATE_READY. + if (info.m_candidate_ready && info.m_candidate_best) { + assert(info.m_priority_candidate_best >= + info.m_priority_best_candidate_ready); + } + + // No txid can have been announced by the same peer twice. + std::sort(info.m_peers.begin(), info.m_peers.end()); + assert( + std::adjacent_find(info.m_peers.begin(), info.m_peers.end()) == + info.m_peers.end()); + } + } /** * Run a time-dependent internal consistency check (testing only). @@ -255,7 +1051,22 @@ * This can only be called immediately after GetRequestable, with the same * 'now' parameter. */ - void PostGetRequestableSanityCheck(std::chrono::microseconds now) const; + void PostGetRequestableSanityCheck(std::chrono::microseconds now) const { + for (const Announcement &ann : m_index) { + if (ann.IsWaiting()) { + // REQUESTED and CANDIDATE_DELAYED must have a time in the + // future (they should have been converted to + // COMPLETED/CANDIDATE_READY respectively). + assert(ann.m_time > now); + } else if (ann.IsSelectable()) { + // CANDIDATE_READY and CANDIDATE_BEST cannot have a time in the + // future (they should have remained CANDIDATE_DELAYED, or + // should have been converted back to it if time went + // backwards). + assert(ann.m_time <= now); + } + } + } }; #endif // BITCOIN_TXREQUEST_H diff --git a/src/txrequest.cpp b/src/txrequest.cpp deleted file mode 100644 --- a/src/txrequest.cpp +++ /dev/null @@ -1,916 +0,0 @@ -// Copyright (c) 2020 The Bitcoin Core 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 -#include - -#include -#include -#include -#include - -namespace { - -/** - * The various states a (txid, peer) pair can be in. - * - * Note that CANDIDATE is split up into 3 substates (DELAYED, BEST, READY), - * allowing more efficient implementation. Also note that the sorting order of - * ByTxIdView relies on the specific order of values in this enum. - * - * Expected behaviour is: - * - When first announced by a peer, the state is CANDIDATE_DELAYED until - * reqtime is reached. - * - Announcements that have reached their reqtime but not been requested will - * be either CANDIDATE_READY or CANDIDATE_BEST. Neither of those has an - * expiration time; they remain in that state until they're requested or no - * longer needed. CANDIDATE_READY announcements are promoted to - * CANDIDATE_BEST when they're the best one left. - * - When requested, an announcement will be in state REQUESTED until expiry - * is reached. - * - If expiry is reached, or the peer replies to the request (either with - * NOTFOUND or the tx), the state becomes COMPLETED. - */ -enum class State : uint8_t { - /** A CANDIDATE announcement whose reqtime is in the future. */ - CANDIDATE_DELAYED, - /** - * A CANDIDATE announcement that's not CANDIDATE_DELAYED or CANDIDATE_BEST. - */ - CANDIDATE_READY, - /** - * The best CANDIDATE for a given txid; only if there is no REQUESTED - * announcement already for that txid. The CANDIDATE_BEST is the - * highest-priority announcement among all CANDIDATE_READY (and _BEST) ones - * for that txid. - */ - CANDIDATE_BEST, - /** A REQUESTED announcement. */ - REQUESTED, - /** A COMPLETED announcement. */ - COMPLETED, -}; - -//! Type alias for sequence numbers. -using SequenceNumber = uint64_t; - -/** - * An announcement. This is the data we track for each txid that is announced - * to us by each peer. - */ -struct Announcement { - /** TxId that was announced. */ - const TxId m_txid; - /** - * For CANDIDATE_{DELAYED,BEST,READY} the reqtime; for REQUESTED the - * expiry. - */ - std::chrono::microseconds m_time; - /** What peer the request was from. */ - const NodeId m_peer; - /** What sequence number this announcement has. */ - const SequenceNumber m_sequence : 60; - /** Whether the request is preferred. */ - const bool m_preferred : 1; - - /** - * What state this announcement is in. - * This is a uint8_t instead of a State to silence a GCC warning in versions - * prior to 8.4 and 9.3. See - * https://gcc.gnu.org/bugzilla/show_bug.cgi?id=61414 - */ - uint8_t m_state : 3; - - /** Convert m_state to a State enum. */ - State GetState() const { return static_cast(m_state); } - - /** Convert a State enum to a uint8_t and store it in m_state. */ - void SetState(State state) { m_state = static_cast(state); } - - /** - * Whether this announcement is selected. There can be at most 1 selected - * peer per txid. - */ - bool IsSelected() const { - return GetState() == State::CANDIDATE_BEST || - GetState() == State::REQUESTED; - } - - /** Whether this announcement is waiting for a certain time to pass. */ - bool IsWaiting() const { - return GetState() == State::REQUESTED || - GetState() == State::CANDIDATE_DELAYED; - } - - /** - * Whether this announcement can feasibly be selected if the current - * IsSelected() one disappears. - */ - bool IsSelectable() const { - return GetState() == State::CANDIDATE_READY || - GetState() == State::CANDIDATE_BEST; - } - - /** - * Construct a new announcement from scratch, initially in - * CANDIDATE_DELAYED state. - */ - Announcement(const TxId &txid, NodeId peer, bool preferred, - std::chrono::microseconds reqtime, SequenceNumber sequence) - : m_txid(txid), m_time(reqtime), m_peer(peer), m_sequence(sequence), - m_preferred(preferred), - m_state(static_cast(State::CANDIDATE_DELAYED)) {} -}; - -//! Type alias for priorities. -using Priority = uint64_t; - -/** - * A functor with embedded salt that computes priority of an announcement. - * - * Higher priorities are selected first. - */ -class PriorityComputer { - const uint64_t m_k0, m_k1; - -public: - explicit PriorityComputer(bool deterministic) - : m_k0{deterministic ? 0 : GetRand(0xFFFFFFFFFFFFFFFF)}, - m_k1{deterministic ? 0 : GetRand(0xFFFFFFFFFFFFFFFF)} {} - - Priority operator()(const TxId &txid, NodeId peer, bool preferred) const { - uint64_t low_bits = CSipHasher(m_k0, m_k1) - .Write(txid.begin(), txid.size()) - .Write(peer) - .Finalize() >> - 1; - return low_bits | uint64_t{preferred} << 63; - } - - Priority operator()(const Announcement &ann) const { - return operator()(ann.m_txid, ann.m_peer, ann.m_preferred); - } -}; - -// Definitions for the 3 indexes used in the main data structure. -// -// Each index has a By* type to identify it, a By*View data type to represent -// the view of announcement it is sorted by, and an By*ViewExtractor type to -// convert an announcement into the By*View type. See -// https://www.boost.org/doc/libs/1_58_0/libs/multi_index/doc/reference/key_extraction.html#key_extractors -// for more information about the key extraction concept. - -// The ByPeer index is sorted by (peer, state == CANDIDATE_BEST, txid) -// -// Uses: -// * Looking up existing announcements by peer/txid, by checking both (peer, -// false, txid) and (peer, true, txid). -// * Finding all CANDIDATE_BEST announcements for a given peer in -// GetRequestable. -struct ByPeer {}; -using ByPeerView = std::tuple; -struct ByPeerViewExtractor { - using result_type = ByPeerView; - result_type operator()(const Announcement &ann) const { - return ByPeerView{ann.m_peer, ann.GetState() == State::CANDIDATE_BEST, - ann.m_txid}; - } -}; - -// The ByTxId index is sorted by (txid, state, priority). -// -// Note: priority == 0 whenever state != CANDIDATE_READY. -// -// Uses: -// * Deleting all announcements with a given txid in ForgetTxId. -// * Finding the best CANDIDATE_READY to convert to CANDIDATE_BEST, when no -// other CANDIDATE_READY or REQUESTED announcement exists for that txid. -// * Determining when no more non-COMPLETED announcements for a given txid -// exist, so the COMPLETED ones can be deleted. -struct ByTxId {}; -using ByTxIdView = std::tuple; -class ByTxIdViewExtractor { - const PriorityComputer &m_computer; - -public: - ByTxIdViewExtractor(const PriorityComputer &computer) - : m_computer(computer) {} - using result_type = ByTxIdView; - result_type operator()(const Announcement &ann) const { - const Priority prio = - (ann.GetState() == State::CANDIDATE_READY) ? m_computer(ann) : 0; - return ByTxIdView{ann.m_txid, ann.GetState(), prio}; - } -}; - -enum class WaitState { - //! Used for announcements that need efficient testing of "is their - //! timestamp in the future?". - FUTURE_EVENT, - //! Used for announcements whose timestamp is not relevant. - NO_EVENT, - //! Used for announcements that need efficient testing of "is their - //! timestamp in the past?". - PAST_EVENT, -}; - -WaitState GetWaitState(const Announcement &ann) { - if (ann.IsWaiting()) { - return WaitState::FUTURE_EVENT; - } - if (ann.IsSelectable()) { - return WaitState::PAST_EVENT; - } - return WaitState::NO_EVENT; -} - -// The ByTime index is sorted by (wait_state, time). -// -// All announcements with a timestamp in the future can be found by iterating -// the index forward from the beginning. All announcements with a timestamp in -// the past can be found by iterating the index backwards from the end. -// -// Uses: -// * Finding CANDIDATE_DELAYED announcements whose reqtime has passed, and -// REQUESTED announcements whose expiry has passed. -// * Finding CANDIDATE_READY/BEST announcements whose reqtime is in the future -// (when the clock time went backwards). -struct ByTime {}; -using ByTimeView = std::pair; -struct ByTimeViewExtractor { - using result_type = ByTimeView; - result_type operator()(const Announcement &ann) const { - return ByTimeView{GetWaitState(ann), ann.m_time}; - } -}; - -/** - * Data type for the main data structure (Announcement objects with - * ByPeer/ByTxId/ByTime indexes). - */ -using Index = boost::multi_index_container< - Announcement, - boost::multi_index::indexed_by< - boost::multi_index::ordered_unique, - ByPeerViewExtractor>, - boost::multi_index::ordered_non_unique, - ByTxIdViewExtractor>, - boost::multi_index::ordered_non_unique, - ByTimeViewExtractor>>>; - -/** Helper type to simplify syntax of iterator types. */ -template using Iter = typename Index::index::type::iterator; - -/** Per-peer statistics object. */ -struct PeerInfo { - //! Total number of announcements for this peer. - size_t m_total = 0; - //! Number of COMPLETED announcements for this peer. - size_t m_completed = 0; - //! Number of REQUESTED announcements for this peer. - size_t m_requested = 0; -}; - -/** Per-txid statistics object. Only used for sanity checking. */ -struct TxIdInfo { - //! Number of CANDIDATE_DELAYED announcements for this txid. - size_t m_candidate_delayed = 0; - //! Number of CANDIDATE_READY announcements for this txid. - size_t m_candidate_ready = 0; - //! Number of CANDIDATE_BEST announcements for this txid (at most one). - size_t m_candidate_best = 0; - //! Number of REQUESTED announcements for this txid (at most one; mutually - //! exclusive with CANDIDATE_BEST). - size_t m_requested = 0; - //! The priority of the CANDIDATE_BEST announcement if one exists, or max() - //! otherwise. - Priority m_priority_candidate_best = std::numeric_limits::max(); - //! The highest priority of all CANDIDATE_READY announcements (or min() if - //! none exist). - Priority m_priority_best_candidate_ready = - std::numeric_limits::min(); - //! All peers we have an announcement for this txid for. - std::vector m_peers; -}; - -/** Compare two PeerInfo objects. Only used for sanity checking. */ -bool operator==(const PeerInfo &a, const PeerInfo &b) { - return std::tie(a.m_total, a.m_completed, a.m_requested) == - std::tie(b.m_total, b.m_completed, b.m_requested); -}; - -/** - * (Re)compute the PeerInfo map from the index. Only used for sanity checking. - */ -std::unordered_map RecomputePeerInfo(const Index &index) { - std::unordered_map ret; - for (const Announcement &ann : index) { - PeerInfo &info = ret[ann.m_peer]; - ++info.m_total; - info.m_requested += (ann.GetState() == State::REQUESTED); - info.m_completed += (ann.GetState() == State::COMPLETED); - } - return ret; -} - -/** Compute the TxIdInfo map. Only used for sanity checking. */ -std::map ComputeTxIdInfo(const Index &index, - const PriorityComputer &computer) { - std::map ret; - for (const Announcement &ann : index) { - TxIdInfo &info = ret[ann.m_txid]; - // Classify how many announcements of each state we have for this txid. - info.m_candidate_delayed += - (ann.GetState() == State::CANDIDATE_DELAYED); - info.m_candidate_ready += (ann.GetState() == State::CANDIDATE_READY); - info.m_candidate_best += (ann.GetState() == State::CANDIDATE_BEST); - info.m_requested += (ann.GetState() == State::REQUESTED); - // And track the priority of the best CANDIDATE_READY/CANDIDATE_BEST - // announcements. - if (ann.GetState() == State::CANDIDATE_BEST) { - info.m_priority_candidate_best = computer(ann); - } - if (ann.GetState() == State::CANDIDATE_READY) { - info.m_priority_best_candidate_ready = - std::max(info.m_priority_best_candidate_ready, computer(ann)); - } - // Also keep track of which peers this txid has an announcement for - // (so we can detect duplicates). - info.m_peers.push_back(ann.m_peer); - } - return ret; -} - -} // namespace - -/** Actual implementation for TxRequestTracker's data structure. */ -class TxRequestTracker::Impl { - //! The current sequence number. Increases for every announcement. This is - //! used to sort txid returned by GetRequestable in announcement order. - SequenceNumber m_current_sequence{0}; - - //! This tracker's priority computer. - const PriorityComputer m_computer; - - //! This tracker's main data structure. See SanityCheck() for the invariants - //! that apply to it. - Index m_index; - - //! Map with this tracker's per-peer statistics. - std::unordered_map m_peerinfo; - -public: - void SanityCheck() const { - // Recompute m_peerdata from m_index. This verifies the data in it as it - // should just be caching statistics on m_index. It also verifies the - // invariant that no PeerInfo announcements with m_total==0 exist. - assert(m_peerinfo == RecomputePeerInfo(m_index)); - - // Calculate per-txid statistics from m_index, and validate - // invariants. - for (auto &item : ComputeTxIdInfo(m_index, m_computer)) { - TxIdInfo &info = item.second; - - // Cannot have only COMPLETED peer (txid should have been forgotten - // already) - assert(info.m_candidate_delayed + info.m_candidate_ready + - info.m_candidate_best + info.m_requested > - 0); - - // Can have at most 1 CANDIDATE_BEST/REQUESTED peer - assert(info.m_candidate_best + info.m_requested <= 1); - - // If there are any CANDIDATE_READY announcements, there must be - // exactly one CANDIDATE_BEST or REQUESTED announcement. - if (info.m_candidate_ready > 0) { - assert(info.m_candidate_best + info.m_requested == 1); - } - - // If there is both a CANDIDATE_READY and a CANDIDATE_BEST - // announcement, the CANDIDATE_BEST one must be at least as good - // (equal or higher priority) as the best CANDIDATE_READY. - if (info.m_candidate_ready && info.m_candidate_best) { - assert(info.m_priority_candidate_best >= - info.m_priority_best_candidate_ready); - } - - // No txid can have been announced by the same peer twice. - std::sort(info.m_peers.begin(), info.m_peers.end()); - assert( - std::adjacent_find(info.m_peers.begin(), info.m_peers.end()) == - info.m_peers.end()); - } - } - - void PostGetRequestableSanityCheck(std::chrono::microseconds now) const { - for (const Announcement &ann : m_index) { - if (ann.IsWaiting()) { - // REQUESTED and CANDIDATE_DELAYED must have a time in the - // future (they should have been converted to - // COMPLETED/CANDIDATE_READY respectively). - assert(ann.m_time > now); - } else if (ann.IsSelectable()) { - // CANDIDATE_READY and CANDIDATE_BEST cannot have a time in the - // future (they should have remained CANDIDATE_DELAYED, or - // should have been converted back to it if time went - // backwards). - assert(ann.m_time <= now); - } - } - } - -private: - //! Wrapper around Index::...::erase that keeps m_peerinfo up to date. - template Iter Erase(Iter it) { - auto peerit = m_peerinfo.find(it->m_peer); - peerit->second.m_completed -= it->GetState() == State::COMPLETED; - peerit->second.m_requested -= it->GetState() == State::REQUESTED; - if (--peerit->second.m_total == 0) { - m_peerinfo.erase(peerit); - } - return m_index.get().erase(it); - } - - //! Wrapper around Index::...::modify that keeps m_peerinfo up to date. - template - void Modify(Iter it, Modifier modifier) { - auto peerit = m_peerinfo.find(it->m_peer); - peerit->second.m_completed -= it->GetState() == State::COMPLETED; - peerit->second.m_requested -= it->GetState() == State::REQUESTED; - m_index.get().modify(it, std::move(modifier)); - peerit->second.m_completed += it->GetState() == State::COMPLETED; - peerit->second.m_requested += it->GetState() == State::REQUESTED; - } - - //! Convert a CANDIDATE_DELAYED announcement into a CANDIDATE_READY. If this - //! makes it the new best CANDIDATE_READY (and no REQUESTED exists) and - //! better than the CANDIDATE_BEST (if any), it becomes the new - //! CANDIDATE_BEST. - void PromoteCandidateReady(Iter it) { - assert(it != m_index.get().end()); - assert(it->GetState() == State::CANDIDATE_DELAYED); - // Convert CANDIDATE_DELAYED to CANDIDATE_READY first. - Modify(it, [](Announcement &ann) { - ann.SetState(State::CANDIDATE_READY); - }); - // The following code relies on the fact that the ByTxId is sorted by - // txid, and then by state (first _DELAYED, then _READY, then - // _BEST/REQUESTED). Within the _READY announcements, the best one - // (highest priority) comes last. Thus, if an existing _BEST exists for - // the same txid that this announcement may be preferred over, it must - // immediately follow the newly created _READY. - auto it_next = std::next(it); - if (it_next == m_index.get().end() || - it_next->m_txid != it->m_txid || - it_next->GetState() == State::COMPLETED) { - // This is the new best CANDIDATE_READY, and there is no - // IsSelected() announcement for this txid already. - Modify(it, [](Announcement &ann) { - ann.SetState(State::CANDIDATE_BEST); - }); - } else if (it_next->GetState() == State::CANDIDATE_BEST) { - Priority priority_old = m_computer(*it_next); - Priority priority_new = m_computer(*it); - if (priority_new > priority_old) { - // There is a CANDIDATE_BEST announcement already, but this one - // is better. - Modify(it_next, [](Announcement &ann) { - ann.SetState(State::CANDIDATE_READY); - }); - Modify(it, [](Announcement &ann) { - ann.SetState(State::CANDIDATE_BEST); - }); - } - } - } - - //! Change the state of an announcement to something non-IsSelected(). If it - //! was IsSelected(), the next best announcement will be marked - //! CANDIDATE_BEST. - void ChangeAndReselect(Iter it, State new_state) { - assert(new_state == State::COMPLETED || - new_state == State::CANDIDATE_DELAYED); - assert(it != m_index.get().end()); - if (it->IsSelected() && it != m_index.get().begin()) { - auto it_prev = std::prev(it); - // The next best CANDIDATE_READY, if any, immediately precedes the - // REQUESTED or CANDIDATE_BEST announcement in the ByTxId index. - if (it_prev->m_txid == it->m_txid && - it_prev->GetState() == State::CANDIDATE_READY) { - // If one such CANDIDATE_READY exists (for this txid), convert - // it to CANDIDATE_BEST. - Modify(it_prev, [](Announcement &ann) { - ann.SetState(State::CANDIDATE_BEST); - }); - } - } - Modify( - it, [new_state](Announcement &ann) { ann.SetState(new_state); }); - } - - //! Check if 'it' is the only announcement for a given txid that isn't - //! COMPLETED. - bool IsOnlyNonCompleted(Iter it) { - assert(it != m_index.get().end()); - // Not allowed to call this on COMPLETED announcements. - assert(it->GetState() != State::COMPLETED); - - // This announcement has a predecessor that belongs to the same txid. - // Due to ordering, and the fact that 'it' is not COMPLETED, its - // predecessor cannot be COMPLETED here. - if (it != m_index.get().begin() && - std::prev(it)->m_txid == it->m_txid) { - return false; - } - - // This announcement has a successor that belongs to the same txid, - // and is not COMPLETED. - if (std::next(it) != m_index.get().end() && - std::next(it)->m_txid == it->m_txid && - std::next(it)->GetState() != State::COMPLETED) { - return false; - } - - return true; - } - - /** - * Convert any announcement to a COMPLETED one. If there are no - * non-COMPLETED announcements left for this txid, they are deleted. If - * this was a REQUESTED announcement, and there are other CANDIDATEs left, - * the best one is made CANDIDATE_BEST. Returns whether the announcement - * still exists. - */ - bool MakeCompleted(Iter it) { - assert(it != m_index.get().end()); - - // Nothing to be done if it's already COMPLETED. - if (it->GetState() == State::COMPLETED) { - return true; - } - - if (IsOnlyNonCompleted(it)) { - // This is the last non-COMPLETED announcement for this txid. - // Delete all. - TxId txid = it->m_txid; - do { - it = Erase(it); - } while (it != m_index.get().end() && it->m_txid == txid); - return false; - } - - // Mark the announcement COMPLETED, and select the next best - // announcement (the first CANDIDATE_READY) if needed. - ChangeAndReselect(it, State::COMPLETED); - - return true; - } - - //! Make the data structure consistent with a given point in time: - //! - REQUESTED annoucements with expiry <= now are turned into COMPLETED. - //! - CANDIDATE_DELAYED announcements with reqtime <= now are turned into - //! CANDIDATE_{READY,BEST}. - //! - CANDIDATE_{READY,BEST} announcements with reqtime > now are turned - //! into CANDIDATE_DELAYED. - void SetTimePoint(std::chrono::microseconds now, - std::vector> *expired) { - if (expired) { - expired->clear(); - } - // Iterate over all CANDIDATE_DELAYED and REQUESTED from old to new, as - // long as they're in the past, and convert them to CANDIDATE_READY and - // COMPLETED respectively. - while (!m_index.empty()) { - auto it = m_index.get().begin(); - if (it->GetState() == State::CANDIDATE_DELAYED && - it->m_time <= now) { - PromoteCandidateReady(m_index.project(it)); - } else if (it->GetState() == State::REQUESTED && - it->m_time <= now) { - if (expired) { - expired->emplace_back(it->m_peer, it->m_txid); - } - MakeCompleted(m_index.project(it)); - } else { - break; - } - } - - while (!m_index.empty()) { - // If time went backwards, we may need to demote CANDIDATE_BEST and - // CANDIDATE_READY announcements back to CANDIDATE_DELAYED. This is - // an unusual edge case, and unlikely to matter in production. - // However, it makes it much easier to specify and test - // TxRequestTracker::Impl's behaviour. - auto it = std::prev(m_index.get().end()); - if (it->IsSelectable() && it->m_time > now) { - ChangeAndReselect(m_index.project(it), - State::CANDIDATE_DELAYED); - } else { - break; - } - } - } - -public: - Impl(bool deterministic) - : m_computer(deterministic), - // Explicitly initialize m_index as we need to pass a reference to - // m_computer to ByTxHashViewExtractor. - m_index(boost::make_tuple( - boost::make_tuple(ByPeerViewExtractor(), std::less()), - boost::make_tuple(ByTxIdViewExtractor(m_computer), - std::less()), - boost::make_tuple(ByTimeViewExtractor(), - std::less()))) {} - - // Disable copying and assigning (a default copy won't work due the stateful - // ByTxIdViewExtractor). - Impl(const Impl &) = delete; - Impl &operator=(const Impl &) = delete; - - void DisconnectedPeer(NodeId peer) { - auto &index = m_index.get(); - auto it = - index.lower_bound(ByPeerView{peer, false, TxId(uint256::ZERO)}); - while (it != index.end() && it->m_peer == peer) { - // Check what to continue with after this iteration. 'it' will be - // deleted in what follows, so we need to decide what to continue - // with afterwards. There are a number of cases to consider: - // - std::next(it) is end() or belongs to a different peer. In that - // case, this is the last iteration of the loop (denote this by - // setting it_next to end()). - // - 'it' is not the only non-COMPLETED announcement for its txid. - // This means it will be deleted, but no other Announcement - // objects will be modified. Continue with std::next(it) if it - // belongs to the same peer, but decide this ahead of time (as - // 'it' may change position in what follows). - // - 'it' is the only non-COMPLETED announcement for its txid. This - // means it will be deleted along with all other announcements for - // the same txid - which may include std::next(it). However, other - // than 'it', no announcements for the same peer can be affected - // (due to (peer, txid) uniqueness). In other words, the situation - // where std::next(it) is deleted can only occur if std::next(it) - // belongs to a different peer but the same txid as 'it'. This is - // covered by the first bulletpoint already, and we'll have set - // it_next to end(). - auto it_next = - (std::next(it) == index.end() || std::next(it)->m_peer != peer) - ? index.end() - : std::next(it); - // If the announcement isn't already COMPLETED, first make it - // COMPLETED (which will mark other CANDIDATEs as CANDIDATE_BEST, or - // delete all of a txid's announcements if no non-COMPLETED ones are - // left). - if (MakeCompleted(m_index.project(it))) { - // Then actually delete the announcement (unless it was already - // deleted by MakeCompleted). - Erase(it); - } - it = it_next; - } - } - - void ForgetTxId(const TxId &txid) { - auto it = m_index.get().lower_bound( - ByTxIdView{txid, State::CANDIDATE_DELAYED, 0}); - while (it != m_index.get().end() && it->m_txid == txid) { - it = Erase(it); - } - } - - void ReceivedInv(NodeId peer, const TxId &txid, bool preferred, - std::chrono::microseconds reqtime) { - // Bail out if we already have a CANDIDATE_BEST announcement for this - // (txid, peer) combination. The case where there is a - // non-CANDIDATE_BEST announcement already will be caught by the - // uniqueness property of the ByPeer index when we try to emplace the - // new object below. - if (m_index.get().count(ByPeerView{peer, true, txid})) { - return; - } - - // Try creating the announcement with CANDIDATE_DELAYED state (which - // will fail due to the uniqueness of the ByPeer index if a - // non-CANDIDATE_BEST announcement already exists with the same txid - // and peer). Bail out in that case. - auto ret = m_index.get().emplace(txid, peer, preferred, reqtime, - m_current_sequence); - if (!ret.second) { - return; - } - - // Update accounting metadata. - ++m_peerinfo[peer].m_total; - ++m_current_sequence; - } - - //! Find the TxIds to request now from peer. - std::vector - GetRequestable(NodeId peer, std::chrono::microseconds now, - std::vector> *expired) { - // Move time. - SetTimePoint(now, expired); - - // Find all CANDIDATE_BEST announcements for this peer. - std::vector selected; - auto it_peer = m_index.get().lower_bound( - ByPeerView{peer, true, TxId(uint256::ZERO)}); - while (it_peer != m_index.get().end() && - it_peer->m_peer == peer && - it_peer->GetState() == State::CANDIDATE_BEST) { - selected.emplace_back(&*it_peer); - ++it_peer; - } - - // Sort by sequence number. - std::sort(selected.begin(), selected.end(), - [](const Announcement *a, const Announcement *b) { - return a->m_sequence < b->m_sequence; - }); - - // Convert to TxId and return. - std::vector ret; - ret.reserve(selected.size()); - std::transform(selected.begin(), selected.end(), - std::back_inserter(ret), - [](const Announcement *ann) { return ann->m_txid; }); - return ret; - } - - void RequestedTx(NodeId peer, const TxId &txid, - std::chrono::microseconds expiry) { - auto it = m_index.get().find(ByPeerView{peer, true, txid}); - if (it == m_index.get().end()) { - // There is no CANDIDATE_BEST announcement, look for a _READY or - // _DELAYED instead. If the caller only ever invokes RequestedTx - // with the values returned by GetRequestable, and no other - // non-const functions other than ForgetTxHash and GetRequestable in - // between, this branch will never execute (as txids returned by - // GetRequestable always correspond to CANDIDATE_BEST - // announcements). - - it = m_index.get().find(ByPeerView{peer, false, txid}); - if (it == m_index.get().end() || - (it->GetState() != State::CANDIDATE_DELAYED && - it->GetState() != State::CANDIDATE_READY)) { - // There is no CANDIDATE announcement tracked for this peer, so - // we have nothing to do. Either this txid wasn't tracked at - // all (and the caller should have called ReceivedInv), or it - // was already requested and/or completed for other reasons and - // this is just a superfluous RequestedTx call. - return; - } - - // Look for an existing CANDIDATE_BEST or REQUESTED with the same - // txid. We only need to do this if the found announcement had a - // different state than CANDIDATE_BEST. If it did, invariants - // guarantee that no other CANDIDATE_BEST or REQUESTED can exist. - auto it_old = m_index.get().lower_bound( - ByTxIdView{txid, State::CANDIDATE_BEST, 0}); - if (it_old != m_index.get().end() && - it_old->m_txid == txid) { - if (it_old->GetState() == State::CANDIDATE_BEST) { - // The data structure's invariants require that there can be - // at most one CANDIDATE_BEST or one REQUESTED announcement - // per txid (but not both simultaneously), so we have to - // convert any existing CANDIDATE_BEST to another - // CANDIDATE_* when constructing another REQUESTED. It - // doesn't matter whether we pick CANDIDATE_READY or - // _DELAYED here, as SetTimePoint() will correct it at - // GetRequestable() time. If time only goes forward, it will - // always be _READY, so pick that to avoid extra work in - // SetTimePoint(). - Modify(it_old, [](Announcement &ann) { - ann.SetState(State::CANDIDATE_READY); - }); - } else if (it_old->GetState() == State::REQUESTED) { - // As we're no longer waiting for a response to the previous - // REQUESTED announcement, convert it to COMPLETED. This - // also helps guaranteeing progress. - Modify(it_old, [](Announcement &ann) { - ann.SetState(State::COMPLETED); - }); - } - } - } - - Modify(it, [expiry](Announcement &ann) { - ann.SetState(State::REQUESTED); - ann.m_time = expiry; - }); - } - - void ReceivedResponse(NodeId peer, const TxId &txid) { - // We need to search the ByPeer index for both (peer, false, txid) and - // (peer, true, txid). - auto it = m_index.get().find(ByPeerView{peer, false, txid}); - if (it == m_index.get().end()) { - it = m_index.get().find(ByPeerView{peer, true, txid}); - } - if (it != m_index.get().end()) { - MakeCompleted(m_index.project(it)); - } - } - - size_t CountInFlight(NodeId peer) const { - auto it = m_peerinfo.find(peer); - if (it != m_peerinfo.end()) { - return it->second.m_requested; - } - return 0; - } - - size_t CountCandidates(NodeId peer) const { - auto it = m_peerinfo.find(peer); - if (it != m_peerinfo.end()) { - return it->second.m_total - it->second.m_requested - - it->second.m_completed; - } - return 0; - } - - size_t Count(NodeId peer) const { - auto it = m_peerinfo.find(peer); - if (it != m_peerinfo.end()) { - return it->second.m_total; - } - return 0; - } - - //! Count how many announcements are being tracked in total across all peers - //! and transactions. - size_t Size() const { return m_index.size(); } - - uint64_t ComputePriority(const TxId &txid, NodeId peer, - bool preferred) const { - // Return Priority as a uint64_t as Priority is internal. - return uint64_t{m_computer(txid, peer, preferred)}; - } -}; - -TxRequestTracker::TxRequestTracker(bool deterministic) - : m_impl{std::make_unique(deterministic)} {} - -TxRequestTracker::~TxRequestTracker() = default; - -void TxRequestTracker::ForgetTxId(const TxId &txid) { - m_impl->ForgetTxId(txid); -} -void TxRequestTracker::DisconnectedPeer(NodeId peer) { - m_impl->DisconnectedPeer(peer); -} -size_t TxRequestTracker::CountInFlight(NodeId peer) const { - return m_impl->CountInFlight(peer); -} -size_t TxRequestTracker::CountCandidates(NodeId peer) const { - return m_impl->CountCandidates(peer); -} -size_t TxRequestTracker::Count(NodeId peer) const { - return m_impl->Count(peer); -} -size_t TxRequestTracker::Size() const { - return m_impl->Size(); -} -void TxRequestTracker::SanityCheck() const { - m_impl->SanityCheck(); -} - -void TxRequestTracker::PostGetRequestableSanityCheck( - std::chrono::microseconds now) const { - m_impl->PostGetRequestableSanityCheck(now); -} - -void TxRequestTracker::ReceivedInv(NodeId peer, const TxId &txid, - bool preferred, - std::chrono::microseconds reqtime) { - m_impl->ReceivedInv(peer, txid, preferred, reqtime); -} - -void TxRequestTracker::RequestedTx(NodeId peer, const TxId &txid, - std::chrono::microseconds expiry) { - m_impl->RequestedTx(peer, txid, expiry); -} - -void TxRequestTracker::ReceivedResponse(NodeId peer, const TxId &txid) { - m_impl->ReceivedResponse(peer, txid); -} - -std::vector TxRequestTracker::GetRequestable( - NodeId peer, std::chrono::microseconds now, - std::vector> *expired) { - return m_impl->GetRequestable(peer, now, expired); -} - -uint64_t TxRequestTracker::ComputePriority(const TxId &txid, NodeId peer, - bool preferred) const { - return m_impl->ComputePriority(txid, peer, preferred); -}