Commit fe9b773d authored by Mygod's avatar Mygod

Shutdown service on process exit too fast or restart failures

Also handles possible double shutdown.
parent b6880efd
...@@ -77,6 +77,7 @@ object BaseService { ...@@ -77,6 +77,7 @@ object BaseService {
var closeReceiverRegistered = false var closeReceiverRegistered = false
val binder = Binder(this) val binder = Binder(this)
var connectingJob: Job? = null
fun changeState(s: Int, msg: String? = null) { fun changeState(s: Int, msg: String? = null) {
if (state == s && msg == null) return if (state == s && msg == null) return
...@@ -239,6 +240,7 @@ object BaseService { ...@@ -239,6 +240,7 @@ object BaseService {
} }
fun stopRunner(restart: Boolean = false, msg: String? = null) { fun stopRunner(restart: Boolean = false, msg: String? = null) {
if (data.state == STOPPING) return
// channge the state // channge the state
data.changeState(STOPPING) data.changeState(STOPPING)
GlobalScope.launch(Dispatchers.Main, CoroutineStart.UNDISPATCHED) { GlobalScope.launch(Dispatchers.Main, CoroutineStart.UNDISPATCHED) {
...@@ -302,14 +304,17 @@ object BaseService { ...@@ -302,14 +304,17 @@ object BaseService {
Core.analytics.logEvent("start", bundleOf(Pair(FirebaseAnalytics.Param.METHOD, tag))) Core.analytics.logEvent("start", bundleOf(Pair(FirebaseAnalytics.Param.METHOD, tag)))
data.changeState(CONNECTING) data.changeState(CONNECTING)
data.connectingJob = GlobalScope.launch(Dispatchers.Main) {
GlobalScope.launch(Dispatchers.Main) {
try { try {
proxy.init() proxy.init()
data.udpFallback?.init() data.udpFallback?.init()
killProcesses() killProcesses()
data.processes = GuardedProcessPool() data.processes = GuardedProcessPool {
printLog(it)
data.connectingJob?.apply { runBlocking { cancelAndJoin() } }
stopRunner(false, it.localizedMessage)
}
startProcesses() startProcesses()
proxy.scheduleUpdate() proxy.scheduleUpdate()
...@@ -324,6 +329,8 @@ object BaseService { ...@@ -324,6 +329,8 @@ object BaseService {
printLog(exc) printLog(exc)
} }
stopRunner(false, "${getString(R.string.service_failed)}: ${exc.localizedMessage}") stopRunner(false, "${getString(R.string.service_failed)}: ${exc.localizedMessage}")
} finally {
data.connectingJob = null
} }
} }
return Service.START_NOT_STICKY return Service.START_NOT_STICKY
......
...@@ -38,7 +38,7 @@ import java.io.IOException ...@@ -38,7 +38,7 @@ import java.io.IOException
import java.io.InputStream import java.io.InputStream
import kotlin.concurrent.thread import kotlin.concurrent.thread
class GuardedProcessPool : CoroutineScope { class GuardedProcessPool(private val onFatal: (IOException) -> Unit) : CoroutineScope {
companion object { companion object {
private const val TAG = "GuardedProcessPool" private const val TAG = "GuardedProcessPool"
private val pid by lazy { private val pid by lazy {
...@@ -47,14 +47,12 @@ class GuardedProcessPool : CoroutineScope { ...@@ -47,14 +47,12 @@ class GuardedProcessPool : CoroutineScope {
} }
private inner class Guard(private val cmd: List<String>) { private inner class Guard(private val cmd: List<String>) {
val abortChannel = Channel<Unit>() private val abortChannel = Channel<Unit>()
private val exitChannel = Channel<Int>()
private val cmdName = File(cmd.first()).nameWithoutExtension
private var startTime: Long = -1 private var startTime: Long = -1
private lateinit var process: Process private lateinit var process: Process
private fun streamLogger(input: InputStream, logger: (String, String) -> Int) = try { private fun streamLogger(input: InputStream, logger: (String) -> Unit) = try {
input.bufferedReader().forEachLine { logger(TAG, it) } input.bufferedReader().forEachLine(logger)
} catch (_: IOException) { } // ignore } catch (_: IOException) { } // ignore
fun start() { fun start() {
...@@ -62,14 +60,20 @@ class GuardedProcessPool : CoroutineScope { ...@@ -62,14 +60,20 @@ class GuardedProcessPool : CoroutineScope {
process = ProcessBuilder(cmd).directory(Core.deviceStorage.noBackupFilesDir).start() process = ProcessBuilder(cmd).directory(Core.deviceStorage.noBackupFilesDir).start()
} }
suspend fun abort() {
if (!abortChannel.isClosedForSend) abortChannel.send(Unit)
}
suspend fun looper(onRestartCallback: (() -> Unit)?) { suspend fun looper(onRestartCallback: (() -> Unit)?) {
var running = true var running = true
val cmdName = File(cmd.first()).nameWithoutExtension
val exitChannel = Channel<Int>()
try { try {
while (true) { while (true) {
thread(name = "stderr-$cmdName") { streamLogger(process.errorStream, Log::e) } thread(name = "stderr-$cmdName") { streamLogger(process.errorStream) { Log.e(cmdName, it) } }
thread(name = "stdout-$cmdName") { thread(name = "stdout-$cmdName") {
runBlocking { runBlocking {
streamLogger(process.inputStream, Log::i) streamLogger(process.inputStream) { Log.i(cmdName, it) }
exitChannel.send(process.waitFor()) // this thread also acts as a daemon thread for waitFor exitChannel.send(process.waitFor()) // this thread also acts as a daemon thread for waitFor
} }
} }
...@@ -78,30 +82,31 @@ class GuardedProcessPool : CoroutineScope { ...@@ -78,30 +82,31 @@ class GuardedProcessPool : CoroutineScope {
exitChannel.onReceive { false } exitChannel.onReceive { false }
}) break }) break
running = false running = false
if (SystemClock.elapsedRealtime() - startTime < 1000) { if (SystemClock.elapsedRealtime() - startTime < 1000) throw IOException("$cmdName exits too fast")
Crashlytics.log(Log.WARN, TAG, "process exit too fast, stop guard: $cmdName")
break
}
Crashlytics.log(Log.DEBUG, TAG, "restart process: " + Commandline.toString(cmd)) Crashlytics.log(Log.DEBUG, TAG, "restart process: " + Commandline.toString(cmd))
start() start()
running = true running = true
onRestartCallback?.invoke() onRestartCallback?.invoke()
} }
} catch (e: IOException) {
Crashlytics.log(Log.WARN, TAG, "error occurred. stop guard: " + Commandline.toString(cmd))
// calling callback without closing channel first will cause deadlock, therefore we defer it
GlobalScope.launch(Dispatchers.Main) { onFatal(e) }
} finally { } finally {
if (!running) return // process already exited, nothing to be done abortChannel.close()
if (!running) return // process already exited, nothing to be done
if (Build.VERSION.SDK_INT < 24) { if (Build.VERSION.SDK_INT < 24) {
val pid = pid.get(process) as Int
try { try {
Os.kill(pid, OsConstants.SIGTERM) Os.kill(pid.get(process) as Int, OsConstants.SIGTERM)
} catch (e: ErrnoException) { } catch (e: ErrnoException) {
if (e.errno != OsConstants.ESRCH) throw e if (e.errno != OsConstants.ESRCH) throw e
} }
if (withTimeoutOrNull(500) { exitChannel.receive() } != null) return if (withTimeoutOrNull(500) { exitChannel.receive() } != null) return
} }
process.destroy() // kill the process process.destroy() // kill the process
if (Build.VERSION.SDK_INT >= 26) { if (Build.VERSION.SDK_INT >= 26) {
if (withTimeoutOrNull(1000) { exitChannel.receive() } != null) return if (withTimeoutOrNull(1000) { exitChannel.receive() } != null) return
process.destroyForcibly() // Force to kill the process if it's still alive process.destroyForcibly() // Force to kill the process if it's still alive
} }
exitChannel.receive() exitChannel.receive()
} }
...@@ -123,7 +128,7 @@ class GuardedProcessPool : CoroutineScope { ...@@ -123,7 +128,7 @@ class GuardedProcessPool : CoroutineScope {
@MainThread @MainThread
suspend fun close() { suspend fun close() {
guards.forEach { it.abortChannel.send(Unit) } guards.forEach { it.abort() }
supervisor.children.forEach { it.join() } supervisor.children.forEach { it.join() } // we can't cancel the supervisor as we need it to do clean up
} }
} }
...@@ -23,6 +23,7 @@ package com.github.shadowsocks.bg ...@@ -23,6 +23,7 @@ package com.github.shadowsocks.bg
import android.net.LocalServerSocket import android.net.LocalServerSocket
import android.net.LocalSocket import android.net.LocalSocket
import android.net.LocalSocketAddress import android.net.LocalSocketAddress
import android.system.ErrnoException
import android.system.Os import android.system.Os
import android.system.OsConstants import android.system.OsConstants
import com.github.shadowsocks.utils.printLog import com.github.shadowsocks.utils.printLog
...@@ -56,7 +57,11 @@ abstract class LocalSocketListener(name: String, socketFile: File) : Thread(name ...@@ -56,7 +57,11 @@ abstract class LocalSocketListener(name: String, socketFile: File) : Thread(name
override fun close() { override fun close() {
running = false running = false
// see also: https://issuetracker.google.com/issues/36945762#comment15 // see also: https://issuetracker.google.com/issues/36945762#comment15
Os.shutdown(localSocket.fileDescriptor, OsConstants.SHUT_RDWR) try {
Os.shutdown(localSocket.fileDescriptor, OsConstants.SHUT_RDWR)
} catch (e: ErrnoException) {
if (e.errno != OsConstants.EBADF) throw e // suppress fd already closed
}
join() 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