| 1008 | } |
| 1009 | |
| 1010 | void NodeImpl::fetch_sync_items_loop() |
| 1011 | { |
| 1012 | VERIFY_CORRECT_THREAD(); |
| 1013 | while (!_fetch_sync_items_loop_done.canceled()) |
| 1014 | { |
| 1015 | _sync_items_to_fetch_updated = false; |
| 1016 | dlog("beginning another iteration of the sync items loop"); |
| 1017 | |
| 1018 | if (!_suspend_fetching_sync_blocks) |
| 1019 | { |
| 1020 | std::map<PeerConnectionPtr, std::vector<ItemHashType> > sync_item_requests_to_send; |
| 1021 | |
| 1022 | { |
| 1023 | ASSERT_TASK_NOT_PREEMPTED(); |
| 1024 | std::set<ItemHashType> sync_items_to_request; |
| 1025 | |
| 1026 | // for each idle peer that we're syncing with |
| 1027 | for (const PeerConnectionPtr& peer : _active_connections) |
| 1028 | { |
| 1029 | if (peer->we_need_sync_items_from_peer && |
| 1030 | sync_item_requests_to_send.find(peer) == sync_item_requests_to_send.end() && // if we've already scheduled a request for this peer, don't consider scheduling another |
| 1031 | peer->idle()) |
| 1032 | { |
| 1033 | if (!peer->inhibit_fetching_sync_blocks) |
| 1034 | { |
| 1035 | // loop through the items it has that we don't yet have on our blockchain |
| 1036 | for (unsigned i = 0; i < peer->ids_of_items_to_get.size(); ++i) |
| 1037 | { |
| 1038 | ItemHashType item_to_potentially_request = peer->ids_of_items_to_get[i]; |
| 1039 | // if we don't already have this item in our temporary storage and we haven't requested from another syncing peer |
| 1040 | if (!have_already_received_sync_item(item_to_potentially_request) && // already got it, but for some reson it's still in our list of items to fetch |
| 1041 | sync_items_to_request.find(item_to_potentially_request) == sync_items_to_request.end() && // we have already decided to request it from another peer during this iteration |
| 1042 | _active_sync_requests.find(item_to_potentially_request) == _active_sync_requests.end()) // we've requested it in a previous iteration and we're still waiting for it to arrive |
| 1043 | { |
| 1044 | // then schedule a request from this peer |
| 1045 | sync_item_requests_to_send[peer].push_back(item_to_potentially_request); |
| 1046 | sync_items_to_request.insert(item_to_potentially_request); |
| 1047 | if (sync_item_requests_to_send[peer].size() >= _maximum_blocks_per_peer_during_syncing) |
| 1048 | break; |
| 1049 | } |
| 1050 | } |
| 1051 | } |
| 1052 | } |
| 1053 | } |
| 1054 | } // end non-preemptable section |
| 1055 | |
| 1056 | // make all the requests we scheduled in the loop above |
| 1057 | for (auto sync_item_request : sync_item_requests_to_send) |
| 1058 | request_sync_items_from_peer(sync_item_request.first, sync_item_request.second); |
| 1059 | sync_item_requests_to_send.clear(); |
| 1060 | } |
| 1061 | else |
| 1062 | dlog("fetch_sync_items_loop is suspended pending backlog processing"); |
| 1063 | |
| 1064 | if (!_sync_items_to_fetch_updated) |
| 1065 | { |
| 1066 | dlog("no sync items to fetch right now, going to sleep"); |
| 1067 | _retrigger_fetch_sync_items_loop_promise = fc::promise<void>::ptr(new fc::promise<void>("thinkyoung::net::retrigger_fetch_sync_items_loop")); |