Experimental BDB workaround optimizations

src/blockchain_db/berkeleydb/db_bdb.cpp

@@ -105,6 +105,8 @@ const char* const BDB_OUTPUT_KEYS = "output_keys";
 
 const char* const BDB_SPENT_KEYS = "spent_keys";
 
+const int BUFFER_LENGTH	= 32 * 1024 * 1024;
+
 template<typename T>
 struct Dbt_copy: public Dbt
 {
@@ -575,7 +577,9 @@ void BlockchainBDB::check_open() const
 BlockchainBDB::~BlockchainBDB()
 {
   LOG_PRINT_L3("BlockchainBDB::" << __func__);
-
+    if(m_buffer != NULL)
+        free(m_buffer);
+    m_buffer = NULL;
   if (m_open)
   {
     close();
@@ -584,6 +588,7 @@ BlockchainBDB::~BlockchainBDB()
 
 BlockchainBDB::BlockchainBDB(bool batch_transactions)
 {
+    m_buffer = malloc(BUFFER_LENGTH);
   LOG_PRINT_L3("BlockchainBDB::" << __func__);
   // initialize folder to something "safe" just in case
   // someone accidentally misuses this class...
@@ -1389,7 +1394,9 @@ tx_out BlockchainBDB::get_output(const uint64_t& index) const
   return tx_out();
 }
 
-tx_out_index BlockchainBDB::get_output_tx_and_index_from_global(const uint64_t& index) const
+void BlockchainBDB::get_output_tx_and_index(const uint64_t& amount,
+        std::vector<uint64_t> &offsets,
+        std::vector<tx_out_index> &indices) const
 {
   LOG_PRINT_L3("BlockchainBDB::" << __func__);
   check_open();
@@ -1398,27 +1405,181 @@ tx_out_index BlockchainBDB::get_output_tx_and_index_from_global(const uint64_t&
   if (m_env->txn_begin(NULL, txn, 0))
     throw0(DB_ERROR("Failed to create a transaction for the db"));
 
-  Dbt_copy<uint32_t> k(index);
+    bdb_cur cur(txn, m_output_amounts);
-  Dbt_copy<crypto::hash > v;
+    uint64_t max = 0;
+    for(const uint64_t& index : offsets)
+    {
+        if(index > max)
+            max = index;
+    }
 
-  auto get_result = m_output_txs->get(txn, &k, &v, 0);
+	// ??? might be a bug, don't always treat as uint64_t
-  if (get_result == DB_NOTFOUND)
+	#define DBT_VALUE(dbt) v.get_size() == sizeof(uint64_t) ? \
-    throw1(OUTPUT_DNE("output with given index not in db"));
+		*((uint64_t *)v.get_data()) : *((uint32_t *)v.get_data()) \
-  else if (get_result)
-    throw0(DB_ERROR("DB error attempting to fetch output tx hash"));
 
-  crypto::hash tx_hash = v;
+	// get returned keypairs count
+	#define	DB_COUNT_RECORDS(dbt, cnt) \
+	do { \
+		uint32_t *_p = (uint32_t *) ((uint8_t *)(dbt)->data + \
+		(dbt)->ulen - sizeof(uint32_t)); \
+		cnt = 0; \
+		while(*_p != (uint32_t) -1) { \
+		_p -= 2; \
+		++cnt; \
+		} \
+	} while(0); \
 
-  Dbt_copy<uint64_t> result;
+	Dbt_copy<uint64_t> k(amount);
-  get_result = m_output_indices->get(txn, &k, &result, 0);
+    Dbt_copy<uint64_t> v;
-  if (get_result == DB_NOTFOUND)
+    uint64_t buflen = 0;
-    throw1(OUTPUT_DNE("output with given index not in db"));
+	uint64_t t_dbmul = 0;
-  else if (get_result)
+	uint64_t t_dbscan = 0;
-    throw0(DB_ERROR("DB error attempting to fetch output tx index"));
+    TIME_MEASURE_START(db2);
+    if(max <= 1)
+    {
+    	for (const uint64_t& index : offsets)
+        {
+            TIME_MEASURE_START(t_seek);
 
+            auto result = cur->get(&k, &v, DB_SET);
+            if (result == DB_NOTFOUND)
+                throw1(OUTPUT_DNE("Attempting to get an output index by amount and amount index, but amount not found"));
+            else if (result)
+                throw0(DB_ERROR("DB error attempting to get an output"));
+
+            db_recno_t num_elems = 0;
+            cur->count(&num_elems, 0);
+
+            if (num_elems <= index)
+                throw1(OUTPUT_DNE("Attempting to get an output index by amount and amount index, but output not found"));
+
+            for (uint64_t i = 0; i < index; ++i)
+            {
+                cur->get(&k, &v, DB_NEXT_DUP);
+            }
+
+            uint64_t glob_index = DBT_VALUE(v);
+
+            LOG_PRINT_L1("L0->v: " << glob_index);
+            tx_out_index res =  get_output_tx_and_index_from_global(glob_index);
+            indices.push_back(res);
+
+            TIME_MEASURE_FINISH(t_seek);
+        }
+    }
+    else
+    {
+		// setup a 256KB minimum buffer size
+		uint32_t pagesize = 256 * 1024;
+
+		// Retrieve only a suitable portion of the kvp data, up to somewhere near
+		// the maximum offset value being retrieved
+		buflen = (max + 1) * 4 * sizeof(uint64_t);
+		buflen = ((buflen / pagesize) + ((buflen % pagesize) > 0 ? 1 : 0)) * pagesize;
+		bool singlebuff = buflen <= BUFFER_LENGTH;
+		buflen = buflen < BUFFER_LENGTH ? buflen : BUFFER_LENGTH;
+
+		Dbt data;
+		data.set_data(m_buffer);
+		data.set_ulen(buflen);
+		data.set_size(buflen);
+		data.set_flags(DB_DBT_USERMEM);
+
+		uint32_t curcount = 0;
+    	uint32_t blockstart = 0;
+        for (const uint64_t& index : offsets)
+        {
+    		// fixme! for whatever reason, the first call to DB_MULTIPLE | DB_SET does not
+			// retrieve the first value.
+            if(index <= 1)
+            {
+                auto result = cur->get(&k, &v, DB_SET);
+                if (result == DB_NOTFOUND)
+                    throw1(OUTPUT_DNE("Attempting to get an output index by amount and amount index, but amount not found"));
+                else if (result)
+                    throw0(DB_ERROR("DB error attempting to get an output"));
+
+                db_recno_t num_elems = 0;
+                cur->count(&num_elems, 0);
+
+                if (num_elems <= index)
+                    throw1(OUTPUT_DNE("Attempting to get an output index by amount and amount index, but output not found"));
+                if(index > 0)
+                    cur->get(&k, &v, DB_NEXT_DUP);
+            }
+            else
+            {
+				while(index >= curcount)
+				{
+					TIME_MEASURE_START(t_db1);
+					try
+					{
+						cur->get(&k, &data, DB_MULTIPLE | (curcount == 0 ? DB_SET : DB_NEXT_DUP));
+						blockstart = curcount;
+						// skip counting if using single buffer, it actually adds some overhead on some systems.
+						if(!singlebuff)
+						{
+							int count = 0;
+							DB_COUNT_RECORDS((DBT *) &data, count);
+							curcount += count;
+						}
+					}
+					catch (const std::exception &e)
+					{
+						LOG_PRINT_L0("DB_EXCEPTION: " << e.what());
+					}
+
+					TIME_MEASURE_FINISH(t_db1);
+					t_dbmul += t_db1;
+					if(singlebuff)
+						break;
+				}
+
+				LOG_PRINT_L1("Records returned: " << curcount << " Index: " << index);
+				TIME_MEASURE_START(t_db2);
+				DBT *pdata = (DBT *) &data;
+
+                uint8_t *value;
+                uint64_t dlen = 0;
+
+				void *pbase = ((uint8_t *)(pdata->data)) + pdata->ulen - sizeof(uint32_t);
+                uint32_t *p = (uint32_t *) pbase;
+                if (*p == (uint32_t) -1)
+                {
+                    value = NULL;
+                }
+                else
+                {
+                    p -= (index - blockstart) * 2; // index * 4 + 2; <- if DB_MULTIPLE_KEY
+                    value = (uint8_t *) pdata->data + *p--;
+                    dlen = *p--;
+                    if (value == (uint8_t *) pdata->data)
+                        value = NULL;
+                }
+
+                if (value != NULL)
+                {
+                    v = dlen == sizeof(uint64_t) ?  *((uint64_t *) value)
+                    		: *((uint32_t *) value);
+                }
+	            TIME_MEASURE_FINISH(t_db2);
+				t_dbscan += t_db2;
+			}
+
+            uint64_t glob_index = DBT_VALUE(v);
+
+            LOG_PRINT_L1("L1->v: " << glob_index);
+            tx_out_index res =  get_output_tx_and_index_from_global(glob_index);
+            indices.push_back(res);
+        }
+    }
+
+    TIME_MEASURE_FINISH(db2);
+
+    LOG_PRINT_L1("blen: " << buflen  << " db1: " << t_dbmul << " db2: " << t_dbscan);
+
+    cur.close();
   txn.commit();
-
-  return tx_out_index(tx_hash, result);
 }
 
 tx_out_index BlockchainBDB::get_output_tx_and_index(const uint64_t& amount, const uint64_t& index) const
@@ -1571,6 +1732,37 @@ std::vector<uint64_t> BlockchainBDB::get_tx_amount_output_indices(const crypto::
 }
 
 
+tx_out_index BlockchainBDB::get_output_tx_and_index_from_global(const uint64_t& index) const
+{
+    LOG_PRINT_L3("BlockchainBDB::" << __func__);
+    check_open();
+
+    bdb_txn_safe txn;
+    if (m_env->txn_begin(NULL, txn, 0))
+        throw0(DB_ERROR("Failed to create a transaction for the db"));
+
+    Dbt_copy<uint32_t> k(index);
+    Dbt_copy<crypto::hash > v;
+
+    auto get_result = m_output_txs->get(txn, &k, &v, 0);
+    if (get_result == DB_NOTFOUND)
+        throw1(OUTPUT_DNE("output with given index not in db"));
+    else if (get_result)
+        throw0(DB_ERROR("DB error attempting to fetch output tx hash"));
+
+    crypto::hash tx_hash = v;
+
+    Dbt_copy<uint64_t> result;
+    get_result = m_output_indices->get(txn, &k, &result, 0);
+    if (get_result == DB_NOTFOUND)
+        throw1(OUTPUT_DNE("output with given index not in db"));
+    else if (get_result)
+        throw0(DB_ERROR("DB error attempting to fetch output tx index"));
+
+    txn.commit();
+
+    return tx_out_index(tx_hash, result);
+}
 
 bool BlockchainBDB::has_key_image(const crypto::key_image& img) const
 {

src/blockchain_db/berkeleydb/db_bdb.h

@@ -177,6 +177,7 @@ public:
   virtual tx_out_index get_output_tx_and_index_from_global(const uint64_t& index) const;
 
   virtual tx_out_index get_output_tx_and_index(const uint64_t& amount, const uint64_t& index) const;
+  virtual void get_output_tx_and_index(const uint64_t& amount, std::vector<uint64_t> &offsets, std::vector<tx_out_index> &indices) const;
 
   virtual std::vector<uint64_t> get_tx_output_indices(const crypto::hash& h) const;
   virtual std::vector<uint64_t> get_tx_amount_output_indices(const crypto::hash& h) const;
@@ -197,6 +198,7 @@ public:
   virtual void batch_abort();
 
   virtual void pop_block(block& blk, std::vector<transaction>& txs);
+  virtual bool has_bulk_indices() const { return true; }
 
 private:
   virtual void add_block( const block& blk
@@ -254,6 +256,7 @@ private:
   uint64_t get_output_global_index(const uint64_t& amount, const uint64_t& index) const;
 
   void check_open() const;
+  void *m_buffer;
 
   DbEnv* m_env;
 

src/blockchain_db/blockchain_db.h

@@ -472,6 +472,8 @@ public:
   // returns the transaction-local reference for the output with <amount> at <index>
   // return type is pair of tx hash and index
   virtual tx_out_index get_output_tx_and_index(const uint64_t& amount, const uint64_t& index) const = 0;
+  virtual void get_output_tx_and_index(const uint64_t& amount, std::vector<uint64_t> &offsets, std::vector<tx_out_index> &indices) const = 0;
+  virtual bool  has_bulk_indices() const = 0;
 
   // return a vector of indices corresponding to the global output index for
   // each output in the transaction with hash <h>

src/blockchain_db/lmdb/db_lmdb.h

@@ -177,6 +177,10 @@ public:
   virtual tx_out_index get_output_tx_and_index_from_global(const uint64_t& index) const;
 
   virtual tx_out_index get_output_tx_and_index(const uint64_t& amount, const uint64_t& index) const;
+  virtual void get_output_tx_and_index(const uint64_t& amount, std::vector<uint64_t> &offsets, std::vector<tx_out_index> &indices) const
+  {
+	  // do nothing
+  };
 
   virtual std::vector<uint64_t> get_tx_output_indices(const crypto::hash& h) const;
   virtual std::vector<uint64_t> get_tx_amount_output_indices(const crypto::hash& h) const;
@@ -198,6 +202,7 @@ public:
 
   virtual void pop_block(block& blk, std::vector<transaction>& txs);
 
+  virtual bool has_bulk_indices() const { return false; }
 private:
   void do_resize(uint64_t size_increase=0);
 

src/cryptonote_core/blockchain.cpp

@@ -39,6 +39,7 @@
 #include "tx_pool.h"
 #include "blockchain.h"
 #include "blockchain_db/blockchain_db.h"
+#include "blockchain_db/berkeleydb/db_bdb.h"
 #include "cryptonote_format_utils.h"
 #include "cryptonote_boost_serialization.h"
 #include "cryptonote_config.h"
@@ -159,7 +160,13 @@ bool Blockchain::scan_outputkeys_for_indexes(const txin_to_key& tx_in_to_key, vi
   // #1 plus relative offset #2.
   // TODO: Investigate if this is necessary / why this is done.
   std::vector<uint64_t> absolute_offsets = relative_output_offsets_to_absolute(tx_in_to_key.key_offsets);
+  std::vector<tx_out_index> indices;
 
+  // if(typeid(*m_db) == typeid(BlockchainBDB))
+  if(m_db->has_bulk_indices())
+  {
+	  m_db->get_output_tx_and_index(tx_in_to_key.amount, absolute_offsets, indices);
+  }
 
   //std::vector<std::pair<crypto::hash, size_t> >& amount_outs_vec = it->second;
   size_t count = 0;
@@ -168,7 +175,12 @@ bool Blockchain::scan_outputkeys_for_indexes(const txin_to_key& tx_in_to_key, vi
     try
     {
       // get tx hash and output index for output
-      auto output_index = m_db->get_output_tx_and_index(tx_in_to_key.amount, i);
+      tx_out_index output_index;
+
+      if(indices.size() == absolute_offsets.size())
+    	  output_index = indices.at(count);
+      else
+    	  output_index = m_db->get_output_tx_and_index(tx_in_to_key.amount, i);
 
       // get tx that output is from
       auto tx = m_db->get_tx(output_index.first);
@@ -2103,6 +2115,7 @@ bool Blockchain::handle_block_to_main_chain(const block& bl, const crypto::hash&
   key_images_container keys;
 
   uint64_t fee_summary = 0;
+  uint64_t t_checktx = 0;
 
   // Iterate over the block's transaction hashes, grabbing each
   // from the tx_pool and validating them.  Each is then added
@@ -2133,6 +2146,7 @@ bool Blockchain::handle_block_to_main_chain(const block& bl, const crypto::hash&
     // taken from the tx_pool back to it if the block fails verification.
     txs.push_back(tx);
 
+    TIME_MEASURE_START(aa);
     // validate that transaction inputs and the keys spending them are correct.
     if(!check_tx_inputs(tx))
     {
@@ -2144,6 +2158,8 @@ bool Blockchain::handle_block_to_main_chain(const block& bl, const crypto::hash&
       bvc.m_verifivation_failed = true;
       break;
     }
+    TIME_MEASURE_FINISH(aa);
+    t_checktx += aa;
 
     if (!check_for_double_spend(tx, keys))
     {
@@ -2220,6 +2236,9 @@ bool Blockchain::handle_block_to_main_chain(const block& bl, const crypto::hash&
     << "), coinbase_blob_size: " << coinbase_blob_size << ", cumulative size: " << cumulative_block_size
     << ", " << block_processing_time << "("<< target_calculating_time << "/" << longhash_calculating_time << ")ms");
 
+  LOG_PRINT_L0("Height: " << new_height << " blob: " << coinbase_blob_size <<
+		  " cumm: " << cumulative_block_size << " p/t: " << block_processing_time
+		  << " ("<< target_calculating_time << "/" << longhash_calculating_time << "/" << t_checktx << ")ms");
   bvc.m_added_to_main_chain = true;
 
   // appears to be a NOP *and* is called elsewhere.  wat?