Manage all processes with GuardedProcessPool

This way we can destroy the processes and wait for them to exit in parallel, and ensure the process is destroyed so that we won't encounter strange port being used issue at the same time.
2018-04-09 21:54:11 -07:00
parent 7226f8aab4
commit 68f3f8b20d
5 changed files with 61 additions and 87 deletions
@@ -76,7 +76,7 @@ object BaseService {
        @Volatile var state = STOPPED
        @Volatile var plugin = PluginOptions()
        @Volatile var pluginPath: String? = null
-        var sslocalProcess: GuardedProcess? = null
+        val processes = GuardedProcessPool()

        var timer: Timer? = null
        var trafficMonitorThread: TrafficMonitorThread? = null
@@ -274,7 +274,7 @@ object BaseService {

            if (TcpFastOpen.sendEnabled) cmd += "--fast-open"

-            data.sslocalProcess = GuardedProcess(cmd).start()
+            data.processes.start(cmd)
        }

        fun createNotification(profileName: String): ServiceNotification
@@ -285,11 +285,7 @@ object BaseService {
            else startService(Intent(this, javaClass))
        }

-        fun killProcesses() {
-            val data = data
-            data.sslocalProcess?.destroy()
-            data.sslocalProcess = null
-        }
+        fun killProcesses() = data.processes.killAll()

        fun stopRunner(stopService: Boolean, msg: String? = null) {
            // channge the state
@@ -31,31 +31,34 @@ import com.github.shadowsocks.utils.thread
 import java.io.File
 import java.io.IOException
 import java.io.InputStream
-import java.util.concurrent.Semaphore
+import java.util.*
+import java.util.concurrent.ArrayBlockingQueue

-class GuardedProcess(private val cmd: List<String>) {
+class GuardedProcessPool {
    companion object {
-        private const val TAG = "GuardedProcess"
+        private const val TAG = "GuardedProcessPool"
+        private val dummy = IOException()
    }

-    private lateinit var guardThread: Thread
-    @Volatile
-    private var isDestroyed = false
-    @Volatile
-    private lateinit var process: Process
-    private val name = File(cmd.first()).nameWithoutExtension
+    private inner class Guard(private val cmd: List<String>, private val onRestartCallback: (() -> Unit)?) {
+        val cmdName = File(cmd.first()).nameWithoutExtension
+        val excQueue = ArrayBlockingQueue<IOException>(1)   // ArrayBlockingQueue doesn't want null
+        private var pushed = false

-    private fun streamLogger(input: InputStream, logger: (String, String) -> Int) = thread("StreamLogger-$name") {
-        try {
-            input.bufferedReader().useLines { it.forEach { logger(TAG, it) } }
-        } catch (_: IOException) { }    // ignore
-    }
+        private fun streamLogger(input: InputStream, logger: (String, String) -> Int) =
+                thread("StreamLogger-$cmdName") {
+                    try {
+                        input.bufferedReader().useLines { it.forEach { logger(TAG, it) } }
+                    } catch (_: IOException) { }    // ignore
+                }
+        private fun pushException(ioException: IOException?) {
+            if (pushed) return
+            excQueue.put(ioException ?: dummy)
+            pushed = true
+        }

-    fun start(onRestartCallback: (() -> Unit)? = null): GuardedProcess {
-        val semaphore = Semaphore(1)
-        semaphore.acquire()
-        var ioException: IOException? = null
-        guardThread = thread("GuardThread-$name") {
+        fun looper() {
+            var process: Process? = null
            try {
                var callback: (() -> Unit)? = null
                while (!isDestroyed) {
@@ -72,44 +75,54 @@ class GuardedProcess(private val cmd: List<String>) {

                    if (callback == null) callback = onRestartCallback else callback()

-                    semaphore.release()
+                    pushException(null)
                    process.waitFor()

                    synchronized(this) {
                        if (SystemClock.elapsedRealtime() - startTime < 1000) {
-                            Log.w(TAG, "process exit too fast, stop guard: " + Commandline.toString(cmd))
+                            Log.w(TAG, "process exit too fast, stop guard: $cmdName")
                            isDestroyed = true
                        }
                    }
                }
            } catch (_: InterruptedException) {
-                if (BuildConfig.DEBUG) Log.d(TAG, "thread interrupt, destroy process: " + Commandline.toString(cmd))
-                destroyProcess()
+                if (BuildConfig.DEBUG) Log.d(TAG, "thread interrupt, destroy process: $cmdName")
            } catch (e: IOException) {
-                ioException = e
+                pushException(e)
            } finally {
-                semaphore.release()
+                if (process != null) {
+                    if (Build.VERSION.SDK_INT < 24) @Suppress("DEPRECATION") {
+                        JniHelper.sigtermCompat(process)
+                        JniHelper.waitForCompat(process, 500)
+                    }
+                    process.destroy()
+                    process.waitFor()   // ensure the process is destroyed
+                }
+                pushException(null)
            }
        }
-        semaphore.acquire()
-        if (ioException != null) throw ioException!!
+    }
+
+    private val guardThreads = HashSet<Thread>()
+    @Volatile
+    private var isDestroyed = false
+
+    fun start(cmd: List<String>, onRestartCallback: (() -> Unit)? = null): GuardedProcessPool {
+        val guard = Guard(cmd, onRestartCallback)
+        guardThreads.add(thread("GuardThread-${guard.cmdName}", block = guard::looper))
+        val ioException = guard.excQueue.take()
+        if (ioException !== dummy) throw ioException
        return this
    }

-    fun destroy() {
+    fun killAll() {
        isDestroyed = true
-        guardThread.interrupt()
-        destroyProcess()
+        guardThreads.forEach { it.interrupt() }
        try {
-            guardThread.join()
+            guardThreads.forEach { it.join() }
        } catch (_: InterruptedException) { }
-    }

-    private fun destroyProcess() {
-        if (Build.VERSION.SDK_INT < 24) @Suppress("DEPRECATION") {
-            JniHelper.sigtermCompat(process)
-            JniHelper.waitForCompat(process, 500)
-        }
-        process.destroy()
+        guardThreads.clear()
+        isDestroyed = false
    }
 }
@@ -28,19 +28,9 @@ import org.json.JSONArray
 import org.json.JSONObject
 import java.io.File
 import java.net.Inet6Address
-import java.util.*

-/**
- * This object also uses WeakMap to simulate the effects of multi-inheritance, but more lightweight.
- */
 object LocalDnsService {
    interface Interface : BaseService.Interface {
-        var overtureProcess: GuardedProcess?
-            get() = overtureProcesses[this]
-            set(value) {
-                if (value == null) overtureProcesses.remove(this) else overtureProcesses[this] = value
-            }
-
        override fun startNativeProcesses() {
            super.startNativeProcesses()
            val data = data
@@ -96,18 +86,10 @@ object LocalDnsService {
                return file
            }

-            if (!profile.udpdns) overtureProcess = GuardedProcess(buildAdditionalArguments(arrayListOf(
+            if (!profile.udpdns) data.processes.start(buildAdditionalArguments(arrayListOf(
                    File(app.applicationInfo.nativeLibraryDir, Executable.OVERTURE).absolutePath,
                    "-c", buildOvertureConfig("overture.conf")
-            ))).start()
-        }
-
-        override fun killProcesses() {
-            super.killProcesses()
-            overtureProcess?.destroy()
-            overtureProcess = null
+            )))
        }
    }
-
-    private val overtureProcesses = WeakHashMap<Interface, GuardedProcess>()
 }
@@ -40,18 +40,14 @@ class TransproxyService : Service(), LocalDnsService.Interface {
    override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int =
            super<LocalDnsService.Interface>.onStartCommand(intent, flags, startId)

-    private var sstunnelProcess: GuardedProcess? = null
-    private var redsocksProcess: GuardedProcess? = null
-
    private fun startDNSTunnel() {
-        val cmd = arrayListOf(File(applicationInfo.nativeLibraryDir, Executable.SS_TUNNEL).absolutePath,
+        data.processes.start(listOf(File(applicationInfo.nativeLibraryDir, Executable.SS_TUNNEL).absolutePath,
                "-t", "10",
                "-b", "127.0.0.1",
                "-u",
                "-l", DataStore.portLocalDns.toString(),            // ss-tunnel listens on the same port as overture
                "-L", data.profile!!.remoteDns.split(",").first().trim() + ":53",
-                "-c", data.shadowsocksConfigFile!!.absolutePath)    // config is already built by BaseService.Interface
-        sstunnelProcess = GuardedProcess(cmd).start()
+                "-c", data.shadowsocksConfigFile!!.absolutePath))   // config is already built by BaseService.Interface
    }

    private fun startRedsocksDaemon() {
@@ -70,10 +66,8 @@ redsocks {
 type = socks5;
 }
 """)
-        redsocksProcess = GuardedProcess(arrayListOf(
-                File(applicationInfo.nativeLibraryDir, Executable.REDSOCKS).absolutePath,
-                "-c", "redsocks.conf")
-        ).start()
+        data.processes.start(listOf(
+                File(applicationInfo.nativeLibraryDir, Executable.REDSOCKS).absolutePath, "-c", "redsocks.conf"))
    }

    override fun startNativeProcesses() {
@@ -81,12 +75,4 @@ redsocks {
        super.startNativeProcesses()
        if (data.profile!!.udpdns) startDNSTunnel()
    }
-
-    override fun killProcesses() {
-        super.killProcesses()
-        sstunnelProcess?.destroy()
-        sstunnelProcess = null
-        redsocksProcess?.destroy()
-        redsocksProcess = null
-    }
 }
@@ -116,7 +116,6 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {

    private var conn: ParcelFileDescriptor? = null
    private var worker: ProtectWorker? = null
-    private var tun2socksProcess: GuardedProcess? = null
    private var underlyingNetwork: Network? = null
        @TargetApi(28)
        set(value) {
@@ -155,8 +154,6 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {
        worker?.stopThread()
        worker = null
        super.killProcesses()
-        tun2socksProcess?.destroy()
-        tun2socksProcess = null
        conn?.close()
        conn = null
    }
@@ -256,7 +253,7 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {
            cmd += "--dnsgw"
            cmd += "127.0.0.1:${DataStore.portLocalDns}"
        }
-        tun2socksProcess = GuardedProcess(cmd).start { sendFd(fd) }
+        data.processes.start(cmd) { sendFd(fd) }
        return fd
    }