wallet2: speedup refresh a bit

Use the NoodleDoodle threading technique to speedup a couple
code blocks on the main path when refreshing blocks without
any transactions for us.

1 files changed, 148 insertions, 2 deletions

diff --git a/src/wallet/wallet2.cpp b/src/wallet/wallet2.cpp
index f26809af8..005af0a9a 100644
--- a/src/wallet/wallet2.cpp
+++ b/src/wallet/wallet2.cpp
@@ -68,6 +68,13 @@ using namespace cryptonote;
 // arbitrary, used to generate different hashes from the same input
 #define CHACHA8_KEY_TAIL 0x8c
 
+#define KILL_IOSERVICE()  \
+    do { \
+      work.reset(); \
+      threadpool.join_all(); \
+      ioservice.stop(); \
+    } while(0)
+
 namespace
 {
 void do_prepare_file_names(const std::string& file_path, std::string& keys_file, std::string& wallet_file)
@@ -148,6 +155,25 @@ bool wallet2::is_deprecated() const
   return is_old_file_format;
 }
 //----------------------------------------------------------------------------------------------------
+void wallet2::check_acc_out(const account_keys &acc, const tx_out &o, const crypto::public_key &tx_pub_key, size_t i, uint64_t &money_transfered, bool &error) const
+{
+  if (o.target.type() !=  typeid(txout_to_key))
+  {
+     error = true;
+     LOG_ERROR("wrong type id in transaction out");
+     return;
+  }
+  if(is_out_to_acc(acc, boost::get<txout_to_key>(o.target), tx_pub_key, i))
+  {
+    money_transfered = o.amount;
+  }
+  else
+  {
+    money_transfered = 0;
+  }
+  error = false;
+}
+//----------------------------------------------------------------------------------------------------
 void wallet2::process_new_transaction(const cryptonote::transaction& tx, uint64_t height)
 {
   process_unconfirmed(tx, height);
@@ -175,7 +201,48 @@ void wallet2::process_new_transaction(const cryptonote::transaction& tx, uint64_
     }
 
     tx_pub_key = pub_key_field.pub_key;
-    bool r = lookup_acc_outs(m_account.get_keys(), tx, tx_pub_key, outs, tx_money_got_in_outs);
+    bool r = true;
+    int threads = std::thread::hardware_concurrency();
+    if (threads > 1)
+    {
+      boost::asio::io_service ioservice;
+      boost::thread_group threadpool;
+      std::auto_ptr < boost::asio::io_service::work > work(new boost::asio::io_service::work(ioservice));
+      for (int i = 0; i < threads; i++)
+      {
+        threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &ioservice));
+      }
+
+      const account_keys &keys = m_account.get_keys();
+      uint64_t *money_transfered = new uint64_t[tx.vout.size()];
+      bool *error = new bool[tx.vout.size()];
+      for (size_t i = 0; i < tx.vout.size(); ++i)
+      {
+        ioservice.dispatch(boost::bind(&wallet2::check_acc_out, this, std::cref(keys), std::cref(tx.vout[i]), std::cref(tx_pub_key), i,
+          std::ref(money_transfered[i]), std::ref(error[i])));
+      }
+      KILL_IOSERVICE();
+      tx_money_got_in_outs = 0;
+      for (size_t i = 0; i < tx.vout.size(); ++i)
+      {
+        if (error[i])
+        {
+          r = false;
+          break;
+        }
+        if (money_transfered[i])
+        {
+          outs.push_back(i);
+          tx_money_got_in_outs += money_transfered[i];
+        }
+      }
+      delete[] error;
+      delete[] money_transfered;
+    }
+    else
+    {
+      r = lookup_acc_outs(m_account.get_keys(), tx, tx_pub_key, outs, tx_money_got_in_outs);
+    }
     THROW_WALLET_EXCEPTION_IF(!r, error::acc_outs_lookup_error, tx, tx_pub_key, m_account.get_keys());
 
     if(!outs.empty() && tx_money_got_in_outs)
@@ -320,7 +387,7 @@ void wallet2::process_outgoing(const cryptonote::transaction &tx, uint64_t heigh
   ctd.m_block_height = height;
 }
 //----------------------------------------------------------------------------------------------------
-void wallet2::process_new_blockchain_entry(const cryptonote::block& b, cryptonote::block_complete_entry& bche, crypto::hash& bl_id, uint64_t height)
+void wallet2::process_new_blockchain_entry(const cryptonote::block& b, const cryptonote::block_complete_entry& bche, const crypto::hash& bl_id, uint64_t height)
 {
   //handle transactions from new block
     
@@ -379,6 +446,13 @@ void wallet2::get_short_chain_history(std::list<crypto::hash>& ids) const
     ids.push_back(m_blockchain[0]);
 }
 //----------------------------------------------------------------------------------------------------
+void wallet2::parse_block_round(const cryptonote::blobdata &blob, cryptonote::block &bl, crypto::hash &bl_id, bool &error) const
+{
+  error = !cryptonote::parse_and_validate_block_from_blob(blob, bl);
+  if (!error)
+    bl_id = get_block_hash(bl);
+}
+//----------------------------------------------------------------------------------------------------
 void wallet2::pull_blocks(uint64_t start_height, uint64_t& blocks_added)
 {
   blocks_added = 0;
@@ -392,6 +466,77 @@ void wallet2::pull_blocks(uint64_t start_height, uint64_t& blocks_added)
   THROW_WALLET_EXCEPTION_IF(res.status != CORE_RPC_STATUS_OK, error::get_blocks_error, res.status);
 
   size_t current_index = res.start_height;
+
+  int threads = std::thread::hardware_concurrency();
+  if (threads > 1)
+  {
+    crypto::hash *round_block_hashes = new crypto::hash[threads];
+    cryptonote::block *round_blocks = new cryptonote::block[threads];
+    bool *error = new bool[threads];
+    const std::list<block_complete_entry> &blocks = res.blocks;
+    size_t blocks_size = blocks.size();
+    std::list<block_complete_entry>::const_iterator blocki = blocks.begin();
+    for (size_t b = 0; b < blocks_size; b += threads)
+    {
+      size_t round_size = std::min((size_t)threads, blocks_size - b);
+
+      boost::asio::io_service ioservice;
+      boost::thread_group threadpool;
+      std::auto_ptr < boost::asio::io_service::work > work(new boost::asio::io_service::work(ioservice));
+      for (size_t i = 0; i < round_size; i++)
+      {
+        threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &ioservice));
+      }
+
+      std::list<block_complete_entry>::const_iterator tmpblocki = blocki;
+      for (size_t i = 0; i < round_size; ++i)
+      {
+        ioservice.dispatch(boost::bind(&wallet2::parse_block_round, this, std::cref(tmpblocki->block),
+          std::ref(round_blocks[i]), std::ref(round_block_hashes[i]), std::ref(error[i])));
+        ++tmpblocki;
+      }
+      KILL_IOSERVICE();
+      tmpblocki = blocki;
+      for (size_t i = 0; i < round_size; ++i)
+      {
+        THROW_WALLET_EXCEPTION_IF(error[i], error::block_parse_error, tmpblocki->block);
+        ++tmpblocki;
+      }
+      for (size_t i = 0; i < round_size; ++i)
+      {
+        const crypto::hash &bl_id = round_block_hashes[i];
+        cryptonote::block &bl = round_blocks[i];
+
+        if(current_index >= m_blockchain.size())
+        {
+          process_new_blockchain_entry(bl, *blocki, bl_id, current_index);
+          ++blocks_added;
+        }
+        else if(bl_id != m_blockchain[current_index])
+        {
+          //split detected here !!!
+          THROW_WALLET_EXCEPTION_IF(current_index == start_height, error::wallet_internal_error,
+            "wrong daemon response: split starts from the first block in response " + string_tools::pod_to_hex(bl_id) +
+            " (height " + std::to_string(start_height) + "), local block id at this height: " +
+            string_tools::pod_to_hex(m_blockchain[current_index]));
+
+          detach_blockchain(current_index);
+          process_new_blockchain_entry(bl, *blocki, bl_id, current_index);
+        }
+        else
+        {
+          LOG_PRINT_L2("Block is already in blockchain: " << string_tools::pod_to_hex(bl_id));
+        }
+        ++current_index;
+        ++blocki;
+      }
+    }
+    delete[] error;
+    delete[] round_blocks;
+    delete[] round_block_hashes;
+  }
+  else
+  {
   BOOST_FOREACH(auto& bl_entry, res.blocks)
   {
     cryptonote::block bl;
@@ -422,6 +567,7 @@ void wallet2::pull_blocks(uint64_t start_height, uint64_t& blocks_added)
 
     ++current_index;
   }
+  }
 }
 //----------------------------------------------------------------------------------------------------
 void wallet2::refresh()