Progress #6: Integrate P2P

Define a new p2p_plugin as the adaptor between appbase and the old P2P
library, and wire things together. This is mostly working, but for some
reason peers won't stay connected to sync, but instead just drop their
connection (with a TCP reset, no less) right before syncing, without
bothering to log about why. I suspect the failure is at a lower level. :
(

libraries/net/CMakeLists.txt

@@ -10,7 +10,7 @@ set(SOURCES node.cpp
 add_library( eos_net ${SOURCES} ${HEADERS} )
 
 target_link_libraries( eos_net 
-  PUBLIC fc chainbase )
+  PUBLIC fc chainbase appbase )
 target_include_directories( eos_net 
   PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/include"
   PRIVATE "${CMAKE_SOURCE_DIR}/libraries/chain/include"

libraries/net/node.cpp

@@ -80,14 +80,16 @@
 
 #include <eos/chain/config.hpp>
 
+#include <appbase/application.hpp>
+
 #include <fc/git_revision.hpp>
 
 //#define ENABLE_DEBUG_ULOGS
 
-#ifdef DEFAULT_LOGGER
-# undef DEFAULT_LOGGER
-#endif
-#define DEFAULT_LOGGER "p2p"
+//#ifdef DEFAULT_LOGGER
+//# undef DEFAULT_LOGGER
+//#endif
+//#define DEFAULT_LOGGER "p2p"
 
 #define P2P_IN_DEDICATED_THREAD 1
 
@@ -2278,7 +2280,7 @@ namespace eos { namespace net { namespace detail {
 
       if (!originating_peer->peer_needs_sync_items_from_us)
       {
-        dlog("sync: peer is already in sync with us");
+        dlog("sync: peer is already in sync with us ${p}", ("p", originating_peer->we_need_sync_items_from_peer));
         // if we thought we had all the items this peer had, but now it turns out that we don't
         // have the last item it requested to send from,
         // we need to kick off another round of synchronization
@@ -5325,6 +5327,38 @@ namespace eos { namespace net { namespace detail {
       return statistics;
     }
 
+    template<typename T>
+    struct invoke_in_appbase {
+       template<typename Lambda>
+       T operator()( Lambda&& l )const {
+          typename fc::promise<T>::ptr p(new fc::promise<T>( "invoke in appbase" ));
+          appbase::app().get_io_service().post( [&](){
+             try {
+                p->set_value( l() );
+             } catch ( const fc::exception& e ) {
+                p->set_exception( e.dynamic_copy_exception() );
+             }
+           } );
+           return p->wait();
+       }
+    };
+    template<>
+    struct invoke_in_appbase<void> {
+       template<typename Lambda>
+       void operator()( Lambda&& l )const {
+          fc::promise<void>::ptr p(new fc::promise<void>( "invoke in appbase" ));
+              appbase::app().get_io_service().post( [&](){
+             try {
+                l();
+                p->set_value();
+             } catch ( const fc::exception& e ) {
+                p->set_exception( e.dynamic_copy_exception() );
+             }
+           } );
+           p->wait();
+       }
+    };
+
 // define VERBOSE_NODE_DELEGATE_LOGGING to log whenever the node delegate throws exceptions
 //#define VERBOSE_NODE_DELEGATE_LOGGING
 #ifdef VERBOSE_NODE_DELEGATE_LOGGING
@@ -5363,22 +5397,25 @@ namespace eos { namespace net { namespace detail {
     }
 #else
 #  define INVOKE_AND_COLLECT_STATISTICS(method_name, ...) \
-    call_statistics_collector statistics_collector(#method_name, \
-                                                   &_ ## method_name ## _execution_accumulator, \
-                                                   &_ ## method_name ## _delay_before_accumulator, \
-                                                   &_ ## method_name ## _delay_after_accumulator); \
-    if (_thread->is_current()) \
-    { \
-      call_statistics_collector::actual_execution_measurement_helper helper(statistics_collector); \
-      return _node_delegate->method_name(__VA_ARGS__); \
-    } \
-    else \
-      return _thread->async([&](){ \
-        call_statistics_collector::actual_execution_measurement_helper helper(statistics_collector); \
-        return _node_delegate->method_name(__VA_ARGS__); \
-      }, "invoke " BOOST_STRINGIZE(method_name)).wait()
-#endif
+   call_statistics_collector statistics_collector(#method_name, \
+                                                  &_ ## method_name ## _execution_accumulator, \
+                                                  &_ ## method_name ## _delay_before_accumulator, \
+                                                  &_ ## method_name ## _delay_after_accumulator); \
+   if (_thread->is_current()) \
+   { \
+     call_statistics_collector::actual_execution_measurement_helper helper(statistics_collector); \
+     return _node_delegate->method_name(__VA_ARGS__); \
+   } \
+   else \
+   { \
+      using T = decltype( _node_delegate->method_name(__VA_ARGS__) ); \
+      return invoke_in_appbase<T>()( [&]() mutable { \
+         call_statistics_collector::actual_execution_measurement_helper helper(statistics_collector); \
+         return _node_delegate->method_name(__VA_ARGS__); \
+      }); \
+   }
 
+#endif
     bool statistics_gathering_node_delegate_wrapper::has_item( const net::item_id& id )
     {
       INVOKE_AND_COLLECT_STATISTICS(has_item, id);

plugins/CMakeLists.txt

@@ -1,4 +1,5 @@
 add_subdirectory(net_plugin)
+add_subdirectory(p2p_plugin)
 add_subdirectory(http_plugin)
 add_subdirectory(chain_plugin)
 add_subdirectory(producer_plugin)

plugins/chain_plugin/chain_plugin.cpp

@@ -124,4 +124,27 @@ void chain_plugin::plugin_shutdown() {
    ilog("database closed successfully");
 }
 
+bool chain_plugin::accept_block(const chain::signed_block& block, bool currently_syncing) {
+   if (currently_syncing && block.block_num() % 10000 == 0) {
+      ilog("Syncing Blockchain --- Got block: #${n} time: ${t} producer: ${p}",
+           ("t", block.timestamp)
+           ("n", block.block_num())
+           ("p", block.producer));
+   }
+
+   return db().push_block(block);
+}
+
+void chain_plugin::accept_transaction(const chain::signed_transaction& trx) {
+   db().push_transaction(trx);
+}
+
+bool chain_plugin::block_is_on_preferred_chain(const chain::block_id_type& block_id) {
+   // If it's not known, it's not preferred.
+   if (!db().is_known_block(block_id)) return false;
+   // Extract the block number from block_id, and fetch that block number's ID from the database.
+   // If the database's block ID matches block_id, then block_id is on the preferred chain. Otherwise, it's on a fork.
+   return db().get_block_id_for_num(chain::block_header::num_from_id(block_id)) == block_id;
+}
+
 }

plugins/chain_plugin/include/eos/chain_plugin/chain_plugin.hpp

@@ -13,14 +13,19 @@ namespace eos {
         APPBASE_PLUGIN_REQUIRES()
 
         chain_plugin();
-        ~chain_plugin();
+        virtual ~chain_plugin();
 
         virtual void set_program_options(options_description& cli, options_description& cfg) override;
 
         void plugin_initialize(const variables_map& options);
         void plugin_startup();
         void plugin_shutdown();
 
+        bool accept_block(const chain::signed_block& block, bool currently_syncing);
+        void accept_transaction(const chain::signed_transaction& trx);
+
+        bool block_is_on_preferred_chain(const chain::block_id_type& block_id);
+
         database& db();
 
       private:

plugins/http_plugin/include/eos/http_plugin/http_plugin.hpp

@@ -34,7 +34,7 @@ namespace eos {
    {
       public:
         http_plugin();
-        ~http_plugin();
+        virtual ~http_plugin();
 
         APPBASE_PLUGIN_REQUIRES()
         virtual void set_program_options(options_description&, options_description& cfg) override;

plugins/net_plugin/include/eos/net_plugin/net_plugin.hpp

@@ -9,7 +9,7 @@ namespace eos {
    {
       public:
         net_plugin();
-        ~net_plugin();
+        virtual ~net_plugin();
 
         APPBASE_PLUGIN_REQUIRES((chain_plugin))
         virtual void set_program_options(options_description& cli, options_description& cfg) override;

plugins/p2p_plugin/CMakeLists.txt

@@ -0,0 +1,17 @@
+file(GLOB HEADERS "include/eos/p2p_plugin/*.hpp")
+
+add_library( p2p_plugin
+             p2p_plugin.cpp
+           )
+
+target_link_libraries( p2p_plugin chain_plugin appbase eos_chain eos_utilities eos_net )
+target_include_directories( p2p_plugin
+                            PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/include" )
+
+install( TARGETS
+   p2p_plugin
+
+   RUNTIME DESTINATION bin
+   LIBRARY DESTINATION lib
+   ARCHIVE DESTINATION lib
+)

plugins/p2p_plugin/include/eos/p2p_plugin/p2p_plugin.hpp

@@ -0,0 +1,51 @@
+/*
+ * Copyright (c) 2015 Cryptonomex, Inc., and contributors.
+ *
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+#pragma once
+
+#include <eos/chain_plugin/chain_plugin.hpp>
+
+#include <appbase/application.hpp>
+
+namespace eos {
+namespace bpo = boost::program_options;
+
+class p2p_plugin : public appbase::plugin<p2p_plugin> {
+public:
+   APPBASE_PLUGIN_REQUIRES((chain_plugin))
+
+   p2p_plugin();
+   virtual ~p2p_plugin();
+
+   virtual void set_program_options(bpo::options_description &,
+                                    bpo::options_description &config_file_options) override;
+
+   virtual void plugin_initialize(const bpo::variables_map& options);
+   virtual void plugin_startup();
+   virtual void plugin_shutdown();
+
+private:
+   std::unique_ptr<class p2p_plugin_impl> my;
+};
+
+} //eos