fzorb/REDACTED-rig
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reimplement whole stratum network communication with boost::asio (#90)

Browse source 
Logger is now thread safe
This commit is contained in:
Ben Gräf 2018-04-14 19:55:13 +02:00 • committed by GitHub
parent 15d752d9e0
commit df084acff6
40 changed files with 915 additions and 1942 deletions
Download patch file Download diff file Expand all files Collapse all files

144
src/net/BoostConnection.h Normal file
View file

@ -0,0 +1,144 @@
/* XMRigCC
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
* Copyright 2018- BenDr0id <ben@graef.in>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __BOOSTCONNECTION_H__
#define __BOOSTCONNECTION_H__
#include "net/Connection.h"
#include "log/Log.h"
template <class SOCKET>
class BoostConnection : public Connection, public std::enable_shared_from_this<BoostConnection<SOCKET> >
{
public:
BoostConnection(const ConnectionListener::Ptr& listener)
: Connection(listener)
, socket_(ioService_)
{
}
~BoostConnection()
{
disconnect();
}
void connect(const std::string& server, uint16_t port) override
{
LOG_DEBUG("[%s:%d] Connecting", server.c_str(), port);
boost::asio::ip::tcp::resolver resolver(ioService_);
boost::asio::ip::tcp::resolver::query query(server, std::to_string(port));
boost::asio::ip::tcp::resolver::iterator iterator = resolver.resolve(query);
socket_.connect(iterator,
boost::bind(&BoostConnection::handleConnect, this->shared_from_this(),
boost::asio::placeholders::error));
std::thread([this]() { ioService_.run(); }).detach();
}
void handleConnect(const boost::system::error_code& error)
{
if (!error) {
startReading();
LOG_DEBUG("[%s:%d] Connected", getConnectedIp().c_str(), getConnectedPort());
notifyConnected();
} else {
notifyError(error.message());
}
}
void disconnect() override
{
if (isConnected()) {
LOG_DEBUG("[%s:%d] Disconnecting", getConnectedIp().c_str(), getConnectedPort());
socket_.get().lowest_layer().close();
}
ioService_.stop();
}
bool isConnected() const override
{
boost::system::error_code ec;
socket_.get().lowest_layer().remote_endpoint(ec);
return !ec && socket_.get().lowest_layer().is_open();
}
std::string getConnectedIp() const override
{
return isConnected() ? socket_.get().lowest_layer().remote_endpoint().address().to_string() : "";
}
uint16_t getConnectedPort() const override
{
return isConnected() ? socket_.get().lowest_layer().remote_endpoint().port() : 0;
}
void send(const char* data, std::size_t size) override
{
LOG_DEBUG("[%s:%d] Sending: %.*s", getConnectedIp().c_str(), getConnectedPort(), size, data);
boost::asio::async_write(socket_.get(),
boost::asio::buffer(data, size),
boost::bind(&BoostConnection::handleWrite, this->shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
void handleWrite(const boost::system::error_code& error,
size_t bytes_transferred)
{
if (error) {
LOG_DEBUG_ERR("[%s:%d] Sending failed: %s", getConnectedIp().c_str(), getConnectedPort(), error.message().c_str());
notifyError(error.message());
}
}
void startReading()
{
boost::asio::async_read(socket_.get(),
boost::asio::buffer(receiveBuffer_, sizeof(receiveBuffer_)),
boost::asio::transfer_at_least(1),
boost::bind(&BoostConnection::handleRead, this->shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
void handleRead(const boost::system::error_code& error,
size_t bytes_transferred)
{
if (!error) {
LOG_DEBUG("[%s:%d] Read: %.*s", getConnectedIp().c_str(), getConnectedPort(), bytes_transferred, receiveBuffer_);
notifyRead(receiveBuffer_, bytes_transferred);
startReading();
} else {
LOG_DEBUG_ERR("[%s:%d] Read failed: %s", getConnectedIp().c_str(), getConnectedPort(), error.message().c_str());
notifyError(error.message());
}
}
private:
boost::asio::io_service ioService_;
SOCKET socket_;
char receiveBuffer_[2048];
};
#endif /* __BOOSTCONNECTION_H__ */

72
src/net/BoostTcpConnection.cpp Normal file
View file

@ -0,0 +1,72 @@
/* XMRigCC
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <iostream>
#include <thread>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include "net/BoostConnection.h"
#include "net/BoostTcpConnection.h"
class BoostTcpSocket
{
public:
typedef boost::asio::ip::tcp::socket SocketType;
public:
BoostTcpSocket(boost::asio::io_service& ioService)
: socket_(ioService)
{
}
template <class ITERATOR, class HANDLER>
void connect(ITERATOR& iterator, HANDLER handler)
{
boost::asio::async_connect(
socket_, iterator,
[this, handler](const boost::system::error_code& ec, const ITERATOR& iterator)
{
if (!ec) {
socket_.set_option(boost::asio::ip::tcp::no_delay(true));
socket_.set_option(boost::asio::socket_base::keep_alive(true));
}
handler(ec);
}
);
}
SocketType& get()
{
return socket_;
}
const SocketType& get() const
{
return socket_;
}
private:
SocketType socket_;
};
Connection::Ptr establishBoostTcpConnection(const ConnectionListener::Ptr& listener)
{
return std::make_shared<BoostConnection<BoostTcpSocket> >(listener);
}

26
src/net/BoostTcpConnection.h Normal file
View file

@ -0,0 +1,26 @@
/* XMRigCC
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __BOOSTTCPCONNECTION_H__
#define __BOOSTTCPCONNECTION_H__
#include "net/Connection.h"
Connection::Ptr establishBoostTcpConnection(const ConnectionListener::Ptr& listener);
#endif /* __BOOSTTCPCONNECTION_H__ */

83
src/net/BoostTlsConnection.cpp Normal file
View file

@ -0,0 +1,83 @@
/* XMRigCC
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <iostream>
#include <thread>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include "net/BoostConnection.h"
#include "net/BoostTlsConnection.h"
class BoostTlsSocket
{
public:
typedef boost::asio::ssl::stream<boost::asio::ip::tcp::socket> SocketType;
public:
BoostTlsSocket(boost::asio::io_service& ioService)
: sslContext_(boost::asio::ssl::context::sslv23_client)
, socket_(ioService, sslContext_)
{
socket_.set_verify_mode(boost::asio::ssl::verify_none);
}
template <class ITERATOR, class HANDLER>
void connect(ITERATOR& iterator, HANDLER handler)
{
boost::asio::async_connect(
socket_.lowest_layer(), iterator,
[this, handler](const boost::system::error_code& ec, const ITERATOR& iterator)
{
if (!ec) {
socket_.lowest_layer().set_option(boost::asio::ip::tcp::no_delay(true));
socket_.lowest_layer().set_option(boost::asio::socket_base::keep_alive(true));
socket_.async_handshake(
boost::asio::ssl::stream_base::client,
[handler](const boost::system::error_code& ec) {
handler(ec);
});
} else {
handler(ec);
}
}
);
}
SocketType& get()
{
return socket_;
}
const SocketType& get() const
{
return socket_;
}
private:
boost::asio::ssl::context sslContext_;
SocketType socket_;
};
Connection::Ptr establishBoostTlsConnection(const ConnectionListener::Ptr& listener)
{
return std::make_shared<BoostConnection<BoostTlsSocket> >(listener);
}

26
src/net/BoostTlsConnection.h Normal file
View file

@ -0,0 +1,26 @@
/* XMRigCC
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __BOOSTTLSCONNECTION_H__
#define __BOOSTTLSCONNECTION_H__
#include "net/Connection.h"
Connection::Ptr establishBoostTlsConnection(const ConnectionListener::Ptr& listener);
#endif /* __BOOSTTLSCONNECTION_H__ */

439
src/net/Client.cpp
View file

@ -5,6 +5,7 @@
* Copyright 2014-2016 Wolf9466 <https://github.com/OhGodAPet>
* Copyright 2016 Jay D Dee <jayddee246@gmail.com>
* Copyright 2016-2017 XMRig <support@xmrig.com>
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
* Copyright 2018- BenDr0id <ben@graef.in>
*
*
@ -32,19 +33,12 @@
#include "interfaces/IClientListener.h"
#include "log/Log.h"
#include "net/Client.h"
#include "net/JobResult.h"
#include "rapidjson/document.h"
#include "rapidjson/error/en.h"
#include "rapidjson/stringbuffer.h"
#include "rapidjson/writer.h"
#ifdef XMRIG_PROXY_PROJECT
# include "proxy/JobResult.h"
#else
# include "net/JobResult.h"
#endif
#ifdef _MSC_VER
# define strncasecmp(x,y,z) _strnicmp(x,y,z)
#endif
@ -54,54 +48,65 @@ int64_t Client::m_sequence = 1;
Client::Client(int id, const char *agent, IClientListener *listener) :
m_quiet(false),
m_nicehash(false),
m_agent(agent),
m_listener(listener),
m_id(id),
m_retryPause(5000),
m_failures(0),
m_jobs(0),
m_recvBufPos(0),
m_expire(0)
m_quiet(false),
m_nicehash(false),
m_agent(agent),
m_listener(listener),
m_id(id),
m_retryPause(5000),
m_failures(0),
m_jobs(0),
m_recvBufPos(0),
m_expire(0)
{
m_recvBuf.base = m_buf;
m_recvBuf.len = sizeof(m_buf);
# ifndef XMRIG_PROXY_PROJECT
m_keepAliveTimer.data = this;
uv_timer_init(uv_default_loop(), &m_keepAliveTimer);
# endif
uv_mutex_init(&m_mutex);
uv_async_init(uv_default_loop(), &onConnectedAsync, Client::onConnected);
uv_async_init(uv_default_loop(), &onReceivedAsync, Client::onReceived);
uv_async_init(uv_default_loop(), &onErrorAsync, Client::onError);
}
Client::~Client()
{
if (m_net) {
net_free(m_net);
}
uv_close((uv_handle_t*) &onConnectedAsync, NULL);
uv_close((uv_handle_t*) &onReceivedAsync, NULL);
uv_close((uv_handle_t*) &onErrorAsync, NULL);
uv_mutex_destroy(&m_mutex);
}
/**
* @brief Connect to server.
*
* @param url
*/
void Client::connect(const Url *url)
{
LOG_DEBUG("connect %s", url);
setUrl(url);
connect();
}
void Client::connect()
{
LOG_DEBUG("connect");
m_connection = establishConnection(shared_from_this(),
m_url.useTls() ? CONNECTION_TYPE_TLS : CONNECTION_TYPE_TCP,
m_url.host(), m_url.port());
}
void Client::disconnect()
{
LOG_DEBUG("Client::disconnect");
LOG_DEBUG("disconnect");
# ifndef XMRIG_PROXY_PROJECT
uv_timer_stop(&m_keepAliveTimer);
# endif
m_expire = 0;
m_failures = -1;
@ -112,6 +117,8 @@ void Client::disconnect()
void Client::setUrl(const Url *url)
{
LOG_DEBUG("setUrl");
if (!url || !url->isValid()) {
return;
}
@ -126,12 +133,15 @@ void Client::tick(uint64_t now)
return;
}
if (m_net) {
LOG_DEBUG("tick expired");
if (m_connection) {
LOG_WARN("[%s:%u] timeout", m_url.host(), m_url.port());
close();
LOG_DEBUG("tick -> reconnect");
reconnect();
}
else {
LOG_DEBUG("Client::tick -> connect");
LOG_DEBUG("tick -> connect");
connect();
}
}
@ -139,10 +149,6 @@ void Client::tick(uint64_t now)
int64_t Client::submit(const JobResult &result)
{
# ifdef XMRIG_PROXY_PROJECT
const char *nonce = result.nonce;
const char *data = result.result;
# else
char nonce[9];
char data[65];
@ -151,13 +157,24 @@ int64_t Client::submit(const JobResult &result)
Job::toHex(result.result, 32, data);
data[64] = '\0';
# endif
const size_t size = snprintf(m_sendBuf, sizeof(m_sendBuf), "{\"id\":%" PRIu64 ",\"jsonrpc\":\"2.0\",\"method\":\"submit\",\"params\":{\"id\":\"%s\",\"job_id\":\"%s\",\"nonce\":\"%s\",\"result\":\"%s\"}}\n",
m_sequence, m_rpcId, result.jobId.data(), nonce, data);
const int size = snprintf(m_sendBuf, sizeof(m_sendBuf),
"{"
"\"id\":%" PRIu64 ","
"\"jsonrpc\":\"2.0\","
"\"method\":\"submit\","
"\"params\":"
"{"
"\"id\":\"%s\","
"\"job_id\":\"%s\","
"\"nonce\":\"%s\","
"\"result\":\"%s\""
"}"
"}\n",
m_sequence, m_rpcId, result.jobId.data(), nonce, data);
m_results[m_sequence] = SubmitResult(m_sequence, result.diff, result.actualDiff());
return send(size);
return send(m_sendBuf, size);
}
@ -191,6 +208,7 @@ bool Client::parseJob(const rapidjson::Value &params, int *code)
}
Job job(m_id, m_nicehash);
if (!job.setId(params["job_id"].GetString())) {
*code = 3;
return false;
@ -211,17 +229,19 @@ bool Client::parseJob(const rapidjson::Value &params, int *code)
switch (variantFromProxy) {
case -1:
Options::i()->setForcePowVersion(Options::POW_AUTODETECT);
job.setPowVersion(Options::POW_AUTODETECT);
break;
case 0:
Options::i()->setForcePowVersion(Options::POW_V1);
job.setPowVersion(Options::POW_V1);
break;
case 1:
Options::i()->setForcePowVersion(Options::POW_V2);
job.setPowVersion(Options::POW_V2);
break;
default:
break;
}
} else {
job.setPowVersion(Options::i()->forcePowVersion());
}
if (m_job != job) {
@ -238,11 +258,29 @@ bool Client::parseJob(const rapidjson::Value &params, int *code)
LOG_WARN("[%s:%u] duplicate job received, reconnect", m_url.host(), m_url.port());
}
close();
reconnect();
return false;
}
void Client::parseExtensions(const rapidjson::Value &value)
{
if (!value.IsArray()) {
return;
}
for (const rapidjson::Value &ext : value.GetArray()) {
if (!ext.IsString()) {
continue;
}
if (strcmp(ext.GetString(), "nicehash") == 0) {
m_nicehash = true;
}
}
}
bool Client::parseLogin(const rapidjson::Value &result, int *code)
{
const char *id = result["id"].GetString();
@ -251,9 +289,7 @@ bool Client::parseLogin(const rapidjson::Value &result, int *code)
return false;
}
# ifndef XMRIG_PROXY_PROJECT
m_nicehash = m_url.isNicehash();
# endif
if (result.HasMember("extensions")) {
parseExtensions(result["extensions"]);
@ -268,139 +304,31 @@ bool Client::parseLogin(const rapidjson::Value &result, int *code)
return rc;
}
int64_t Client::send(size_t size)
int64_t Client::send(char* buf, size_t size)
{
LOG_DEBUG("Client::send");
LOG_DEBUG("[%s:%u] send (%d bytes): \"%s\"", m_url.host(), m_url.port(), size, m_sendBuf);
if (!m_net) {
LOG_DEBUG_ERR("[%s:%u] send failed", m_url.host(), m_url.port());
return -1;
if (m_connection)
{
m_connection->send(buf, size);
m_expire = uv_now(uv_default_loop()) + kResponseTimeout;
m_sequence++;
}
if (net_write2(m_net, m_sendBuf, static_cast<unsigned int>(size)) < 0) {
close();
return -1;
}
m_expire = uv_now(uv_default_loop()) + kResponseTimeout;
return m_sequence++;
return m_sequence;
}
void Client::close()
{
LOG_DEBUG("Client::close");
LOG_DEBUG("close");
if (m_net) {
net_close(m_net, nullptr);
} else {
reconnect();
}
m_connection.reset();
}
void Client::connect()
{
LOG_DEBUG("Client::connect");
m_net = net_new(const_cast<char *>(m_url.host()), m_url.port());
m_net->data = this;
m_net->conn_cb = Client::onConnect;
m_net->read_cb = Client::onRead;
m_net->close_cb = Client::onClose;
m_net->error_cb = Client::onError;
#ifndef XMRIG_NO_TLS
if (m_url.useTls()) {
tls_ctx* tls_ctx = tls_ctx_new();
net_set_tls(m_net, tls_ctx);
}
#endif
net_connect(m_net);
}
void Client::onRead(net_t *net, size_t size, char *buf)
{
LOG_DEBUG("Client::onRead");
auto client = getClient(net->data);
if (size == 0) {
if (size != UV_EOF && !client->m_quiet) {
LOG_ERR("[%s:%u] read error: \"%s\"", client->m_url.host(), client->m_url.port(), uv_strerror((int) size));
}
return client->close();
}
client->m_recvBufPos += size;
char* end;
char* start = buf;
size_t remaining = client->m_recvBufPos;
while ((end = static_cast<char*>(memchr(start, '\n', remaining))) != nullptr) {
end++;
size_t len = end - start;
client->parse(start, len);
remaining -= len;
start = end;
}
if (remaining == 0) {
client->m_recvBufPos = 0;
return;
}
if (start == buf) {
return;
}
memcpy(buf, start, remaining);
client->m_recvBufPos = remaining;
}
void Client::onConnect(net_t *net) {
LOG_DEBUG("Client::onConnect");
auto client = getClient(net->data);
client->login();
}
void Client::onError(net_t *net, int err, char *errStr)
{
LOG_DEBUG("Client::onError");
if (net) {
auto client = getClient(net->data);
if (!client->m_quiet) {
LOG_ERR("[%s:%u] error: \"%s\"", client->m_url.host(), client->m_url.port(), errStr);
}
client->close();
}
}
void Client::onClose(net_t *net)
{
LOG_DEBUG("Client::onClose");
if (net) {
auto client = getClient(net->data);
net_free(net);
client->m_net = nullptr;
client->reconnect();
}
}
void Client::login()
{
LOG_DEBUG("Client::login");
LOG_DEBUG("login");
m_results.clear();
@ -433,20 +361,54 @@ void Client::login()
m_sendBuf[size] = '\n';
m_sendBuf[size + 1] = '\0';
send(size + 1);
send(m_sendBuf, size + 1);
}
void Client::processReceivedData(char* data, size_t size)
{
LOG_DEBUG("processReceivedData");
if ((size_t) size > (sizeof(m_buf) - 8 - m_recvBufPos)) {
reconnect();
return;
}
m_recvBufPos += size;
char* end;
char* start = data;
size_t remaining = m_recvBufPos;
while ((end = static_cast<char*>(memchr(start, '\n', remaining))) != nullptr) {
end++;
size_t len = end - start;
parse(start, len);
remaining -= len;
start = end;
}
if (remaining == 0) {
m_recvBufPos = 0;
return;
}
if (start == data) {
return;
}
memcpy(data, start, remaining);
m_recvBufPos = remaining;
}
void Client::parse(char *line, size_t len)
{
LOG_DEBUG("Client::parse");
LOG_DEBUG("parse");
startTimeout();
line[len - 1] = '\0';
LOG_DEBUG("[%s:%u] received (%d bytes): \"%s\"", m_url.host(), m_url.port(), len, line);
rapidjson::Document doc;
if (doc.ParseInsitu(line).HasParseError()) {
if (!m_quiet) {
@ -470,28 +432,11 @@ void Client::parse(char *line, size_t len)
}
void Client::parseExtensions(const rapidjson::Value &value)
{
if (!value.IsArray()) {
return;
}
for (const rapidjson::Value &ext : value.GetArray()) {
if (!ext.IsString()) {
continue;
}
if (strcmp(ext.GetString(), "nicehash") == 0) {
m_nicehash = true;
}
}
}
void Client::parseNotification(const char *method, const rapidjson::Value &params, const rapidjson::Value &error)
{
if (error.IsObject()) {
if (!m_quiet) {
LOG_ERR("[%s:%u] error: \"%s\", code: %d", m_url.host(), m_url.port(), error["message"].GetString(), error["code"].GetInt());
LOG_ERR("[%s:%u] Parse notification failed: \"%s\", code: %d", m_url.host(), m_url.port(), error["message"].GetString(), error["code"].GetInt());
}
return;
}
@ -509,7 +454,7 @@ void Client::parseNotification(const char *method, const rapidjson::Value &param
return;
}
LOG_WARN("[%s:%u] unsupported method: \"%s\"", m_url.host(), m_url.port(), method);
LOG_WARN("[%s:%u] Unsupported method: \"%s\"", m_url.host(), m_url.port(), method);
}
@ -525,11 +470,11 @@ void Client::parseResponse(int64_t id, const rapidjson::Value &result, const rap
m_results.erase(it);
}
else if (!m_quiet) {
LOG_ERR("[%s:%u] error: \"%s\", code: %d", m_url.host(), m_url.port(), message, error["code"].GetInt());
LOG_ERR("[%s:%u] Parse response failed: \"%s\", code: %d", m_url.host(), m_url.port(), message, error["code"].GetInt());
}
if (id == 1 || isCriticalError(message)) {
close();
reconnect();
}
return;
@ -543,10 +488,10 @@ void Client::parseResponse(int64_t id, const rapidjson::Value &result, const rap
int code = -1;
if (!parseLogin(result, &code)) {
if (!m_quiet) {
LOG_ERR("[%s:%u] login error code: %d", m_url.host(), m_url.port(), code);
LOG_ERR("[%s:%u] Login error code: %d", m_url.host(), m_url.port(), code);
}
return close();
return reconnect();
}
m_failures = 0;
@ -566,26 +511,35 @@ void Client::parseResponse(int64_t id, const rapidjson::Value &result, const rap
void Client::ping()
{
LOG_DEBUG("Client::ping");
send(snprintf(m_sendBuf, sizeof(m_sendBuf), "{\"id\":%" PRId64 ",\"jsonrpc\":\"2.0\",\"method\":\"keepalived\",\"params\":{\"id\":\"%s\"}}\n", m_sequence, m_rpcId));
LOG_DEBUG("ping");
const int size = snprintf(m_sendBuf, sizeof(m_sendBuf),
"{"
"\"id\":%" PRId64 ","
"\"jsonrpc\":\"2.0\","
"\"method\":\"keepalived\","
"\"params\":"
"{"
"\"id\":\"%s\""
"}"
"}\n",
m_sequence, m_rpcId);
send(m_sendBuf, size);
}
void Client::reconnect() {
void Client::reconnect()
{
LOG_DEBUG("reconnect");
LOG_DEBUG("Client::reconnect");
close();
# ifndef XMRIG_PROXY_PROJECT
if (m_url.isKeepAlive()) {
uv_timer_stop(&m_keepAliveTimer);
}
# endif
if (m_failures == -1) {
LOG_DEBUG("Client::reconnect -> m_failures == -1");
m_failures = 0;
m_expire = 0;
LOG_DEBUG("reconnect -> m_failures == -1");
return m_listener->onClose(this, -1);
}
@ -595,17 +549,86 @@ void Client::reconnect() {
m_expire = uv_now(uv_default_loop()) + m_retryPause;
}
void Client::startTimeout()
{
LOG_DEBUG("Client::startTimeout");
LOG_DEBUG("startTimeout");
m_expire = 0;
# ifndef XMRIG_PROXY_PROJECT
if (!m_url.isKeepAlive()) {
return;
}
uv_timer_start(&m_keepAliveTimer, [](uv_timer_t *handle) { getClient(handle->data)->ping(); }, kKeepAliveTimeout, 0);
# endif
}
void Client::onConnected(uv_async_t *handle)
{
LOG_DEBUG("onConnected");
auto client = getClient(handle->data);
if (client) {
client->login();
}
}
void Client::scheduleOnConnected()
{
LOG_DEBUG("scheduleOnConnected");
onConnectedAsync.data = this;
uv_async_send(&onConnectedAsync);
}
void Client::onReceived(uv_async_t *handle)
{
LOG_DEBUG("onReceived");
auto client = getClient(handle->data);
if (client) {
uv_mutex_lock(&client->m_mutex);
while (!client->m_readQueue.empty()) {
std::string data = client->m_readQueue.front();
client->processReceivedData(const_cast<char *>(data.c_str()), data.size());
client->m_readQueue.pop_front();
}
uv_mutex_unlock(&client->m_mutex);
}
}
void Client::scheduleOnReceived(char* data, std::size_t size)
{
LOG_DEBUG("scheduleOnReceived");
uv_mutex_lock(&m_mutex);
m_readQueue.emplace_back(data, size);
uv_mutex_unlock(&m_mutex);
onReceivedAsync.data = this;
uv_async_send(&onReceivedAsync);
}
void Client::onError(uv_async_t *handle)
{
LOG_DEBUG("onError");
auto client = getClient(handle->data);
if (client) {
client->reconnect();
}
}
void Client::scheduleOnError(const std::string &error)
{
LOG_DEBUG("scheduleOnError");
if (!m_quiet) {
LOG_ERR("[%s:%u] Error: \"%s\"", m_url.host(), m_url.port(), error.c_str());
}
onErrorAsync.data = this;
uv_async_send(&onErrorAsync);
}

54
src/net/Client.h
View file

@ -29,34 +29,25 @@
#include <map>
#include <uv.h>
#include <memory>
#include <list>
#include "net/Connection.h"
#include "net/Job.h"
#include "net/SubmitResult.h"
#include "net/Url.h"
#include "rapidjson/fwd.h"
#ifdef __cplusplus
extern "C"
{
#endif
#include "net.h"
#ifndef XMRIG_NO_TLS
#include "tls.h"
#endif
#ifdef __cplusplus
}
#endif
class IClientListener;
class JobResult;
class Client
class Client : public ConnectionListener,
public std::enable_shared_from_this<Client>
{
public:
typedef std::shared_ptr<Client> Ptr;
public:
constexpr static int kResponseTimeout = 20 * 1000;
constexpr static int kKeepAliveTimeout = 60 * 1000;
@ -78,25 +69,29 @@ public:
inline void setQuiet(bool quiet) { m_quiet = quiet; }
inline void setRetryPause(int ms) { m_retryPause = ms; }
static void onConnected(uv_async_t *handle);
static void onReceived(uv_async_t *handle);
static void onError(uv_async_t *handle);
private:
bool isCriticalError(const char *message);
bool parseJob(const rapidjson::Value &params, int *code);
bool parseLogin(const rapidjson::Value &result, int *code);
int64_t send(size_t size);
int64_t send(char* buf, size_t size);
void close();
void reconnect();
void login();
void processReceivedData(char* data, size_t size);
void parse(char *line, size_t len);
void parseExtensions(const rapidjson::Value &value);
void parseNotification(const char *method, const rapidjson::Value &params, const rapidjson::Value &error);
void parseExtensions(const rapidjson::Value &value);
void parseResponse(int64_t id, const rapidjson::Value &result, const rapidjson::Value &error);
void ping();
void reconnect();
void startTimeout();
static void onRead(net_t *net, size_t read, char *buf);
static void onConnect(net_t *net);
static void onError(net_t *net, int err, char *errStr);
static void onClose(net_t *net);
virtual void scheduleOnConnected();
virtual void scheduleOnReceived(char *data, size_t size);
virtual void scheduleOnError(const std::string &error);
static inline Client *getClient(void *data) { return static_cast<Client*>(data); }
@ -119,13 +114,18 @@ private:
Url m_url;
uv_buf_t m_recvBuf;
net_t* m_net;
uv_mutex_t m_mutex;
std::list<std::string> m_readQueue;
Connection::Ptr m_connection;
uv_async_t onConnectedAsync;
uv_async_t onReceivedAsync;
uv_async_t onErrorAsync;
# ifndef XMRIG_PROXY_PROJECT
uv_timer_t m_keepAliveTimer;
# endif
};
#endif /* __CLIENT_H__ */
#endif /* __CLIENT_H__ */

89
src/net/Connection.cpp Normal file
View file

@ -0,0 +1,89 @@
/* XMRigCC
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
* Copyright 2018- BenDr0id <ben@graef.in>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <iostream>
#include <boost/exception/diagnostic_information.hpp>
#include <log/Log.h>
#include "Connection.h"
#include "BoostTcpConnection.h"
#ifndef XMRIG_NO_TLS
#include "BoostTlsConnection.h"
#endif
Connection::Connection(const ConnectionListener::Ptr &listener)
: listener_(listener)
{
}
void Connection::notifyConnected()
{
ConnectionListener::Ptr listener = listener_.lock();
if (listener)
{
listener->scheduleOnConnected();
}
}
void Connection::notifyRead(char* data, size_t size)
{
ConnectionListener::Ptr listener = listener_.lock();
if (listener)
{
listener->scheduleOnReceived(data, size);
}
}
void Connection::notifyError(const std::string& error)
{
ConnectionListener::Ptr listener = listener_.lock();
if (listener)
{
listener->scheduleOnError(error);
}
}
Connection::Ptr establishConnection(const ConnectionListener::Ptr& listener,
ConnectionType type, const std::string& host, uint16_t port)
{
Connection::Ptr connection;
try {
switch (type) {
case CONNECTION_TYPE_TLS:
#ifndef XMRIG_NO_TLS
connection = establishBoostTlsConnection(listener);
break;
#endif
case CONNECTION_TYPE_TCP:
connection = establishBoostTcpConnection(listener);
break;
}
connection->connect(host, port);
}
catch (...) {
LOG_ERR("[%s:%d] Failed to establish connection: %s", host.c_str(), port, boost::current_exception_diagnostic_information().c_str());
}
return connection;
}

78
src/net/Connection.h Normal file
View file

@ -0,0 +1,78 @@
/* XMRig
* Copyright 2018 Sebastian Stolzenberg <https://github.com/sebastianstolzenberg>
*
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __CONNECTION_H__
#define __CONNECTION_H__
#include <cstdint>
#include <memory>
#include <string>
#include <boost/noncopyable.hpp>
class ConnectionListener
{
public:
typedef std::shared_ptr<ConnectionListener> Ptr;
typedef std::weak_ptr<ConnectionListener> WeakPtr;
protected:
ConnectionListener() {};
virtual ~ConnectionListener() {};
public:
virtual void scheduleOnConnected() = 0;
virtual void scheduleOnReceived(char *data, std::size_t size) = 0;
virtual void scheduleOnError(const std::string &error) = 0;
};
class Connection : private boost::noncopyable
{
public:
typedef std::shared_ptr<Connection> Ptr;
protected:
Connection(const ConnectionListener::Ptr& listener);
virtual ~Connection() {};
void notifyConnected();
void notifyRead(char* data, size_t size);
void notifyError(const std::string& error);
public:
virtual void connect(const std::string& server, uint16_t port) = 0;
virtual void disconnect() = 0;
virtual bool isConnected() const = 0;
virtual std::string getConnectedIp() const = 0;
virtual uint16_t getConnectedPort() const = 0;
virtual void send(const char* data, std::size_t size) = 0;
private:
ConnectionListener::WeakPtr listener_;
};
enum ConnectionType
{
CONNECTION_TYPE_TCP,
CONNECTION_TYPE_TLS
};
Connection::Ptr establishConnection(const ConnectionListener::Ptr& listener,
ConnectionType type,
const std::string& host, uint16_t port);
#endif /* __CONNECTION_H__ */

3
src/net/Job.cpp
View file

@ -61,7 +61,8 @@ Job::Job(int poolId, bool nicehash) :
m_threadId(-1),
m_size(0),
m_diff(0),
m_target(0)
m_target(0),
m_powVersion(Options::POW_AUTODETECT)
{
}

9
src/net/Job.h
View file

@ -27,6 +27,7 @@
#include <stddef.h>
#include <stdint.h>
#include <Options.h>
#include "net/JobId.h"
@ -49,17 +50,14 @@ public:
inline const uint8_t *blob() const { return m_blob; }
inline int poolId() const { return m_poolId; }
inline int threadId() const { return m_threadId; }
inline Options::PowVersion powVersion() const { return m_powVersion; }
inline size_t size() const { return m_size; }
inline uint32_t *nonce() { return reinterpret_cast<uint32_t*>(m_blob + 39); }
inline uint32_t diff() const { return (uint32_t) m_diff; }
inline uint64_t target() const { return m_target; }
inline void setNicehash(bool nicehash) { m_nicehash = nicehash; }
inline void setThreadId(int threadId) { m_threadId = threadId; }
# ifdef XMRIG_PROXY_PROJECT
inline char *rawBlob() { return m_rawBlob; }
inline const char *rawTarget() const { return m_rawTarget; }
# endif
inline void setPowVersion(Options::PowVersion powVersion) { m_powVersion = powVersion; }
static bool fromHex(const char* in, unsigned int len, unsigned char* out);
static inline uint32_t *nonce(uint8_t *blob) { return reinterpret_cast<uint32_t*>(blob + 39); }
@ -79,6 +77,7 @@ private:
size_t m_size;
uint64_t m_diff;
uint64_t m_target;
Options::PowVersion m_powVersion;
# ifdef XMRIG_PROXY_PROJECT
VAR_ALIGN(16, char m_rawBlob[169]);

10
src/net/Network.cpp
View file

@ -57,7 +57,7 @@ Network::Network(const Options *options) :
const std::vector<Url*> &pools = options->pools();
#ifndef XMRIG_NO_TLS
ssl_init();
//ssl_init();
#endif
if (pools.size() > 1) {
@ -81,7 +81,7 @@ Network::Network(const Options *options) :
Network::~Network()
{
#ifndef XMRIG_NO_TLS
ssl_destroy();
//ssl_destroy();
#endif
}
@ -170,10 +170,12 @@ void Network::onResultAccepted(Client *client, const SubmitResult &result, const
void Network::setJob(Client *client, const Job &job)
{
if (m_options->colors()) {
LOG_INFO("\x1B[01;35mnew job\x1B[0m from \x1B[01;37m%s:%d\x1B[0m diff \x1B[01;37m%d", client->host(), client->port(), job.diff());
LOG_INFO("\x1B[01;35mnew job\x1B[0m from \x1B[01;37m%s:%d\x1B[0m with diff \x1B[01;37m%d\x1B[0m and PoW \x1B[01;37mv%d",
client->host(), client->port(), job.diff(), job.powVersion());
}
else {
LOG_INFO("new job from %s:%d diff %d", client->host(), client->port(), job.diff());
LOG_INFO("new job from %s:%d with diff %d and PoW v%d",
client->host(), client->port(), job.diff(), job.powVersion());
}
m_state.diff = job.diff();

2
src/net/strategies/DonateStrategy.cpp
View file

@ -81,7 +81,7 @@ DonateStrategy::DonateStrategy(const char *agent, IStrategyListener *listener) :
}
#endif
m_client = new Client(-1, agent, this);
m_client = std::make_shared<Client>(-1, agent, this);
m_client->setUrl(url);
m_client->setRetryPause(Options::i()->retryPause() * 1000);
m_client->setQuiet(true);

2
src/net/strategies/DonateStrategy.h
View file

@ -65,7 +65,7 @@ private:
static void onSuspendTimer(uv_timer_t *handle);
bool m_active;
Client *m_client;
Client::Ptr m_client;
const int m_donateTime;
const int m_idleTime;
IStrategyListener *m_listener;

6
src/net/strategies/FailoverStrategy.cpp
View file

@ -57,7 +57,7 @@ void FailoverStrategy::resume()
return;
}
m_listener->onJob( m_pools[m_active], m_pools[m_active]->job());
m_listener->onJob(m_pools[m_active].get(), m_pools[m_active]->job());
}
@ -76,7 +76,7 @@ void FailoverStrategy::stop()
void FailoverStrategy::tick(uint64_t now)
{
for (Client *client : m_pools) {
for (auto client : m_pools) {
client->tick(now);
}
}
@ -140,7 +140,7 @@ void FailoverStrategy::onResultAccepted(Client *client, const SubmitResult &resu
void FailoverStrategy::add(const Url *url, const char *agent)
{
Client *client = new Client((int) m_pools.size(), agent, this);
Client::Ptr client = std::make_shared<Client>((int) m_pools.size(), agent, this);
client->setUrl(url);
client->setRetryPause(Options::i()->retryPause() * 1000);

2
src/net/strategies/FailoverStrategy.h
View file

@ -63,7 +63,7 @@ private:
int m_active;
int m_index;
IStrategyListener *m_listener;
std::vector<Client*> m_pools;
std::vector<Client::Ptr> m_pools;
};
#endif /* __FAILOVERSTRATEGY_H__ */

4
src/net/strategies/SinglePoolStrategy.cpp
View file

@ -32,7 +32,7 @@ SinglePoolStrategy::SinglePoolStrategy(const Url *url, const char *agent, IStrat
m_active(false),
m_listener(listener)
{
m_client = new Client(0, agent, this);
m_client = std::make_shared<Client>(0, agent, this);
m_client->setUrl(url);
m_client->setRetryPause(Options::i()->retryPause() * 1000);
}
@ -56,7 +56,7 @@ void SinglePoolStrategy::resume()
return;
}
m_listener->onJob(m_client, m_client->job());
m_listener->onJob(m_client.get(), m_client->job());
}

2
src/net/strategies/SinglePoolStrategy.h
View file

@ -56,7 +56,7 @@ protected:
private:
bool m_active;
Client *m_client;
Client::Ptr m_client;
IStrategyListener *m_listener;
};