diff --git a/src/cryptonote_core/cryptonote_core.cpp b/src/cryptonote_core/cryptonote_core.cpp index d697859b7..473afa450 100644 --- a/src/cryptonote_core/cryptonote_core.cpp +++ b/src/cryptonote_core/cryptonote_core.cpp @@ -124,6 +124,11 @@ namespace cryptonote , "Set maximum size of block download queue in bytes (0 for default)" , 0 }; + const command_line::arg_descriptor arg_span_limit = { + "span-limit" + , "Defines how many minutes of block synchronization data to request at a time (default is 2 minutes)" + , 2 + }; const command_line::arg_descriptor arg_sync_pruned_blocks = { "sync-pruned-blocks" , "Allow syncing from nodes with only pruned blocks" @@ -334,6 +339,7 @@ namespace cryptonote command_line::add_arg(desc, arg_offline); command_line::add_arg(desc, arg_disable_dns_checkpoints); command_line::add_arg(desc, arg_block_download_max_size); + command_line::add_arg(desc, arg_span_limit); command_line::add_arg(desc, arg_sync_pruned_blocks); command_line::add_arg(desc, arg_max_txpool_weight); command_line::add_arg(desc, arg_block_notify); diff --git a/src/cryptonote_core/cryptonote_core.h b/src/cryptonote_core/cryptonote_core.h index 290422d61..fd10f6200 100644 --- a/src/cryptonote_core/cryptonote_core.h +++ b/src/cryptonote_core/cryptonote_core.h @@ -69,6 +69,7 @@ namespace cryptonote extern const command_line::arg_descriptor arg_fixed_difficulty; extern const command_line::arg_descriptor arg_offline; extern const command_line::arg_descriptor arg_block_download_max_size; + extern const command_line::arg_descriptor arg_span_limit; extern const command_line::arg_descriptor arg_sync_pruned_blocks; /************************************************************************/ diff --git a/src/cryptonote_protocol/block_queue.cpp b/src/cryptonote_protocol/block_queue.cpp index 52ad94d7a..5ed9e997d 100644 --- a/src/cryptonote_protocol/block_queue.cpp +++ b/src/cryptonote_protocol/block_queue.cpp @@ -83,6 +83,7 @@ void block_queue::flush_spans(const boost::uuids::uuid &connection_id, bool all) void block_queue::erase_block(block_map::iterator j) { + MDEBUG("Erasing span starting from block " << j->start_block_height); CHECK_AND_ASSERT_THROW_MES(j != blocks.end(), "Invalid iterator"); for (const crypto::hash &h: j->hashes) { @@ -152,7 +153,10 @@ uint64_t block_queue::get_next_needed_height(uint64_t blockchain_height) const { boost::unique_lock lock(mutex); if (blocks.empty()) + { + MDEBUG("Span queue is empty"); return blockchain_height; + } uint64_t covered_until = blockchain_height; @@ -165,7 +169,10 @@ uint64_t block_queue::get_next_needed_height(uint64_t blockchain_height) const // If this span starts after what we already have/scheduled, we found the first gap if (span.start_block_height > covered_until) + { + MDEBUG("Found gap in the span queue from " << covered_until << " to " << span.start_block_height); return covered_until; + } // This span overlaps or is adjacent; extend coverage regardless of filled/scheduled if (span.start_block_height <= covered_until) diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.h b/src/cryptonote_protocol/cryptonote_protocol_handler.h index dbf1c870d..a7dc77c27 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.h +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.h @@ -180,6 +180,7 @@ namespace cryptonote size_t skip_unneeded_hashes(cryptonote_connection_context& context, bool check_block_queue) const; bool request_txpool_complement(cryptonote_connection_context &context); void hit_score(cryptonote_connection_context &context, int32_t score); + void calculate_dynamic_span(const double blocks_per_seconds); t_core& m_core; @@ -204,6 +205,9 @@ namespace cryptonote uint64_t m_sync_download_chain_size, m_sync_download_objects_size; size_t m_block_download_max_size; bool m_sync_pruned_blocks; + size_t m_span_time; + std::atomic m_span_limit; + std::atomic m_bss; // Values for sync time estimates boost::posix_time::ptime m_sync_start_time; diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.inl b/src/cryptonote_protocol/cryptonote_protocol_handler.inl index 0cf1bb52a..f85171f0d 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.inl +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.inl @@ -70,9 +70,8 @@ #define MLOG_PEER_STATE(x) \ MCINFO(MONERO_DEFAULT_LOG_CATEGORY, context << "[" << epee::string_tools::to_string_hex(context.m_pruning_seed) << "] state: " << x << " in state " << cryptonote::get_protocol_state_string(context.m_state)) -#define BLOCK_QUEUE_NSPANS_THRESHOLD 10 // chunks of N blocks +#define BLOCK_QUEUE_NSPANS_MINIMUM 10 // minimum number of spans #define BLOCK_QUEUE_SIZE_THRESHOLD (100*1024*1024) // MB -#define BLOCK_QUEUE_FORCE_DOWNLOAD_NEAR_BLOCKS 1000 #define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD_STANDBY (5 * 1000000) // microseconds #define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD (30 * 1000000) // microseconds #define IDLE_PEER_KICK_TIME (240 * 1000000) // microseconds @@ -189,7 +188,10 @@ namespace cryptonote m_synchronized(offline), m_ask_for_txpool_complement(true), m_stopping(false), - m_no_sync(false) + m_no_sync(false), + m_span_limit(BLOCK_QUEUE_NSPANS_MINIMUM), + m_span_time(0), + m_bss(0) { if(!m_p2p) @@ -212,11 +214,25 @@ namespace cryptonote m_block_download_max_size = command_line::get_arg(vm, cryptonote::arg_block_download_max_size); m_sync_pruned_blocks = command_line::get_arg(vm, cryptonote::arg_sync_pruned_blocks); + m_span_time = command_line::get_arg(vm, cryptonote::arg_span_limit); return true; } //------------------------------------------------------------------------------------------------------------------------ template + void t_cryptonote_protocol_handler::calculate_dynamic_span(const double blocks_per_seconds) + { + size_t current_bss = m_bss.load(); + size_t current_span_limit = m_span_limit.load(); + MINFO("m_bss : " << current_bss << ", blocks_per_seconds : " << blocks_per_seconds << ", current_span_limit : " << current_span_limit); + current_span_limit = (current_bss && blocks_per_seconds) ? (( blocks_per_seconds * 60 * m_span_time ) / current_bss) : BLOCK_QUEUE_NSPANS_MINIMUM; + if (current_span_limit < BLOCK_QUEUE_NSPANS_MINIMUM) + current_span_limit = BLOCK_QUEUE_NSPANS_MINIMUM; + m_span_limit = current_span_limit; + MINFO("calculated dynamic span limit is span_limit : " << m_span_limit); + } + //------------------------------------------------------------------------------------------------------------------------ + template bool t_cryptonote_protocol_handler::deinit() { return true; @@ -1588,6 +1604,8 @@ namespace cryptonote { const uint64_t target_blockchain_height = m_core.get_target_blockchain_height(); const boost::posix_time::time_duration dt = boost::posix_time::microsec_clock::universal_time() - start; + const double blocks_per_seconds = (((current_blockchain_height - previous_height) * 1e6) / dt.total_microseconds()); + calculate_dynamic_span(blocks_per_seconds); std::string progress_message = ""; if (current_blockchain_height < target_blockchain_height) { @@ -1611,7 +1629,7 @@ namespace cryptonote std::string timing_message = ""; if (ELPP->vRegistry()->allowed(el::Level::Info, "sync-info")) timing_message = std::string(" (") + std::to_string(dt.total_microseconds()/1e6) + " sec, " - + std::to_string((current_blockchain_height - previous_height) * 1e6 / dt.total_microseconds()) + + std::to_string(blocks_per_seconds) + " blocks/sec), " + std::to_string(m_block_queue.get_data_size() / 1048576.f) + " MB queued in " + std::to_string(m_block_queue.get_num_filled_spans()) + " spans, stripe " + std::to_string(previous_stripe) + " -> " + std::to_string(current_stripe); @@ -2021,15 +2039,15 @@ skip: { do { - size_t nspans = m_block_queue.get_num_filled_spans(); - size_t size = m_block_queue.get_data_size(); + const size_t nspans = m_block_queue.get_num_filled_spans(); + const size_t size = m_block_queue.get_data_size(); const uint64_t bc_height = m_core.get_current_blockchain_height(); const auto next_needed_pruning_stripe = get_next_needed_pruning_stripe(); const uint32_t add_stripe = tools::get_pruning_stripe(bc_height, context.m_remote_blockchain_height, CRYPTONOTE_PRUNING_LOG_STRIPES); const uint32_t peer_stripe = tools::get_pruning_stripe(context.m_pruning_seed); const uint32_t local_stripe = tools::get_pruning_stripe(m_core.get_blockchain_pruning_seed()); const size_t block_queue_size_threshold = m_block_download_max_size ? m_block_download_max_size : BLOCK_QUEUE_SIZE_THRESHOLD; - bool queue_proceed = nspans < BLOCK_QUEUE_NSPANS_THRESHOLD || size < block_queue_size_threshold; + const bool queue_proceed_init = (nspans < m_span_limit.load()) && (size < block_queue_size_threshold); // get rid of blocks we already requested, or already have if (skip_unneeded_hashes(context, true) && context.m_needed_objects.empty() && context.m_num_requested == 0) { @@ -2044,15 +2062,16 @@ skip: on_connection_synchronized(); return true; } - uint64_t next_needed_height = m_block_queue.get_next_needed_height(bc_height); - uint64_t next_block_height; - if (context.m_needed_objects.empty()) - next_block_height = next_needed_height; - else - next_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1; - bool stripe_proceed_main = ((m_sync_pruned_blocks && local_stripe && add_stripe != local_stripe) || add_stripe == 0 || peer_stripe == 0 || add_stripe == peer_stripe) && (next_block_height < bc_height + BLOCK_QUEUE_FORCE_DOWNLOAD_NEAR_BLOCKS || next_needed_height < bc_height + BLOCK_QUEUE_FORCE_DOWNLOAD_NEAR_BLOCKS); - bool stripe_proceed_secondary = tools::has_unpruned_block(next_block_height, context.m_remote_blockchain_height, context.m_pruning_seed); - bool proceed = stripe_proceed_main || (queue_proceed && stripe_proceed_secondary); + const uint64_t next_needed_height = m_block_queue.get_next_needed_height(bc_height); + const uint64_t next_block_height = context.m_needed_objects.empty() + ? next_needed_height + : context.m_last_response_height - context.m_needed_objects.size() + 1; + const bool next_height_proceed = next_needed_height < std::max(next_block_height, bc_height + 1); + const bool stripe_proceed_main = next_height_proceed && ((m_sync_pruned_blocks && local_stripe && add_stripe != local_stripe) || add_stripe == 0 || peer_stripe == 0 || add_stripe == peer_stripe); + const bool stripe_proceed_secondary = tools::has_unpruned_block(next_block_height, context.m_remote_blockchain_height, context.m_pruning_seed); + // override queue_proceed_init if we need the immediate block(s) + const bool queue_proceed = (next_needed_height == bc_height) ? stripe_proceed_main : queue_proceed_init; + const bool proceed = queue_proceed && (stripe_proceed_main || stripe_proceed_secondary); if (!stripe_proceed_main && !stripe_proceed_secondary && should_drop_connection(context, tools::get_pruning_stripe(next_block_height, context.m_remote_blockchain_height, CRYPTONOTE_PRUNING_LOG_STRIPES))) { if (!context.m_is_income) @@ -2060,12 +2079,21 @@ skip: return false; // drop outgoing connections } - MDEBUG(context << "proceed " << proceed << " (queue " << queue_proceed << ", stripe " << stripe_proceed_main << "/" << - stripe_proceed_secondary << "), " << next_needed_pruning_stripe.first << "-" << next_needed_pruning_stripe.second << - " needed, bc add stripe " << add_stripe << ", we have " << peer_stripe << "), bc_height " << bc_height); - MDEBUG(context << " - next_block_height " << next_block_height << ", seed " << epee::string_tools::to_string_hex(context.m_pruning_seed) << - ", next_needed_height "<< next_needed_height); - MDEBUG(context << " - last_response_height " << context.m_last_response_height << ", m_needed_objects size " << context.m_needed_objects.size()); + MDEBUG(context + << "add_stripe : " << add_stripe + << ", peer_stripe : " << peer_stripe + << ", local_stripe : " << local_stripe + << ", next_needed_pruning_stripe first-second : " << next_needed_pruning_stripe.first << "-" << next_needed_pruning_stripe.second << " needed" + << ", seed " << epee::string_tools::to_string_hex(context.m_pruning_seed) + << ", last_response_height " << context.m_last_response_height << ", m_needed_objects size " << context.m_needed_objects.size() + << ", proceed : " << proceed + << ", queue_proceed : " << queue_proceed + << ", stripe_proceed_main : " << stripe_proceed_main + << ", stripe_proceed_secondary : " << stripe_proceed_secondary + << ", next_height_proceed : " << next_height_proceed + << ", next_block_height/next_needed_height/bc_height : " << next_block_height << "/" << next_needed_height << "/" << bc_height + << ", nspans/span_limit : " << nspans << "/" << m_span_limit + << ", queue size/size_limit : " << size << "/" << block_queue_size_threshold); // if we're waiting for next span, try to get it before unblocking threads below, // or a runaway downloading of future spans might happen @@ -2147,7 +2175,7 @@ skip: NOTIFY_REQUEST_GET_OBJECTS::request req; bool is_next = false; size_t count = 0; - size_t l_m_bss = m_core.get_block_sync_size(m_core.get_current_blockchain_height(), max_average_of_blocksize_in_queue()); + size_t l_m_bss = m_bss = m_core.get_block_sync_size(m_core.get_current_blockchain_height(), max_average_of_blocksize_in_queue()); std::pair span = std::make_pair(0, 0); if (force_next_span) { @@ -2407,7 +2435,7 @@ skip: { synced_seconds = 1; } - float blocks_per_second = (1000 * synced_blocks / synced_seconds) / 1000.0f; + const float blocks_per_second = (1000 * synced_blocks / synced_seconds) / 1000.0f; MGINFO_YELLOW("Synced " << synced_blocks << " blocks in " << tools::get_human_readable_timespan(synced_seconds) << " (" << blocks_per_second << " blocks per second)"); }