Commit dac2cc4f authored by Mygod's avatar Mygod

Allow clean-ups in parallel

parent a3e9dcb0
...@@ -233,9 +233,9 @@ object BaseService { ...@@ -233,9 +233,9 @@ object BaseService {
else startService(Intent(this, javaClass)) else startService(Intent(this, javaClass))
} }
suspend fun killProcesses() { fun killProcesses(scope: CoroutineScope) {
data.processes?.run { data.processes?.run {
close() close(scope)
data.processes = null data.processes = null
} }
} }
...@@ -246,26 +246,27 @@ object BaseService { ...@@ -246,26 +246,27 @@ object BaseService {
data.changeState(STOPPING) data.changeState(STOPPING)
GlobalScope.launch(Dispatchers.Main, CoroutineStart.UNDISPATCHED) { GlobalScope.launch(Dispatchers.Main, CoroutineStart.UNDISPATCHED) {
Core.analytics.logEvent("stop", bundleOf(Pair(FirebaseAnalytics.Param.METHOD, tag))) Core.analytics.logEvent("stop", bundleOf(Pair(FirebaseAnalytics.Param.METHOD, tag)))
killProcesses()
// clean up recevier
this@Interface as Service this@Interface as Service
val data = data // we use a coroutineScope here to allow clean-up in parallel
if (data.closeReceiverRegistered) { coroutineScope {
unregisterReceiver(data.closeReceiver) killProcesses(this)
data.closeReceiverRegistered = false // clean up receivers
} val data = data
if (data.closeReceiverRegistered) {
unregisterReceiver(data.closeReceiver)
data.closeReceiverRegistered = false
}
data.notification?.destroy() data.notification?.destroy()
data.notification = null data.notification = null
val ids = listOfNotNull(data.proxy, data.udpFallback).map { val ids = listOfNotNull(data.proxy, data.udpFallback).map {
it.close() it.close()
it.profile.id it.profile.id
}
data.proxy = null
data.binder.trafficPersisted(ids)
} }
data.proxy = null
data.binder.trafficPersisted(ids)
// change the state // change the state
data.changeState(STOPPED, msg) data.changeState(STOPPED, msg)
......
...@@ -32,7 +32,6 @@ import com.github.shadowsocks.Core ...@@ -32,7 +32,6 @@ import com.github.shadowsocks.Core
import com.github.shadowsocks.utils.Commandline import com.github.shadowsocks.utils.Commandline
import kotlinx.coroutines.* import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.selects.select
import java.io.File import java.io.File
import java.io.IOException import java.io.IOException
import java.io.InputStream import java.io.InputStream
...@@ -47,7 +46,6 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C ...@@ -47,7 +46,6 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
} }
private inner class Guard(private val cmd: List<String>) { private inner class Guard(private val cmd: List<String>) {
private val abortChannel = Channel<Unit>()
private lateinit var process: Process private lateinit var process: Process
private fun streamLogger(input: InputStream, logger: (String) -> Unit) = try { private fun streamLogger(input: InputStream, logger: (String) -> Unit) = try {
...@@ -58,12 +56,6 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C ...@@ -58,12 +56,6 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
process = ProcessBuilder(cmd).directory(Core.deviceStorage.noBackupFilesDir).start() process = ProcessBuilder(cmd).directory(Core.deviceStorage.noBackupFilesDir).start()
} }
suspend fun abort() {
if (abortChannel.isClosedForSend) return
abortChannel.send(Unit)
abortChannel.close()
}
suspend fun looper(onRestartCallback: (suspend () -> Unit)?) { suspend fun looper(onRestartCallback: (suspend () -> Unit)?) {
var running = true var running = true
val cmdName = File(cmd.first()).nameWithoutExtension val cmdName = File(cmd.first()).nameWithoutExtension
...@@ -77,13 +69,13 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C ...@@ -77,13 +69,13 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
runBlocking { exitChannel.send(process.waitFor()) } runBlocking { exitChannel.send(process.waitFor()) }
} }
val startTime = SystemClock.elapsedRealtime() val startTime = SystemClock.elapsedRealtime()
if (select { val exitCode = exitChannel.receive()
abortChannel.onReceive { true } // prefer abort to save work
exitChannel.onReceive { false }
}) break
running = false running = false
if (SystemClock.elapsedRealtime() - startTime < 1000) throw IOException("$cmdName exits too fast") if (SystemClock.elapsedRealtime() - startTime < 1000) {
Crashlytics.log(Log.DEBUG, TAG, "restart process: " + Commandline.toString(cmd)) throw IOException("$cmdName exits too fast (exit code: $exitCode)")
}
Crashlytics.log(Log.DEBUG, TAG,
"restart process: ${Commandline.toString(cmd)} (last exit code: $exitCode)")
start() start()
running = true running = true
onRestartCallback?.invoke() onRestartCallback?.invoke()
...@@ -92,7 +84,6 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C ...@@ -92,7 +84,6 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
Crashlytics.log(Log.WARN, TAG, "error occurred. stop guard: " + Commandline.toString(cmd)) Crashlytics.log(Log.WARN, TAG, "error occurred. stop guard: " + Commandline.toString(cmd))
GlobalScope.launch(Dispatchers.Main) { onFatal(e) } GlobalScope.launch(Dispatchers.Main) { onFatal(e) }
} finally { } finally {
abortChannel.close()
if (!running) return // process already exited, nothing to be done if (!running) return // process already exited, nothing to be done
withContext(NonCancellable) { // clean-up cannot be cancelled withContext(NonCancellable) { // clean-up cannot be cancelled
if (Build.VERSION.SDK_INT < 24) { if (Build.VERSION.SDK_INT < 24) {
...@@ -128,8 +119,8 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C ...@@ -128,8 +119,8 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
} }
@MainThread @MainThread
suspend fun close() { fun close(scope: CoroutineScope) {
guards.forEach { it.abort() } job.cancel()
job.cancelAndJoin() scope.launch { job.join() }
} }
} }
...@@ -27,6 +27,7 @@ import com.github.shadowsocks.net.LocalDnsServer ...@@ -27,6 +27,7 @@ import com.github.shadowsocks.net.LocalDnsServer
import com.github.shadowsocks.net.Socks5Endpoint import com.github.shadowsocks.net.Socks5Endpoint
import com.github.shadowsocks.net.Subnet import com.github.shadowsocks.net.Subnet
import com.github.shadowsocks.preference.DataStore import com.github.shadowsocks.preference.DataStore
import kotlinx.coroutines.CoroutineScope
import java.net.InetSocketAddress import java.net.InetSocketAddress
import java.net.URI import java.net.URI
import java.util.* import java.util.*
...@@ -61,9 +62,9 @@ object LocalDnsService { ...@@ -61,9 +62,9 @@ object LocalDnsService {
}.also { servers[this] = it }.start(InetSocketAddress(DataStore.listenAddress, DataStore.portLocalDns)) }.also { servers[this] = it }.start(InetSocketAddress(DataStore.listenAddress, DataStore.portLocalDns))
} }
override suspend fun killProcesses() { override fun killProcesses(scope: CoroutineScope) {
servers.remove(this)?.shutdown() servers.remove(this)?.shutdown(scope)
super.killProcesses() super.killProcesses(scope)
} }
} }
} }
...@@ -40,7 +40,9 @@ import com.github.shadowsocks.net.Subnet ...@@ -40,7 +40,9 @@ import com.github.shadowsocks.net.Subnet
import com.github.shadowsocks.preference.DataStore import com.github.shadowsocks.preference.DataStore
import com.github.shadowsocks.utils.Key import com.github.shadowsocks.utils.Key
import com.github.shadowsocks.utils.printLog import com.github.shadowsocks.utils.printLog
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import java.io.Closeable import java.io.Closeable
import java.io.File import java.io.File
import java.io.FileDescriptor import java.io.FileDescriptor
...@@ -112,11 +114,11 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface { ...@@ -112,11 +114,11 @@ class VpnService : BaseVpnService(), LocalDnsService.Interface {
override fun onRevoke() = stopRunner() override fun onRevoke() = stopRunner()
override suspend fun killProcesses() { override fun killProcesses(scope: CoroutineScope) {
super.killProcesses() super.killProcesses(scope)
active = false active = false
DefaultNetworkListener.stop(this) scope.launch { DefaultNetworkListener.stop(this) }
worker?.shutdown() worker?.shutdown(scope)
worker = null worker = null
conn?.close() conn?.close()
conn = null conn = null
......
...@@ -104,11 +104,13 @@ class ChannelMonitor { ...@@ -104,11 +104,13 @@ class ChannelMonitor {
await() await()
} }
suspend fun close() { fun close(scope: CoroutineScope) {
running = false running = false
selector.wakeup() selector.wakeup()
job.join() scope.launch {
selector.keys().forEach { it.channel().close() } job.join()
selector.close() selector.keys().forEach { it.channel().close() }
selector.close()
}
} }
} }
...@@ -34,10 +34,10 @@ abstract class ConcurrentLocalSocketListener(name: String, socketFile: File) : L ...@@ -34,10 +34,10 @@ abstract class ConcurrentLocalSocketListener(name: String, socketFile: File) : L
launch { super.accept(socket) } launch { super.accept(socket) }
} }
suspend fun shutdown() { fun shutdown(scope: CoroutineScope) {
running = false running = false
job.cancel() job.cancel()
close() close()
job.join() scope.launch { job.join() }
} }
} }
...@@ -161,9 +161,9 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd ...@@ -161,9 +161,9 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd
} }
} }
suspend fun shutdown() { fun shutdown(scope: CoroutineScope) {
job.cancel() job.cancel()
monitor.close() monitor.close(scope)
job.join() scope.launch { job.join() }
} }
} }
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