584 lines
19 KiB
C++
584 lines
19 KiB
C++
#include "EthGetworkClient.h"
|
|
|
|
#include <chrono>
|
|
|
|
#include <boost/bind/bind.hpp>
|
|
|
|
#include <ethash/ethash.hpp>
|
|
|
|
using namespace std;
|
|
using namespace dev;
|
|
using namespace eth;
|
|
|
|
using boost::asio::ip::tcp;
|
|
|
|
EthGetworkClient::EthGetworkClient(int worktimeout, unsigned farmRecheckPeriod)
|
|
: PoolClient(),
|
|
m_farmRecheckPeriod(farmRecheckPeriod),
|
|
m_io_strand(g_io_service),
|
|
m_socket(g_io_service),
|
|
m_resolver(g_io_service),
|
|
m_endpoints(),
|
|
m_getwork_timer(g_io_service),
|
|
m_worktimeout(worktimeout)
|
|
{
|
|
m_jSwBuilder.settings_["indentation"] = "";
|
|
|
|
Json::Value jGetWork;
|
|
jGetWork["id"] = unsigned(1);
|
|
jGetWork["jsonrpc"] = "2.0";
|
|
jGetWork["method"] = "eth_getWork";
|
|
jGetWork["params"] = Json::Value(Json::arrayValue);
|
|
m_jsonGetWork = std::string(Json::writeString(m_jSwBuilder, jGetWork));
|
|
}
|
|
|
|
EthGetworkClient::~EthGetworkClient()
|
|
{
|
|
// Do not stop io service.
|
|
// It's global
|
|
}
|
|
|
|
void EthGetworkClient::connect()
|
|
{
|
|
// Prevent unnecessary and potentially dangerous recursion
|
|
bool expected = false;
|
|
if (!m_connecting.compare_exchange_weak(expected, true, memory_order::memory_order_relaxed))
|
|
return;
|
|
|
|
// Reset status flags
|
|
m_getwork_timer.cancel();
|
|
|
|
// Initialize a new queue of end points
|
|
m_endpoints = std::queue<boost::asio::ip::basic_endpoint<boost::asio::ip::tcp>>();
|
|
m_endpoint = boost::asio::ip::basic_endpoint<boost::asio::ip::tcp>();
|
|
|
|
if (m_conn->HostNameType() == dev::UriHostNameType::Dns ||
|
|
m_conn->HostNameType() == dev::UriHostNameType::Basic)
|
|
{
|
|
// Begin resolve all ips associated to hostname
|
|
// calling the resolver each time is useful as most
|
|
// load balancers will give Ips in different order
|
|
m_resolver = boost::asio::ip::tcp::resolver(g_io_service);
|
|
boost::asio::ip::tcp::resolver::query q(m_conn->Host(), toString(m_conn->Port()));
|
|
|
|
// Start resolving async
|
|
m_resolver.async_resolve(
|
|
q, m_io_strand.wrap(boost::bind(&EthGetworkClient::handle_resolve, this,
|
|
boost::asio::placeholders::error, boost::asio::placeholders::iterator)));
|
|
}
|
|
else
|
|
{
|
|
// No need to use the resolver if host is already an IP address
|
|
m_endpoints.push(boost::asio::ip::tcp::endpoint(
|
|
boost::asio::ip::address::from_string(m_conn->Host()), m_conn->Port()));
|
|
send(m_jsonGetWork);
|
|
}
|
|
}
|
|
|
|
void EthGetworkClient::disconnect()
|
|
{
|
|
// Release session
|
|
m_connected.store(false, memory_order_relaxed);
|
|
m_conn->addDuration(m_session->duration());
|
|
m_session = nullptr;
|
|
|
|
m_connecting.store(false, std::memory_order_relaxed);
|
|
m_txPending.store(false, std::memory_order_relaxed);
|
|
m_getwork_timer.cancel();
|
|
|
|
m_txQueue.consume_all([](std::string* l) { delete l; });
|
|
m_request.consume(m_request.capacity());
|
|
m_response.consume(m_response.capacity());
|
|
|
|
if (m_onDisconnected)
|
|
m_onDisconnected();
|
|
}
|
|
|
|
void EthGetworkClient::begin_connect()
|
|
{
|
|
if (!m_endpoints.empty())
|
|
{
|
|
// Pick the first endpoint in list.
|
|
// Eventually endpoints get discarded on connection errors
|
|
m_endpoint = m_endpoints.front();
|
|
m_socket.async_connect(
|
|
m_endpoint, m_io_strand.wrap(boost::bind(&EthGetworkClient::handle_connect, this, boost::placeholders::_1)));
|
|
}
|
|
else
|
|
{
|
|
cwarn << "No more IP addresses to try for host: " << m_conn->Host();
|
|
disconnect();
|
|
}
|
|
}
|
|
|
|
void EthGetworkClient::handle_connect(const boost::system::error_code& ec)
|
|
{
|
|
if (!ec && m_socket.is_open())
|
|
{
|
|
|
|
// If in "connecting" phase raise the proper event
|
|
if (m_connecting.load(std::memory_order_relaxed))
|
|
{
|
|
// Initialize new session
|
|
m_connected.store(true, memory_order_relaxed);
|
|
m_session = unique_ptr<Session>(new Session);
|
|
m_session->subscribed.store(true, memory_order_relaxed);
|
|
m_session->authorized.store(true, memory_order_relaxed);
|
|
|
|
m_connecting.store(false, std::memory_order_relaxed);
|
|
|
|
if (m_onConnected)
|
|
m_onConnected();
|
|
m_current_tstamp = std::chrono::steady_clock::now();
|
|
}
|
|
|
|
// Retrieve 1st line waiting in the queue and submit
|
|
// if other lines waiting they will be processed
|
|
// at the end of the processed request
|
|
Json::Reader jRdr;
|
|
std::string* line;
|
|
std::ostream os(&m_request);
|
|
if (!m_txQueue.empty())
|
|
{
|
|
while (m_txQueue.pop(line))
|
|
{
|
|
if (line->size())
|
|
{
|
|
|
|
jRdr.parse(*line, m_pendingJReq);
|
|
m_pending_tstamp = std::chrono::steady_clock::now();
|
|
|
|
// Make sure path begins with "/"
|
|
string _path = (m_conn->Path().empty() ? "/" : m_conn->Path());
|
|
|
|
os << "POST " << _path << " HTTP/1.0\r\n";
|
|
os << "Host: " << m_conn->Host() << "\r\n";
|
|
os << "Content-Type: application/json"
|
|
<< "\r\n";
|
|
os << "Content-Length: " << line->length() << "\r\n";
|
|
os << "Connection: close\r\n\r\n"; // Double line feed to mark the
|
|
// beginning of body
|
|
// The payload
|
|
os << *line;
|
|
|
|
// Out received message only for debug purpouses
|
|
if (g_logOptions & LOG_JSON)
|
|
cnote << " >> " << *line;
|
|
|
|
delete line;
|
|
|
|
async_write(m_socket, m_request,
|
|
m_io_strand.wrap(boost::bind(&EthGetworkClient::handle_write, this,
|
|
boost::asio::placeholders::error)));
|
|
break;
|
|
}
|
|
delete line;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
m_txPending.store(false, std::memory_order_relaxed);
|
|
}
|
|
|
|
}
|
|
else
|
|
{
|
|
if (ec != boost::asio::error::operation_aborted)
|
|
{
|
|
// This endpoint does not respond
|
|
// Pop it and retry
|
|
cwarn << "Error connecting to " << m_conn->Host() << ":" << toString(m_conn->Port())
|
|
<< " : " << ec.message();
|
|
m_endpoints.pop();
|
|
begin_connect();
|
|
}
|
|
}
|
|
}
|
|
|
|
void EthGetworkClient::handle_write(const boost::system::error_code& ec)
|
|
{
|
|
if (!ec)
|
|
{
|
|
// Transmission succesfully sent.
|
|
// Read the response async.
|
|
async_read(m_socket, m_response, boost::asio::transfer_at_least(1),
|
|
m_io_strand.wrap(boost::bind(&EthGetworkClient::handle_read, this,
|
|
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)));
|
|
}
|
|
else
|
|
{
|
|
if (ec != boost::asio::error::operation_aborted)
|
|
{
|
|
cwarn << "Error writing to " << m_conn->Host() << ":" << toString(m_conn->Port())
|
|
<< " : " << ec.message();
|
|
m_endpoints.pop();
|
|
begin_connect();
|
|
}
|
|
}
|
|
}
|
|
|
|
void EthGetworkClient::handle_read(
|
|
const boost::system::error_code& ec, std::size_t bytes_transferred)
|
|
{
|
|
if (!ec)
|
|
{
|
|
// Close socket
|
|
if (m_socket.is_open())
|
|
m_socket.close();
|
|
|
|
// Get the whole message
|
|
std::string rx_message(
|
|
boost::asio::buffer_cast<const char*>(m_response.data()), bytes_transferred);
|
|
m_response.consume(bytes_transferred);
|
|
|
|
// Empty response ?
|
|
if (!rx_message.size())
|
|
{
|
|
cwarn << "Invalid response from " << m_conn->Host() << ":" << toString(m_conn->Port());
|
|
disconnect();
|
|
return;
|
|
}
|
|
|
|
// Read message by lines.
|
|
// First line is http status
|
|
// Other lines are headers
|
|
// A double "\r\n" identifies begin of body
|
|
// The rest is body
|
|
std::string line;
|
|
std::string linedelimiter = "\r\n";
|
|
std::size_t delimiteroffset = rx_message.find(linedelimiter);
|
|
|
|
unsigned int linenum = 0;
|
|
bool isHeader = true;
|
|
while (rx_message.length() && delimiteroffset != std::string::npos)
|
|
{
|
|
linenum++;
|
|
line = rx_message.substr(0, delimiteroffset);
|
|
rx_message.erase(0, delimiteroffset + 2);
|
|
|
|
// This identifies the beginning of body
|
|
if (line.empty())
|
|
{
|
|
isHeader = false;
|
|
delimiteroffset = rx_message.find(linedelimiter);
|
|
if (delimiteroffset != std::string::npos)
|
|
continue;
|
|
boost::replace_all(rx_message, "\n", "");
|
|
line = rx_message;
|
|
}
|
|
|
|
// Http status
|
|
if (isHeader && linenum == 1)
|
|
{
|
|
if (line.substr(0, 7) != "HTTP/1.")
|
|
{
|
|
cwarn << "Invalid response from " << m_conn->Host() << ":"
|
|
<< toString(m_conn->Port());
|
|
disconnect();
|
|
return;
|
|
}
|
|
std::size_t spaceoffset = line.find(' ');
|
|
if (spaceoffset == std::string::npos)
|
|
{
|
|
cwarn << "Invalid response from " << m_conn->Host() << ":"
|
|
<< toString(m_conn->Port());
|
|
disconnect();
|
|
return;
|
|
}
|
|
std::string status = line.substr(spaceoffset + 1);
|
|
if (status.substr(0, 3) != "200")
|
|
{
|
|
cwarn << m_conn->Host() << ":" << toString(m_conn->Port())
|
|
<< " reported status " << status;
|
|
disconnect();
|
|
return;
|
|
}
|
|
}
|
|
|
|
// Body
|
|
if (!isHeader)
|
|
{
|
|
// Out received message only for debug purpouses
|
|
if (g_logOptions & LOG_JSON)
|
|
cnote << " << " << line;
|
|
|
|
// Test validity of chunk and process
|
|
Json::Value jRes;
|
|
Json::Reader jRdr;
|
|
if (jRdr.parse(line, jRes))
|
|
{
|
|
// Run in sync so no 2 different async reads may overlap
|
|
processResponse(jRes);
|
|
}
|
|
else
|
|
{
|
|
string what = jRdr.getFormattedErrorMessages();
|
|
boost::replace_all(what, "\n", " ");
|
|
cwarn << "Got invalid Json message : " << what;
|
|
}
|
|
|
|
}
|
|
|
|
delimiteroffset = rx_message.find(linedelimiter);
|
|
}
|
|
|
|
// Is there anything else in the queue
|
|
if (!m_txQueue.empty())
|
|
{
|
|
begin_connect();
|
|
}
|
|
else
|
|
{
|
|
// Signal end of async send/receive operations
|
|
m_txPending.store(false, std::memory_order_relaxed);
|
|
}
|
|
|
|
}
|
|
else
|
|
{
|
|
if (ec != boost::asio::error::operation_aborted)
|
|
{
|
|
cwarn << "Error reading from :" << m_conn->Host() << ":" << toString(m_conn->Port())
|
|
<< " : "
|
|
<< ec.message();
|
|
disconnect();
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
void EthGetworkClient::handle_resolve(
|
|
const boost::system::error_code& ec, tcp::resolver::iterator i)
|
|
{
|
|
if (!ec)
|
|
{
|
|
while (i != tcp::resolver::iterator())
|
|
{
|
|
m_endpoints.push(i->endpoint());
|
|
i++;
|
|
}
|
|
m_resolver.cancel();
|
|
|
|
// Resolver has finished so invoke connection asynchronously
|
|
send(m_jsonGetWork);
|
|
}
|
|
else
|
|
{
|
|
cwarn << "Could not resolve host " << m_conn->Host() << ", " << ec.message();
|
|
disconnect();
|
|
}
|
|
}
|
|
|
|
void EthGetworkClient::processResponse(Json::Value& JRes)
|
|
{
|
|
unsigned _id = 0; // This SHOULD be the same id as the request it is responding to
|
|
bool _isSuccess = false; // Whether or not this is a succesful or failed response
|
|
string _errReason = ""; // Content of the error reason
|
|
|
|
if (!JRes.isMember("id"))
|
|
{
|
|
cwarn << "Missing id member in response from " << m_conn->Host() << ":"
|
|
<< toString(m_conn->Port());
|
|
return;
|
|
}
|
|
// We get the id from pending jrequest
|
|
// It's not guaranteed we get response labelled with same id
|
|
// For instance Dwarfpool always responds with "id":0
|
|
_id = m_pendingJReq.get("id", unsigned(0)).asUInt();
|
|
_isSuccess = JRes.get("error", Json::Value::null).empty();
|
|
_errReason = (_isSuccess ? "" : processError(JRes));
|
|
|
|
// We have only theese possible ids
|
|
// 0 or 1 as job notification
|
|
// 9 as response for eth_submitHashrate
|
|
// 40+ for responses to mining submissions
|
|
if (_id == 0 || _id == 1)
|
|
{
|
|
// Getwork might respond with an error to
|
|
// a request. (eg. node is still syncing)
|
|
// In such case delay further requests
|
|
// by 30 seconds.
|
|
// Otherwise resubmit another getwork request
|
|
// with a delay of m_farmRecheckPeriod ms.
|
|
if (!_isSuccess)
|
|
{
|
|
cwarn << "Got " << _errReason << " from " << m_conn->Host() << ":"
|
|
<< toString(m_conn->Port());
|
|
m_getwork_timer.expires_from_now(boost::posix_time::seconds(30));
|
|
m_getwork_timer.async_wait(
|
|
m_io_strand.wrap(boost::bind(&EthGetworkClient::getwork_timer_elapsed, this,
|
|
boost::asio::placeholders::error)));
|
|
}
|
|
else
|
|
{
|
|
if (!JRes.isMember("result"))
|
|
{
|
|
cwarn << "Missing data for eth_getWork request from " << m_conn->Host() << ":"
|
|
<< toString(m_conn->Port());
|
|
}
|
|
else
|
|
{
|
|
Json::Value JPrm = JRes.get("result", Json::Value::null);
|
|
WorkPackage newWp;
|
|
|
|
newWp.header = h256(JPrm.get(Json::Value::ArrayIndex(0), "").asString());
|
|
newWp.seed = h256(JPrm.get(Json::Value::ArrayIndex(1), "").asString());
|
|
newWp.boundary = h256(JPrm.get(Json::Value::ArrayIndex(2), "").asString());
|
|
newWp.block = strtoul(JPrm.get(Json::Value::ArrayIndex(3), "").asString().c_str(), nullptr, 0);
|
|
newWp.job = newWp.header.hex();
|
|
if (m_current.header != newWp.header)
|
|
{
|
|
m_current = newWp;
|
|
m_current_tstamp = std::chrono::steady_clock::now();
|
|
|
|
if (m_onWorkReceived)
|
|
m_onWorkReceived(m_current);
|
|
}
|
|
m_getwork_timer.expires_from_now(boost::posix_time::milliseconds(m_farmRecheckPeriod));
|
|
m_getwork_timer.async_wait(
|
|
m_io_strand.wrap(boost::bind(&EthGetworkClient::getwork_timer_elapsed, this,
|
|
boost::asio::placeholders::error)));
|
|
}
|
|
}
|
|
|
|
}
|
|
else if (_id == 9)
|
|
{
|
|
// Response to hashrate submission
|
|
// Actually don't do anything
|
|
}
|
|
else if (_id >= 40 && _id <= m_solution_submitted_max_id)
|
|
{
|
|
if (_isSuccess && JRes["result"].isConvertibleTo(Json::ValueType::booleanValue))
|
|
_isSuccess = JRes["result"].asBool();
|
|
|
|
std::chrono::milliseconds _delay = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
std::chrono::steady_clock::now() - m_pending_tstamp);
|
|
|
|
const unsigned miner_index = _id - 40;
|
|
if (_isSuccess)
|
|
{
|
|
if (m_onSolutionAccepted)
|
|
m_onSolutionAccepted(_delay, miner_index, false);
|
|
}
|
|
else
|
|
{
|
|
if (m_onSolutionRejected)
|
|
m_onSolutionRejected(_delay, miner_index);
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
std::string EthGetworkClient::processError(Json::Value& JRes)
|
|
{
|
|
std::string retVar;
|
|
|
|
if (JRes.isMember("error") &&
|
|
!JRes.get("error", Json::Value::null).isNull())
|
|
{
|
|
if (JRes["error"].isConvertibleTo(Json::ValueType::stringValue))
|
|
{
|
|
retVar = JRes.get("error", "Unknown error").asString();
|
|
}
|
|
else if (JRes["error"].isConvertibleTo(Json::ValueType::arrayValue))
|
|
{
|
|
for (auto i : JRes["error"])
|
|
{
|
|
retVar += i.asString() + " ";
|
|
}
|
|
}
|
|
else if (JRes["error"].isConvertibleTo(Json::ValueType::objectValue))
|
|
{
|
|
for (Json::Value::iterator i = JRes["error"].begin(); i != JRes["error"].end(); ++i)
|
|
{
|
|
Json::Value k = i.key();
|
|
Json::Value v = (*i);
|
|
retVar += (std::string)i.name() + ":" + v.asString() + " ";
|
|
}
|
|
}
|
|
}
|
|
else
|
|
{
|
|
retVar = "Unknown error";
|
|
}
|
|
|
|
return retVar;
|
|
}
|
|
|
|
void EthGetworkClient::send(Json::Value const& jReq)
|
|
{
|
|
send(std::string(Json::writeString(m_jSwBuilder, jReq)));
|
|
}
|
|
|
|
void EthGetworkClient::send(std::string const& sReq)
|
|
{
|
|
std::string* line = new std::string(sReq);
|
|
m_txQueue.push(line);
|
|
|
|
bool ex = false;
|
|
if (m_txPending.compare_exchange_weak(ex, true, std::memory_order_relaxed))
|
|
begin_connect();
|
|
}
|
|
|
|
void EthGetworkClient::submitHashrate(uint64_t const& rate, string const& id)
|
|
{
|
|
// No need to check for authorization
|
|
if (m_session)
|
|
{
|
|
Json::Value jReq;
|
|
jReq["id"] = unsigned(9);
|
|
jReq["jsonrpc"] = "2.0";
|
|
jReq["method"] = "eth_submitHashrate";
|
|
jReq["params"] = Json::Value(Json::arrayValue);
|
|
jReq["params"].append(toHex(rate, HexPrefix::Add)); // Already expressed as hex
|
|
jReq["params"].append(id); // Already prefixed by 0x
|
|
send(jReq);
|
|
}
|
|
|
|
}
|
|
|
|
void EthGetworkClient::submitSolution(const Solution& solution)
|
|
{
|
|
|
|
if (m_session)
|
|
{
|
|
Json::Value jReq;
|
|
string nonceHex = toHex(solution.nonce);
|
|
|
|
unsigned id = 40 + solution.midx;
|
|
jReq["id"] = id;
|
|
jReq["jsonrpc"] = "2.0";
|
|
m_solution_submitted_max_id = max(m_solution_submitted_max_id, id);
|
|
jReq["method"] = "eth_submitWork";
|
|
jReq["params"] = Json::Value(Json::arrayValue);
|
|
jReq["params"].append("0x" + nonceHex);
|
|
jReq["params"].append("0x" + solution.work.header.hex());
|
|
jReq["params"].append("0x" + solution.mixHash.hex());
|
|
send(jReq);
|
|
}
|
|
|
|
}
|
|
|
|
void EthGetworkClient::getwork_timer_elapsed(const boost::system::error_code& ec)
|
|
{
|
|
// Triggers the resubmission of a getWork request
|
|
if (!ec)
|
|
{
|
|
// Check if last work is older than timeout
|
|
std::chrono::seconds _delay = std::chrono::duration_cast<std::chrono::seconds>(
|
|
std::chrono::steady_clock::now() - m_current_tstamp);
|
|
if (_delay.count() > m_worktimeout)
|
|
{
|
|
cwarn << "No new work received in " << m_worktimeout << " seconds.";
|
|
m_endpoints.pop();
|
|
disconnect();
|
|
}
|
|
else
|
|
{
|
|
send(m_jsonGetWork);
|
|
}
|
|
|
|
}
|
|
}
|