Cache get_coinbase_tx_sum-from-0 result

This caches the result of a get_coinbase_tx_sum to H-30 (if the last
request started from 0 and retrieved up to at least H-30).  This makes
get_coinbase_tx_sum calls to get the full chain values massively faster
for all but the first call.

The "first call" is kind of tricky, though, because it can take a couple
minutes, during which if we get multiple calls (e.g. from the block
explorer) we might get multiple threads trying to create the cache all
at once, and *each* of those takes minutes (and chew up an admin rpc
thread).  So this commit also blocks out other threads from getting a
cacheable result while the cache is being built; instead those calls get
a null optional back.

Once the cache is built, requests start returning pretty much instantly
(on my desktop system with the blockchain data cached in RAM I process
around 5k blocks per second).

src/cryptonote_core/cryptonote_core.cpp

@@ -1672,20 +1672,88 @@ namespace cryptonote
     return m_mempool.check_for_key_images(key_im, spent);
   }
   //-----------------------------------------------------------------------------------------------
-  std::tuple<uint64_t, uint64_t, uint64_t> core::get_coinbase_tx_sum(const uint64_t start_offset, const size_t count)
+  std::optional<std::tuple<uint64_t, uint64_t, uint64_t>> core::get_coinbase_tx_sum(uint64_t start_offset, size_t count)
   {
-    uint64_t emission_amount = 0;
-    uint64_t total_fee_amount = 0;
-    uint64_t burnt_loki = 0;
-    if (count)
-    {
-      const uint64_t end = start_offset + count - 1;
-      m_blockchain_storage.for_blocks_range(start_offset, end,
-        [this, &emission_amount, &total_fee_amount, &burnt_loki](uint64_t, const crypto::hash& hash, const block& b){
+    std::tuple<uint64_t, uint64_t, uint64_t> result{0, 0, 0};
+    if (count == 0)
+      return result;
+
+    auto& [emission_amount, total_fee_amount, burnt_loki] = result;
+
+    // Caching.
+    //
+    // Requesting this value from the beginning of the chain is very slow, so we cache it.  That
+    // still means the first request will be slow, but that's okay.  To prevent a bunch of threads
+    // getting backed up trying to calculate this, we lock out more than one thread building the
+    // cache at a time if we're requesting a large number of block values at once.  Any other thread
+    // requesting will get a nullopt back.
+
+    constexpr uint64_t CACHE_LAG = 30; // We cache the values up to this many blocks ago; we lag so that we don't have to worry about small reorgs
+    constexpr uint64_t CACHE_EXCLUSIVE = 1000; // If we need to load more than this, we block out other threads
+
+    // Check if we have a cacheable from-the-beginning result
+    uint64_t cache_to = 0;
+    std::chrono::steady_clock::time_point cache_build_started;
+    if (start_offset == 0) {
+      uint64_t height = m_blockchain_storage.get_current_blockchain_height();
+      if (count > height) count = height;
+      cache_to = height - std::min(CACHE_LAG, height);
+      {
+        std::shared_lock lock{m_coinbase_cache.mutex};
+        if (count >= m_coinbase_cache.height) {
+          emission_amount = m_coinbase_cache.emissions;
+          total_fee_amount = m_coinbase_cache.fees;
+          burnt_loki = m_coinbase_cache.burnt;
+          start_offset = m_coinbase_cache.height;
+          count -= m_coinbase_cache.height;
+        }
+        // else don't change anything; we need a subset of blocks that ends before the cache.
+
+        if (cache_to <= m_coinbase_cache.height)
+          cache_to = 0; // Cache doesn't need updating
+      }
+
+      // If we're loading a lot then acquire an exclusive lock, recheck our variables, and block out
+      // other threads until we're done.  (We don't do this if we're only loading a few because even
+      // if we have some competing cache updates they don't hurt anything).
+      if (cache_to > 0 && count > CACHE_EXCLUSIVE) {
+        std::unique_lock lock{m_coinbase_cache.mutex};
+        if (m_coinbase_cache.building)
+          return std::nullopt; // Another thread is already updating the cache
+
+        if (m_coinbase_cache.height > start_offset) {
+          // Someone else updated the cache while we were acquiring the unique lock, so update our variables
+          if (m_coinbase_cache.height >= start_offset + count) {
+            // The cache is now *beyond* us, which means we can't use it, so reset start/count back
+            // to what they were originally.
+            count += start_offset;
+            start_offset = 0;
+            cache_to = 0;
+          } else {
+            // The cache is updated and we can still use it, so update our variables.
+            emission_amount = m_coinbase_cache.emissions;
+            total_fee_amount = m_coinbase_cache.fees;
+            burnt_loki = m_coinbase_cache.burnt;
+            count -= m_coinbase_cache.height - start_offset;
+            start_offset = m_coinbase_cache.height;
+          }
+        }
+        if (cache_to > 0 && count > CACHE_EXCLUSIVE) {
+          cache_build_started = std::chrono::steady_clock::now();
+          m_coinbase_cache.building = true; // Block out other threads until we're done
+          MINFO("Starting slow cache build request for get_coinbase_tx_sum(" << start_offset << ", " << count << ")");
+        }
+      }
+    }
+
+    const uint64_t end = start_offset + count - 1;
+    m_blockchain_storage.for_blocks_range(start_offset, end,
+      [this, &cache_to, &result, &cache_build_started](uint64_t height, const crypto::hash& hash, const block& b){
+      auto& [emission_amount, total_fee_amount, burnt_loki] = result;
       std::vector<transaction> txs;
       std::vector<crypto::hash> missed_txs;
       uint64_t coinbase_amount = get_outs_money_amount(b.miner_tx);
-      this->get_transactions(b.tx_hashes, txs, missed_txs);      
+      get_transactions(b.tx_hashes, txs, missed_txs);
       uint64_t tx_fee_amount = 0;
       for(const auto& tx: txs)
       {
@@ -1695,14 +1763,28 @@ namespace cryptonote
           burnt_loki += get_burned_amount_from_tx_extra(tx.extra);
         }
       }
-      
+
       emission_amount += coinbase_amount - tx_fee_amount;
       total_fee_amount += tx_fee_amount;
+      if (cache_to && cache_to == height)
+      {
+        std::unique_lock lock{m_coinbase_cache.mutex};
+        m_coinbase_cache.height = height;
+        m_coinbase_cache.emissions = emission_amount;
+        m_coinbase_cache.fees = total_fee_amount;
+        m_coinbase_cache.burnt = burnt_loki;
+        if (m_coinbase_cache.building)
+        {
+          m_coinbase_cache.building = false;
+          MINFO("Finishing cache build for get_coinbase_tx_sum in " <<
+              std::chrono::duration<double>{std::chrono::steady_clock::now() - cache_build_started}.count() << "s");
+          cache_to = 0;
+        }
+      }
       return true;
-      });
-    }
+    });
 
-    return std::tuple<uint64_t, uint64_t, uint64_t>(emission_amount, total_fee_amount, burnt_loki);
+    return result;
   }
   //-----------------------------------------------------------------------------------------------
   bool core::check_tx_inputs_keyimages_diff(const transaction& tx) const

src/cryptonote_core/cryptonote_core.h

@@ -795,9 +795,20 @@ namespace cryptonote
      /**
       * @brief get the sum of coinbase tx amounts between blocks
       *
-      * @return the number of blocks to sync in one go
+      * @param start_offset the height to start counting from
+      * @param count the number of blocks to include
+      *
+      * When requesting from the beginning of the chain (i.e. with `start_offset=0` and count >=
+      * current height) the first thread to call this will take a very long time; during this
+      * initial calculation any other threads that attempt to make a similar request will fail
+      * immediately (getting back std::nullopt) until the first thread to calculate it has finished,
+      * after which we use the cached value and only calculate for the last few blocks.
+      *
+      * @return optional tuple of: coin emissions, total fees, and total burned coins in the
+      * requested range.  The optional value will be empty only if requesting the full chain *and*
+      * another thread is already calculating it.
       */
-     std::tuple<uint64_t, uint64_t, uint64_t> get_coinbase_tx_sum(const uint64_t start_offset, const size_t count);
+     std::optional<std::tuple<uint64_t, uint64_t, uint64_t>> get_coinbase_tx_sum(uint64_t start_offset, size_t count);
 
      /**
       * @brief get the network type we're on
@@ -1212,6 +1223,11 @@ namespace cryptonote
 
      std::shared_ptr<tools::Notify> m_block_rate_notify;
 
+     struct {
+       std::shared_mutex mutex;
+       bool building = false;
+       uint64_t height = 0, emissions = 0, fees = 0, burnt = 0;
+     } m_coinbase_cache;
    };
 }
 

src/rpc/core_rpc_server.cpp

@@ -2187,8 +2187,12 @@ namespace cryptonote { namespace rpc {
     GET_COINBASE_TX_SUM::response res{};
 
     PERF_TIMER(on_get_coinbase_tx_sum);
-    std::tie(res.emission_amount, res.fee_amount, res.burn_amount) = m_core.get_coinbase_tx_sum(req.height, req.count);
-    res.status = STATUS_OK;
+    if (auto sums = m_core.get_coinbase_tx_sum(req.height, req.count)) {
+        std::tie(res.emission_amount, res.fee_amount, res.burn_amount) = *sums;
+        res.status = STATUS_OK;
+    } else {
+        res.status = STATUS_BUSY; // some other request is already calculating it
+    }
     return res;
   }
   //------------------------------------------------------------------------------------------------------------------------------