Git-Mirrors/monero
1
0
Fork 0
mirror of https://github.com/monero-project/monero.git synced 2025-08-18 14:27:56 -04:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #2149

Browse source 
158c3ecf core: thread most of handle_incoming_tx (moneromooo-monero)
f57ee382 cryptonote_protocol: retry stale spans early (moneromooo-monero)
90df52e1 cryptonote_protocol: light cleanup (moneromooo-monero)
84e23156 cryptonote_protocol: avoid spurious SYNCHRONIZED OK messages (moneromooo-monero)
5be43fcd cryptonote_protocol_handler: sync speedup (moneromooo-monero)
This commit is contained in:
Riccardo Spagni 2017-08-07 15:24:58 +02:00
commit 6db8a60a18
No known key found for this signature in database
GPG key ID: 55432DF31CCD4FCD
26 changed files with 1616 additions and 186 deletions
Download patch file Download diff file Expand all files Collapse all files

628
src/cryptonote_protocol/cryptonote_protocol_handler.inl
View file

@ -48,6 +48,10 @@
#define MLOG_P2P_MESSAGE(x) MCINFO("net.p2p.msg", context << x)
#define BLOCK_QUEUE_NBLOCKS_THRESHOLD 10 // chunks of N blocks
#define BLOCK_QUEUE_SIZE_THRESHOLD (100*1024*1024) // MB
#define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD (5 * 1000000) // microseconds
namespace cryptonote
{
@ -233,6 +237,8 @@ namespace cryptonote
cnx.current_download = cntxt.m_current_speed_down / 1024;
cnx.current_upload = cntxt.m_current_speed_up / 1024;
cnx.connection_id = cntxt.m_connection_id;
connections.push_back(cnx);
return true;
@ -310,6 +316,11 @@ namespace cryptonote
MLOG_P2P_MESSAGE("Received NOTIFY_NEW_BLOCK (hop " << arg.hop << ", " << arg.b.txs.size() << " txes)");
if(context.m_state != cryptonote_connection_context::state_normal)
return 1;
if(!is_synchronized()) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
{
LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
return 1;
}
m_core.pause_mine();
std::list<block_complete_entry> blocks;
blocks.push_back(arg.b);
@ -324,6 +335,7 @@ namespace cryptonote
m_p2p->drop_connection(context);
m_core.cleanup_handle_incoming_blocks();
m_core.resume_mine();
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
}
@ -336,6 +348,7 @@ namespace cryptonote
{
LOG_PRINT_CCONTEXT_L0("Block verification failed, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
if(bvc.m_added_to_main_chain)
@ -361,6 +374,11 @@ namespace cryptonote
MLOG_P2P_MESSAGE("Received NOTIFY_NEW_FLUFFY_BLOCK (height " << arg.current_blockchain_height << ", hop " << arg.hop << ", " << arg.b.txs.size() << " txes)");
if(context.m_state != cryptonote_connection_context::state_normal)
return 1;
if(!is_synchronized()) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
{
LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
return 1;
}
m_core.pause_mine();
@ -384,6 +402,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -417,6 +436,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -431,6 +451,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -455,6 +476,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -472,6 +494,7 @@ namespace cryptonote
{
LOG_PRINT_CCONTEXT_L1("Block verification failed: transaction verification failed, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -494,6 +517,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -514,6 +538,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
m_core.resume_mine();
return 1;
}
@ -591,6 +616,7 @@ namespace cryptonote
{
LOG_PRINT_CCONTEXT_L0("Block verification failed, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
if( bvc.m_added_to_main_chain )
@ -624,6 +650,7 @@ namespace cryptonote
m_core.resume_mine();
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
@ -644,6 +671,7 @@ namespace cryptonote
{
LOG_ERROR_CCONTEXT("failed to find block: " << arg.block_hash << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
@ -671,6 +699,7 @@ namespace cryptonote
);
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
}
@ -682,6 +711,7 @@ namespace cryptonote
LOG_ERROR_CCONTEXT("Failed to handle request NOTIFY_REQUEST_FLUFFY_MISSING_TX, "
<< "failed to get requested transactions");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
if (!missed.empty() || txs.size() != txids.size())
@ -689,6 +719,7 @@ namespace cryptonote
LOG_ERROR_CCONTEXT("Failed to handle request NOTIFY_REQUEST_FLUFFY_MISSING_TX, "
<< missed.size() << " requested transactions not found" << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
@ -732,6 +763,7 @@ namespace cryptonote
{
LOG_PRINT_CCONTEXT_L1("Tx verification failed, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
if(tvc.m_should_be_relayed)
@ -758,6 +790,8 @@ namespace cryptonote
{
LOG_ERROR_CCONTEXT("failed to handle request NOTIFY_REQUEST_GET_OBJECTS, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
LOG_PRINT_CCONTEXT_L2("-->>NOTIFY_RESPONSE_GET_OBJECTS: blocks.size()=" << rsp.blocks.size() << ", txs.size()=" << rsp.txs.size()
<< ", rsp.m_current_blockchain_height=" << rsp.current_blockchain_height << ", missed_ids.size()=" << rsp.missed_ids.size());
@ -770,21 +804,15 @@ namespace cryptonote
template<class t_core>
double t_cryptonote_protocol_handler<t_core>::get_avg_block_size() {
// return m_core.get_blockchain_storage().get_avg_block_size(count); // this does not count too well the actuall network-size of data we need to download
double t_cryptonote_protocol_handler<t_core>::get_avg_block_size()
{
CRITICAL_REGION_LOCAL(m_buffer_mutex);
double avg = 0;
if (m_avg_buffer.size() == 0) {
_warn("m_avg_buffer.size() == 0");
if (m_avg_buffer.empty()) {
MWARNING("m_avg_buffer.size() == 0");
return 500;
}
const bool dbg_poke_lock = 0; // debug: try to trigger an error by poking around with locks. TODO: configure option
long int dbg_repeat=0;
do {
for (auto element : m_avg_buffer) avg += element;
} while(dbg_poke_lock && (dbg_repeat++)<100000); // in debug/poke mode, repeat this calculation to trigger hidden locking error if there is one
double avg = 0;
for (const auto &element : m_avg_buffer) avg += element;
return avg / m_avg_buffer.size();
}
@ -794,31 +822,30 @@ namespace cryptonote
{
MLOG_P2P_MESSAGE("Received NOTIFY_RESPONSE_GET_OBJECTS (" << arg.blocks.size() << " blocks, " << arg.txs.size() << " txes)");
// calculate size of request - mainly for logging/debug
bool force_next_span = false;
// calculate size of request
size_t size = 0;
for (auto element : arg.txs) size += element.size();
for (const auto &element : arg.txs) size += element.size();
for (auto element : arg.blocks) {
size += element.block.size();
for (auto tx : element.txs)
size += tx.size();
size_t blocks_size = 0;
for (const auto &element : arg.blocks) {
blocks_size += element.block.size();
for (const auto &tx : element.txs)
blocks_size += tx.size();
}
size += blocks_size;
for (auto element : arg.missed_ids)
for (const auto &element : arg.missed_ids)
size += sizeof(element.data);
size += sizeof(arg.current_blockchain_height);
{
CRITICAL_REGION_LOCAL(m_buffer_mutex);
m_avg_buffer.push_back(size);
const bool dbg_poke_lock = 0; // debug: try to trigger an error by poking around with locks. TODO: configure option
long int dbg_repeat=0;
do {
m_avg_buffer.push_back(666); // a test value
m_avg_buffer.erase_end(1);
} while(dbg_poke_lock && (dbg_repeat++)<100000); // in debug/poke mode, repeat this calculation to trigger hidden locking error if there is one
}
MDEBUG(context << " downloaded " << size << " bytes worth of blocks");
/*using namespace boost::chrono;
auto point = steady_clock::now();
auto time_from_epoh = point.time_since_epoch();
@ -831,14 +858,17 @@ namespace cryptonote
LOG_ERROR_CCONTEXT("sent wrong NOTIFY_HAVE_OBJECTS: arg.m_current_blockchain_height=" << arg.current_blockchain_height
<< " < m_last_response_height=" << context.m_last_response_height << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
context.m_remote_blockchain_height = arg.current_blockchain_height;
size_t count = 0;
std::vector<crypto::hash> block_hashes;
block_hashes.reserve(arg.blocks.size());
const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
uint64_t start_height = std::numeric_limits<uint64_t>::max();
cryptonote::block b;
for(const block_complete_entry& block_entry: arg.blocks)
{
if (m_stopping)
@ -846,35 +876,33 @@ namespace cryptonote
return 1;
}
++count;
block b;
if(!parse_and_validate_block_from_blob(block_entry.block, b))
{
LOG_ERROR_CCONTEXT("sent wrong block: failed to parse and validate block: "
<< epee::string_tools::buff_to_hex_nodelimer(block_entry.block) << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
//to avoid concurrency in core between connections, suspend connections which delivered block later then first one
const crypto::hash block_hash = get_block_hash(b);
if(count == 2)
if (b.miner_tx.vin.size() != 1 || b.miner_tx.vin.front().type() != typeid(txin_gen))
{
if(m_core.have_block(block_hash))
{
context.m_state = cryptonote_connection_context::state_idle;
context.m_needed_objects.clear();
context.m_requested_objects.clear();
LOG_PRINT_CCONTEXT_L1("Connection set to idle state.");
return 1;
}
LOG_ERROR_CCONTEXT("sent wrong block: block: miner tx does not have exactly one txin_gen input"
<< epee::string_tools::buff_to_hex_nodelimer(block_entry.block) << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
if (start_height == std::numeric_limits<uint64_t>::max())
start_height = boost::get<txin_gen>(b.miner_tx.vin[0]).height;
const crypto::hash block_hash = get_block_hash(b);
auto req_it = context.m_requested_objects.find(block_hash);
if(req_it == context.m_requested_objects.end())
{
LOG_ERROR_CCONTEXT("sent wrong NOTIFY_RESPONSE_GET_OBJECTS: block with id=" << epee::string_tools::pod_to_hex(get_blob_hash(block_entry.block))
<< " wasn't requested, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
if(b.tx_hashes.size() != block_entry.txs.size())
@ -882,6 +910,7 @@ namespace cryptonote
LOG_ERROR_CCONTEXT("sent wrong NOTIFY_RESPONSE_GET_OBJECTS: block with id=" << epee::string_tools::pod_to_hex(get_blob_hash(block_entry.block))
<< ", tx_hashes.size()=" << b.tx_hashes.size() << " mismatch with block_complete_entry.m_txs.size()=" << block_entry.txs.size() << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
@ -894,114 +923,189 @@ namespace cryptonote
MERROR("returned not all requested objects (context.m_requested_objects.size()="
<< context.m_requested_objects.size() << "), dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
// get the last parsed block, which should be the highest one
if(m_core.have_block(cryptonote::get_block_hash(b)))
{
const uint64_t subchain_height = start_height + arg.blocks.size();
LOG_DEBUG_CC(context, "These are old blocks, ignoring: blocks " << start_height << " - " << (subchain_height-1) << ", blockchain height " << m_core.get_current_blockchain_height());
goto skip;
}
{
MLOG_YELLOW(el::Level::Debug, "Got NEW BLOCKS inside of " << __FUNCTION__ << ": size: " << arg.blocks.size());
// add that new span to the block queue
const boost::posix_time::time_duration dt = now - context.m_last_request_time;
const float rate = size * 1e6 / (dt.total_microseconds() + 1);
MDEBUG(context << " adding span: " << arg.blocks.size() << " at height " << start_height << ", " << dt.total_microseconds()/1e6 << " seconds, " << (rate/1e3) << " kB/s, size now " << (m_block_queue.get_data_size() + blocks_size) / 1048576.f << " MB");
m_block_queue.add_blocks(start_height, arg.blocks, context.m_connection_id, rate, blocks_size);
if (m_core.get_test_drop_download() && m_core.get_test_drop_download_height()) { // DISCARD BLOCKS for testing
// we lock all the rest to avoid having multiple connections redo a lot
// of the same work, and one of them doing it for nothing: subsequent
// connections will wait until the current one's added its blocks, then
// will add any extra it has, if any
CRITICAL_REGION_LOCAL(m_sync_lock);
// We try to lock the sync lock. If we can, it means no other thread is
// currently adding blocks, so we do that for as long as we can from the
// block queue. Then, we go back to download.
const boost::unique_lock<boost::mutex> sync{m_sync_lock, boost::try_to_lock};
if (!sync.owns_lock())
{
MINFO("Failed to lock m_sync_lock, going back to download");
goto skip;
}
MDEBUG(context << " lock m_sync_lock, adding blocks to chain...");
m_core.pause_mine();
epee::misc_utils::auto_scope_leave_caller scope_exit_handler = epee::misc_utils::create_scope_leave_handler(
boost::bind(&t_core::resume_mine, &m_core));
const uint64_t previous_height = m_core.get_current_blockchain_height();
// dismiss what another connection might already have done (likely everything)
uint64_t top_height;
crypto::hash top_hash;
if (m_core.get_blockchain_top(top_height, top_hash)) {
uint64_t dismiss = 1;
for (const auto &h: block_hashes) {
if (top_hash == h) {
LOG_DEBUG_CC(context, "Found current top block in synced blocks, dismissing "
<< dismiss << "/" << arg.blocks.size() << " blocks");
while (dismiss--)
arg.blocks.pop_front();
break;
}
++dismiss;
}
}
if (arg.blocks.empty())
goto skip;
m_core.prepare_handle_incoming_blocks(arg.blocks);
for(const block_complete_entry& block_entry: arg.blocks)
while (1)
{
if (m_stopping)
const uint64_t previous_height = m_core.get_current_blockchain_height();
uint64_t start_height;
std::list<cryptonote::block_complete_entry> blocks;
boost::uuids::uuid span_connection_id;
m_block_queue.mark_last_block(previous_height - 1);
if (!m_block_queue.get_next_span(start_height, blocks, span_connection_id))
{
m_core.cleanup_handle_incoming_blocks();
return 1;
MDEBUG(context << " no next span found, going back to download");
break;
}
MDEBUG(context << " next span in the queue has blocks " << start_height << "-" << (start_height + blocks.size() - 1)
<< ", we need " << previous_height);
if (previous_height < start_height || previous_height >= start_height + blocks.size())
{
MDEBUG(context << " this span is not what we need, going back to download");
break;
}
// process transactions
TIME_MEASURE_START(transactions_process_time);
for(auto& tx_blob: block_entry.txs)
const boost::posix_time::ptime start = boost::posix_time::microsec_clock::universal_time();
m_core.prepare_handle_incoming_blocks(blocks);
uint64_t block_process_time_full = 0, transactions_process_time_full = 0;
size_t num_txs = 0;
for(const block_complete_entry& block_entry: blocks)
{
tx_verification_context tvc = AUTO_VAL_INIT(tvc);
m_core.handle_incoming_tx(tx_blob, tvc, true, true, false);
if(tvc.m_verifivation_failed)
if (m_stopping)
{
LOG_ERROR_CCONTEXT("transaction verification failed on NOTIFY_RESPONSE_GET_OBJECTS, tx_id = "
<< epee::string_tools::pod_to_hex(get_blob_hash(tx_blob)) << ", dropping connection");
m_p2p->drop_connection(context);
m_core.cleanup_handle_incoming_blocks();
return 1;
}
// process transactions
TIME_MEASURE_START(transactions_process_time);
num_txs += block_entry.txs.size();
std::vector<tx_verification_context> tvc;
m_core.handle_incoming_txs(block_entry.txs, tvc, true, true, false);
std::list<blobdata>::const_iterator it = block_entry.txs.begin();
for (size_t i = 0; i < tvc.size(); ++i, ++it)
{
if(tvc[i].m_verifivation_failed)
{
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
LOG_ERROR_CCONTEXT("transaction verification failed on NOTIFY_RESPONSE_GET_OBJECTS, tx_id = "
<< epee::string_tools::pod_to_hex(get_blob_hash(*it)) << ", dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id, true);
return true;
}))
LOG_ERROR_CCONTEXT("span connection id not found");
m_core.cleanup_handle_incoming_blocks();
// in case the peer had dropped beforehand, remove the span anyway so other threads can wake up and get it
m_block_queue.remove_spans(span_connection_id, start_height);
return 1;
}
}
TIME_MEASURE_FINISH(transactions_process_time);
transactions_process_time_full += transactions_process_time;
// process block
TIME_MEASURE_START(block_process_time);
block_verification_context bvc = boost::value_initialized<block_verification_context>();
m_core.handle_incoming_block(block_entry.block, bvc, false); // <--- process block
if(bvc.m_verifivation_failed)
{
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
LOG_PRINT_CCONTEXT_L1("Block verification failed, dropping connection");
m_p2p->drop_connection(context);
m_p2p->add_host_fail(context.m_remote_address);
m_block_queue.flush_spans(context.m_connection_id, true);
return true;
}))
LOG_ERROR_CCONTEXT("span connection id not found");
m_core.cleanup_handle_incoming_blocks();
// in case the peer had dropped beforehand, remove the span anyway so other threads can wake up and get it
m_block_queue.remove_spans(span_connection_id, start_height);
return 1;
}
}
TIME_MEASURE_FINISH(transactions_process_time);
if(bvc.m_marked_as_orphaned)
{
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
LOG_PRINT_CCONTEXT_L1("Block received at sync phase was marked as orphaned, dropping connection");
m_p2p->drop_connection(context);
m_p2p->add_host_fail(context.m_remote_address);
m_block_queue.flush_spans(context.m_connection_id, true);
return true;
}))
LOG_ERROR_CCONTEXT("span connection id not found");
// process block
m_core.cleanup_handle_incoming_blocks();
// in case the peer had dropped beforehand, remove the span anyway so other threads can wake up and get it
m_block_queue.remove_spans(span_connection_id, start_height);
return 1;
}
TIME_MEASURE_START(block_process_time);
block_verification_context bvc = boost::value_initialized<block_verification_context>();
TIME_MEASURE_FINISH(block_process_time);
block_process_time_full += block_process_time;
m_core.handle_incoming_block(block_entry.block, bvc, false); // <--- process block
} // each download block
if(bvc.m_verifivation_failed)
MCINFO("sync-info", "Block process time (" << blocks.size() << " blocks, " << num_txs << " txs): " << block_process_time_full + transactions_process_time_full << " (" << transactions_process_time_full << "/" << block_process_time_full << ") ms");
m_core.cleanup_handle_incoming_blocks();
m_block_queue.mark_last_block(m_core.get_current_blockchain_height() - 1);
if (m_core.get_current_blockchain_height() > previous_height)
{
LOG_PRINT_CCONTEXT_L1("Block verification failed, dropping connection");
m_p2p->drop_connection(context);
m_p2p->add_host_fail(context.m_remote_address);
m_core.cleanup_handle_incoming_blocks();
return 1;
const boost::posix_time::time_duration dt = boost::posix_time::microsec_clock::universal_time() - start;
std::string timing_message = "";
if (ELPP->vRegistry()->allowed(el::Level::Info, "sync-info"))
timing_message = std::string(" (") + std::to_string(dt.total_microseconds()/1e6) + " sec, "
+ std::to_string((m_core.get_current_blockchain_height() - previous_height) * 1e6 / dt.total_microseconds())
+ " blocks/sec), " + std::to_string(m_block_queue.get_data_size() / 1048576.f) + " MB queued";
if (ELPP->vRegistry()->allowed(el::Level::Debug, "sync-info"))
timing_message += std::string(": ") + m_block_queue.get_overview();
MGINFO_YELLOW(context << " Synced " << m_core.get_current_blockchain_height() << "/" << m_core.get_target_blockchain_height()
<< timing_message);
}
if(bvc.m_marked_as_orphaned)
{
LOG_PRINT_CCONTEXT_L1("Block received at sync phase was marked as orphaned, dropping connection");
m_p2p->drop_connection(context);
m_p2p->add_host_fail(context.m_remote_address);
m_core.cleanup_handle_incoming_blocks();
return 1;
}
TIME_MEASURE_FINISH(block_process_time);
LOG_PRINT_CCONTEXT_L2("Block process time: " << block_process_time + transactions_process_time << "(" << transactions_process_time << "/" << block_process_time << ")ms");
} // each download block
m_core.cleanup_handle_incoming_blocks();
if (m_core.get_current_blockchain_height() > previous_height)
{
MGINFO_YELLOW(context << " Synced " << m_core.get_current_blockchain_height() << "/" << m_core.get_target_blockchain_height());
}
} // if not DISCARD BLOCK
if (should_download_next_span(context))
{
context.m_needed_objects.clear();
context.m_last_response_height = 0;
force_next_span = true;
}
}
skip:
request_missing_objects(context, true);
if (!request_missing_objects(context, true, force_next_span))
{
LOG_ERROR_CCONTEXT("Failed to request missing objects, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
return 1;
}
//------------------------------------------------------------------------------------------------------------------------
@ -1020,6 +1124,7 @@ skip:
{
LOG_ERROR_CCONTEXT("Failed to handle NOTIFY_REQUEST_CHAIN.");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
LOG_PRINT_CCONTEXT_L2("-->>NOTIFY_RESPONSE_CHAIN_ENTRY: m_start_height=" << r.start_height << ", m_total_height=" << r.total_height << ", m_block_ids.size()=" << r.m_block_ids.size());
@ -1028,54 +1133,260 @@ skip:
}
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
bool t_cryptonote_protocol_handler<t_core>::request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks)
bool t_cryptonote_protocol_handler<t_core>::should_download_next_span(cryptonote_connection_context& context) const
{
//if (!m_one_request == false)
//return true;
m_one_request = false;
// save request size to log (dr monero)
/*using namespace boost::chrono;
auto point = steady_clock::now();
auto time_from_epoh = point.time_since_epoch();
auto sec = duration_cast< seconds >( time_from_epoh ).count();*/
std::list<crypto::hash> hashes;
boost::uuids::uuid span_connection_id;
boost::posix_time::ptime request_time;
std::pair<uint64_t, uint64_t> span = m_block_queue.get_next_span_if_scheduled(hashes, span_connection_id, request_time);
if(context.m_needed_objects.size())
// if the next span is not scheduled (or there is none)
if (span.second == 0)
{
// we might be in a weird case where there is a filled next span,
// but it starts higher than the current height
uint64_t height;
std::list<cryptonote::block_complete_entry> bcel;
if (!m_block_queue.get_next_span(height, bcel, span_connection_id, true))
return false;
if (height > m_core.get_current_blockchain_height())
{
MDEBUG(context << " we should download it as the next block isn't scheduled");
return true;
}
return false;
}
// if it was scheduled by this particular peer
if (span_connection_id == context.m_connection_id)
return false;
float span_speed = m_block_queue.get_speed(span_connection_id);
float speed = m_block_queue.get_speed(context.m_connection_id);
MDEBUG(context << " next span is scheduled for " << span_connection_id << ", speed " << span_speed << ", ours " << speed);
// we try for that span too if:
// - we're substantially faster, or:
// - we're the fastest and the other one isn't (avoids a peer being waaaay slow but yet unmeasured)
// - the other one asked at least 5 seconds ago
if (span_speed < .25 && speed > .75f)
{
MDEBUG(context << " we should download it as we're substantially faster");
return true;
}
if (speed == 1.0f && span_speed != 1.0f)
{
MDEBUG(context << " we should download it as we're the fastest peer");
return true;
}
const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
if ((now - request_time).total_microseconds() > REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD)
{
MDEBUG(context << " we should download it as this span was requested long ago");
return true;
}
return false;
}
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
bool t_cryptonote_protocol_handler<t_core>::request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks, bool force_next_span)
{
m_block_queue.mark_last_block(m_core.get_current_blockchain_height() - 1);
// flush stale spans
std::set<boost::uuids::uuid> live_connections;
m_p2p->for_each_connection([&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t support_flags)->bool{
live_connections.insert(context.m_connection_id);
return true;
});
m_block_queue.flush_stale_spans(live_connections);
// if we don't need to get next span, and the block queue is full enough, wait a bit
bool start_from_current_chain = false;
if (!force_next_span)
{
bool first = true;
while (1)
{
size_t nblocks = m_block_queue.get_num_filled_spans();
size_t size = m_block_queue.get_data_size();
if (nblocks < BLOCK_QUEUE_NBLOCKS_THRESHOLD || size < BLOCK_QUEUE_SIZE_THRESHOLD)
{
if (!first)
{
LOG_DEBUG_CC(context, "Block queue is " << nblocks << " and " << size << ", resuming");
}
break;
}
if (should_download_next_span(context))
{
MDEBUG(context << " we should try for that next span too, we think we could get it faster, resuming");
force_next_span = true;
break;
}
if (first)
{
LOG_DEBUG_CC(context, "Block queue is " << nblocks << " and " << size << ", pausing");
first = false;
}
for (size_t n = 0; n < 50; ++n)
{
if (m_stopping)
return 1;
boost::this_thread::sleep_for(boost::chrono::milliseconds(100));
}
}
}
MDEBUG(context << " request_missing_objects: check " << check_having_blocks << ", force_next_span " << force_next_span << ", m_needed_objects " << context.m_needed_objects.size() << " lrh " << context.m_last_response_height << ", chain " << m_core.get_current_blockchain_height());
if(context.m_needed_objects.size() || force_next_span)
{
//we know objects that we need, request this objects
NOTIFY_REQUEST_GET_OBJECTS::request req;
bool is_next = false;
size_t count = 0;
auto it = context.m_needed_objects.begin();
const size_t count_limit = m_core.get_block_sync_size();
MDEBUG("Setting count_limit: " << count_limit);
while(it != context.m_needed_objects.end() && count < count_limit)
std::pair<uint64_t, uint64_t> span = std::make_pair(0, 0);
if (force_next_span)
{
if( !(check_having_blocks && m_core.have_block(*it)))
MDEBUG(context << " force_next_span is true, trying next span");
span = m_block_queue.get_start_gap_span();
if (span.second > 0)
{
req.blocks.push_back(*it);
++count;
context.m_requested_objects.insert(*it);
const uint64_t first_block_height_known = context.m_last_response_height - context.m_needed_objects.size() + 1;
const uint64_t last_block_height_known = context.m_last_response_height;
const uint64_t first_block_height_needed = span.first;
const uint64_t last_block_height_needed = span.first + std::min(span.second, (uint64_t)count_limit) - 1;
MDEBUG(context << " gap found, span: " << span.first << " - " << span.first + span.second - 1 << " (" << last_block_height_needed << ")");
MDEBUG(context << " current known hashes from from " << first_block_height_known << " to " << last_block_height_known);
if (first_block_height_needed < first_block_height_known || last_block_height_needed > last_block_height_known)
{
MDEBUG(context << " we are missing some of the necessary hashes for this gap, requesting chain again");
context.m_needed_objects.clear();
context.m_last_response_height = 0;
start_from_current_chain = true;
goto skip;
}
MDEBUG(context << " we have the hashes for this gap");
}
if (span.second == 0)
{
std::list<crypto::hash> hashes;
boost::uuids::uuid span_connection_id;
boost::posix_time::ptime time;
span = m_block_queue.get_next_span_if_scheduled(hashes, span_connection_id, time);
if (span.second > 0)
{
is_next = true;
for (const auto &hash: hashes)
{
req.blocks.push_back(hash);
context.m_requested_objects.insert(hash);
}
}
}
context.m_needed_objects.erase(it++);
}
LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size() << ", txs.size()=" << req.txs.size()
<< "requested blocks count=" << count << " / " << count_limit);
//epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
if (span.second == 0)
{
MDEBUG(context << " span size is 0");
if (context.m_last_response_height + 1 < context.m_needed_objects.size())
{
MERROR(context << " ERROR: inconsistent context: lrh " << context.m_last_response_height << ", nos " << context.m_needed_objects.size());
context.m_needed_objects.clear();
context.m_last_response_height = 0;
goto skip;
}
const uint64_t first_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
span = m_block_queue.reserve_span(first_block_height, context.m_last_response_height, count_limit, context.m_connection_id);
MDEBUG(context << " span from " << first_block_height << ": " << span.first << "/" << span.second);
}
if (span.second == 0 && !force_next_span)
{
MDEBUG(context << " still no span reserved, we may be in the corner case of next span scheduled and everything else scheduled/filled");
std::list<crypto::hash> hashes;
boost::uuids::uuid span_connection_id;
boost::posix_time::ptime time;
span = m_block_queue.get_next_span_if_scheduled(hashes, span_connection_id, time);
if (span.second > 0)
{
is_next = true;
for (const auto &hash: hashes)
{
req.blocks.push_back(hash);
context.m_requested_objects.insert(hash);
}
}
}
MDEBUG(context << " span: " << span.first << "/" << span.second << " (" << span.first << " - " << (span.first + span.second - 1) << ")");
if (span.second > 0)
{
if (!is_next)
{
const uint64_t first_context_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
uint64_t skip = span.first - first_context_block_height;
if (skip > context.m_needed_objects.size())
{
MERROR("ERROR: skip " << skip << ", m_needed_objects " << context.m_needed_objects.size() << ", first_context_block_height" << first_context_block_height);
return false;
}
while (skip--)
context.m_needed_objects.pop_front();
if (context.m_needed_objects.size() < span.second)
{
MERROR("ERROR: span " << span.first << "/" << span.second << ", m_needed_objects " << context.m_needed_objects.size());
return false;
}
post_notify<NOTIFY_REQUEST_GET_OBJECTS>(req, context);
}else if(context.m_last_response_height < context.m_remote_blockchain_height-1)
std::list<crypto::hash> hashes;
auto it = context.m_needed_objects.begin();
for (size_t n = 0; n < span.second; ++n)
{
req.blocks.push_back(*it);
++count;
context.m_requested_objects.insert(*it);
hashes.push_back(*it);
auto j = it++;
context.m_needed_objects.erase(j);
}
m_block_queue.set_span_hashes(span.first, context.m_connection_id, hashes);
}
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size() << ", txs.size()=" << req.txs.size()
<< "requested blocks count=" << count << " / " << count_limit << " from " << span.first);
//epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
post_notify<NOTIFY_REQUEST_GET_OBJECTS>(req, context);
return true;
}
}
skip:
context.m_needed_objects.clear();
if(context.m_last_response_height < context.m_remote_blockchain_height-1)
{//we have to fetch more objects ids, request blockchain entry
NOTIFY_REQUEST_CHAIN::request r = boost::value_initialized<NOTIFY_REQUEST_CHAIN::request>();
m_core.get_short_chain_history(r.block_ids);
if (!start_from_current_chain)
{
// we'll want to start off from where we are on the download side, which may not be added yet
crypto::hash last_known_hash = m_block_queue.get_last_known_hash();
if (last_known_hash != cryptonote::null_hash && r.block_ids.front() != last_known_hash)
r.block_ids.push_front(last_known_hash);
}
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
//std::string blob; // for calculate size of request
//epee::serialization::store_t_to_binary(r, blob);
//epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
LOG_PRINT_CCONTEXT_L1("r = " << 200);
//LOG_PRINT_CCONTEXT_L1("r = " << 200);
LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() << ", start_from_current_chain " << start_from_current_chain);
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
}else
{
@ -1089,8 +1400,18 @@ skip:
<< "\r\non connection [" << epee::net_utils::print_connection_context_short(context)<< "]");
context.m_state = cryptonote_connection_context::state_normal;
MGINFO_GREEN("SYNCHRONIZED OK");
on_connection_synchronized();
if (context.m_remote_blockchain_height >= m_core.get_target_blockchain_height())
{
if (m_core.get_current_blockchain_height() >= m_core.get_target_blockchain_height())
{
MGINFO_GREEN("SYNCHRONIZED OK");
on_connection_synchronized();
}
}
else
{
MINFO(context << " we've reached this peer's blockchain height");
}
}
return true;
}
@ -1134,15 +1455,7 @@ skip:
LOG_ERROR_CCONTEXT("sent empty m_block_ids, dropping connection");
m_p2p->drop_connection(context);
m_p2p->add_host_fail(context.m_remote_address);
return 1;
}
if(!m_core.have_block(arg.m_block_ids.front()))
{
LOG_ERROR_CCONTEXT("sent m_block_ids starting from unknown id: "
<< epee::string_tools::pod_to_hex(arg.m_block_ids.front()) << " , dropping connection");
m_p2p->drop_connection(context);
m_p2p->add_host_fail(context.m_remote_address);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
@ -1154,15 +1467,22 @@ skip:
<< ", m_start_height=" << arg.start_height
<< ", m_block_ids.size()=" << arg.m_block_ids.size());
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
for(auto& bl_id: arg.m_block_ids)
{
if(!m_core.have_block(bl_id))
context.m_needed_objects.push_back(bl_id);
context.m_needed_objects.push_back(bl_id);
}
request_missing_objects(context, false);
if (!request_missing_objects(context, false))
{
LOG_ERROR_CCONTEXT("Failed to request missing objects, dropping connection");
m_p2p->drop_connection(context);
m_block_queue.flush_spans(context.m_connection_id);
return 1;
}
return 1;
}
//------------------------------------------------------------------------------------------------------------------------