diff options
Diffstat (limited to '')
30 files changed, 714 insertions, 383 deletions
diff --git a/.gitignore b/.gitignore index 3aed0e11b..755cb1043 100644 --- a/.gitignore +++ b/.gitignore @@ -20,7 +20,7 @@ cscope.out cscope.in.out cscope.po.out - +external/miniupnpc/Makefile miniupnpcstrings.h version/ # Created by https://www.gitignore.io diff --git a/CMakeLists.txt b/CMakeLists.txt index bdb967467..84a48c116 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -50,24 +50,17 @@ list(INSERT CMAKE_MODULE_PATH 0 if (NOT DEFINED ENV{DEVELOPER_LOCAL_TOOLS}) message(STATUS "Could not find DEVELOPER_LOCAL_TOOLS in env (not required)") - set(BOOST_IGNORE_SYSTEM_PATHS OFF) -elseif ("$ENV{DEVELOPER_LOCAL_TOOLS}" STREQUAL "1") + set(BOOST_IGNORE_SYSTEM_PATHS_DEFAULT OFF) +elseif ("$ENV{DEVELOPER_LOCAL_TOOLS}" EQUAL 1) message(STATUS "Found: env DEVELOPER_LOCAL_TOOLS = 1") - set(BOOST_IGNORE_SYSTEM_PATHS ON) + set(BOOST_IGNORE_SYSTEM_PATHS_DEFAULT ON) else() message(STATUS "Found: env DEVELOPER_LOCAL_TOOLS = 0") - set(BOOST_IGNORE_SYSTEM_PATHS OFF) + set(BOOST_IGNORE_SYSTEM_PATHS_DEFAULT OFF) endif() -#message(STATUS "BOOST_IGNORE_SYSTEM_PATHS defaults to ${BOOST_IGNORE_SYSTEM_PATHS_DEFAULT}") -#option(BOOST_IGNORE_SYSTEM_PATHS "Ignore boost system paths for local boost installation" ${BOOST_IGNORE_SYSTEM_PATHS_DEFAULT}) -message(STATUS "BOOST_IGNORE_SYSTEM_PATHS: ${BOOST_IGNORE_SYSTEM_PATHS}") - -# Options (for external/otshell_utils/) -option(WITH_TERMCOLORS "Build with support for unix terminal console colors VT100" ON) -if (WITH_TERMCOLORS) - add_definitions( -DCFG_WITH_TERMCOLORS ) -endif () +message(STATUS "BOOST_IGNORE_SYSTEM_PATHS defaults to ${BOOST_IGNORE_SYSTEM_PATHS_DEFAULT}") +option(BOOST_IGNORE_SYSTEM_PATHS "Ignore boost system paths for local boost installation" ${BOOST_IGNORE_SYSTEM_PATHS_DEFAULT}) set_property(GLOBAL PROPERTY USE_FOLDERS ON) enable_testing() @@ -162,9 +155,9 @@ else() else() set(ARCH_FLAG "-march=${ARCH}") endif() - set(WARNINGS "-Wall -Wextra -Wpointer-arith -Wundef -Wvla -Wwrite-strings -Wno-error=extra -Wno-error=deprecated-declarations -Wno-error=sign-compare -Wno-error=strict-aliasing -Wno-error=type-limits -Wno-unused-parameter -Wno-error=unused-variable -Wno-error=undef -Wno-error=uninitialized") + set(WARNINGS "-Wall -pedantic -Wextra -Wpointer-arith -Wundef -Wvla -Wwrite-strings -Wno-error=extra -Wno-error=deprecated-declarations -Wno-error=sign-compare -Wno-error=strict-aliasing -Wno-error=type-limits -Wno-unused-parameter -Wno-error=unused-variable -Wno-error=undef -Wno-error=uninitialized") if(NOT MINGW) - set(WARNINGS "${WARNINGS} -Werror") + # set(WARNINGS "${WARNINGS} -Werror") # to allow pedantic but not stop compilation endif() if(CMAKE_C_COMPILER_ID STREQUAL "Clang") set(WARNINGS "${WARNINGS} -Wno-error=mismatched-tags -Wno-error=null-conversion -Wno-overloaded-shift-op-parentheses -Wno-error=shift-count-overflow -Wno-error=tautological-constant-out-of-range-compare -Wno-error=unused-private-field -Wno-error=unneeded-internal-declaration") diff --git a/contrib/epee/include/net/abstract_tcp_server2.h b/contrib/epee/include/net/abstract_tcp_server2.h index 1e6223212..3e6ea2171 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.h +++ b/contrib/epee/include/net/abstract_tcp_server2.h @@ -56,6 +56,7 @@ #include "syncobj.h" #include "../../../../src/p2p/connection_basic.hpp" #include "../../../../contrib/otshell_utils/utils.hpp" +#include "../../../../src/p2p/network_throttle-detail.hpp" #define ABSTRACT_SERVER_SEND_QUE_MAX_COUNT 1000 @@ -130,6 +131,7 @@ namespace net_utils /// Buffer for incoming data. boost::array<char, 8192> buffer_; + //boost::array<char, 1024> buffer_; t_connection_context context; i_connection_filter* &m_pfilter; @@ -143,6 +145,12 @@ namespace net_utils critical_section m_chunking_lock; // held while we add small chunks of the big do_send() to small do_send_chunk() t_server_role m_connection_type; + + // for calculate speed (last 60 sec) + network_throttle m_throttle_speed_in; + network_throttle m_throttle_speed_out; + std::mutex m_throttle_speed_in_mutex; + std::mutex m_throttle_speed_out_mutex; public: void setRPcStation(); @@ -285,6 +293,7 @@ namespace net_utils critical_section m_threads_lock; volatile uint32_t m_thread_index; // TODO change to std::atomic t_server_role type; + void detach_threads(); /// The next connection to be accepted connection_ptr new_connection_; diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl index 8dff192b1..5855f7f86 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.inl +++ b/contrib/epee/include/net/abstract_tcp_server2.inl @@ -52,6 +52,7 @@ #include "../../../../src/cryptonote_core/cryptonote_core.h" // e.g. for the send_stop_signal() #include "../../../../contrib/otshell_utils/utils.hpp" +#include "../../../../src/p2p/data_logger.hpp" using namespace nOT::nUtils; // TODO PRAGMA_WARNING_PUSH @@ -75,7 +76,9 @@ PRAGMA_WARNING_DISABLE_VS(4355) connection_basic(io_service, ref_sock_count, sock_number), m_protocol_handler(this, config, context), m_pfilter( pfilter ), - m_connection_type(NET) + m_connection_type(NET), + m_throttle_speed_in("speed_in", "throttle_speed_in"), + m_throttle_speed_out("speed_out", "throttle_speed_out") { _info_c("net/sleepRPC", "connection constructor set m_connection_type="<<m_connection_type); } @@ -162,6 +165,9 @@ PRAGMA_WARNING_DISABLE_VS(4355) socket_.set_option( optionTos ); //_dbg1("Set ToS flag to " << tos); + boost::asio::ip::tcp::no_delay noDelayOption(false); + socket_.set_option(noDelayOption); + return true; CATCH_ENTRY_L0("connection<t_protocol_handler>::start()", false); @@ -240,6 +246,32 @@ PRAGMA_WARNING_DISABLE_VS(4355) if (!e) { + { + CRITICAL_REGION_LOCAL(m_throttle_speed_in_mutex); + m_throttle_speed_in.handle_trafic_exact(bytes_transferred); + context.m_current_speed_down = m_throttle_speed_in.get_current_speed(); + } + + { + CRITICAL_REGION_LOCAL( epee::net_utils::network_throttle_manager::network_throttle_manager::m_lock_get_global_throttle_in ); + epee::net_utils::network_throttle_manager::network_throttle_manager::get_global_throttle_in().handle_trafic_exact(bytes_transferred * 1024); + } + double delay=0; // will be calculated + do + { + { //_scope_dbg1("CRITICAL_REGION_LOCAL"); + CRITICAL_REGION_LOCAL( epee::net_utils::network_throttle_manager::m_lock_get_global_throttle_in ); + delay = epee::net_utils::network_throttle_manager::get_global_throttle_in().get_sleep_time_after_tick( bytes_transferred ); // decission from global + } + + delay *= 0.5; + if (delay > 0) { + long int ms = (long int)(delay * 100); + epee::net_utils::data_logger::get_instance().add_data("sleep_down", ms); + std::this_thread::sleep_for(std::chrono::milliseconds(ms)); + } + } while(delay > 0); + //_info("[sock " << socket_.native_handle() << "] RECV " << bytes_transferred); logger_handle_net_read(bytes_transferred); context.m_last_recv = time(NULL); @@ -398,6 +430,11 @@ PRAGMA_WARNING_DISABLE_VS(4355) return false; if(m_was_shutdown) return false; + { + CRITICAL_REGION_LOCAL(m_throttle_speed_out_mutex); + m_throttle_speed_out.handle_trafic_exact(cb); + context.m_current_speed_up = m_throttle_speed_out.get_current_speed(); + } //_info("[sock " << socket_.native_handle() << "] SEND " << cb); context.m_last_send = time(NULL); @@ -405,8 +442,7 @@ PRAGMA_WARNING_DISABLE_VS(4355) //some data should be wrote to stream //request complete - do_send_handler_start( ptr , cb ); // (((H))) - + sleep_before_packet(cb, 1, 1); epee::critical_region_t<decltype(m_send_que_lock)> send_guard(m_send_que_lock); // *** critical *** long int retry=0; const long int retry_limit = 5*4; @@ -440,8 +476,9 @@ PRAGMA_WARNING_DISABLE_VS(4355) { // active operation should be in progress, nothing to do, just wait last operation callback auto size_now = cb; _info_c("net/out/size", "do_send() NOW just queues: packet="<<size_now<<" B, is added to queue-size="<<m_send_que.size()); - do_send_handler_delayed( ptr , size_now ); // (((H))) - + //do_send_handler_delayed( ptr , size_now ); // (((H))) // empty function + + LOG_PRINT_L4("[sock " << socket_.native_handle() << "] Async send requested " << m_send_que.front().size()); } else { // no active operation @@ -463,11 +500,11 @@ PRAGMA_WARNING_DISABLE_VS(4355) //) ); //_dbg3("(chunk): " << size_now); - logger_handle_net_write(size_now); + //logger_handle_net_write(size_now); //_info("[sock " << socket_.native_handle() << "] Async send requested " << m_send_que.front().size()); } - do_send_handler_stop( ptr , cb ); + //do_send_handler_stop( ptr , cb ); // empty function return true; @@ -516,7 +553,8 @@ PRAGMA_WARNING_DISABLE_VS(4355) shutdown(); return; } - + logger_handle_net_write(cb); + sleep_before_packet(cb, 1, 1); bool do_shutdown = false; CRITICAL_REGION_BEGIN(m_send_que_lock); if(m_send_que.empty()) @@ -545,7 +583,6 @@ PRAGMA_WARNING_DISABLE_VS(4355) // ) ); //_dbg3("(normal)" << size_now); - logger_handle_net_write(size_now); } CRITICAL_REGION_END(); @@ -784,6 +821,10 @@ POP_WARNINGS template<class t_protocol_handler> void boosted_tcp_server<t_protocol_handler>::send_stop_signal() { + if (::cryptonote::core::get_fast_exit() == true) + { + detach_threads(); + } m_stop_signal_sent = true; TRY_ENTRY(); io_service_.stop(); @@ -988,6 +1029,15 @@ POP_WARNINGS return true; CATCH_ENTRY_L0("boosted_tcp_server<t_protocol_handler>::connect_async", false); } + //--------------------------------------------------------------------------------- + template<class t_protocol_handler> + void boosted_tcp_server<t_protocol_handler>::detach_threads() + { + for (auto thread : m_threads) + thread->detach(); + } + + } // namespace } // namespace PRAGMA_WARNING_POP diff --git a/contrib/epee/include/net/levin_protocol_handler_async.h b/contrib/epee/include/net/levin_protocol_handler_async.h index 5e5e803f5..92f75161a 100644 --- a/contrib/epee/include/net/levin_protocol_handler_async.h +++ b/contrib/epee/include/net/levin_protocol_handler_async.h @@ -81,7 +81,7 @@ public: async_protocol_handler_config():m_pcommands_handler(NULL), m_max_packet_size(LEVIN_DEFAULT_MAX_PACKET_SIZE) {} - void del_connections(size_t count); + void del_out_connections(size_t count); }; @@ -670,10 +670,30 @@ void async_protocol_handler_config<t_connection_context>::del_connection(async_p } //------------------------------------------------------------------------------------------ template<class t_connection_context> -void async_protocol_handler_config<t_connection_context>::del_connections(size_t count) // TODO +void async_protocol_handler_config<t_connection_context>::del_out_connections(size_t count) { + std::vector <boost::uuids::uuid> out_connections; CRITICAL_REGION_BEGIN(m_connects_lock); - m_connects.clear(); + for (auto& c: m_connects) + { + if (!c.second->m_connection_context.m_is_income) + out_connections.push_back(c.first); + } + + if (out_connections.size() == 0) + return; + + // close random out connections + unsigned seed = std::chrono::system_clock::now().time_since_epoch().count(); + shuffle(out_connections.begin(), out_connections.end(), std::default_random_engine(seed)); + while (count > 0 && out_connections.size() > 0) + { + close(*out_connections.begin()); + del_connection(m_connects.at(*out_connections.begin())); + out_connections.erase(out_connections.begin()); + --count; + } + CRITICAL_REGION_END(); } //------------------------------------------------------------------------------------------ diff --git a/contrib/epee/include/net/net_utils_base.h b/contrib/epee/include/net/net_utils_base.h index 90e352787..f963e7746 100644 --- a/contrib/epee/include/net/net_utils_base.h +++ b/contrib/epee/include/net/net_utils_base.h @@ -55,6 +55,8 @@ namespace net_utils time_t m_last_send; uint64_t m_recv_cnt; uint64_t m_send_cnt; + double m_current_speed_down; + double m_current_speed_up; connection_context_base(boost::uuids::uuid connection_id, long remote_ip, int remote_port, bool is_income, @@ -68,7 +70,9 @@ namespace net_utils m_last_recv(last_recv), m_last_send(last_send), m_recv_cnt(recv_cnt), - m_send_cnt(send_cnt) + m_send_cnt(send_cnt), + m_current_speed_down(0), + m_current_speed_up(0) {} connection_context_base(): m_connection_id(), @@ -79,7 +83,9 @@ namespace net_utils m_last_recv(0), m_last_send(0), m_recv_cnt(0), - m_send_cnt(0) + m_send_cnt(0), + m_current_speed_down(0), + m_current_speed_up(0) {} connection_context_base& operator=(const connection_context_base& a) diff --git a/contrib/epee/include/storages/levin_abstract_invoke2.h b/contrib/epee/include/storages/levin_abstract_invoke2.h index 1b32c51d1..73ede1b12 100644 --- a/contrib/epee/include/storages/levin_abstract_invoke2.h +++ b/contrib/epee/include/storages/levin_abstract_invoke2.h @@ -185,7 +185,7 @@ namespace epee } return res; - }; + } template<class t_owner, class t_in_type, class t_context, class callback_t> int buff_to_t_adapter(t_owner* powner, int command, const std::string& in_buff, callback_t cb, t_context& context) @@ -199,7 +199,7 @@ namespace epee boost::value_initialized<t_in_type> in_struct; static_cast<t_in_type&>(in_struct).load(strg); return cb(command, in_struct, context); - }; + } #define CHAIN_LEVIN_INVOKE_MAP2(context_type) \ int invoke(int command, const std::string& in_buff, std::string& buff_out, context_type& context) \ diff --git a/contrib/otshell_utils/runoptions.cpp b/contrib/otshell_utils/runoptions.cpp index 28e7ceb58..ffd37eae4 100644 --- a/contrib/otshell_utils/runoptions.cpp +++ b/contrib/otshell_utils/runoptions.cpp @@ -7,7 +7,7 @@ namespace nOT { -INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1; // <=== namespaces +INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1 // <=== namespaces // (no debug - this is the default) // +nodebug (no debug) @@ -64,6 +64,6 @@ void cRunOptions::Normalize() { cRunOptions gRunOptions; // (extern) -}; // namespace OT +} // namespace OT diff --git a/contrib/otshell_utils/runoptions.hpp b/contrib/otshell_utils/runoptions.hpp index f3306283a..219d3b509 100644 --- a/contrib/otshell_utils/runoptions.hpp +++ b/contrib/otshell_utils/runoptions.hpp @@ -10,7 +10,7 @@ Template for new files, replace word "template" and later delete this line here. namespace nOT { -INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1; // <=== namespaces +INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1 // <=== namespaces /** Global options to run this program main() Eg used for developer's special options like +setdemo +setdebug. This is NOT for all the other options that are parsed and executed by program. */ @@ -50,7 +50,7 @@ class cRunOptions { extern cRunOptions gRunOptions; -}; // namespace nOT +} // namespace nOT diff --git a/contrib/otshell_utils/utils.cpp b/contrib/otshell_utils/utils.cpp index 489fb3076..1d26075c4 100644 --- a/contrib/otshell_utils/utils.cpp +++ b/contrib/otshell_utils/utils.cpp @@ -26,8 +26,7 @@ #elif defined(__unix__) || defined(__posix) || defined(__linux) || defined(__darwin) || defined(__APPLE__) || defined(__clang__) #define OS_TYPE_POSIX #else - #warning "Compiler/OS platform is not recognized" - #warning "Just assuming it will work as POSIX then" + #warning "Compiler/OS platform is not recognized. Just assuming it will work as POSIX then" #define OS_TYPE_POSIX #endif @@ -44,7 +43,7 @@ namespace nOT { namespace nUtils { -INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1; // <=== namespaces +INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1 // <=== namespaces myexception::myexception(const char * what) : std::runtime_error(what) @@ -78,26 +77,37 @@ std::string & trim(std::string &s) { return ltrim(rtrim(s)); } -std::string get_current_time() -{ - std::stringstream stream; - struct tm * date; - - std::chrono::high_resolution_clock::time_point now = std::chrono::high_resolution_clock::now(); - time_t time_now; - time_now = std::chrono::high_resolution_clock::to_time_t(now); - date = std::localtime(& time_now); - - char date_buff[32]; - std::strftime(date_buff, sizeof(date_buff), "%d-%b-%Y %H:%M:%S.", date); - stream << date_buff; +std::string get_current_time() { + std::chrono::system_clock::time_point now = std::chrono::system_clock::now(); + time_t time_now = std::chrono::system_clock::to_time_t(now); + std::chrono::high_resolution_clock::duration duration = now.time_since_epoch(); + int64_t micro = std::chrono::duration_cast<std::chrono::microseconds>(duration).count(); - std::chrono::high_resolution_clock::duration duration = now.time_since_epoch(); - int64_t micro = std::chrono::duration_cast<std::chrono::microseconds>(duration).count(); - micro %= 1000000; - stream << std::setfill('0') << std::setw(3) << micro; + // std::localtime() - This function may not be thread-safe. + #ifdef OS_TYPE_WINDOWS + struct tm * tm_pointer = std::localtime( &time_now ); // thread-safe on mingw-w64 (thread local variable) and on MSVC btw + // http://stackoverflow.com/questions/18551409/localtime-r-support-on-mingw + // tm_pointer points to thread-local data, memory is owned/managed by the system/library + #else + // linux, freebsd, have this + struct tm tm_object; // automatic storage duration http://en.cppreference.com/w/cpp/language/storage_duration + struct tm * tm_pointer = & tm_object; // just point to our data + auto x = localtime_r( &time_now , tm_pointer ); // modifies our own (this thread) data in tm_object, this is safe http://linux.die.net/man/3/localtime_r + if (x != tm_pointer) return "(internal error in get_current_time)"; // redundant check in case of broken implementation of localtime_r + #endif + // tm_pointer now points to proper time data, and that memory is automatically managed + if (!tm_pointer) return "(internal error in get_current_time - NULL)"; // redundant check in case of broken implementation of used library methods - return stream.str(); + std::stringstream stream; + stream << std::setfill('0') + << std::setw(2) << tm_pointer->tm_year+1900 + << '-' << std::setw(2) << tm_pointer->tm_mon+1 + << '-' << std::setw(2) << tm_pointer->tm_mday + << ' ' << std::setw(2) << tm_pointer->tm_hour + << ':' << std::setw(2) << tm_pointer->tm_min + << ':' << std::setw(2) << tm_pointer->tm_sec + << '.' << std::setw(6) << (micro%1000000); // 6 because microseconds + return stream.str(); } cNullstream g_nullstream; // extern a stream that does nothing (eats/discards data) @@ -213,7 +223,7 @@ void cDebugScopeGuard::Assign(const string &chan, const int level, const string mMsg=msg; } -}; // namespace nDetail +} // namespace nDetail // ==================================================================== @@ -591,10 +601,10 @@ string stringToColor(const string &hash) { // algorthms -}; // namespace nUtil +} // namespace nUtil -}; // namespace OT +} // namespace OT // global namespace diff --git a/contrib/otshell_utils/utils.hpp b/contrib/otshell_utils/utils.hpp index 6cfd11ee1..bb984320b 100644 --- a/contrib/otshell_utils/utils.hpp +++ b/contrib/otshell_utils/utils.hpp @@ -14,7 +14,7 @@ #endif #ifndef CFG_WITH_TERMCOLORS - #error "You requested to turn off terminal colors (CFG_WITH_TERMCOLORS), however currently they are hardcoded (this option to turn them off is not yet implemented)." + //#error "You requested to turn off terminal colors (CFG_WITH_TERMCOLORS), however currently they are hardcoded (this option to turn them off is not yet implemented)." #endif ///Macros related to automatic deduction of class name etc; @@ -35,7 +35,7 @@ class myexception : public std::runtime_error { }; /// @macro Use this macro INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1 as a shortcut for various using std::string etc. -INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1; // <=== namespaces +INJECT_OT_COMMON_USING_NAMESPACE_COMMON_1 // <=== namespaces // ====================================================================================== /// text trimming functions (they do mutate the passes string); they trim based on std::isspace. also return it's reference again @@ -87,6 +87,7 @@ extern std::mutex gLoggerGuard; #define _warn(VAR) _debug_level( 90,VAR) // some problem #define _erro(VAR) _debug_level(100,VAR) // error - report + #define _dbg3_c(C,VAR) _debug_level_c(C, 20,VAR) #define _dbg2_c(C,VAR) _debug_level_c(C, 30,VAR) #define _dbg1_c(C,VAR) _debug_level_c(C, 40,VAR) // details @@ -140,7 +141,7 @@ class cDebugScopeGuard { const char* DbgShortenCodeFileName(const char *s); ///< Returns a pointer to some part of the string that was given, skipping directory names, for log/debug -}; // namespace nDetail +} // namespace nDetail // ========== logger ========== @@ -423,9 +424,9 @@ class value_init { template <class T, T INIT> value_init<T, INIT>::value_init() : data(INIT) { } -}; // namespace nUtils +} // namespace nUtils -}; // namespace nOT +} // namespace nOT // global namespace diff --git a/src/cryptonote_core/blockchain_storage.cpp b/src/cryptonote_core/blockchain_storage.cpp index 78419121f..fe80d75ac 100644 --- a/src/cryptonote_core/blockchain_storage.cpp +++ b/src/cryptonote_core/blockchain_storage.cpp @@ -178,6 +178,22 @@ bool blockchain_storage::store_genesis_block(bool testnet) { return true; } //------------------------------------------------------------------ +void blockchain_storage::logger_handle(long int ms) +{ + std::ofstream log_file; + log_file.open("log/dr-monero/blockchain_log.data", std::ofstream::out | std::ofstream::app); + log_file.precision(7); + + using namespace boost::chrono; + auto point = steady_clock::now(); + auto time_from_epoh = point.time_since_epoch(); + auto m_ms = duration_cast< milliseconds >( time_from_epoh ).count(); + double ms_f = m_ms; + ms_f /= 1000.; + + log_file << ms_f << " " << ms << std::endl; +} +//------------------------------------------------------------------ bool blockchain_storage::store_blockchain() { m_is_blockchain_storing = true; @@ -1760,6 +1776,8 @@ bool blockchain_storage::handle_block_to_main_chain(const block& bl, const crypt << "), coinbase_blob_size: " << coinbase_blob_size << ", cumulative size: " << cumulative_block_size << ", " << block_processing_time << "("<< target_calculating_time << "/" << longhash_calculating_time << ")ms"); + logger_handle(block_processing_time); + bvc.m_added_to_main_chain = true; /*if(!m_orphanes_reorganize_in_work) review_orphaned_blocks_with_new_block_id(id, true);*/ diff --git a/src/cryptonote_core/blockchain_storage.h b/src/cryptonote_core/blockchain_storage.h index 38bdfbce7..6456689b9 100644 --- a/src/cryptonote_core/blockchain_storage.h +++ b/src/cryptonote_core/blockchain_storage.h @@ -249,6 +249,7 @@ namespace cryptonote bool complete_timestamps_vector(uint64_t start_height, std::vector<uint64_t>& timestamps); bool update_next_comulative_size_limit(); bool store_genesis_block(bool testnet); + void logger_handle(long int ms); }; diff --git a/src/cryptonote_core/cryptonote_core.cpp b/src/cryptonote_core/cryptonote_core.cpp index b8b5dc008..c8daa3510 100644 --- a/src/cryptonote_core/cryptonote_core.cpp +++ b/src/cryptonote_core/cryptonote_core.cpp @@ -187,12 +187,35 @@ namespace cryptonote //----------------------------------------------------------------------------------------------- bool core::deinit() { - m_miner.stop(); - m_mempool.deinit(); - m_blockchain_storage.deinit(); + m_miner.stop(); + m_mempool.deinit(); + if (!m_fast_exit) + { + m_blockchain_storage.deinit(); + } return true; } //----------------------------------------------------------------------------------------------- + void core::set_fast_exit() + { + m_fast_exit = true; + } + //----------------------------------------------------------------------------------------------- + bool core::get_fast_exit() + { + return m_fast_exit; + } + //----------------------------------------------------------------------------------------------- + void core::no_check_blocks() + { + m_check_blocks = false; + } + //----------------------------------------------------------------------------------------------- + bool core::get_check_blocks() + { + return m_check_blocks; + } + //----------------------------------------------------------------------------------------------- bool core::handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, bool keeped_by_block) { tvc = boost::value_initialized<tx_verification_context>(); @@ -595,4 +618,6 @@ namespace cryptonote { raise(SIGTERM); } + + std::atomic<bool> core::m_fast_exit(false); } diff --git a/src/cryptonote_core/cryptonote_core.h b/src/cryptonote_core/cryptonote_core.h index 748f2b665..9218aef0f 100644 --- a/src/cryptonote_core/cryptonote_core.h +++ b/src/cryptonote_core/cryptonote_core.h @@ -75,6 +75,10 @@ namespace cryptonote bool init(const boost::program_options::variables_map& vm, bool testnet); bool set_genesis_block(const block& b); bool deinit(); + static void set_fast_exit(); + static bool get_fast_exit(); + void no_check_blocks(); + bool get_check_blocks(); uint64_t get_current_blockchain_height(); bool get_blockchain_top(uint64_t& heeight, crypto::hash& top_id); bool get_blocks(uint64_t start_offset, size_t count, std::list<block>& blocks, std::list<transaction>& txs); @@ -146,6 +150,8 @@ namespace cryptonote bool on_update_blocktemplate_interval(); bool check_tx_inputs_keyimages_diff(const transaction& tx); void graceful_exit(); + static std::atomic<bool> m_fast_exit; + bool m_check_blocks = true; tx_memory_pool m_mempool; diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp b/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp index 6b25cb681..b5a5ceea9 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp +++ b/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp @@ -104,11 +104,11 @@ namespace cryptonote { double cryptonote_protocol_handler_base::estimate_one_block_size() noexcept { // for estimating size of blocks to downloa const double size_min = 500; // XXX 500 - const int history_len = 20; // how many blocks to average over + //const int history_len = 20; // how many blocks to average over double avg=0; try { - avg = get_avg_block_size(history_len); + avg = get_avg_block_size(/*history_len*/); } catch (...) { } avg = std::max( size_min , avg); return avg; @@ -120,96 +120,6 @@ cryptonote_protocol_handler_base::cryptonote_protocol_handler_base() { cryptonote_protocol_handler_base::~cryptonote_protocol_handler_base() { } -void cryptonote_protocol_handler_base::handler_request_blocks_now(size_t &count_limit) { - using namespace epee::net_utils; - size_t est_req_size=0; // how much data are we now requesting (to be soon send to us) - - const auto count_limit_default = count_limit; - - bool allowed_now = false; // are we now allowed to request or are we limited still - // long int size_limit; - - while (!allowed_now) { - /* if ( ::cryptonote::core::get_is_stopping() ) { // TODO fast exit - _fact("ABORT sleep (before sending requeset) due to stopping"); - break; - }*/ - - //LOG_PRINT_RED("[DBG]" << get_avg_block_size(1), LOG_LEVEL_0); - //{ - long int size_limit1=0, size_limit2=0; - //LOG_PRINT_RED("calculating REQUEST size:", LOG_LEVEL_0); - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_in ); - network_throttle_manager::get_global_throttle_in().tick(); - size_limit1 = network_throttle_manager::get_global_throttle_in().get_recommended_size_of_planned_transport(); - } - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_inreq ); - network_throttle_manager::get_global_throttle_inreq().tick(); - size_limit2 = network_throttle_manager::get_global_throttle_inreq().get_recommended_size_of_planned_transport(); - } - - long int one_block_estimated_size = estimate_one_block_size(); - long int limit_small = std::min( size_limit1 , size_limit2 ); - long int size_limit = limit_small/3 + size_limit1/3 + size_limit2/3; - if (limit_small <= 0) size_limit = 0; - const double estimated_peers = 1.2; // how many peers/threads we want to talk to, in order to not grab entire b/w by 1 thread - const double knob = 1.000; - size_limit /= (estimated_peers / estimated_peers) * knob; - _note_c("net/req-calc" , "calculating REQUEST size:" << size_limit1 << " " << size_limit2 << " small=" << limit_small << " final size_limit="<<size_limit); - - double L = size_limit / one_block_estimated_size; // calculating item limit (some heuristics) - //LOG_PRINT_RED("L1 = " << L , LOG_LEVEL_0); - //double L2=0; if (L>1) L2=std::log(L); - //L = L/10. + L2*5; - //LOG_PRINT_RED("L2 = " << L , LOG_LEVEL_0); - L = std::min( (double)count_limit_default, (double)L); - //LOG_PRINT_RED("L3 = " << L , LOG_LEVEL_0); - - const long int hard_limit = 500; // never get more blocks at once ; TODO depend on speed limit. Must be low or limiting is too bursty. - - L = std::min(L, (double) hard_limit); - - count_limit = (int)L; - - est_req_size = count_limit * one_block_estimated_size ; // how much data did we just requested? - - //LOG_PRINT_RED("est_req_size = " << est_req_size , LOG_LEVEL_0); - //LOG_PRINT_RED("count_limit = " << count_limit , LOG_LEVEL_0); - //LOG_PRINT_RED("one_block_estimated_size = " << one_block_estimated_size , LOG_LEVEL_0); - //} - - if (count_limit > 0) allowed_now = true; - // XXX if (!allowed_now) { // XXX DOWNLOAD - //long int ms = 3000; // XXX 2000 - //LOG_PRINT_RED("size_limit = " << size_limit , LOG_LEVEL_0); - long int ms = network_throttle_manager::get_global_throttle_in().get_sleep_time_after_tick(one_block_estimated_size); // XXX too long - //long int ms = network_throttle_manager::get_global_throttle_in().get_sleep_time(count_limit); // XXX - //long int ms = network_throttle_manager::get_global_throttle_in().get_sleep_time(size_limit); // XXX best - - //ms /= 100; // XXX - _info_c("net/sleep", "Sleeping in " << __FUNCTION__ << " for " << ms << " ms"); // XXX debug sleep - //LOG_PRINT_RED("ms = " << ms , LOG_LEVEL_0); - boost::this_thread::sleep(boost::posix_time::milliseconds( ms ) ); // TODO randomize sleeps - //} - } - // done waiting&sleeping ^ - - // ok we are allowed to send now - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_inreq ); - network_throttle_manager::get_global_throttle_inreq().handle_trafic_tcp( est_req_size ); // increase countere of the global requested input - } - - // TODO remove debug - LOG_PRINT_YELLOW("*************************************************************************", LOG_LEVEL_0); - LOG_PRINT_RED("### RRRR ### sending request (type 1), CALCULATED limit = " << count_limit << " = estimated " << est_req_size << " b", LOG_LEVEL_0); - LOG_PRINT_YELLOW("*************************************************************************", LOG_LEVEL_0); - LOG_PRINT_RED("\n", LOG_LEVEL_0); - _note_c("net/req", "### RRRR ### sending request (type 1), CALCULATED limit = " << count_limit << " = estimated " << est_req_size << " b"); -} - void cryptonote_protocol_handler_base::handler_request_blocks_history(std::list<crypto::hash>& ids) { using namespace epee::net_utils; LOG_PRINT_L0("### ~~~RRRR~~~~ ### sending request (type 2), limit = " << ids.size()); diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.h b/src/cryptonote_protocol/cryptonote_protocol_handler.h index b3393928c..0be864dfa 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.h +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.h @@ -46,6 +46,7 @@ #include "cryptonote_core/cryptonote_stat_info.h" #include "cryptonote_core/verification_context.h" #include <netinet/in.h> +#include <boost/circular_buffer.hpp> PUSH_WARNINGS DISABLE_VS_WARNINGS(4355) @@ -61,11 +62,10 @@ namespace cryptonote public: cryptonote_protocol_handler_base(); virtual ~cryptonote_protocol_handler_base(); - void handler_request_blocks_now(size_t & count_limit); // before asking for blocks, can adjust the limit of download void handler_request_blocks_history(std::list<crypto::hash>& ids); // before asking for list of objects, we can change the list still void handler_response_blocks_now(size_t packet_size); - virtual double get_avg_block_size( size_t count) const = 0; + virtual double get_avg_block_size() = 0; virtual double estimate_one_block_size() noexcept; // for estimating size of blocks to download virtual std::ofstream& get_logreq() const =0; @@ -129,10 +129,12 @@ namespace cryptonote nodetool::i_p2p_endpoint<connection_context>* m_p2p; std::atomic<uint32_t> m_syncronized_connections_count; std::atomic<bool> m_synchronized; + bool m_one_request = true; // static std::ofstream m_logreq; - - double get_avg_block_size(size_t count) const; + std::mutex m_buffer_mutex; + double get_avg_block_size(); + boost::circular_buffer<size_t> m_avg_buffer = boost::circular_buffer<size_t>(10); template<class t_parametr> bool post_notify(typename t_parametr::request& arg, cryptonote_connection_context& context) diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.inl b/src/cryptonote_protocol/cryptonote_protocol_handler.inl index aebfcd10d..0130cb384 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.inl +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.inl @@ -41,6 +41,7 @@ #include "cryptonote_core/cryptonote_format_utils.h" #include "profile_tools.h" #include "../../contrib/otshell_utils/utils.hpp" +#include "../../src/p2p/network_throttle-detail.hpp" using namespace nOT::nUtils; namespace cryptonote @@ -115,28 +116,66 @@ namespace cryptonote void t_cryptonote_protocol_handler<t_core>::log_connections() { std::stringstream ss; + ss.precision(1); + + double down_sum = 0.0; + double down_curr_sum = 0.0; + double up_sum = 0.0; + double up_curr_sum = 0.0; ss << std::setw(30) << std::left << "Remote Host" << std::setw(20) << "Peer id" - << std::setw(25) << "Recv/Sent (inactive,sec)" + << std::setw(30) << "Recv/Sent (inactive,sec)" << std::setw(25) << "State" - << std::setw(20) << "Livetime(seconds)" << ENDL; + << std::setw(20) << "Livetime(sec)" + << std::setw(12) << "Down (kB/s)" + << std::setw(14) << "Down(now)" + << std::setw(10) << "Up (kB/s)" + << std::setw(13) << "Up(now)" + << ENDL; uint32_t ip; m_p2p->for_each_connection([&](const connection_context& cntxt, nodetool::peerid_type peer_id) { + bool local_ip = false; ip = ntohl(cntxt.m_remote_ip); + // TODO: local ip in calss A, B + if (ip > 3232235520 && ip < 3232301055) // 192.168.x.x + local_ip = true; + auto connection_time = time(NULL) - cntxt.m_started; ss << std::setw(30) << std::left << std::string(cntxt.m_is_income ? " [INC]":"[OUT]") + epee::string_tools::get_ip_string_from_int32(cntxt.m_remote_ip) + ":" + std::to_string(cntxt.m_remote_port) << std::setw(20) << std::hex << peer_id - << std::setw(25) << std::to_string(cntxt.m_recv_cnt)+ "(" + std::to_string(time(NULL) - cntxt.m_last_recv) + ")" + "/" + std::to_string(cntxt.m_send_cnt) + "(" + std::to_string(time(NULL) - cntxt.m_last_send) + ")" + << std::setw(30) << std::to_string(cntxt.m_recv_cnt)+ "(" + std::to_string(time(NULL) - cntxt.m_last_recv) + ")" + "/" + std::to_string(cntxt.m_send_cnt) + "(" + std::to_string(time(NULL) - cntxt.m_last_send) + ")" << std::setw(25) << get_protocol_state_string(cntxt.m_state) << std::setw(20) << std::to_string(time(NULL) - cntxt.m_started) - << std::setw(10) << (ip > 3232235520 && ip < 3232301055 ? " [LAN]" : "") //TODO: local ip in calss A, B - << ENDL; + << std::setw(12) << std::fixed << (connection_time == 0 ? 0.0 : cntxt.m_recv_cnt / connection_time / 1024) + << std::setw(14) << std::fixed << cntxt.m_current_speed_down / 1024 + << std::setw(10) << std::fixed << (connection_time == 0 ? 0.0 : cntxt.m_send_cnt / connection_time / 1024) + << std::setw(13) << std::fixed << cntxt.m_current_speed_up / 1024 + << (local_ip ? "[LAN]" : "") + << std::left << (ip == 2130706433 ? "[LOCALHOST]" : "") // 127.0.0.1 + << ENDL; + + if (connection_time > 0) + { + down_sum += (cntxt.m_recv_cnt / connection_time / 1024); + up_sum += (cntxt.m_send_cnt / connection_time / 1024); + } + + down_curr_sum += (cntxt.m_current_speed_down / 1024); + up_curr_sum += (cntxt.m_current_speed_up / 1024); + return true; }); - LOG_PRINT_L0("Connections: " << ENDL << ss.str()); + ss << ENDL + << std::setw(125) << " " + << std::setw(12) << down_sum + << std::setw(14) << down_curr_sum + << std::setw(10) << up_sum + << std::setw(13) << up_curr_sum + << ENDL; + LOG_PRINT_L0("Connections: " << ENDL << ss.str()); } //------------------------------------------------------------------------------------------------------------------------ // Returns a list of connection_info objects describing each open p2p connection @@ -332,8 +371,22 @@ namespace cryptonote template<class t_core> - double t_cryptonote_protocol_handler<t_core>::get_avg_block_size( size_t count) const { - return m_core.get_blockchain_storage().get_avg_block_size(count); + 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 + + CRITICAL_REGION_LOCAL(m_buffer_mutex); + double avg = 0; + if (m_avg_buffer.size() == 0) { + _warn("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 + return avg / m_avg_buffer.size(); } @@ -341,6 +394,41 @@ namespace cryptonote int t_cryptonote_protocol_handler<t_core>::handle_response_get_objects(int command, NOTIFY_RESPONSE_GET_OBJECTS::request& arg, cryptonote_connection_context& context) { LOG_PRINT_CCONTEXT_L2("NOTIFY_RESPONSE_GET_OBJECTS"); + + // calculate size of request - mainly for logging/debug + size_t size = 0; + for (auto element : arg.txs) + size += element.size(); + + for (auto element : arg.blocks) + { + size += element.block.size(); + for (auto tx : element.txs) + size += tx.size(); + } + + for (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 + } + /*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();*/ + + //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(context.m_last_response_height > arg.current_blockchain_height) { LOG_ERROR_CCONTEXT("sent wrong NOTIFY_HAVE_OBJECTS: arg.m_current_blockchain_height=" << arg.current_blockchain_height @@ -430,8 +518,9 @@ namespace cryptonote //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); + + if (m_core.get_check_blocks()) + m_core.handle_incoming_block(block_entry.block, bvc, false); if(bvc.m_verifivation_failed) { @@ -448,10 +537,21 @@ namespace cryptonote 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"); + + std::ofstream log_file; + log_file.open("log/dr-monero/get_objects_calc_time.data", std::ofstream::out | std::ofstream::app); + log_file.precision(7); + + using namespace boost::chrono; + auto point = steady_clock::now(); + auto time_from_epoh = point.time_since_epoch(); + auto m_ms = duration_cast< milliseconds >( time_from_epoh ).count(); + double ms_f = m_ms; + ms_f /= 1000.; + + log_file << static_cast<int>(ms_f) << " " << block_process_time + transactions_process_time << std::endl; } } - size_t count_limit = BLOCKS_SYNCHRONIZING_DEFAULT_COUNT; - handler_request_blocks_now(count_limit); // XXX request_missing_objects(context, true); return 1; } @@ -480,6 +580,15 @@ namespace cryptonote template<class t_core> bool t_cryptonote_protocol_handler<t_core>::request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks) { + //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();*/ + if(context.m_needed_objects.size()) { //we know objects that we need, request this objects @@ -487,11 +596,8 @@ namespace cryptonote size_t count = 0; auto it = context.m_needed_objects.begin(); - size_t count_limit = BLOCKS_SYNCHRONIZING_DEFAULT_COUNT; - //handler_request_blocks_now( count_limit ); // change the limit, sleep(?) XXX - // XXX - count_limit=200; // XXX - _note_c("net/req-calc" , "Setting count_limit: " << count_limit); + size_t count_limit = BLOCKS_SYNCHRONIZING_DEFAULT_COUNT; + _note_c("net/req-calc" , "Setting count_limit: " << count_limit); while(it != context.m_needed_objects.end() && count < BLOCKS_SYNCHRONIZING_DEFAULT_COUNT) { if( !(check_having_blocks && m_core.have_block(*it))) @@ -504,6 +610,8 @@ namespace cryptonote } LOG_PRINT_CCONTEXT_L0("-->>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()); + post_notify<NOTIFY_REQUEST_GET_OBJECTS>(req, context); }else if(context.m_last_response_height < context.m_remote_blockchain_height-1) {//we have to fetch more objects ids, request blockchain entry @@ -511,6 +619,12 @@ namespace cryptonote NOTIFY_REQUEST_CHAIN::request r = boost::value_initialized<NOTIFY_REQUEST_CHAIN::request>(); m_core.get_short_chain_history(r.block_ids); 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_L0("r = " << 200); + LOG_PRINT_CCONTEXT_L0("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() ); post_notify<NOTIFY_REQUEST_CHAIN>(r, context); }else diff --git a/src/daemon/daemon.cpp b/src/daemon/daemon.cpp index 9f2ae3167..27a1129dc 100644 --- a/src/daemon/daemon.cpp +++ b/src/daemon/daemon.cpp @@ -69,7 +69,9 @@ namespace , "Run on testnet. The wallet must be launched with --testnet flag." , false }; - const command_line::arg_descriptor<bool> arg_dns_checkpoints = {"enforce-dns-checkpointing", "checkpoints from DNS server will be enforced", false}; + const command_line::arg_descriptor<bool> arg_dns_checkpoints = {"enforce-dns-checkpointing", "checkpoints from DNS server will be enforced", false}; + const command_line::arg_descriptor<bool> arg_test_drop_download = {"test-drop-download", "For network testing, drop downloaded blocks instead checking/adding them to blockchain. Can fake-download blocks very fast."}; + const command_line::arg_descriptor<bool> arg_save_graph = {"save-graph", "Save data for dr monero", false}; } bool command_line_preprocessor(const boost::program_options::variables_map& vm) @@ -99,6 +101,8 @@ bool command_line_preprocessor(const boost::program_options::variables_map& vm) else if (log_space::get_set_log_detalisation_level(false) != new_log_level) { log_space::get_set_log_detalisation_level(true, new_log_level); + int otshell_utils_log_level = 100 - (new_log_level * 25); + gCurrentLogger.setDebugLevel(otshell_utils_log_level); LOG_PRINT_L0("LOG_LEVEL set to " << new_log_level); } @@ -107,7 +111,7 @@ bool command_line_preprocessor(const boost::program_options::variables_map& vm) int main(int argc, char* argv[]) { - + string_tools::set_module_name_and_folder(argv[0]); #ifdef WIN32 _CrtSetDbgFlag ( _CRTDBG_ALLOC_MEM_DF | _CRTDBG_LEAK_CHECK_DF ); @@ -137,6 +141,8 @@ int main(int argc, char* argv[]) command_line::add_arg(desc_cmd_sett, arg_console); command_line::add_arg(desc_cmd_sett, arg_testnet_on); command_line::add_arg(desc_cmd_sett, arg_dns_checkpoints); + command_line::add_arg(desc_cmd_sett, arg_test_drop_download); + command_line::add_arg(desc_cmd_sett, arg_save_graph); cryptonote::core::init_options(desc_cmd_sett); cryptonote::core_rpc_server::init_options(desc_cmd_sett); @@ -231,7 +237,17 @@ int main(int argc, char* argv[]) cryptonote::core_rpc_server rpc_server {ccore, p2psrv, testnet_mode}; cprotocol.set_p2p_endpoint(&p2psrv); ccore.set_cryptonote_protocol(&cprotocol); - daemon_cmmands_handler dch(p2psrv, testnet_mode); + std::shared_ptr<daemon_cmmands_handler> dch(new daemon_cmmands_handler(p2psrv, testnet_mode)); + if(command_line::has_arg(vm, arg_save_graph)) + p2psrv.set_save_graph(true); + + //initialize core here + LOG_PRINT_L0("Initializing core..."); + res = ccore.init(vm, testnet_mode); + CHECK_AND_ASSERT_MES(res, 1, "Failed to initialize core"); + if (command_line::get_arg(vm, arg_test_drop_download)) + ccore.no_check_blocks(); + LOG_PRINT_L0("Core initialized OK"); //initialize objects LOG_PRINT_L0("Initializing P2P server..."); @@ -248,17 +264,11 @@ int main(int argc, char* argv[]) res = rpc_server.init(vm); CHECK_AND_ASSERT_MES(res, 1, "Failed to initialize core RPC server."); LOG_PRINT_GREEN("Core RPC server initialized OK on port: " << rpc_server.get_binded_port(), LOG_LEVEL_0); - - //initialize core here - LOG_PRINT_L0("Initializing core..."); - res = ccore.init(vm, testnet_mode); - CHECK_AND_ASSERT_MES(res, 1, "Failed to initialize core"); - LOG_PRINT_L0("Core initialized OK"); - + // start components if(!command_line::has_arg(vm, arg_console)) { - dch.start_handling(); + dch->start_handling(); } LOG_PRINT_L0("Starting core RPC server..."); @@ -267,7 +277,7 @@ int main(int argc, char* argv[]) LOG_PRINT_L0("Core RPC server started ok"); tools::signal_handler::install([&dch, &p2psrv] { - dch.stop_handling(); + dch->stop_handling(); p2psrv.send_stop_signal(); }); @@ -276,6 +286,8 @@ int main(int argc, char* argv[]) LOG_PRINT_L0("P2P net loop stopped"); //stop components + dch->stop_handling(); + dch.reset(); LOG_PRINT_L0("Stopping core rpc server..."); rpc_server.send_stop_signal(); rpc_server.timed_wait_server_stop(5000); diff --git a/src/daemon/daemon_commands_handler.h b/src/daemon/daemon_commands_handler.h index 7cba4ec5a..6afbbb07f 100644 --- a/src/daemon/daemon_commands_handler.h +++ b/src/daemon/daemon_commands_handler.h @@ -79,6 +79,10 @@ public: m_cmd_binder.set_handler("limit_up", boost::bind(&daemon_cmmands_handler::limit_up, this, _1), "Set upload limit [kB/s]"); m_cmd_binder.set_handler("limit_down", boost::bind(&daemon_cmmands_handler::limit_down, this, _1), "Set download limit [kB/s]"); m_cmd_binder.set_handler("limit", boost::bind(&daemon_cmmands_handler::limit, this, _1), "Set download and upload limit [kB/s]"); + m_cmd_binder.set_handler("fast_exit", boost::bind(&daemon_cmmands_handler::fast_exit, this, _1), "Exit"); + m_cmd_binder.set_handler("test_drop_download", boost::bind(&daemon_cmmands_handler::test_drop_download, this, _1), "For network testing, drop downloaded blocks instead checking/adding them to blockchain. Can fake-download blocks very fast."); + m_cmd_binder.set_handler("start_save_graph", boost::bind(&daemon_cmmands_handler::start_save_graph, this, _1), ""); + m_cmd_binder.set_handler("stop_save_graph", boost::bind(&daemon_cmmands_handler::stop_save_graph, this, _1), ""); } bool start_handling() @@ -240,6 +244,8 @@ private: } log_space::log_singletone::get_set_log_detalisation_level(true, l); + int otshell_utils_log_level = 100 - (l * 25); + gCurrentLogger.setDebugLevel(otshell_utils_log_level); return true; } @@ -406,7 +412,7 @@ private: //-------------------------------------------------------------------------------- bool out_peers_limit(const std::vector<std::string>& args) { if(args.size()!=1) { - std::cout << "Usage: limit_down <speed>" << ENDL; + std::cout << "Usage: out_peers <number_of_peers>" << ENDL; return true; } @@ -420,13 +426,26 @@ private: return false; } + using namespace boost::chrono; + auto point = steady_clock::now(); + auto time_from_epoh = point.time_since_epoch(); + auto ms = duration_cast< milliseconds >( time_from_epoh ).count(); + double ms_f = ms; + ms_f /= 1000.; + + std::ofstream limitFile("log/dr-monero/peers_limit.info", std::ios::app); + limitFile.precision(7); + limitFile << ms_f << " " << static_cast<int>(limit) << std::endl; if (m_srv.m_config.m_net_config.connections_count > limit) { - int count = m_srv.m_config.m_net_config.connections_count - limit; m_srv.m_config.m_net_config.connections_count = limit; - m_srv.delete_connections(count); - } - else + if (m_srv.m_number_of_out_peers > limit) + { + int count = m_srv.m_number_of_out_peers - limit; + m_srv.delete_connections(count); + } + } + else m_srv.m_config.m_net_config.connections_count = limit; return true; @@ -527,4 +546,29 @@ private: return true; } + //-------------------------------------------------------------------------------- + bool fast_exit(const std::vector<std::string>& args) + { + m_srv.get_payload_object().get_core().set_fast_exit(); + m_srv.send_stop_signal(); + return true; + } + //-------------------------------------------------------------------------------- + bool test_drop_download(const std::vector<std::string>& args) + { + m_srv.get_payload_object().get_core().no_check_blocks(); + return true; + } + //-------------------------------------------------------------------------------- + bool start_save_graph(const std::vector<std::string>& args) + { + m_srv.set_save_graph(true); + return true; + } + //-------------------------------------------------------------------------------- + bool stop_save_graph(const std::vector<std::string>& args) + { + m_srv.set_save_graph(false); + return true; + } }; diff --git a/src/p2p/connection_basic.cpp b/src/p2p/connection_basic.cpp index 35b0d4c8e..0e2fd5942 100644 --- a/src/p2p/connection_basic.cpp +++ b/src/p2p/connection_basic.cpp @@ -78,6 +78,7 @@ #include "../../contrib/epee/include/net/abstract_tcp_server2.h" #include "../../contrib/otshell_utils/utils.hpp" +#include "data_logger.hpp" using namespace nOT::nUtils; // TODO: @@ -146,31 +147,31 @@ connection_basic::connection_basic(boost::asio::io_service& io_service, std::ato { ++ref_sock_count; // increase the global counter mI->m_peer_number = sock_number.fetch_add(1); // use, and increase the generated number - _note("Spawned connection p2p#"<<mI->m_peer_number<<" currently we have sockets count:" << m_ref_sock_count); + + string remote_addr_str = "?"; + try { remote_addr_str = socket_.remote_endpoint().address().to_string(); } catch(...){} ; + + _note("Spawned connection p2p#"<<mI->m_peer_number<<" to " << remote_addr_str << " currently we have sockets count:" << m_ref_sock_count); boost::filesystem::create_directories("log/dr-monero/net/"); - /*boost::asio::SettableSocketOption option;// = new boost::asio::SettableSocketOption(); - option.level(IPPROTO_IP); - option.name(IP_TOS); - option.value(&tos); - option.size = sizeof(tos); - socket_.set_option(option);*/ - // TODO socket options } connection_basic::~connection_basic() { - _note("Destructing connection p2p#"<<mI->m_peer_number); + string remote_addr_str = "?"; + try { remote_addr_str = socket_.remote_endpoint().address().to_string(); } catch(...){} ; + _note("Destructing connection p2p#"<<mI->m_peer_number << " to " << remote_addr_str); } void connection_basic::set_rate_up_limit(uint64_t limit) { - save_limit_to_file(limit); + + // TODO remove __SCALING_FACTOR... + const double SCALING_FACTOR = 2.1; // to acheve the best performance + limit *= SCALING_FACTOR; { - // TODO remove __SCALING_FACTOR... - const double SCALING_FACTOR = 2.25; // to acheve the best performance - limit *= SCALING_FACTOR; CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_out ); network_throttle_manager::get_global_throttle_out().set_target_speed(limit); + network_throttle_manager::get_global_throttle_out().set_real_target_speed(limit / SCALING_FACTOR); } - // connection_basic_pimpl::m_throttle_global.m_out.set_target_speed(limit); + save_limit_to_file(limit); } void connection_basic::set_rate_down_limit(uint64_t limit) { @@ -186,36 +187,30 @@ void connection_basic::set_rate_down_limit(uint64_t limit) { save_limit_to_file(limit); } -void connection_basic::set_rate_limit(uint64_t limit) { - // TODO -} -void connection_basic::set_kill_limit (uint64_t limit) { - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_in ); - network_throttle_manager::get_global_throttle_in().set_target_kill(limit); - } - - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_out ); - network_throttle_manager::get_global_throttle_out().set_target_kill(limit); - } - - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_inreq ); - network_throttle_manager::get_global_throttle_inreq().set_target_kill(limit); - } -} void connection_basic::save_limit_to_file(int limit) { // saving limit to file - std::ofstream file; - file.open("log/dr-monero/limit.info"); - file << limit; -} - -void connection_basic::set_rate_autodetect(uint64_t limit) { - // TODO - LOG_PRINT_L0("inside connection_basic we set autodetect (this is additional notification).."); + if (!epee::net_utils::data_logger::m_save_graph) + return; + std::ofstream file_up, file_down; + file_up.open("log/dr-monero/limit_up.info", std::ofstream::out | std::ofstream::app); + file_up.precision(8); + file_down.open("log/dr-monero/limit_down.info", std::ofstream::out | std::ofstream::app); + file_down.precision(8); + using namespace boost::chrono; + auto point = steady_clock::now(); + auto time_from_epoh = point.time_since_epoch(); + auto s = duration_cast< seconds >( time_from_epoh ).count(); + + { + CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_out ); + file_up << s << " " << network_throttle_manager::get_global_throttle_out().get_terget_speed() / 1024 << "\n"; + } + + { + CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_in ); + file_down << s << " " << network_throttle_manager::get_global_throttle_in().get_terget_speed() / 1024 << "\n"; + } } void connection_basic::set_tos_flag(int tos) { @@ -230,39 +225,30 @@ void connection_basic::sleep_before_packet(size_t packet_size, int phase, int q double delay=0; // will be calculated do { // rate limiting - //XXX - /*if (::cryptonote::core::get_is_stopping()) { - _dbg1("We are stopping - so abort sleep"); - return; - }*/ if (m_was_shutdown) { _dbg2("m_was_shutdown - so abort sleep"); return; } { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_out ); + CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_out ); delay = network_throttle_manager::get_global_throttle_out().get_sleep_time_after_tick( packet_size ); // decission from global } - delay *= 0.50; - delay = 0; // XXX if (delay > 0) { - //delay += rand2*0.1; - long int ms = (long int)(delay * 1000); - _info_c("net/sleep", "Sleeping in " << __FUNCTION__ << " for " << ms << " ms before packet_size="<<packet_size); // XXX debug sleep + long int ms = (long int)(delay * 1000); + _info_c("net/sleep", "Sleeping in " << __FUNCTION__ << " for " << ms << " ms before packet_size="<<packet_size); // debug sleep _dbg1("sleep in sleep_before_packet"); - boost::this_thread::sleep(boost::posix_time::milliseconds( ms ) ); // TODO randomize sleeps + epee::net_utils::data_logger::get_instance().add_data("sleep_up", ms); + boost::this_thread::sleep(boost::posix_time::milliseconds( ms ) ); } } while(delay > 0); // XXX LATER XXX { CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_out ); - network_throttle_manager::get_global_throttle_out().handle_trafic_tcp( packet_size ); // increase counter - global - //epee::critical_region_t<decltype(m_throttle_global_lock)> guard(m_throttle_global_lock); // *** critical *** - //m_throttle_global.m_out.handle_trafic_tcp( packet_size ); // increase counter - global + network_throttle_manager::get_global_throttle_out().handle_trafic_exact( packet_size * 700); // increase counter - global } } @@ -271,34 +257,12 @@ void connection_basic::set_start_time() { m_start_time = network_throttle_manager::get_global_throttle_out().get_time_seconds(); } -void connection_basic::do_send_handler_start(const void* ptr , size_t cb ) { - _fact_c("net/out/size", "*** do_sen() called for packet="<<cb<<" B"); - sleep_before_packet(cb,1,-1); - // set_start_time(); -} - -void connection_basic::do_send_handler_delayed(const void* ptr , size_t cb ) { - // CRITICAL_REGION_LOCAL(network_throttle_manager::m_lock_get_global_throttle_out); - // auto sending_time = network_throttle_manager::get_global_throttle_out().get_time_seconds() - m_start_time; // wrong? --r -} - void connection_basic::do_send_handler_write(const void* ptr , size_t cb ) { sleep_before_packet(cb,1,-1); _info_c("net/out/size", "handler_write (direct) - before ASIO write, for packet="<<cb<<" B (after sleep)"); set_start_time(); } -void connection_basic::do_send_handler_stop(const void* ptr , size_t cb ) { -} - -void connection_basic::do_send_handler_after_write(const boost::system::error_code& e, size_t cb) { - // CRITICAL_REGION_LOCAL(network_throttle_manager::m_lock_get_global_throttle_out); - // auto sending_time = network_throttle_manager::get_global_throttle_out().get_time_seconds() - m_start_time; - // lag: if current sending time > max sending time - //if (sending_time > 0.1) network_throttle_manager::get_global_throttle_out().set_overheat(sending_time); // TODO - -} - void connection_basic::do_send_handler_write_from_queue( const boost::system::error_code& e, size_t cb, int q_len ) { sleep_before_packet(cb,2,q_len); _info_c("net/out/size", "handler_write (after write, from queue="<<q_len<<") - before ASIO write, for packet="<<cb<<" B (after sleep)"); @@ -306,49 +270,14 @@ void connection_basic::do_send_handler_write_from_queue( const boost::system::er set_start_time(); } -void connection_basic::do_read_handler_start(const boost::system::error_code& e, std::size_t bytes_transferred) { // from read, after read completion - const size_t packet_size = bytes_transferred; - { - CRITICAL_REGION_LOCAL( network_throttle_manager::m_lock_get_global_throttle_in ); - // sleep_before_packet(packet_size * __SCALING_FACTOR, 1, -1); // TODO remove __SCALING_FACTOR - network_throttle_manager::get_global_throttle_in().handle_trafic_tcp( packet_size ); // increase counter - global - // epee::critical_region_t<decltype(mI->m_throttle_global_lock)> guard(mI->m_throttle_global_lock); // *** critical *** - // mI->m_throttle_global.m_in.handle_trafic_tcp( packet_size ); // increase counter - global - } -} - -void connection_basic::logger_handle_net_peer(size_t size, bool io) { // network data written - // TODO OPTIMIZE! do NOT reopen idiotically :) - std::ostringstream oss; - std::string filename; - if (io) { // write - double time = network_throttle_manager::get_global_throttle_in().get_time_seconds() ; - oss << "log/dr-monero/net/in-peer-" << (mI->m_peer_number) << ".dat" << std::ends; - filename = oss.str(); - network_throttle_manager::get_global_throttle_out().logger_handle_net(filename,time,size); - } - else { // read - double time = network_throttle_manager::get_global_throttle_out().get_time_seconds() ; - oss << "log/dr-monero/net/out-peer-" << (mI->m_peer_number) << ".dat" << std::ends; - filename = oss.str(); - network_throttle_manager::get_global_throttle_in().logger_handle_net(filename,time,size); - } -} - void connection_basic::logger_handle_net_read(size_t size) { // network data read - std::string filename = "log/dr-monero/net/in-all.data"; - - double time = network_throttle_manager::get_global_throttle_in().get_time_seconds() ; - network_throttle_manager::get_global_throttle_in().logger_handle_net(filename, time, size); - logger_handle_net_peer(size,0); + size /= 1024; + epee::net_utils::data_logger::get_instance().add_data("download", size); } void connection_basic::logger_handle_net_write(size_t size) { - std::string filename = "log/dr-monero/net/out-all.data"; - double time = network_throttle_manager::get_global_throttle_out().get_time_seconds() ; - network_throttle_manager::get_global_throttle_out().logger_handle_net(filename, time, size); - logger_handle_net_peer(size,1); - + size /= 1024; + epee::net_utils::data_logger::get_instance().add_data("upload", size); } double connection_basic::get_sleep_time(size_t cb) { @@ -356,6 +285,10 @@ double connection_basic::get_sleep_time(size_t cb) { return t; } +void connection_basic::set_save_graph(bool save_graph) { + epee::net_utils::data_logger::m_save_graph = save_graph; +} + } // namespace } // namespace diff --git a/src/p2p/connection_basic.hpp b/src/p2p/connection_basic.hpp index 1b5a2c8ad..e9fdc3add 100644 --- a/src/p2p/connection_basic.hpp +++ b/src/p2p/connection_basic.hpp @@ -99,17 +99,11 @@ class connection_basic { // not-templated base class for rapid developmet of som virtual ~connection_basic(); // various handlers to be called from connection class: - void do_send_handler_start(const void * ptr , size_t cb); - void do_send_handler_delayed(const void * ptr , size_t cb); void do_send_handler_write(const void * ptr , size_t cb); - void do_send_handler_stop(const void * ptr , size_t cb); - void do_send_handler_after_write( const boost::system::error_code& e, size_t cb ); // from handle_write void do_send_handler_write_from_queue(const boost::system::error_code& e, size_t cb , int q_len); // from handle_write, sending next part - void do_read_handler_start(const boost::system::error_code& e, std::size_t bytes_transferred); // from read, after read completion void logger_handle_net_write(size_t size); // network data written void logger_handle_net_read(size_t size); // network data read - void logger_handle_net_peer(size_t size, bool io); void set_start_time(); @@ -117,9 +111,6 @@ class connection_basic { // not-templated base class for rapid developmet of som static void set_rate_up_limit(uint64_t limit); static void set_rate_down_limit(uint64_t limit); - static void set_rate_limit(uint64_t limit); - static void set_rate_autodetect(uint64_t limit); - static void set_kill_limit (uint64_t limit); // config misc static void set_tos_flag(int tos); // ToS / QoS flag @@ -129,6 +120,8 @@ class connection_basic { // not-templated base class for rapid developmet of som void sleep_before_packet(size_t packet_size, int phase, int q_len); // execute a sleep ; phase is not really used now(?) static void save_limit_to_file(int limit); ///< for dr-monero static double get_sleep_time(size_t cb); + + static void set_save_graph(bool save_graph); }; } // nameserver diff --git a/src/p2p/data_logger.cpp b/src/p2p/data_logger.cpp new file mode 100644 index 000000000..6a8eb25be --- /dev/null +++ b/src/p2p/data_logger.cpp @@ -0,0 +1,81 @@ +#include "data_logger.hpp" + +#include <boost/chrono.hpp> +#include <chrono> + +namespace epee +{ +namespace net_utils +{ + data_logger &data_logger::get_instance() + { + static data_logger instance; + return instance; + } + + data_logger::data_logger() + { + //create timer + std::shared_ptr<std::thread> logger_thread(new std::thread([&]() + { + while (true) + { + std::this_thread::sleep_for(std::chrono::seconds(1)); + saveToFile(); + } + })); + logger_thread->detach(); + + mFilesMap["peers"] = data_logger::fileData("log/dr-monero/peers.data"); + mFilesMap["download"] = data_logger::fileData("log/dr-monero/net/in-all.data"); + mFilesMap["upload"] = data_logger::fileData("log/dr-monero/net/out-all.data"); + mFilesMap["request"] = data_logger::fileData("log/dr-monero/net/req-all.data"); + mFilesMap["sleep_down"] = data_logger::fileData("log/dr-monero/down_sleep_log.data"); + mFilesMap["sleep_up"] = data_logger::fileData("log/dr-monero/up_sleep_log.data"); + + } + + void data_logger::add_data(std::string filename, unsigned int data) + { + if (mFilesMap.find(filename) == mFilesMap.end()) + return; // TODO: exception + + mFilesMap[filename].mDataToSave += data; + } + + double data_logger::fileData::get_current_time() + { + using namespace boost::chrono; + auto point = steady_clock::now(); + auto time_from_epoh = point.time_since_epoch(); + auto ms = duration_cast< milliseconds >( time_from_epoh ).count(); + double ms_f = ms; + return ms_f / 1000.; + } + + data_logger::fileData::fileData(std::string pFile) + { + mFile = std::make_shared<std::ofstream> (pFile); + } + + void data_logger::fileData::save() + { + if (!data_logger::m_save_graph) + return; + *mFile << static_cast<int>(get_current_time()) << " " << mDataToSave << std::endl; + } + + void data_logger::saveToFile() + { + std::lock_guard<std::mutex> lock(mSaveMutex); + for (auto &element : mFilesMap) + { + element.second.save(); + element.second.mDataToSave = 0; + } + } + +std::atomic<bool> data_logger::m_save_graph(false); + +} // namespace +} // namespace diff --git a/src/p2p/data_logger.hpp b/src/p2p/data_logger.hpp new file mode 100644 index 000000000..2b8503df3 --- /dev/null +++ b/src/p2p/data_logger.hpp @@ -0,0 +1,46 @@ +#ifndef INCLUDED_p2p_data_logger_hpp +#define INCLUDED_p2p_data_logger_hpp + +#include <string> +#include <map> +#include <fstream> +#include <memory> +#include <thread> +#include <mutex> +#include <atomic> + +namespace epee +{ +namespace net_utils +{ + + class data_logger { + public: + static data_logger &get_instance(); + data_logger(const data_logger &ob) = delete; + data_logger(data_logger &&ob) = delete; + void add_data(std::string filename, unsigned int data); + static std::atomic<bool> m_save_graph; + private: + data_logger(); + class fileData + { + public: + fileData(){} + fileData(std::string pFile); + + std::shared_ptr<std::ofstream> mFile; + long int mDataToSave = 0; + static double get_current_time(); + void save(); + }; + + std::map <std::string, fileData> mFilesMap; + std::mutex mSaveMutex; + void saveToFile(); + }; + +} // namespace +} // namespace + +#endif diff --git a/src/p2p/net_node.h b/src/p2p/net_node.h index 48737193e..ea7d5c383 100644 --- a/src/p2p/net_node.h +++ b/src/p2p/net_node.h @@ -86,7 +86,10 @@ namespace nodetool m_no_igd(false), m_hide_my_port(false), m_network_id(std::move(network_id)) - {} + { + m_number_of_out_peers = 0; + m_save_graph = false; + } static void init_options(boost::program_options::options_description& desc); @@ -225,6 +228,12 @@ namespace nodetool public: config m_config; // TODO was private, add getters? + std::atomic<unsigned int> m_number_of_out_peers; + void set_save_graph(bool save_graph) + { + m_save_graph = save_graph; + epee::net_utils::connection_basic::set_save_graph(save_graph); + } private: std::string m_config_folder; @@ -237,6 +246,7 @@ namespace nodetool bool m_allow_local_ip; bool m_hide_my_port; bool m_no_igd; + std::atomic<bool> m_save_graph; //critical_section m_connections_lock; //connections_indexed_container m_connections; diff --git a/src/p2p/net_node.inl b/src/p2p/net_node.inl index ce70e241a..60eed1f36 100644 --- a/src/p2p/net_node.inl +++ b/src/p2p/net_node.inl @@ -46,6 +46,7 @@ #include "net/local_ip.h" #include "crypto/crypto.h" #include "storages/levin_abstract_invoke2.h" +#include "data_logger.hpp" // We have to look for miniupnpc headers in different places, dependent on if its compiled or external #ifdef UPNP_STATIC @@ -85,8 +86,8 @@ namespace nodetool const command_line::arg_descriptor<std::vector<std::string> > arg_p2p_seed_node = {"seed-node", "Connect to a node to retrieve peer addresses, and disconnect"}; const command_line::arg_descriptor<bool> arg_p2p_hide_my_port = {"hide-my-port", "Do not announce yourself as peerlist candidate", false, true}; - const command_line::arg_descriptor<bool> arg_no_igd = {"no-igd", "Disable UPnP port mapping"}; - const command_line::arg_descriptor<int64_t> arg_out_peers = {"out-peers", "set max limit of out peers", -1}; + const command_line::arg_descriptor<bool> arg_no_igd = {"no-igd", "Disable UPnP port mapping"}; + const command_line::arg_descriptor<int64_t> arg_out_peers = {"out-peers", "set max limit of out peers", -1}; const command_line::arg_descriptor<int> arg_tos_flag = {"tos-flag", "set TOS flag", -1}; const command_line::arg_descriptor<int64_t> arg_limit_rate_up = {"limit-rate-up", "set limit-rate-up [kB/s]", -1}; @@ -289,6 +290,31 @@ namespace nodetool std::vector<std::vector<std::string>> dns_results; dns_results.resize(m_seed_nodes_list.size()); + + std::shared_ptr<std::thread> peersLoggerThread (new std::thread([&]() + { + unsigned int number_of_peers; + while (1) + { + if (m_save_graph) + { + //number_of_peers = m_net_server.get_config_object().get_connections_count(); + number_of_peers = 0; + m_net_server.get_config_object().foreach_connection([&](const p2p_connection_context& cntxt) + { + if(!cntxt.m_is_income) + ++number_of_peers; + return true; + }); // lambda + + m_number_of_out_peers = number_of_peers; + epee::net_utils::data_logger::get_instance().add_data("peers", number_of_peers); + } + std::this_thread::sleep_for(std::chrono::seconds(1)); + } + })); // lambda + + peersLoggerThread->detach(); std::list<boost::thread*> dns_threads; uint64_t result_index = 0; @@ -487,6 +513,7 @@ namespace nodetool { m_peerlist.deinit(); m_net_server.deinit_server(); + return store_config(); } //----------------------------------------------------------------------------------- @@ -697,6 +724,16 @@ namespace nodetool template<class t_payload_net_handler> bool node_server<t_payload_net_handler>::try_to_connect_and_handshake_with_new_peer(const net_address& na, bool just_take_peerlist, uint64_t last_seen_stamp, bool white) { + if (m_number_of_out_peers == m_config.m_net_config.connections_count) // out peers limit + { + return false; + } + else if (m_number_of_out_peers > m_config.m_net_config.connections_count) + { + m_net_server.get_config_object().del_out_connections(1); + m_number_of_out_peers --; // atomic variable, update time = 1s + return false; + } LOG_PRINT_L1("Connecting to " << epee::string_tools::get_ip_string_from_int32(na.ip) << ":" << epee::string_tools::num_to_string_fast(na.port) << "(white=" << white << ", last_seen: " << (last_seen_stamp ? epee::misc_utils::get_time_interval_string(time(NULL) - last_seen_stamp):"never") @@ -784,16 +821,22 @@ namespace nodetool ++try_count; - if(is_peer_used(pe)) + _note("Considering connecting (out) to peer: " << pe.id << " " << epee::string_tools::get_ip_string_from_int32(pe.adr.ip) << ":" << boost::lexical_cast<std::string>(pe.adr.port)); + + if(is_peer_used(pe)) { + _note("Peer is used"); continue; + } LOG_PRINT_L1("Selected peer: " << pe.id << " " << epee::string_tools::get_ip_string_from_int32(pe.adr.ip) << ":" << boost::lexical_cast<std::string>(pe.adr.port) << "[white=" << use_white_list << "] last_seen: " << (pe.last_seen ? epee::misc_utils::get_time_interval_string(time(NULL) - pe.last_seen) : "never")); - if(!try_to_connect_and_handshake_with_new_peer(pe.adr, false, pe.last_seen, use_white_list)) + if(!try_to_connect_and_handshake_with_new_peer(pe.adr, false, pe.last_seen, use_white_list)) { + _note("Handshake failed"); continue; + } return true; } @@ -1336,20 +1379,31 @@ namespace nodetool template<class t_payload_net_handler> bool node_server<t_payload_net_handler>::set_max_out_peers(const boost::program_options::variables_map& vm, int64_t max) { + using namespace std::chrono; + auto point = steady_clock::now(); + auto time_from_epoh = point.time_since_epoch(); + auto ms = duration_cast< milliseconds >( time_from_epoh ).count(); + double ms_f = ms; + ms_f /= 1000.; + + std::ofstream limitFile("log/dr-monero/peers_limit.info", std::ios::app); + limitFile.precision(7); if(max == -1) { m_config.m_net_config.connections_count = P2P_DEFAULT_CONNECTIONS_COUNT; + if (m_save_graph) + limitFile << static_cast<int>(ms_f) << " " << P2P_DEFAULT_CONNECTIONS_COUNT << std::endl; return true; } m_config.m_net_config.connections_count = max; - LOG_PRINT_RED_L0("connections_count: " << m_config.m_net_config.connections_count); + limitFile << static_cast<int>(ms_f) << " " << max << std::endl; return true; } template<class t_payload_net_handler> void node_server<t_payload_net_handler>::delete_connections(size_t count) { - m_net_server.get_config_object().del_connections(count); + m_net_server.get_config_object().del_out_connections(count); } template<class t_payload_net_handler> diff --git a/src/p2p/network_throttle-detail.cpp b/src/p2p/network_throttle-detail.cpp index 6ea3076a9..6b2ee698e 100644 --- a/src/p2p/network_throttle-detail.cpp +++ b/src/p2p/network_throttle-detail.cpp @@ -78,6 +78,7 @@ #include "../../src/p2p/network_throttle-detail.hpp" #include "../../contrib/otshell_utils/utils.hpp" +#include "data_logger.hpp" using namespace nOT::nUtils; // ################################################################################################ @@ -152,8 +153,6 @@ network_throttle::network_throttle(const std::string &nameshort, const std::stri m_any_packet_yet = false; m_slot_size = 1.0; // hard coded in few places m_target_speed = 16 * 1024; // other defaults are probably defined in the command-line parsing code when this class is used e.g. as main global throttle - m_target_MB = 0; - } void network_throttle::set_name(const std::string &name) @@ -163,16 +162,20 @@ void network_throttle::set_name(const std::string &name) void network_throttle::set_target_speed( network_speed_kbps target ) { - m_target_speed = target; + m_target_speed = target * 1024; _note_c("net/"+m_nameshort, "Setting LIMIT: " << target << " kbps"); + set_real_target_speed(target); } -void network_throttle::set_target_kill( network_MB target ) +void network_throttle::set_real_target_speed( network_speed_kbps real_target ) { - _note_c("net/"+m_nameshort, "Setting KILL: " << target << " MB hard limit"); - m_target_MB = target; + m_real_target_speed = real_target * 1024; } +network_speed_kbps network_throttle::get_terget_speed() +{ + return m_real_target_speed / 1024; +} void network_throttle::tick() { @@ -187,7 +190,7 @@ void network_throttle::tick() // TODO optimize when moving few slots at once while ( (!m_any_packet_yet) || (last_sample_time_slot < current_sample_time_slot)) { - LOG_PRINT_L4("Moving counter buffer by 1 second " << last_sample_time_slot << " < " << current_sample_time_slot << " (last time " << m_last_sample_time<<")"); + _dbg3("Moving counter buffer by 1 second " << last_sample_time_slot << " < " << current_sample_time_slot << " (last time " << m_last_sample_time<<")"); // rotate buffer for (size_t i=m_history.size()-1; i>=1; --i) m_history[i] = m_history[i-1]; m_history[0] = packet_info(); @@ -217,7 +220,6 @@ void network_throttle::_handle_trafic_exact(size_t packet_size, size_t orginal_s std::ostringstream oss; oss << "["; for (auto sample: m_history) oss << sample.m_size << " "; oss << "]" << std::ends; std::string history_str = oss.str(); - logger_handle_net("log/dr-monero/net/inreq-all.data",get_time_seconds(),packet_size); _info_c( "net/" + m_nameshort , "Throttle " << m_name << ": packet of ~"<<packet_size<<"b " << " (from "<<orginal_size<<" b)" << " Speed AVG=" << std::setw(4) << ((long int)(cts .average/1024)) <<"[w="<<cts .window<<"]" << " " << std::setw(4) << ((long int)(cts2.average/1024)) <<"[w="<<cts2.window<<"]" @@ -233,23 +235,22 @@ void network_throttle::handle_trafic_tcp(size_t packet_size) _handle_trafic_exact( all_size , packet_size ); } -void network_throttle::handle_congestion(double overheat) { - // TODO -} - network_time_seconds network_throttle::get_sleep_time_after_tick(size_t packet_size) { tick(); return get_sleep_time(packet_size); } void network_throttle::logger_handle_net(const std::string &filename, double time, size_t size) { + if (! epee::net_utils::data_logger::m_save_graph) + return; std::mutex mutex; mutex.lock(); { std::fstream file; file.open(filename.c_str(), std::ios::app | std::ios::out ); + file.precision(6); if(!file.is_open()) _warn("Can't open file " << filename); - file << time << " " << size/1024 << "\n"; + file << static_cast<int>(time) << " " << static_cast<double>(size/1024) << "\n"; file.close(); } mutex.unlock(); } @@ -257,27 +258,11 @@ void network_throttle::logger_handle_net(const std::string &filename, double tim // fine tune this to decide about sending speed: network_time_seconds network_throttle::get_sleep_time(size_t packet_size) const { - //_scope_mark(""); double D2=0; calculate_times_struct cts = { 0, 0, 0, 0}; - //calculate_times(packet_size, cts, false, m_window_size/2); D2=cts.delay; - //calculate_times(packet_size, cts, true, m_window_size/2); D2=cts.delay; calculate_times(packet_size, cts, true, m_window_size); D2=cts.delay; return D2; } -double network_throttle::get_current_overheat() const { - auto now = get_time_seconds(); - auto diff = now - m_overheat_time; - auto overheat = m_overheat - diff; - overheat = std::max(m_overheat, 0.); - return overheat; -} - -void network_throttle::set_overheat(double lag) { - m_overheat += lag; - m_overheat_time = get_time_seconds(); - LOG_PRINT_L0("Lag: " << lag << ", overheat: " << m_overheat ); -} // MAIN LOGIC: void network_throttle::calculate_times(size_t packet_size, calculate_times_struct &cts, bool dbg, double force_window) const @@ -310,9 +295,7 @@ void network_throttle::calculate_times(size_t packet_size, calculate_times_struc const double D1 = (Epast - M*cts.window) / M; // delay - how long to sleep to get back to target speed const double D2 = (Enow - M*cts.window) / M; // delay - how long to sleep to get back to target speed (including current packet) - auto O = get_current_overheat(); - auto Ouse = O * 0 ; // XXX TODO - cts.delay = (D1*0.80 + D2*0.20) + Ouse; // finall sleep depends on both with/without current packet + cts.delay = (D1*0.80 + D2*0.20); // finall sleep depends on both with/without current packet // update_overheat(); cts.average = Epast/cts.window; // current avg. speed (for info) @@ -329,13 +312,13 @@ void network_throttle::calculate_times(size_t packet_size, calculate_times_struc if (dbg) { std::ostringstream oss; oss << "["; for (auto sample: m_history) oss << sample.m_size << " "; oss << "]" << std::ends; std::string history_str = oss.str(); - _dbg1_c( "net/"+m_nameshort+"_c" , - "dbg " << m_name << ": " + _info_c( "net/"+m_nameshort+"_c" , + (cts.delay > 0 ? "SLEEP" : "") + << "dbg " << m_name << ": " << "speed is A=" << std::setw(8) <<cts.average<<" vs " << "Max=" << std::setw(8) <<M<<" " << " so sleep: " << "D=" << std::setw(8) <<cts.delay<<" sec " - << "Overheat=" << std::setw(8) <<O<<" sec " << "E="<< std::setw(8) << E << " (Enow="<<std::setw(8)<<Enow<<") " << "M=" << std::setw(8) << M <<" W="<< std::setw(8) << cts.window << " " << "R=" << std::setw(8) << cts.recomendetDataSize << " Wgood" << std::setw(8) << Wgood << " " @@ -347,7 +330,7 @@ void network_throttle::calculate_times(size_t packet_size, calculate_times_struc } double network_throttle::get_time_seconds() const { - using namespace boost::chrono; + using namespace std::chrono; auto point = steady_clock::now(); auto time_from_epoh = point.time_since_epoch(); auto ms = duration_cast< milliseconds >( time_from_epoh ).count(); @@ -368,14 +351,28 @@ size_t network_throttle::get_recommended_size_of_planned_transport_window(double size_t network_throttle::get_recommended_size_of_planned_transport() const { size_t R1=0,R2=0,R3=0; R1 = get_recommended_size_of_planned_transport_window( -1 ); - R2 = get_recommended_size_of_planned_transport_window( m_window_size/2); - R3 = get_recommended_size_of_planned_transport_window( 8 ); + R2 = get_recommended_size_of_planned_transport_window(m_window_size / 2); + R3 = get_recommended_size_of_planned_transport_window( 5 ); auto RM = std::min(R1, std::min(R2,R3)); - const double a1=70, a2=10, a3=10, am=10; // weight of the various windows in decisssion + const double a1=20, a2=10, a3=10, am=10; // weight of the various windows in decisssion // TODO 70 => 20 return (R1*a1 + R2*a2 + R3*a3 + RM*am) / (a1+a2+a3+am); } +double network_throttle::get_current_speed() const { + unsigned int bytes_transferred = 0; + if (m_history.size() == 0 || m_slot_size == 0) + return 0; + + auto it = m_history.begin(); + while (it < m_history.end() - 1) + { + bytes_transferred += it->m_size; + it ++; + } + + return bytes_transferred / ((m_history.size() - 1) * m_slot_size); +} } // namespace } // namespace diff --git a/src/p2p/network_throttle-detail.hpp b/src/p2p/network_throttle-detail.hpp index 9d492c534..063dac850 100644 --- a/src/p2p/network_throttle-detail.hpp +++ b/src/p2p/network_throttle-detail.hpp @@ -54,7 +54,7 @@ class network_throttle : public i_network_throttle { network_speed_kbps m_target_speed; - network_MB m_target_MB; + network_speed_kbps m_real_target_speed; size_t m_network_add_cost; // estimated add cost of headers size_t m_network_minimal_segment; // estimated minimal cost of sending 1 byte to round up to size_t m_network_max_segment; // recommended max size of 1 TCP transmission @@ -80,18 +80,16 @@ class network_throttle : public i_network_throttle { virtual ~network_throttle(); virtual void set_name(const std::string &name); virtual void set_target_speed( network_speed_kbps target ); - virtual void set_target_kill( network_MB target ); + virtual void set_real_target_speed( network_speed_kbps real_target ); // only for throttle_out + virtual network_speed_kbps get_terget_speed(); // add information about events: virtual void handle_trafic_exact(size_t packet_size); ///< count the new traffic/packet; the size is exact considering all network costs virtual void handle_trafic_tcp(size_t packet_size); ///< count the new traffic/packet; the size is as TCP, we will consider MTU etc - virtual void handle_congestion(double overheat); ///< call this when congestion is detected; see example use virtual void tick(); ///< poke and update timers/history (recalculates, moves the history if needed, checks the real clock etc) virtual double get_time_seconds() const ; ///< timer that we use, time in seconds, monotionic - virtual double get_current_overheat() const; ///< did we detected congestion now. NOT USED NOW TODO - virtual void set_overheat(double lag); ///< did we detected congestion now. NOT USED NOW TODO. rename to add_overheat ? // time calculations: virtual void calculate_times(size_t packet_size, calculate_times_struct &cts, bool dbg, double force_window) const; ///< MAIN LOGIC (see base class for info) @@ -101,7 +99,7 @@ class network_throttle : public i_network_throttle { virtual size_t get_recommended_size_of_planned_transport() const; ///< what should be the size (bytes) of next data block to be transported virtual size_t get_recommended_size_of_planned_transport_window(double force_window) const; ///< ditto, but for given windows time frame - //virtual void add_planned_transport(size_t size); + virtual double get_current_speed() const; private: virtual network_time_seconds time_to_slot(network_time_seconds t) const { return std::floor( t ); } // convert exact time eg 13.7 to rounded time for slot number in history 13 diff --git a/src/p2p/network_throttle.hpp b/src/p2p/network_throttle.hpp index dc25a2c45..add4daa86 100644 --- a/src/p2p/network_throttle.hpp +++ b/src/p2p/network_throttle.hpp @@ -100,7 +100,7 @@ struct calculate_times_struct { typedef calculate_times_struct calculate_times_struct; -namespace cryptonote { class cryptonote_protocol_handler_base; }; // a friend class // TODO friend not working +namespace cryptonote { class cryptonote_protocol_handler_base; } // a friend class // TODO friend not working /*** @brief Access to simple throttles, with singlton to access global network limits @@ -146,11 +146,11 @@ class i_network_throttle { public: virtual void set_name(const std::string &name)=0; virtual void set_target_speed( network_speed_kbps target )=0; - virtual void set_target_kill( network_MB target )=0; + virtual void set_real_target_speed(network_speed_kbps real_target)=0; + virtual network_speed_kbps get_terget_speed()=0; virtual void handle_trafic_exact(size_t packet_size) =0; // count the new traffic/packet; the size is exact considering all network costs virtual void handle_trafic_tcp(size_t packet_size) =0; // count the new traffic/packet; the size is as TCP, we will consider MTU etc - virtual void handle_congestion(double overheat) =0; // call this when congestion is detected; see example use virtual void tick() =0; // poke and update timers/history // time calculations: @@ -166,8 +166,6 @@ class i_network_throttle { virtual size_t get_recommended_size_of_planned_transport() const =0; // what should be the recommended limit of data size that we can transport over current network_throttle in near future virtual double get_time_seconds() const =0; // a timer - virtual double get_current_overheat() const =0; - virtual void set_overheat(double lag) =0; virtual void logger_handle_net(const std::string &filename, double time, size_t size)=0; diff --git a/src/serialization/binary_archive.h b/src/serialization/binary_archive.h index 6e00ad664..5cd4988e4 100644 --- a/src/serialization/binary_archive.h +++ b/src/serialization/binary_archive.h @@ -42,8 +42,8 @@ #include "warnings.h" /* I have no clue what these lines means */ -PUSH_WARNINGS; -DISABLE_VS_WARNINGS(4244); +PUSH_WARNINGS +DISABLE_VS_WARNINGS(4244) //TODO: fix size_t warning in x32 platform |