13#include <boost/multi_index/indexed_by.hpp>
14#include <boost/multi_index/ordered_index.hpp>
15#include <boost/multi_index/sequenced_index.hpp>
16#include <boost/multi_index/tag.hpp>
17#include <boost/multi_index_container.hpp>
18#include <boost/tuple/tuple.hpp>
21#include <unordered_map>
65 std::chrono::microseconds m_time;
69 const SequenceNumber m_sequence : 59;
71 const bool m_preferred : 1;
73 State m_state : 3 {State::CANDIDATE_DELAYED};
74 State GetState()
const {
return m_state; }
78 bool IsSelected()
const
80 return GetState() == State::CANDIDATE_BEST || GetState() == State::REQUESTED;
86 return GetState() == State::REQUESTED || GetState() == State::CANDIDATE_DELAYED;
90 bool IsSelectable()
const
92 return GetState() == State::CANDIDATE_READY || GetState() == State::CANDIDATE_BEST;
108class PriorityComputer {
121 Priority operator()(
const Announcement&
ann)
const
123 return operator()(
ann.m_gtxid.ToUint256(),
ann.m_peer,
ann.m_preferred);
141using ByPeerView = std::tuple<NodeId, bool, const uint256&>;
142struct ByPeerViewExtractor
144 using result_type = ByPeerView;
145 result_type operator()(
const Announcement&
ann)
const
147 return ByPeerView{
ann.m_peer,
ann.GetState() == State::CANDIDATE_BEST,
ann.m_gtxid.ToUint256()};
162using ByTxHashView = std::tuple<const uint256&, State, Priority>;
163class ByTxHashViewExtractor {
164 const PriorityComputer& m_computer;
166 explicit ByTxHashViewExtractor(
const PriorityComputer&
computer) : m_computer(
computer) {}
168 result_type operator()(
const Announcement&
ann)
const
170 const Priority
prio = (
ann.GetState() == State::CANDIDATE_READY) ? m_computer(
ann) : 0;
175enum class WaitState {
186 if (
ann.IsWaiting())
return WaitState::FUTURE_EVENT;
187 if (
ann.IsSelectable())
return WaitState::PAST_EVENT;
188 return WaitState::NO_EVENT;
201using ByTimeView = std::pair<WaitState, std::chrono::microseconds>;
202struct ByTimeViewExtractor
205 result_type operator()(
const Announcement&
ann)
const
213using Index = boost::multi_index_container<
215 boost::multi_index::indexed_by<
216 boost::multi_index::ordered_unique<boost::multi_index::tag<ByPeer>, ByPeerViewExtractor>,
217 boost::multi_index::ordered_non_unique<boost::multi_index::tag<ByTxHash>, ByTxHashViewExtractor>,
218 boost::multi_index::ordered_non_unique<boost::multi_index::tag<ByTime>, ByTimeViewExtractor>
223template<
typename Tag>
224using Iter =
typename Index::index<Tag>::type::iterator;
229 size_t m_completed = 0;
230 size_t m_requested = 0;
237 size_t m_candidate_delayed = 0;
239 size_t m_candidate_ready = 0;
241 size_t m_candidate_best = 0;
243 size_t m_requested = 0;
245 Priority m_priority_candidate_best = std::numeric_limits<Priority>::max();
247 Priority m_priority_best_candidate_ready = std::numeric_limits<Priority>::min();
249 std::vector<NodeId> m_peers;
253bool operator==(
const PeerInfo&
a,
const PeerInfo& b)
255 return std::tie(
a.m_total,
a.m_completed,
a.m_requested) ==
256 std::tie(b.m_total, b.m_completed, b.m_requested);
262 std::unordered_map<NodeId, PeerInfo>
ret;
263 for (
const Announcement&
ann : index) {
264 PeerInfo& info =
ret[
ann.m_peer];
266 info.m_requested += (
ann.GetState() == State::REQUESTED);
267 info.m_completed += (
ann.GetState() == State::COMPLETED);
275 std::map<uint256, TxHashInfo>
ret;
276 for (
const Announcement&
ann : index) {
277 TxHashInfo& info =
ret[
ann.m_gtxid.ToUint256()];
279 info.m_candidate_delayed += (
ann.GetState() == State::CANDIDATE_DELAYED);
280 info.m_candidate_ready += (
ann.GetState() == State::CANDIDATE_READY);
281 info.m_candidate_best += (
ann.GetState() == State::CANDIDATE_BEST);
282 info.m_requested += (
ann.GetState() == State::REQUESTED);
284 if (
ann.GetState() == State::CANDIDATE_BEST) {
287 if (
ann.GetState() == State::CANDIDATE_READY) {
288 info.m_priority_best_candidate_ready = std::max(info.m_priority_best_candidate_ready,
computer(
ann));
291 info.m_peers.push_back(
ann.m_peer);
322 TxHashInfo& info =
item.second;
325 assert(info.m_candidate_delayed + info.m_candidate_ready + info.m_candidate_best + info.m_requested > 0);
328 assert(info.m_candidate_best + info.m_requested <= 1);
332 if (info.m_candidate_ready > 0) {
333 assert(info.m_candidate_best + info.m_requested == 1);
338 if (info.m_candidate_ready && info.m_candidate_best) {
339 assert(info.m_priority_candidate_best >= info.m_priority_best_candidate_ready);
343 std::sort(info.m_peers.begin(), info.m_peers.end());
344 assert(std::adjacent_find(info.m_peers.begin(), info.m_peers.end()) == info.m_peers.end());
351 if (
ann.IsWaiting()) {
355 }
else if (
ann.IsSelectable()) {
365 template<
typename Tag>
369 peerit->second.m_completed -= it->GetState() == State::COMPLETED;
370 peerit->second.m_requested -= it->GetState() == State::REQUESTED;
376 template<
typename Tag,
typename Modifier>
380 peerit->second.m_completed -= it->GetState() == State::COMPLETED;
381 peerit->second.m_requested -= it->GetState() == State::REQUESTED;
383 peerit->second.m_completed += it->GetState() == State::COMPLETED;
384 peerit->second.m_requested += it->GetState() == State::REQUESTED;
393 assert(it->GetState() == State::CANDIDATE_DELAYED);
401 if (
it_next ==
m_index.get<ByTxHash>().end() ||
it_next->m_gtxid.ToUint256() != it->m_gtxid.ToUint256() ||
402 it_next->GetState() == State::COMPLETED) {
406 }
else if (
it_next->GetState() == State::CANDIDATE_BEST) {
423 if (it->IsSelected() && it !=
m_index.get<ByTxHash>().begin()) {
427 if (
it_prev->m_gtxid.ToUint256() == it->m_gtxid.ToUint256() &&
it_prev->GetState() == State::CANDIDATE_READY) {
439 assert(it->GetState() != State::COMPLETED);
443 if (it !=
m_index.get<ByTxHash>().begin() && std::prev(it)->m_gtxid.ToUint256() == it->m_gtxid.ToUint256())
return false;
446 if (std::next(it) !=
m_index.get<ByTxHash>().end() && std::next(it)->m_gtxid.ToUint256() == it->m_gtxid.ToUint256() &&
447 std::next(it)->GetState() != State::COMPLETED)
return false;
460 if (it->GetState() == State::COMPLETED)
return true;
467 }
while (it !=
m_index.get<ByTxHash>().end() && it->m_gtxid.ToUint256() ==
txhash);
482 void SetTimePoint(std::chrono::microseconds now, std::vector<std::pair<NodeId, GenTxid>>* expired)
484 if (expired) expired->clear();
489 auto it =
m_index.get<ByTime>().begin();
490 if (it->GetState() == State::CANDIDATE_DELAYED && it->m_time <= now) {
492 }
else if (it->GetState() == State::REQUESTED && it->m_time <= now) {
493 if (expired) expired->emplace_back(it->m_peer, it->m_gtxid);
504 auto it = std::prev(
m_index.get<ByTime>().end());
505 if (it->IsSelectable() && it->m_time > now) {
529 auto& index =
m_index.get<ByPeer>();
530 auto it = index.lower_bound(ByPeerView{peer,
false,
uint256::ZERO});
531 while (it != index.end() && it->m_peer == peer) {
545 auto it_next = (std::next(it) == index.end() || std::next(it)->m_peer != peer) ? index.end() :
561 while (it !=
m_index.get<ByTxHash>().end() && it->m_gtxid.ToUint256() ==
txhash) {
569 while (it !=
m_index.get<ByTxHash>().end() && it->m_gtxid.ToUint256() ==
txhash && it->GetState() != State::COMPLETED) {
576 std::chrono::microseconds
reqtime)
581 if (
m_index.get<ByPeer>().count(ByPeerView{peer, true, gtxid.ToUint256()}))
return;
587 if (!
ret.second)
return;
596 std::vector<std::pair<NodeId, GenTxid>>* expired)
602 std::vector<const Announcement*>
selected;
605 it_peer->GetState() == State::CANDIDATE_BEST) {
611 std::sort(
selected.begin(),
selected.end(), [](
const Announcement*
a,
const Announcement* b) {
612 return a->m_sequence < b->m_sequence;
616 std::vector<GenTxid>
ret;
626 auto it =
m_index.get<ByPeer>().find(ByPeerView{peer,
true,
txhash});
627 if (it ==
m_index.get<ByPeer>().end()) {
633 it =
m_index.get<ByPeer>().find(ByPeerView{peer,
false,
txhash});
634 if (it ==
m_index.get<ByPeer>().end() || (it->GetState() != State::CANDIDATE_DELAYED &&
635 it->GetState() != State::CANDIDATE_READY)) {
647 if (
it_old->GetState() == State::CANDIDATE_BEST) {
655 }
else if (
it_old->GetState() == State::REQUESTED) {
664 ann.SetState(State::REQUESTED);
672 auto it =
m_index.get<ByPeer>().find(ByPeerView{peer,
false,
txhash});
673 if (it ==
m_index.get<ByPeer>().end()) {
674 it =
m_index.get<ByPeer>().find(ByPeerView{peer,
true,
txhash});
682 if (it !=
m_peerinfo.end())
return it->second.m_requested;
689 if (it !=
m_peerinfo.end())
return it->second.m_total - it->second.m_requested - it->second.m_completed;
696 if (it !=
m_peerinfo.end())
return it->second.m_total;
727 m_impl->PostGetRequestableSanityCheck(now);
731 std::chrono::microseconds
reqtime)
747 std::vector<std::pair<NodeId, GenTxid>>* expired)
749 return m_impl->GetRequestable(peer, now, expired);
General SipHash-2-4 implementation.
uint64_t Finalize() const
Compute the 64-bit SipHash-2-4 of the data written so far.
CSipHasher & Write(uint64_t data)
Hash a 64-bit integer worth of data.
Actual implementation for TxRequestTracker's data structure.
const PriorityComputer m_computer
This tracker's priority computer.
std::vector< GenTxid > GetRequestable(NodeId peer, std::chrono::microseconds now, std::vector< std::pair< NodeId, GenTxid > > *expired)
Find the GenTxids to request now from peer.
void SetTimePoint(std::chrono::microseconds now, std::vector< std::pair< NodeId, GenTxid > > *expired)
Make the data structure consistent with a given point in time:
void PromoteCandidateReady(Iter< ByTxHash > it)
Convert a CANDIDATE_DELAYED announcement into a CANDIDATE_READY.
SequenceNumber m_current_sequence
The current sequence number.
Iter< Tag > Erase(Iter< Tag > it)
Wrapper around Index::...::erase that keeps m_peerinfo up to date.
void GetCandidatePeers(const uint256 &txhash, std::vector< NodeId > &result_peers) const
void ReceivedResponse(NodeId peer, const uint256 &txhash)
size_t CountInFlight(NodeId peer) const
size_t Size() const
Count how many announcements are being tracked in total across all peers and transactions.
void ReceivedInv(NodeId peer, const GenTxid >xid, bool preferred, std::chrono::microseconds reqtime)
Impl(const Impl &)=delete
bool MakeCompleted(Iter< ByTxHash > it)
Convert any announcement to a COMPLETED one.
void PostGetRequestableSanityCheck(std::chrono::microseconds now) const
Index m_index
This tracker's main data structure. See SanityCheck() for the invariants that apply to it.
bool IsOnlyNonCompleted(Iter< ByTxHash > it)
Check if 'it' is the only announcement for a given txhash that isn't COMPLETED.
void ForgetTxHash(const uint256 &txhash)
uint64_t ComputePriority(const uint256 &txhash, NodeId peer, bool preferred) const
std::unordered_map< NodeId, PeerInfo > m_peerinfo
Map with this tracker's per-peer statistics.
void RequestedTx(NodeId peer, const uint256 &txhash, std::chrono::microseconds expiry)
void ChangeAndReselect(Iter< ByTxHash > it, State new_state)
Change the state of an announcement to something non-IsSelected().
size_t CountCandidates(NodeId peer) const
Impl & operator=(const Impl &)=delete
size_t Count(NodeId peer) const
void Modify(Iter< Tag > it, Modifier modifier)
Wrapper around Index::...::modify that keeps m_peerinfo up to date.
void DisconnectedPeer(NodeId peer)
Data structure to keep track of, and schedule, transaction downloads from peers.
void ReceivedInv(NodeId peer, const GenTxid >xid, bool preferred, std::chrono::microseconds reqtime)
Adds a new CANDIDATE announcement.
void SanityCheck() const
Run internal consistency check (testing only).
size_t CountInFlight(NodeId peer) const
Count how many REQUESTED announcements a peer has.
void GetCandidatePeers(const uint256 &txhash, std::vector< NodeId > &result_peers) const
For some txhash (txid or wtxid), finds all peers with non-COMPLETED announcements and appends them to...
size_t CountCandidates(NodeId peer) const
Count how many CANDIDATE announcements a peer has.
TxRequestTracker(bool deterministic=false)
Construct a TxRequestTracker.
const std::unique_ptr< Impl > m_impl
void DisconnectedPeer(NodeId peer)
Deletes all announcements for a given peer.
void ReceivedResponse(NodeId peer, const uint256 &txhash)
Converts a CANDIDATE or REQUESTED announcement to a COMPLETED one.
uint64_t ComputePriority(const uint256 &txhash, NodeId peer, bool preferred) const
Access to the internal priority computation (testing only)
void PostGetRequestableSanityCheck(std::chrono::microseconds now) const
Run a time-dependent internal consistency check (testing only).
void RequestedTx(NodeId peer, const uint256 &txhash, std::chrono::microseconds expiry)
Marks a transaction as requested, with a specified expiry.
size_t Count(NodeId peer) const
Count how many announcements a peer has (REQUESTED, CANDIDATE, and COMPLETED combined).
size_t Size() const
Count how many announcements are being tracked in total across all peers and transaction hashes.
std::vector< GenTxid > GetRequestable(NodeId peer, std::chrono::microseconds now, std::vector< std::pair< NodeId, GenTxid > > *expired=nullptr)
Find the txids to request now from peer.
void ForgetTxHash(const uint256 &txhash)
Deletes all announcements for a given txhash (both txid and wtxid ones).
static const uint256 ZERO
bool operator==(const CNetAddr &a, const CNetAddr &b)
constexpr auto Ticks(Dur2 d)
Helper to count the seconds of a duration/time_point.