| 1023 | } |
| 1024 | |
| 1025 | void node_impl::fetch_sync_items_loop() |
| 1026 | { |
| 1027 | VERIFY_CORRECT_THREAD(); |
| 1028 | while( !_fetch_sync_items_loop_done.canceled() ) |
| 1029 | { |
| 1030 | _sync_items_to_fetch_updated = false; |
| 1031 | dlog( "beginning another iteration of the sync items loop" ); |
| 1032 | |
| 1033 | if (!_suspend_fetching_sync_blocks) |
| 1034 | { |
| 1035 | std::map<peer_connection_ptr, std::vector<item_hash_t> > sync_item_requests_to_send; |
| 1036 | |
| 1037 | { |
| 1038 | ASSERT_TASK_NOT_PREEMPTED(); |
| 1039 | std::set<item_hash_t> sync_items_to_request; |
| 1040 | |
| 1041 | // for each idle peer that we're syncing with |
| 1042 | for( const peer_connection_ptr& peer : _active_connections ) |
| 1043 | { |
| 1044 | if( peer->we_need_sync_items_from_peer && |
| 1045 | 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 |
| 1046 | peer->idle() ) |
| 1047 | { |
| 1048 | if (!peer->inhibit_fetching_sync_blocks) |
| 1049 | { |
| 1050 | // loop through the items it has that we don't yet have on our blockchain |
| 1051 | for( unsigned i = 0; i < peer->ids_of_items_to_get.size(); ++i ) |
| 1052 | { |
| 1053 | item_hash_t item_to_potentially_request = peer->ids_of_items_to_get[i]; |
| 1054 | // if we don't already have this item in our temporary storage and we haven't requested from another syncing peer |
| 1055 | 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 |
| 1056 | 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 |
| 1057 | _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 |
| 1058 | { |
| 1059 | // then schedule a request from this peer |
| 1060 | sync_item_requests_to_send[peer].push_back(item_to_potentially_request); |
| 1061 | sync_items_to_request.insert( item_to_potentially_request ); |
| 1062 | if (sync_item_requests_to_send[peer].size() >= _maximum_blocks_per_peer_during_syncing) |
| 1063 | break; |
| 1064 | } |
| 1065 | } |
| 1066 | } |
| 1067 | } |
| 1068 | } |
| 1069 | } // end non-preemptable section |
| 1070 | |
| 1071 | // make all the requests we scheduled in the loop above |
| 1072 | for( auto sync_item_request : sync_item_requests_to_send ) |
| 1073 | request_sync_items_from_peer( sync_item_request.first, sync_item_request.second ); |
| 1074 | sync_item_requests_to_send.clear(); |
| 1075 | } |
| 1076 | else |
| 1077 | dlog("fetch_sync_items_loop is suspended pending backlog processing"); |
| 1078 | |
| 1079 | if( !_sync_items_to_fetch_updated ) |
| 1080 | { |
| 1081 | dlog( "no sync items to fetch right now, going to sleep" ); |
| 1082 | _retrigger_fetch_sync_items_loop_promise = fc::promise<void>::ptr( new fc::promise<void>("graphene::net::retrigger_fetch_sync_items_loop") ); |