[bloc-157] integrate grandpa (#23)

* [bloc-157] add sync version of grandpa

* [bloc-157] integrate grandpa

plugins/grandpa_plugin/include/eosio/grandpa_plugin/grandpa.hpp

@@ -1,5 +1,6 @@
-#include <eosio/grandpa_plugin/network_messages.hpp>
-#include <eosio/grandpa_plugin/prefix_chain_tree.hpp>
+#pragma once
+#include "network_messages.hpp"
+#include "prefix_chain_tree.hpp"
 #include <fc/exception/exception.hpp>
 #include <fc/io/json.hpp>
 #include <fc/bitutil.hpp>
@@ -8,6 +9,7 @@
 #include <atomic>
 #include <mutex>
 #include <thread>
+#include <condition_variable>
 
 
 using ::fc::static_variant;
@@ -228,19 +230,23 @@ class grandpa {
         );
         update_lib(lib_id);
 
+#ifndef SYNC_GRANDPA
         _thread_ptr.reset(new std::thread([this]() {
             wlog("Grandpa thread started");
             loop();
             wlog("Grandpa thread terminated");
         }));
+#endif
 
         subscribe();
     }
 
     void stop() {
+#ifndef SYNC_GRANDPA
         _done = true;
         _message_queue.terminate();
         _thread_ptr->join();
+#endif
     }
 
 private:
@@ -250,7 +256,10 @@ class grandpa {
     prefix_chain_tree_ptr _prefix_tree_ptr;
     std::map<uint32_t, peer_info> _peers;
 
+#ifndef SYNC_GRANDPA
     message_queue<grandpa_message> _message_queue;
+#endif
+
     net_channel_ptr _in_net_channel;
     net_channel_ptr _out_net_channel;
     event_channel_ptr _in_event_channel;
@@ -261,12 +270,20 @@ class grandpa {
     void subscribe() {
         _in_net_channel->subscribe([&](const grandpa_net_msg& msg) {
             dlog("Grandpa received net message, type: ${type}", ("type", msg.data.which()));
+#ifdef SYNC_GRANDPA
+            process_msg(std::make_shared<grandpa_message>(msg));
+#else
             _message_queue.push_message(msg);
+#endif
         });
 
         _in_event_channel->subscribe([&](const grandpa_event& event) {
             dlog("Grandpa received event, type: ${type}", ("type", event.data.which()));
+#ifdef SYNC_GRANDPA
+            process_msg(std::make_shared<grandpa_message>(event));
+#else
             _message_queue.push_message(event);
+#endif
         });
     }
 
@@ -288,6 +305,7 @@ class grandpa {
         }
     }
 
+#ifndef SYNC_GRANDPA
     void loop() {
         while (true) {
             auto msg = _message_queue.get_next_msg_wait();
@@ -301,6 +319,7 @@ class grandpa {
             process_msg(msg);
         }
     }
+#endif
 
     // void do_bft_finalize(const block_id_type& block_id) {
     //     app().get_io_service().post([this, bid = block_id]() {

plugins/grandpa_plugin/include/eosio/grandpa_plugin/network_messages.hpp

@@ -1,8 +1,7 @@
 #pragma once
 
-#include <eosio/grandpa_plugin/types.hpp>
-#include <eosio/grandpa_plugin/prefix_chain_tree.hpp>
-#include <eosio/chain/types.hpp>
+#include "types.hpp"
+#include "prefix_chain_tree.hpp"
 #include <fc/reflect/reflect.hpp>
 
 using std::vector;

plugins/grandpa_plugin/include/eosio/grandpa_plugin/prefix_chain_tree.hpp

@@ -1,6 +1,6 @@
 #pragma once
 
-#include <eosio/grandpa_plugin/types.hpp>
+#include "types.hpp"
 #include <memory>
 #include <vector>
 #include <map>

simulator/CMakeLists.txt

@@ -16,6 +16,7 @@ ExternalProject_Get_Property(gtest source_dir binary_dir)
 
 add_dependencies(simulator gtest)
 include_directories(${source_dir}/googletest/include)
+include_directories(${CMAKE_CURRENT_SOURCE_DIR}/../plugins/grandpa_plugin/include)
 
 target_link_libraries(simulator ${binary_dir}/googlemock/gtest/libgtest.a pthread fc)
 
@@ -25,4 +26,4 @@ target_link_libraries(simulator ${binary_dir}/googlemock/gtest/libgtest.a pthrea
 
 enable_testing()
 add_test(NAME    simulator
-        COMMAND simulator)
+        COMMAND simulator)

simulator/include/database.hpp

@@ -27,7 +27,7 @@ struct fork_db_node {
     }
 };
 
-struct node_info {
+struct block_info {
     fork_db_node_ptr node;
     size_t height;
 };
@@ -131,8 +131,8 @@ class fork_db {
         return result;
     }
 
-    node_info find_master_head(const fork_db_node_ptr& node, size_t depth) const {
-        auto result = node_info{node, depth};
+    block_info find_master_head(const fork_db_node_ptr& node, size_t depth) const {
+        auto result = block_info{node, depth};
         for (const auto& adjacent_node : node->adjacent_nodes) {
             const auto head_node = find_master_head(adjacent_node, depth + 1);
             if (head_node.height > result.height) {
@@ -168,4 +168,4 @@ class fork_db {
         }
         return node;
     }
-};
+};

simulator/include/grandpa.hpp

@@ -0,0 +1,90 @@
+#pragma once
+#include "simulator.hpp"
+#include "database.hpp"
+#define SYNC_GRANDPA //SYNC mode
+#include <eosio/grandpa_plugin/grandpa.hpp>
+#include <mutex>
+
+
+
+class GrandpaNode: public Node {
+public:
+    explicit GrandpaNode(int id, Network && net, fork_db && db):
+        Node(id, std::move(net), std::move(db))
+    {
+        init_channels();
+        init_providers();
+        init_grandpa();
+
+        grandpa.start();
+    }
+
+    ~GrandpaNode() {
+        grandpa.stop();
+    }
+
+    void on_receive(uint32_t from, void* msg) override {
+        auto data = *static_cast<grandpa_net_msg*>(msg);
+        data.ses_id = from;
+        in_net_ch->send(data);
+    }
+
+    void on_new_peer_event(uint32_t id) override {
+        ev_ch->send(grandpa_event { ::on_new_peer_event { id } });
+    }
+
+    void on_accepted_block_event(block_id_type id) override {
+        ev_ch->send(grandpa_event { ::on_accepted_block_event { id } });
+    }
+
+private:
+    void init_channels() {
+        in_net_ch = std::make_shared<net_channel>();
+        out_net_ch = std::make_shared<net_channel>();
+        ev_ch = std::make_shared<event_channel>();
+
+        out_net_ch->subscribe([this](const grandpa_net_msg& msg) {
+            send<grandpa_net_msg>(msg.ses_id, msg);
+        });
+    }
+
+    void init_providers() {
+        prev_block_prov = std::make_shared<prev_block_prodiver>([this](const block_id_type& id) -> fc::optional<block_id_type> {
+            auto block = db.find(id);
+            if (!block || !block->parent)
+                return {};
+            else
+                return block->parent->block_id;
+        });
+
+        lib_prov = std::make_shared<lib_prodiver>([this]() -> block_id_type {
+            return db.last_irreversible_block_id();
+        });
+
+        prods_prov = std::make_shared<prods_provider>([]() -> vector<public_key_type> {
+            return {};
+        });
+    }
+
+    void init_grandpa() {
+        grandpa
+            .set_event_channel(ev_ch)
+            .set_in_net_channel(in_net_ch)
+            .set_out_net_channel(out_net_ch)
+            .set_prev_block_provider(prev_block_prov)
+            .set_lib_provider(lib_prov)
+            .set_prods_provider(prods_prov)
+            .set_private_key(private_key_type::generate());
+    }
+
+    net_channel_ptr in_net_ch;
+    net_channel_ptr out_net_ch;
+    event_channel_ptr ev_ch;
+
+    prev_block_prodiver_ptr prev_block_prov;
+    lib_prodiver_ptr lib_prov;
+    prods_provider_ptr prods_prov;
+
+    grandpa grandpa;
+};
+

simulator/include/simulator.hpp

@@ -42,10 +42,8 @@ class Clock {
 
 class TestRunner;
 class Node;
-
 using NodePtr = std::shared_ptr<Node>;
 
-
 struct Task {
     uint32_t from;
     uint32_t to;
@@ -87,6 +85,7 @@ class Node {
 public:
     Node() = default;
     explicit Node(int id, Network && net, fork_db&& db): id(id), net(std::move(net)), db(std::move(db)) {}
+    virtual ~Node() = default;
 
     template <typename T>
     void send(uint32_t to, const T& msg) {
@@ -104,17 +103,22 @@ class Node {
         }
         cout << "]" << endl;
         db.insert(chain);
+        for (auto& block_id : chain.blocks) {
+            on_accepted_block_event(block_id);
+        }
     }
 
-    virtual void on_receive(void *) {
-        std::cout << "Received by " << id << std::endl;
+    virtual void on_receive(uint32_t from, void *) {
+        std::cout << "Received from " << from << std::endl;
     }
 
-    virtual void on_new_peer_event(uint32_t from)  {
+    virtual void on_new_peer_event(uint32_t from) {
         std::cout << "On new peer event handled by " << id << " at " << tester_clock.now() << endl;
     }
 
-    virtual ~Node() = default;
+    virtual void on_accepted_block_event(block_id_type id) {
+        std::cout << "On accepted block event handled by " << id << " at " << tester_clock.now() << endl;
+    }
 
     uint32_t id;
     bool is_producer = true;
@@ -280,7 +284,7 @@ class TestRunner {
                 task.cb(nodes[task.to]);
             }
 
-            this_thread::sleep_for(chrono::milliseconds(3000));
+            this_thread::sleep_for(chrono::milliseconds(1000));
         }
     }
 
@@ -385,13 +389,13 @@ void Network::send(uint32_t to, const T& msg) {
         node_id,
         to,
         tester_clock.now() + matrix[node_id][to],
-        [msg](NodePtr n) {
-            n->on_receive(msg);
+        [node_id = node_id, msg = msg](NodePtr n) {
+            n->on_receive(node_id, (void*)&msg);
         }
     });
 }
 
 template <typename T>
 void Network::bcast(const T& msg) {
     //TODO bcast to all nodes with calculate routes
-}
+}

simulator/tests/no_byzantine_nodes.cpp

@@ -7,17 +7,18 @@
 #include <cstdlib>
 #include <ctime>
 #include <random>
+#include <grandpa.hpp>
 
 using namespace std;
 
 using std::string;
 
 TEST(A, B) {
-    srand(66);
-    TestRunner t(3);
-    vector<pair<int, int> > v0{{1, 2}, {2, 5}};
+    auto t = TestRunner(3);
+    vector<pair<int, int> > v0{{1, 2}, {2, 10}};
     graph_type g;
     g.push_back(v0);
     t.load_graph(g);
-    t.run<>();
+    assert(t.get_delay_matrix()[0][1] == 2);
+    t.run<GrandpaNode>();
 }