MCPcopy Create free account
hub / github.com/cryptonomex/graphene / fetch_items_loop

Method fetch_items_loop

libraries/net/node.cpp:1108–1225  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

1106 }
1107
1108 void node_impl::fetch_items_loop()
1109 {
1110 VERIFY_CORRECT_THREAD();
1111 while (!_fetch_item_loop_done.canceled())
1112 {
1113 _items_to_fetch_updated = false;
1114 dlog("beginning an iteration of fetch items (${count} items to fetch)",
1115 ("count", _items_to_fetch.size()));
1116
1117 fc::time_point oldest_timestamp_to_fetch = fc::time_point::now() - fc::seconds(_recent_block_interval_in_seconds * GRAPHENE_NET_MESSAGE_CACHE_DURATION_IN_BLOCKS);
1118 fc::time_point next_peer_unblocked_time = fc::time_point::maximum();
1119
1120 // we need to construct a list of items to request from each peer first,
1121 // then send the messages (in two steps, to avoid yielding while iterating)
1122 // we want to evenly distribute our requests among our peers.
1123 struct requested_item_count_index {};
1124 struct peer_and_items_to_fetch
1125 {
1126 peer_connection_ptr peer;
1127 std::vector<item_id> item_ids;
1128 peer_and_items_to_fetch(const peer_connection_ptr& peer) : peer(peer) {}
1129 bool operator<(const peer_and_items_to_fetch& rhs) const { return peer < rhs.peer; }
1130 size_t number_of_items() const { return item_ids.size(); }
1131 };
1132 typedef boost::multi_index_container<peer_and_items_to_fetch,
1133 boost::multi_index::indexed_by<boost::multi_index::ordered_unique<boost::multi_index::member<peer_and_items_to_fetch, peer_connection_ptr, &peer_and_items_to_fetch::peer> >,
1134 boost::multi_index::ordered_non_unique<boost::multi_index::tag<requested_item_count_index>,
1135 boost::multi_index::const_mem_fun<peer_and_items_to_fetch, size_t, &peer_and_items_to_fetch::number_of_items> > > > fetch_messages_to_send_set;
1136 fetch_messages_to_send_set items_by_peer;
1137
1138 // initialize the fetch_messages_to_send with an empty set of items for all idle peers
1139 for (const peer_connection_ptr& peer : _active_connections)
1140 if (peer->idle())
1141 items_by_peer.insert(peer_and_items_to_fetch(peer));
1142
1143 // now loop over all items we want to fetch
1144 for (auto item_iter = _items_to_fetch.begin(); item_iter != _items_to_fetch.end();)
1145 {
1146 if (item_iter->timestamp < oldest_timestamp_to_fetch)
1147 {
1148 // this item has probably already fallen out of our peers' caches, we'll just ignore it.
1149 // this can happen during flooding, and the _items_to_fetch could otherwise get clogged
1150 // with a bunch of items that we'll never be able to request from any peer
1151 wlog("Unable to fetch item ${item} before its likely expiration time, removing it from our list of items to fetch", ("item", item_iter->item));
1152 item_iter = _items_to_fetch.erase(item_iter);
1153 }
1154 else
1155 {
1156 // find a peer that has it, we'll use the one who has the least requests going to it to load balance
1157 bool item_fetched = false;
1158 for (auto peer_iter = items_by_peer.get<requested_item_count_index>().begin(); peer_iter != items_by_peer.get<requested_item_count_index>().end(); ++peer_iter)
1159 {
1160 const peer_connection_ptr& peer = peer_iter->peer;
1161 // if they have the item and we haven't already decided to ask them for too many other items
1162 if (peer_iter->item_ids.size() < GRAPHENE_NET_MAX_ITEMS_PER_PEER_DURING_NORMAL_OPERATION &&
1163 peer->inventory_peer_advertised_to_us.find(item_iter->item) != peer->inventory_peer_advertised_to_us.end())
1164 {
1165 if (item_iter->item.item_type == graphene::net::trx_message_type && peer->is_transaction_fetching_inhibited())

Callers

nothing calls this directly

Calls 15

nowFunction · 0.85
fetch_items_messageClass · 0.85
idleMethod · 0.80
eraseMethod · 0.80
resetMethod · 0.80
sizeMethod · 0.45
beginMethod · 0.45
endMethod · 0.45
findMethod · 0.45
modifyMethod · 0.45

Tested by

no test coverage detected