Commit 490e0432 authored by Mygod's avatar Mygod

Fix ClosedChannelException

parent 00dfb925
......@@ -310,7 +310,7 @@ object BaseService {
data.changeState(CONNECTING)
data.connectingJob = GlobalScope.launch(Dispatchers.Main) {
try {
killProcesses()
Executable.killAll() // clean up old processes
preInit()
proxy.init(this@Interface::resolver)
data.udpFallback?.init(this@Interface::resolver)
......
......@@ -21,22 +21,28 @@
package com.github.shadowsocks.net
import com.github.shadowsocks.utils.printLog
import kotlinx.coroutines.suspendCancellableCoroutine
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import java.io.IOException
import java.lang.IllegalStateException
import java.nio.ByteBuffer
import java.nio.channels.*
import java.util.concurrent.ConcurrentLinkedQueue
import kotlin.coroutines.resume
import java.util.concurrent.Executors
class ChannelMonitor : Thread("ChannelMonitor"), AutoCloseable {
class ChannelMonitor {
private data class Registration(val channel: SelectableChannel,
val ops: Int,
val listener: suspend (SelectionKey) -> Unit) {
val result = CompletableDeferred<SelectionKey>()
}
private val job: Job
private val selector = Selector.open()
private val registrationPipe = Pipe.open()
private val pendingRegistrations = ConcurrentLinkedQueue<Triple<SelectableChannel, Int, (SelectionKey) -> Unit>>()
private val pendingRegistrations = Channel<Registration>()
@Volatile
private var running = true
private fun registerInternal(channel: SelectableChannel, ops: Int, block: (SelectionKey) -> Unit) =
private fun registerInternal(channel: SelectableChannel, ops: Int, block: suspend (SelectionKey) -> Unit) =
channel.register(selector, ops, block)
init {
......@@ -45,31 +51,18 @@ class ChannelMonitor : Thread("ChannelMonitor"), AutoCloseable {
registerInternal(this, SelectionKey.OP_READ) {
val junk = ByteBuffer.allocateDirect(1)
while (read(junk) > 0) {
val (channel, ops, block) = pendingRegistrations.remove()
registerInternal(channel, ops, block)
junk.clear()
}
}
pendingRegistrations.receive().apply {
try {
result.complete(registerInternal(channel, ops, listener))
} catch (e: ClosedChannelException) {
result.completeExceptionally(e)
}
start()
}
fun register(channel: SelectableChannel, ops: Int, block: (SelectionKey) -> Unit) {
pendingRegistrations.add(Triple(channel, ops, block))
val junk = ByteBuffer.allocateDirect(1)
while (running && registrationPipe.sink().write(junk) == 0);
junk.clear()
}
suspend fun wait(channel: SelectableChannel, ops: Int) = suspendCancellableCoroutine<Unit> { cont ->
register(channel, ops) {
if (it.isValid) it.interestOps(0) // stop listening
try {
cont.resume(Unit)
} catch (_: IllegalStateException) { } // already resumed by a timeout, maybe should use tryResume?
}
}
override fun run() {
job = GlobalScope.launch(Executors.newSingleThreadExecutor().asCoroutineDispatcher()) {
while (running) {
val num = try {
selector.select()
......@@ -82,15 +75,39 @@ class ChannelMonitor : Thread("ChannelMonitor"), AutoCloseable {
while (iterator.hasNext()) {
val key = iterator.next()
iterator.remove()
(key.attachment() as (SelectionKey) -> Unit)(key)
(key.attachment() as suspend (SelectionKey) -> Unit)(key)
}
}
}
}
suspend fun register(channel: SelectableChannel, ops: Int, block: suspend (SelectionKey) -> Unit): SelectionKey {
ByteBuffer.allocateDirect(1).also { junk ->
loop@ while (running) when (registrationPipe.sink().write(junk)) {
0 -> yield()
1 -> break@loop
else -> throw IOException("Failed to register in the channel")
}
}
if (!running) throw ClosedChannelException()
return Registration(channel, ops, block).run {
pendingRegistrations.send(this)
result.await()
}
}
suspend fun wait(channel: SelectableChannel, ops: Int) = CompletableDeferred<SelectionKey>().run {
register(channel, ops) {
if (it.isValid) it.interestOps(0) // stop listening
complete(it)
}
await()
}
override fun close() {
suspend fun close() {
running = false
selector.wakeup()
join()
job.join()
selector.keys().forEach { it.channel().close() }
selector.close()
}
......
......@@ -67,14 +67,14 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd
private val job = SupervisorJob()
override val coroutineContext = Dispatchers.Default + job + CoroutineExceptionHandler { _, t -> printLog(t) }
fun start(listen: SocketAddress) = DatagramChannel.open().apply {
suspend fun start(listen: SocketAddress) = DatagramChannel.open().apply {
configureBlocking(false)
socket().bind(listen)
monitor.register(this, SelectionKey.OP_READ) {
val buffer = ByteBuffer.allocateDirect(UDP_PACKET_SIZE)
val source = receive(buffer)!!
buffer.flip()
launch {
this@LocalDnsServer.launch {
val reply = resolve(buffer)
while (send(reply, source) <= 0) monitor.wait(this@apply, SelectionKey.OP_WRITE)
}
......
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