From 8a1acf7d88f91b9a9c4f6dd27a3ba7a74158dc45 Mon Sep 17 00:00:00 2001 From: NickLD Date: Thu, 6 Aug 2026 15:26:46 -0500 Subject: [PATCH 1/2] Move relay socket I/O off the shared Dispatchers.IO pool Every accepted connection's handshake, TCP copy loops, and UDP receive loops ran as blocking socket calls on Dispatchers.IO, whose default parallelism caps concurrently-running tasks at 64. With ~150 tunnels each holding 2 long-lived blocking reads open for the connection's lifetime, that ceiling is reached well before the connection count is: most connections queue behind the 64 that happened to get a thread, and a blocked read on an idle socket never yields it back. This is the throughput collapse seen at high concurrency (~144 connections) while a single connection stays fast. RelayDispatcher is a cached thread pool dedicated to this data path, so concurrently-active connections each get a real thread instead of contending for a fixed 64-thread limit. Threads are daemon and reclaimed after 60s idle, so it doesn't need explicit lifecycle management across proxy start/stop cycles. --- .../com/dataproxy/proxy/RelayDispatcher.kt | 22 +++++++++++++++++++ .../com/dataproxy/proxy/Socks5Connection.kt | 11 +++++----- .../com/dataproxy/proxy/Socks5UdpRelay.kt | 5 ++--- 3 files changed, 29 insertions(+), 9 deletions(-) create mode 100644 app/src/main/java/com/dataproxy/proxy/RelayDispatcher.kt diff --git a/app/src/main/java/com/dataproxy/proxy/RelayDispatcher.kt b/app/src/main/java/com/dataproxy/proxy/RelayDispatcher.kt new file mode 100644 index 0000000..7fb2594 --- /dev/null +++ b/app/src/main/java/com/dataproxy/proxy/RelayDispatcher.kt @@ -0,0 +1,22 @@ +package com.dataproxy.proxy + +import kotlinx.coroutines.asCoroutineDispatcher +import java.util.concurrent.Executors + +/** + * Dispatcher for blocking socket I/O on the SOCKS5 data path: handshake + * reads/writes, TCP relay copy loops, UDP receive loops. Each of these is a + * synchronous blocking call, so it occupies one OS thread for as long as it + * runs — for a relay copy loop, that's the lifetime of the connection. + * + * Dispatchers.IO caps concurrently-running tasks at its parallelism limit + * (64 by default). With ~150 tunnels each holding 2 blocking reads open for + * their whole lifetime, that limit is exhausted well before the connection + * count is, so most connections queue behind the 64 that got a thread — + * this was the root cause of throughput collapsing under concurrency. A + * cached pool grows with actual concurrent load instead of hitting a fixed + * ceiling, and reclaims idle threads after 60s. + */ +val RelayDispatcher = Executors.newCachedThreadPool { runnable -> + Thread(runnable, "socks5-relay").apply { isDaemon = true } +}.asCoroutineDispatcher() diff --git a/app/src/main/java/com/dataproxy/proxy/Socks5Connection.kt b/app/src/main/java/com/dataproxy/proxy/Socks5Connection.kt index 655fcee..5a135b4 100644 --- a/app/src/main/java/com/dataproxy/proxy/Socks5Connection.kt +++ b/app/src/main/java/com/dataproxy/proxy/Socks5Connection.kt @@ -3,7 +3,6 @@ package com.dataproxy.proxy import android.util.Log import com.dataproxy.network.CellularNetworkProvider import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll @@ -36,7 +35,7 @@ class Socks5Connection( private var outbound: Socket? = null private var udpRelay: Socks5UdpRelay? = null - fun handle(): Job = scope.launch(Dispatchers.IO) { + fun handle(): Job = scope.launch(RelayDispatcher) { try { clientSocket.tcpNoDelay = true clientSocket.soTimeout = HANDSHAKE_TIMEOUT_MS @@ -184,7 +183,7 @@ class Socks5Connection( val resolved: InetAddress? = when (target) { is Target.Ipv4 -> target.addr is Target.Ipv6 -> target.addr - is Target.Domain -> withContext(Dispatchers.IO) { + is Target.Domain -> withContext(RelayDispatcher) { cellular.resolveHost(target.host) } } @@ -213,7 +212,7 @@ class Socks5Connection( } return try { - withContext(Dispatchers.IO) { + withContext(RelayDispatcher) { remote.connect(InetSocketAddress(resolved, port), CONNECT_TIMEOUT_MS) } reply(output, REP_SUCCEEDED, remote.localSocketAddress as? InetSocketAddress) @@ -278,7 +277,7 @@ class Socks5Connection( // EOF or throws), tear down the UDP relay. clientSocket.soTimeout = 0 runCatching { - withContext(Dispatchers.IO) { + withContext(RelayDispatcher) { val buf = ByteArray(64) while (true) { val n = input.read(buf) @@ -291,7 +290,7 @@ class Socks5Connection( // ----------------------------------------------------------------- relay - private suspend fun relay(remote: Socket) = withContext(Dispatchers.IO) { + private suspend fun relay(remote: Socket) = withContext(RelayDispatcher) { val client = clientSocket val tracker = entry diff --git a/app/src/main/java/com/dataproxy/proxy/Socks5UdpRelay.kt b/app/src/main/java/com/dataproxy/proxy/Socks5UdpRelay.kt index 79e9db8..f788442 100644 --- a/app/src/main/java/com/dataproxy/proxy/Socks5UdpRelay.kt +++ b/app/src/main/java/com/dataproxy/proxy/Socks5UdpRelay.kt @@ -3,7 +3,6 @@ package com.dataproxy.proxy import android.util.Log import com.dataproxy.network.CellularNetworkProvider import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.launch import java.io.Closeable @@ -44,8 +43,8 @@ class Socks5UdpRelay( val port: Int get() = clientSocket.localPort fun start() { - clientLoopJob = scope.launch(Dispatchers.IO) { clientLoop() } - remoteLoopJob = scope.launch(Dispatchers.IO) { remoteLoop() } + clientLoopJob = scope.launch(RelayDispatcher) { clientLoop() } + remoteLoopJob = scope.launch(RelayDispatcher) { remoteLoop() } } private fun clientLoop() { From d8458b66dde4ce275d8a312c48b98ce5f9ba63da Mon Sep 17 00:00:00 2001 From: NickLD Date: Thu, 6 Aug 2026 15:27:43 -0500 Subject: [PATCH 2/2] Fix lost-update race in per-device byte/connection counters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DeviceAccumulator's fields were plain vars mutated via read-modify- write (`it.bytesUp += n`, `acc.activeConnections - 1`) from every connection coroutine sharing a client host, outside the ConcurrentHashMap.compute() lock that only open() used. Concurrent traffic from the same device (the normal case — a browser alone opens many simultaneous connections) could lose increments, making the Devices screen under-report bytes and active-connection counts. Switching every mutable field to AtomicInteger/AtomicLong makes each update self-contained regardless of what else is touching that device's stats concurrently. --- .../com/dataproxy/proxy/ConnectionRegistry.kt | 48 ++++++++++--------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/app/src/main/java/com/dataproxy/proxy/ConnectionRegistry.kt b/app/src/main/java/com/dataproxy/proxy/ConnectionRegistry.kt index 57fa058..6986366 100644 --- a/app/src/main/java/com/dataproxy/proxy/ConnectionRegistry.kt +++ b/app/src/main/java/com/dataproxy/proxy/ConnectionRegistry.kt @@ -4,6 +4,7 @@ import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicReference @@ -59,9 +60,9 @@ class ConnectionRegistry { deviceStats.compute(clientHost) { _, prev -> val acc = prev ?: DeviceAccumulator(firstSeenMs = now) - acc.activeConnections += 1 - acc.totalConnections += 1 - acc.lastSeenMs = now + acc.activeConnections.incrementAndGet() + acc.totalConnections.incrementAndGet() + acc.lastSeenMs.set(now) acc } refresh() @@ -73,8 +74,8 @@ class ConnectionRegistry { conn.bytesUp.addAndGet(n.toLong()) totalUp.addAndGet(n.toLong()) deviceStats[conn.clientHost]?.let { - it.bytesUp += n - it.lastSeenMs = System.currentTimeMillis() + it.bytesUp.addAndGet(n.toLong()) + it.lastSeenMs.set(System.currentTimeMillis()) } } @@ -83,16 +84,16 @@ class ConnectionRegistry { conn.bytesDown.addAndGet(n.toLong()) totalDown.addAndGet(n.toLong()) deviceStats[conn.clientHost]?.let { - it.bytesDown += n - it.lastSeenMs = System.currentTimeMillis() + it.bytesDown.addAndGet(n.toLong()) + it.lastSeenMs.set(System.currentTimeMillis()) } } fun close(conn: Connection) { connections.remove(conn.id) deviceStats[conn.clientHost]?.let { acc -> - acc.activeConnections = (acc.activeConnections - 1).coerceAtLeast(0) - acc.lastSeenMs = System.currentTimeMillis() + acc.activeConnections.updateAndGet { (it - 1).coerceAtLeast(0) } + acc.lastSeenMs.set(System.currentTimeMillis()) } refresh() } @@ -117,12 +118,12 @@ class ConnectionRegistry { .map { (host, acc) -> DeviceSummary( clientHost = host, - activeConnections = acc.activeConnections, - totalConnections = acc.totalConnections, - bytesUp = acc.bytesUp, - bytesDown = acc.bytesDown, + activeConnections = acc.activeConnections.get(), + totalConnections = acc.totalConnections.get(), + bytesUp = acc.bytesUp.get(), + bytesDown = acc.bytesDown.get(), firstSeenMs = acc.firstSeenMs, - lastSeenMs = acc.lastSeenMs, + lastSeenMs = acc.lastSeenMs.get(), ) } .sortedWith( @@ -137,14 +138,17 @@ class ConnectionRegistry { ) } - private class DeviceAccumulator( - var activeConnections: Int = 0, - var totalConnections: Int = 0, - var bytesUp: Long = 0L, - var bytesDown: Long = 0L, - var firstSeenMs: Long = 0L, - var lastSeenMs: Long = 0L, - ) + // Mutated concurrently from every connection coroutine sharing this + // client host, so every field that changes after construction needs to + // be a real atomic, not a plain var — a `+=` here is a lost-update race + // under concurrent traffic from the same device. + private class DeviceAccumulator(val firstSeenMs: Long) { + val activeConnections = AtomicInteger(0) + val totalConnections = AtomicInteger(0) + val bytesUp = AtomicLong(0L) + val bytesDown = AtomicLong(0L) + val lastSeenMs = AtomicLong(firstSeenMs) + } } /**