diff --git a/src/wallet/wallet2.cpp b/src/wallet/wallet2.cpp index d4b375464..e74e68153 100644 --- a/src/wallet/wallet2.cpp +++ b/src/wallet/wallet2.cpp @@ -1208,6 +1208,7 @@ wallet2::wallet2(network_type nettype, uint64_t kdf_rounds, bool unattended, std m_http_client(http_client_factory->create()), m_upper_transaction_weight_limit(0), m_run(true), + m_stopped(false), m_callback(0), m_trusted_daemon(false), m_nettype(nettype), @@ -3958,7 +3959,7 @@ bool wallet2::fast_refresh(uint64_t stop_height, uint64_t &blocks_start_height, } size_t current_index = m_blockchain.size(); - while(m_run.load(std::memory_order_relaxed) && current_index < stop_height) + while(refresh_running() && current_index < stop_height) { if (max_pulls > 0 && num_pulls++ >= max_pulls) return false; // pull budget reached, caller may resume on a later call @@ -4123,7 +4124,7 @@ void wallet2::refresh(bool trusted_daemon, uint64_t start_height, uint64_t & blo } // If stop() is called during fast refresh we don't need to continue - if(!m_run.load(std::memory_order_relaxed)) + if(!refresh_running()) return; // always reset start_height to 0 to force short_chain_ history to be used on // subsequent pulls in this refresh. @@ -4146,7 +4147,7 @@ void wallet2::refresh(bool trusted_daemon, uint64_t start_height, uint64_t & blo // infer when we get an incoming output bool first = true, last = false; - while(m_run.load(std::memory_order_relaxed) && blocks_fetched < max_blocks) + while(refresh_running() && blocks_fetched < max_blocks) { uint64_t next_blocks_start_height; std::vector next_blocks; @@ -4287,7 +4288,7 @@ void wallet2::refresh(bool trusted_daemon, uint64_t start_height, uint64_t & blo try { // If stop() is called we don't need to check pending transactions - if (check_pool && m_run.load(std::memory_order_relaxed) && !process_pool_txs.empty()) + if (check_pool && refresh_running() && !process_pool_txs.empty()) process_pool_state(process_pool_txs); } catch (...) diff --git a/src/wallet/wallet2.h b/src/wallet/wallet2.h index 91e076b0a..5c4b414b2 100644 --- a/src/wallet/wallet2.h +++ b/src/wallet/wallet2.h @@ -797,6 +797,7 @@ private: bool set_proxy(const std::string &address); void stop() { m_run.store(false, std::memory_order_relaxed); m_message_store.stop(); } + void shutdown() { m_stopped.store(true, std::memory_order_relaxed); stop(); } i_wallet2_callback* callback() const { return m_callback; } void callback(i_wallet2_callback* callback) { m_callback = callback; } @@ -1533,6 +1534,7 @@ private: void pull_blocks(bool first, bool try_incremental, uint64_t start_height, uint64_t& blocks_start_height, const std::list &short_chain_history, std::vector &blocks, std::vector &o_indices, uint64_t ¤t_height, std::vector>& process_pool_txs); void pull_hashes(uint64_t start_height, uint64_t& blocks_start_height, const std::list &short_chain_history, std::vector &hashes); bool fast_refresh(uint64_t stop_height, uint64_t &blocks_start_height, std::list &short_chain_history, bool force = false, uint64_t max_pulls = 0); + bool refresh_running() const { return m_run.load(std::memory_order_relaxed) && !m_stopped.load(std::memory_order_relaxed); } void pull_and_parse_next_blocks(bool first, bool try_incremental, uint64_t start_height, uint64_t &blocks_start_height, std::list &short_chain_history, const std::vector &prev_blocks, const std::vector &prev_parsed_blocks, std::vector &blocks, std::vector &parsed_blocks, std::vector>& process_pool_txs, bool &last, bool &error, std::exception_ptr &exception); void process_parsed_blocks(const uint64_t start_height, const std::vector &blocks, const std::vector &parsed_blocks, uint64_t& blocks_added, std::map, size_t> *output_tracker_cache = NULL); bool accept_pool_tx_for_processing(const crypto::hash &txid, const std::unordered_set &payments_tx_hashes); @@ -1659,6 +1661,7 @@ private: std::unordered_map m_cold_key_images; std::atomic m_run; + std::atomic m_stopped; mutable boost::recursive_mutex m_daemon_rpc_mutex; diff --git a/src/wallet/wallet_rpc_server.cpp b/src/wallet/wallet_rpc_server.cpp index e8f49a3be..8239aa0c5 100644 --- a/src/wallet/wallet_rpc_server.cpp +++ b/src/wallet/wallet_rpc_server.cpp @@ -35,6 +35,7 @@ #include #include #include +#include #include "version.h" #include "wallet_rpc_server.h" @@ -194,19 +195,24 @@ namespace tools } //------------------------------------------------------------------------------------------------------------------------------ - wallet_rpc_server::wallet_rpc_server():m_wallet(NULL), rpc_login_file(), m_stop(false), m_restricted(false), m_vm(NULL) + wallet_rpc_server::wallet_rpc_server():m_wallet(NULL), rpc_login_file(), m_stop(false), m_teardown(false), m_stop_refresh_active(0), m_wallet_swap_active(false), m_restricted(false), m_vm(NULL) { } //------------------------------------------------------------------------------------------------------------------------------ wallet_rpc_server::~wallet_rpc_server() { - if (m_wallet) - delete m_wallet; + set_wallet(NULL); } //------------------------------------------------------------------------------------------------------------------------------ void wallet_rpc_server::set_wallet(wallet2 *cr) { + // pause stop_refresh() and wait out in-flight calls so they cannot stop a deleted wallet + m_wallet_swap_active = true; + while (m_stop_refresh_active > 0) + std::this_thread::yield(); + delete m_wallet; m_wallet = cr; + m_wallet_swap_active = false; } //------------------------------------------------------------------------------------------------------------------------------ bool wallet_rpc_server::run() @@ -293,8 +299,21 @@ namespace tools return epee::http_server_impl_base::run(1, true); } //------------------------------------------------------------------------------------------------------------------------------ + void wallet_rpc_server::stop_refresh() + { + // may run on a signal/service thread: skip while the wallet is replaced or torn down + ++m_stop_refresh_active; + if (!m_teardown && !m_wallet_swap_active && m_wallet) + m_wallet->shutdown(); + --m_stop_refresh_active; + } + //------------------------------------------------------------------------------------------------------------------------------ void wallet_rpc_server::stop() { + // disarm stop_refresh() and wait out in-flight calls before deleting the wallet under them + m_teardown = true; + while (m_stop_refresh_active > 0) + std::this_thread::yield(); if (m_wallet) { m_wallet->store(); @@ -3665,9 +3684,8 @@ namespace tools handle_rpc_exception(std::current_exception(), er, WALLET_RPC_ERROR_CODE_UNKNOWN_ERROR); return false; } - delete m_wallet; } - m_wallet = wal.release(); + set_wallet(wal.release()); return true; } //------------------------------------------------------------------------------------------------------------------------------ @@ -3742,9 +3760,7 @@ namespace tools return false; } - if (m_wallet) - delete m_wallet; - m_wallet = wal.release(); + set_wallet(wal.release()); return true; } //------------------------------------------------------------------------------------------------------------------------------ @@ -3770,8 +3786,7 @@ namespace tools return false; } } - delete m_wallet; - m_wallet = NULL; + set_wallet(NULL); return true; } //------------------------------------------------------------------------------------------------------------------------------ @@ -4077,9 +4092,7 @@ namespace tools return false; } - if (m_wallet) - delete m_wallet; - m_wallet = wal.release(); + set_wallet(wal.release()); res.address = m_wallet->get_account().get_public_address_str(m_wallet->nettype()); return true; } @@ -4293,9 +4306,7 @@ namespace tools return false; } - if (m_wallet) - delete m_wallet; - m_wallet = wal.release(); + set_wallet(wal.release()); res.address = m_wallet->get_account().get_public_address_str(m_wallet->nettype()); res.info = "Wallet has been restored successfully."; return true; @@ -5045,7 +5056,7 @@ public: tools::signal_handler::install([&wal, &quit](int) { assert(wal); quit = true; - wal->stop(); + wal->shutdown(); }); try @@ -5057,6 +5068,8 @@ public: { LOG_ERROR(tools::wallet_rpc_server::tr("Initial refresh failed: ") << e.what()); } + // swap handlers before the quit check so a late signal cannot permanently stop the wallet the server serves + tools::signal_handler::install([&quit](int) { quit = true; }); // if we ^C during potentially length load/refresh, there's no server loop yet if (quit) { @@ -5077,6 +5090,7 @@ public: bool r = wrpc->init(&vm); CHECK_AND_ASSERT_MES(r, false, tools::wallet_rpc_server::tr("Failed to initialize wallet RPC server")); tools::signal_handler::install([this](int) { + wrpc->stop_refresh(); // a running refresh blocks server exit wrpc->send_stop_signal(); }); LOG_PRINT_L0(tools::wallet_rpc_server::tr("Starting wallet RPC server")); @@ -5106,6 +5120,7 @@ public: void stop() { + wrpc->stop_refresh(); // a running refresh blocks server exit wrpc->send_stop_signal(); } }; diff --git a/src/wallet/wallet_rpc_server.h b/src/wallet/wallet_rpc_server.h index 29b46ae08..5e87d545f 100644 --- a/src/wallet/wallet_rpc_server.h +++ b/src/wallet/wallet_rpc_server.h @@ -59,6 +59,7 @@ namespace tools bool init(const boost::program_options::variables_map *vm); bool run(); + void stop_refresh(); void stop(); void set_wallet(wallet2 *cr); @@ -286,6 +287,9 @@ namespace tools std::string m_wallet_dir; tools::private_file rpc_login_file; std::atomic m_stop; + std::atomic m_teardown; + std::atomic m_stop_refresh_active; + std::atomic m_wallet_swap_active; bool m_restricted; const boost::program_options::variables_map *m_vm; std::atomic m_auto_refresh_period;