fix android http cancelation

valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt

@@ -1,17 +1,64 @@
 package com.snap.valdi.network
 
 import android.content.Context
-import com.snap.valdi.utils.ExecutorsUtil
+import androidx.annotation.VisibleForTesting
+import java.io.IOException
 import java.io.InputStream
 import java.net.HttpURLConnection
 import java.net.URL
-import java.util.concurrent.Executors
+import java.net.URLConnection
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
 import com.snapchat.client.valdi.*
 import com.snapchat.client.valdi_core.*
 
-class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
+class DefaultHTTPRequestManager @VisibleForTesting constructor(
+    context: Context,
+    private val keepAliveMs: Long = DEFAULT_KEEP_ALIVE_MS,
+    private val openConnection: (URL) -> URLConnection,
+): HTTPRequestManager() {
 
-    private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map<String, String>, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable {
+    constructor(context: Context): this(context, openConnection = { it.openConnection() })
+
+    private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map<String, String>, val openConnection: (URL) -> URLConnection, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable {
+
+        private var connection: HttpURLConnection? = null
+        private var cancelled = false
+
+        override fun cancel() {
+            super.cancel()
+
+            val connectionToClose = synchronized(this) {
+                cancelled = true
+                connection
+            }
+
+            // Closing from another thread makes the worker's blocked read throw, which is
+            // how the thread gets reclaimed. Do it outside the lock so a slow close cannot stall
+            // the task binding its connection. Swallow like the worker's own teardown does: this
+            // races that teardown, and native is the only caller left to hand a throw to.
+            try {
+                connectionToClose?.disconnect()
+            } catch (exc: Exception) {}
+        }
+
+        private fun isCancelled(): Boolean = synchronized(this) { cancelled }
+
+        /**
+         * Hands the connection to [cancel] so it can be torn down. Returns false when the request
+         * was already cancelled, in which case the caller must not go on to perform it.
+         */
+        private fun bindConnection(urlConnection: HttpURLConnection): Boolean = synchronized(this) {
+            if (cancelled) {
+                false
+            } else {
+                connection = urlConnection
+                true
+            }
+        }
 
         private fun doPerformRequestWithURLConnection(urlConnection: HttpURLConnection): HTTPResponse {
             urlConnection.instanceFollowRedirects = true
@@ -25,6 +72,14 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
 
                 urlConnection.doInput = true
 
+                // Nothing above this point touches the network. Writing the body connects, and so
+                // does reading responseCode when there is no body, and disconnect() cannot tear
+                // down a connection that has not connected yet -- so this is the last point a
+                // cancel that arrived during setup can be honoured.
+                if (isCancelled()) {
+                    throw IOException("Request was cancelled")
+                }
+
                 if (body != null) {
                     urlConnection.doOutput = true
                     urlConnection.outputStream.write(body)
@@ -53,9 +108,14 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
         }
 
         private fun performRequest(): HTTPResponse {
-            val urlConnection = url.openConnection()
+            val urlConnection = openConnection(url)
 
             if (urlConnection is HttpURLConnection) {
+                if (!bindConnection(urlConnection)) {
+                    urlConnection.disconnect()
+                    throw IOException("Request was cancelled")
+                }
+
                 return doPerformRequestWithURLConnection(urlConnection)
             } else {
                 urlConnection.doInput = true
@@ -66,17 +126,26 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
         }
 
         override fun run() {
+            // Cancelled while queued: opening the connection at all would be wasted work.
+            if (isCancelled()) {
+                return
+            }
+
             try {
                 val response = performRequest()
                 notifySuccess(response)
-            } catch (error: Exception) {
+            } catch (error: Throwable) {
+                // Once cancelled the completion is already gone, so a teardown IOException lands
+                // here and goes nowhere -- matching how iOS swallows NSURLErrorCancelled.
+                // Catches Throwable so that reading an oversized body into memory fails the
+                // request rather than leaving the completion unsettled.
                 notifyFailure("HTTP Request failed: ${error.message}")
             }
         }
 
         companion object {
 
-            fun from(request: HTTPRequest, completion: HTTPRequestManagerCompletion): RequestTask {
+            fun from(request: HTTPRequest, openConnection: (URL) -> URLConnection, completion: HTTPRequestManagerCompletion): RequestTask {
                 val url = URL(request.url)
                 val method = request.method
                 val body = request.body
@@ -93,34 +162,93 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
                     }
                 }
 
-                return RequestTask(url, method, body, headers, completion)
+                return RequestTask(url, method, body, headers, openConnection, completion)
+            }
+        }
+    }
+
+    private val threadCount = AtomicInteger(0)
+
+    // One pool per host, matching NSURLSession.httpMaximumConnectionsPerHost, so a stalled host
+    // cannot starve requests to other hosts.
+    private val executors = ConcurrentHashMap<String, ThreadPoolExecutor>()
+
+    private fun newHostPool(): ThreadPoolExecutor =
+        ThreadPoolExecutor(
+            MAX_CONCURRENT_REQUESTS_PER_HOST,
+            MAX_CONCURRENT_REQUESTS_PER_HOST,
+            keepAliveMs,
+            TimeUnit.MILLISECONDS,
+            LinkedBlockingQueue(),
+        ) { r ->
+            Thread(r).apply {
+                name = "Valdi Network Thread ${threadCount.incrementAndGet()}"
+                priority = Thread.NORM_PRIORITY
             }
+        }.apply {
+            // An unbounded queue never rejects, so a pool only ever grows to its core size and
+            // maximumPoolSize is inert. Concurrency has to come from the core size, and this
+            // restores the idle reaping that the previous core size of zero provided.
+            allowCoreThreadTimeOut(true)
+        }
+
+    // Picking the pool and submitting to it as two steps would let a sweep shut it down in between,
+    // so compute() does both under the bin lock for the host.
+    private fun submitToHostPool(task: RequestTask) {
+        executors.compute(task.url.host.orEmpty().lowercase()) { _, pool ->
+            (pool ?: newHostPool()).also { it.execute(task) }
         }
     }
 
-    private var executors = ExecutorsUtil.newSingleThreadCachedExecutor { r ->
-        Thread(r).apply {
-            name = "Valdi Network Thread"
-            priority = Thread.NORM_PRIORITY
+    /**
+     * A pool with no threads has had none for a full keep-alive window, and [ThreadPoolExecutor]
+     * adds a worker back whenever the queue is non-empty, so no threads also means nothing queued.
+     *
+     * Runs after a request has been submitted rather than on a timer: [HTTPRequestManager] has no
+     * dispose hook, so there would be nowhere to stop a scheduler thread. Sweeping after the submit
+     * rather than before it keeps the pool this request just used out of the sweep.
+     */
+    private fun evictDormantHostPools() {
+        for (host in executors.keys) {
+            executors.computeIfPresent(host) { _, pool ->
+                if (pool.poolSize == 0) {
+                    pool.shutdown()
+                    null
+                } else {
+                    pool
+                }
+            }
         }
     }
 
+    @VisibleForTesting
+    fun hostPools(): Set<String> = executors.keys.toSet()
+
     override fun performRequest(request: HTTPRequest, completion: HTTPRequestManagerCompletion): Cancelable {
-        val task: RequestTask
         try {
-            task = RequestTask.from(request, completion)
-        } catch (exception: Exception) {
-            completion.onFail("Failed to build request: ${exception.message}")
+            val task = RequestTask.from(request, openConnection, completion)
+
+            submitToHostPool(task)
+            evictDormantHostPools()
+
+            return task
+        } catch (throwable: Throwable) {
+            // Anything escaping to JNI leaves the completion unsettled, which hangs the native
+            // request instead of failing it. Catches Throwable because starting a pool thread
+            // fails with OutOfMemoryError, not an Exception.
+            completion.onFail("Failed to perform request: ${throwable.message}")
 
             return object: Cancelable() {
                 override fun cancel() {
                 }
             }
         }
+    }
 
-        executors.submit(task)
+    companion object {
+        const val MAX_CONCURRENT_REQUESTS_PER_HOST = 4
 
-        return task
+        private const val DEFAULT_KEEP_ALIVE_MS = 60_000L
     }
 
 }