Commit 1def9e69 authored by Mygod's avatar Mygod

Finally we fix cancelling

parent 96dddba3
......@@ -31,10 +31,10 @@ import java.nio.ByteOrder
class TrafficMonitor(statFile: File) : AutoCloseable {
private val thread = object : LocalSocketListener("TrafficMonitor", statFile) {
private val buffer = ByteArray(16)
private val stat = ByteBuffer.wrap(buffer).order(ByteOrder.LITTLE_ENDIAN)
override fun acceptInternal(socket: LocalSocket) {
val buffer = ByteArray(16)
if (socket.inputStream.read(buffer) != 16) throw IOException("Unexpected traffic stat length")
val stat = ByteBuffer.wrap(buffer).order(ByteOrder.LITTLE_ENDIAN)
val tx = stat.getLong(0)
val rx = stat.getLong(8)
if (current.txTotal != tx) {
......
......@@ -20,6 +20,7 @@
package com.github.shadowsocks.net
import kotlinx.coroutines.suspendCancellableCoroutine
import java.nio.ByteBuffer
import java.nio.channels.Pipe
import java.nio.channels.SelectableChannel
......@@ -27,7 +28,6 @@ import java.nio.channels.SelectionKey
import java.nio.channels.Selector
import java.util.concurrent.ConcurrentLinkedQueue
import kotlin.coroutines.resume
import kotlin.coroutines.suspendCoroutine
class ChannelMonitor : Thread("ChannelMonitor"), AutoCloseable {
private val selector = Selector.open()
......@@ -60,10 +60,10 @@ class ChannelMonitor : Thread("ChannelMonitor"), AutoCloseable {
while (registrationPipe.sink().write(junk) == 0);
}
suspend fun wait(channel: SelectableChannel, ops: Int) = suspendCoroutine<Unit> { continuation ->
suspend fun wait(channel: SelectableChannel, ops: Int) = suspendCancellableCoroutine<Unit> { cont ->
register(channel, ops) {
it.interestOps(0) // stop listening
continuation.resume(Unit)
cont.resume(Unit)
}
}
......
......@@ -25,7 +25,6 @@ import com.github.shadowsocks.utils.printLog
import kotlinx.coroutines.*
import org.xbill.DNS.*
import java.io.IOException
import java.lang.RuntimeException
import java.net.*
import java.nio.ByteBuffer
import java.nio.channels.DatagramChannel
......@@ -63,7 +62,6 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd
private const val TTL = 120L
private const val UDP_PACKET_SIZE = 512
}
private val monitor = ChannelMonitor()
private val job = SupervisorJob()
......@@ -79,8 +77,8 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd
launch {
try {
val reply = resolve(buffer)
while (isActive && send(reply, source) <= 0) monitor.wait(this@apply, SelectionKey.OP_WRITE)
} catch (e: RuntimeException) {
while (send(reply, source) <= 0) monitor.wait(this@apply, SelectionKey.OP_WRITE)
} catch (e: Exception) {
printLog(e)
}
}
......@@ -144,8 +142,8 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd
channel.configureBlocking(false)
channel.connect(proxy)
val wrapped = remoteDns.tcpWrap(packet)
while (isActive && !channel.finishConnect()) monitor.wait(channel, SelectionKey.OP_CONNECT)
while (isActive && channel.write(wrapped) >= 0 && wrapped.hasRemaining()) {
while (!channel.finishConnect()) monitor.wait(channel, SelectionKey.OP_CONNECT)
while (channel.write(wrapped) >= 0 && wrapped.hasRemaining()) {
monitor.wait(channel, SelectionKey.OP_WRITE)
}
val result = remoteDns.tcpReceiveBuffer(UDP_PACKET_SIZE)
......@@ -161,7 +159,7 @@ class LocalDnsServer(private val localResolver: suspend (String) -> Array<InetAd
result.flip()
remoteDns.udpUnwrap(result)
result
}.also { Log.d("forward", "completed $it") }
}
}
}
......
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