Added embedded miner

pull/112/head
SChernykh 2 years ago
parent 95c268be4c
commit e2931f94b5

@ -27,6 +27,7 @@ set(HEADERS
src/keccak.h
src/log.h
src/mempool.h
src/miner.h
src/p2p_server.h
src/p2pool.h
src/p2pool_api.h
@ -60,6 +61,7 @@ set(SOURCES
src/main.cpp
src/memory_leak_debug.cpp
src/mempool.cpp
src/miner.cpp
src/p2p_server.cpp
src/p2pool.cpp
src/p2pool_api.cpp

@ -1002,7 +1002,7 @@ void BlockTemplate::submit_sidechain_block(uint32_t template_id, uint32_t nonce,
m_poolBlockTemplate->m_nonce = nonce;
m_poolBlockTemplate->m_extraNonce = extra_nonce;
memcpy(m_poolBlockTemplate->m_mainChainData.data() + m_nonceOffset, &nonce, NONCE_SIZE);
memcpy(m_poolBlockTemplate->m_mainChainData.data() + m_extraNonceOffsetInTemplate, &extra_nonce, NONCE_SIZE);
memcpy(m_poolBlockTemplate->m_mainChainData.data() + m_extraNonceOffsetInTemplate, &extra_nonce, EXTRA_NONCE_SIZE);
SideChain& side_chain = m_pool->side_chain();

@ -21,6 +21,7 @@
#include "p2pool.h"
#include "stratum_server.h"
#include "p2p_server.h"
#include "miner.h"
#include "side_chain.h"
#include <iostream>
@ -69,7 +70,7 @@ typedef struct cmd {
cmdfunc *func;
} cmd;
static cmdfunc do_help, do_status, do_loglevel, do_addpeers, do_droppeers, do_showpeers, do_outpeers, do_inpeers, do_exit;
static cmdfunc do_help, do_status, do_loglevel, do_addpeers, do_droppeers, do_showpeers, do_outpeers, do_inpeers, do_start_mining, do_stop_mining, do_exit;
static cmd cmds[] = {
{ STRCONST("help"), "", "display list of commands", do_help },
@ -80,17 +81,18 @@ static cmd cmds[] = {
{ STRCONST("peers"), "", "show all peers", do_showpeers },
{ STRCONST("outpeers"), "", "set maximum number of outgoing connections", do_outpeers },
{ STRCONST("inpeers"), "", "set maximum number of incoming connections", do_inpeers },
{ STRCONST("start_mining"), "<threads>", "start mining", do_start_mining },
{ STRCONST("stop_mining"), "", "stop mining", do_stop_mining },
{ STRCONST("exit"), "", "terminate p2pool", do_exit },
{ STRCNULL, NULL, NULL, NULL }
};
static int do_help(p2pool * /* m_pool */, const char * /* args */)
{
int i;
LOGINFO(0, "List of commands");
for (i=0; cmds[i].name.len; i++)
for (int i = 0; cmds[i].name.len; ++i) {
LOGINFO(0, cmds[i].name.str << " " << cmds[i].arg << "\t" << cmds[i].descr);
}
return 0;
}
@ -103,13 +105,16 @@ static int do_status(p2pool *m_pool, const char * /* args */)
if (m_pool->p2p_server()) {
m_pool->p2p_server()->print_status();
}
if (m_pool->miner()) {
m_pool->miner()->print_status();
}
bkg_jobs_tracker.print_status();
return 0;
}
static int do_loglevel(p2pool * /* m_pool */, const char *args)
{
int level = atoi(args);
int level = strtol(args, nullptr, 10);
level = std::min(std::max(level, 0), log::MAX_GLOBAL_LOG_LEVEL);
log::GLOBAL_LOG_LEVEL = level;
LOGINFO(0, "log level set to " << level);
@ -158,6 +163,20 @@ static int do_inpeers(p2pool* m_pool, const char* args)
return 0;
}
static int do_start_mining(p2pool* m_pool, const char* args)
{
uint32_t threads = strtoul(args, nullptr, 10);
threads = std::min(std::max(threads, 1u), 64u);
m_pool->start_mining(threads);
return 0;
}
static int do_stop_mining(p2pool* m_pool, const char* /*args*/)
{
m_pool->stop_mining();
return 0;
}
static int do_exit(p2pool *m_pool, const char * /* args */)
{
bkg_jobs_tracker.wait();

@ -40,8 +40,9 @@ static void usage()
"--no-cache Disable p2pool.cache\n"
"--no-color Disable colors in console output\n"
"--no-randomx Disable internal RandomX hasher: p2pool will use RPC calls to monerod to check PoW hashes\n"
"--out-peers Maximum number of outgoing connections for p2p server (any value between 10 and 1000)\n"
"--in-peers Maximum number of incoming connections for p2p server (any value between 10 and 1000)\n"
"--out-peers N Maximum number of outgoing connections for p2p server (any value between 10 and 1000)\n"
"--in-peers N Maximum number of incoming connections for p2p server (any value between 10 and 1000)\n"
"--start-mining N Start built-in miner using N threads (any value between 1 and 64)\n"
"--help Show this help message\n\n"
"Example command line:\n\n"
"%s --host 127.0.0.1 --rpc-port 18081 --zmq-port 18083 --wallet YOUR_WALLET_ADDRESS --stratum 0.0.0.0:%d --p2p 0.0.0.0:%d\n\n",

@ -0,0 +1,196 @@
/*
* This file is part of the Monero P2Pool <https://github.com/SChernykh/p2pool>
* Copyright (c) 2021 SChernykh <https://github.com/SChernykh>
*
* 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, version 3.
*
* 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 "common.h"
#include "miner.h"
#include "p2pool.h"
#include "stratum_server.h"
#include "block_template.h"
#include "pow_hash.h"
#include "randomx.h"
#include <thread>
static constexpr char log_category_prefix[] = "Miner ";
namespace p2pool {
Miner::Miner(p2pool* pool, uint32_t threads)
: m_pool(pool)
, m_threads(threads)
, m_stopped(false)
, m_nonce(0)
, m_nonceTimestamp(std::chrono::high_resolution_clock::now())
, m_extraNonce(0xF19E3779U)
, m_job{}
{
on_block(m_pool->block_template());
m_minerThreads.clear();
m_minerThreads.reserve(threads);
for (uint32_t i = 0; i < threads; ++i) {
WorkerData* data = new WorkerData{ this, i + 1, threads, {} };
const int err = uv_thread_create(&data->m_worker, run, data);
if (err) {
LOGERR(1, "failed to start worker thread " << data->m_index << '/' << threads << ", error " << uv_err_name(err));
delete data;
continue;
}
m_minerThreads.push_back(data);
}
}
Miner::~Miner()
{
m_stopped = true;
for (WorkerData* data : m_minerThreads) {
uv_thread_join(&data->m_worker);
delete data;
}
m_minerThreads.clear();
}
void Miner::print_status()
{
const uint32_t hash_count = 0 - m_nonce.load();
using namespace std::chrono;
const double dt = static_cast<double>(duration_cast<nanoseconds>(high_resolution_clock::now() - m_nonceTimestamp).count()) / 1e9;
const uint64_t hr = (dt > 0.0) ? static_cast<uint64_t>(hash_count / dt) : 0;
LOGINFO(0, "status" <<
"\nThreads = " << m_threads <<
"\nHashrate = " << log::Hashrate(hr)
);
}
void Miner::on_block(const BlockTemplate& block)
{
Job j;
hash seed;
j.m_blobSize = block.get_hashing_blob(m_extraNonce, j.m_blob, j.m_height, j.m_diff, j.m_sidechainDiff, seed, j.m_nonceOffset, j.m_templateId);
memcpy(&m_job, &j, sizeof(j));
m_nonce.exchange(0);
m_nonceTimestamp = std::chrono::high_resolution_clock::now();
}
void Miner::run(void* data)
{
WorkerData* d = static_cast<WorkerData*>(data);
LOGINFO(1, "worker thread " << d->m_index << '/' << d->m_count << " started");
make_thread_background();
d->m_miner->run(d);
LOGINFO(1, "worker thread " << d->m_index << '/' << d->m_count << " stopped");
}
void Miner::run(WorkerData* data)
{
RandomX_Hasher_Base* hasher = m_pool->hasher();
randomx_cache* cache = hasher->cache();
randomx_dataset* dataset = hasher->dataset();
if (!cache && !dataset) {
LOGERR(1, "worker thread " << data->m_index << '/' << data->m_count << ": RandomX cache and dataset are not ready");
return;
}
randomx_flags flags = randomx_get_flags();
if (dataset) {
flags |= RANDOMX_FLAG_FULL_MEM;
}
randomx_vm* vm = randomx_create_vm(flags | RANDOMX_FLAG_LARGE_PAGES, dataset ? nullptr : cache, dataset);
if (!vm) {
LOGWARN(1, "couldn't allocate RandomX VM using large pages");
vm = randomx_create_vm(flags, dataset ? nullptr : cache, dataset);
if (!vm) {
LOGERR(1, "couldn't allocate RandomX VM");
return;
}
}
uint32_t index = 0;
Job job[2];
uint32_t seed_counter = 0;
bool first = true;
Miner* miner = data->m_miner;
while (!m_stopped) {
if (hasher->seed_counter() != seed_counter) {
LOGINFO(5, "worker thread " << data->m_index << '/' << data->m_count << " paused (waiting for RandomX cache/dataset update)");
hasher->sync_wait();
seed_counter = hasher->seed_counter();
if (flags & RANDOMX_FLAG_FULL_MEM) {
dataset = hasher->dataset();
randomx_vm_set_dataset(vm, dataset);
}
else {
cache = hasher->cache();
randomx_vm_set_cache(vm, cache);
}
LOGINFO(5, "worker thread " << data->m_index << '/' << data->m_count << " resumed");
}
if (first) {
first = false;
memcpy(&job[index], &miner->m_job, sizeof(m_job));
job[index].set_nonce(miner->m_nonce.fetch_sub(1), miner->m_extraNonce);
randomx_calculate_hash_first(vm, job[index].m_blob, job[index].m_blobSize);
}
const Job& j = job[index];
index ^= 1;
memcpy(&job[index], &miner->m_job, sizeof(m_job));
job[index].set_nonce(miner->m_nonce.fetch_sub(1), miner->m_extraNonce);
hash h;
randomx_calculate_hash_next(vm, job[index].m_blob, job[index].m_blobSize, &h);
if (j.m_diff.check_pow(h)) {
LOGINFO(0, log::Green() << "worker thread " << data->m_index << '/' << data->m_count << " found a mainchain block, submitting it");
m_pool->submit_block_async(j.m_templateId, j.m_nonce, j.m_extraNonce);
m_pool->block_template().update_tx_keys();
}
if (j.m_sidechainDiff.check_pow(h)) {
LOGINFO(0, log::Green() << "SHARE FOUND: mainchain height " << j.m_height << ", diff " << j.m_sidechainDiff << ", worker thread " << data->m_index << '/' << data->m_count);
m_pool->submit_sidechain_block(j.m_templateId, j.m_nonce, j.m_extraNonce);
}
std::this_thread::yield();
}
randomx_destroy_vm(vm);
}
void Miner::Job::set_nonce(uint32_t nonce, uint32_t extra_nonce)
{
m_nonce = nonce;
m_extraNonce = extra_nonce;
uint8_t* p = m_blob + m_nonceOffset;
p[0] = static_cast<uint8_t>(nonce);
p[1] = static_cast<uint8_t>(nonce >> 8);
p[2] = static_cast<uint8_t>(nonce >> 16);
p[3] = static_cast<uint8_t>(nonce >> 24);
}
} // namespace p2pool

@ -0,0 +1,77 @@
/*
* This file is part of the Monero P2Pool <https://github.com/SChernykh/p2pool>
* Copyright (c) 2021 SChernykh <https://github.com/SChernykh>
*
* 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, version 3.
*
* 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/>.
*/
#pragma once
#include "uv_util.h"
#include <chrono>
namespace p2pool {
class p2pool;
class BlockTemplate;
class Miner
{
public:
Miner(p2pool* pool, uint32_t threads);
~Miner();
void print_status();
void on_block(const BlockTemplate& block);
private:
static void run(void* data);
p2pool* m_pool;
uint32_t m_threads;
struct WorkerData
{
Miner* m_miner;
uint32_t m_index;
uint32_t m_count;
uv_thread_t m_worker;
};
std::vector<WorkerData*> m_minerThreads;
volatile bool m_stopped;
std::atomic<uint32_t> m_nonce;
std::chrono::time_point<std::chrono::high_resolution_clock> m_nonceTimestamp;
const uint32_t m_extraNonce;
struct Job
{
uint8_t m_blob[128] = {};
uint32_t m_blobSize = 0;
uint32_t m_templateId = 0;
difficulty_type m_diff = {};
difficulty_type m_sidechainDiff = {};
uint64_t m_height = 0;
size_t m_nonceOffset = 0;
uint32_t m_nonce = 0;
uint32_t m_extraNonce = 0;
void set_nonce(uint32_t nonce, uint32_t extra_nonce);
};
Job m_job;
void run(WorkerData* data);
};
} // namespace p2pool

@ -27,6 +27,7 @@
#include "side_chain.h"
#include "stratum_server.h"
#include "p2p_server.h"
#include "miner.h"
#include "params.h"
#include "console_commands.h"
#include "crypto.h"
@ -590,6 +591,9 @@ void p2pool::download_block_headers(uint64_t current_height)
m_ZMQReader = new ZMQReader(m_params->m_host.c_str(), m_params->m_zmqPort, this);
m_stratumServer = new StratumServer(this);
m_p2pServer = new P2PServer(this);
if (m_params->m_minerThreads) {
start_mining(m_params->m_minerThreads);
}
api_update_network_stats();
}
}
@ -656,22 +660,12 @@ void p2pool::update_median_timestamp()
void p2pool::stratum_on_block()
{
#if 0
uint8_t hashing_blob[128];
uint64_t height;
difficulty_type difficulty;
difficulty_type sidechain_difficulty;
hash seed_hash;
size_t nonce_offset;
uint32_t template_id;
m_blockTemplate->get_hashing_blob(0, hashing_blob, height, difficulty, sidechain_difficulty, seed_hash, nonce_offset, template_id);
submit_block(template_id, 0, 0);
#else
if (m_miner) {
m_miner->on_block(*m_blockTemplate);
}
if (m_stratumServer) {
m_stratumServer->on_block(*m_blockTemplate);
}
#endif
}
void p2pool::get_info()
@ -1214,6 +1208,21 @@ bool p2pool::get_difficulty_at_height(uint64_t height, difficulty_type& diff)
return true;
}
void p2pool::start_mining(uint32_t threads)
{
stop_mining();
m_miner = new Miner(this, threads);
}
void p2pool::stop_mining()
{
Miner* miner = m_miner;
if (miner) {
m_miner = nullptr;
delete miner;
}
}
static void on_signal(uv_signal_t* handle, int signum)
{
p2pool* pool = reinterpret_cast<p2pool*>(handle->data);
@ -1322,6 +1331,7 @@ int p2pool::run()
bkg_jobs_tracker.wait();
delete m_miner;
delete m_stratumServer;
delete m_p2pServer;

@ -29,6 +29,7 @@ class Mempool;
class SideChain;
class StratumServer;
class P2PServer;
class Miner;
class ConsoleCommands;
class p2pool_api;
class ZMQReader;
@ -58,6 +59,7 @@ public:
StratumServer* stratum_server() const { return m_stratumServer; }
P2PServer* p2p_server() const { return m_p2pServer; }
Miner* miner() const { return m_miner; }
virtual void handle_tx(TxMempoolData& tx) override;
virtual void handle_miner_data(MinerData& data) override;
@ -78,6 +80,9 @@ public:
bool get_difficulty_at_height(uint64_t height, difficulty_type& diff);
void start_mining(uint32_t threads);
void stop_mining();
time_t zmq_last_active() const { return m_zmqLastActive; }
time_t start_time() const { return m_startTime; }
@ -155,6 +160,7 @@ private:
std::atomic<uint32_t> m_serversStarted{ 0 };
StratumServer* m_stratumServer = nullptr;
P2PServer* m_p2pServer = nullptr;
Miner* m_miner = nullptr;
ConsoleCommands* m_consoleCommands;

@ -93,6 +93,10 @@ Params::Params(int argc, char* argv[])
if ((strcmp(argv[i], "--in-peers") == 0) && (i + 1 < argc)) {
m_maxIncomingPeers = std::min(std::max(strtoul(argv[++i], nullptr, 10), 10UL), 1000UL);
}
if ((strcmp(argv[i], "--start-mining") == 0) && (i + 1 < argc)) {
m_minerThreads = std::min(std::max(strtoul(argv[++i], nullptr, 10), 1UL), 64UL);
}
}
if (m_stratumAddresses.empty()) {

@ -42,6 +42,7 @@ struct Params
bool m_disableRandomX = false;
uint32_t m_maxOutgoingPeers = 10;
uint32_t m_maxIncomingPeers = 1000;
uint32_t m_minerThreads = 0;
};
} // namespace p2pool

@ -37,7 +37,7 @@ RandomX_Hasher::RandomX_Hasher(p2pool* pool)
, m_dataset(nullptr)
, m_seed{}
, m_index(0)
, m_setSeedCounter(0)
, m_seedCounter(0)
{
uint64_t memory_allocated = 0;
@ -119,6 +119,10 @@ RandomX_Hasher::~RandomX_Hasher()
void RandomX_Hasher::set_seed_async(const hash& seed)
{
if (m_seed[m_index] == seed) {
return;
}
struct Work
{
p2pool* pool;
@ -127,10 +131,7 @@ void RandomX_Hasher::set_seed_async(const hash& seed)
uv_work_t req;
};
Work* work = new Work{};
work->pool = m_pool;
work->hasher = this;
work->seed = seed;
Work* work = new Work{ m_pool, this, seed, {} };
work->req.data = work;
const int err = uv_queue_work(uv_default_loop_checked(), &work->req,
@ -167,7 +168,7 @@ void RandomX_Hasher::set_seed(const hash& seed)
WriteLock lock(m_datasetLock);
uv_rwlock_wrlock(&m_cacheLock);
m_setSeedCounter.fetch_add(1);
m_seedCounter.fetch_add(1);
if (m_seed[m_index] == seed) {
uv_rwlock_wrunlock(&m_cacheLock);
@ -269,7 +270,7 @@ void RandomX_Hasher::set_seed(const hash& seed)
void RandomX_Hasher::set_old_seed(const hash& seed)
{
// set_seed() must go first, wait for it
while (m_setSeedCounter.load() == 0) {
while (m_seedCounter.load() == 0) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
@ -305,6 +306,12 @@ void RandomX_Hasher::set_old_seed(const hash& seed)
LOGINFO(1, log::LightCyan() << "old cache updated");
}
void RandomX_Hasher::sync_wait()
{
ReadLock lock(m_datasetLock);
ReadLock lock2(m_cacheLock);
}
bool RandomX_Hasher::calculate(const void* data, size_t size, uint64_t /*height*/, const hash& seed, hash& result)
{
// First try to use the dataset if it's ready
@ -353,7 +360,6 @@ bool RandomX_Hasher::calculate(const void* data, size_t size, uint64_t /*height*
RandomX_Hasher_RPC::RandomX_Hasher_RPC(p2pool* pool)
: m_pool(pool)
, m_loopStopped(false)
, m_loopThread{}
{
int err = uv_loop_init(&m_loop);
@ -385,11 +391,7 @@ RandomX_Hasher_RPC::RandomX_Hasher_RPC(p2pool* pool)
RandomX_Hasher_RPC::~RandomX_Hasher_RPC()
{
uv_async_send(&m_shutdownAsync);
using namespace std::chrono;
while (!m_loopStopped) {
std::this_thread::sleep_for(milliseconds(1));
}
uv_thread_join(&m_loopThread);
uv_mutex_destroy(&m_requestMutex);
uv_mutex_destroy(&m_condMutex);
@ -405,7 +407,6 @@ void RandomX_Hasher_RPC::loop(void* data)
uv_run(&hasher->m_loop, UV_RUN_DEFAULT);
uv_loop_close(&hasher->m_loop);
LOGINFO(1, "event loop stopped");
hasher->m_loopStopped = true;
}
bool RandomX_Hasher_RPC::calculate(const void* data_ptr, size_t size, uint64_t height, const hash& /*seed*/, hash& h)

@ -35,6 +35,11 @@ public:
virtual void set_seed_async(const hash&) {}
virtual void set_old_seed(const hash&) {}
virtual randomx_cache* cache() const { return nullptr; }
virtual randomx_dataset* dataset() const { return nullptr; }
virtual uint32_t seed_counter() const { return 0; }
virtual void sync_wait() {}
virtual bool calculate(const void* data, size_t size, uint64_t height, const hash& seed, hash& result) = 0;
};
@ -49,6 +54,11 @@ public:
void set_old_seed(const hash& seed) override;
randomx_cache* cache() const override { return m_cache[m_index]; }
randomx_dataset* dataset() const override { return m_dataset; }
uint32_t seed_counter() const override { return m_seedCounter.load(); }
void sync_wait() override;
bool calculate(const void* data, size_t size, uint64_t height, const hash& seed, hash& result) override;
private:
@ -78,7 +88,7 @@ private:
hash m_seed[2];
uint32_t m_index;
std::atomic<uint32_t> m_setSeedCounter;
std::atomic<uint32_t> m_seedCounter;
};
class RandomX_Hasher_RPC : public RandomX_Hasher_Base
@ -96,7 +106,6 @@ private:
uv_mutex_t m_requestMutex;
uv_loop_t m_loop;
volatile bool m_loopStopped;
uv_thread_t m_loopThread;
uv_mutex_t m_condMutex;

@ -86,6 +86,7 @@ set(SOURCES
../src/log.cpp
../src/memory_leak_debug.cpp
../src/mempool.cpp
../src/miner.cpp
../src/p2p_server.cpp
../src/p2pool.cpp
../src/p2pool_api.cpp

Loading…
Cancel
Save