merge

.gitignore

@@ -18,4 +18,3 @@ build/*
 *cmake-build-debug*
 cmake-build-debug/*
 docker/build/iroha.tar
-

.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

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/repository/consensus/CMakeLists.txt

@@ -1,7 +1,6 @@
 SET(LIBRARY_OUTPUT_PATH ${PROJECT_BINARY_DIR}/lib)
 
 add_library(event_repository STATIC
-    event_repository.cpp
     transaction_repository.cpp
 )
 target_link_libraries(event_repository