src: dynamic span, to calculate span limit dynamically

Co-authored-by: nahuhh
    - jberman review
    - cryptonote_protocol: don't arbitrarily download 1000 blocks ahead
    - further restrict `proceed` to require `queue_proceed` in all cases.
      ensure queue_proceed is true if we need the next block, even if we
      already exceed the span and size limits

cryptonote_protocol: improved logging + const usage in span downloader

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<size_t> 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<bool> 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);

src/cryptonote_core/cryptonote_core.h

@@ -69,6 +69,7 @@ namespace cryptonote
   extern const command_line::arg_descriptor<difficulty_type> arg_fixed_difficulty;
   extern const command_line::arg_descriptor<bool> arg_offline;
   extern const command_line::arg_descriptor<size_t> arg_block_download_max_size;
+  extern const command_line::arg_descriptor<size_t> arg_span_limit;
   extern const command_line::arg_descriptor<bool> arg_sync_pruned_blocks;
 
   /************************************************************************/

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<boost::recursive_mutex> 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)

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<size_t> m_span_limit;
+    std::atomic<size_t> m_bss;
 
     // Values for sync time estimates
     boost::posix_time::ptime m_sync_start_time;

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<class t_core>
+  void t_cryptonote_protocol_handler<t_core>::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<class t_core>
   bool t_cryptonote_protocol_handler<t_core>::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();
+        const size_t nspans = m_block_queue.get_num_filled_spans();
-        size_t size = m_block_queue.get_data_size();
+        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);
+        const uint64_t next_needed_height = m_block_queue.get_next_needed_height(bc_height);
-        uint64_t next_block_height;
+        const uint64_t next_block_height = context.m_needed_objects.empty()
-        if (context.m_needed_objects.empty())
+          ? next_needed_height
-          next_block_height = next_needed_height;
+          : context.m_last_response_height - context.m_needed_objects.size() + 1;
-        else
+        const bool next_height_proceed = next_needed_height < std::max(next_block_height, bc_height + 1);
-          next_block_height = context.m_last_response_height - context.m_needed_objects.size() + 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);
-        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);
+        const bool stripe_proceed_secondary = tools::has_unpruned_block(next_block_height, context.m_remote_blockchain_height, context.m_pruning_seed);
-        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)
-        bool proceed = stripe_proceed_main || (queue_proceed && stripe_proceed_secondary);
+        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 << "/" <<
+        MDEBUG(context
-          stripe_proceed_secondary << "), " << next_needed_pruning_stripe.first << "-" << next_needed_pruning_stripe.second <<
+               << "add_stripe : " << add_stripe
-          " needed, bc add stripe " << add_stripe << ", we have " << peer_stripe << "), bc_height " << bc_height);
+               << ", peer_stripe : " << peer_stripe
-        MDEBUG(context << "  - next_block_height " << next_block_height << ", seed " << epee::string_tools::to_string_hex(context.m_pruning_seed) <<
+               << ", local_stripe : " << local_stripe
-          ", next_needed_height "<< next_needed_height);
+               << ", next_needed_pruning_stripe first-second : " << next_needed_pruning_stripe.first << "-" << next_needed_pruning_stripe.second << " needed"
-        MDEBUG(context << "  - last_response_height " << context.m_last_response_height << ", m_needed_objects size " << context.m_needed_objects.size());
+               << ", 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<uint64_t, uint64_t> 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)");
         }