Unverified Commit 8fc37015 authored by kladko's avatar kladko

Adding zmq

parent d522173a
......@@ -19,3 +19,6 @@
[submodule "libzmq"]
path = libzmq
url = https://github.com/zeromq/libzmq.git
[submodule "cppzmq"]
path = cppzmq
url = https://github.com/zeromq/cppzmq.git
......@@ -49,7 +49,7 @@ AM_CFLAGS = -DUSER_SPACE -g -Og -rdynamic -Wl,--no-as-needed -lSegFault -DSGXWAL
AM_CXXFLAGS = ${AM_CPPFLAGS} -rdynamic -Wl,--no-as-needed -lSegFault -DSGXWALLET_VERSION="$(WALLET_VERSION)"
AM_CPPFLAGS += -DSGXWALLET_VERSION="$(WALLET_VERSION)" -Wall -DSKALE_SGX=1 -DBINARY_OUTPUT=1 -Ileveldb/include -IlibBLS/bls -IlibBLS/libff -IlibBLS -fno-builtin-memset $(GMP_CPPFLAGS) -I. -I./libBLS/deps/deps_inst/x86_or_x64/include -I./libzmq/include
AM_CPPFLAGS += -DSGXWALLET_VERSION="$(WALLET_VERSION)" -Wall -DSKALE_SGX=1 -DBINARY_OUTPUT=1 -Ileveldb/include -IlibBLS/bls -IlibBLS/libff -IlibBLS -fno-builtin-memset $(GMP_CPPFLAGS) -I. -I./libBLS/deps/deps_inst/x86_or_x64/include -I./libzmq/include -I./cppzmq -I./third_party/zguide
## Additional targets to remove with 'make clean'. You must list
## any edger8r generated files here.
......
......@@ -21,9 +21,12 @@
@date 2019
*/
#include "sgxwallet.h"
#include "ServerWorker.h"
#include "ServerTask.h"
#include "sgxwallet_common.h"
using namespace std;
ServerTask::ServerTask()
: ctx_(1),
......@@ -35,12 +38,12 @@ void ServerTask::run() {
frontend_.bind("tcp://*:" + to_string(BASE_PORT + 4)) ;
backend_.bind("inproc://backend");
std::vector < server_worker * > worker;
std::vector < ServerWorker * > worker;
std::vector < std::thread * > worker_thread;
for (int i = 0; i < kMaxThread; ++i) {
worker.push_back(new server_worker(ctx_, ZMQ_DEALER));
worker.push_back(new ServerWorker(ctx_, ZMQ_DEALER));
worker_thread.push_back(new std::thread(std::bind(&server_worker::work, worker[i])));
worker_thread.push_back(new std::thread(std::bind(&ServerWorker::work, worker[i])));
worker_thread[i]->detach();
}
......@@ -58,7 +61,4 @@ void ServerTask::run() {
}
}
zmq::context_t ServerTask::ctx_;
zmq::socket_t ServerTask::frontend_;
zmq::socket_t ServerTask::backend_;
......@@ -34,20 +34,22 @@
#include <zmq.hpp>
#include "zhelpers.hpp"
class ServerTask {
class ServerTask {
public:
ServerTask();
enum { kMaxThread = 5 };
void run();
class ServerTask {
public:
ServerTask();
private:
zmq::context_t ctx_;
zmq::socket_t frontend_;
zmq::socket_t backend_;
enum {
kMaxThread = 5
};
void run();
private:
zmq::context_t ctx_;
zmq::socket_t frontend_;
zmq::socket_t backend_;
};
#endif //SGXWALLET_SERVERTASK_H
......@@ -5,7 +5,8 @@
#include "ServerWorker.h"
ServerWorker::ServerWorker(zmq::context_t &ctx, int sock_type);
ServerWorker::ServerWorker(zmq::context_t &ctx, int sock_type) : ctx_(ctx),
worker_(ctx_, sock_type){};
void ServerWorker::work() {
worker_.connect("inproc://backend");
......@@ -32,6 +33,4 @@ void ServerWorker::work() {
catch (std::exception &e) {}
}
zmq::context_t &ServerWorker::ctx_;
zmq::socket_t ServerWorker::worker_;
Subproject commit 18db4568f9ff1568193d3a496d87bed4bc9ec59e
......@@ -36,6 +36,9 @@ GMP_DIR = topDir + "/sgx-gmp"
SGX_SDK_DIR_SSL = topDir + "/sgx-sdk-build/sgxsdk"
ZMQ_DIR = topDir + "/libzmq"
ZMQ_BUILD_DIR = ZMQ_DIR + "/build"
CZMQ_DIR = topDir + "/cppzmq"
CZMQ_BUILD_DIR = CZMQ_DIR + "/build"
LEVELDB_DIR = topDir + "/leveldb"
LEVELDB_BUILD_DIR = LEVELDB_DIR + "/build"
GMP_BUILD_DIR = topDir + "/gmp-build"
......@@ -72,6 +75,7 @@ assert subprocess.call(["bash", "-c", "mkdir -p build"]) == 0
os.chdir(ZMQ_BUILD_DIR)
assert subprocess.call(["bash", "-c", "cmake -DDZMQ_EXPERIMENTAL=1 -DCMAKE_BUILD_TYPE=Release .. && cmake --build ."]) == 0
print("Build LevelDB");
os.chdir(LEVELDB_DIR)
......
Copyright (c) 2010-2013 iMatix Corporation and Contributors
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
\ No newline at end of file
#ifndef __ZHELPERS_HPP_INCLUDED__
#define __ZHELPERS_HPP_INCLUDED__
// Include a bunch of headers that we will need in the examples
#include <zmq.hpp> // https://github.com/zeromq/cppzmq
#include <iostream>
#include <iomanip>
#include <string>
#include <sstream>
#include <time.h>
#include <assert.h>
#include <stdlib.h> // random() RAND_MAX
#include <stdio.h>
#include <stdarg.h>
#include <signal.h>
#if (!defined(WIN32))
# include <sys/time.h>
# include <unistd.h>
#endif
// Bring Windows MSVC up to C99 scratch
#if (defined (WIN32))
typedef unsigned long ulong;
typedef unsigned int uint;
typedef __int64 int64_t;
#endif
// On some version of Windows, POSIX subsystem is not installed by default.
// So define srandom and random ourself.
//
#if (defined (WIN32))
# define srandom srand
# define random rand
#endif
// Visual Studio versions below 2015 do not support sprintf properly. This is a workaround.
// Taken from http://stackoverflow.com/questions/2915672/snprintf-and-visual-studio-2010
#if defined(_MSC_VER) && _MSC_VER < 1900
#define snprintf c99_snprintf
#define vsnprintf c99_vsnprintf
inline int c99_vsnprintf(char *outBuf, size_t size, const char *format, va_list ap)
{
int count = -1;
if (size != 0)
count = _vsnprintf_s(outBuf, size, _TRUNCATE, format, ap);
if (count == -1)
count = _vscprintf(format, ap);
return count;
}
inline int c99_snprintf(char *outBuf, size_t size, const char *format, ...)
{
int count;
va_list ap;
va_start(ap, format);
count = c99_vsnprintf(outBuf, size, format, ap);
va_end(ap);
return count;
}
#endif
// Provide random number from 0..(num-1)
#define within(num) (int) ((float)((num) * random ()) / (RAND_MAX + 1.0))
// Receive 0MQ string from socket and convert into C string
// Caller must free returned string.
inline static char *
s_recv(void *socket, int flags = 0) {
zmq_msg_t message;
zmq_msg_init(&message);
int rc = zmq_msg_recv(&message, socket, flags);
if (rc < 0)
return nullptr; // Context terminated, exit
size_t size = zmq_msg_size(&message);
char *string = (char*)malloc(size + 1);
memcpy(string, zmq_msg_data(&message), size);
zmq_msg_close(&message);
string[size] = 0;
return (string);
}
// Receive 0MQ string from socket and convert into string
inline static std::string
s_recv (zmq::socket_t & socket, int flags = 0) {
zmq::message_t message;
socket.recv(&message, flags);
return std::string(static_cast<char*>(message.data()), message.size());
}
inline static bool s_recv(zmq::socket_t & socket, std::string & ostring, int flags = 0)
{
zmq::message_t message;
bool rc = socket.recv(&message, flags);
if (rc) {
ostring = std::string(static_cast<char*>(message.data()), message.size());
}
return (rc);
}
// Convert C string to 0MQ string and send to socket
inline static int
s_send(void *socket, const char *string, int flags = 0) {
int rc;
zmq_msg_t message;
zmq_msg_init_size(&message, strlen(string));
memcpy(zmq_msg_data(&message), string, strlen(string));
rc = zmq_msg_send(&message, socket, flags);
assert(-1 != rc);
zmq_msg_close(&message);
return (rc);
}
// Convert string to 0MQ string and send to socket
inline static bool
s_send (zmq::socket_t & socket, const std::string & string, int flags = 0) {
zmq::message_t message(string.size());
memcpy (message.data(), string.data(), string.size());
bool rc = socket.send (message, flags);
return (rc);
}
// Sends string as 0MQ string, as multipart non-terminal
inline static int
s_sendmore(void *socket, char *string) {
int rc;
zmq_msg_t message;
zmq_msg_init_size(&message, strlen(string));
memcpy(zmq_msg_data(&message), string, strlen(string));
//rc = zmq_send(socket, string, strlen(string), ZMQ_SNDMORE);
rc = zmq_msg_send(&message, socket, ZMQ_SNDMORE);
assert(-1 != rc);
zmq_msg_close(&message);
return (rc);
}
// Sends string as 0MQ string, as multipart non-terminal
inline static bool
s_sendmore (zmq::socket_t & socket, const std::string & string) {
zmq::message_t message(string.size());
memcpy (message.data(), string.data(), string.size());
bool rc = socket.send (message, ZMQ_SNDMORE);
return (rc);
}
// Receives all message parts from socket, prints neatly
//
inline static void
s_dump (zmq::socket_t & socket)
{
std::cout << "----------------------------------------" << std::endl;
while (1) {
// Process all parts of the message
zmq::message_t message;
socket.recv(&message);
// Dump the message as text or binary
size_t size = message.size();
std::string data(static_cast<char*>(message.data()), size);
bool is_text = true;
size_t char_nbr;
unsigned char byte;
for (char_nbr = 0; char_nbr < size; char_nbr++) {
byte = data [char_nbr];
if (byte < 32 || byte > 127)
is_text = false;
}
std::cout << "[" << std::setfill('0') << std::setw(3) << size << "]";
for (char_nbr = 0; char_nbr < size; char_nbr++) {
if (is_text)
std::cout << (char)data [char_nbr];
else
std::cout << std::setfill('0') << std::setw(2)
<< std::hex << (unsigned int) data [char_nbr];
}
std::cout << std::endl;
int more = 0; // Multipart detection
size_t more_size = sizeof (more);
socket.getsockopt (ZMQ_RCVMORE, &more, &more_size);
if (!more)
break; // Last message part
}
}
#if (!defined (WIN32))
// Set simple random printable identity on socket
// Caution:
// DO NOT call this version of s_set_id from multiple threads on MS Windows
// since s_set_id will call rand() on MS Windows. rand(), however, is not
// reentrant or thread-safe. See issue #521.
inline std::string
s_set_id (zmq::socket_t & socket)
{
std::stringstream ss;
ss << std::hex << std::uppercase
<< std::setw(4) << std::setfill('0') << within (0x10000) << "-"
<< std::setw(4) << std::setfill('0') << within (0x10000);
socket.setsockopt(ZMQ_IDENTITY, ss.str().c_str(), ss.str().length());
return ss.str();
}
#else
// Fix #521
inline std::string
s_set_id(zmq::socket_t & socket, intptr_t id)
{
std::stringstream ss;
ss << std::hex << std::uppercase
<< std::setw(4) << std::setfill('0') << id;
socket.setsockopt(ZMQ_IDENTITY, ss.str().c_str(), ss.str().length());
return ss.str();
}
#endif
// Report 0MQ version number
//
inline static void
s_version (void)
{
int major, minor, patch;
zmq_version (&major, &minor, &patch);
std::cout << "Current 0MQ version is " << major << "." << minor << "." << patch << std::endl;
}
inline static void
s_version_assert (int want_major, int want_minor)
{
int major, minor, patch;
zmq_version (&major, &minor, &patch);
if (major < want_major
|| (major == want_major && minor < want_minor)) {
std::cout << "Current 0MQ version is " << major << "." << minor << std::endl;
std::cout << "Application needs at least " << want_major << "." << want_minor
<< " - cannot continue" << std::endl;
exit (EXIT_FAILURE);
}
}
// Return current system clock as milliseconds
inline static int64_t
s_clock (void)
{
#if (defined (WIN32))
FILETIME fileTime;
GetSystemTimeAsFileTime(&fileTime);
unsigned __int64 largeInt = fileTime.dwHighDateTime;
largeInt <<= 32;
largeInt |= fileTime.dwLowDateTime;
largeInt /= 10000; // FILETIME is in units of 100 nanoseconds
return (int64_t)largeInt;
#else
struct timeval tv;
gettimeofday (&tv, NULL);
return (int64_t) (tv.tv_sec * 1000 + tv.tv_usec / 1000);
#endif
}
// Sleep for a number of milliseconds
inline static void
s_sleep (int msecs)
{
#if (defined (WIN32))
Sleep (msecs);
#else
struct timespec t;
t.tv_sec = msecs / 1000;
t.tv_nsec = (msecs % 1000) * 1000000;
nanosleep (&t, NULL);
#endif
}
inline static void
s_console (const char *format, ...)
{
time_t curtime = time (NULL);
struct tm *loctime = localtime (&curtime);
char *formatted = new char[20];
strftime (formatted, 20, "%y-%m-%d %H:%M:%S ", loctime);
printf ("%s", formatted);
delete[] formatted;
va_list argptr;
va_start (argptr, format);
vprintf (format, argptr);
va_end (argptr);
printf ("\n");
}
// ---------------------------------------------------------------------
// Signal handling
//
// Call s_catch_signals() in your application at startup, and then exit
// your main loop if s_interrupted is ever 1. Works especially well with
// zmq_poll.
static int s_interrupted = 0;
inline static void s_signal_handler (int signal_value)
{
s_interrupted = 1;
}
inline static void s_catch_signals ()
{
#if (!defined(WIN32))
struct sigaction action;
action.sa_handler = s_signal_handler;
action.sa_flags = 0;
sigemptyset (&action.sa_mask);
sigaction (SIGINT, &action, NULL);
sigaction (SIGTERM, &action, NULL);
#endif
}
#endif
\ No newline at end of file
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