Merge pull request shadowsocks#1747 from Mygod/guarded-process-pool

Manage all processes with GuardedProcessPool

mobile/src/main/java/com/github/shadowsocks/bg/BaseService.kt

@@ -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

mobile/src/main/java/com/github/shadowsocks/bg/GuardedProcess.kt → mobile/src/main/java/com/github/shadowsocks/bg/GuardedProcessPool.kt

@@ -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
  }
 }

mobile/src/main/java/com/github/shadowsocks/bg/LocalDnsService.kt

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

mobile/src/main/java/com/github/shadowsocks/bg/TransproxyService.kt

@@ -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,23 +66,13 @@ 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() {
  startRedsocksDaemon()
  super.startNativeProcesses()
  if (data.profile!!.udpdns) startDNSTunnel()
  }
-
- override fun killProcesses() {
- super.killProcesses()
- sstunnelProcess?.destroy()
- sstunnelProcess = null
- redsocksProcess?.destroy()
- redsocksProcess = null
- }
 }

mobile/src/main/java/com/github/shadowsocks/bg/VpnService.kt

@@ -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
  }