p2pool v2.2.1

16-bit wrap around wasn't processed correctly

CMakeLists.txt

@@ -145,14 +145,7 @@ endif()
 
 add_definitions(/DZMQ_STATIC)
 
-include(CheckSymbolExists)
-
 set(CMAKE_REQUIRED_FLAGS "${GENERAL_FLAGS}")
-check_symbol_exists(pthread_cancel pthread.h HAVE_PTHREAD_CANCEL)
-
-if (HAVE_PTHREAD_CANCEL)
-	add_definitions(/DHAVE_PTHREAD_CANCEL)
-endif()
 
 include(CheckCXXSourceCompiles)
 

README.md

@@ -66,12 +66,12 @@ First you need to find a pool share. This share will stay in PPLNS window for 21
 
 ## Monero version support
 
-Monero will undergo a network upgrade on July 16th, 2022 (block 2,668,888). In order to continue mining after that date, you must update both Monero and P2Pool software to the latest available versions as soon as they are released.
+Monero will undergo a network upgrade on August 13th, 2022 (block 2,688,888). In order to continue mining after that date, you must update both Monero and P2Pool software to the latest available versions as soon as they are released.
 
 |Monero protocol version|Required Monero software version|Required P2Pool version
 |-|-|-|
-|v14 (active until July 16th, 2022)|v0.17.3.0 or newer|v1.0 or newer
-|v15, v16  (active after July 16th, 2022)|v0.18.0.0 or newer|v2.0 or newer
+|v14 (active until August 13th, 2022)|v0.17.3.0 or newer|v1.0 or newer
+|v15, v16 (active after August 13th, 2022)|v0.18.0.0 or newer|v2.2 or newer
 
 ## How to mine on P2Pool
 

src/console_commands.cpp

@@ -27,35 +27,61 @@
 #include "side_chain.h"
 #include <iostream>
 
-#ifdef HAVE_PTHREAD_CANCEL
-#include <pthread.h>
-#endif
-
 static constexpr char log_category_prefix[] = "ConsoleCommands ";
 
 namespace p2pool {
 
-bool ConsoleCommands::stopped = false;
-
 ConsoleCommands::ConsoleCommands(p2pool* pool)
 	: m_pool(pool)
+	, m_loop{}
+	, m_shutdownAsync{}
+	, m_tty{}
+	, m_loopThread{}
+	, m_readBuf{}
+	, m_readBufInUse(false)
 {
-	m_worker = new std::thread(&ConsoleCommands::run, this);
+	if (uv_guess_handle(0) != UV_TTY) {
+		LOGERR(1, "tty is not available");
+		throw std::exception();
+	}
+
+	int err = uv_loop_init(&m_loop);
+	if (err) {
+		LOGERR(1, "failed to create event loop, error " << uv_err_name(err));
+		throw std::exception();
+	}
+
+	err = uv_async_init(&m_loop, &m_shutdownAsync, on_shutdown);
+	if (err) {
+		LOGERR(1, "uv_async_init failed, error " << uv_err_name(err));
+		throw std::exception();
+	}
+	m_shutdownAsync.data = this;
+
+	err = uv_tty_init(&m_loop, &m_tty, 0, 1);
+	if (err) {
+		LOGERR(1, "uv_tty_init failed, error " << uv_err_name(err));
+		throw std::exception();
+	}
+	m_tty.data = this;
+
+	err = uv_read_start(reinterpret_cast<uv_stream_t*>(&m_tty), allocCallback, stdinReadCallback);
+	if (err) {
+		LOGERR(1, "uv_read_start failed, error " << uv_err_name(err));
+		throw std::exception();
+	}
+
+	err = uv_thread_create(&m_loopThread, loop, this);
+	if (err) {
+		LOGERR(1, "failed to start event loop thread, error " << uv_err_name(err));
+		throw std::exception();
+	}
 }
 
 ConsoleCommands::~ConsoleCommands()
 {
-	stopped = true;
-
-#ifdef _WIN32
-	TerminateThread(reinterpret_cast<HANDLE>(m_worker->native_handle()), 0);
-#elif defined HAVE_PTHREAD_CANCEL
-	pthread_cancel(m_worker->native_handle());
-#endif
-
-	m_worker->join();
-	delete m_worker;
-
+	uv_async_send(&m_shutdownAsync);
+	uv_thread_join(&m_loopThread);
 	LOGINFO(1, "stopped");
 }
 
@@ -67,7 +93,7 @@ typedef struct strconst {
 #define STRCONST(x)	{x, sizeof(x)-1}
 #define STRCNULL	{NULL, 0}
 
-typedef int (cmdfunc)(p2pool *pool, const char *args);
+typedef void (cmdfunc)(p2pool *pool, const char *args);
 
 typedef struct cmd {
 	strconst name;
@@ -101,16 +127,15 @@ static cmd cmds[] = {
 	{ STRCNULL, NULL, NULL, NULL }
 };
 
-static int do_help(p2pool * /* m_pool */, const char * /* args */)
+static void do_help(p2pool * /* m_pool */, const char * /* args */)
 {
 	LOGINFO(0, "List of commands");
 	for (int i = 0; cmds[i].name.len; ++i) {
 		LOGINFO(0, cmds[i].name.str << " " << cmds[i].arg << "\t" << cmds[i].descr);
 	}
-	return 0;
 }
 
-static int do_status(p2pool *m_pool, const char * /* args */)
+static void do_status(p2pool *m_pool, const char * /* args */)
 {
 	m_pool->side_chain().print_status();
 	if (m_pool->stratum_server()) {
@@ -123,51 +148,45 @@ static int do_status(p2pool *m_pool, const char * /* args */)
 	m_pool->print_miner_status();
 #endif
 	bkg_jobs_tracker.print_status();
-	return 0;
 }
 
-static int do_loglevel(p2pool * /* m_pool */, const char *args)
+static void do_loglevel(p2pool * /* m_pool */, const char *args)
 {
 	int level = strtol(args, nullptr, 10);
 	level = std::min(std::max(level, 0), log::MAX_GLOBAL_LOG_LEVEL);
 	log::GLOBAL_LOG_LEVEL = level;
 	LOGINFO(0, "log level set to " << level);
-	return 0;
 }
 
-static int do_addpeers(p2pool *m_pool, const char *args)
+static void do_addpeers(p2pool *m_pool, const char *args)
 {
 	if (m_pool->p2p_server()) {
 		m_pool->p2p_server()->connect_to_peers(args);
 	}
-	return 0;
 }
 
-static int do_droppeers(p2pool *m_pool, const char * /* args */)
+static void do_droppeers(p2pool *m_pool, const char * /* args */)
 {
 	if (m_pool->p2p_server()) {
 		m_pool->p2p_server()->drop_connections();
 	}
-	return 0;
 }
 
-static int do_showpeers(p2pool* m_pool, const char* /* args */)
+static void do_showpeers(p2pool* m_pool, const char* /* args */)
 {
 	if (m_pool->p2p_server()) {
 		m_pool->p2p_server()->show_peers();
 	}
-	return 0;
 }
 
-static int do_showworkers(p2pool* m_pool, const char* /* args */)
+static void do_showworkers(p2pool* m_pool, const char* /* args */)
 {
 	if (m_pool->stratum_server()) {
 		m_pool->stratum_server()->show_workers();
 	}
-	return 0;
 }
 
-static int do_showbans(p2pool* m_pool, const char* /* args */)
+static void do_showbans(p2pool* m_pool, const char* /* args */)
 {
 	if (m_pool->stratum_server()) {
 		m_pool->stratum_server()->print_bans();
@@ -175,87 +194,98 @@ static int do_showbans(p2pool* m_pool, const char* /* args */)
 	if (m_pool->p2p_server()) {
 		m_pool->p2p_server()->print_bans();
 	}
-	return 0;
 }
 
-static int do_outpeers(p2pool* m_pool, const char* args)
+static void do_outpeers(p2pool* m_pool, const char* args)
 {
 	if (m_pool->p2p_server()) {
 		m_pool->p2p_server()->set_max_outgoing_peers(strtoul(args, nullptr, 10));
 		LOGINFO(0, "max outgoing peers set to " << m_pool->p2p_server()->max_outgoing_peers());
 	}
-	return 0;
 }
 
-static int do_inpeers(p2pool* m_pool, const char* args)
+static void do_inpeers(p2pool* m_pool, const char* args)
 {
 	if (m_pool->p2p_server()) {
 		m_pool->p2p_server()->set_max_incoming_peers(strtoul(args, nullptr, 10));
 		LOGINFO(0, "max incoming peers set to " << m_pool->p2p_server()->max_incoming_peers());
 	}
-	return 0;
 }
 
 #ifdef WITH_RANDOMX
-static int do_start_mining(p2pool* m_pool, const char* args)
+static void do_start_mining(p2pool* m_pool, const char* args)
 {
 	uint32_t threads = strtoul(args, nullptr, 10);
 	threads = std::min(std::max(threads, 1u), 64u);
 	m_pool->start_mining(threads);
-	return 0;
 }
 
-static int do_stop_mining(p2pool* m_pool, const char* /*args*/)
+static void do_stop_mining(p2pool* m_pool, const char* /*args*/)
 {
 	m_pool->stop_mining();
-	return 0;
 }
 #endif
 
-static int do_exit(p2pool *m_pool, const char * /* args */)
+static void do_exit(p2pool *m_pool, const char * /* args */)
 {
 	bkg_jobs_tracker.wait();
 	m_pool->stop();
-
-	return 1;
 }
 
-void ConsoleCommands::run()
+void ConsoleCommands::allocCallback(uv_handle_t* handle, size_t /*suggested_size*/, uv_buf_t* buf)
 {
-	LOGINFO(1, "started");
+	ConsoleCommands* pThis = static_cast<ConsoleCommands*>(handle->data);
 
-	std::string command;
-	command.reserve(1024);
+	if (pThis->m_readBufInUse) {
+		buf->len = 0;
+		buf->base = nullptr;
+		return;
+	}
 
-	do {
-		std::getline(std::cin, command);
+	buf->len = sizeof(pThis->m_readBuf);
+	buf->base = pThis->m_readBuf;
+	pThis->m_readBufInUse = true;
+}
 
-		if (std::cin.eof()) {
-			LOGINFO(1, "EOF, stopping");
-			return;
-		}
+void ConsoleCommands::stdinReadCallback(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf)
+{
+	ConsoleCommands* pThis = static_cast<ConsoleCommands*>(stream->data);
 
-		if (stopped) {
-			LOGINFO(1, "stopping");
-			return;
-		}
+	if (nread > 0) {
+		for (size_t i = 0; i < static_cast<size_t>(nread); ++i) {
+			if ((buf->base[i] == '\r') || (buf->base[i] == '\n')) {
+				buf->base[i] = '\0';
 
-		cmd* c = cmds;
-		for (; c->name.len; ++c) {
-			if (!strncmp(command.c_str(), c->name.str, c->name.len)) {
-				const char *args = command.c_str() + c->name.len + 1;
-				if (c->func(m_pool, args)) {
-					LOGINFO(1, "exit requested, stopping");
-					return;
+				cmd* c = cmds;
+				for (; c->name.len; ++c) {
+					if (!strncmp(buf->base, c->name.str, c->name.len)) {
+						const char* args = (c->name.len + 1 <= i) ? (buf->base + c->name.len + 1) : "";
+						c->func(pThis->m_pool, args);
+						break;
+					}
+				}
+
+				if (!c->name.len) {
+					LOGWARN(0, "Unknown command " << buf->base);
 				}
 				break;
 			}
 		}
+	}
+	else if (nread < 0) {
+		LOGWARN(4, "read error " << uv_err_name(static_cast<int>(nread)));
+	}
 
-		if (!c->name.len) {
-			LOGWARN(0, "Unknown command " << command);
-		}
-	} while (true);
+	pThis->m_readBufInUse = false;
+}
+
+void ConsoleCommands::loop(void* data)
+{
+	LOGINFO(1, "event loop started");
+	ConsoleCommands* pThis = static_cast<ConsoleCommands*>(data);
+	uv_run(&pThis->m_loop, UV_RUN_DEFAULT);
+	uv_loop_close(&pThis->m_loop);
+	LOGINFO(1, "event loop stopped");
 }
 
 } // namespace p2pool

src/console_commands.h

@@ -17,7 +17,7 @@
 
 #pragma once
 
-#include <thread>
+#include "uv_util.h"
 
 namespace p2pool {
 
@@ -31,10 +31,26 @@ public:
 
 private:
 	p2pool* m_pool;
-	std::thread* m_worker;
 
-	static bool stopped;
-	void run();
+	uv_loop_t m_loop;
+	uv_async_t m_shutdownAsync;
+	uv_tty_t m_tty;
+	uv_thread_t m_loopThread;
+
+	char m_readBuf[64];
+	bool m_readBufInUse;
+
+	static void loop(void* data);
+
+	static void on_shutdown(uv_async_t* async)
+	{
+		ConsoleCommands* pThis = reinterpret_cast<ConsoleCommands*>(async->data);
+		uv_close(reinterpret_cast<uv_handle_t*>(&pThis->m_shutdownAsync), nullptr);
+		uv_close(reinterpret_cast<uv_handle_t*>(&pThis->m_tty), nullptr);
+	}
+
+	static void allocCallback(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
+	static void stdinReadCallback(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf);
 };
 
 } // namespace p2pool

src/json_rpc_request.cpp

@@ -58,6 +58,8 @@ struct CurlContext
 
 	static void on_close(uv_handle_t* h);
 
+	void close_handles();
+
 	uv_poll_t m_pollHandle;
 	curl_socket_t m_socket;
 
@@ -76,6 +78,8 @@ struct CurlContext
 
 	std::vector<char> m_response;
 	std::string m_error;
+
+	curl_slist* m_headers;
 };
 
 CurlContext::CurlContext(const std::string& address, int port, const std::string& req, const std::string& auth, CallbackBase* cb, CallbackBase* close_cb, uv_loop_t* loop)
@@ -90,6 +94,7 @@ CurlContext::CurlContext(const std::string& address, int port, const std::string
 	, m_handle(nullptr)
 	, m_req(req)
 	, m_auth(auth)
+	, m_headers(nullptr)
 {
 	{
 		char buf[log::Stream::BUF_SIZE + 1];
@@ -177,6 +182,11 @@ CurlContext::CurlContext(const std::string& address, int port, const std::string
 	curl_easy_setopt_checked(m_handle, CURLOPT_CONNECTTIMEOUT, 1);
 	curl_easy_setopt_checked(m_handle, CURLOPT_TIMEOUT, 10);
 
+	m_headers = curl_slist_append(m_headers, "Content-Type: application/json");
+	if (m_headers) {
+		curl_easy_setopt_checked(m_handle, CURLOPT_HTTPHEADER, m_headers);
+	}
+
 	if (!m_auth.empty()) {
 		curl_easy_setopt_checked(m_handle, CURLOPT_HTTPAUTH, CURLAUTH_DIGEST | CURLAUTH_ONLY);
 		curl_easy_setopt_checked(m_handle, CURLOPT_USERPWD, m_auth.c_str());
@@ -202,6 +212,8 @@ CurlContext::~CurlContext()
 
 	(*m_closeCallback)(m_error.c_str(), m_error.length());
 	delete m_closeCallback;
+
+	curl_slist_free_all(m_headers);
 }
 
 int CurlContext::on_socket(CURL* /*easy*/, curl_socket_t s, int action)
@@ -235,10 +247,7 @@ int CurlContext::on_socket(CURL* /*easy*/, curl_socket_t s, int action)
 	case CURL_POLL_REMOVE:
 	default:
 		curl_multi_assign(m_multiHandle, s, nullptr);
-		uv_poll_stop(&m_pollHandle);
-		uv_close(reinterpret_cast<uv_handle_t*>(&m_async), on_close);
-		uv_close(reinterpret_cast<uv_handle_t*>(&m_timer), on_close);
-		uv_close(reinterpret_cast<uv_handle_t*>(&m_pollHandle), on_close);
+		close_handles();
 		break;
 	}
 
@@ -266,9 +275,13 @@ void CurlContext::on_timeout(uv_handle_t* req)
 {
 	CurlContext* ctx = reinterpret_cast<CurlContext*>(req->data);
 
-	int running_handles;
+	int running_handles = 0;
 	curl_multi_socket_action(ctx->m_multiHandle, CURL_SOCKET_TIMEOUT, 0, &running_handles);
 	ctx->check_multi_info();
+
+	if (running_handles == 0) {
+		ctx->close_handles();
+	}
 }
 
 size_t CurlContext::on_write(const void* buffer, size_t size, size_t count)
@@ -340,13 +353,29 @@ void CurlContext::on_close(uv_handle_t* h)
 	delete ctx;
 }
 
+void CurlContext::close_handles()
+{
+	if (m_pollHandle.data && !uv_is_closing(reinterpret_cast<uv_handle_t*>(&m_pollHandle))) {
+		uv_poll_stop(&m_pollHandle);
+		uv_close(reinterpret_cast<uv_handle_t*>(&m_pollHandle), on_close);
+	}
+
+	if (m_async.data && !uv_is_closing(reinterpret_cast<uv_handle_t*>(&m_async))) {
+		uv_close(reinterpret_cast<uv_handle_t*>(&m_async), on_close);
+	}
+
+	if (m_timer.data && !uv_is_closing(reinterpret_cast<uv_handle_t*>(&m_timer))) {
+		uv_close(reinterpret_cast<uv_handle_t*>(&m_timer), on_close);
+	}
+}
+
 void Call(const std::string& address, int port, const std::string& req, const std::string& auth, CallbackBase* cb, CallbackBase* close_cb, uv_loop_t* loop)
 {
 	if (!loop) {
 		loop = uv_default_loop();
 	}
 
-	CallOnLoop(loop,
+	const bool result = CallOnLoop(loop,
 		[=]()
 		{
 			try {
@@ -357,6 +386,10 @@ void Call(const std::string& address, int port, const std::string& req, const st
 				(*close_cb)(msg, strlen(msg));
 			}
 		});
+
+	if (!result) {
+		LOGERR(1, "JSON RPC \"" << req << "\" failed");
+	}
 }
 
 } // namespace JSONRPCRequest

src/p2pool.cpp

@@ -156,7 +156,14 @@ p2pool::p2pool(int argc, char* argv[])
 
 	m_blockTemplate = new BlockTemplate(this);
 	m_mempool = new Mempool();
-	m_consoleCommands = new ConsoleCommands(this);
+
+	try {
+		m_consoleCommands = new ConsoleCommands(this);
+	}
+	catch (...) {
+		LOGERR(1, "Couldn't start console commands handler");
+		m_consoleCommands = nullptr;
+	}
 }
 
 p2pool::~p2pool()
@@ -775,8 +782,10 @@ void p2pool::get_info()
 		{
 			if (size > 0) {
 				LOGWARN(1, "get_info RPC request failed: error " << log::const_buf(data, size) << ", trying again in 1 second");
-				std::this_thread::sleep_for(std::chrono::milliseconds(1000));
-				get_info();
+				if (!m_stopped) {
+					std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+					get_info();
+				}
 			}
 		});
 }
@@ -820,6 +829,10 @@ void p2pool::load_found_blocks()
 
 void p2pool::parse_get_info_rpc(const char* data, size_t size)
 {
+	if (m_stopped) {
+		return;
+	}
+
 	rapidjson::Document doc;
 	doc.Parse<rapidjson::kParseCommentsFlag | rapidjson::kParseTrailingCommasFlag>(data, size);
 
@@ -881,14 +894,20 @@ void p2pool::get_version()
 		{
 			if (size > 0) {
 				LOGWARN(1, "get_version RPC request failed: error " << log::const_buf(data, size) << ", trying again in 1 second");
-				std::this_thread::sleep_for(std::chrono::milliseconds(1000));
-				get_version();
+				if (!m_stopped) {
+					std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+					get_version();
+				}
 			}
 		});
 }
 
 void p2pool::parse_get_version_rpc(const char* data, size_t size)
 {
+	if (m_stopped) {
+		return;
+	}
+
 	rapidjson::Document doc;
 	doc.Parse<rapidjson::kParseCommentsFlag | rapidjson::kParseTrailingCommasFlag>(data, size);
 
@@ -945,8 +964,10 @@ void p2pool::get_miner_data()
 		{
 			if (size > 0) {
 				LOGWARN(1, "get_miner_data RPC request failed: error " << log::const_buf(data, size) << ", trying again in 1 second");
-				std::this_thread::sleep_for(std::chrono::milliseconds(1000));
-				get_miner_data();
+				if (!m_stopped) {
+					std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+					get_miner_data();
+				}
 			}
 			else {
 				m_getMinerDataPending = false;
@@ -956,6 +977,10 @@ void p2pool::get_miner_data()
 
 void p2pool::parse_get_miner_data_rpc(const char* data, size_t size)
 {
+	if (m_stopped) {
+		return;
+	}
+
 	hash h;
 	keccak(reinterpret_cast<const uint8_t*>(data), static_cast<int>(size), h.h, HASH_SIZE);
 	if (h == m_getMinerDataHash) {

src/stratum_server.cpp

@@ -614,14 +614,14 @@ void StratumServer::update_auto_diff(StratumClient* client, const uint64_t times
 
 	if (k >= N) {
 		// Full window
-		const uint64_t dt = t2 - t1;
+		const uint16_t dt = t2 - t1;
 		client->m_autoDiff.lo = std::max<uint64_t>((client->m_autoDiffWindowHashes * AUTO_DIFF_TARGET_TIME) / (dt ? dt : 1), MIN_DIFF);
 		client->m_autoDiff.hi = 0;
 	}
 	else if (k >= 10) {
 		// Partial window
 		const uint64_t h0 = hash_uncompress(client->m_autoDiffData[0].m_hashes);
-		const uint64_t dt = client->m_autoDiffData[k].m_timestamp - client->m_autoDiffData[0].m_timestamp;
+		const uint16_t dt = client->m_autoDiffData[k].m_timestamp - client->m_autoDiffData[0].m_timestamp;
 
 		client->m_autoDiff.lo = std::max<uint64_t>(((client->m_autoDiffWindowHashes - h0) * AUTO_DIFF_TARGET_TIME) / (dt ? dt : 1), MIN_DIFF);
 		client->m_autoDiff.hi = 0;

src/tcp_server.h

@@ -83,6 +83,7 @@ public:
 		bool m_isV6;
 		bool m_isIncoming;
 		bool m_readBufInUse;
+		bool m_isClosing;
 		uint32_t m_numRead;
 
 		raw_ip m_addr;

src/tcp_server.inl

@@ -556,6 +556,11 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::send_internal(Client* client, Sen
 		LOGERR(1, "sending data from another thread, this is not thread safe");
 	}
 
+	if (client->m_isClosing) {
+		LOGWARN(5, "client " << static_cast<const char*>(client->m_addrString) << " is being disconnected, can't send any more data");
+		return true;
+	}
+
 	WriteBuf* buf = nullptr;
 
 	{
@@ -845,6 +850,7 @@ TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::Client()
 	, m_isV6(false)
 	, m_isIncoming(false)
 	, m_readBufInUse(false)
+	, m_isClosing(false)
 	, m_numRead(0)
 	, m_addr{}
 	, m_port(0)
@@ -867,6 +873,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::reset()
 	m_isV6 = false;
 	m_isIncoming = false;
 	m_readBufInUse = false;
+	m_isClosing = false;
 	m_numRead = 0;
 	m_addr = {};
 	m_port = -1;
@@ -903,6 +910,11 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_read(uv_stream_t* stre
 	Client* pThis = static_cast<Client*>(stream->data);
 	pThis->m_readBufInUse = false;
 
+	if (pThis->m_isClosing) {
+		LOGWARN(5, "client " << static_cast<const char*>(pThis->m_addrString) << " is being disconnected but data received from it, nread = " << nread << ". Ignoring it.");
+		return;
+	}
+
 	if (nread > 0) {
 		if (pThis->m_owner && !pThis->m_owner->m_finished.load()) {
 			if (!pThis->on_read(buf->base, static_cast<uint32_t>(nread))) {
@@ -944,11 +956,13 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_write(uv_write_t* req,
 template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
 void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::close()
 {
-	if (!m_owner) {
+	if (m_isClosing || !m_owner) {
 		// Already closed
 		return;
 	}
 
+	m_isClosing = true;
+
 	uv_read_stop(reinterpret_cast<uv_stream_t*>(&m_socket));
 
 	uv_tcp_t* s = &m_socket;

src/util.cpp

@@ -32,7 +32,7 @@ namespace p2pool {
 #define STR2(X) STR(X)
 #define STR(X) #X
 
-const char* VERSION = "v2.2 (built"
+const char* VERSION = "v2.2.1 (built"
 #if defined(__clang__)
 	" with clang/" __clang_version__
 #elif defined(__GNUC__)

src/uv_util.h

@@ -137,9 +137,12 @@ struct UV_LoopUserData
 UV_LoopUserData* GetLoopUserData(uv_loop_t* loop, bool create = true);
 
 template<typename T>
-void CallOnLoop(uv_loop_t* loop, T&& callback)
+bool CallOnLoop(uv_loop_t* loop, T&& callback)
 {
 	UV_LoopUserData* data = GetLoopUserData(loop, false);
+	if (!data) {
+		return false;
+	}
 
 	UV_LoopCallbackBase* cb = new UV_LoopCallback<T>(std::move(callback));
 	{
@@ -147,7 +150,27 @@ void CallOnLoop(uv_loop_t* loop, T&& callback)
 		data->m_callbacks.push_back(cb);
 	}
 
-	uv_async_send(data->m_async);
+	if (uv_async_send(data->m_async) == 0) {
+		return true;
+	}
+
+	// Clean up after uv_async_send error
+	bool found = false;
+	{
+		MutexLock lock(data->m_callbacksLock);
+
+		auto it = std::find(data->m_callbacks.begin(), data->m_callbacks.end(), cb);
+		if (it != data->m_callbacks.end()) {
+			found = true;
+			data->m_callbacks.erase(it);
+		}
+	}
+
+	if (found) {
+		delete cb;
+	}
+
+	return false;
 }
 
 } // namespace p2pool