cryptonote_protocol: limit queued blocks dynamically

### src/cryptonote_core/cryptonote_core.cpp
@@ -124,9 +124,9 @@ 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)"
+  const command_line::arg_descriptor<size_t> arg_block_sync_queue_time  = {
+    "block-sync-queue-time"
+  , "Target duration of block synchronization data to keep queued, in minutes (default is 2 minutes)"
   , 2
   };
   const command_line::arg_descriptor<bool> arg_sync_pruned_blocks  = {
@@ -326,7 +326,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_block_sync_queue_time);
     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,7 +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<size_t> arg_block_sync_queue_time;
   extern const command_line::arg_descriptor<bool> arg_sync_pruned_blocks;
 
   /************************************************************************/

### src/cryptonote_protocol/block_queue.cpp
@@ -439,6 +439,18 @@ size_t block_queue::get_num_filled_spans() const
   return size;
 }
 
+uint64_t block_queue::get_num_filled_blocks() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  uint64_t size = 0;
+  for (const auto &span: blocks)
+  {
+    if (!span.blocks.empty())
+      size += span.nblocks;
+  }
+  return size;
+}
+
 float block_queue::get_speed(const boost::uuids::uuid &connection_id) const
 {
   boost::unique_lock<boost::recursive_mutex> lock(mutex);

### src/cryptonote_protocol/block_queue.h
@@ -89,6 +89,7 @@ namespace cryptonote
     bool has_next_span(uint64_t height, bool &filled, boost::posix_time::ptime &time, boost::uuids::uuid &connection_id) const;
     size_t get_data_size() const;
     size_t get_num_filled_spans() const;
+    uint64_t get_num_filled_blocks() const;
     float get_speed(const boost::uuids::uuid &connection_id) const;
     bool foreach(std::function<bool(const span&)> f) const;
     bool requested(const crypto::hash &hash) const;

### src/cryptonote_protocol/cryptonote_protocol_handler.h
@@ -178,7 +178,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);
+    void calculate_block_queue_limit(double blocks_per_second);
 
     t_core& m_core;
 
@@ -204,9 +204,8 @@ 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;
+    size_t m_block_sync_queue_time;
+    std::atomic<uint64_t> m_block_queue_limit;
 
     // Values for sync time estimates
     boost::posix_time::ptime m_sync_start_time;

### src/cryptonote_protocol/cryptonote_protocol_handler.inl
@@ -38,7 +38,9 @@
 #include <boost/optional/optional.hpp>
 #include <boost/uuid/uuid.hpp>
 #include <boost/uuid/uuid_generators.hpp>
+#include <cmath>
 #include <list>
+#include <limits>
 #include <ctime>
 
 #include <cryptonote_core/cryptonote_core.h>
@@ -182,9 +184,8 @@ namespace cryptonote
                                                                                                               m_ask_for_txpool_complement(true),
                                                                                                               m_stopping(false),
                                                                                                               m_no_sync(false),
-                                                                                                              m_span_limit(BLOCK_QUEUE_NSPANS_MINIMUM),
-                                                                                                              m_span_time(0),
-                                                                                                              m_bss(0)
+                                                                                                              m_block_sync_queue_time(0),
+                                                                                                              m_block_queue_limit(0)
 
   {
     if(!m_p2p)
@@ -207,22 +208,29 @@ 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);
+    m_block_sync_queue_time = command_line::get_arg(vm, cryptonote::arg_block_sync_queue_time);
 
     return true;
   }
   //------------------------------------------------------------------------------------------------------------------------
   template<class t_core>
-  void t_cryptonote_protocol_handler<t_core>::calculate_dynamic_span(const double blocks_per_seconds)
+  void t_cryptonote_protocol_handler<t_core>::calculate_block_queue_limit(double blocks_per_second)
   {
-    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);
+    if (!(blocks_per_second > 0.0) || !std::isfinite(blocks_per_second))
+    {
+      MWARNING("Not updating block queue limit from invalid sync rate " << blocks_per_second);
+      return;
+    }
+
+    // Multiply the duration first so a zero target stays zero even for very large rates.
+    const long double requested_blocks = blocks_per_second * (60.0L * m_block_sync_queue_time);
+    const long double max_block_queue_limit = static_cast<long double>(std::numeric_limits<uint64_t>::max());
+    const uint64_t block_queue_limit = requested_blocks >= max_block_queue_limit
+      ? std::numeric_limits<uint64_t>::max()
+      : static_cast<uint64_t>(requested_blocks);
+    m_block_queue_limit = block_queue_limit;
+    MINFO("Calculated dynamic block queue limit: " << block_queue_limit << " blocks at "
+        << blocks_per_second << " blocks per second");
   }
   //------------------------------------------------------------------------------------------------------------------------
   template<class t_core>
@@ -1607,8 +1615,11 @@ 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);
+            const int64_t elapsed_us = dt.total_microseconds();
+            const double blocks_per_seconds = elapsed_us > 0
+              ? ((current_blockchain_height - previous_height) * 1e6) / elapsed_us
+              : 0.0;
+            calculate_block_queue_limit(blocks_per_seconds);
             std::string progress_message = "";
             if (current_blockchain_height < target_blockchain_height)
             {
@@ -2055,14 +2066,16 @@ skip:
         boost::unique_lock<boost::mutex> check_span_lock{m_check_span_queue_mutex};
 
         const size_t nspans = m_block_queue.get_num_filled_spans();
+        const uint64_t nblocks = m_block_queue.get_num_filled_blocks();
         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;
-        const bool queue_proceed_init = (nspans < m_span_limit.load()) && (size < block_queue_size_threshold);
+        const bool queue_proceed_init = (nspans < BLOCK_QUEUE_NSPANS_MINIMUM || nblocks < m_block_queue_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)
         {
@@ -2108,7 +2121,8 @@ skip:
                << ", 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
+               << ", nspans/minimum : " << nspans << "/" << BLOCK_QUEUE_NSPANS_MINIMUM
+               << ", nblocks/block_queue_limit : " << nblocks << "/" << m_block_queue_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,
@@ -2192,7 +2206,7 @@ skip:
       NOTIFY_REQUEST_GET_OBJECTS::request req;
       bool is_next = false;
       size_t count = 0;
-      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());
+      const size_t l_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)
       {

### tests/unit_tests/block_queue.cpp
@@ -164,3 +164,16 @@ TEST(block_queue, reserve_span_skips_requested_prefix)
   ASSERT_TRUE(bq.requested(hashes[2].first));
   ASSERT_TRUE(bq.requested(hashes[3].first));
 }
+
+TEST(block_queue, count_filled_blocks)
+{
+  cryptonote::block_queue bq;
+  epee::net_utils::network_address na;
+
+  bq.add_blocks(0, std::vector<cryptonote::block_complete_entry>(3), uuid1(), na, 0.0f, 0);
+  bq.add_blocks(3, 2, uuid2(), na);
+  bq.add_blocks(5, std::vector<cryptonote::block_complete_entry>(4), uuid2(), na, 0.0f, 0);
+
+  ASSERT_EQ(bq.get_num_filled_spans(), 2);
+  ASSERT_EQ(bq.get_num_filled_blocks(), 7);
+}