p2pool v3.3

P2Pool could fail to sync if uncles of depth 3 were present at a certain depth in the chain

.github/workflows/c-cpp.yml

@@ -42,7 +42,10 @@ jobs:
         make -j$(nproc)
 
     - name: Run tests
-      run: cd tests/build && ./p2pool_tests
+      run: |
+        cd tests/build
+        gunzip *.gz
+        ./p2pool_tests
 
     - name: Archive binary
       uses: actions/upload-artifact@v3
@@ -88,7 +91,7 @@ jobs:
         cd external/src/libzmq
         mkdir build
         cd build
-        cmake .. -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF
+        cmake .. -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF -DWITH_DOCS=OFF -DENABLE_DRAFTS=OFF -DWITH_PERF_TOOL=OFF
         make -j$(nproc)
 
     - name: Build p2pool
@@ -107,7 +110,10 @@ jobs:
         make -j$(nproc)
 
     - name: Run tests
-      run: cd tests/build && ./p2pool_tests
+      run: |
+        cd tests/build
+        gunzip *.gz
+        ./p2pool_tests
 
     - name: Archive binary
       uses: actions/upload-artifact@v3
@@ -157,7 +163,7 @@ jobs:
         cd external/src/libzmq
         mkdir build
         cd build
-        cmake .. -DCMAKE_C_COMPILER=aarch64-linux-gnu-gcc -DCMAKE_CXX_COMPILER=aarch64-linux-gnu-g++ -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF
+        cmake .. -DCMAKE_C_COMPILER=aarch64-linux-gnu-gcc -DCMAKE_CXX_COMPILER=aarch64-linux-gnu-g++ -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF -DWITH_DOCS=OFF -DENABLE_DRAFTS=OFF -DWITH_PERF_TOOL=OFF
         make -j$(nproc)
 
     - name: Build p2pool
@@ -214,7 +220,7 @@ jobs:
         cd external/src/libzmq
         mkdir build
         cd build
-        cmake .. -G "Unix Makefiles" -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF -DZMQ_HAVE_IPC=OFF
+        cmake .. -G "Unix Makefiles" -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF -DZMQ_HAVE_IPC=OFF -DWITH_DOCS=OFF -DENABLE_DRAFTS=OFF -DWITH_PERF_TOOL=OFF
         make -j$(nproc)
 
     - name: Build p2pool
@@ -235,6 +241,7 @@ jobs:
     - name: Run tests
       run: |
         cd tests/build
+        gunzip *.gz
         ./p2pool_tests.exe
 
     - name: Archive binary
@@ -282,6 +289,7 @@ jobs:
     - name: Run tests
       run: |
         cd tests/build/Debug
+        7z x *.gz
         ./p2pool_tests.exe
 
     - name: Archive binary
@@ -328,7 +336,7 @@ jobs:
         cd external/src/libzmq
         mkdir build
         cd build
-        cmake .. -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF
+        cmake .. -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF -DWITH_DOCS=OFF -DENABLE_DRAFTS=OFF -DWITH_PERF_TOOL=OFF
         make -j3
 
     - name: Build p2pool
@@ -349,6 +357,7 @@ jobs:
     - name: Run tests
       run: |
         cd tests/build
+        gunzip *.gz
         ./p2pool_tests
 
     - name: Archive binary
@@ -397,7 +406,7 @@ jobs:
           cd ../../libzmq
           mkdir build
           cd build
-          cmake .. -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF
+          cmake .. -DCMAKE_C_FLAGS='-Os' -DCMAKE_CXX_FLAGS='-Os' -DWITH_TLS=OFF -DWITH_LIBSODIUM=OFF -DWITH_LIBBSD=OFF -DBUILD_TESTS=OFF -DWITH_DOCS=OFF -DENABLE_DRAFTS=OFF -DWITH_PERF_TOOL=OFF
           make -j2
           cd ../../../..
           mkdir build
@@ -409,6 +418,7 @@ jobs:
           cd build
           cmake .. -DSTATIC_LIBS=ON
           make -j2
+          gunzip *.gz
           ./p2pool_tests
 
     - name: Archive binary
@@ -452,6 +462,7 @@ jobs:
           mkdir build && cd build
           cmake ..
           make -j2
+          gunzip *.gz
           ./p2pool_tests
 
     - name: Archive binary

.github/workflows/test-sync.yml

@@ -33,6 +33,7 @@ jobs:
         cd build
         mkdir data
         ./p2pool --host p2pmd.xmrvsbeast.com --zmq-port 18084 --wallet 44MnN1f3Eto8DZYUWuE5XZNUtE3vcRzt2j6PzqWpPau34e6Cf4fAxt6X2MBmrm6F9YMEiMNjN6W4Shn4pLcfNAja621jwyg --no-cache --data-api data --local-api --loglevel 6
+        grep 'Synchronization finished successfully' p2pool.log
 
     - name: Archive p2pool.log
       uses: actions/upload-artifact@v3
@@ -69,6 +70,7 @@ jobs:
         cd build
         mkdir data
         ./p2pool --host p2pmd.xmrvsbeast.com --zmq-port 18084 --wallet 44MnN1f3Eto8DZYUWuE5XZNUtE3vcRzt2j6PzqWpPau34e6Cf4fAxt6X2MBmrm6F9YMEiMNjN6W4Shn4pLcfNAja621jwyg --no-cache --data-api data --local-api --loglevel 6
+        grep 'Synchronization finished successfully' p2pool.log
 
     - name: Archive p2pool.log
       uses: actions/upload-artifact@v3
@@ -105,6 +107,7 @@ jobs:
         cd build/Debug
         mkdir data
         ./p2pool.exe --host p2pmd.xmrvsbeast.com --zmq-port 18084 --wallet 44MnN1f3Eto8DZYUWuE5XZNUtE3vcRzt2j6PzqWpPau34e6Cf4fAxt6X2MBmrm6F9YMEiMNjN6W4Shn4pLcfNAja621jwyg --no-cache --data-api data --local-api --loglevel 6
+        findstr /C:"Synchronization finished successfully" p2pool.log
 
     - name: Archive p2pool.log
       uses: actions/upload-artifact@v3

CMakeLists.txt

@@ -56,7 +56,6 @@ set(HEADERS
 	src/side_chain.h
 	src/stratum_server.h
 	src/tcp_server.h
-	src/tcp_server.inl
 	src/util.h
 	src/uv_util.h
 	src/wallet.h
@@ -85,6 +84,7 @@ set(SOURCES
 	src/pow_hash.cpp
 	src/side_chain.cpp
 	src/stratum_server.cpp
+	src/tcp_server.cpp
 	src/util.cpp
 	src/wallet.cpp
 	src/zmq_reader.cpp

README.md

@@ -4,11 +4,13 @@ Decentralized pool for Monero mining.
 
 Pool status and monitoring pages can be found at https://p2pool.io/, https://p2pool.io/mini/ and https://p2pool.observer/, https://mini.p2pool.observer/
 
-# P2Pool hardfork on March 18, 2023
+These are 3rd-party pages. If they are down, it doesn't mean there is a problem with P2Pool itself - it keeps mining always thanks to its decentralized nature.
 
-P2Pool (not Monero!) will hardfork to new consensus rules on March 18th at 21:00 UTC (use [this link](https://dateful.com/convert/utc?t=9pm&d=2023-03-18) to convert to your local time). The hardfork is necessary to improve P2Pool scalability and reduce the impact on Monero blockchain. You'll need to update to **P2Pool v3.0** or newer version before this time.
+# P2Pool hardforked on March 18, 2023
 
-Reddit discussion: https://www.reddit.com/r/MoneroMining/comments/1095730/psa_p2pool_network_upgrade_aka_hardfork_on_march/
+P2Pool (not Monero!) hardforked to new consensus rules on March 18th at 21:00 UTC. The hardfork was necessary to improve P2Pool scalability and reduce the impact on Monero blockchain. You need to update to the latest P2Pool version to continue mining on P2Pool.
+
+Reddit discussions: [original announcement](https://www.reddit.com/r/MoneroMining/comments/1095730/psa_p2pool_network_upgrade_aka_hardfork_on_march/), [final announcement](https://www.reddit.com/r/MoneroMining/comments/11tln6z/psa_p2pool_miners_update_to_the_latest_version/)
 
 ### Build Status
 
@@ -70,7 +72,7 @@ First you need to find a pool share. This share will stay in PPLNS window for 21
 ## Default P2Pool parameters
 
 * Block time: 10 seconds
-* PPLNS window: 2160 blocks (6 hours)
+* PPLNS window: up to 2160 blocks (6 hours), auto adjustable to balance payout sizes and frequency
 * Minimum payout = Monero block reward/2160, ~0.0003 XMR
 
 ## Monero version support
@@ -79,7 +81,7 @@ Monero network upgrade happened on August 13th, 2022 (block 2,688,888). In order
 
 |Monero protocol version|Required Monero software version|Required P2Pool version
 |-|-|-|
-|v15, v16 (active after August 13th, 2022)|v0.18.0.0 or newer|v2.2 or newer
+|v15, v16 (active after August 13th, 2022)|v0.18.0.0 or newer, v0.18.2.2 recommended|v3.0 or newer
 
 ## How to mine on P2Pool
 
@@ -227,6 +229,8 @@ PAUSE
 ```
 
 ## Build instructions
+Only 64-bit builds are supported, in particular ARMv7 or older CPUs are not supported. The reason is that RandomX hashing algorithm is too slow in 32-bit mode, and P2Pool needs to check new blocks very fast to keep up with other nodes.
+
 Please see the relevant instructions for your platform:
 
 ### Ubuntu 20.04

docker-compose/cfg/configure.py

@@ -620,7 +620,7 @@ class MoneroConfigForm(ConfigFormBase):
         self.fast_sync = self.add(
             npyscreen.Checkbox,
             name="Fast Block Sync",
-            value=False,
+            value=True,
             begin_entry_at=self.name_size,
             relx=self.indent,
         )

docker-compose/cfg/defaults

@@ -22,7 +22,7 @@
     "rate_limit_up": "2048",
     "rate_limit_down": "8192",
     "sync_pruned_blocks": false,
-    "fast_sync": false,
+    "fast_sync": true,
     "monero_options": "",
     "public_monero_node": "",
     "monero_node_login": "",

docker-compose/current_config

@@ -24,7 +24,7 @@
     "rate_limit_up": "2048",
     "rate_limit_down": "8192",
     "sync_pruned_blocks": false,
-    "fast_sync": false,
+    "fast_sync": true,
     "monero_options": "",
     "public_monero_node": "",
     "monero_node_login": "",

docker-compose/docker-compose.yml

@@ -115,7 +115,7 @@ services:
 
 
 
-      --fast-block-sync=0
+      --fast-block-sync=1
 
 
 

docker-compose/statistics/app/p2pool_statistics.py

@@ -58,7 +58,7 @@ def render():
             pool_stats = json.loads(reader.read())
         with open("/data/network/stats", "r") as reader:
             network_stats = json.loads(reader.read())
-        with open("/data/local/stats", "r") as reader:
+        with open("/data/local/stratum", "r") as reader:
             local_stats = json.loads(reader.read())
         return render_template(
             "index.html",

docs/COMMAND_LINE.MD

@@ -29,6 +29,7 @@
 --no-upnp            Disable UPnP port forwarding
 --no-igd             An alias for --no-upnp
 --upnp-stratum       Port forward Stratum port (it's not forwarded by default)
+--version            Print p2pool's version and build details
 ```
 
 ### Example command line

docs/SYSTEMD.MD

@@ -0,0 +1,49 @@
+## Setting up P2Pool as a systemd service (Linux)
+
+With this config, P2Pool will run as a service with redirected I/O. It uses `p2pool` user which you should setup beforehand.
+
+You can execute console commands by running `echo "status" > /run/p2pool/p2pool.control`
+
+### p2pool.service
+```
+[Unit]
+Description=p2pool daemon
+After=network-online.target systemd-modules-load.service monerod.service
+Wants=network-online.target systemd-modules-load.service monerod.service
+Requires=p2pool.socket
+
+[Service]
+Type=exec
+Sockets=p2pool.socket
+StandardInput=socket
+StandardOutput=journal
+StandardError=journal
+
+# /usr/local/bin/p2pool.sh simply invokes the p2pool executable with your chosen command-line options. 
+ExecStart=/usr/local/bin/p2pool.sh
+TimeoutStopSec=60
+
+# Run as p2pool:p2pool
+User=p2pool
+Group=p2pool
+
+WorkingDirectory=/var/lib/p2pool
+
+[Install]
+WantedBy=multi-user.target
+```
+
+### p2pool.socket
+```
+# FIFO socket for controlling p2pool daemon
+[Unit]
+Description=Command FIFO for p2pool daemon
+
+[Socket]
+ListenFIFO=/run/p2pool/p2pool.control
+DirectoryMode=0755
+SocketMode=0666
+SocketUser=p2pool
+SocketGroup=p2pool
+RemoveOnStop=true
+```

docs/html/index.html

@@ -85,7 +85,10 @@ async function load_data() {
 	let network_stats = JSON.parse(await network_stats_req.text());
 
 	let pool_hashrate = pool_stats.pool_statistics.hashRate / 1e6;
+	let pplns_duration = pool_stats.pool_statistics.pplnsWeight / pool_stats.pool_statistics.hashRate;
+
 	document.getElementById("pool_hashrate").innerHTML = pool_hashrate.toFixed(3) + ' MH/s';
+	document.getElementById("pplns_duration").innerHTML = elapsedTimeStr(0, pplns_duration, 0) + ' ' + Math.floor(pplns_duration % 60) + 's';
 	document.getElementById("pool_miners").innerHTML = pool_stats.pool_statistics.miners;
 
 	let ts = Math.floor(Date.now() / 1000);
@@ -198,6 +201,7 @@ function toggle_menu(name) {
 <tr><td style="padding-bottom: 1rem;"></td><td></td><td></td></tr>
 <tr><td>Hashrate</td><td><span id="pool_hashrate">N/A</span></td><td><span id="monero_hashrate">N/A</span></td></tr>
 <tr><td>Miners</td><td><span id="pool_miners">N/A</span></td><td></td></tr>
+<tr><td>PPLNS window</td><td><span id="pplns_duration">N/A</span></td><td></td></tr>
 <tr><td>Last block</td><td><span id="pool_last_block">N/A</span></td><td><span id="monero_last_block">N/A</span></td></tr>
 <tr><td>Total blocks</td><td><span id="pool_total_blocks">N/A</span></td><td></td></tr>
 <tr><td>Current effort</td><td><span id="current_effort">N/A</span></td><td></td></tr>

docs/html/mini/index.html

@@ -85,7 +85,10 @@ async function load_data() {
 	let network_stats = JSON.parse(await network_stats_req.text());
 
 	let pool_hashrate = pool_stats.pool_statistics.hashRate / 1e6;
+	let pplns_duration = pool_stats.pool_statistics.pplnsWeight / pool_stats.pool_statistics.hashRate;
+
 	document.getElementById("pool_hashrate").innerHTML = pool_hashrate.toFixed(3) + ' MH/s';
+	document.getElementById("pplns_duration").innerHTML = elapsedTimeStr(0, pplns_duration, 0) + ' ' + Math.floor(pplns_duration % 60) + 's';
 	document.getElementById("pool_miners").innerHTML = pool_stats.pool_statistics.miners;
 
 	let ts = Math.floor(Date.now() / 1000);
@@ -198,6 +201,7 @@ function toggle_menu(name) {
 <tr><td style="padding-bottom: 1rem;"></td><td></td><td></td></tr>
 <tr><td>Hashrate</td><td><span id="pool_hashrate">N/A</span></td><td><span id="monero_hashrate">N/A</span></td></tr>
 <tr><td>Miners</td><td><span id="pool_miners">N/A</span></td><td></td></tr>
+<tr><td>PPLNS window</td><td><span id="pplns_duration">N/A</span></td><td></td></tr>
 <tr><td>Last block</td><td><span id="pool_last_block">N/A</span></td><td><span id="monero_last_block">N/A</span></td></tr>
 <tr><td>Total blocks</td><td><span id="pool_total_blocks">N/A</span></td><td></td></tr>
 <tr><td>Current effort</td><td><span id="current_effort">N/A</span></td><td></td></tr>

src/block_cache.cpp

@@ -183,7 +183,7 @@ void BlockCache::store(const PoolBlock& block)
 	memcpy(data + sizeof(uint32_t) + n1, sidechain_data.data(), n2);
 }
 
-void BlockCache::load_all(SideChain& side_chain, P2PServer& server)
+void BlockCache::load_all(const SideChain& side_chain, P2PServer& server)
 {
 	if (!m_impl->m_data) {
 		return;

src/block_cache.h

@@ -30,7 +30,7 @@ public:
 	~BlockCache();
 
 	void store(const PoolBlock& block);
-	void load_all(SideChain& side_chain, P2PServer& server);
+	void load_all(const SideChain& side_chain, P2PServer& server);
 	void flush();
 
 private:

src/block_template.cpp

@@ -34,9 +34,6 @@
 
 static constexpr char log_category_prefix[] = "BlockTemplate ";
 
-// Max P2P message size (128 KB) minus BLOCK_RESPONSE header (5 bytes)
-static constexpr size_t MAX_BLOCK_TEMPLATE_SIZE = 128 * 1024 - (1 + sizeof(uint32_t));
-
 namespace p2pool {
 
 BlockTemplate::BlockTemplate(SideChain* sidechain, RandomX_Hasher_Base* hasher)
@@ -168,9 +165,7 @@ BlockTemplate& BlockTemplate::operator=(const BlockTemplate& b)
 static FORCEINLINE uint64_t get_base_reward(uint64_t already_generated_coins)
 {
 	const uint64_t result = ~already_generated_coins >> 19;
-
-	constexpr uint64_t min_reward = 600000000000ULL;
-	return (result < min_reward) ? min_reward : result;
+	return (result < BASE_BLOCK_REWARD) ? BASE_BLOCK_REWARD : result;
 }
 
 static FORCEINLINE uint64_t get_block_reward(uint64_t base_reward, uint64_t median_weight, uint64_t fees, uint64_t weight)
@@ -353,14 +348,27 @@ void BlockTemplate::update(const MinerData& data, const Mempool& mempool, const
 		b->m_transactions.resize(1);
 		b->m_outputs.clear();
 
-		// Block template size without coinbase outputs and transactions (add 1+1 more bytes for output and tx count if they go above 128)
-		size_t k = b->serialize_mainchain_data().size() + b->serialize_sidechain_data().size() + 2;
+		// Block template size without coinbase outputs and transactions (minus 2 bytes for output and tx count dummy varints)
+		size_t k = b->serialize_mainchain_data().size() + b->serialize_sidechain_data().size() - 2;
 
-		// a rough estimation of outputs' size
-		// all outputs have <= 5 bytes for each output's reward, and up to 18 outputs can have 6 bytes for output's reward
-		k += m_shares.size() * (5 /* reward */ + 1 /* tx_type */ + HASH_SIZE /* stealth address */ + 1 /* viewtag */) + 18;
+		// Add output and tx count real varints
+		writeVarint(m_shares.size(), [&k](uint8_t) { ++k; });
+		writeVarint(m_mempoolTxs.size(), [&k](uint8_t) { ++k; });
 
-		const size_t max_transactions = (MAX_BLOCK_TEMPLATE_SIZE > k) ? ((MAX_BLOCK_TEMPLATE_SIZE - k) / HASH_SIZE) : 0;
+		// Add a rough upper bound estimation of outputs' size. All outputs have <= 5 bytes for each output's reward (< 0.034359738368 XMR per output)
+		k += m_shares.size() * (5 /* reward */ + 1 /* tx_type */ + HASH_SIZE /* stealth address */ + 1 /* viewtag */);
+
+		// >= 0.034359738368 XMR is required for a 6 byte varint, add 1 byte per each potential 6-byte varint
+		{
+			uint64_t r = BASE_BLOCK_REWARD;
+			for (const auto& tx : m_mempoolTxs) {
+				r += tx.fee;
+			}
+			k += r / 34359738368ULL;
+		}
+
+		const size_t max_transactions = (MAX_BLOCK_SIZE > k) ? ((MAX_BLOCK_SIZE - k) / HASH_SIZE) : 0;
+		LOGINFO(6, max_transactions << " transactions can be taken with current block size limit");
 
 		if (max_transactions == 0) {
 			m_mempoolTxs.clear();
@@ -678,6 +686,7 @@ void BlockTemplate::update(const MinerData& data, const Mempool& mempool, const
 
 	m_fullDataBlob = m_blockTemplateBlob;
 	m_fullDataBlob.insert(m_fullDataBlob.end(), sidechain_data.begin(), sidechain_data.end());
+	LOGINFO(6, "blob size = " << m_fullDataBlob.size());
 
 	m_poolBlockTemplate->m_sidechainId = calc_sidechain_hash(0);
 
@@ -705,9 +714,6 @@ void BlockTemplate::update(const MinerData& data, const Mempool& mempool, const
 		if (result != 0) {
 			LOGERR(1, "pool block blob generation and/or parsing is broken, error " << result);
 		}
-		else {
-			LOGINFO(6, "blob size = " << m_fullDataBlob.size());
-		}
 	}
 
 	memset(m_minerTxKeccakState, 0, sizeof(m_minerTxKeccakState));

src/common.h

@@ -56,7 +56,6 @@
 #include <algorithm>
 #include <atomic>
 #include <chrono>
-#include <iostream>
 #include <random>
 
 #include <signal.h>

src/console_commands.cpp

@@ -25,20 +25,21 @@
 #include "miner.h"
 #endif
 #include "side_chain.h"
-#include <iostream>
+#include "p2pool_api.h"
+#include "params.h"
 
 static constexpr char log_category_prefix[] = "ConsoleCommands ";
 
+static constexpr int DEFAULT_BACKLOG = 1;
+
 namespace p2pool {
 
 ConsoleCommands::ConsoleCommands(p2pool* pool)
-	: m_pool(pool)
-	, m_loop{}
-	, m_shutdownAsync{}
+	: TCPServer(DEFAULT_BACKLOG, ConsoleClient::allocate)
+	, m_pool(pool)
 	, m_tty{}
 	, m_stdin_pipe{}
 	, m_stdin_handle(nullptr)
-	, m_loopThread{}
 	, m_readBuf{}
 	, m_readBufInUse(false)
 {
@@ -49,18 +50,30 @@ ConsoleCommands::ConsoleCommands(p2pool* pool)
 		throw std::exception();
 	}
 
-	int err = uv_loop_init(&m_loop);
-	if (err) {
-		LOGERR(1, "failed to create event loop, error " << uv_err_name(err));
+	std::random_device rd;
+
+	for (int i = 0; i < 10; ++i) {
+		if (start_listening(false, "127.0.0.1", 49152 + (rd() % 16384))) {
+			break;
+		}
+	}
+
+	if (m_listenPort < 0) {
+		LOGERR(1, "failed to listen on TCP port");
 		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();
+	if (m_pool->api() && m_pool->params().m_localStats) {
+		m_pool->api()->set(p2pool_api::Category::LOCAL, "console",
+			[stdin_type, this](log::Stream& s)
+			{
+				s << "{\"mode\":" << ((stdin_type == UV_TTY) ? "\"tty\"" : "\"pipe\"")
+					<< ",\"tcp_port\":" << m_listenPort
+					<< "}";
+			});
 	}
-	m_shutdownAsync.data = this;
+
+	int err;
 
 	if (stdin_type == UV_TTY) {
 		LOGINFO(3, "processing stdin as UV_TTY");
@@ -102,9 +115,12 @@ ConsoleCommands::ConsoleCommands(p2pool* pool)
 
 ConsoleCommands::~ConsoleCommands()
 {
-	uv_async_send(&m_shutdownAsync);
-	uv_thread_join(&m_loopThread);
-	LOGINFO(1, "stopped");
+	shutdown_tcp();
+}
+
+void ConsoleCommands::on_shutdown()
+{
+	uv_close(reinterpret_cast<uv_handle_t*>(m_stdin_handle), nullptr);
 }
 
 typedef struct strconst {
@@ -274,46 +290,7 @@ void ConsoleCommands::stdinReadCallback(uv_stream_t* stream, ssize_t nread, cons
 	ConsoleCommands* pThis = static_cast<ConsoleCommands*>(stream->data);
 
 	if (nread > 0) {
-		std::string& command = pThis->m_command;
-		command.append(buf->base, nread);
-
-		do {
-			size_t k = command.find_first_of("\r\n");
-			if (k == std::string::npos) {
-				break;
-			}
-			command[k] = '\0';
-
-			cmd* c = cmds;
-			for (; c->name.len; ++c) {
-				if (!strncmp(command.c_str(), c->name.str, c->name.len)) {
-					const char* args = (c->name.len + 1 <= k) ? (command.c_str() + c->name.len + 1) : "";
-
-					// Skip spaces
-					while ((args[0] == ' ') || (args[0] == '\t')) {
-						++args;
-					}
-
-					// Check if an argument is required
-					if (strlen(c->arg) && !strlen(args)) {
-						LOGWARN(0, c->name.str << " requires arguments");
-						do_help(nullptr, nullptr);
-						break;
-					}
-
-					c->func(pThis->m_pool, args);
-					break;
-				}
-			}
-
-			if (!c->name.len) {
-				LOGWARN(0, "Unknown command " << command.c_str());
-				do_help(nullptr, nullptr);
-			}
-
-			k = command.find_first_not_of("\r\n", k + 1);
-			command.erase(0, k);
-		} while (true);
+		pThis->process_input(pThis->m_command, buf->base, static_cast<uint32_t>(nread));
 	}
 	else if (nread < 0) {
 		LOGWARN(4, "read error " << uv_err_name(static_cast<int>(nread)));
@@ -322,23 +299,48 @@ void ConsoleCommands::stdinReadCallback(uv_stream_t* stream, ssize_t nread, cons
 	pThis->m_readBufInUse = false;
 }
 
-void ConsoleCommands::loop(void* data)
+
+void ConsoleCommands::process_input(std::string& command, char* data, uint32_t size)
 {
-	LOGINFO(1, "event loop started");
+	command.append(data, size);
 
-	ConsoleCommands* pThis = static_cast<ConsoleCommands*>(data);
+	do {
+		size_t k = command.find_first_of("\r\n");
+		if (k == std::string::npos) {
+			break;
+		}
+		command[k] = '\0';
 
-	int err = uv_run(&pThis->m_loop, UV_RUN_DEFAULT);
-	if (err) {
-		LOGWARN(1, "uv_run returned " << err);
-	}
+		cmd* c = cmds;
+		for (; c->name.len; ++c) {
+			if (!strncmp(command.c_str(), c->name.str, c->name.len)) {
+				const char* args = (c->name.len + 1 <= k) ? (command.c_str() + c->name.len + 1) : "";
 
-	err = uv_loop_close(&pThis->m_loop);
-	if (err) {
-		LOGWARN(1, "uv_loop_close returned error " << uv_err_name(err));
-	}
+				// Skip spaces
+				while ((args[0] == ' ') || (args[0] == '\t')) {
+					++args;
+				}
 
-	LOGINFO(1, "event loop stopped");
+				// Check if an argument is required
+				if (strlen(c->arg) && !strlen(args)) {
+					LOGWARN(0, c->name.str << " requires arguments");
+					do_help(nullptr, nullptr);
+					break;
+				}
+
+				c->func(m_pool, args);
+				break;
+			}
+		}
+
+		if (!c->name.len) {
+			LOGWARN(0, "Unknown command " << command.c_str());
+			do_help(nullptr, nullptr);
+		}
+
+		k = command.find_first_not_of("\r\n", k + 1);
+		command.erase(0, k);
+	} while (true);
 }
 
 } // namespace p2pool

src/console_commands.h

@@ -18,43 +18,55 @@
 #pragma once
 
 #include "uv_util.h"
+#include "tcp_server.h"
 
 namespace p2pool {
 
 class p2pool;
 
-class ConsoleCommands : public nocopy_nomove
+class ConsoleCommands : public TCPServer
 {
 public:
 	explicit ConsoleCommands(p2pool* pool);
 	~ConsoleCommands();
 
+	struct ConsoleClient : public Client
+	{
+		ConsoleClient() : Client(m_consoleReadBuf, sizeof(m_consoleReadBuf)) {}
+		~ConsoleClient() {}
+
+		static Client* allocate() { return new ConsoleClient(); }
+
+		size_t size() const override { return sizeof(ConsoleClient); }
+
+		bool on_connect() override { return true; };
+		bool on_read(char* data, uint32_t size) override { static_cast<ConsoleCommands*>(m_owner)->process_input(m_command, data, size); return true; };
+
+		char m_consoleReadBuf[1024] = {};
+
+		std::string m_command;
+	};
+
+	void on_shutdown() override;
+
 private:
+	const char* get_category() const override { return "ConsoleCommands "; }
+
 	p2pool* m_pool;
 
-	uv_loop_t m_loop;
-	uv_async_t m_shutdownAsync;
 	uv_tty_t m_tty;
 	uv_pipe_t m_stdin_pipe;
 	uv_stream_t* m_stdin_handle;
-	uv_thread_t m_loopThread;
 
-	char m_readBuf[64];
+	char m_readBuf[1024];
 	bool m_readBufInUse;
 
 	std::string m_command;
 
-	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_stdin_handle), 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);
+
+	void process_input(std::string& command, char* data, uint32_t size);
 };
 
 } // namespace p2pool

src/crypto.cpp

@@ -80,7 +80,7 @@ static FORCEINLINE bool less32(const uint8_t* k0, const uint8_t* k1)
 	return false;
 }
 
-// cppcheck-suppress constParameter
+// cppcheck-suppress constParameterReference
 void generate_keys(hash& pub, hash& sec)
 {
 	do {
@@ -356,6 +356,7 @@ private:
 		hash m_derivation;
 		uint32_t m_viewTags1[2] = { 0xFFFFFFFFUL, 0xFFFFFFFFUL };
 		std::vector<uint32_t> m_viewTags2;
+		// cppcheck-suppress unusedStructMember
 		uint64_t m_timestamp;
 
 		FORCEINLINE bool find_view_tag(size_t output_index, uint8_t& view_tag) const

src/json_rpc_request.h

@@ -20,29 +20,17 @@
 namespace p2pool {
 namespace JSONRPCRequest {
 
-struct CallbackBase
-{
-	virtual ~CallbackBase() {}
-	virtual void operator()(const char* data, size_t size) = 0;
-};
-
-template<typename T>
-struct Callback : public CallbackBase
-{
-	explicit FORCEINLINE Callback(T&& cb) : m_cb(std::move(cb)) {}
-	void operator()(const char* data, size_t size) override { m_cb(data, size); }
-
-private:
-	Callback& operator=(Callback&&) = delete;
-	T m_cb;
-};
+typedef Callback<void, const char*, size_t>::Base CallbackBase;
 
 void Call(const std::string& address, int port, const std::string& req, const std::string& auth, const std::string& proxy, CallbackBase* cb, CallbackBase* close_cb, uv_loop_t* loop);
 
 template<typename T, typename U>
 FORCEINLINE void call(const std::string& address, int port, const std::string& req, const std::string& auth, const std::string& proxy, T&& cb, U&& close_cb, uv_loop_t* loop = nullptr)
 {
-	Call(address, port, req, auth, proxy, new Callback<T>(std::move(cb)), new Callback<U>(std::move(close_cb)), loop);
+	typedef Callback<void, const char*, size_t>::Derived<T> CallbackT;
+	typedef Callback<void, const char*, size_t>::Derived<U> CallbackU;
+
+	Call(address, port, req, auth, proxy, new CallbackT(std::move(cb)), new CallbackU(std::move(close_cb)), loop);
 }
 
 } // namespace JSONRPCRequest

src/main.cpp

@@ -58,6 +58,7 @@ void p2pool_usage()
 		"--no-igd             An alias for --no-upnp\n"
 		"--upnp-stratum       Port forward Stratum port (it's not forwarded by default)\n"
 #endif
+		"--version            Print p2pool's version and build details\n"
 		"--help               Show this help message\n\n"
 		"Example command line:\n\n"
 		"%s --host 127.0.0.1 --rpc-port 18081 --zmq-port 18083 --wallet YOUR_WALLET_ADDRESS --stratum 0.0.0.0:%d --p2p 0.0.0.0:%d\n\n",
@@ -75,6 +76,11 @@ void p2pool_usage()
 	);
 }
 
+void p2pool_version()
+{
+	printf("P2Pool %s\n", p2pool::VERSION);
+}
+
 void memory_tracking_start();
 void memory_tracking_stop();
 
@@ -90,6 +96,11 @@ int main(int argc, char* argv[])
 			p2pool_usage();
 			return 0;
 		}
+
+		if (!strcmp(argv[i], "--version") || !strcmp(argv[i], "/version") || !strcmp(argv[i], "-v") || !strcmp(argv[i], "/v")) {
+			p2pool_version();
+			return 0;
+		}
 	}
 
 	memory_tracking_start();

src/p2p_server.cpp

@@ -39,14 +39,10 @@ static const char* seed_nodes_mini[] = { "seeds-mini.p2pool.io", "" };
 static constexpr int DEFAULT_BACKLOG = 16;
 static constexpr uint64_t DEFAULT_BAN_TIME = 600;
 
-static constexpr size_t SEND_BUF_MIN_SIZE = 256;
-
-#include "tcp_server.inl"
-
 namespace p2pool {
 
 P2PServer::P2PServer(p2pool* pool)
-	: TCPServer(P2PClient::allocate)
+	: TCPServer(DEFAULT_BACKLOG, P2PClient::allocate)
 	, m_pool(pool)
 	, m_cache(pool->params().m_blockCache ? new BlockCache() : nullptr)
 	, m_cacheLoaded(false)
@@ -62,7 +58,8 @@ P2PServer::P2PServer(p2pool* pool)
 	, m_lookForMissingBlocks(true)
 	, m_fastestPeer(nullptr)
 {
-	m_blockDeserializeBuf.reserve(131072);
+	m_callbackBuf.resize(P2P_BUF_SIZE);
+	m_blockDeserializeBuf.reserve(MAX_BLOCK_SIZE);
 
 	// Diffuse the initial state in case it has low quality
 	m_rng.discard(10000);
@@ -383,7 +380,7 @@ void P2PServer::send_peer_list_request(P2PClient* client, uint64_t cur_time)
 		{
 			LOGINFO(6, "sending PEER_LIST_REQUEST to " << static_cast<char*>(client->m_addrString));
 
-			if (buf_size < SEND_BUF_MIN_SIZE) {
+			if (buf_size < 1) {
 				return 0;
 			}
 
@@ -594,7 +591,7 @@ void P2PServer::load_peer_list()
 			p.m_numFailedConnections = 0;
 			p.m_lastSeen = seconds_since_epoch();
 
-			if (!already_added && !is_banned(p.m_addr)) {
+			if (!already_added && !is_banned(p.m_isV6, p.m_addr)) {
 				m_peerList.push_back(p);
 			}
 		});
@@ -664,7 +661,7 @@ void P2PServer::load_monerod_peer_list()
 				p.m_port = port;
 				p.m_numFailedConnections = 0;
 
-				if (!is_banned(p.m_addr)) {
+				if (!is_banned(p.m_isV6, p.m_addr)) {
 					m_peerListMonero.push_back(p);
 				}
 			}
@@ -697,7 +694,7 @@ void P2PServer::update_peer_in_list(bool is_v6, const raw_ip& ip, int port)
 		}
 	}
 
-	if (!is_banned(ip)) {
+	if (!is_banned(is_v6, ip)) {
 		m_peerList.emplace_back(Peer{ is_v6, ip, port, 0, cur_time });
 	}
 }
@@ -891,7 +888,7 @@ void P2PServer::on_broadcast()
 					const std::vector<uint8_t>& blob = send_compact ? data->compact_blob : data->pruned_blob;
 
 					const uint32_t len = static_cast<uint32_t>(blob.size());
-					if (buf_size < SEND_BUF_MIN_SIZE + 1 + sizeof(uint32_t) + len) {
+					if (buf_size < 1 + sizeof(uint32_t) + len) {
 						return 0;
 					}
 
@@ -909,7 +906,7 @@ void P2PServer::on_broadcast()
 					LOGINFO(5, "sending BLOCK_BROADCAST (full)   to " << log::Gray() << static_cast<char*>(client->m_addrString));
 
 					const uint32_t len = static_cast<uint32_t>(data->blob.size());
-					if (buf_size < SEND_BUF_MIN_SIZE + 1 + sizeof(uint32_t) + len) {
+					if (buf_size < 1 + sizeof(uint32_t) + len) {
 						return 0;
 					}
 
@@ -1131,7 +1128,7 @@ void P2PServer::download_missing_blocks()
 			{
 				LOGINFO(5, "sending BLOCK_REQUEST for id = " << id << " to " << static_cast<char*>(client->m_addrString));
 
-				if (buf_size < SEND_BUF_MIN_SIZE) {
+				if (buf_size < 1 + HASH_SIZE) {
 					return 0;
 				}
 
@@ -1188,7 +1185,8 @@ void P2PServer::check_block_template()
 }
 
 P2PServer::P2PClient::P2PClient()
-	: m_peerId(0)
+	: Client(m_p2pReadBuf, sizeof(m_p2pReadBuf))
+	, m_peerId(0)
 	, m_connectedTime(0)
 	, m_broadcastMaxHeight(0)
 	, m_expectedMessage(MessageId::HANDSHAKE_CHALLENGE)
@@ -1211,6 +1209,7 @@ P2PServer::P2PClient::P2PClient()
 	, m_lastBlockrequestTimestamp(0)
 	, m_broadcastedHashes{}
 {
+	m_p2pReadBuf[0] = '\0';
 }
 
 void P2PServer::on_shutdown()
@@ -1352,7 +1351,7 @@ bool P2PServer::P2PClient::on_read(char* data, uint32_t size)
 		return false;
 	}
 
-	if ((data != m_readBuf + m_numRead) || (data + size > m_readBuf + sizeof(m_readBuf))) {
+	if ((data != m_readBuf + m_numRead) || (data + size > m_readBuf + m_readBufSize)) {
 		LOGERR(1, "peer " << static_cast<char*>(m_addrString) << " invalid data pointer or size in on_read()");
 		ban(DEFAULT_BAN_TIME);
 		server->remove_peer_from_list(this);
@@ -1609,7 +1608,7 @@ bool P2PServer::P2PClient::send_handshake_challenge()
 		{
 			LOGINFO(5, "sending HANDSHAKE_CHALLENGE to " << static_cast<char*>(m_addrString));
 
-			if (buf_size < SEND_BUF_MIN_SIZE) {
+			if (buf_size < 1 + CHALLENGE_SIZE + sizeof(uint64_t)) {
 				return 0;
 			}
 
@@ -1733,7 +1732,7 @@ void P2PServer::P2PClient::send_handshake_solution(const uint8_t (&challenge)[CH
 				{
 					LOGINFO(5, "sending HANDSHAKE_SOLUTION to " << static_cast<char*>(work->client->m_addrString));
 
-					if (buf_size < SEND_BUF_MIN_SIZE) {
+					if (buf_size < 1 + HASH_SIZE + CHALLENGE_SIZE + 1 + sizeof(int32_t) + 1 + HASH_SIZE) {
 						return 0;
 					}
 
@@ -1811,9 +1810,8 @@ bool P2PServer::P2PClient::check_handshake_solution(const hash& solution, const
 
 bool P2PServer::P2PClient::on_handshake_challenge(const uint8_t* buf)
 {
-	check_event_loop_thread(__func__);
-
 	P2PServer* server = static_cast<P2PServer*>(m_owner);
+	server->check_event_loop_thread(__func__);
 
 	uint8_t challenge[CHALLENGE_SIZE];
 	memcpy(challenge, buf, CHALLENGE_SIZE);
@@ -1881,7 +1879,7 @@ bool P2PServer::P2PClient::on_handshake_solution(const uint8_t* buf)
 			{
 				LOGINFO(5, "sending LISTEN_PORT and BLOCK_REQUEST for the chain tip to " << static_cast<char*>(m_addrString));
 
-				if (buf_size < SEND_BUF_MIN_SIZE) {
+				if (buf_size < 1 + sizeof(int32_t) + 1 + HASH_SIZE) {
 					return 0;
 				}
 
@@ -1952,7 +1950,7 @@ bool P2PServer::P2PClient::on_block_request(const uint8_t* buf)
 
 			const uint32_t len = static_cast<uint32_t>(blob.size());
 
-			if (buf_size < SEND_BUF_MIN_SIZE + 1 + sizeof(uint32_t) + len) {
+			if (buf_size < 1 + sizeof(uint32_t) + len) {
 				return 0;
 			}
 
@@ -2084,9 +2082,9 @@ bool P2PServer::P2PClient::on_block_broadcast(const uint8_t* buf, uint32_t size,
 
 bool P2PServer::P2PClient::on_peer_list_request(const uint8_t*)
 {
-	check_event_loop_thread(__func__);
-
 	P2PServer* server = static_cast<P2PServer*>(m_owner);
+	server->check_event_loop_thread(__func__);
+
 	const uint64_t cur_time = seconds_since_epoch();
 	const bool first = (m_prevIncomingPeerListRequest == 0);
 
@@ -2156,7 +2154,7 @@ bool P2PServer::P2PClient::on_peer_list_request(const uint8_t*)
 		{
 			LOGINFO(6, "sending PEER_LIST_RESPONSE to " << static_cast<char*>(m_addrString));
 
-			if (buf_size < SEND_BUF_MIN_SIZE + 2 + num_selected_peers * 19) {
+			if (buf_size < 2 + static_cast<size_t>(num_selected_peers) * 19) {
 				return 0;
 			}
 
@@ -2239,7 +2237,7 @@ void P2PServer::P2PClient::on_peer_list_response(const uint8_t* buf)
 			}
 		}
 
-		if (!already_added && !server->is_banned(ip)) {
+		if (!already_added && !server->is_banned(is_v6, ip)) {
 			server->m_peerList.emplace_back(Peer{ is_v6, ip, port, 0, cur_time });
 		}
 	}
@@ -2308,11 +2306,12 @@ bool P2PServer::P2PClient::handle_incoming_block_async(const PoolBlock* block, u
 		P2PClient* client;
 		P2PServer* server;
 		uint32_t client_reset_counter;
+		bool client_isV6;
 		raw_ip client_ip;
 		std::vector<hash> missing_blocks;
 	};
 
-	Work* work = new Work{ {}, *block, this, server, m_resetCounter.load(), m_addr, {} };
+	Work* work = new Work{ {}, *block, this, server, m_resetCounter.load(), m_isV6, m_addr, {} };
 	work->req.data = work;
 
 	const int err = uv_queue_work(&server->m_loop, &work->req,
@@ -2320,7 +2319,7 @@ bool P2PServer::P2PClient::handle_incoming_block_async(const PoolBlock* block, u
 		{
 			BACKGROUND_JOB_START(P2PServer::handle_incoming_block_async);
 			Work* work = reinterpret_cast<Work*>(req->data);
-			work->client->handle_incoming_block(work->server->m_pool, work->block, work->client_reset_counter, work->client_ip, work->missing_blocks);
+			work->client->handle_incoming_block(work->server->m_pool, work->block, work->client_reset_counter, work->client_isV6, work->client_ip, work->missing_blocks);
 		},
 		[](uv_work_t* req, int /*status*/)
 		{
@@ -2339,7 +2338,7 @@ bool P2PServer::P2PClient::handle_incoming_block_async(const PoolBlock* block, u
 	return true;
 }
 
-void P2PServer::P2PClient::handle_incoming_block(p2pool* pool, PoolBlock& block, const uint32_t reset_counter, const raw_ip& addr, std::vector<hash>& missing_blocks)
+void P2PServer::P2PClient::handle_incoming_block(p2pool* pool, PoolBlock& block, const uint32_t reset_counter, bool is_v6, const raw_ip& addr, std::vector<hash>& missing_blocks)
 {
 	if (!pool->side_chain().add_external_block(block, missing_blocks)) {
 		// Client sent bad data, disconnect and ban it
@@ -2353,7 +2352,7 @@ void P2PServer::P2PClient::handle_incoming_block(p2pool* pool, PoolBlock& block,
 		}
 
 		P2PServer* server = pool->p2p_server();
-		server->ban(addr, DEFAULT_BAN_TIME);
+		server->ban(is_v6, addr, DEFAULT_BAN_TIME);
 		server->remove_peer_from_list(addr);
 	}
 }
@@ -2396,7 +2395,7 @@ void P2PServer::P2PClient::post_handle_incoming_block(const uint32_t reset_count
 			{
 				LOGINFO(5, "sending BLOCK_REQUEST for id = " << id << " to " << static_cast<char*>(m_addrString));
 
-				if (buf_size < SEND_BUF_MIN_SIZE + 1 + HASH_SIZE) {
+				if (buf_size < 1 + HASH_SIZE) {
 					return 0;
 				}
 

src/p2p_server.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include "tcp_server.h"
+#include "pool_block.h"
 #include <list>
 
 namespace p2pool {
@@ -26,7 +27,10 @@ class p2pool;
 struct PoolBlock;
 class BlockCache;
 
-static constexpr size_t P2P_BUF_SIZE = 128 * 1024;
+// Max block size plus BLOCK_RESPONSE header (5 bytes)
+static constexpr uint64_t P2P_BUF_SIZE = MAX_BLOCK_SIZE + (1 + sizeof(uint32_t));
+static_assert((P2P_BUF_SIZE & (P2P_BUF_SIZE - 1)) == 0, "P2P_BUF_SIZE is not a power of 2, fix MAX_BLOCK_SIZE");
+
 static constexpr size_t PEER_LIST_RESPONSE_MAX_PEERS = 16;
 static constexpr int DEFAULT_P2P_PORT = 37889;
 static constexpr int DEFAULT_P2P_PORT_MINI = 37888;
@@ -36,7 +40,7 @@ static constexpr uint32_t PROTOCOL_VERSION_1_1 = 0x00010001UL;
 
 static constexpr uint32_t SUPPORTED_PROTOCOL_VERSION = PROTOCOL_VERSION_1_1;
 
-class P2PServer : public TCPServer<P2P_BUF_SIZE, P2P_BUF_SIZE>
+class P2PServer : public TCPServer
 {
 public:
 	enum class MessageId {
@@ -107,13 +111,15 @@ public:
 		void on_peer_list_response(const uint8_t* buf);
 
 		bool handle_incoming_block_async(const PoolBlock* block, uint64_t max_time_delta = 0);
-		void handle_incoming_block(p2pool* pool, PoolBlock& block, const uint32_t reset_counter, const raw_ip& addr, std::vector<hash>& missing_blocks);
+		void handle_incoming_block(p2pool* pool, PoolBlock& block, const uint32_t reset_counter, bool is_v6, const raw_ip& addr, std::vector<hash>& missing_blocks);
 		void post_handle_incoming_block(const uint32_t reset_counter, std::vector<hash>& missing_blocks);
 
 		bool is_good() const { return m_handshakeComplete && !m_handshakeInvalid && (m_listenPort >= 0); }
 
 		const char* software_name() const;
 
+		alignas(8) char m_p2pReadBuf[P2P_BUF_SIZE];
+
 		uint64_t m_peerId;
 		uint64_t m_connectedTime;
 		uint64_t m_broadcastMaxHeight;
@@ -167,6 +173,8 @@ public:
 	const PoolBlock* get_block() const { return m_block; }
 
 private:
+	const char* get_category() const override { return "P2PServer "; }
+
 	p2pool* m_pool;
 	BlockCache* m_cache;
 	bool m_cacheLoaded;

src/p2pool_api.cpp

@@ -103,7 +103,7 @@ void p2pool_api::on_stop()
 	uv_close(reinterpret_cast<uv_handle_t*>(&m_dumpToFileAsync), nullptr);
 }
 
-void p2pool_api::dump_to_file_async_internal(Category category, const char* filename, DumpFileCallbackBase&& callback)
+void p2pool_api::dump_to_file_async_internal(Category category, const char* filename, Callback<void, log::Stream&>::Base&& callback)
 {
 	std::vector<char> buf(16384);
 	log::Stream s(buf.data(), buf.size());

src/p2pool_api.h

@@ -38,7 +38,7 @@ public:
 	void on_stop();
 
 	template<typename T>
-	void set(Category category, const char* filename, T&& callback) { dump_to_file_async_internal(category, filename, DumpFileCallback<T>(std::move(callback))); }
+	void set(Category category, const char* filename, T&& callback) { dump_to_file_async_internal(category, filename, Callback<void, log::Stream&>::Derived<T>(std::move(callback))); }
 
 private:
 	void create_dir(const std::string& path);
@@ -54,25 +54,7 @@ private:
 		std::vector<char> buf;
 	};
 
-	struct DumpFileCallbackBase
-	{
-		virtual ~DumpFileCallbackBase() {}
-		virtual void operator()(log::Stream&) = 0;
-	};
-
-	template<typename T>
-	struct DumpFileCallback : public DumpFileCallbackBase
-	{
-		explicit FORCEINLINE DumpFileCallback(T&& callback) : m_callback(std::move(callback)) {}
-		void operator()(log::Stream& s) override { m_callback(s); }
-
-	private:
-		DumpFileCallback& operator=(DumpFileCallback&&) = delete;
-
-		T m_callback;
-	};
-
-	void dump_to_file_async_internal(Category category, const char* filename, DumpFileCallbackBase&& callback);
+	void dump_to_file_async_internal(Category category, const char* filename, Callback<void, log::Stream&>::Base&& callback);
 	void dump_to_file();
 	static void on_fs_open(uv_fs_t* req);
 	static void on_fs_write(uv_fs_t* req);

src/params.cpp

@@ -24,7 +24,7 @@ void p2pool_usage();
 
 namespace p2pool {
 
-Params::Params(int argc, char* argv[])
+Params::Params(int argc, char* const argv[])
 {
 	for (int i = 1; i < argc; ++i) {
 		bool ok = false;

src/params.h

@@ -23,7 +23,7 @@ namespace p2pool {
 
 struct Params
 {
-	Params(int argc, char* argv[]);
+	Params(int argc, char* const argv[]);
 
 	bool valid() const;
 

src/pool_block.h

@@ -47,6 +47,12 @@ class SideChain;
 * Pool block's PoW hash is calculated from the Monero block template part using Monero's consensus rules
 */
 
+// 128 KB minus BLOCK_RESPONSE P2P protocol header (5 bytes)
+static constexpr uint64_t MAX_BLOCK_SIZE = 128 * 1024 - 5;
+
+// 0.6 XMR
+static constexpr uint64_t BASE_BLOCK_REWARD = 600000000000ULL;
+
 struct DifficultyData
 {
 	FORCEINLINE DifficultyData(uint64_t t, const difficulty_type& d) : m_timestamp(t), m_cumulativeDifficulty(d) {}

src/pool_block_parser.inl

@@ -27,7 +27,7 @@ int PoolBlock::deserialize(const uint8_t* data, size_t size, const SideChain& si
 {
 	try {
 		// Sanity check
-		if (!data || (size > 128 * 1024)) {
+		if (!data || (size > MAX_BLOCK_SIZE)) {
 			return __LINE__;
 		}
 
@@ -139,7 +139,7 @@ int PoolBlock::deserialize(const uint8_t* data, size_t size, const SideChain& si
 			READ_VARINT(tmp);
 
 			// Sanity check
-			if ((tmp == 0) || (tmp > 128 * 1024)) {
+			if ((tmp == 0) || (tmp > MAX_BLOCK_SIZE)) {
 				return __LINE__;
 			}
 
@@ -148,7 +148,7 @@ int PoolBlock::deserialize(const uint8_t* data, size_t size, const SideChain& si
 
 		// Technically some p2pool node could keep stuffing block with transactions until reward is less than 0.6 XMR
 		// But default transaction picking algorithm never does that. It's better to just ban such nodes
-		if (total_reward < 600000000000ULL) {
+		if (total_reward < BASE_BLOCK_REWARD) {
 			return __LINE__;
 		}
 
@@ -341,6 +341,11 @@ int PoolBlock::deserialize(const uint8_t* data, size_t size, const SideChain& si
 		hash check;
 		const std::vector<uint8_t>& consensus_id = sidechain.consensus_id();
 		const int data_size = static_cast<int>((data_end - data_begin) + outputs_blob_size_diff + transactions_blob_size_diff);
+
+		if (data_size > static_cast<int>(MAX_BLOCK_SIZE)) {
+			return __LINE__;
+		}
+
 		keccak_custom(
 			[nonce_offset, extra_nonce_offset, sidechain_hash_offset, data_begin, data_size, &consensus_id, &outputs_blob, outputs_blob_size_diff, outputs_offset, outputs_blob_size, transactions_blob, transactions_blob_size_diff, transactions_offset, transactions_blob_size](int offset) -> uint8_t
 			{

src/side_chain.cpp

@@ -197,6 +197,7 @@ SideChain::SideChain(p2pool* pool, NetworkType type, const char* pool_name)
 	}
 
 	m_uniquePrecalcInputs = new unordered_set<size_t>();
+	m_uniquePrecalcInputs->reserve(1 << 18);
 }
 
 SideChain::~SideChain()
@@ -1122,6 +1123,8 @@ uint64_t SideChain::bottom_height(const PoolBlock* tip) const
 	uint64_t bottom_height;
 	std::vector<MinerShare> shares;
 
+	ReadLock lock(m_sidechainLock);
+
 	if (!get_shares(tip, shares, &bottom_height, true)) {
 		return 0;
 	}
@@ -1367,10 +1370,10 @@ void SideChain::verify(PoolBlock* block)
 	// Deep block
 	//
 	// Blocks in PPLNS window (m_chainWindowSize) require up to m_chainWindowSize earlier blocks to verify
-	// If a block is deeper than m_chainWindowSize * 2 - 1 it can't influence blocks in PPLNS window
+	// If a block is deeper than (m_chainWindowSize - 1) * 2 + UNCLE_BLOCK_DEPTH it can't influence blocks in PPLNS window
 	// Also, having so many blocks on top of this one means it was verified by the network at some point
 	// We skip checks in this case to make pruning possible
-	if (block->m_depth >= m_chainWindowSize * 2) {
+	if (block->m_depth > (m_chainWindowSize - 1) * 2 + UNCLE_BLOCK_DEPTH) {
 		LOGINFO(4, "block " << block->m_sidechainId << " skipped verification");
 		block->m_verified = true;
 		block->m_invalid = false;
@@ -1924,7 +1927,7 @@ void SideChain::update_depths(PoolBlock* block)
 		blocks_to_update.pop_back();
 
 		// Verify this block and possibly other blocks on top of it when we're sure it will get verified
-		if (!block->m_verified && ((block->m_depth >= m_chainWindowSize * 2) || (block->m_sidechainHeight == 0))) {
+		if (!block->m_verified && ((block->m_depth > (m_chainWindowSize - 1) * 2 + UNCLE_BLOCK_DEPTH) || (block->m_sidechainHeight == 0))) {
 			verify_loop(block);
 		}
 
@@ -2156,7 +2159,7 @@ void SideChain::launch_precalc(const PoolBlock* block)
 		return;
 	}
 
-	for (int h = UNCLE_BLOCK_DEPTH - 1; h >= 0; --h) {
+	for (int h = UNCLE_BLOCK_DEPTH; h >= 0; --h) {
 		auto it = m_blocksByHeight.find(block->m_sidechainHeight + m_chainWindowSize + h - 1);
 		if (it == m_blocksByHeight.end()) {
 			continue;
@@ -2183,6 +2186,7 @@ void SideChain::precalc_worker()
 {
 	do {
 		PrecalcJob* job;
+		size_t num_inputs;
 		{
 			MutexLock lock(m_precalcJobsMutex);
 
@@ -2205,22 +2209,29 @@ void SideChain::precalc_worker()
 			uint8_t t[HASH_SIZE * 2 + sizeof(size_t)];
 			memcpy(t, job->b->m_txkeySec.h, HASH_SIZE);
 
-			for (size_t i = 0, n = job->shares.size(); i < n; ++i) {
+			const size_t n = job->shares.size();
+			num_inputs = n;
+
+			for (size_t i = 0; i < n; ++i) {
 				memcpy(t + HASH_SIZE, job->shares[i].m_wallet->view_public_key().h, HASH_SIZE);
 				memcpy(t + HASH_SIZE * 2, &i, sizeof(i));
 				if (!m_uniquePrecalcInputs->insert(robin_hood::hash_bytes(t, array_size(t))).second) {
 					job->shares[i].m_wallet = nullptr;
+					--num_inputs;
 				}
 			}
 		}
 
-		for (size_t i = 0, n = job->shares.size(); i < n; ++i) {
-			if (job->shares[i].m_wallet) {
-				hash eph_public_key;
-				uint8_t view_tag;
-				job->shares[i].m_wallet->get_eph_public_key(job->b->m_txkeySec, i, eph_public_key, view_tag);
+		if (num_inputs) {
+			for (size_t i = 0, n = job->shares.size(); i < n; ++i) {
+				if (job->shares[i].m_wallet) {
+					hash eph_public_key;
+					uint8_t view_tag;
+					job->shares[i].m_wallet->get_eph_public_key(job->b->m_txkeySec, i, eph_public_key, view_tag);
+				}
 			}
 		}
+
 		delete job;
 	} while (true);
 }

src/stratum_server.cpp

@@ -37,12 +37,10 @@ static constexpr int32_t BAD_SHARE_POINTS = -5;
 static constexpr int32_t GOOD_SHARE_POINTS = 1;
 static constexpr int32_t BAN_THRESHOLD_POINTS = -15;
 
-#include "tcp_server.inl"
-
 namespace p2pool {
 
 StratumServer::StratumServer(p2pool* pool)
-	: TCPServer(StratumClient::allocate)
+	: TCPServer(DEFAULT_BACKLOG, StratumClient::allocate)
 	, m_pool(pool)
 	, m_autoDiff(pool->params().m_autoDiff)
 	, m_rng(RandomDeviceSeed::instance)
@@ -57,6 +55,8 @@ StratumServer::StratumServer(p2pool* pool)
 	, m_totalFailedShares(0)
 	, m_apiLastUpdateTime(0)
 {
+	m_callbackBuf.resize(STRATUM_BUF_SIZE);
+
 	// Diffuse the initial state in case it has low quality
 	m_rng.discard(10000);
 
@@ -107,6 +107,7 @@ void StratumServer::on_block(const BlockTemplate& block)
 	const uint32_t num_connections = m_numConnections;
 	if (num_connections == 0) {
 		LOGINFO(4, "no clients connected");
+		api_update_local_stats(seconds_since_epoch());
 		return;
 	}
 
@@ -186,6 +187,8 @@ void StratumServer::on_block(const BlockTemplate& block)
 			delete blobs_data;
 		}
 	}
+
+	api_update_local_stats(seconds_since_epoch());
 }
 
 template<size_t N>
@@ -407,6 +410,7 @@ bool StratumServer::on_submit(StratumClient* client, uint32_t id, const char* jo
 		share->m_req.data = share;
 		share->m_server = this;
 		share->m_client = client;
+		share->m_clientIPv6 = client->m_isV6;
 		share->m_clientAddr = client->m_addr;
 		share->m_clientResetCounter = client->m_resetCounter.load();
 		share->m_rpcId = client->m_rpcId;
@@ -840,7 +844,7 @@ void StratumServer::on_share_found(uv_work_t* req)
 	SubmittedShare* share = reinterpret_cast<SubmittedShare*>(req->data);
 	StratumServer* server = share->m_server;
 
-	if (server->is_banned(share->m_clientAddr)) {
+	if (server->is_banned(share->m_clientIPv6, share->m_clientAddr)) {
 		share->m_highEnoughDifficulty = false;
 		share->m_result = SubmittedShare::Result::BANNED;
 		return;
@@ -1016,7 +1020,7 @@ void StratumServer::on_after_share_found(uv_work_t* req, int /*status*/)
 		}
 	}
 	else if (bad_share) {
-		server->ban(share->m_clientAddr, DEFAULT_BAN_TIME);
+		server->ban(share->m_clientIPv6, share->m_clientAddr, DEFAULT_BAN_TIME);
 	}
 }
 
@@ -1027,7 +1031,8 @@ void StratumServer::on_shutdown()
 }
 
 StratumServer::StratumClient::StratumClient()
-	: m_rpcId(0)
+	: Client(m_stratumReadBuf, sizeof(m_stratumReadBuf))
+	, m_rpcId(0)
 	, m_perConnectionJobId(0)
 	, m_connectedTime(0)
 	, m_jobs{}
@@ -1040,6 +1045,7 @@ StratumServer::StratumClient::StratumClient()
 	, m_lastJobTarget(0)
 	, m_score(0)
 {
+	m_stratumReadBuf[0] = '\0';
 }
 
 void StratumServer::StratumClient::reset()
@@ -1072,7 +1078,7 @@ bool StratumServer::StratumClient::on_connect()
 
 bool StratumServer::StratumClient::on_read(char* data, uint32_t size)
 {
-	if ((data != m_readBuf + m_numRead) || (data + size > m_readBuf + sizeof(m_readBuf))) {
+	if ((data != m_readBuf + m_numRead) || (data + size > m_readBuf + m_readBufSize)) {
 		LOGERR(1, "client: invalid data pointer or size in on_read()");
 		ban(DEFAULT_BAN_TIME);
 		return false;
@@ -1259,9 +1265,9 @@ void StratumServer::api_update_local_stats(uint64_t timestamp)
 		return;
 	}
 
-	// Rate limit to no more than once in 60 seconds.
+	// Rate limit to no more than once in 20 seconds.
 	uint64_t t = m_apiLastUpdateTime.load();
-	if (timestamp < t + 60) {
+	if (timestamp < t + 20) {
 		return;
 	}
 
@@ -1316,8 +1322,10 @@ void StratumServer::api_update_local_stats(uint64_t timestamp)
 	uint32_t connections = m_numConnections;
 	uint32_t incoming_connections = m_numIncomingConnections;
 
+	const double block_reward_share_percent = m_pool->side_chain().get_reward_share(m_pool->params().m_wallet) * 100.0;
+
 	m_pool->api()->set(p2pool_api::Category::LOCAL, "stratum",
-		[hashrate_15m, hashrate_1h, hashrate_24h, total_hashes, shares_found, shares_failed, average_effort, current_effort, connections, incoming_connections](log::Stream& s)
+		[hashrate_15m, hashrate_1h, hashrate_24h, total_hashes, shares_found, shares_failed, average_effort, current_effort, connections, incoming_connections, block_reward_share_percent](log::Stream& s)
 		{
 			s << "{\"hashrate_15m\":" << hashrate_15m
 				<< ",\"hashrate_1h\":" << hashrate_1h
@@ -1329,6 +1337,7 @@ void StratumServer::api_update_local_stats(uint64_t timestamp)
 				<< ",\"current_effort\":" << current_effort
 				<< ",\"connections\":" << connections
 				<< ",\"incoming_connections\":" << incoming_connections
+				<< ",\"block_reward_share_percent\":" << block_reward_share_percent
 				<< "}";
 		});
 }

src/stratum_server.h

@@ -28,7 +28,7 @@ class BlockTemplate;
 static constexpr size_t STRATUM_BUF_SIZE = log::Stream::BUF_SIZE + 1;
 static constexpr int DEFAULT_STRATUM_PORT = 3333;
 
-class StratumServer : public TCPServer<STRATUM_BUF_SIZE, STRATUM_BUF_SIZE>
+class StratumServer : public TCPServer
 {
 public:
 	explicit StratumServer(p2pool *pool);
@@ -52,6 +52,8 @@ public:
 		bool process_login(rapidjson::Document& doc, uint32_t id);
 		bool process_submit(rapidjson::Document& doc, uint32_t id);
 
+		alignas(8) char m_stratumReadBuf[STRATUM_BUF_SIZE];
+
 		uint32_t m_rpcId;
 		uint32_t m_perConnectionJobId;
 		uint64_t m_connectedTime;
@@ -95,6 +97,8 @@ public:
 	void reset_share_counters();
 
 private:
+	const char* get_category() const override { return "StratumServer "; }
+
 	void print_stratum_status() const;
 	void update_auto_diff(StratumClient* client, const uint64_t timestamp, const uint64_t hashes);
 
@@ -138,6 +142,7 @@ private:
 		uv_work_t m_req;
 		StratumServer* m_server;
 		StratumClient* m_client;
+		bool m_clientIPv6;
 		raw_ip m_clientAddr;
 		uint32_t m_clientResetCounter;
 		uint32_t m_rpcId;

src/tcp_server.cpp

@@ -15,15 +15,17 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <thread>
+#include "common.h"
+#include "tcp_server.h"
 
-static thread_local bool server_event_loop_thread = false;
+static thread_local void* server_event_loop_thread = nullptr;
+static thread_local const char* log_category_prefix = "TCPServer ";
 
 namespace p2pool {
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::TCPServer(allocate_client_callback allocate_new_client)
+TCPServer::TCPServer(int default_backlog, allocate_client_callback allocate_new_client)
 	: m_allocateNewClient(allocate_new_client)
+	, m_defaultBacklog(default_backlog)
 	, m_loopThread{}
 #ifdef WITH_UPNP
 	, m_portMapping(0)
@@ -71,9 +73,7 @@ TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::TCPServer(allocate_client_callback all
 	m_connectedClientsList->m_prev = m_connectedClientsList;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-// cppcheck-suppress functionStatic
-TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::~TCPServer()
+TCPServer::~TCPServer()
 {
 	if (m_finished.load() == 0) {
 		LOGERR(1, "TCP wasn't shutdown properly");
@@ -83,10 +83,7 @@ TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::~TCPServer()
 	delete m_connectedClientsList;
 }
 
-
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-template<typename T>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::parse_address_list(const std::string& address_list, T callback)
+void TCPServer::parse_address_list_internal(const std::string& address_list, Callback<void, bool, const std::string&, const std::string&, int>::Base&& callback)
 {
 	if (address_list.empty()) {
 		return;
@@ -120,7 +117,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::parse_address_list(const std::str
 				callback(is_v6, address, ip, port);
 			}
 			else {
-				LOGWARN(1, "invalid IP:port " << address);
+				error_invalid_ip(address);
 			}
 		}
 
@@ -130,8 +127,102 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::parse_address_list(const std::str
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::start_listening(const std::string& listen_addresses, bool upnp)
+bool TCPServer::start_listening(bool is_v6, const std::string& ip, int port, std::string address)
+{
+	if ((m_listenPort >= 0) && (m_listenPort != port)) {
+		LOGERR(1, "all sockets must be listening on the same port number, fix the command line");
+		return false;
+	}
+
+	uv_tcp_t* socket = new uv_tcp_t();
+
+	int err = uv_tcp_init(&m_loop, socket);
+	if (err) {
+		LOGERR(1, "failed to create tcp server handle, error " << uv_err_name(err));
+		delete socket;
+		return false;
+	}
+	socket->data = this;
+
+	ON_SCOPE_LEAVE([is_v6, this, socket]()
+	{
+		const std::vector<uv_tcp_t*>& v = is_v6 ? m_listenSockets6 : m_listenSockets;
+		if (std::find(v.begin(), v.end(), socket) == v.end()) {
+			uv_close(reinterpret_cast<uv_handle_t*>(socket), [](uv_handle_t* h) { delete reinterpret_cast<uv_tcp_t*>(h); });
+		}
+	});
+
+	err = uv_tcp_nodelay(socket, 1);
+	if (err) {
+		LOGERR(1, "failed to set tcp_nodelay on tcp server handle, error " << uv_err_name(err));
+		return false;
+	}
+
+	if (is_v6) {
+		if (address.empty()) {
+			char buf[64] = {};
+			log::Stream s(buf);
+			s << '[' << ip << "]:" << port;
+			address = buf;
+		}
+
+		sockaddr_in6 addr6;
+		err = uv_ip6_addr(ip.c_str(), port, &addr6);
+		if (err) {
+			LOGERR(1, "failed to parse IPv6 address " << ip << ", error " << uv_err_name(err));
+			return false;
+		}
+
+		err = uv_tcp_bind(socket, reinterpret_cast<sockaddr*>(&addr6), UV_TCP_IPV6ONLY);
+		if (err) {
+			LOGERR(1, "failed to bind tcp server IPv6 socket " << address << ", error " << uv_err_name(err));
+			return false;
+		}
+	}
+	else {
+		if (address.empty()) {
+			char buf[64] = {};
+			log::Stream s(buf);
+			s << ip << ':' << port;
+			address = buf;
+		}
+
+		sockaddr_in addr;
+		err = uv_ip4_addr(ip.c_str(), port, &addr);
+		if (err) {
+			LOGERR(1, "failed to parse IPv4 address " << ip << ", error " << uv_err_name(err));
+			return false;
+		}
+
+		err = uv_tcp_bind(socket, reinterpret_cast<sockaddr*>(&addr), 0);
+		if (err) {
+			LOGERR(1, "failed to bind tcp server IPv4 socket " << address << ", error " << uv_err_name(err));
+			return false;
+		}
+	}
+
+	err = uv_listen(reinterpret_cast<uv_stream_t*>(socket), m_defaultBacklog, on_new_connection);
+	if (err) {
+		LOGERR(1, "failed to listen on tcp server socket " << address << ", error " << uv_err_name(err));
+		return false;
+	}
+
+	if (is_v6) {
+		m_listenSockets6.push_back(socket);
+	}
+	else {
+		m_listenSockets.push_back(socket);
+	}
+
+	if (m_listenPort < 0) {
+		m_listenPort = port;
+	}
+
+	LOGINFO(1, "listening on " << log::Gray() << address);
+	return true;
+}
+
+void TCPServer::start_listening(const std::string& listen_addresses, bool upnp)
 {
 	if (listen_addresses.empty()) {
 		LOGERR(1, "listen address not set");
@@ -139,74 +230,11 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::start_listening(const std::string
 	}
 
 	parse_address_list(listen_addresses,
-		[this](bool is_v6, const std::string& address, const std::string& ip, int port)
+		[this](bool is_v6, const std::string& /*address*/, const std::string& ip, int port)
 		{
-			if (m_listenPort < 0) {
-				m_listenPort = port;
-			}
-			else if (m_listenPort != port) {
-				LOGERR(1, "all sockets must be listening on the same port number, fix the command line");
+			if (!start_listening(is_v6, ip, port)) {
 				PANIC_STOP();
 			}
-
-			uv_tcp_t* socket = new uv_tcp_t();
-
-			if (is_v6) {
-				m_listenSockets6.push_back(socket);
-			}
-			else {
-				m_listenSockets.push_back(socket);
-			}
-
-			int err = uv_tcp_init(&m_loop, socket);
-			if (err) {
-				LOGERR(1, "failed to create tcp server handle, error " << uv_err_name(err));
-				PANIC_STOP();
-			}
-			socket->data = this;
-
-			err = uv_tcp_nodelay(socket, 1);
-			if (err) {
-				LOGERR(1, "failed to set tcp_nodelay on tcp server handle, error " << uv_err_name(err));
-				PANIC_STOP();
-			}
-
-			if (is_v6) {
-				sockaddr_in6 addr6;
-				err = uv_ip6_addr(ip.c_str(), port, &addr6);
-				if (err) {
-					LOGERR(1, "failed to parse IPv6 address " << ip << ", error " << uv_err_name(err));
-					PANIC_STOP();
-				}
-
-				err = uv_tcp_bind(socket, reinterpret_cast<sockaddr*>(&addr6), UV_TCP_IPV6ONLY);
-				if (err) {
-					LOGERR(1, "failed to bind tcp server IPv6 socket " << address << ", error " << uv_err_name(err));
-					PANIC_STOP();
-				}
-			}
-			else {
-				sockaddr_in addr;
-				err = uv_ip4_addr(ip.c_str(), port, &addr);
-				if (err) {
-					LOGERR(1, "failed to parse IPv4 address " << ip << ", error " << uv_err_name(err));
-					PANIC_STOP();
-				}
-
-				err = uv_tcp_bind(socket, reinterpret_cast<sockaddr*>(&addr), 0);
-				if (err) {
-					LOGERR(1, "failed to bind tcp server IPv4 socket " << address << ", error " << uv_err_name(err));
-					PANIC_STOP();
-				}
-			}
-
-			err = uv_listen(reinterpret_cast<uv_stream_t*>(socket), DEFAULT_BACKLOG, on_new_connection);
-			if (err) {
-				LOGERR(1, "failed to listen on tcp server socket " << address << ", error " << uv_err_name(err));
-				PANIC_STOP();
-			}
-
-			LOGINFO(1, "listening on " << log::Gray() << address);
 		});
 
 #ifdef WITH_UPNP
@@ -224,8 +252,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::start_listening(const std::string
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(bool is_v6, const char* ip, int port)
+bool TCPServer::connect_to_peer(bool is_v6, const char* ip, int port)
 {
 	if (!ip || (strlen(ip) > sizeof(Client::m_addrString) - 16)) {
 		LOGERR(1, "failed to parse IP address, too long");
@@ -257,8 +284,7 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(bool is_v6, const
 	return connect_to_peer(client);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(bool is_v6, const raw_ip& ip, int port)
+bool TCPServer::connect_to_peer(bool is_v6, const raw_ip& ip, int port)
 {
 	if (m_finished.load()) {
 		return false;
@@ -274,13 +300,17 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(bool is_v6, const
 	return connect_to_peer(client);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::is_banned(const raw_ip& ip)
+bool TCPServer::is_banned(bool is_v6, raw_ip ip)
 {
 	if (ip.is_localhost()) {
 		return false;
 	}
 
+	// If it's IPv6, check the whole /64 prefix
+	if (is_v6 && !ip.is_ipv4_prefix()) {
+		memset(ip.data + 8, 0, sizeof(ip.data) - 8);
+	}
+
 	const auto cur_time = std::chrono::steady_clock::now();
 
 	MutexLock lock(m_bansLock);
@@ -297,10 +327,9 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::is_banned(const raw_ip& ip)
 	return false;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(Client* client)
+bool TCPServer::connect_to_peer(Client* client)
 {
-	if (is_banned(client->m_addr)) {
+	if (is_banned(client->m_isV6, client->m_addr)) {
 		LOGINFO(5, "peer " << log::Gray() << static_cast<char*>(client->m_addrString) << log::NoColor() << " is banned, not connecting to it");
 		return_client(client);
 		return false;
@@ -327,7 +356,11 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(Client* client)
 		return false;
 	}
 
-	static_assert(sizeof(client->m_readBuf) >= sizeof(uv_connect_t), "READ_BUF_SIZE must be large enough");
+	if (client->m_readBufSize < sizeof(uv_connect_t)) {
+		LOGERR(1, "client read buf size is too small (" << client->m_readBufSize << " bytes), expected at least " << sizeof(uv_connect_t) << " bytes");
+		uv_close(reinterpret_cast<uv_handle_t*>(&client->m_socket), on_connection_error);
+		return false;
+	}
 
 	uv_connect_t* connect_request = reinterpret_cast<uv_connect_t*>(client->m_readBuf);
 	memset(connect_request, 0, sizeof(uv_connect_t));
@@ -379,17 +412,15 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::connect_to_peer(Client* client)
 }
 
 #ifdef P2POOL_DEBUGGING
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::check_event_loop_thread(const char* func)
+void TCPServer::check_event_loop_thread(const char* func) const
 {
-	if (!server_event_loop_thread) {
+	if (server_event_loop_thread != this) {
 		LOGERR(1, func << " called from another thread, this is not thread safe");
 	}
 }
 #endif
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::close_sockets(bool listen_sockets)
+void TCPServer::close_sockets(bool listen_sockets)
 {
 	check_event_loop_thread(__func__);
 
@@ -423,8 +454,12 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::close_sockets(bool listen_sockets
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::shutdown_tcp()
+void TCPServer::error_invalid_ip(const std::string& address)
+{
+	LOGERR(1, "invalid IP:port " << address);
+}
+
+void TCPServer::shutdown_tcp()
 {
 	if (m_finished.exchange(1)) {
 		return;
@@ -445,29 +480,31 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::shutdown_tcp()
 	LOGINFO(1, "stopped");
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::print_status()
+void TCPServer::print_status()
 {
 	LOGINFO(0, "status" <<
 		"\nConnections = " << m_numConnections.load() << " (" << m_numIncomingConnections.load() << " incoming)"
 	);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::ban(const raw_ip& ip, uint64_t seconds)
+void TCPServer::ban(bool is_v6, raw_ip ip, uint64_t seconds)
 {
 	if (ip.is_localhost()) {
 		return;
 	}
 
+	// If it's IPv6, ban the whole /64 prefix
+	if (is_v6 && !ip.is_ipv4_prefix()) {
+		memset(ip.data + 8, 0, sizeof(ip.data) - 8);
+	}
+
 	const auto ban_time = std::chrono::steady_clock::now() + std::chrono::seconds(seconds);
 
 	MutexLock lock(m_bansLock);
 	m_bans[ip] = ban_time;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::print_bans()
+void TCPServer::print_bans()
 {
 	using namespace std::chrono;
 	const auto cur_time = steady_clock::now();
@@ -482,8 +519,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::print_bans()
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::send_internal(Client* client, SendCallbackBase&& callback)
+bool TCPServer::send_internal(Client* client, Callback<size_t, uint8_t*, size_t>::Base&& callback)
 {
 	check_event_loop_thread(__func__);
 
@@ -492,23 +528,20 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::send_internal(Client* client, Sen
 		return true;
 	}
 
-	WriteBuf* buf = get_write_buffer();
+	const size_t bytes_written = callback(m_callbackBuf.data(), m_callbackBuf.size());
 
-	// callback_buf is used in only 1 thread, so it's safe
-	static uint8_t callback_buf[WRITE_BUF_SIZE];
-	const size_t bytes_written = callback(callback_buf, sizeof(callback_buf));
-
-	if (bytes_written > WRITE_BUF_SIZE) {
-		LOGERR(0, "send callback wrote " << bytes_written << " bytes, expected no more than " << WRITE_BUF_SIZE << " bytes");
+	if (bytes_written > m_callbackBuf.size()) {
+		LOGERR(0, "send callback wrote " << bytes_written << " bytes, expected no more than " << m_callbackBuf.size() << " bytes");
 		PANIC_STOP();
 	}
 
 	if (bytes_written == 0) {
 		LOGWARN(1, "send callback wrote 0 bytes, nothing to do");
-		return_write_buffer(buf);
 		return true;
 	}
 
+	WriteBuf* buf = get_write_buffer(bytes_written);
+
 	buf->m_write.data = buf;
 	buf->m_client = client;
 
@@ -521,7 +554,7 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::send_internal(Client* client, Sen
 		}
 	}
 
-	memcpy(buf->m_data, callback_buf, bytes_written);
+	memcpy(buf->m_data, m_callbackBuf.data(), bytes_written);
 
 	uv_buf_t bufs[1];
 	bufs[0].base = reinterpret_cast<char*>(buf->m_data);
@@ -537,21 +570,25 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::send_internal(Client* client, Sen
 	return true;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::loop(void* data)
+void TCPServer::loop(void* data)
 {
-	LOGINFO(1, "event loop started");
-	server_event_loop_thread = true;
 	TCPServer* server = static_cast<TCPServer*>(data);
+	log_category_prefix = server->get_category();
 
-	server->m_writeBuffers.resize(DEFAULT_BACKLOG);
-	server->m_preallocatedClients.reserve(DEFAULT_BACKLOG);
-	for (size_t i = 0; i < DEFAULT_BACKLOG; ++i) {
+	LOGINFO(1, "event loop started");
+	server_event_loop_thread = data;
+
+	server->m_preallocatedClients.reserve(server->m_defaultBacklog);
+	for (int i = 0; i < server->m_defaultBacklog; ++i) {
 		WriteBuf* wb = new WriteBuf();
+		const size_t capacity = wb->m_dataCapacity;
+
 		Client* c = server->m_allocateNewClient();
+
 		ASAN_POISON_MEMORY_REGION(wb, sizeof(WriteBuf));
 		ASAN_POISON_MEMORY_REGION(c, c->size());
-		server->m_writeBuffers[i] = wb;
+
+		server->m_writeBuffers.emplace(capacity, wb);
 		server->m_preallocatedClients.emplace_back(c);
 	}
 
@@ -565,7 +602,9 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::loop(void* data)
 		LOGWARN(1, "uv_loop_close returned error " << uv_err_name(err));
 	}
 
-	for (WriteBuf* buf : server->m_writeBuffers) {
+	for (const auto& it : server->m_writeBuffers) {
+		WriteBuf* buf = it.second;
+
 		ASAN_UNPOISON_MEMORY_REGION(buf, sizeof(WriteBuf));
 		if (buf->m_data) {
 			ASAN_UNPOISON_MEMORY_REGION(buf->m_data, buf->m_dataCapacity);
@@ -585,8 +624,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::loop(void* data)
 	LOGINFO(1, "event loop stopped");
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_connection(uv_stream_t* server, int status)
+void TCPServer::on_new_connection(uv_stream_t* server, int status)
 {
 	TCPServer* pThis = static_cast<TCPServer*>(server->data);
 
@@ -602,17 +640,16 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_connection(uv_stream_t* se
 	pThis->on_new_client(server);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_connection_close(uv_handle_t* handle)
+void TCPServer::on_connection_close(uv_handle_t* handle)
 {
-	check_event_loop_thread(__func__);
-
 	Client* client = static_cast<Client*>(handle->data);
 	TCPServer* owner = client->m_owner;
 
 	LOGINFO(5, "peer " << log::Gray() << static_cast<char*>(client->m_addrString) << log::NoColor() << " disconnected");
 
 	if (owner) {
+		owner->check_event_loop_thread(__func__);
+
 		Client* prev_in_list = client->m_prev;
 		Client* next_in_list = client->m_next;
 
@@ -635,15 +672,13 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_connection_close(uv_handle_t*
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_connection_error(uv_handle_t* handle)
+void TCPServer::on_connection_error(uv_handle_t* handle)
 {
 	Client* client = reinterpret_cast<Client*>(handle->data);
 	client->m_owner->return_client(client);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_connect(uv_connect_t* req, int status)
+void TCPServer::on_connect(uv_connect_t* req, int status)
 {
 	Client* client = reinterpret_cast<Client*>(req->data);
 
@@ -669,8 +704,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_connect(uv_connect_t* req, int
 	server->on_new_client(nullptr, client);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_client(uv_stream_t* server)
+void TCPServer::on_new_client(uv_stream_t* server)
 {
 	if (m_finished.load()) {
 		return;
@@ -704,8 +738,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_client(uv_stream_t* server
 	on_new_client(server, client);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_client(uv_stream_t* server, Client* client)
+void TCPServer::on_new_client(uv_stream_t* server, Client* client)
 {
 	check_event_loop_thread(__func__);
 
@@ -749,7 +782,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_client(uv_stream_t* server
 
 	LOGINFO(5, "new connection " << (client->m_isIncoming ? "from " : "to ") << log::Gray() << static_cast<char*>(client->m_addrString));
 
-	if (is_banned(client->m_addr)) {
+	if (is_banned(client->m_isV6, client->m_addr)) {
 		LOGINFO(5, "peer " << log::Gray() << static_cast<char*>(client->m_addrString) << log::NoColor() << " is banned, disconnecting");
 		client->close();
 		return;
@@ -799,8 +832,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_new_client(uv_stream_t* server
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_shutdown(uv_async_t* async)
+void TCPServer::on_shutdown(uv_async_t* async)
 {
 	TCPServer* s = reinterpret_cast<TCPServer*>(async->data);
 	s->on_shutdown();
@@ -860,14 +892,20 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::on_shutdown(uv_async_t* async)
 		});
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-typename TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::WriteBuf* TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::get_write_buffer()
+TCPServer::WriteBuf* TCPServer::get_write_buffer(size_t size_hint)
 {
 	WriteBuf* buf;
 
 	if (!m_writeBuffers.empty()) {
-		buf = m_writeBuffers.back();
-		m_writeBuffers.pop_back();
+		// Try to find the smallest buffer that still has enough capacity
+		// If there is no buffer with enough capacity, just take the largest available buffer
+		auto it = m_writeBuffers.lower_bound(size_hint);
+		if (it == m_writeBuffers.end()) {
+			it = std::prev(it);
+		}
+
+		buf = it->second;
+		m_writeBuffers.erase(it);
 
 		ASAN_UNPOISON_MEMORY_REGION(buf, sizeof(WriteBuf));
 		if (buf->m_data) {
@@ -881,19 +919,19 @@ typename TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::WriteBuf* TCPServer<READ_BUF_
 	return buf;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::return_write_buffer(WriteBuf* buf)
+void TCPServer::return_write_buffer(WriteBuf* buf)
 {
+	const size_t capacity = buf->m_dataCapacity;
+
 	if (buf->m_data) {
-		ASAN_POISON_MEMORY_REGION(buf->m_data, buf->m_dataCapacity);
+		ASAN_POISON_MEMORY_REGION(buf->m_data, capacity);
 	}
 	ASAN_POISON_MEMORY_REGION(buf, sizeof(WriteBuf));
 
-	m_writeBuffers.push_back(buf);
+	m_writeBuffers.emplace(capacity, buf);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-typename TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client* TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::get_client()
+TCPServer::Client* TCPServer::get_client()
 {
 	Client* c;
 
@@ -911,16 +949,16 @@ typename TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client* TCPServer<READ_BUF_SI
 	return c;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::return_client(Client* c)
+void TCPServer::return_client(Client* c)
 {
 	ASAN_POISON_MEMORY_REGION(c, c->size());
 	m_preallocatedClients.push_back(c);
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::Client()
-	: m_owner(nullptr)
+TCPServer::Client::Client(char* read_buf, size_t size)
+	: m_readBuf(read_buf)
+	, m_readBufSize(static_cast<uint32_t>(size))
+	, m_owner(nullptr)
 	, m_prev(nullptr)
 	, m_next(nullptr)
 	, m_socket{}
@@ -936,11 +974,10 @@ TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::Client()
 	, m_resetCounter{ 0 }
 {
 	m_readBuf[0] = '\0';
-	m_readBuf[READ_BUF_SIZE - 1] = '\0';
+	m_readBuf[m_readBufSize - 1] = '\0';
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::reset()
+void TCPServer::Client::reset()
 {
 	m_resetCounter.fetch_add(1);
 
@@ -958,11 +995,10 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::reset()
 	m_addrString[0] = '\0';
 	m_socks5ProxyState = Socks5ProxyState::Default;
 	m_readBuf[0] = '\0';
-	m_readBuf[READ_BUF_SIZE - 1] = '\0';
+	m_readBuf[m_readBufSize - 1] = '\0';
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_alloc(uv_handle_t* handle, size_t /*suggested_size*/, uv_buf_t* buf)
+void TCPServer::Client::on_alloc(uv_handle_t* handle, size_t /*suggested_size*/, uv_buf_t* buf)
 {
 	Client* pThis = static_cast<Client*>(handle->data);
 
@@ -973,20 +1009,19 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_alloc(uv_handle_t* han
 		return;
 	}
 
-	if (pThis->m_numRead >= sizeof(pThis->m_readBuf)) {
+	if (pThis->m_numRead >= pThis->m_readBufSize) {
 		LOGWARN(4, "client " << static_cast<const char*>(pThis->m_addrString) << " read buffer is full");
 		buf->len = 0;
 		buf->base = nullptr;
 		return;
 	}
 
-	buf->len = sizeof(pThis->m_readBuf) - pThis->m_numRead;
+	buf->len = pThis->m_readBufSize - pThis->m_numRead;
 	buf->base = pThis->m_readBuf + pThis->m_numRead;
 	pThis->m_readBufInUse = true;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_read(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf)
+void TCPServer::Client::on_read(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf)
 {
 	Client* client = static_cast<Client*>(stream->data);
 	client->m_readBufInUse = false;
@@ -1021,10 +1056,9 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_read(uv_stream_t* stre
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_proxy_handshake(char* data, uint32_t size)
+bool TCPServer::Client::on_proxy_handshake(char* data, uint32_t size)
 {
-	if ((data != m_readBuf + m_numRead) || (data + size > m_readBuf + sizeof(m_readBuf))) {
+	if ((data != m_readBuf + m_numRead) || (data + size > m_readBuf + m_readBufSize)) {
 		LOGERR(1, "peer " << static_cast<char*>(m_addrString) << " invalid data pointer or size in on_read()");
 		return false;
 	}
@@ -1143,8 +1177,7 @@ bool TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_proxy_handshake(char*
 	return true;
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::on_write(uv_write_t* req, int status)
+void TCPServer::Client::on_write(uv_write_t* req, int status)
 {
 	WriteBuf* buf = static_cast<WriteBuf*>(req->data);
 	Client* client = buf->m_client;
@@ -1160,8 +1193,7 @@ 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()
+void TCPServer::Client::close()
 {
 	if (m_isClosing || !m_owner) {
 		// Already closed
@@ -1179,8 +1211,7 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::close()
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::ban(uint64_t seconds)
+void TCPServer::Client::ban(uint64_t seconds)
 {
 	if (m_addr.is_localhost()) {
 		return;
@@ -1188,12 +1219,11 @@ void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::ban(uint64_t seconds)
 
 	if (m_owner) {
 		LOGWARN(3, "peer " << static_cast<char*>(m_addrString) << " banned for " << seconds << " seconds");
-		m_owner->ban(m_addr, seconds);
+		m_owner->ban(m_isV6, m_addr, seconds);
 	}
 }
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
-void TCPServer<READ_BUF_SIZE, WRITE_BUF_SIZE>::Client::init_addr_string()
+void TCPServer::Client::init_addr_string()
 {
 	const char* addr_str;
 	char addr_str_buf[64];

src/tcp_server.h

@@ -18,22 +18,19 @@
 #pragma once
 
 #include "uv_util.h"
+#include <map>
 
 namespace p2pool {
 
-template<size_t READ_BUF_SIZE, size_t WRITE_BUF_SIZE>
 class TCPServer : public nocopy_nomove
 {
 public:
 	struct Client;
 	typedef Client* (*allocate_client_callback)();
 
-	explicit TCPServer(allocate_client_callback allocate_new_client);
+	TCPServer(int default_backlog, allocate_client_callback allocate_new_client);
 	virtual ~TCPServer();
 
-	template<typename T>
-	void parse_address_list(const std::string& address_list, T callback);
-
 	bool connect_to_peer(bool is_v6, const char* ip, int port);
 
 	void drop_connections_async() { if (m_finished.load() == 0) { uv_async_send(&m_dropConnectionsAsync); } }
@@ -47,12 +44,12 @@ public:
 	bool connect_to_peer(bool is_v6, const raw_ip& ip, int port);
 	virtual void on_connect_failed(bool /*is_v6*/, const raw_ip& /*ip*/, int /*port*/) {}
 
-	void ban(const raw_ip& ip, uint64_t seconds);
+	void ban(bool is_v6, raw_ip ip, uint64_t seconds);
 	virtual void print_bans();
 
 	struct Client
 	{
-		Client();
+		Client(char* read_buf, size_t size);
 		virtual ~Client() {}
 
 		virtual size_t size() const = 0;
@@ -73,7 +70,8 @@ public:
 
 		void init_addr_string();
 
-		alignas(8) char m_readBuf[READ_BUF_SIZE];
+		char* m_readBuf;
+		uint32_t m_readBufSize;
 
 		TCPServer* m_owner;
 
@@ -110,34 +108,21 @@ public:
 		size_t m_dataCapacity = 0;
 	};
 
-	std::vector<WriteBuf*> m_writeBuffers;
+	std::multimap<size_t, WriteBuf*> m_writeBuffers;
 
-	WriteBuf* get_write_buffer();
+	WriteBuf* get_write_buffer(size_t size_hint);
 	void return_write_buffer(WriteBuf* buf);
 
-	struct SendCallbackBase
+	template<typename T>
+	FORCEINLINE static void parse_address_list(const std::string& address_list, T&& callback)
 	{
-		virtual ~SendCallbackBase() {}
-		virtual size_t operator()(void*, size_t) = 0;
-	};
+		return parse_address_list_internal(address_list, Callback<void, bool, const std::string&, const std::string&, int>::Derived<T>(std::move(callback)));
+	}
 
 	template<typename T>
-	struct SendCallback : public SendCallbackBase
-	{
-		explicit FORCEINLINE SendCallback(T&& callback) : m_callback(std::move(callback)) {}
-		size_t operator()(void* buf, size_t buf_size) override { return m_callback(buf, buf_size); }
-
-	private:
-		SendCallback& operator=(SendCallback&&) = delete;
-
-		T m_callback;
-	};
-
-	template<typename T>
-	FORCEINLINE bool send(Client* client, T&& callback) { return send_internal(client, SendCallback<T>(std::move(callback))); }
+	FORCEINLINE bool send(Client* client, T&& callback) { return send_internal(client, Callback<size_t, uint8_t*, size_t>::Derived<T>(std::move(callback))); }
 
 private:
-	static void loop(void* data);
 	static void on_new_connection(uv_stream_t* server, int status);
 	static void on_connection_close(uv_handle_t* handle);
 	static void on_connection_error(uv_handle_t* handle);
@@ -147,18 +132,29 @@ private:
 
 	bool connect_to_peer(Client* client);
 
-	bool send_internal(Client* client, SendCallbackBase&& callback);
+	bool send_internal(Client* client, Callback<size_t, uint8_t*, size_t>::Base&& callback);
 
 	allocate_client_callback m_allocateNewClient;
 
 	void close_sockets(bool listen_sockets);
+	static void error_invalid_ip(const std::string& address);
 
 	std::vector<uv_tcp_t*> m_listenSockets6;
 	std::vector<uv_tcp_t*> m_listenSockets;
-	uv_thread_t m_loopThread;
 
 protected:
+	virtual const char* get_category() const { return "TCPServer "; }
+
+	std::vector<uint8_t> m_callbackBuf;
+	int m_defaultBacklog;
+
+	uv_thread_t m_loopThread;
+
+	static void loop(void* data);
+
+	static void parse_address_list_internal(const std::string& address_list, Callback<void, bool, const std::string&, const std::string&, int>::Base&& callback);
 	void start_listening(const std::string& listen_addresses, bool upnp);
+	bool start_listening(bool is_v6, const std::string& ip, int port, std::string address = std::string());
 
 #ifdef WITH_UPNP
 	int m_portMapping;
@@ -175,7 +171,7 @@ protected:
 	uv_loop_t m_loop;
 
 #ifdef P2POOL_DEBUGGING
-	static void check_event_loop_thread(const char *func);
+	void check_event_loop_thread(const char *func) const;
 #else
 	static FORCEINLINE void check_event_loop_thread(const char*) {}
 #endif
@@ -192,7 +188,7 @@ protected:
 	uv_mutex_t m_bansLock;
 	unordered_map<raw_ip, std::chrono::steady_clock::time_point> m_bans;
 
-	bool is_banned(const raw_ip& ip);
+	bool is_banned(bool is_v6, raw_ip ip);
 
 	unordered_set<raw_ip> m_pendingConnections;
 

src/util.cpp

@@ -19,7 +19,8 @@
 #include "util.h"
 #include "uv_util.h"
 #include <map>
-#include <thread>
+#include <istream>
+#include <ostream>
 
 #if !defined(_WIN32) && defined(HAVE_SCHED)
 #include <sched.h>

src/util.h

@@ -35,7 +35,7 @@
 namespace p2pool {
 
 #define P2POOL_VERSION_MAJOR 3
-#define P2POOL_VERSION_MINOR 2
+#define P2POOL_VERSION_MINOR 3
 
 extern const char* VERSION;
 
@@ -275,11 +275,33 @@ struct PerfTimer
 #define PERFLOG(level, name) PerfTimer CONCAT(perf_timer_, __LINE__)(level, name)
 #endif
 
+template<typename R, typename ...Args>
+struct Callback
+{
+	struct Base
+	{
+		virtual ~Base() {}
+		virtual R operator()(Args...) = 0;
+	};
+
+	template<typename T>
+	struct Derived : public Base
+	{
+		explicit FORCEINLINE Derived(T&& cb) : m_cb(std::move(cb)) {}
+		R operator()(Args... args) override { return m_cb(args...); }
+
+	private:
+		Derived& operator=(Derived&&) = delete;
+		T m_cb;
+	};
+};
+
 } // namespace p2pool
 
 void memory_tracking_start();
 void memory_tracking_stop();
 void p2pool_usage();
+void p2pool_version();
 
 namespace robin_hood {
 

src/uv_util.h

@@ -64,22 +64,7 @@ void uv_rwlock_init_checked(uv_rwlock_t* lock);
 void uv_async_init_checked(uv_loop_t* loop, uv_async_t* async, uv_async_cb async_cb);
 uv_loop_t* uv_default_loop_checked();
 
-struct UV_LoopCallbackBase
-{
-	virtual ~UV_LoopCallbackBase() {}
-	virtual void operator()() = 0;
-};
-
-template<typename T>
-struct UV_LoopCallback : public UV_LoopCallbackBase
-{
-	explicit FORCEINLINE UV_LoopCallback(T&& cb) : m_cb(std::move(cb)) {}
-	void operator()() override { m_cb(); }
-
-private:
-	UV_LoopCallback& operator=(UV_LoopCallback&&) = delete;
-	T m_cb;
-};
+typedef Callback<void>::Base UV_LoopCallbackBase;
 
 struct UV_LoopUserData
 {
@@ -147,7 +132,7 @@ bool CallOnLoop(uv_loop_t* loop, T&& callback)
 		return false;
 	}
 
-	UV_LoopCallbackBase* cb = new UV_LoopCallback<T>(std::move(callback));
+	UV_LoopCallbackBase* cb = new Callback<void>::Derived<T>(std::move(callback));
 	{
 		MutexLock lock(data->m_callbacksLock);
 		data->m_callbacks.push_back(cb);

tests/CMakeLists.txt

@@ -9,6 +9,9 @@ if (${CMAKE_VERSION} VERSION_GREATER "3.5.2")
 	set_property(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} PROPERTY VS_STARTUP_PROJECT p2pool_tests)
 endif()
 
+set(BUILD_GMOCK OFF CACHE BOOL "Builds the googlemock subproject" FORCE)
+set(INSTALL_GTEST OFF CACHE BOOL "Enable installation of googletest. (Projects embedding googletest may want to turn this OFF.)" FORCE)
+
 add_subdirectory(googletest)
 set(LIBS gtest)
 
@@ -55,6 +58,7 @@ set(SOURCES
 	../src/pow_hash.cpp
 	../src/side_chain.cpp
 	../src/stratum_server.cpp
+	../src/tcp_server.cpp
 	../src/util.cpp
 	../src/wallet.cpp
 	../src/zmq_reader.cpp
@@ -174,5 +178,5 @@ add_executable(${CMAKE_PROJECT_NAME} ${HEADERS} ${SOURCES})
 target_link_libraries(${CMAKE_PROJECT_NAME} debug ${ZMQ_LIBRARY_DEBUG} debug ${UV_LIBRARY_DEBUG} debug ${CURL_LIBRARY_DEBUG} optimized ${ZMQ_LIBRARY} optimized ${UV_LIBRARY} optimized ${CURL_LIBRARY} ${LIBS})
 add_custom_command(TARGET ${CMAKE_PROJECT_NAME} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${CMAKE_SOURCE_DIR}/src/crypto_tests.txt" $<TARGET_FILE_DIR:${CMAKE_PROJECT_NAME}>)
 add_custom_command(TARGET ${CMAKE_PROJECT_NAME} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${CMAKE_SOURCE_DIR}/src/block.dat" $<TARGET_FILE_DIR:${CMAKE_PROJECT_NAME}>)
-add_custom_command(TARGET ${CMAKE_PROJECT_NAME} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${CMAKE_SOURCE_DIR}/src/sidechain_dump.dat" $<TARGET_FILE_DIR:${CMAKE_PROJECT_NAME}>)
-add_custom_command(TARGET ${CMAKE_PROJECT_NAME} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${CMAKE_SOURCE_DIR}/src/sidechain_dump_mini.dat" $<TARGET_FILE_DIR:${CMAKE_PROJECT_NAME}>)
+add_custom_command(TARGET ${CMAKE_PROJECT_NAME} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${CMAKE_SOURCE_DIR}/src/sidechain_dump.dat.gz" $<TARGET_FILE_DIR:${CMAKE_PROJECT_NAME}>)
+add_custom_command(TARGET ${CMAKE_PROJECT_NAME} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${CMAKE_SOURCE_DIR}/src/sidechain_dump_mini.dat.gz" $<TARGET_FILE_DIR:${CMAKE_PROJECT_NAME}>)

tests/src/pool_block_tests.cpp

@@ -87,7 +87,7 @@ TEST(pool_block, deserialize)
 	class RandomX_Hasher_Test : public RandomX_Hasher_Base
 	{
 	public:
-		bool calculate(const void* data, size_t size, uint64_t, const hash&, hash& result, bool force_light_mode) override
+		bool calculate(const void* data, size_t size, uint64_t, const hash&, hash& result, bool /*force_light_mode*/) override
 		{
 			if (size == 76) {
 				char buf[76 * 2 + 1];
@@ -131,8 +131,8 @@ TEST(pool_block, verify)
 		uint64_t m_txinGenHeight;
 		uint64_t m_sidechainHeight;
 	} tests[2] = {
-		{ "default", "sidechain_dump.dat", 2845288, 4674368 },
-		{ "mini", "sidechain_dump_mini.dat", 2845293, 4129185 },
+		{ "default", "sidechain_dump.dat", 2870010, 4957203 },
+		{ "mini", "sidechain_dump_mini.dat", 2870010, 4414446 },
 	};
 
 	for (const STest& t : tests)
@@ -140,8 +140,8 @@ TEST(pool_block, verify)
 		PoolBlock b;
 		SideChain sidechain(nullptr, NetworkType::Mainnet, t.m_poolName);
 
-		// Difficulty of block 2844672
-		sidechain.m_testMainChainDiff = difficulty_type(321967641416ULL, 0ULL);
+		// Difficulty of block 2869248
+		sidechain.m_testMainChainDiff = difficulty_type(345786476185ULL, 0ULL);
 
 		std::ifstream f(t.m_fileName, std::ios::binary | std::ios::ate);
 		ASSERT_EQ(f.good() && f.is_open(), true);