| 3176 | } |
| 3177 | |
| 3178 | void node_impl::process_backlog_of_sync_blocks() |
| 3179 | { |
| 3180 | VERIFY_CORRECT_THREAD(); |
| 3181 | // garbage-collect the list of async tasks here for lack of a better place |
| 3182 | for (auto calls_iter = _handle_message_calls_in_progress.begin(); |
| 3183 | calls_iter != _handle_message_calls_in_progress.end();) |
| 3184 | { |
| 3185 | if (calls_iter->ready()) |
| 3186 | calls_iter = _handle_message_calls_in_progress.erase(calls_iter); |
| 3187 | else |
| 3188 | ++calls_iter; |
| 3189 | } |
| 3190 | |
| 3191 | dlog("in process_backlog_of_sync_blocks"); |
| 3192 | if (_handle_message_calls_in_progress.size() >= _maximum_number_of_blocks_to_handle_at_one_time) |
| 3193 | { |
| 3194 | dlog("leaving process_backlog_of_sync_blocks because we're already processing too many blocks"); |
| 3195 | return; // we will be rescheduled when the next block finishes its processing |
| 3196 | } |
| 3197 | dlog("currently ${count} blocks in the process of being handled", ("count", _handle_message_calls_in_progress.size())); |
| 3198 | |
| 3199 | |
| 3200 | if (_suspend_fetching_sync_blocks) |
| 3201 | { |
| 3202 | dlog("resuming processing sync block backlog because we only ${count} blocks in progress", |
| 3203 | ("count", _handle_message_calls_in_progress.size())); |
| 3204 | _suspend_fetching_sync_blocks = false; |
| 3205 | } |
| 3206 | |
| 3207 | |
| 3208 | // when syncing with multiple peers, it's possible that we'll have hundreds of blocks ready to push |
| 3209 | // to the client at once. This can be slow, and we need to limit the number we push at any given |
| 3210 | // time to allow network traffic to continue so we don't end up disconnecting from peers |
| 3211 | //fc::time_point start_time = fc::time_point::now(); |
| 3212 | //fc::time_point when_we_should_yield = start_time + fc::seconds(1); |
| 3213 | |
| 3214 | bool block_processed_this_iteration; |
| 3215 | unsigned blocks_processed = 0; |
| 3216 | |
| 3217 | std::set<peer_connection_ptr> peers_with_newly_empty_item_lists; |
| 3218 | std::set<peer_connection_ptr> peers_we_need_to_sync_to; |
| 3219 | std::map<peer_connection_ptr, fc::oexception> peers_with_rejected_block; |
| 3220 | |
| 3221 | do |
| 3222 | { |
| 3223 | std::copy(std::make_move_iterator(_new_received_sync_items.begin()), |
| 3224 | std::make_move_iterator(_new_received_sync_items.end()), |
| 3225 | std::front_inserter(_received_sync_items)); |
| 3226 | _new_received_sync_items.clear(); |
| 3227 | dlog("currently ${count} sync items to consider", ("count", _received_sync_items.size())); |
| 3228 | |
| 3229 | block_processed_this_iteration = false; |
| 3230 | for (auto received_block_iter = _received_sync_items.begin(); |
| 3231 | received_block_iter != _received_sync_items.end(); |
| 3232 | ++received_block_iter) |
| 3233 | { |
| 3234 | |
| 3235 | // find out if this block is the next block on the active chain or one of the forks |