Added boost lockfree queue and consum thread #172

plugins/db_plugin/db_plugin.cpp

@@ -15,6 +15,8 @@
 #include <boost/range/adaptors.hpp>
 #include <boost/range/algorithm.hpp>
 #include <boost/range/algorithm_ext.hpp>
+#include <boost/thread/thread.hpp>
+#include <boost/lockfree/spsc_queue.hpp>
 
 namespace fc { class variant; }
 
@@ -32,19 +34,15 @@ using namespace boost::multi_index;
 class db_plugin_impl {
 public:
    void applied_irreversible_block(const signed_block&);
+   void process_irreversible_block(const signed_block&);
 
-   chain_plugin* chain_plug;
    std::set<AccountName> filter_on;
+   std::unique_ptr<boost::lockfree::spsc_queue<signed_block>> queue;
+   boost::thread consum_thread;
+   boost::atomic<bool> startup{true};
+   boost::atomic<bool> done{false};
 
-private:
-   struct block_comp
-   {
-      bool operator()(const block_id_type& a, const block_id_type& b) const
-      {
-         return chain::block_header::num_from_id(a) > chain::block_header::num_from_id(b);
-      }
-   };
-   typedef std::multimap<block_id_type, transaction_id_type, block_comp> block_transaction_id_map;
+   void consum_blocks();
 
    bool is_scope_relevant(const eos::types::Vector<AccountName>& scope);
    static void add(chainbase::database& db, const vector<types::KeyPermissionWeight>& keys, const AccountName& account_name, const PermissionName& permission);
@@ -77,7 +75,32 @@ const PermissionName db_plugin_impl::OWNER = "owner";
 const PermissionName db_plugin_impl::ACTIVE = "active";
 const PermissionName db_plugin_impl::RECOVERY = "recovery";
 
-void db_plugin_impl::applied_irreversible_block(const signed_block& block)
+void db_plugin_impl::applied_irreversible_block(const signed_block& block) {
+   if (startup) {
+      // on startup we don't want to queue, insteas push back on caller
+      process_irreversible_block(block);
+   } else {
+      if (!queue->push(block)) {
+         // TODO what to do if full
+         elog("queue is full!!!!!");
+      }
+   }
+}
+
+void db_plugin_impl::consum_blocks() {
+   signed_block block;
+   while (!done) {
+      while (queue->pop(block)) {
+         process_irreversible_block(block);
+      }
+   }
+   while (queue->pop(block)) {
+      process_irreversible_block(block);
+   }
+   ilog("db_plugin consum thread shutdown gracefully");
+}
+
+void db_plugin_impl::process_irreversible_block(const signed_block& block)
 {
    const auto block_id = block.id();
    ilog("block ${bid}", ("bid", block_id));
@@ -193,6 +216,8 @@ void db_plugin::set_program_options(options_description& cli, options_descriptio
    cfg.add_options()
          ("filter_on_accounts,f", bpo::value<vector<string>>()->composing(),
           "Track only transactions whose scopes involve the listed accounts. Default is to track all transactions.")
+         ("queue_size,q", bpo::value<uint>()->default_value(1024),
+         "The block queue size")
          ;
 }
 
@@ -215,6 +240,9 @@ void db_plugin::plugin_initialize(const variables_map& options)
       auto foa = options.at("filter_on_accounts").as<vector<string>>();
       for(auto filter_account : foa)
          my->filter_on.emplace(filter_account);
+   } else if (options.count("queue_size")) {
+      auto size = options.at("queue_size").as<uint>();
+      my->queue = std::make_unique<boost::lockfree::spsc_queue<signed_block>>(size);
    }
 }
 
@@ -223,17 +251,23 @@ void db_plugin::plugin_startup()
    ilog("starting db plugin");
    // TODO: during chain startup we want to pushback on apply so that chainbase has to wait for db.
    // TODO: assert that last irreversible in db is one less than received (on startup only?, every so often?)
-   my->chain_plug = app().find_plugin<chain_plugin>();
+//   my->chain_plug = app().find_plugin<chain_plugin>();
 //   auto& db = my->chain_plug->chain().get_mutable_database();
 //   db.add_index<account_control_history_multi_index>();
 //   db.add_index<account_transaction_history_multi_index>();
 //   db.add_index<public_key_history_multi_index>();
 //   db.add_index<transaction_history_multi_index>();
 
+   my->consum_thread = boost::thread([this]{ my->consum_blocks(); });
+
+   // chain_controller is created and has resynced or replayed if needed
+   my->startup = false;
 }
 
 void db_plugin::plugin_shutdown()
 {
+   my->done = true;
+   my->consum_thread.join();
 }
 
 } // namespace eos