bug/SKALE-3751-enable-zeromq

ServerWorker.cpp

@@ -13,14 +13,14 @@
 
 #include "ServerWorker.h"
 
-std::atomic<uint64_t>  ServerWorker::workerCount(1);
+std::atomic <uint64_t>  ServerWorker::workerCount(1);
 
 ServerWorker::ServerWorker(zmq::context_t &ctx, int sock_type) : ctx_(ctx),
                                                                  worker_(ctx_, sock_type),
                                                                  isExitRequested(false) {
     index = workerCount.fetch_add(1);
     int linger = 0;
-    zmq_setsockopt (worker_, ZMQ_LINGER, &linger, sizeof (linger));
+    zmq_setsockopt(worker_, ZMQ_LINGER, &linger, sizeof(linger));
 };
 
 void ServerWorker::work() {
@@ -42,17 +42,30 @@ void ServerWorker::work() {
         zmq::message_t copied_id;
 
         try {
+
+            zmq_pollitem_t items[1];
+            items[0].socket = worker_;
+            items[0].events = ZMQ_POLLIN;
+
+            int pollResult = 0;
+
+            do {
+                pollResult = zmq_poll(items, 1, 1000);
+                if (isExitRequested) {
+                    goto clean;
+                }
+            } while (pollResult == 0);
+
+
             zmq::message_t msg;
             zmq::message_t copied_msg;
             worker_.recv(&identity);
-
-            cerr << identity.size();
             copied_id.copy(&identity);
             worker_.recv(&msg);
 
             int64_t more;
             size_t more_size = sizeof(more);
-            auto rc = zmq_getsockopt (worker_, ZMQ_RCVMORE, &more, &more_size);
+            auto rc = zmq_getsockopt(worker_, ZMQ_RCVMORE, &more, &more_size);
             CHECK_STATE(rc == 0);
 
 
@@ -60,17 +73,13 @@ void ServerWorker::work() {
 
             memcpy(msgData.data(), msg.data(), msg.size());
 
-
-            cerr << "Received:" << msgData.data();
-
-
             CHECK_STATE(msg.size() > 5 || msgData.at(0) == '{' || msgData[msg.size()] == '}');
 
 
             memcpy(msgData.data(), msg.data(), msg.size());
 
             auto parsedMsg = ZMQMessage::parse(
-                    (const char*) msgData.data(), msg.size(), true);
+                    (const char *) msgData.data(), msg.size(), true);
 
             CHECK_STATE(parsedMsg);
 
@@ -100,7 +109,7 @@ void ServerWorker::work() {
             Json::FastWriter fastWriter;
 
             replyStr = fastWriter.write(result);
-            replyStr = replyStr.substr(0, replyStr.size()  - 1 );
+            replyStr = replyStr.substr(0, replyStr.size() - 1);
 
             CHECK_STATE(replyStr.size() > 2);
             CHECK_STATE(replyStr.front() == '{');
@@ -123,7 +132,7 @@ void ServerWorker::work() {
         }
     }
 
-clean:
+    clean:
     spdlog::info("Exited worker thread {}", index);
 }
 

ZMQClient.cpp

@@ -72,7 +72,7 @@ string ZMQClient::doZmqRequestReply(string &_req) {
         reconnect();
     CHECK_STATE(clientSocket);
 
-    spdlog::info("ZMQ client sending: \n {}" , _req);
+    spdlog::debug("ZMQ client sending: \n {}" , _req);
 
     s_send(*clientSocket, _req);
 
@@ -87,7 +87,7 @@ string ZMQClient::doZmqRequestReply(string &_req) {
 
             CHECK_STATE(reply.size() > 5);
             reply = reply.substr(0, reply.size() - 1);
-            spdlog::info("ZMQ client received reply:{}",  reply);
+            spdlog::debug("ZMQ client received reply:{}",  reply);
             CHECK_STATE(reply.front() == '{');
             CHECK_STATE(reply.back() == '}');
 

ZMQServer.cpp

@@ -45,12 +45,12 @@ ZMQServer::ZMQServer()
 
 void ZMQServer::run() {
 
-    auto port = BASE_PORT + 4;
+    auto port = BASE_PORT + 5;
 
     spdlog::info("Starting zmq server ...");
 
     try {
-        frontend_.bind("tcp://*:" + to_string(BASE_PORT + 5));
+        frontend_.bind("tcp://*:" + to_string(port));
     } catch (...) {
         spdlog::error("Server task could not bind to port:{}", port);
         exit(-100);
@@ -107,10 +107,6 @@ void ZMQServer::exitWorkers() {
 
 
 
-
-
-
-
         spdlog::info("Tell workers to exit");
 
         for (auto &&worker : workers) {
@@ -126,14 +122,20 @@ void ZMQServer::exitWorkers() {
 
         spdlog::info("Closed server sockets");
 
+
         spdlog::info("Terminating context ...");
 
+
         // terminate context (destructor will be called)
         ctx_ = nullptr;
+
         spdlog::info("Terminated context ...");
-    }
+
         spdlog::info("Deleting threads ...");
         worker_threads.empty();
+
+
+    }
     spdlog::info("Deleting workers ...");
     spdlog::info("Deleted workers ...");
 }

ZMQServer.h

@@ -47,7 +47,7 @@ public:
     ZMQServer();
 
     enum {
-        kMaxThread = 1
+        kMaxThread = 16
     };
 
     void run();

testw.cpp

@@ -77,6 +77,7 @@ public:
     }
 
     ~TestFixture() {
+        exitZMQServer();
         TestUtils::destroyEnclave();
     }
 };
@@ -1125,7 +1126,7 @@ TEST_CASE_METHOD(TestFixtureNoReset, "Second run", "[second-run]") {
 
 
 
-TEST_CASE_METHOD(TestFixtureNoReset, "ZMQ-ecdsa", "[zmq-ecdsa-run]") {
+TEST_CASE_METHOD(TestFixture, "ZMQ-ecdsa", "[zmq-ecdsa]") {
 
     HttpClient htp(RPC_ENDPOINT);
     StubClient c(htp, JSONRPC_CLIENT_V2);
@@ -1138,9 +1139,22 @@ TEST_CASE_METHOD(TestFixtureNoReset, "ZMQ-ecdsa", "[zmq-ecdsa-run]") {
 
     PRINT_SRC_LINE
     keyName = genECDSAKeyAPI(c);
+
+
+
+    int end = 100000;
+    string sh = string(SAMPLE_HASH);
+
+
+
+
+
     PRINT_SRC_LINE
-    for (int i = 1; i < 1000; i++) {
-        auto sig = client->ecdsaSignMessageHash(16, keyName, SAMPLE_HASH);
+    for (int i = 1; i < 10000; i++) {
+
+        auto hash = sh.substr(0, sh.size() - 6) + to_string(end + i);
+
+        auto sig = client->ecdsaSignMessageHash(16, keyName, hash);
         REQUIRE(sig.size() > 10);
     }