Boost fiber

libraries/app/api.cpp

@@ -39,7 +39,9 @@
 #include <fc/crypto/base64.hpp>
 #include <fc/crypto/hex.hpp>
 #include <fc/rpc/api_connection.hpp>
-#include <fc/thread/future.hpp>
+#include <fc/thread/async.hpp>
+
+#include <boost/fiber/future.hpp>
 
 template class fc::api<graphene::app::block_api>;
 template class fc::api<graphene::app::network_broadcast_api>;
@@ -187,12 +189,13 @@ namespace graphene { namespace app {
 
     fc::variant network_broadcast_api::broadcast_transaction_synchronous(const precomputable_transaction& trx)
     {
-       fc::promise<fc::variant>::ptr prom = fc::promise<fc::variant>::create();
-       broadcast_transaction_with_callback( [prom]( const fc::variant& v ){
-        prom->set_value(v);
+       boost::fibers::promise<fc::variant> prom;
+       boost::fibers::future<fc::variant> result = prom.get_future();
+       broadcast_transaction_with_callback( [&prom]( const fc::variant& v ) {
+        prom.set_value(v);
        }, trx );
 
-       return fc::future<fc::variant>(prom).wait();
+       return result.get();
     }
 
     void network_broadcast_api::broadcast_block( const signed_block& b )
@@ -356,12 +359,7 @@ namespace graphene { namespace app {
        if(_app.is_plugin_enabled("elasticsearch")) {
           auto es = _app.get_plugin<elasticsearch::elasticsearch_plugin>("elasticsearch");
           if(es.get()->get_running_mode() != elasticsearch::mode::only_save) {
-             if(!_app.elasticsearch_thread)
-                _app.elasticsearch_thread= std::make_shared<fc::thread>("elasticsearch");
-
-             return _app.elasticsearch_thread->async([&es, &account, &stop, &limit, &start]() {
-                return es->get_account_history(account, stop, limit, start);
-             }, "thread invoke for method " BOOST_PP_STRINGIZE(method_name)).wait();
+             return es->get_account_history(account, stop, limit, start);
           }
        }
 

libraries/app/database_api.cpp

@@ -31,6 +31,7 @@
 
 #include <fc/crypto/hex.hpp>
 #include <fc/rpc/api_connection.hpp>
+#include <fc/thread/async.hpp>
 
 #include <boost/range/iterator_range.hpp>
 

libraries/app/include/graphene/app/application.hpp

@@ -139,8 +139,6 @@ namespace graphene { namespace app {
 
          bool is_plugin_enabled(const string& name) const;
 
-         std::shared_ptr<fc::thread> elasticsearch_thread;
-
    private:
          void add_available_plugin( std::shared_ptr<abstract_plugin> p );
          std::shared_ptr<detail::application_impl> my;

libraries/chain/db_block.cpp

@@ -39,6 +39,7 @@
 
 #include <graphene/protocol/fee_schedule.hpp>
 
+#include <fc/asio.hpp>
 #include <fc/io/raw.hpp>
 #include <fc/thread/parallel.hpp>
 
@@ -822,9 +823,9 @@ void database::_precompute_parallel( const Trx* trx, const size_t count, const u
    }
 }
 
-fc::future<void> database::precompute_parallel( const signed_block& block, const uint32_t skip )const
+boost::fibers::future<void> database::precompute_parallel( const signed_block& block, const uint32_t skip )const
 { try {
-   std::vector<fc::future<void>> workers;
+   std::vector<boost::fibers::future<void>> workers;
    if( !block.transactions.empty() )
    {
       if( (skip & skip_expensive) == skip_expensive )
@@ -850,16 +851,20 @@ fc::future<void> database::precompute_parallel( const signed_block& block, const
    block.id();
 
    if( workers.empty() )
-      return fc::future< void >( fc::promise< void >::create( true ) );
+   {
+      boost::fibers::promise< void > done;
+      done.set_value();
+      return done.get_future();
+   }
 
    auto first = workers.begin();
    auto worker = first;
    while( ++worker != workers.end() )
       worker->wait();
-   return *first;
+   return std::move( *first );
 } FC_LOG_AND_RETHROW() }
 
-fc::future<void> database::precompute_parallel( const precomputable_transaction& trx )const
+boost::fibers::future<void> database::precompute_parallel( const precomputable_transaction& trx )const
 {
    return fc::do_parallel([this,&trx] () {
       _precompute_parallel( &trx, 1, skip_nothing );

libraries/chain/db_management.cpp

@@ -80,7 +80,7 @@ void database::reindex( fc::path data_dir )
 
    size_t total_block_size = _block_id_to_block.total_block_size();
    const auto& gpo = get_global_properties();
-   std::queue< std::tuple< size_t, signed_block, fc::future< void > > > blocks;
+   std::queue< std::tuple< size_t, signed_block, boost::fibers::future< void > > > blocks;
    uint32_t next_block_num = head_block_num() + 1;
    uint32_t i = next_block_num;
    while( next_block_num <= last_block_num || !blocks.empty() )
@@ -93,7 +93,7 @@ void database::reindex( fc::path data_dir )
          {
             if( block->timestamp >= last_block->timestamp - gpo.parameters.maximum_time_until_expiration )
                skip &= ~skip_transaction_dupe_check;
-            blocks.emplace( processed_block_size, std::move(*block), fc::future<void>() );
+            blocks.emplace( processed_block_size, std::move(*block), boost::fibers::future<void>() );
             std::get<2>(blocks.back()) = precompute_parallel( std::get<1>(blocks.back()), skip );
          }
          else

libraries/chain/include/graphene/chain/database.hpp

@@ -37,10 +37,12 @@
 #include <graphene/db/object_database.hpp>
 #include <graphene/db/object.hpp>
 #include <graphene/db/simple_index.hpp>
-#include <fc/signals.hpp>
 
+#include <fc/signals.hpp>
 #include <fc/log/logger.hpp>
 
+#include <boost/fiber/future.hpp>
+
 #include <map>
 
 namespace graphene { namespace chain {
@@ -453,7 +455,8 @@ namespace graphene { namespace chain {
           * @return a future that will resolve to the input block with
           *         precomputations applied
           */
-         fc::future<void> precompute_parallel( const signed_block& block, const uint32_t skip = skip_nothing )const;
+         boost::fibers::future<void> precompute_parallel( const signed_block& block,
+                                                          const uint32_t skip = skip_nothing )const;
 
          /** Precomputes digests, signatures and operation validations.
           *  "Expensive" computations may be done in a parallel thread.
@@ -462,7 +465,7 @@ namespace graphene { namespace chain {
           * @return a future that will resolve to the input transaction with
           *         precomputations applied
           */
-         fc::future<void> precompute_parallel( const precomputable_transaction& trx )const;
+         boost::fibers::future<void> precompute_parallel( const precomputable_transaction& trx )const;
    private:
          template<typename Trx>
          void _precompute_parallel( const Trx* trx, const size_t count, const uint32_t skip )const;

libraries/db/object_database.cpp

@@ -72,7 +72,7 @@ void object_database::flush()
 {
 //   ilog("Save object_database in ${d}", ("d", _data_dir));
    fc::create_directories( _data_dir / "object_database.tmp" / "lock" );
-   std::vector<fc::future<void>> tasks;
+   std::vector<boost::fibers::future<void>> tasks;
    tasks.reserve(200);
    for( uint32_t space = 0; space < _index.size(); ++space )
    {
@@ -109,7 +109,7 @@ void object_database::open(const fc::path& data_dir)
        wlog("Ignoring locked object_database");
        return;
    }
-   std::vector<fc::future<void>> tasks;
+   std::vector<boost::fibers::future<void>> tasks;
    tasks.reserve(200);
    ilog("Opening object database from ${d} ...", ("d", data_dir));
    for( uint32_t space = 0; space < _index.size(); ++space )

libraries/net/CMakeLists.txt

@@ -12,7 +12,7 @@ set(SOURCES node.cpp
 add_library( graphene_net ${SOURCES} ${HEADERS} )
 
 target_link_libraries( graphene_net 
-  PUBLIC fc graphene_db graphene_protocol )
+  PUBLIC fc graphene_db graphene_protocol graphene_utilities )
 target_include_directories( graphene_net 
   PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/include"
   PRIVATE "${CMAKE_SOURCE_DIR}/libraries/chain/include"

libraries/net/include/graphene/net/node.hpp

@@ -30,6 +30,7 @@
 #include <graphene/protocol/types.hpp>
 
 #include <list>
+#include <thread>
 
 namespace graphene { namespace net {
 
@@ -293,35 +294,10 @@ namespace graphene { namespace net {
         void disable_peer_advertising();
         fc::variant_object get_call_statistics() const;
       private:
-        std::unique_ptr<detail::node_impl, detail::node_impl_deleter> my;
+        std::unique_ptr<detail::node_impl> my;
    };
 
-    class simulated_network : public node
-    {
-    public:
-      ~simulated_network();
-      simulated_network(const std::string& user_agent) : node(user_agent) {}
-      void      listen_to_p2p_network() override {}
-      void      connect_to_p2p_network() override {}
-      void      connect_to_endpoint(const fc::ip::endpoint& ep) override {}
-
-      fc::ip::endpoint get_actual_listening_endpoint() const override { return fc::ip::endpoint(); }
-
-      void      sync_from(const item_id& current_head_block, const std::vector<uint32_t>& hard_fork_block_numbers) override {}
-      void      broadcast(const message& item_to_broadcast) override;
-      void      add_node_delegate(node_delegate* node_delegate_to_add);
-
-      virtual uint32_t get_connection_count() const override { return 8; }
-    private:
-      struct node_info;
-      void message_sender(node_info* destination_node);
-      std::list<node_info*> network_nodes;
-    };
-
-
    typedef std::shared_ptr<node> node_ptr;
-   typedef std::shared_ptr<simulated_network> simulated_network_ptr;
-
 } } // graphene::net
 
 FC_REFLECT(graphene::net::message_propagation_data, (received_time)(validated_time)(originating_peer));

libraries/net/include/graphene/net/peer_connection.hpp

@@ -27,6 +27,7 @@
 #include <graphene/net/peer_database.hpp>
 #include <graphene/net/message_oriented_connection.hpp>
 #include <graphene/net/config.hpp>
+#include <graphene/utilities/recurring_task.hpp>
 
 #include <boost/tuple/tuple.hpp>
 
@@ -39,10 +40,25 @@
 
 #include <queue>
 #include <boost/container/deque.hpp>
-#include <fc/thread/future.hpp>
 
-namespace graphene { namespace net
-  {
+namespace graphene { namespace net {
+   class peer_connection;
+
+   namespace detail {
+
+   class send_queued_messages_task : public graphene::utilities::recurring_task {
+   public:
+      send_queued_messages_task() : _conn(nullptr) {}
+      explicit send_queued_messages_task( peer_connection& conn ) : _conn(&conn) {}
+
+   protected:
+      virtual void run();
+
+      peer_connection* _conn;
+   };
+
+   } // detail
+
     struct firewall_check_state_data
     {
       node_id_t        expected_node_id;
@@ -58,7 +74,6 @@ namespace graphene { namespace net
       node_id_t        requesting_peer;
     };
 
-    class peer_connection;
     class peer_connection_delegate
     {
     public:
@@ -69,7 +84,6 @@ namespace graphene { namespace net
       virtual message get_message_for_item(const item_id& item) = 0;
     };
 
-    class peer_connection;
     typedef std::shared_ptr<peer_connection> peer_connection_ptr;
     class peer_connection : public message_oriented_connection_delegate,
                             public std::enable_shared_from_this<peer_connection>
@@ -166,7 +180,7 @@ namespace graphene { namespace net
 
       size_t _total_queued_messages_size = 0;
       std::queue<std::unique_ptr<queued_message>, std::list<std::unique_ptr<queued_message> > > _queued_messages;
-      fc::future<void> _send_queued_messages_done;
+      detail::send_queued_messages_task _send_queued_messages;
     public:
       fc::time_point connection_initiation_time;
       fc::time_point connection_closed_time;
@@ -260,8 +274,6 @@ namespace graphene { namespace net
 
       uint32_t last_known_fork_block_number = 0;
 
-      fc::future<void> accept_or_connect_task_done;
-
       firewall_check_state_data *firewall_check_state = nullptr;
     private:
 #ifndef NDEBUG
@@ -313,6 +325,7 @@ namespace graphene { namespace net
       void send_queued_messages_task();
       void accept_connection_task();
       void connect_to_task(const fc::ip::endpoint& remote_endpoint);
+      friend class detail::send_queued_messages_task;
     };
     typedef std::shared_ptr<peer_connection> peer_connection_ptr;