diff options
| author | rfree2monero <rfreemonero@op.pl> | 2015-02-12 20:59:39 +0100 |
|---|---|---|
| committer | rfree2monero <rfreemonero@op.pl> | 2015-02-20 22:13:00 +0100 |
| commit | 5ce4256e3d6ff2e1595750e3875865089e20a03b (patch) | |
| tree | 6e26a3ac3285cb3c86c24caa7fa51033d2794085 | |
| parent | 2014 network limit 1.0a +utils +toc -doc -drmonero (diff) | |
| download | monero-5ce4256e3d6ff2e1595750e3875865089e20a03b.tar.xz | |
2014 network limit 1.1 +utils +toc -doc -drmonero
Update of the PR with network limits
works very well for all speeds
(but remember that low download speed can stop upload
because we then slow down downloading of blockchain
requests too)
more debug options
fixed pedantic warnings in our code
should work again on Mac OS X and FreeBSD
fixed warning about size_t
tested on Debian, Ubuntu, Windows(testing now)
TCP options and ToS (QoS) flag
FIXED peer number limit
FIXED some spikes in ingress/download
FIXED problems when other up and down limit
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 |