Address #247 - out-of-order blocks in fork db

- with this change the fork database now caches blocks that do not link
 and attempts to link the unlinked blocks after every new block is
 successfully linked.

 - with this change the fork database only tracks blocks as far back as
 the undo history allows.  This significantly reduces memory usage when
 there is high witness participation.

docs

@@ -1 +1 @@
-Subproject commit f42a917c0cb93784567d649be60d610469c33ee0
+Subproject commit cdc8ea8133a999afef8051700a4ce8edb0988ec4

libraries/chain/CMakeLists.txt

@@ -5,6 +5,7 @@ add_library( graphene_chain
 
              # As database takes the longest to compile, start it first
              database.cpp
+             fork_database.cpp
 
 #             db_balance.cpp
 #             db_block.cpp
@@ -65,7 +66,6 @@ add_library( graphene_chain
              proposal_object.cpp
              vesting_balance_object.cpp
 
-             fork_database.cpp
              block_database.cpp
 
              ${HEADERS}

libraries/chain/db_block.cpp

@@ -97,6 +97,9 @@ bool database::_push_block(const signed_block& new_block)
    uint32_t skip = get_node_properties().skip_flags;
    if( !(skip&skip_fork_db) )
    {
+      /// TODO: if the block is greater than the head block and before the next maitenance interval
+      // verify that the block signer is in the current set of active witnesses.
+
       shared_ptr<fork_item> new_head = _fork_db.push_block(new_block);
       //If the head block from the longest chain does not build off of the current head, we need to switch forks.
       if( new_head->data.previous != head_block_id() )

libraries/chain/db_update.cpp

@@ -73,6 +73,7 @@ void database::update_global_dynamic_data( const signed_block& b )
    }
 
    _undo_db.set_max_size( _dgp.recently_missed_count + GRAPHENE_MIN_UNDO_HISTORY );
+   _fork_db.set_max_size( _dgp.recently_missed_count + GRAPHENE_MIN_UNDO_HISTORY );
 }
 
 void database::update_signing_witness(const witness_object& signing_witness, const signed_block& new_block)

libraries/chain/fork_database.cpp

@@ -16,6 +16,7 @@
  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 #include <graphene/chain/fork_database.hpp>
+#include <graphene/chain/exceptions.hpp>
 #include <graphene/chain/protocol/fee_schedule.hpp>
 #include <fc/smart_ref_impl.hpp>
 
@@ -41,14 +42,36 @@ void     fork_database::start_block(signed_block b)
    _head = item;
 }
 
+/**
+ * Pushes the block into the fork database and caches it if it doesn't link
+ *
+ */
 shared_ptr<fork_item>  fork_database::push_block(const signed_block& b)
 {
    auto item = std::make_shared<fork_item>(b);
-
-   if( _head && b.previous != block_id_type() )
+   try {
+      _push_block(item);
+   } 
+   catch ( const unlinkable_block_exception& e )
    {
-      auto itr = _index.get<block_id>().find(b.previous);
-      FC_ASSERT(itr != _index.get<block_id>().end());
+      wlog( "Pushing block to fork database that failed to link." );
+      _unlinked_index.insert( item );
+   }
+   return _head;
+}
+
+void  fork_database::_push_block(const item_ptr& item)
+{
+   if( _head ) // make sure the block is within the range that we are caching
+   {
+      FC_ASSERT( item->num > std::max<int64_t>( 0, int64_t(_head->num) - (_max_size) ) );
+      FC_ASSERT( item->num < _head->num + 32 );
+   }
+
+   if( _head && item->previous_id() != block_id_type() )
+   {
+      auto itr = _index.get<block_id>().find(item->previous_id());
+      GRAPHENE_ASSERT(itr != _index.get<block_id>().end(), unlinkable_block_exception, "block does not link to known chain");
       FC_ASSERT(!(*itr)->invalid);
       item->prev = *itr;
    }
@@ -58,10 +81,69 @@ shared_ptr<fork_item>  fork_database::push_block(const signed_block& b)
    else if( item->num > _head->num )
    {
       _head = item;
-      _index.get<block_num>().erase(_head->num - 1024);
+      _index.get<block_num>().erase(_head->num - _max_size);
+      _unlinked_index.get<block_num>().erase(_head->num - _max_size);
    }
-   return _head;
+
+   _push_next( item );
 }
+
+/**
+ *  Iterate through the unlinked cache and insert anything that
+ *  links to the newly inserted item.  This will start a recursive
+ *  set of calls performing a depth-first insertion of pending blocks as
+ *  _push_next(..) calls _push_block(...) which will in turn call _push_next
+ */
+void fork_database::_push_next( const item_ptr& new_item )
+{
+    auto& prev_idx = _unlinked_index.get<by_previous>();
+    vector<item_ptr> newly_inserted;
+
+    auto itr = prev_idx.find( new_item->id );
+    while( itr != prev_idx.end() )
+    {
+       auto tmp = *itr;
+       prev_idx.erase( itr );
+       _push_block( tmp ); 
+
+       itr = prev_idx.find( new_item->id );
+    }
+}
+
+void fork_database::set_max_size( uint32_t s )
+{
+   _max_size = s;
+   if( !_head ) return;
+
+   { /// index
+      auto& by_num_idx = _index.get<block_num>();
+      auto itr = by_num_idx.begin();
+      while( itr != by_num_idx.end() )
+      {
+         if( (*itr)->num < std::max(int64_t(0),int64_t(_head->num) - _max_size) )
+            by_num_idx.erase(itr);
+         else
+            break;
+         itr = by_num_idx.begin();
+      }
+   }
+   { /// unlinked_index
+      auto& by_num_idx = _unlinked_index.get<block_num>();
+      auto itr = by_num_idx.begin();
+      while( itr != by_num_idx.end() )
+      {
+         if( (*itr)->num < std::max(int64_t(0),int64_t(_head->num) - _max_size) )
+            by_num_idx.erase(itr);
+         else
+            break;
+         itr = by_num_idx.begin();
+      }
+   }
+}
+
+
+
+
 bool fork_database::is_known_block(const block_id_type& id)const
 {
    auto& index = _index.get<block_id>();

libraries/chain/include/graphene/chain/exceptions.hpp

@@ -69,6 +69,7 @@ namespace graphene { namespace chain {
    FC_DECLARE_DERIVED_EXCEPTION( operation_evaluate_exception,      graphene::chain::chain_exception, 3050000, "operation evaluation exception" )
    FC_DECLARE_DERIVED_EXCEPTION( utility_exception,                 graphene::chain::chain_exception, 3060000, "utility method exception" )
    FC_DECLARE_DERIVED_EXCEPTION( undo_database_exception,           graphene::chain::chain_exception, 3070000, "undo database exception" )
+   FC_DECLARE_DERIVED_EXCEPTION( unlinkable_block_exception,          graphene::chain::chain_exception, 3080000, "unlinkable block" )
 
    FC_DECLARE_DERIVED_EXCEPTION( tx_missing_active_auth,            graphene::chain::transaction_exception, 3030001, "missing required active authority" )
    FC_DECLARE_DERIVED_EXCEPTION( tx_missing_owner_auth,             graphene::chain::transaction_exception, 3030002, "missing required owner authority" )

libraries/chain/include/graphene/chain/fork_database.hpp

@@ -22,6 +22,8 @@
 #include <boost/multi_index/member.hpp>
 #include <boost/multi_index/ordered_index.hpp>
 #include <boost/multi_index/hashed_index.hpp>
+#include <boost/multi_index/mem_fun.hpp>
+
 
 namespace graphene { namespace chain {
    using boost::multi_index_container;
@@ -32,6 +34,8 @@ namespace graphene { namespace chain {
       fork_item( signed_block d )
       :num(d.block_num()),id(d.id()),data( std::move(d) ){}
 
+      block_id_type previous_id()const { return data.previous; }
+
       weak_ptr< fork_item > prev;
       uint32_t              num;
       /**
@@ -44,6 +48,7 @@ namespace graphene { namespace chain {
    };
    typedef shared_ptr<fork_item> item_ptr;
 
+
    /**
     *  As long as blocks are pushed in order the fork
     *  database will maintain a linked tree of all blocks
@@ -68,6 +73,10 @@ namespace graphene { namespace chain {
          bool                             is_known_block(const block_id_type& id)const;
          shared_ptr<fork_item>            fetch_block(const block_id_type& id)const;
          vector<item_ptr>                 fetch_block_by_number(uint32_t n)const;
+
+         /**
+          *  @return the new head block ( the longest fork )
+          */
          shared_ptr<fork_item>            push_block(const signed_block& b);
          shared_ptr<fork_item>            head()const { return _head; }
          void                             pop_block();
@@ -81,15 +90,26 @@ namespace graphene { namespace chain {
 
          struct block_id;
          struct block_num;
+         struct by_previous;
          typedef multi_index_container<
             item_ptr,
             indexed_by<
                hashed_unique<tag<block_id>, member<fork_item, block_id_type, &fork_item::id>, std::hash<fc::ripemd160>>,
+               hashed_non_unique<tag<by_previous>, const_mem_fun<fork_item, block_id_type, &fork_item::previous_id>, std::hash<fc::ripemd160>>,
                ordered_non_unique<tag<block_num>, member<fork_item,uint32_t,&fork_item::num>>
             >
          > fork_multi_index_type;
 
+         void set_max_size( uint32_t s );
+
       private:
+         /** @return a pointer to the newly pushed item */
+         void _push_block(const item_ptr& b );
+         void _push_next(const item_ptr& newly_inserted);
+
+         uint32_t                 _max_size = 1024;
+
+         fork_multi_index_type    _unlinked_index;
          fork_multi_index_type    _index;
          shared_ptr<fork_item>    _head;
    };