Unverified Commit 89c5d942 authored by Max Lv's avatar Max Lv Committed by GitHub

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

Manage all processes with GuardedProcessPool
parents 0b98a2d4 68f3f8b2
...@@ -76,7 +76,7 @@ object BaseService { ...@@ -76,7 +76,7 @@ object BaseService {
@Volatile var state = STOPPED @Volatile var state = STOPPED
@Volatile var plugin = PluginOptions() @Volatile var plugin = PluginOptions()
@Volatile var pluginPath: String? = null @Volatile var pluginPath: String? = null
var sslocalProcess: GuardedProcess? = null val processes = GuardedProcessPool()
var timer: Timer? = null var timer: Timer? = null
var trafficMonitorThread: TrafficMonitorThread? = null var trafficMonitorThread: TrafficMonitorThread? = null
...@@ -274,7 +274,7 @@ object BaseService { ...@@ -274,7 +274,7 @@ object BaseService {
if (TcpFastOpen.sendEnabled) cmd += "--fast-open" if (TcpFastOpen.sendEnabled) cmd += "--fast-open"
data.sslocalProcess = GuardedProcess(cmd).start() data.processes.start(cmd)
} }
fun createNotification(profileName: String): ServiceNotification fun createNotification(profileName: String): ServiceNotification
...@@ -285,11 +285,7 @@ object BaseService { ...@@ -285,11 +285,7 @@ object BaseService {
else startService(Intent(this, javaClass)) else startService(Intent(this, javaClass))
} }
fun killProcesses() { fun killProcesses() = data.processes.killAll()
val data = data
data.sslocalProcess?.destroy()
data.sslocalProcess = null
}
fun stopRunner(stopService: Boolean, msg: String? = null) { fun stopRunner(stopService: Boolean, msg: String? = null) {
// channge the state // channge the state
......
...@@ -31,31 +31,34 @@ import com.github.shadowsocks.utils.thread ...@@ -31,31 +31,34 @@ import com.github.shadowsocks.utils.thread
import java.io.File import java.io.File
import java.io.IOException import java.io.IOException
import java.io.InputStream 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 { companion object {
private const val TAG = "GuardedProcess" private const val TAG = "GuardedProcessPool"
private val dummy = IOException()
} }
private lateinit var guardThread: Thread private inner class Guard(private val cmd: List<String>, private val onRestartCallback: (() -> Unit)?) {
@Volatile val cmdName = File(cmd.first()).nameWithoutExtension
private var isDestroyed = false val excQueue = ArrayBlockingQueue<IOException>(1) // ArrayBlockingQueue doesn't want null
@Volatile private var pushed = false
private lateinit var process: Process
private val name = File(cmd.first()).nameWithoutExtension
private fun streamLogger(input: InputStream, logger: (String, String) -> Int) = thread("StreamLogger-$name") { private fun streamLogger(input: InputStream, logger: (String, String) -> Int) =
try { thread("StreamLogger-$cmdName") {
input.bufferedReader().useLines { it.forEach { logger(TAG, it) } } try {
} catch (_: IOException) { } // ignore 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 { fun looper() {
val semaphore = Semaphore(1) var process: Process? = null
semaphore.acquire()
var ioException: IOException? = null
guardThread = thread("GuardThread-$name") {
try { try {
var callback: (() -> Unit)? = null var callback: (() -> Unit)? = null
while (!isDestroyed) { while (!isDestroyed) {
...@@ -72,44 +75,54 @@ class GuardedProcess(private val cmd: List<String>) { ...@@ -72,44 +75,54 @@ class GuardedProcess(private val cmd: List<String>) {
if (callback == null) callback = onRestartCallback else callback() if (callback == null) callback = onRestartCallback else callback()
semaphore.release() pushException(null)
process.waitFor() process.waitFor()
synchronized(this) { synchronized(this) {
if (SystemClock.elapsedRealtime() - startTime < 1000) { 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 isDestroyed = true
} }
} }
} }
} catch (_: InterruptedException) { } catch (_: InterruptedException) {
if (BuildConfig.DEBUG) Log.d(TAG, "thread interrupt, destroy process: " + Commandline.toString(cmd)) if (BuildConfig.DEBUG) Log.d(TAG, "thread interrupt, destroy process: $cmdName")
destroyProcess()
} catch (e: IOException) { } catch (e: IOException) {
ioException = e pushException(e)
} finally { } 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 return this
} }
fun destroy() { fun killAll() {
isDestroyed = true isDestroyed = true
guardThread.interrupt() guardThreads.forEach { it.interrupt() }
destroyProcess()
try { try {
guardThread.join() guardThreads.forEach { it.join() }
} catch (_: InterruptedException) { } } catch (_: InterruptedException) { }
}
private fun destroyProcess() { guardThreads.clear()
if (Build.VERSION.SDK_INT < 24) @Suppress("DEPRECATION") { isDestroyed = false
JniHelper.sigtermCompat(process)
JniHelper.waitForCompat(process, 500)
}
process.destroy()
} }
} }
...@@ -28,19 +28,9 @@ import org.json.JSONArray ...@@ -28,19 +28,9 @@ import org.json.JSONArray
import org.json.JSONObject import org.json.JSONObject
import java.io.File import java.io.File
import java.net.Inet6Address import java.net.Inet6Address
import java.util.*
/**
* This object also uses WeakMap to simulate the effects of multi-inheritance, but more lightweight.
*/
object LocalDnsService { object LocalDnsService {
interface Interface : BaseService.Interface { 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() { override fun startNativeProcesses() {
super.startNativeProcesses() super.startNativeProcesses()
val data = data val data = data
...@@ -96,18 +86,10 @@ object LocalDnsService { ...@@ -96,18 +86,10 @@ object LocalDnsService {
return file return file
} }
if (!profile.udpdns) overtureProcess = GuardedProcess(buildAdditionalArguments(arrayListOf( if (!profile.udpdns) data.processes.start(buildAdditionalArguments(arrayListOf(
File(app.applicationInfo.nativeLibraryDir, Executable.OVERTURE).absolutePath, File(app.applicationInfo.nativeLibraryDir, Executable.OVERTURE).absolutePath,
"-c", buildOvertureConfig("overture.conf") "-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 { ...@@ -40,18 +40,14 @@ class TransproxyService : Service(), LocalDnsService.Interface {
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int = override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int =
super<LocalDnsService.Interface>.onStartCommand(intent, flags, startId) super<LocalDnsService.Interface>.onStartCommand(intent, flags, startId)
private var sstunnelProcess: GuardedProcess? = null
private var redsocksProcess: GuardedProcess? = null
private fun startDNSTunnel() { 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", "-t", "10",
"-b", "127.0.0.1", "-b", "127.0.0.1",
"-u", "-u",
"-l", DataStore.portLocalDns.toString(), // ss-tunnel listens on the same port as overture "-l", DataStore.portLocalDns.toString(), // ss-tunnel listens on the same port as overture
"-L", data.profile!!.remoteDns.split(",").first().trim() + ":53", "-L", data.profile!!.remoteDns.split(",").first().trim() + ":53",
"-c", data.shadowsocksConfigFile!!.absolutePath) // config is already built by BaseService.Interface "-c", data.shadowsocksConfigFile!!.absolutePath)) // config is already built by BaseService.Interface
sstunnelProcess = GuardedProcess(cmd).start()
} }
private fun startRedsocksDaemon() { private fun startRedsocksDaemon() {
...@@ -70,10 +66,8 @@ redsocks { ...@@ -70,10 +66,8 @@ redsocks {
type = socks5; type = socks5;
} }
""") """)
redsocksProcess = GuardedProcess(arrayListOf( data.processes.start(listOf(
File(applicationInfo.nativeLibraryDir, Executable.REDSOCKS).absolutePath, File(applicationInfo.nativeLibraryDir, Executable.REDSOCKS).absolutePath, "-c", "redsocks.conf"))
"-c", "redsocks.conf")
).start()
} }
override fun startNativeProcesses() { override fun startNativeProcesses() {
...@@ -81,12 +75,4 @@ redsocks { ...@@ -81,12 +75,4 @@ redsocks {
super.startNativeProcesses() super.startNativeProcesses()
if (data.profile!!.udpdns) startDNSTunnel() 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 { ...@@ -116,7 +116,6 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {
private var conn: ParcelFileDescriptor? = null private var conn: ParcelFileDescriptor? = null
private var worker: ProtectWorker? = null private var worker: ProtectWorker? = null
private var tun2socksProcess: GuardedProcess? = null
private var underlyingNetwork: Network? = null private var underlyingNetwork: Network? = null
@TargetApi(28) @TargetApi(28)
set(value) { set(value) {
...@@ -155,8 +154,6 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface { ...@@ -155,8 +154,6 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {
worker?.stopThread() worker?.stopThread()
worker = null worker = null
super.killProcesses() super.killProcesses()
tun2socksProcess?.destroy()
tun2socksProcess = null
conn?.close() conn?.close()
conn = null conn = null
} }
...@@ -256,7 +253,7 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface { ...@@ -256,7 +253,7 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {
cmd += "--dnsgw" cmd += "--dnsgw"
cmd += "127.0.0.1:${DataStore.portLocalDns}" cmd += "127.0.0.1:${DataStore.portLocalDns}"
} }
tun2socksProcess = GuardedProcess(cmd).start { sendFd(fd) } data.processes.start(cmd) { sendFd(fd) }
return fd return fd
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment