cryptonote: defer tx verification to benefit from batching

src/blockchain_utilities/blockchain_import.cpp

@@ -213,7 +213,7 @@ int check_flush(cryptonote::core &core, std::vector<block_complete_entry> &block
     for(auto& tx_blob: block_entry.txs)
     {
       tx_verification_context tvc = AUTO_VAL_INIT(tvc);
-      core.handle_incoming_tx(tx_blob, tvc, true, true, false);
+      core.handle_incoming_tx(tx_blob, tvc, false, true, true, false);
       if(tvc.m_verifivation_failed)
       {
         MERROR("transaction verification failed, tx_id = "

src/cryptonote_config.h

@@ -174,6 +174,8 @@
 #define CRYPTONOTE_PRUNING_TIP_BLOCKS           5500 // the smaller, the more space saved
 //#define CRYPTONOTE_PRUNING_DEBUG_SPOOF_SEED
 
+#define DEFER_TX_THRESHOLD                      4
+
 // New constants are intended to go here
 namespace config
 {

src/cryptonote_core/blockchain.cpp

@@ -2878,7 +2878,7 @@ bool Blockchain::expand_transaction_2(transaction &tx, const crypto::hash &tx_pr
   }
   else
   {
-    CHECK_AND_ASSERT_MES(false, false, "Unsupported rct tx type: " + boost::lexical_cast<std::string>(rv.type));
+    CHECK_AND_ASSERT_MES(false, false, "Unsupported rct tx type: " << (unsigned)rv.type);
   }
 
   // II
@@ -2906,7 +2906,7 @@ bool Blockchain::expand_transaction_2(transaction &tx, const crypto::hash &tx_pr
   }
   else
   {
-    CHECK_AND_ASSERT_MES(false, false, "Unsupported rct tx type: " + boost::lexical_cast<std::string>(rv.type));
+    CHECK_AND_ASSERT_MES(false, false, "Unsupported rct tx type: " << (unsigned)rv.type);
   }
 
   // outPk was already done by handle_incoming_tx

src/cryptonote_core/cryptonote_core.cpp

@@ -848,19 +848,19 @@ namespace cryptonote
     return true;
   }
   //-----------------------------------------------------------------------------------------------
-  bool core::handle_incoming_tx_accumulated_batch(std::vector<tx_verification_batch_info> &tx_info, bool keeped_by_block)
+  bool core::handle_incoming_tx_accumulated_batch(std::vector<tx_verification_batch_info> &tx_info)
   {
     bool ret = true;
-    if (keeped_by_block && get_blockchain_storage().is_within_compiled_block_hash_area())
-    {
-      MTRACE("Skipping semantics check for tx kept by block in embedded hash area");
-      return true;
-    }
 
     std::vector<const rct::rctSig*> rvv;
     for (size_t n = 0; n < tx_info.size(); ++n)
     {
-      if (!check_tx_semantic(*tx_info[n].tx, keeped_by_block))
+      if (tx_info[n].keeped_by_block && get_blockchain_storage().is_within_compiled_block_hash_area())
+      {
+        MTRACE("Skipping semantics check for tx kept by block in embedded hash area");
+        continue;
+      }
+      if (!check_tx_semantic(*tx_info[n].tx, tx_info[n].keeped_by_block))
       {
         set_semantics_failed(tx_info[n].tx_hash);
         tx_info[n].tvc.m_verifivation_failed = true;
@@ -942,23 +942,58 @@ namespace cryptonote
     return ret;
   }
   //-----------------------------------------------------------------------------------------------
-  bool core::handle_incoming_txs(const std::vector<tx_blob_entry>& tx_blobs, std::vector<tx_verification_context>& tvc, bool keeped_by_block, bool relayed, bool do_not_relay)
+  bool core::handle_incoming_txs(std::vector<tx_blob_entry> tx_blobs, std::vector<tx_verification_context>& tvc, bool allow_defering, bool keeped_by_block, bool relayed, bool do_not_relay)
   {
     TRY_ENTRY();
     CRITICAL_REGION_LOCAL(m_incoming_tx_lock);
 
-    struct result { bool res; cryptonote::transaction tx; crypto::hash hash; };
-    std::vector<result> results(tx_blobs.size());
+    const size_t prior_txs = m_defered_txs.size();
+
+    for (size_t i = 0; i < tx_blobs.size(); ++i)
+    {
+      // slow linear search, but preserves ordering
+      if (std::find_if(m_defered_txs.rbegin(), m_defered_txs.rend(), [&](const defered_tx_t &dtx) { return dtx.blob.blob == tx_blobs[i].blob; }) == m_defered_txs.rend())
+        m_defered_txs.push_back({std::move(tx_blobs[i]), keeped_by_block, relayed, do_not_relay});
+    }
 
+    const bool defer = allow_defering && !keeped_by_block && m_defered_txs.size() < DEFER_TX_THRESHOLD;
+    if (defer)
+    {
+      MDEBUG("defering tx verification");
+      return true;
+    }
+
+    std::vector<tx_verification_context> tvc_all;
+    bool ret = verify_defered_txes(tvc_all);
+    CHECK_AND_ASSERT_MES(tvc_all.size() == tx_blobs.size() + prior_txs, false, "Unexpected number of returned tvc");
     tvc.resize(tx_blobs.size());
+    for (size_t i = 0; i < tvc.size(); ++i)
+      tvc[i] = tvc_all[i + prior_txs];
+    return ret;
+    CATCH_ENTRY_L0("core::handle_incoming_txs()", false);
+  }
+  //-----------------------------------------------------------------------------------------------
+  bool core::verify_defered_txes(std::vector<tx_verification_context> &tvc)
+  {
+    PERF_TIMER(verify_defered_txes);
+    TRY_ENTRY();
+    CRITICAL_REGION_LOCAL(m_incoming_tx_lock);
+
+    if (m_defered_txs.empty())
+      return true;
+
+    struct result { bool res; cryptonote::transaction tx; crypto::hash hash; };
+    std::vector<result> results(m_defered_txs.size());
+
+    tvc.resize(m_defered_txs.size());
     tools::threadpool& tpool = tools::threadpool::getInstance();
     tools::threadpool::waiter waiter;
-    std::vector<tx_blob_entry>::const_iterator it = tx_blobs.begin();
-    for (size_t i = 0; i < tx_blobs.size(); i++, ++it) {
+    std::vector<defered_tx_t>::const_iterator it = m_defered_txs.begin();
+    for (size_t i = 0; i < m_defered_txs.size(); i++, ++it) {
       tpool.submit(&waiter, [&, i, it] {
         try
         {
-          results[i].res = handle_incoming_tx_pre(*it, tvc[i], results[i].tx, results[i].hash, keeped_by_block, relayed, do_not_relay);
+          results[i].res = handle_incoming_tx_pre(it->blob, tvc[i], results[i].tx, results[i].hash, it->keeped_by_block, it->relayed, it->do_not_relay);
         }
         catch (const std::exception &e)
         {
@@ -969,9 +1004,9 @@ namespace cryptonote
       });
     }
     waiter.wait(&tpool);
-    it = tx_blobs.begin();
-    std::vector<bool> already_have(tx_blobs.size(), false);
-    for (size_t i = 0; i < tx_blobs.size(); i++, ++it) {
+    it = m_defered_txs.begin();
+    std::vector<bool> already_have(m_defered_txs.size(), false);
+    for (size_t i = 0; i < m_defered_txs.size(); i++, ++it) {
       if (!results[i].res)
         continue;
       if(m_mempool.have_tx(results[i].hash))
@@ -989,7 +1024,7 @@ namespace cryptonote
         tpool.submit(&waiter, [&, i, it] {
           try
           {
-            results[i].res = handle_incoming_tx_post(*it, tvc[i], results[i].tx, results[i].hash, keeped_by_block, relayed, do_not_relay);
+            results[i].res = handle_incoming_tx_post(it->blob, tvc[i], results[i].tx, results[i].hash, it->keeped_by_block, it->relayed, it->do_not_relay);
           }
           catch (const std::exception &e)
           {
@@ -1003,24 +1038,25 @@ namespace cryptonote
     waiter.wait(&tpool);
 
     std::vector<tx_verification_batch_info> tx_info;
-    tx_info.reserve(tx_blobs.size());
-    for (size_t i = 0; i < tx_blobs.size(); i++) {
+    tx_info.reserve(m_defered_txs.size());
+    it = m_defered_txs.begin();
+    for (size_t i = 0; i < m_defered_txs.size(); i++, ++it) {
       if (!results[i].res || already_have[i])
         continue;
-      tx_info.push_back({&results[i].tx, results[i].hash, tvc[i], results[i].res});
+      tx_info.push_back({&results[i].tx, results[i].hash, it->keeped_by_block, tvc[i], results[i].res});
     }
     if (!tx_info.empty())
-      handle_incoming_tx_accumulated_batch(tx_info, keeped_by_block);
+      handle_incoming_tx_accumulated_batch(tx_info);
 
     bool ok = true;
-    it = tx_blobs.begin();
-    for (size_t i = 0; i < tx_blobs.size(); i++, ++it) {
+    it = m_defered_txs.begin();
+    for (size_t i = 0; i < m_defered_txs.size(); i++, ++it) {
       if (!results[i].res)
       {
         ok = false;
         continue;
       }
-      if (keeped_by_block)
+      if (it->keeped_by_block)
         get_blockchain_storage().on_new_tx_from_block(results[i].tx);
       if (already_have[i])
         continue;
@@ -1029,8 +1065,8 @@ namespace cryptonote
       // different from the actual transaction weight, but it's OK for our use. Those txes
       // will be ignored when mining, and using that "pruned" weight seems appropriate for
       // keeping the txpool size constrained
-      const uint64_t weight = results[i].tx.pruned ? 0 : get_transaction_weight(results[i].tx, it->blob.size());
-      ok &= add_new_tx(results[i].tx, results[i].hash, tx_blobs[i].blob, weight, tvc[i], keeped_by_block, relayed, do_not_relay);
+      const uint64_t weight = results[i].tx.pruned ? 0 : get_transaction_weight(results[i].tx, it->blob.blob.size());
+      ok &= add_new_tx(results[i].tx, results[i].hash, it->blob.blob, weight, tvc[i], it->keeped_by_block, it->relayed, it->do_not_relay);
       if(tvc[i].m_verifivation_failed)
       {MERROR_VER("Transaction verification failed: " << results[i].hash);}
       else if(tvc[i].m_verifivation_impossible)
@@ -1039,24 +1075,47 @@ namespace cryptonote
       if(tvc[i].m_added_to_pool)
         MDEBUG("tx added: " << results[i].hash);
     }
+
+    // relay those that are valid
+    NOTIFY_NEW_TRANSACTIONS::request r;
+    it = m_defered_txs.begin();
+    for (size_t i = 0; i < m_defered_txs.size(); i++, ++it) {
+      if (!results[i].res || it->do_not_relay || it->keeped_by_block || already_have[i])
+        continue;
+      r.txs.push_back(it->blob.blob);
+    }
+    if (!r.txs.empty())
+    {
+      cryptonote_connection_context fake_context = AUTO_VAL_INIT(fake_context);
+      get_protocol()->relay_transactions(r, fake_context);
+    }
+
+    m_defered_txs.clear();
+
     return ok;
 
-    CATCH_ENTRY_L0("core::handle_incoming_txs()", false);
+    CATCH_ENTRY_L0("core::verify_defered_txes()", false);
+  }
+  //-----------------------------------------------------------------------------------------------
+  bool core::verify_defered_txes()
+  {
+    std::vector<tx_verification_context> tvc;
+    return verify_defered_txes(tvc);
   }
   //-----------------------------------------------------------------------------------------------
-  bool core::handle_incoming_tx(const tx_blob_entry& tx_blob, tx_verification_context& tvc, bool keeped_by_block, bool relayed, bool do_not_relay)
+  bool core::handle_incoming_tx(const tx_blob_entry& tx_blob, tx_verification_context& tvc, bool allow_defering, bool keeped_by_block, bool relayed, bool do_not_relay)
   {
     std::vector<tx_blob_entry> tx_blobs;
     tx_blobs.push_back(tx_blob);
     std::vector<tx_verification_context> tvcv(1);
-    bool r = handle_incoming_txs(tx_blobs, tvcv, keeped_by_block, relayed, do_not_relay);
+    bool r = handle_incoming_txs(tx_blobs, tvcv, allow_defering, keeped_by_block, relayed, do_not_relay);
     tvc = tvcv[0];
     return r;
   }
   //-----------------------------------------------------------------------------------------------
-  bool core::handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, bool keeped_by_block, bool relayed, bool do_not_relay)
+  bool core::handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, bool allow_defering, bool keeped_by_block, bool relayed, bool do_not_relay)
   {
-    return handle_incoming_tx({tx_blob, crypto::null_hash}, tvc, keeped_by_block, relayed, do_not_relay);
+    return handle_incoming_tx({tx_blob, crypto::null_hash}, tvc, allow_defering, keeped_by_block, relayed, do_not_relay);
   }
   //-----------------------------------------------------------------------------------------------
   bool core::get_stat_info(core_stat_info& st_inf) const
@@ -1650,6 +1709,7 @@ namespace cryptonote
     m_check_disk_space_interval.do_call(boost::bind(&core::check_disk_space, this));
     m_block_rate_interval.do_call(boost::bind(&core::check_block_rate, this));
     m_blockchain_pruning_interval.do_call(boost::bind(&core::update_blockchain_pruning, this));
+    m_defered_tx_verification_interval.do_call(boost::bind(&core::verify_defered_txes, this));
     m_miner.on_idle();
     m_mempool.on_idle();
     return true;

src/cryptonote_core/cryptonote_core.h

@@ -79,6 +79,14 @@ namespace cryptonote
     */
    class core: public i_miner_handler
    {
+     struct defered_tx_t
+     {
+       cryptonote::tx_blob_entry blob;
+       bool keeped_by_block;
+       bool relayed;
+       bool do_not_relay;
+     };
+
    public:
 
       /**
@@ -115,14 +123,15 @@ namespace cryptonote
       *
       * @param tx_blob the tx to handle
       * @param tvc metadata about the transaction's validity
+      * @param allow_defering whether to allow defering verification
       * @param keeped_by_block if the transaction has been in a block
       * @param relayed whether or not the transaction was relayed to us
       * @param do_not_relay whether to prevent the transaction from being relayed
       *
       * @return true if the transaction was accepted, false otherwise
       */
-     bool handle_incoming_tx(const tx_blob_entry& tx_blob, tx_verification_context& tvc, bool keeped_by_block, bool relayed, bool do_not_relay);
-     bool handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, bool keeped_by_block, bool relayed, bool do_not_relay);
+     bool handle_incoming_tx(const tx_blob_entry& tx_blob, tx_verification_context& tvc, bool allow_defering, bool keeped_by_block, bool relayed, bool do_not_relay);
+     bool handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, bool allow_defering, bool keeped_by_block, bool relayed, bool do_not_relay);
 
      /**
       * @brief handles a list of incoming transactions
@@ -132,13 +141,14 @@ namespace cryptonote
       *
       * @param tx_blobs the txs to handle
       * @param tvc metadata about the transactions' validity
+      * @param allow_defering whether to allow defering verification
       * @param keeped_by_block if the transactions have been in a block
       * @param relayed whether or not the transactions were relayed to us
       * @param do_not_relay whether to prevent the transactions from being relayed
       *
       * @return true if the transactions were accepted, false otherwise
       */
-     bool handle_incoming_txs(const std::vector<tx_blob_entry>& tx_blobs, std::vector<tx_verification_context>& tvc, bool keeped_by_block, bool relayed, bool do_not_relay);
+     bool handle_incoming_txs(std::vector<tx_blob_entry> tx_blobs, std::vector<tx_verification_context>& tvc, bool allow_defering, bool keeped_by_block, bool relayed, bool do_not_relay);
 
      /**
       * @brief handles an incoming block
@@ -926,8 +936,8 @@ namespace cryptonote
 
      bool handle_incoming_tx_pre(const tx_blob_entry& tx_blob, tx_verification_context& tvc, cryptonote::transaction &tx, crypto::hash &tx_hash, bool keeped_by_block, bool relayed, bool do_not_relay);
      bool handle_incoming_tx_post(const tx_blob_entry &tx_blob, tx_verification_context& tvc, cryptonote::transaction &tx, crypto::hash &tx_hash, bool keeped_by_block, bool relayed, bool do_not_relay);
-     struct tx_verification_batch_info { const cryptonote::transaction *tx; crypto::hash tx_hash; tx_verification_context &tvc; bool &result; };
-     bool handle_incoming_tx_accumulated_batch(std::vector<tx_verification_batch_info> &tx_info, bool keeped_by_block);
+     struct tx_verification_batch_info { const cryptonote::transaction *tx; crypto::hash tx_hash; bool keeped_by_block; tx_verification_context &tvc; bool &result; };
+     bool handle_incoming_tx_accumulated_batch(std::vector<tx_verification_batch_info> &tx_info);
 
      /**
       * @copydoc miner::on_block_chain_update
@@ -1015,6 +1025,16 @@ namespace cryptonote
       */
      bool check_block_rate();
 
+     /**
+      * @brief verifies defered txes, and add them to the txpool if valid
+      *
+      * @param tvc return-by-reference flags with verification results
+      *
+      * @return true if all txes are valid, false otherwise
+      */
+     bool verify_defered_txes(std::vector<cryptonote::tx_verification_context> &tvc);
+     bool verify_defered_txes();
+
      bool m_test_drop_download = true; //!< whether or not to drop incoming blocks (for testing)
 
      uint64_t m_test_drop_download_height = 0; //!< height under which to drop incoming blocks, if doing so
@@ -1040,6 +1060,7 @@ namespace cryptonote
      epee::math_helper::once_a_time_seconds<60*10, true> m_check_disk_space_interval; //!< interval for checking for disk space
      epee::math_helper::once_a_time_seconds<90, false> m_block_rate_interval; //!< interval for checking block rate
      epee::math_helper::once_a_time_seconds<60*60*5, true> m_blockchain_pruning_interval; //!< interval for incremental blockchain pruning
+     epee::math_helper::once_a_time_seconds<10, false> m_defered_tx_verification_interval; //!< interval for verifying defered txes
 
      std::atomic<bool> m_starter_message_showed; //!< has the "daemon will sync now" message been shown?
 
@@ -1079,6 +1100,8 @@ namespace cryptonote
      bool m_pad_transactions;
 
      std::shared_ptr<tools::Notify> m_block_rate_notify;
+
+     std::vector<defered_tx_t> m_defered_txs;
    };
 }
 

src/cryptonote_protocol/cryptonote_protocol_defs.h

@@ -120,6 +120,7 @@ namespace cryptonote
   {
     blobdata blob;
     crypto::hash prunable_hash;
+
     BEGIN_KV_SERIALIZE_MAP()
       KV_SERIALIZE(blob)
       KV_SERIALIZE_VAL_POD_AS_BLOB(prunable_hash)