bug/SKALE-3751-enable-zeromq

ZMQClient.cpp

@@ -22,6 +22,9 @@
 */
 
 #include "sys/random.h"
+#include <sys/types.h>
+#include <sys/syscall.h>
+
 
 #include "common.h"
 #include "BLSSignReqMessage.h"
@@ -62,14 +65,25 @@ shared_ptr <ZMQMessage> ZMQClient::doRequestReply(Json::Value &_req) {
         throw;
     }
 
+
+
 }
 
+
+
 string ZMQClient::doZmqRequestReply(string &_req) {
 
     stringstream request;
 
-    if (!clientSocket)
+    shared_ptr <zmq::socket_t>  clientSocket = nullptr;
+
+    {
+        lock_guard <recursive_mutex> m(mutex);
+        if (!clientSockets.count(getProcessID()))
             reconnect();
+        clientSocket = clientSockets.at(getProcessID());
+        CHECK_STATE(clientSocket);
+    }
     CHECK_STATE(clientSocket);
 
     spdlog::debug("ZMQ client sending: \n {}" , _req);
@@ -108,18 +122,24 @@ ZMQClient::ZMQClient(string &ip, uint16_t port) : ctx(1) {
 
 void ZMQClient::reconnect() {
 
-    getrandom(identity, 10, 0);
+    lock_guard<recursive_mutex> lock(mutex);
 
+    auto pid = getProcessID();
+
+    if (clientSockets.count(pid) > 0) {
+        clientSockets.erase(pid);
+    }
 
 
-    clientSocket = nullptr; // delete previous
-    clientSocket = make_unique<zmq::socket_t>(ctx, ZMQ_DEALER);
+    char identity[10];
+    getrandom(identity, 10, 0);
+    auto clientSocket = make_shared<zmq::socket_t>(ctx, ZMQ_DEALER);
     clientSocket->setsockopt(ZMQ_IDENTITY, identity, 10);
     //  Configure socket to not wait at close time
     int linger = 0;
     clientSocket->setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
     clientSocket->connect(url);
-
+    clientSockets.insert({pid, clientSocket});
 }
 
 
@@ -148,3 +168,9 @@ string ZMQClient::ecdsaSignMessageHash(int base, const std::string &keyName, con
     CHECK_STATE(result->getStatus() == 0);
     return result->getSignature();
 }
+
+
+
+uint64_t ZMQClient::getProcessID() {
+    return  syscall(__NR_gettid);
+}
\ No newline at end of file

ZMQClient.h

@@ -44,23 +44,29 @@ class ZMQClient {
 
 private:
 
+
+
+    recursive_mutex mutex;
+
     zmq::context_t ctx;
-    std::unique_ptr <zmq::socket_t> clientSocket;
+
     string url;
 
     // generate random identity
-    char identity[10] = {};
+
+    map<uint64_t , shared_ptr <zmq::socket_t>> clientSockets;
 
     shared_ptr <ZMQMessage> doRequestReply(Json::Value &_req);
 
     string doZmqRequestReply(string &_req);
 
+    uint64_t getProcessID();
+
 
 public:
 
 
     ZMQClient(string &ip, uint16_t port);
-
     void reconnect() ;
 
     string blsSignMessageHash(const std::string &keyShareName, const std::string &messageHash, int t, int n);

ZMQServer.h

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

testw.cpp

@@ -381,7 +381,7 @@ TEST_CASE_METHOD(TestFixture, "DKG AES encrypted secret shares test", "[dkg-aes-
 
     vector<char> s_shareG2(BUF_LEN, 0);
     PRINT_SRC_LINE
-    status = trustedGetEncryptedSecretShare(eid, &errStatus,errMsg.data(),
+    status = trustedGetEncryptedSecretShare(eid, &errStatus, errMsg.data(),
                                             encryptedDKGSecret.data(), encLen,
                                             encrPRDHKey.data(), &encLen,
                                             result.data(),
@@ -411,7 +411,7 @@ TEST_CASE_METHOD(TestFixture, "DKG AES encrypted secret shares version 2 test",
 
     vector<char> s_shareG2(BUF_LEN, 0);
     PRINT_SRC_LINE
-    status = trustedGetEncryptedSecretShareV2(eid, &errStatus,errMsg.data(),
+    status = trustedGetEncryptedSecretShareV2(eid, &errStatus, errMsg.data(),
                                               encryptedDKGSecret.data(), encLen,
                                               encrPRDHKey.data(), &encLen,
                                               result.data(),
@@ -510,7 +510,8 @@ TEST_CASE_METHOD(TestFixture, "Import ECDSA Key", "[import-ecdsa-key]") {
     StubClient c(client, JSONRPC_CLIENT_V2);
 
     std::string name = "NEK:abcdef";
-    auto response = c.importECDSAKey("6507625568967977077291849236396320012317305261598035438182864059942098934847", name);
+    auto response = c.importECDSAKey("6507625568967977077291849236396320012317305261598035438182864059942098934847",
+                                     name);
     REQUIRE(response["status"] != 0);
 
     string key_str = "0xe632f7fde2c90a073ec43eaa90dca7b82476bf28815450a11191484934b9c3f";
@@ -781,7 +782,7 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG test", "[aes-dkg]") {
     string shareG2 = complaintResponse["share*G2"].asString();
     string secretShare = secretShares[1]["secretShare"].asString().substr(0, 192);
 
-    vector<char> message (65, 0);
+    vector<char> message(65, 0);
 
     SAFE_CHAR_BUF(encr_sshare, BUF_LEN)
     strncpy(encr_sshare, pubEthKeys[0].asString().c_str(), 128);
@@ -804,7 +805,7 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG test", "[aes-dkg]") {
 
     mpz_clear(hex_share);
 
-    REQUIRE( convertG2ToString(decrypted_share_G2) == shareG2 );
+    REQUIRE(convertG2ToString(decrypted_share_G2) == shareG2);
 
     Json::Value verificationVectorMult = complaintResponse["verificationVectorMult"];
 
@@ -819,13 +820,14 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG test", "[aes-dkg]") {
         verificationValue = verificationValue + value;
     }
     verificationValue.to_affine_coordinates();
-    REQUIRE( verificationValue == decrypted_share_G2 );
+    REQUIRE(verificationValue == decrypted_share_G2);
 
     BLSSigShareSet sigShareSet(t, n);
 
     string hash = SAMPLE_HASH;
 
-    auto hash_arr = make_shared < array < uint8_t, 32 > >();
+    auto hash_arr = make_shared < array < uint8_t,
+    32 > > ();
 
     uint64_t binLen;
 
@@ -926,7 +928,8 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG V2 test", "[aes-dkg-v2]") {
             string secretShare = secretShares[i]["secretShare"].asString().substr(192 * j, 192);
             secShares[i] += secretShares[j]["secretShare"].asString().substr(192 * i, 192);
             PRINT_SRC_LINE
-            Json::Value verif = c.dkgVerificationV2(pubShares[i], ethKeys[j]["keyName"].asString(), secretShare, t, n, j);
+            Json::Value verif = c.dkgVerificationV2(pubShares[i], ethKeys[j]["keyName"].asString(), secretShare, t, n,
+                                                    j);
             REQUIRE(verif["status"] == 0);
             bool res = verif["result"].asBool();
             k++;
@@ -940,7 +943,7 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG V2 test", "[aes-dkg-v2]") {
     string shareG2 = complaintResponse["share*G2"].asString();
     string secretShare = secretShares[1]["secretShare"].asString().substr(0, 192);
 
-    vector<char> message (65, 0);
+    vector<char> message(65, 0);
 
     SAFE_CHAR_BUF(encr_sshare, BUF_LEN)
     strncpy(encr_sshare, pubEthKeys[0].asString().c_str(), 128);
@@ -953,7 +956,7 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG V2 test", "[aes-dkg-v2]") {
     SAFE_CHAR_BUF(derived_key, 33)
 
     uint64_t key_length;
-    REQUIRE(hex2carray(&hashed_key[0], &key_length, (uint8_t*) derived_key, 33));
+    REQUIRE(hex2carray(&hashed_key[0], &key_length, (uint8_t *) derived_key, 33));
 
     SAFE_CHAR_BUF(encr_sshare_check, BUF_LEN)
     strncpy(encr_sshare_check, secretShare.c_str(), ECDSA_SKEY_LEN - 1);
@@ -970,7 +973,7 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG V2 test", "[aes-dkg-v2]") {
 
     mpz_clear(hex_share);
 
-    REQUIRE( convertG2ToString(decrypted_share_G2) == shareG2 );
+    REQUIRE(convertG2ToString(decrypted_share_G2) == shareG2);
 
     Json::Value verificationVectorMult = complaintResponse["verificationVectorMult"];
 
@@ -985,13 +988,14 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG V2 test", "[aes-dkg-v2]") {
         verificationValue = verificationValue + value;
     }
     verificationValue.to_affine_coordinates();
-    REQUIRE( verificationValue == decrypted_share_G2 );
+    REQUIRE(verificationValue == decrypted_share_G2);
 
     BLSSigShareSet sigShareSet(t, n);
 
     string hash = SAMPLE_HASH;
 
-    auto hash_arr = make_shared < array < uint8_t, 32 > >();
+    auto hash_arr = make_shared < array < uint8_t,
+    32 > > ();
 
     uint64_t binLen;
 
@@ -1004,7 +1008,8 @@ TEST_CASE_METHOD(TestFixture, "AES_DKG V2 test", "[aes-dkg-v2]") {
     for (int i = 0; i < t; i++) {
         string endName = polyNames[i].substr(4);
         string blsName = "BLS_KEY" + polyNames[i].substr(4);
-        auto response = c.createBLSPrivateKeyV2(blsName, ethKeys[i]["keyName"].asString(), polyNames[i], secShares[i], t,
+        auto response = c.createBLSPrivateKeyV2(blsName, ethKeys[i]["keyName"].asString(), polyNames[i], secShares[i],
+                                                t,
                                                 n);
         REQUIRE(response["status"] == 0);
 
@@ -1095,8 +1100,7 @@ TEST_CASE_METHOD(TestFixture, "First run", "[first-run]") {
         namefile << keyName;
 
         PRINT_SRC_LINE
-    } catch (JsonRpcException & e)
-    {
+    } catch (JsonRpcException &e) {
         cerr << e.what() << endl;
         throw;
     }
@@ -1125,7 +1129,6 @@ TEST_CASE_METHOD(TestFixtureNoReset, "Second run", "[second-run]") {
 }
 
 
-
 TEST_CASE_METHOD(TestFixture, "ZMQ-ecdsa", "[zmq-ecdsa]") {
 
     HttpClient htp(RPC_ENDPOINT);
@@ -1139,25 +1142,31 @@ TEST_CASE_METHOD(TestFixture, "ZMQ-ecdsa", "[zmq-ecdsa]") {
 
     PRINT_SRC_LINE
     keyName = genECDSAKeyAPI(c);
-
-
-
-    int end = 100000;
+    int end = 10000000;
     string sh = string(SAMPLE_HASH);
 
 
-
+    std::vector <std::thread> workers;
 
 
     PRINT_SRC_LINE
-    for (int i = 1; i < 10000; i++) {
 
-        auto hash = sh.substr(0, sh.size() - 6) + to_string(end + i);
+    for (int j = 0; j < 10; j++) {
+        workers.push_back(std::thread([client, sh, keyName, end, j]() {
+            CHECK_STATE(client);
+            for (int i = (j * 2000); i < (j * 2000) + 1000; i++) {
+
+                auto hash = sh.substr(0, sh.size() - 8) + to_string(end + i);
 
                 auto sig = client->ecdsaSignMessageHash(16, keyName, hash);
                 REQUIRE(sig.size() > 10);
             }
 
+        }));
+    };
+
+    std::for_each(workers.begin(), workers.end(), [](
+            std::thread &t) { t.join(); });
     PRINT_SRC_LINE
 
 }