src: dynamic bss and dynamic span

src/cryptonote_config.h

@@ -93,11 +93,16 @@
 #define DIFFICULTY_BLOCKS_ESTIMATE_TIMESPAN             DIFFICULTY_TARGET_V1 //just alias; used by tests
 
 
-#define BLOCKS_IDS_SYNCHRONIZING_DEFAULT_COUNT          10000  //by default, blocks ids count in synchronizing
-#define BLOCKS_IDS_SYNCHRONIZING_MAX_COUNT              25000  //max blocks ids count in synchronizing
-#define BLOCKS_SYNCHRONIZING_DEFAULT_COUNT_PRE_V4       100    //by default, blocks count in blocks downloading
-#define BLOCKS_SYNCHRONIZING_DEFAULT_COUNT              5     //by default, blocks count in blocks downloading
-#define BLOCKS_SYNCHRONIZING_MAX_COUNT                  2048   //must be a power of 2, greater than 128, equal to SEEDHASH_EPOCH_BLOCKS
+#define BLOCKS_IDS_SYNCHRONIZING_DEFAULT_COUNT          10000    //by default, blocks ids count in synchronizing
+#define BLOCKS_IDS_SYNCHRONIZING_MAX_COUNT              25000    //max blocks ids count in synchronizing
+#define BLOCKS_SYNCHRONIZING_DEFAULT_COUNT_PRE_V4       100      //by default, blocks count in blocks downloading
+#define BLOCKS_SYNCHRONIZING_DEFAULT_COUNT              20       //by default, blocks count in blocks downloading
+#define BLOCKS_MEDIAN_WINDOW                            100      //by default, compute median weights of last 100 blocks
+#define BATCH_MAX_WEIGHT                                20       //by default, maximum size of batch in [mB]
+#define BATCH_MAX_ALLOWED_WEIGHT                        50       //maximum allowed size of batch in [mB]
+#define BLOCKS_HUGE_THRESHOLD_SIZE                      ((BATCH_MAX_WEIGHT * 1000000) / 2) //blocks that we consider huge [B]
+#define BLOCKS_SYNCHRONIZING_MAX_COUNT                  2048     //must be a power of 2, greater than 128, equal to SEEDHASH_EPOCH_BLOCKS
+
 
 #define CRYPTONOTE_MEMPOOL_TX_LIVETIME                    (86400*3) //seconds, three days
 #define CRYPTONOTE_MEMPOOL_TX_FROM_ALT_BLOCK_LIVETIME     604800 //seconds, one week

src/cryptonote_core/blockchain.h

@@ -1460,7 +1460,11 @@ namespace cryptonote
      * @param weights return-by-reference the list of weights
      * @param count the number of blocks to get weights for
      */
+  public:
     void get_last_n_blocks_weights(std::vector<uint64_t>& weights, size_t count) const;
+#ifndef IN_UNIT_TESTS
+  private:
+#endif
 
     /**
      * @brief gets block long term weight median

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"
+  , "Set span limit when syncing, can time (m postfix for minutes), 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"
@@ -168,6 +173,11 @@ namespace cryptonote
   , "How many blocks to sync at once during chain synchronization (0 = adaptive)."
   , 0
   };
+  static const command_line::arg_descriptor<size_t> arg_batch_max_weight  = {
+    "batch-max-weight"
+    , "How many megabytes to sync in one batch during chain synchronization, default is 20 max"
+  , (BATCH_MAX_WEIGHT)
+  };
   static const command_line::arg_descriptor<std::string> arg_check_updates = {
     "check-updates"
   , "Check for new versions of monero: [disabled|notify|download|update]"
@@ -323,11 +333,13 @@ namespace cryptonote
     command_line::add_arg(desc, arg_fast_block_sync);
     command_line::add_arg(desc, arg_show_time_stats);
     command_line::add_arg(desc, arg_block_sync_size);
+    command_line::add_arg(desc, arg_batch_max_weight);
     command_line::add_arg(desc, arg_check_updates);
     command_line::add_arg(desc, arg_test_dbg_lock_sleep);
     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);
@@ -689,6 +701,13 @@ namespace cryptonote
     if (block_sync_size > BLOCKS_SYNCHRONIZING_MAX_COUNT)
       MERROR("Error --block-sync-size cannot be greater than " << BLOCKS_SYNCHRONIZING_MAX_COUNT);
 
+    batch_max_weight = command_line::get_arg(vm, arg_batch_max_weight);
+    if (batch_max_weight > BATCH_MAX_ALLOWED_WEIGHT) {
+      MERROR("Error --batch-max-weight cannot be greater than " << BATCH_MAX_ALLOWED_WEIGHT << " [mB]");
+      batch_max_weight = BATCH_MAX_ALLOWED_WEIGHT;
+    }
+
+    batch_max_weight *= 1000000; // transfer it to byte.
     MGINFO("Loading checkpoints");
 
     // load json & DNS checkpoints, and verify them
@@ -941,17 +960,37 @@ namespace cryptonote
     return true;
   }
   //-----------------------------------------------------------------------------------------------
-  size_t core::get_block_sync_size(uint64_t height) const
+  size_t core::get_block_sync_size(uint64_t height, const uint64_t average_blocksize_of_biggest_batch) const
   {
-    static const uint64_t quick_height = m_nettype == TESTNET ? 2557352 : m_nettype == MAINNET ? 1220516 : 0; // 2557352 is stressnet 251 hardfork height
     size_t res = 0;
     if (block_sync_size > 0)
       res = block_sync_size;
-    else if (height >= quick_height)
-      res = BLOCKS_SYNCHRONIZING_DEFAULT_COUNT;
-    else
-      res = BLOCKS_SYNCHRONIZING_DEFAULT_COUNT_PRE_V4;
-
+    else {
+      size_t number_of_blocks = BLOCKS_MEDIAN_WINDOW;
+      std::vector<uint64_t> last_n_blocks_weights;
+      m_blockchain_storage.get_last_n_blocks_weights(last_n_blocks_weights, number_of_blocks);
+      uint64_t median_weight = epee::misc_utils::median(last_n_blocks_weights);
+      MINFO("Last " << number_of_blocks
+             << " blocks median size is " << median_weight
+             << " bytes and the max average blocksize in the queue is " << average_blocksize_of_biggest_batch << " bytes");
+      uint64_t projected_blocksize = (average_blocksize_of_biggest_batch > median_weight) ? average_blocksize_of_biggest_batch : median_weight;
+      if ((projected_blocksize * BLOCKS_MEDIAN_WINDOW) < batch_max_weight) {
+        res = BLOCKS_MEDIAN_WINDOW;
+        MINFO("blocks are tiny, " << projected_blocksize << " bytes, sync " << res << " blocks in next batch");
+      }
+      else if (projected_blocksize >= batch_max_weight) {
+        res = 1;
+        MINFO("blocks are projected to surpass " << batch_max_weight << " bytes, syncing just a single block in next batch");
+      }
+      else if (projected_blocksize > BLOCKS_HUGE_THRESHOLD_SIZE) {
+        res = 1;
+        MINFO("blocks are huge, sync just a single block in next batch");
+      }
+      else {
+        res = batch_max_weight / projected_blocksize;
+        MINFO("projected blocksize is " << projected_blocksize << " bytes, sync " << res << " blocks in next batch");
+      }
+    }
     static size_t max_block_size = 0;
     if (max_block_size == 0)
     {

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;
 
   /************************************************************************/
@@ -762,9 +763,13 @@ namespace cryptonote
      /**
       * @brief get the number of blocks to sync in one go
       *
+      * @param height the height that we want to get_block_sync_size for
+      * @param average_blocksize_of_biggest_batch is the average blocksize of the biggest batch
+      * we are downloading in current active connections.
+      *
       * @return the number of blocks to sync in one go
       */
-     size_t get_block_sync_size(uint64_t height) const;
+     size_t get_block_sync_size(uint64_t height, const uint64_t average_blocksize_of_biggest_batch = 0) const;
 
      /**
       * @brief get the sum of coinbase tx amounts between blocks
@@ -1066,6 +1071,7 @@ namespace cryptonote
      bool m_disable_dns_checkpoints;
 
      size_t block_sync_size;
+     std::uint64_t batch_max_weight;
 
      time_t start_time;
 

src/cryptonote_protocol/cryptonote_protocol_handler.h

@@ -112,6 +112,15 @@ namespace cryptonote
     void log_connections();
     std::list<connection_info> get_connections();
     const block_queue &get_block_queue() const { return m_block_queue; }
+    const std::uint64_t max_average_of_blocksize_in_queue() {
+      std::vector<std::uint64_t> average_blocksize{0};
+      m_block_queue.foreach([&](const cryptonote::block_queue::span &span) {
+        average_blocksize.push_back(span.size / span.nblocks);
+        return true; // we don't care about the return value
+      });
+      MINFO("Maximum average of blocksize for current batches : " << *std::max_element(average_blocksize.begin(), average_blocksize.end()));
+      return *std::max_element(average_blocksize.begin(), average_blocksize.end());
+    }
     void stop();
     void on_connection_close(cryptonote_connection_context &context);
     void set_max_out_peers(epee::net_utils::zone zone, unsigned int max) { CRITICAL_REGION_LOCAL(m_max_out_peers_lock); m_max_out_peers[zone] = max; }
@@ -167,6 +176,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(double blocks_per_seconds);
 
     t_core& m_core;
 
@@ -191,6 +201,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;
+    size_t m_span_limit;
+    size_t m_bss;
 
     // Values for sync time estimates
     boost::posix_time::ptime m_sync_start_time;

src/cryptonote_protocol/cryptonote_protocol_handler.inl

@@ -62,7 +62,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_THRESHOLD 200 // 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
@@ -157,8 +158,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)
       m_p2p = &m_p2p_stub;
@@ -180,11 +183,22 @@ 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(double blocks_per_seconds)
+  {
+    MINFO("m_bss : " << m_bss << ", blocks_per_seconds : " << blocks_per_seconds << ", m_span_limit : " << m_span_limit);
+    m_span_limit = (m_bss && blocks_per_seconds) ? (( blocks_per_seconds * 60 * m_span_time ) / m_bss) : BLOCK_QUEUE_NSPANS_THRESHOLD;
+    if (m_span_limit < BLOCK_QUEUE_NSPANS_MINIMUM)
+      m_span_limit = BLOCK_QUEUE_NSPANS_MINIMUM;
+    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;
@@ -1559,10 +1573,12 @@ namespace cryptonote
           m_block_queue.remove_spans(span_connection_id, start_height);
 
           const uint64_t current_blockchain_height = m_core.get_current_blockchain_height();
+          const boost::posix_time::time_duration dt = boost::posix_time::microsec_clock::universal_time() - start;
+          double blocks_per_seconds = (((current_blockchain_height - previous_height) * 1e6) / dt.total_microseconds());
+          calculate_dynamic_span(blocks_per_seconds);
           if (current_blockchain_height > previous_height)
           {
             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;
             std::string progress_message = "";
             if (current_blockchain_height < target_blockchain_height)
             {
@@ -1586,7 +1602,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);
@@ -2005,7 +2021,17 @@ skip:
         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;
+        m_span_limit = m_span_limit ? m_span_limit : BLOCK_QUEUE_NSPANS_THRESHOLD;
+        bool queue_proceed = (nspans < m_span_limit) && (size < block_queue_size_threshold);
+        MINFO( "block_queue_size_threshold : " << block_queue_size_threshold
+               << ", queue_proceed : " << queue_proceed
+               << ", size : " << size
+               << ", nspans : " << nspans
+               << ", m_span_limit : " << m_span_limit
+               << ", bc_height : " << bc_height
+               << ", add_stripe : " << add_stripe
+               << ", peer_stripe : " << peer_stripe
+               << ", local_stripe : " << local_stripe);
         // 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)
         {
@@ -2052,7 +2078,6 @@ skip:
           MLOG_PEER_STATE("resuming");
           break;
         }
-
         if (proceed)
         {
           if (context.m_state != cryptonote_connection_context::state_standby)
@@ -2123,7 +2148,7 @@ skip:
       NOTIFY_REQUEST_GET_OBJECTS::request req;
       bool is_next = false;
       size_t count = 0;
-      const size_t count_limit = m_core.get_block_sync_size(m_core.get_current_blockchain_height());
+      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)
       {
@@ -2173,7 +2198,7 @@ skip:
         const uint64_t first_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
         static const uint64_t bp_fork_height = m_core.get_earliest_ideal_height_for_version(8);
         bool sync_pruned_blocks = m_sync_pruned_blocks && first_block_height >= bp_fork_height && m_core.get_blockchain_pruning_seed();
-        span = m_block_queue.reserve_span(first_block_height, context.m_last_response_height, count_limit, context.m_connection_id, context.m_remote_address, sync_pruned_blocks, m_core.get_blockchain_pruning_seed(), context.m_pruning_seed, context.m_remote_blockchain_height, context.m_needed_objects);
+        span = m_block_queue.reserve_span(first_block_height, context.m_last_response_height, m_bss, context.m_connection_id, context.m_remote_address, sync_pruned_blocks, m_core.get_blockchain_pruning_seed(), context.m_pruning_seed, context.m_remote_blockchain_height, context.m_needed_objects);
         MDEBUG(context << " span from " << first_block_height << ": " << span.first << "/" << span.second);
         if (span.second > 0)
         {
@@ -2259,7 +2284,7 @@ skip:
         context.m_expect_height = span.first;
         context.m_expect_response = NOTIFY_RESPONSE_GET_OBJECTS::ID;
         MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size()
-            << "requested blocks count=" << count << " / " << count_limit << " from " << span.first << ", first hash " << req.blocks.front());
+            << "requested blocks count=" << count << " / " << m_bss << " from " << span.first << ", first hash " << req.blocks.front());
         //epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
 
         MDEBUG("Asking for " << (req.prune ? "pruned" : "full") << " data, start/end "

tests/unit_tests/node_server.cpp

@@ -74,7 +74,7 @@ class test_core : public cryptonote::i_core_events
   bool cleanup_handle_incoming_blocks(bool force_sync = false) { return true; }
   bool update_checkpoints(const bool skip_dns = false) { return true; }
   uint64_t get_target_blockchain_height() const { return 1; }
-  size_t get_block_sync_size(uint64_t height) const { return BLOCKS_SYNCHRONIZING_DEFAULT_COUNT; }
+  size_t get_block_sync_size(uint64_t height, const uint64_t average_blocksize_of_biggest_batch = 0) const { return BLOCKS_SYNCHRONIZING_DEFAULT_COUNT; }
   virtual void on_transactions_relayed(epee::span<const cryptonote::blobdata> tx_blobs, cryptonote::relay_method tx_relay) {}
   cryptonote::network_type get_nettype() const { return cryptonote::MAINNET; }
   bool get_pool_transaction(const crypto::hash& id, cryptonote::blobdata& tx_blob, cryptonote::relay_category tx_category) const { return false; }