Commit 6b9f2245 authored by Mygod's avatar Mygod

Join threads non-blockingly

parent 00376c2e
...@@ -261,7 +261,7 @@ object BaseService { ...@@ -261,7 +261,7 @@ object BaseService {
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.shutdown(this)
it.profile.id it.profile.id
} }
data.proxy = null data.proxy = null
......
...@@ -45,7 +45,7 @@ import java.security.MessageDigest ...@@ -45,7 +45,7 @@ import java.security.MessageDigest
/** /**
* This class sets up environment for ss-local. * This class sets up environment for ss-local.
*/ */
class ProxyInstance(val profile: Profile, private val route: String = profile.route) : AutoCloseable { class ProxyInstance(val profile: Profile, private val route: String = profile.route) {
private var configFile: File? = null private var configFile: File? = null
var trafficMonitor: TrafficMonitor? = null var trafficMonitor: TrafficMonitor? = null
private val plugin = PluginConfiguration(profile.plugin ?: "").selectedOptions private val plugin = PluginConfiguration(profile.plugin ?: "").selectedOptions
...@@ -125,9 +125,9 @@ class ProxyInstance(val profile: Profile, private val route: String = profile.ro ...@@ -125,9 +125,9 @@ class ProxyInstance(val profile: Profile, private val route: String = profile.ro
if (route !in arrayOf(Acl.ALL, Acl.CUSTOM_RULES)) AclSyncer.schedule(route) if (route !in arrayOf(Acl.ALL, Acl.CUSTOM_RULES)) AclSyncer.schedule(route)
} }
override fun close() { fun shutdown(scope: CoroutineScope) {
trafficMonitor?.apply { trafficMonitor?.apply {
close() thread.shutdown(scope)
// Make sure update total traffic when stopping the runner // Make sure update total traffic when stopping the runner
try { try {
// profile may have host, etc. modified and thus a re-fetch is necessary (possible race condition) // profile may have host, etc. modified and thus a re-fetch is necessary (possible race condition)
......
...@@ -29,8 +29,8 @@ import java.io.IOException ...@@ -29,8 +29,8 @@ import java.io.IOException
import java.nio.ByteBuffer import java.nio.ByteBuffer
import java.nio.ByteOrder import java.nio.ByteOrder
class TrafficMonitor(statFile: File) : AutoCloseable { class TrafficMonitor(statFile: File) {
private val thread = object : LocalSocketListener("TrafficMonitor-" + statFile.name, statFile) { val thread = object : LocalSocketListener("TrafficMonitor-" + statFile.name, statFile) {
private val buffer = ByteArray(16) private val buffer = ByteArray(16)
private val stat = ByteBuffer.wrap(buffer).order(ByteOrder.LITTLE_ENDIAN) private val stat = ByteBuffer.wrap(buffer).order(ByteOrder.LITTLE_ENDIAN)
override fun acceptInternal(socket: LocalSocket) { override fun acceptInternal(socket: LocalSocket) {
...@@ -79,6 +79,4 @@ class TrafficMonitor(statFile: File) : AutoCloseable { ...@@ -79,6 +79,4 @@ class TrafficMonitor(statFile: File) : AutoCloseable {
} }
return Pair(out, updated) return Pair(out, updated)
} }
override fun close() = thread.close()
} }
...@@ -23,6 +23,7 @@ package com.github.shadowsocks.net ...@@ -23,6 +23,7 @@ package com.github.shadowsocks.net
import com.github.shadowsocks.utils.printLog import com.github.shadowsocks.utils.printLog
import kotlinx.coroutines.* import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.sendBlocking
import java.io.IOException import java.io.IOException
import java.nio.ByteBuffer import java.nio.ByteBuffer
import java.nio.channels.* import java.nio.channels.*
...@@ -37,6 +38,7 @@ class ChannelMonitor : Thread("ChannelMonitor") { ...@@ -37,6 +38,7 @@ class ChannelMonitor : Thread("ChannelMonitor") {
private val selector = Selector.open() private val selector = Selector.open()
private val registrationPipe = Pipe.open() private val registrationPipe = Pipe.open()
private val pendingRegistrations = Channel<Registration>(Channel.UNLIMITED) private val pendingRegistrations = Channel<Registration>(Channel.UNLIMITED)
private val closeChannel = Channel<Unit>(1)
@Volatile @Volatile
private var running = true private var running = true
...@@ -101,13 +103,14 @@ class ChannelMonitor : Thread("ChannelMonitor") { ...@@ -101,13 +103,14 @@ class ChannelMonitor : Thread("ChannelMonitor") {
(key.attachment() as (SelectionKey) -> Unit)(key) (key.attachment() as (SelectionKey) -> Unit)(key)
} }
} }
closeChannel.sendBlocking(Unit)
} }
fun close(scope: CoroutineScope) { fun close(scope: CoroutineScope) {
running = false running = false
selector.wakeup() selector.wakeup()
scope.launch(Dispatchers.IO) { // thread joining is a blocking operation scope.launch {
join() closeChannel.receive()
selector.keys().forEach { it.channel().close() } selector.keys().forEach { it.channel().close() }
selector.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) }
} }
fun shutdown(scope: CoroutineScope) { override fun shutdown(scope: CoroutineScope) {
running = false running = false
job.cancel() job.cancel()
close() super.shutdown(scope)
scope.launch { job.join() } scope.launch { job.join() }
} }
} }
...@@ -23,31 +23,57 @@ package com.github.shadowsocks.net ...@@ -23,31 +23,57 @@ package com.github.shadowsocks.net
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.OsConstants
import com.github.shadowsocks.utils.printLog import com.github.shadowsocks.utils.printLog
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.sendBlocking
import kotlinx.coroutines.launch
import java.io.File import java.io.File
import java.io.IOException import java.io.IOException
abstract class LocalSocketListener(name: String, socketFile: File) : SocketListener(name) { abstract class LocalSocketListener(name: String, socketFile: File) : Thread(name) {
private val localSocket = LocalSocket().apply { private val localSocket = LocalSocket().apply {
socketFile.delete() // It's a must-have to close and reuse previous local socket. socketFile.delete() // It's a must-have to close and reuse previous local socket.
bind(LocalSocketAddress(socketFile.absolutePath, LocalSocketAddress.Namespace.FILESYSTEM)) bind(LocalSocketAddress(socketFile.absolutePath, LocalSocketAddress.Namespace.FILESYSTEM))
} }
private val serverSocket = LocalServerSocket(localSocket.fileDescriptor) private val serverSocket = LocalServerSocket(localSocket.fileDescriptor)
override val fileDescriptor get() = localSocket.fileDescriptor private val closeChannel = Channel<Unit>(1)
@Volatile
protected var running = true
/** /**
* Inherited class do not need to close input/output streams as they will be closed automatically. * Inherited class do not need to close input/output streams as they will be closed automatically.
*/ */
protected open fun accept(socket: LocalSocket) = socket.use { acceptInternal(socket) } protected open fun accept(socket: LocalSocket) = socket.use { acceptInternal(socket) }
protected abstract fun acceptInternal(socket: LocalSocket) protected abstract fun acceptInternal(socket: LocalSocket)
final override fun run() = localSocket.use { final override fun run() {
while (running) { localSocket.use {
try { while (running) {
accept(serverSocket.accept()) try {
} catch (e: IOException) { accept(serverSocket.accept())
if (running) printLog(e) } catch (e: IOException) {
continue if (running) printLog(e)
continue
}
} }
} }
closeChannel.sendBlocking(Unit)
}
open fun shutdown(scope: CoroutineScope) {
running = false
localSocket.fileDescriptor.apply {
// see also: https://issuetracker.google.com/issues/36945762#comment15
if (valid()) try {
Os.shutdown(this, OsConstants.SHUT_RDWR)
} catch (e: ErrnoException) {
// suppress fd inactive or already closed
if (e.errno != OsConstants.EBADF && e.errno != OsConstants.ENOTCONN) throw IOException(e)
}
}
scope.launch { closeChannel.receive() }
} }
} }
package com.github.shadowsocks.net
import android.system.ErrnoException
import android.system.Os
import android.system.OsConstants
import java.io.FileDescriptor
import java.io.IOException
abstract class SocketListener(name: String) : Thread(name), AutoCloseable {
protected abstract val fileDescriptor: FileDescriptor
@Volatile
protected var running = true
private fun FileDescriptor.shutdown() {
// see also: https://issuetracker.google.com/issues/36945762#comment15
if (valid()) try {
Os.shutdown(this, OsConstants.SHUT_RDWR)
} catch (e: ErrnoException) {
// suppress fd inactive or already closed
if (e.errno != OsConstants.EBADF && e.errno != OsConstants.ENOTCONN) throw IOException(e)
}
}
override fun close() {
running = false
fileDescriptor.shutdown()
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