bug/SKALE-3662 Adding libzmq

ServerInit.cpp

@@ -105,6 +105,7 @@ void initUserSpace() {
 
     zmqServer = new ZMQServer();
     static std::thread serverThread(std::bind(&ZMQServer::run, zmqServer));
+    serverThread.detach();
 }
 
 void exitZMQServer() {

ServerWorker.cpp

@@ -13,10 +13,15 @@
 
 #include "ServerWorker.h"
 
+std::atomic<uint64_t>  ServerWorker::workerCount(1);
 
 ServerWorker::ServerWorker(zmq::context_t &ctx, int sock_type) : ctx_(ctx),
                                                                  worker_(ctx_, sock_type),
-                                                                 isExitRequested(false) {};
+                                                                 isExitRequested(false) {
+    index = workerCount.fetch_add(1);
+    int linger = 0;
+    zmq_setsockopt (worker_, ZMQ_LINGER, &linger, sizeof (linger));
+};
 
 void ServerWorker::work() {
     worker_.connect("inproc://backend");
@@ -84,7 +89,7 @@ void ServerWorker::work() {
             spdlog::error("Exception in zmq server worker:{}", e.what());
         } catch (...) {
             if (isExitRequested) {
-                return;
+                goto clean;
             }
             spdlog::error("Error in zmq server worker");
             result["errorMessage"] = "Error in zmq server worker";
@@ -107,21 +112,25 @@ void ServerWorker::work() {
 
         } catch (std::exception &e) {
             if (isExitRequested) {
-                return;
+                goto clean;
             }
             spdlog::error("Exception in zmq server worker send :{}", e.what());
         } catch (...) {
             if (isExitRequested) {
-                return;
+                goto clean;
             }
             spdlog::error("Unklnown exception in zmq server worker send");
         }
-
     }
+
+clean:
+    spdlog::info("Exited worker thread {}", index);
 }
 
 
 void ServerWorker::requestExit() {
     isExitRequested.exchange(true);
+    zmq_close(worker_);
+    spdlog::info("Closed worker socket {}", index);
 }
 

ServerWorker.h

@@ -32,6 +32,9 @@ private:
     zmq::socket_t worker_;
 
     std::atomic<bool> isExitRequested;
+
+    static std::atomic<uint64_t> workerCount;
+    uint64_t index;
 };
 
 

ZMQClient.cpp

@@ -146,6 +146,5 @@ string ZMQClient::ecdsaSignMessageHash(int base, const std::string &keyName, con
     auto result = dynamic_pointer_cast<ECDSASignRspMessage>(doRequestReply(p));
     CHECK_STATE(result);
     CHECK_STATE(result->getStatus() == 0);
-    sleep(5);
     return result->getSignature();
 }
\ No newline at end of file

ZMQMessage.h

@@ -57,7 +57,7 @@ public:
     uint64_t getUint64Rapid(const char *_name);
 
     uint64_t getStatus() {
-        getUint64Rapid("status");
+        return getUint64Rapid("status");
     }
 
     static shared_ptr<ZMQMessage> parse(vector<uint8_t> &_msg, bool _isRequest);

ZMQServer.cpp

@@ -34,6 +34,12 @@ ZMQServer::ZMQServer()
         : isExitRequested(false), ctx_(make_shared<zmq::context_t>(1)),
           frontend_(*ctx_, ZMQ_ROUTER),
           backend_(*ctx_, ZMQ_DEALER) {
+
+
+    int linger = 0;
+    zmq_setsockopt (frontend_, ZMQ_LINGER, &linger, sizeof (linger));
+    zmq_setsockopt (backend_, ZMQ_LINGER, &linger, sizeof (linger));
+
 }
 
 
@@ -65,7 +71,9 @@ void ZMQServer::run() {
     try {
         for (int i = 0; i < kMaxThread; ++i) {
             workers.push_back(make_shared<ServerWorker>(*ctx_, ZMQ_DEALER));
-            worker_threads.push_back(make_shared<std::thread>(std::bind(&ServerWorker::work, workers[i])));
+            auto th = make_shared<std::thread>(std::bind(&ServerWorker::work, workers[i]));
+            th->detach();
+            worker_threads.push_back(th);
         }
     } catch (std::exception &e) {
         spdlog::error("Could not create zmq server workers:{} ", e.what());
@@ -97,28 +105,36 @@ void ZMQServer::exitWorkers() {
     auto doExit = !isExitRequested.exchange(true);
     if (doExit) {
 
+
+
+
+
+
+
         spdlog::info("Tell workers to exit");
 
         for (auto &&worker : workers) {
             worker->requestExit();
         }
 
+        // close server sockets
+
+        spdlog::info("Closing server sockets  ...");
+
+        zmq_close(frontend_);
+        zmq_close(backend_);
+
+        spdlog::info("Closed server sockets");
+
         spdlog::info("Terminating context ...");
 
         // terminate context (destructor will be called)
         ctx_ = nullptr;
-
         spdlog::info("Terminated context ...");
-
-
-        spdlog::info("joining threads ...");
-        for (auto &&thread : worker_threads)
-            thread->join();
     }
     spdlog::info("Deleting threads ...");
     worker_threads.empty();
     spdlog::info("Deleting workers ...");
-    workers.empty();
     spdlog::info("Deleted workers ...");
 }
 

testw.cpp

@@ -1136,30 +1136,16 @@ TEST_CASE_METHOD(TestFixtureNoReset, "ZMQ-ecdsa", "[zmq-ecdsa-run]") {
 
     string keyName = "";
 
-    try {
-        PRINT_SRC_LINE
-        keyName = genECDSAKeyAPI(c);
-        PRINT_SRC_LINE
-    } catch (Exception & e)
-    {
-        cerr << e.what() << endl;
-        throw;
-    }
-
-
-
-    try {
-        PRINT_SRC_LINE
+    PRINT_SRC_LINE
+    keyName = genECDSAKeyAPI(c);
+    PRINT_SRC_LINE
+    for (int i = 1; i < 1000; i++) {
         auto sig = client->ecdsaSignMessageHash(16, keyName, SAMPLE_HASH);
         REQUIRE(sig.size() > 10);
+    }
 
-        cerr << sig << endl;
-
-    } catch (...) {
+    PRINT_SRC_LINE
 
-        client = nullptr;
-        sleep(10000);
-    }
 }