Fix throughput collapse under connection concurrency (relay I/O was serialized on a 64-thread cap)

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)
+    }
 }
 
 /**

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()

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
 

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() {