SKALE-4586 Added Thread Pool

zmq_src/Agent.cpp

@@ -29,13 +29,11 @@ void Agent::notifyAllConditionVariables() {
     queueCond.notify_all();
 }
 
-
-
-Agent::Agent() : startedRun(false) {};
+Agent::Agent() : startedWorkers(false) {};
 
 void Agent::waitOnGlobalStartBarrier() {
     unique_lock<mutex> mlock(queueMutex);
-    while (!startedRun) {
+    while (!startedWorkers) {
         queueCond.wait(mlock);
     }
 }
@@ -44,9 +42,10 @@ Agent::~Agent() {
 }
 
 
-void Agent::releaseGlobalStartBarrier() {
+void Agent::releaseWorkers() {
 
-    if (startedRun.exchange(true)) {
+    if (startedWorkers.exchange(true)) {
+        // already started
         return;
     }
 

zmq_src/Agent.h

@@ -33,7 +33,7 @@ class Agent {
 
 protected:
 
-    atomic_bool startedRun;
+    atomic_bool startedWorkers;
 
     mutex messageMutex;
     condition_variable messageCond;
@@ -52,9 +52,9 @@ public:
 
     virtual ~Agent();
 
-    void releaseGlobalStartBarrier();
+    void releaseWorkers();
 
     void waitOnGlobalStartBarrier();
-    
+
     recursive_mutex& getMainMutex() { return m; }
 };

zmq_src/WorkerThreadPool.cpp

@@ -23,15 +23,16 @@
 
 #include "common.h"
 #include "third_party/spdlog/spdlog.h"
+#include "ZMQServer.h"
 #include "WorkerThreadPool.h"
 
 
 void WorkerThreadPool::startService() {
 
-    lock_guard<recursive_mutex> lock(threadPoolMutex);
-
     CHECK_STATE(!started.exchange(true))
 
+    LOCK(m)
+
     for (uint64_t i = 0; i < (uint64_t) numThreads; i++) {
         createThread(i);
     }
@@ -39,7 +40,7 @@ void WorkerThreadPool::startService() {
 }
 
 
-WorkerThreadPool::WorkerThreadPool(uint64_t _numThreads, Agent *_agent) : started(false), joined(false) {
+WorkerThreadPool::WorkerThreadPool(uint64_t _numThreads, ZMQServer *_agent) : started(false), joined(false) {
     CHECK_STATE(_numThreads > 0);
     CHECK_STATE(_agent);
     spdlog::info("Started thread pool. Threads count:" + to_string(_numThreads));
@@ -52,7 +53,7 @@ void WorkerThreadPool::joinAll() {
     if (joined)
         return;
 
-    lock_guard<recursive_mutex> lock(threadPoolMutex);
+    LOCK(m);
 
     joined = true;
 
@@ -69,3 +70,13 @@ bool WorkerThreadPool::isJoined() const {
 
 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 ) );
+
+    spdlog::info("Started ZMQ worker thread " + to_string(_threadNumber) );
+}

zmq_src/WorkerThreadPool.h

@@ -27,30 +27,30 @@
 #include <atomic>
 #include <thread>
 
-#include "Agent.h"
+
+class Agent;
+class ZMQServer;
 
 class WorkerThreadPool {
 
     atomic_bool started;
 
-    virtual void createThread( uint64_t threadNumber ) = 0;
+    void createThread( uint64_t threadNumber );
+
+    recursive_mutex m;
 
 protected:
 
     atomic_bool joined;
-
     vector<shared_ptr<thread>> threadpool;
 
-    recursive_mutex threadPoolMutex;
     uint64_t numThreads = 0;
-    Agent* agent = nullptr;
-
-protected:
-
-    WorkerThreadPool(uint64_t _numThreads, Agent *_agent);
+    ZMQServer* agent = nullptr;
 
 public:
 
+    WorkerThreadPool(uint64_t _numThreads, ZMQServer *_agent);
+
     virtual ~WorkerThreadPool();
 
     virtual void startService();
@@ -59,5 +59,4 @@ public:
 
     bool isJoined() const;
 
-
 };

zmq_src/ZMQServer.cpp

@@ -55,6 +55,9 @@ ZMQServer::ZMQServer(bool _checkSignature, bool _checkKeyOwnership, const string
     int linger = 0;
 
     zmq_setsockopt(*socket, ZMQ_LINGER, &linger, sizeof(linger));
+
+    threadPool = make_shared<WorkerThreadPool>(1, this);
+
 }
 
 void ZMQServer::run() {
@@ -73,7 +76,7 @@ void ZMQServer::run() {
 
     spdlog::info("Bound port ...");
 
-    waitOnGlobalStartBarrier();
+
 
     spdlog::info("Started zmq read loop ...");
 
@@ -123,11 +126,13 @@ void ZMQServer::initZMQServer(bool _checkSignature, bool _checkKeyOwnership) {
     serverThread = make_shared<thread>(std::bind(&ZMQServer::run, ZMQServer::zmqServer));
     serverThread->detach();
 
+    zmqServer->releaseWorkers();
+
     spdlog::info("Inited zmq server.");
 
     spdlog::info("Starting zmq server ...");
 
-    zmqServer->releaseGlobalStartBarrier();
+    zmqServer->releaseWorkers();
 
     spdlog::info("Started zmq server.");
 
@@ -254,3 +259,12 @@ void ZMQServer::doOneServerLoop() {
         exit(-18);
     }
 }
+
+
+void ZMQServer::workerThreadMessageProcessLoop(ZMQServer* _agent ) {
+    CHECK_STATE(_agent);
+    _agent->waitOnGlobalStartBarrier();
+    while (!isExitRequested) {
+        sleep(100);
+    }
+}
\ No newline at end of file

zmq_src/ZMQServer.h

@@ -36,22 +36,36 @@
 #include "zhelpers.hpp"
 
 #include "Agent.h"
+#include "WorkerThreadPool.h"
 
 using namespace std;
 
 
-class ZMQServer : Agent{
+class ZMQServer : public Agent{
 
     uint64_t workerThreads;
 
+    string caCertFile;
+    string caCert;
+
+    bool checkKeyOwnership = true;
+
+    shared_ptr<zmq::context_t> ctx;
+    shared_ptr<zmq::socket_t> socket;
+
+    static std::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);
@@ -63,15 +77,7 @@ public:
     static void initZMQServer(bool _checkSignature, bool _checkKeyOwnership);
     static void exitZMQServer();
 
-private:
-    bool checkKeyOwnership = true;
-
-    shared_ptr<zmq::context_t> ctx;
-    shared_ptr<zmq::socket_t> socket;
-
-    static std::atomic<bool> isExitRequested;
-
-    void doOneServerLoop();
+    static void workerThreadMessageProcessLoop(ZMQServer* agent );
 
 };