Skip to content

S
sgxwallet
Unverified Commit e0482d51 authored by Oleh Nikolaiev's avatar Oleh Nikolaiev
Browse files
Options

Merge branch 'develop' into feature/SKALE-3951-add-zeromq

parents 6f45b506 5e7570d5
Hide whitespace changes
Inline Side-by-side
Showing with 143 additions and 326 deletions
Makefile.am
View file @ e0482d51
......@@ -71,7 +71,7 @@ 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/ServerWorker.cpp \
zmq_src/ZMQMessage.cpp zmq_src/ZMQServer.cpp \
InvalidStateException.cpp Exception.cpp InvalidArgumentException.cpp Log.cpp \
SGXWalletServer.cpp SGXRegistrationServer.cpp CSRManagerServer.cpp BLSCrypto.cpp \
DKGCrypto.cpp ServerInit.cpp BLSPrivateKeyShareSGX.cpp LevelDB.cpp ServerDataChecker.cpp SEKManager.cpp \
......
scripts/install_packages.sh
View file @ e0482d51
#!/bin/bash
sudo apt install -y build-essential make gcc g++ yasm python libprotobuf10 flex bison automake
sudo apt install -y ccache cmake ccache autoconf texinfo libgcrypt20-dev libgnutls28-dev libtool pkg-config
sudo apt install -y ocaml ocaml-build
\ No newline at end of file
zmq_src/ServerWorker.cpp deleted 100644 → 0
View file @ 6f45b506
/*
Copyright (C) 2019-Present SKALE Labs
This file is part of sgxwallet.
sgxwallet 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.
sgxwallet 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 sgxwallet. If not, see <https://www.gnu.org/licenses/>.
@file ServerWorker.cpp
@author Stan Kladko
@date 2021
*/
#include "common.h"
#include "sgxwallet_common.h"
#include <json/writer.h>
#include <zmq.hpp>
#include "zhelpers.hpp"
#include "Log.h"
#include "ZMQMessage.h"
#include "ServerWorker.h"
std::atomic <uint64_t> ServerWorker::workerCount(1);
ServerWorker::ServerWorker(zmq::context_t &_ctx, int sock_type, bool _checkSignature,
const string& _caCert ) : checkSignature(_checkSignature),
caCert(_caCert), isExitRequested(false) {
worker = make_shared<zmq::socket_t>(_ctx, sock_type);
if (checkSignature) {
CHECK_STATE(!caCert.empty())
}
index = workerCount.fetch_add(1);
int linger = 0;
zmq_setsockopt(*worker, ZMQ_LINGER, &linger, sizeof(linger));
};
void ServerWorker::doOneServerLoop() noexcept {
string replyStr;
Json::Value result;
result["status"] = ZMQ_SERVER_ERROR;
result["errorMessage"] = "";
zmq::message_t identity;
zmq::message_t identit2;
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) {
return;
}
} while (pollResult == 0);
worker->recv(&identity);
copied_id.copy(&identity);
string stringToParse = s_recv(*worker);
auto parsedMsg = ZMQMessage::parse(
stringToParse.c_str(), stringToParse.size(), true, checkSignature);
CHECK_STATE2(parsedMsg, ZMQ_COULD_NOT_PARSE);
result = parsedMsg->process();
} catch (SGXException &e) {
result["status"] = e.getStatus();
result["errorMessage"] = e.what();
spdlog::error("Exception in zmq server worker:{}", e.what());
}
catch (std::exception &e) {
if (isExitRequested) {
return;
}
result["errorMessage"] = string(e.what());
spdlog::error("Exception in zmq server worker:{}", e.what());
} catch (...) {
if (isExitRequested) {
return;
}
spdlog::error("Error in zmq server worker");
result["errorMessage"] = "Error in zmq server worker";
}
try {
Json::FastWriter fastWriter;
fastWriter.omitEndingLineFeed();
replyStr = fastWriter.write(result);
CHECK_STATE(replyStr.size() > 2);
CHECK_STATE(replyStr.front() == '{');
CHECK_STATE(replyStr.back() == '}');
worker->send(copied_id, ZMQ_SNDMORE);
s_send(*worker, replyStr);
} catch (std::exception &e) {
if (isExitRequested) {
return;
}
spdlog::error("Exception in zmq server worker send :{}", e.what());
} catch (...) {
if (isExitRequested) {
return;
}
spdlog::error("Unklnown exception in zmq server worker send");
}
}
void ServerWorker::work() {
worker->connect("inproc://backend");
while (!isExitRequested) {
try {
doOneServerLoop();
} catch (...) {
spdlog::error("doOneServerLoop threw exception. This should never happen!");
}
}
spdlog::info("Exited worker thread {}", index);
}
void ServerWorker::requestExit() {
isExitRequested.exchange(true);
spdlog::info("Closed worker socket {}", index);
}
zmq_src/ServerWorker.h deleted 100644 → 0
View file @ 6f45b506
/*
Copyright (C) 2019-Present SKALE Labs
This file is part of sgxwallet.
sgxwallet 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.
sgxwallet 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 sgxwallet. If not, see <https://www.gnu.org/licenses/>.
@file ServerWorker.h
@author Stan Kladko
@date 2021
*/
#ifndef SGXWALLET_SERVERWORKER_H
#define SGXWALLET_SERVERWORKER_H
#include <vector>
#include <thread>
#include <memory>
#include <functional>
#include "abstractstubserver.h"
#include <zmq.hpp>
#include "zhelpers.hpp"
#include "third_party/spdlog/spdlog.h"
#include "document.h"
class ServerWorker {
bool checkSignature = true;
string caCert = "";
public:
ServerWorker(zmq::context_t &ctx, int sock_type, bool _checkSignature, const string& _caCert );
void work();
void requestExit();
private:
shared_ptr<zmq::socket_t> worker;
std::atomic<bool> isExitRequested;
void doOneServerLoop() noexcept;
static std::atomic<uint64_t> workerCount;
uint64_t index;
};
#endif //SGXWALLET_SERVERWORKER_H
zmq_src/ZMQServer.cpp
View file @ e0482d51
......@@ -30,6 +30,7 @@
#include "common.h"
#include "SGXException.h"
#include "ZMQMessage.h"
#include "ZMQServer.h"
#include "sgxwallet_common.h"
......@@ -39,14 +40,9 @@ shared_ptr <ZMQServer> ZMQServer::zmqServer = nullptr;
ZMQServer::ZMQServer(bool _checkSignature, const string &_caCertFile)
: checkSignature(_checkSignature),
caCertFile(_caCertFile), ctx_(make_shared<zmq::context_t>(1)) {
caCertFile(_caCertFile), ctx(make_shared<zmq::context_t>(1)) {
frontend = make_shared<zmq::socket_t>(*ctx_, ZMQ_ROUTER);
backend = make_shared<zmq::socket_t>(*ctx_, ZMQ_DEALER);
//workerThreads = 2 * thread::hardware_concurrency();
workerThreads = 1; // do one thread for now
socket = make_shared<zmq::socket_t>(*ctx, ZMQ_ROUTER);
if (_checkSignature) {
CHECK_STATE(!_caCertFile.empty());
......@@ -58,9 +54,7 @@ ZMQServer::ZMQServer(bool _checkSignature, const string &_caCertFile)
int linger = 0;
zmq_setsockopt(*frontend, ZMQ_LINGER, &linger, sizeof(linger));
zmq_setsockopt(*backend, ZMQ_LINGER, &linger, sizeof(linger));
zmq_setsockopt(*socket, ZMQ_LINGER, &linger, sizeof(linger));
}
void ZMQServer::run() {
......@@ -70,8 +64,8 @@ void ZMQServer::run() {
spdlog::info("Starting zmq server on port {} ...", port);
try {
CHECK_STATE(frontend);
frontend->bind("tcp://*:" + to_string(port));
CHECK_STATE(socket);
socket->bind("tcp://*:" + to_string(port));
} catch (...) {
spdlog::error("Server task could not bind to port:{}", port);
throw SGXException(ZMQ_COULD_NOT_BIND_FRONT_END, "Server task could not bind.");
......@@ -79,82 +73,29 @@ void ZMQServer::run() {
spdlog::info("Bound port ...");
try {
CHECK_STATE(backend);
backend->bind("inproc://backend");
} catch (exception &e) {
spdlog::error("Could not bind to zmq backend: {}", e.what());
throw SGXException(ZMQ_COULD_NOT_BIND_BACK_END, "Could not bind to zmq backend.");
}
spdlog::info("Creating {} zmq server workers ...", workerThreads);
try {
for (int i = 0; i < workerThreads; ++i) {
workers.push_back(make_shared<ServerWorker>(*ctx_, ZMQ_DEALER,
this->checkSignature, this->caCert));
auto th = make_shared<std::thread>(std::bind(&ServerWorker::work, workers[i]));
worker_threads.push_back(th);
}
} catch (std::exception &e) {
spdlog::error("Could not create zmq server workers:{} ", e.what());
throw SGXException(ZMQ_COULD_NOT_CREATE_WORKERS, "Could not create zmq server workers.");
};
spdlog::info("Created {} zmq server workers ...", workerThreads);
spdlog::info("Creating zmq proxy.");
try {
zmq::proxy(static_cast<void *>(*frontend), static_cast<void *>(*backend), nullptr);
spdlog::info("Exited zmq proxy");
} catch (exception &_e) {
if (isExitRequested) {
spdlog::info("Exited ZMQServer main thread");
return;
}
spdlog::info("Error, exiting zmq server ... {}", _e.what());
return;
} catch (...) {
if (isExitRequested) {
spdlog::info("Exited ZMQServer main thread");
return;
while (!isExitRequested) {
try {
zmqServer->doOneServerLoop();
} catch (...) {
spdlog::error("doOneServerLoop threw exception. This should never happen!");
}
spdlog::info("Error, exiting zmq server ...");
throw SGXException(ZMQ_COULD_NOT_CREATE_PROXY, "Error, exiting zmq server.");
}
}
void ZMQServer::exitAll() {
spdlog::info("Exiting zmq server workers ...");
for (auto &&worker : workers) {
worker->requestExit();
}
for (auto &&workerThread : worker_threads) {
workerThread->join();
}
spdlog::info("Exited zmq server workers ...");
spdlog::info("Exited zmq server loop");
}
std::atomic<bool> ZMQServer::isExitRequested(false);
void ZMQServer::exitZMQServer() {
auto doExit = !isExitRequested.exchange(true);
if (doExit) {
zmqServer->exitAll();
spdlog::info("deleting zmq server");
zmqServer = nullptr;
spdlog::info("deleted zmq server ");
}
isExitRequested.exchange(true);
zmqServer->ctx->shutdown();
zmqServer->socket->close();
zmqServer->ctx->close();
spdlog::info("Exited zmq server.");
}
void ZMQServer::initZMQServer(bool _checkSignature) {
static bool initedServer = false;
CHECK_STATE(!initedServer)
......@@ -165,13 +106,15 @@ void ZMQServer::initZMQServer(bool _checkSignature) {
string rootCAPath = "";
if (_checkSignature) {
rootCAPath = string(SGXDATA_FOLDER) + "cert_data/rootCA.pem";
spdlog::info("Reading root CA from {}", rootCAPath);
CHECK_STATE(access(rootCAPath.c_str(), F_OK) == 0);
};
zmqServer = make_shared<ZMQServer>(_checkSignature, rootCAPath);
CHECK_STATE(zmqServer)
serverThread = make_shared<thread>(std::bind(&ZMQServer::run, ZMQServer::zmqServer));
serverThread->detach();
......@@ -180,27 +123,125 @@ void ZMQServer::initZMQServer(bool _checkSignature) {
shared_ptr <std::thread> ZMQServer::serverThread = nullptr;
ZMQServer::~ZMQServer() {
ZMQServer::~ZMQServer() {}
void ZMQServer::doOneServerLoop() {
string replyStr;
Json::Value result;
result["status"] = ZMQ_SERVER_ERROR;
result["errorMessage"] = "";
zmq::message_t identity;
zmq::message_t identit2;
zmq::message_t copied_id;
spdlog::info("Deleting worker threads");
worker_threads.clear();
spdlog::info("Deleted worker threads");
spdlog::info("Deleting workers ...");
workers.clear();
spdlog::info("Deleted workers ...");
string stringToParse = "";
spdlog::info("Deleting front end and back end");
frontend = nullptr;
backend = nullptr;
spdlog::info("Deleted front end and back end");
try {
zmq_pollitem_t items[1];
items[0].socket = *socket;
items[0].events = ZMQ_POLLIN;
int pollResult = 0;
do {
pollResult = zmq_poll(items, 1, 1000);
if (isExitRequested) {
return;
}
} while (pollResult == 0);
if (!socket->recv(&identity)) {
// something terrible happened
spdlog::error("Fatal error: socket->recv(&identity) returned false");
exit(-11);
}
spdlog::info("Deleting server thread");
ZMQServer::serverThread = nullptr;
spdlog::info("Deleted server thread");
if (!identity.more()) {
// something terrible happened
spdlog::error("Fatal error: zmq_msg_more(identity) returned false");
exit(-12);
}
copied_id.copy(&identity);
zmq::message_t reqMsg;
if (!socket->recv(&reqMsg, 0)) {
// something terrible happened
spdlog::error("Fatal error: socket.recv(&reqMsg, 0) returned false");
exit(-13);
}
stringToParse = string((char *) reqMsg.data(), reqMsg.size());
spdlog::info("Deleting ZMQ context");
ctx_ = nullptr;
spdlog::info("Deleted ZMQ context");
CHECK_STATE(stringToParse.front() == '{')
CHECK_STATE(stringToParse.back() == '}')
auto parsedMsg = ZMQMessage::parse(
stringToParse.c_str(), stringToParse.size(), true, checkSignature);
CHECK_STATE2(parsedMsg, ZMQ_COULD_NOT_PARSE);
result = parsedMsg->process();
} catch (std::exception &e) {
if (isExitRequested) {
return;
}
result["errorMessage"] = string(e.what());
spdlog::error("Exception in zmq server :{}", e.what());
spdlog::error("ID:" + string((char*) identity.data(), identity.size()));
spdlog::error("Client request :" + stringToParse);
} catch (...) {
if (isExitRequested) {
return;
}
spdlog::error("Error in zmq server ");
result["errorMessage"] = "Error in zmq server ";
spdlog::error("ID:" + string((char*) identity.data(), identity.size()));
spdlog::error("Client request :" + stringToParse);
}
try {
Json::FastWriter fastWriter;
fastWriter.omitEndingLineFeed();
replyStr = fastWriter.write(result);
CHECK_STATE(replyStr.size() > 2);
CHECK_STATE(replyStr.front() == '{');
CHECK_STATE(replyStr.back() == '}');
if (!socket->send(copied_id, ZMQ_SNDMORE)) {
if (isExitRequested) {
return;
}
exit(-15);
}
if (!s_send(*socket, replyStr)) {
if (isExitRequested) {
return;
}
exit(-16);
}
} catch ( std::exception &e ) {
if (isExitRequested) {
return;
}
spdlog::error("Exception in zmq server worker send :{}", e.what());
exit(-17);
} catch (...) {
if (isExitRequested) {
return;
}
spdlog::error("Unklnown exception in zmq server worker send");
exit(-18);
}
}
zmq_src/ZMQServer.h
View file @ e0482d51
......@@ -35,9 +35,6 @@
#include <zmq.hpp>
#include "zhelpers.hpp"
#include "ServerWorker.h"
using namespace std;
......@@ -61,21 +58,17 @@ public:
void run();
void exitAll();
static void initZMQServer(bool _checkSignature);
static void exitZMQServer();
private:
shared_ptr<zmq::context_t> ctx_;
shared_ptr<zmq::socket_t> frontend;
shared_ptr<zmq::socket_t> backend;
std::vector<shared_ptr<ServerWorker> > workers;
std::vector<shared_ptr<std::thread>> worker_threads;
shared_ptr<zmq::context_t> ctx;
shared_ptr<zmq::socket_t> socket;
static std::atomic<bool> isExitRequested;
void doOneServerLoop();
};
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment