Merge pull request #354 from skalenetwork/bug/SKALE-4586-no-response

Bug/skale 4586 no response

ExitRequestedException.cpp

+/*
+    Copyright (C) 2019-Present SKALE Labs
+
+    This file is part of sgxwallet.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file ExitRequestedException.cpp
+    @author Stan Kladko
+    @date 2018
+*/
+
+#include "ExitRequestedException.h"
+
+
+ExitRequestedException::ExitRequestedException() {
+
+}

ExitRequestedException.h

+/*
+    Copyright (C) 2019-Present SKALE Labs
+
+    This file is part of sgxwallet.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file ExitRequestedException.h
+    @author Stan Kladko
+    @date 2018
+*/
+
+#pragma once
+
+#include <exception>
+
+class ExitRequestedException : public std::exception {
+public:
+    ExitRequestedException();
+};

Makefile.am

@@ -71,9 +71,9 @@ bin_PROGRAMS = sgxwallet testw sgx_util
 
 
 COMMON_SRC = SGXException.cpp ExitHandler.cpp zmq_src/ZMQClient.cpp zmq_src/RspMessage.cpp zmq_src/ReqMessage.cpp \
-             zmq_src/ZMQMessage.cpp zmq_src/ZMQServer.cpp \
+             zmq_src/ZMQMessage.cpp zmq_src/ZMQServer.cpp zmq_src/Agent.cpp  zmq_src/WorkerThreadPool.cpp ExitRequestedException.cpp \
              InvalidStateException.cpp Exception.cpp InvalidArgumentException.cpp Log.cpp TECrypto.cpp \
-             SGXWalletServer.cpp SGXRegistrationServer.cpp CSRManagerServer.cpp BLSCrypto.cpp CryptoTools.cpp \
+             SGXWalletServer.cpp  SGXRegistrationServer.cpp CSRManagerServer.cpp BLSCrypto.cpp CryptoTools.cpp \
              DKGCrypto.cpp ServerInit.cpp BLSPrivateKeyShareSGX.cpp LevelDB.cpp ServerDataChecker.cpp SEKManager.cpp \
              third_party/intel/sgx_stub.c third_party/intel/sgx_detect_linux.c third_party/intel/create_enclave.c \
              third_party/intel/oc_alloc.c ECDSAImpl.c TestUtils.cpp sgxwallet.c SGXInfoServer.cpp ECDSACrypto.cpp

VERSION

-1.78.0
\ No newline at end of file
+1.83.0
\ No newline at end of file

common.h

@@ -31,16 +31,20 @@ using namespace std;
 #include <iostream>
 #include <map>
 #include <memory>
-
 #include <sys/types.h>
 #include <sys/sysinfo.h>
 #include <string.h>
-
 #include <vector>
-
+#include <json/value.h>
 #include <boost/throw_exception.hpp>
-
 #include <gmp.h>
+#include <thread>
+#include <functional>
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+
+
 #include "secure_enclave/Verify.h"
 #include "InvalidStateException.h"
 #include "SGXException.h"

zmq_src/Agent.cpp

+/*
+    Copyright (C) 2021 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file Agent.cpp
+    @author Stan Kladko
+    @date 2021
+*/
+
+#include "common.h"
+#include "Agent.h"
+
+Agent::Agent() : startedWorkers(false) {};
+
+void Agent::waitOnGlobalStartBarrier() {
+    unique_lock<mutex> lock(startMutex);
+    while (!startedWorkers) {
+        // wait until notified to start running
+        startCond.wait(lock);
+    }
+}
+
+Agent::~Agent() {
+}
+
+
+void Agent::releaseWorkers() {
+
+    if (startedWorkers.exchange(true)) {
+        // already started
+        return;
+    }
+
+    lock_guard<mutex> lock(startMutex);
+
+    startCond.notify_all();
+}

zmq_src/Agent.h

+/*
+    Copyright (C) 2021 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file Agent.h
+    @author Stan Kladko
+    @date 2021
+*/
+
+#pragma once
+
+class Agent {
+
+protected:
+
+    atomic_bool startedWorkers;
+
+    mutex messageMutex;
+    condition_variable messageCond;
+
+    condition_variable startCond;
+    mutex startMutex;
+
+    recursive_mutex m;
+
+
+public:
+
+    Agent();
+
+    virtual ~Agent();
+
+    void releaseWorkers();
+
+    void waitOnGlobalStartBarrier();
+
+    recursive_mutex& getMainMutex() { return m; }
+};

zmq_src/WorkerThreadPool.cpp

+/*
+    Copyright (C) 2021 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file WorkerThreadPool.cpp
+    @author Stan Kladko
+    @date 2021
+*/
+
+#include "document.h"
+#include "stringbuffer.h"
+#include "writer.h"
+
+
+
+#include "common.h"
+#include "sgxwallet_common.h"
+#include "third_party/spdlog/spdlog.h"
+#include "ZMQServer.h"
+#include "WorkerThreadPool.h"
+
+
+WorkerThreadPool::WorkerThreadPool(uint64_t _numThreads, ZMQServer *_agent) : joined(false) {
+    CHECK_STATE(_numThreads > 0);
+    CHECK_STATE(_agent);
+
+    spdlog::info("Creating thread pool. Threads count:" + to_string(_numThreads));
+
+    this->agent = _agent;
+    this->numThreads = _numThreads;;
+
+
+    for (uint64_t i = 0; i < (uint64_t) numThreads; i++) {
+        createThread(i);
+    }
+
+    spdlog::info("Created thread pool");
+
+}
+
+
+void WorkerThreadPool::joinAll() {
+
+    spdlog::info("Joining worker threads ...");
+
+    if (joined.exchange(true))
+        return;
+
+    for (auto &&thread : threadpool) {
+        if (thread->joinable())
+            thread->join();
+        CHECK_STATE(!thread->joinable());
+    }
+
+    spdlog::info("Joined worker threads.");
+}
+
+bool WorkerThreadPool::isJoined() const {
+    return joined;
+}
+
+WorkerThreadPool::~WorkerThreadPool(){
+}
+
+void WorkerThreadPool::createThread(uint64_t _threadNumber) {
+
+    spdlog::info("Starting ZMQ worker thread " + to_string(_threadNumber) );
+
+    this->threadpool.push_back(
+            make_shared< thread >( ZMQServer::workerThreadMessageProcessLoop, agent, _threadNumber ) );
+
+    spdlog::info("Started ZMQ worker thread " + to_string(_threadNumber) );
+}

zmq_src/WorkerThreadPool.h

+/*
+    Copyright (C) 2018-2019 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file WorkerThreadPool.h
+    @author Stan Kladko
+    @date 2018
+*/
+
+#pragma once
+
+class Agent;
+class ZMQServer;
+
+class WorkerThreadPool {
+
+    void createThread( uint64_t threadNumber );
+
+    recursive_mutex m;
+
+protected:
+
+    atomic_bool joined;
+    vector<shared_ptr<thread>> threadpool;
+
+    uint64_t numThreads = 0;
+    ZMQServer* agent = nullptr;
+
+public:
+
+    WorkerThreadPool(uint64_t _numThreads, ZMQServer *_agent);
+
+    virtual ~WorkerThreadPool();
+
+    void joinAll();
+
+    bool isJoined() const;
+
+};

zmq_src/ZMQClient.h

@@ -31,7 +31,6 @@
 #include <openssl/rand.h>
 
 #include "third_party/spdlog/spdlog.h"
-
 #include <zmq.hpp>
 #include "zhelpers.hpp"
 #include <jsonrpccpp/client.h>

zmq_src/ZMQMessage.h

@@ -24,9 +24,6 @@
 #pragma once
 
 
-#include <memory>
-#include <vector>
-
 #include <openssl/pem.h>
 #include <openssl/evp.h>
 #include <openssl/err.h>

zmq_src/ZMQServer.h

@@ -26,30 +26,50 @@
 #define SGXWALLET_ZMQServer_H
 
 
-#include <vector>
-#include <thread>
-#include <memory>
-#include <functional>
-#include <atomic>
+#include "third_party/readerwriterqueue.h"
+#include "third_party/concurrentqueue.h"
+
 
 #include <zmq.hpp>
 #include "zhelpers.hpp"
 
-using namespace std;
+#include "Agent.h"
+#include "WorkerThreadPool.h"
+#include "ZMQMessage.h"
+
+using namespace moodycamel;
+
+static const uint64_t NUM_ZMQ_WORKER_THREADS = 16;
 
 
-class ZMQServer {
+class ZMQServer : public Agent{
 
     uint64_t workerThreads;
 
+    string caCertFile;
+    string caCert;
+
+    ConcurrentQueue<pair<Json::Value, shared_ptr<zmq::message_t>>> outgoingQueue;
+
+    vector<BlockingReaderWriterQueue<pair<shared_ptr<ZMQMessage>, shared_ptr<zmq::message_t>>>> incomingQueue;
+
+    bool checkKeyOwnership = true;
+
+    shared_ptr<zmq::context_t> ctx;
+    shared_ptr<zmq::socket_t> socket;
+
+    static atomic<bool> isExitRequested;
+
+    void doOneServerLoop();
+
 public:
 
     bool checkSignature = false;
-    string caCertFile = "";
-    string caCert = "";
 
     static shared_ptr<ZMQServer> zmqServer;
 
+    shared_ptr<WorkerThreadPool> threadPool = nullptr;
+
     static shared_ptr<std::thread> serverThread;
 
     ZMQServer(bool _checkSignature, bool _checkKeyOwnership, const string& _caCertFile);
@@ -58,18 +78,24 @@ public:
 
     void run();
 
+    void initListenSocket();
+
     static void initZMQServer(bool _checkSignature, bool _checkKeyOwnership);
     static void exitZMQServer();
 
-private:
-    bool checkKeyOwnership = true;
+    static void workerThreadMessageProcessLoop(ZMQServer* agent, uint64_t _threadNumber );
 
-    shared_ptr<zmq::context_t> ctx;
-    shared_ptr<zmq::socket_t> socket;
+    void workerThreadProcessNextMessage(uint64_t _threadNumber);
 
-    static std::atomic<bool> isExitRequested;
+    void checkForExit();
 
-    void doOneServerLoop();
+    void waitForIncomingAndProcessOutgoingMessages();
+
+    pair<string, shared_ptr<zmq::message_t>>  receiveMessage();
+
+    void sendToClient(Json::Value& _result,  shared_ptr<zmq::message_t>& _identity);
+
+    void sendMessagesInOutgoingMessageQueueIfAny();
 
 };