Merge pull request #107 from hyperledger/develop

Develop/refactor core/model, core/util

.gitignore

@@ -17,3 +17,5 @@ build/*
 .vscode/*
 *cmake-build-debug*
 cmake-build-debu/*
+config/sumeragi.json
+CMakeLists.txt.user

.gitmodules

@@ -20,4 +20,6 @@
 [submodule "core/vendor/Cappuccino/lib/json"]
   path = core/vendor/Cappuccino/lib/json
   url = https://github.com/nlohmann/json.git
-
+[submodule "core/vendor/thread_pool_cpp"]
+    path = core/vendor/thread_pool_cpp
+    url = https://github.com/Warchant/thread-pool-cpp.git

CMakeLists.txt

@@ -14,6 +14,10 @@ add_subdirectory(peer)
 
 add_subdirectory(tools)
 
+# at some point we want to include pool as following: #include <worker.hpp>
+# which resides in that folder
+include_directories("${PROJECT_SOURCE_DIR}/core/vendor/thread_pool_cpp/thread_pool")
+
 if(NOT without-test)
   enable_testing()
   include(test/gtest.cmake)

circle.yml

@@ -20,6 +20,9 @@ machine:
 
 checkout:
   post:
+    - git submodule init
+    - git submodule update
+
     - cd $IROHA_HOME
     - cd $IROHA_HOME/core/vendor/leveldb; make -j 4
     - cd $IROHA_HOME/core/vendor/ed25519; make -j 4
@@ -45,4 +48,4 @@ test:
   pre:
     - mkdir build; cd build; cmake ..; make
   override:
-    - ./test.sh
+    - ./test.sh

config/sumeragi.json

@@ -1,7 +1,7 @@
 {
   "me":{
     "ip":"172.17.0.6",
-    "name":"kabohara",
+    "name":"samari",
     "publicKey":"Sht5opDIxbyK+oNuEnXUs5rLbrvVgb2GjSPfqIYGFdU=",
     "privateKey":"aGIuSZRhnGfFyeoKNm/NbTylnAvRfMu3KumOEfyT2HPf36jSF22m2JXWrdCmKiDoshVqjFtZPX3WXaNuo9L8WA=="
   },

core/consensus/CMakeLists.txt

@@ -3,6 +3,11 @@ SET(LIBRARY_OUTPUT_PATH ${PROJECT_BINARY_DIR}/lib)
 ADD_LIBRARY(sumeragi STATIC
   sumeragi.cpp
 )
+
+include_directories(
+  "${PROJECT_SOURCE_DIR}/core/vendor/thread_pool_cpp/thread_pool"
+)
+
 target_link_libraries(sumeragi
   peer_service_with_json
   connection_with_grpc

core/consensus/sumeragi.cpp

@@ -20,9 +20,10 @@ limitations under the License.
 #include <deque>
 #include <cmath>
 
+#include <thread_pool.hpp>
+
 #include "../util/logger.hpp"
 #include "../repository/consensus/merkle_transaction_repository.hpp"
-#include "../repository/consensus/event_repository.hpp"
 #include "../crypto/hash.hpp"
 #include "../crypto/signature.hpp"
 
@@ -56,6 +57,21 @@ namespace sumeragi {
     using namespace command;
     using namespace object;
 
+
+    static size_t concurrency = 
+        std::thread::hardware_concurrency() <= 0
+        ? 1
+        : std::thread::hardware_concurrency();
+
+    //thread pool and a storage of events 
+    static ThreadPool pool(
+        ThreadPoolOptions{
+            .threads_count = concurrency,
+            .worker_queue_size = 1024
+        }
+    );
+
+
     namespace detail{
 
         unsigned int getNumValidSignatures(const Event::ConsensusEvent& event){
@@ -147,7 +163,7 @@ namespace sumeragi {
                 logger::explore("sumeragi", "\033[91m|+-ーー-+|\033[0m");
                 logger::explore("sumeragi", "\033[91m+==ーー==+\033[0m");
         }
-    }
+    } // namespace detail
 
     struct Context {
         bool isSumeragi; // am I the leader or am I not?
@@ -212,16 +228,16 @@ namespace sumeragi {
 
         context->isSumeragi = context->validatingPeers.at(0)->getPublicKey() == context->myPublicKey;
 
-        connection::receive([&](
-            const std::string& from,
-            Event::ConsensusEvent& event
-        ){
+        connection::receive([](const std::string& from, Event::ConsensusEvent& event){
             logger::info("sumeragi", "receive!");
-            auto hash = event.transaction().hash();
-
             logger::info("sumeragi", "received message! sig:[" + std::to_string(event.eventsignatures_size()) +"]");
-            // WIP currently, unuse hash in event repository,
-            repository::event::add( hash, event);
+
+            // send processTransaction(event) as a task to processing pool
+            // this returns std::future<void> object
+            // (std::future).get() method locks processing until result of processTransaction will be available
+            // but processTransaction returns void, so we don't have to call it and wait
+            std::function<void()> &&task = std::bind(processTransaction, std::ref(event)); 
+            pool.process(std::move(task)); 
         });
 
         logger::info("sumeragi", "initialize numValidatingPeers :" + std::to_string(context->numValidatingPeers));
@@ -237,6 +253,7 @@ namespace sumeragi {
         logger::info("sumeragi", "initialize.....  complete!");
     }
 
+
     unsigned long long getNextOrder() {
         return 0l;
         //return merkle_transaction_repository::getLastLeafOrder() + 1;
@@ -439,44 +456,45 @@ namespace sumeragi {
         context->isSumeragi = context->validatingPeers.at(0)->getPublicKey() == context->myPublicKey;
     }
 
+
     void loop() {
         logger::info("sumeragi", "=+=");
         logger::info("sumeragi", "start main loop");
 
-        while (true) {  // 千五百秋　TODO: replace with callback linking the event repository?
-            if(!repository::event::empty()) {
-                // Determine node order
-                determineConsensusOrder();
-
-                logger::info("sumeragi", "event queue not empty");
-
-                auto events = repository::event::findAll();
-                /*
-                logger::info("sumeragi", "event's size " + std::to_string(events.size()));
-                
-                // Sort the events to determine priority to process
-                std::sort(events.begin(), events.end(),
-                    [&](const auto &lhs,const auto &rhs) {
-                        return lhs->getNumValidSignatures() > rhs->getNumValidSignatures()
-                            || (context->isSumeragi && lhs->order == 0)
-                            || lhs->order < rhs->order;
-                    }
-                );
-                */
-                logger::info("sumeragi", "sorted " + std::to_string(events.size()));
-                for (auto& event : events) {
-
-                    logger::info("sumeragi", "evens order:" + std::to_string(event.order()));
-                    /*
-                    if (!transaction_validator::isValid(event)) {
-                        continue;
-                    }
-                    */
-                    // Process transaction
-                    std::thread([&event]{ processTransaction(event); }).join();
-                }
-            }
-        }
+//        while (true) {  // 千五百秋　TODO: replace with callback linking the event repository?
+//            if(!repository::event::empty()) {
+//                // Determine node order
+//                determineConsensusOrder();
+//
+//                logger::info("sumeragi", "event queue not empty");
+//
+//                auto events = repository::event::findAll();
+//                /*
+//                logger::info("sumeragi", "event's size " + std::to_string(events.size()));
+//                
+//                // Sort the events to determine priority to process
+//                std::sort(events.begin(), events.end(),
+//                    [&](const auto &lhs,const auto &rhs) {
+//                        return lhs->getNumValidSignatures() > rhs->getNumValidSignatures()
+//                            || (context->isSumeragi && lhs->order == 0)
+//                            || lhs->order < rhs->order;
+//                    }
+//                );
+//                */
+//                logger::info("sumeragi", "sorted " + std::to_string(events.size()));
+//                for (auto& event : events) {
+//
+//                    logger::info("sumeragi", "evens order:" + std::to_string(event.order()));
+//                    /*
+//                    if (!transaction_validator::isValid(event)) {
+//                        continue;
+//                    }
+//                    */
+//                    // Process transaction
+//                    std::thread([&event]{ processTransaction(event); }).join();
+//                }
+//            }
+//        }
     }
 
 };  // namespace sumeragi

core/consensus/sumeragi.hpp

@@ -21,6 +21,7 @@ limitations under the License.
 #define COMPARATOR(code) [](auto && l, auto && r) -> bool { return code ; }
 
 #include <vector>
+#include <thread>
 #include <memory>
 
 #include "consensus_event.hpp"
@@ -53,7 +54,7 @@ namespace sumeragi {
     void panic(const Event::ConsensusEvent& event);
     void setAwkTimer(const int sleepMillisecs, const std::function<void(void)> action);
     void determineConsensusOrder(/*std::vector<double> trustVector*/);
-    
+
 };  // namespace sumeragi
 
 #endif  // CORE_CONSENSUS_SUMERAGI_HPP_

core/infra/repository/world_state_repository_with_level_db.cpp

@@ -14,8 +14,8 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-#include "../../../repository/world_state_repository.hpp"
-#include "../../../util/exception.hpp"
+#include "../../repository/world_state_repository.hpp"
+#include "../../util/exception.hpp"
 
 #include "../../../util/logger.hpp"
 

core/model/commands/transfer.hpp

@@ -36,7 +36,7 @@ namespace command {
         std::string receiverPublicKey;
 
         template<typename... Args>
-        constexpr explicit Transfer(
+        explicit Transfer(
                 std::string&& sender,
                 std::string&& receiver,
                 Args&&... args

core/model/commands/update.hpp

@@ -33,15 +33,15 @@ namespace command {
         std::string ownerPublicKey;
 
         template<typename... Args>
-        constexpr explicit Update(
+        explicit Update(
             std::string&& ownerPublicKey,
             Args&&... args
         ):
             T(std::forward<Args>(args)...),
             ownerPublicKey(std::move(ownerPublicKey))
         {}
 
-        auto getCommandName() const{
+        auto getCommandName() const {
             return "Update";
         }
 

core/model/objects/account.hpp

@@ -21,15 +21,15 @@ limitations under the License.
 #include <string>
 
 namespace object {
-
+    
     class Account {
 
     public:
         std::string publicKey;
         std::string name;
 
         std::vector<
-            std::tuple<std::string,long >
+            std::tuple<std::string, long>
         > assets;
 
         explicit Account():
@@ -38,35 +38,35 @@ namespace object {
         {}
 
         explicit Account(
-            std::string&& publicKey,
-            std::string&&  name,
-                std::vector<
-                std::tuple<std::string,long>
-            >&& assets
+            std::string publicKey,
+            std::string name,
+            std::vector<
+                std::tuple<std::string, long>
+            > assets
         ):
-            publicKey(publicKey),
-            name(name),
-            assets(assets)
+            publicKey(std::move(publicKey)),
+            name(std::move(name)),
+            assets(std::move(assets))
         {}
 
         explicit Account(
-            std::string&& publicKey,
-            std::string&&  name,
-            std::tuple<std::string,long>&& asset
+            std::string publicKey,
+            std::string name,
+            std::tuple<std::string, long> asset
         ):
-            publicKey(publicKey),
-            name(name)
+            publicKey(std::move(publicKey)),
+            name(std::move(name))
         {
-            assets.push_back(asset);
+            assets.push_back(std::move(asset));
         }
 
 
         explicit Account(
-            std::string&& publicKey,
-            std::string&&  name
+            std::string publicKey,
+            std::string name
         ):
-            publicKey(publicKey),
-            name(name)
+            publicKey(std::move(publicKey)),
+            name(std::move(name))
         {}
 
 

core/model/objects/asset.cpp

@@ -19,23 +19,23 @@ limitations under the License.
 namespace object {
 
 Asset::Asset(
-    std::string&& domain,
-    std::string&&  name,
-    unsigned long long&& value,
-    unsigned int&& precision
+    std::string domain,
+    std::string name,
+    unsigned long long value,
+    unsigned int precision
 ):
-    domain(domain),
-    name(name),
+    domain(std::move(domain)),
+    name(std::move(name)),
     value(value),
     precision(precision)
 {}
 
 Asset::Asset(
-    std::string&& name,
-    unsigned long long&& value
+    std::string name,
+    unsigned long long value
 ):
     domain(""),
-    name(name),
+    name(std::move(name)),
     value(value),
     precision(-1)
 {}