renminghui/punchnet-android
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changed the CoroutineScope from CoroutineScope(Dispatchers.IO) to scope

Browse Source
This commit is contained in:
asxalex 2025-07-14 15:25:28 +08:00
parent f24dd7cfe9
commit 678e9de066
7 changed files with 95 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.idea/.name generated
View File

@ -1 +0,0 @@
punchnet

37
app/src/main/java/com/jihe/punchnet/PunchnetService.kt
View File

@ -31,6 +31,7 @@ import com.jihe.punchnet.sdlan.network.run_sdlan
import com.jihe.punchnet.sdlan.utils.ipToString import com.jihe.punchnet.sdlan.utils.ipToString
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.cancel import kotlinx.coroutines.cancel
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext import kotlinx.coroutines.withContext
@ -49,7 +50,7 @@ class PunchnetService : VpnService() , IfaceTun {
var input: FileInputStream? = null var input: FileInputStream? = null
var output: FileOutputStream? = null var output: FileOutputStream? = null
override val arpTable = ARPTable() override val arpTable = ARPTable(scope)
override val arpWaitList = ARPWaitList() override val arpWaitList = ARPWaitList()
var config: DeviceConfig = DeviceConfig(1400) var config: DeviceConfig = DeviceConfig(1400)
@ -62,18 +63,22 @@ class PunchnetService : VpnService() , IfaceTun {
override suspend fun recv(): ByteArray { override suspend fun recv(): ByteArray {
val result = withContext(Dispatchers.IO) { val result = withContext(Dispatchers.IO) {
val result = ByteArray(1500) val result = ByteArray(1500)
var size = input?.read(result) try {
if (size == null) { var size = input?.read(result)
println("xxx failed to read") if (size == null) {
size = 0 println("xxx failed to read")
} else { size = 0
println("xxx got $size bytes") } else {
} println("xxx got $size bytes")
}
// val size = input?.read(result)?:0 // val size = input?.read(result)?:0
// Log.d(TAG, "RECEIVED $size bytes") // Log.d(TAG, "RECEIVED $size bytes")
result.copyOf(size) result.copyOf(size)
// result.slice(0..<size).toByteArray() // result.slice(0..<size).toByteArray()
} catch (e: Exception) {
ByteArray(0)
}
} }
return result return result
} }
@ -81,7 +86,11 @@ class PunchnetService : VpnService() , IfaceTun {
override suspend fun send(content: ByteArray) { override suspend fun send(content: ByteArray) {
withContext(Dispatchers.IO) { withContext(Dispatchers.IO) {
// Log.d(TAG, "WROTE bytes to vpn service") // Log.d(TAG, "WROTE bytes to vpn service")
output?.write(content) try {
output?.write(content)
} catch (e: Exception) {
TerminalLogger.errorf { "failed to write $e" }
}
} }
} }
@ -159,7 +168,7 @@ class PunchnetService : VpnService() , IfaceTun {
) )
scope.launch { scope.launch {
run_sdlan(iface, argument, startArg) run_sdlan(scope, iface, argument, startArg)
} }
val notification = createNotification() val notification = createNotification()

4
app/src/main/java/com/jihe/punchnet/sdlan/network/arp.kt
View File

@ -16,13 +16,15 @@ const val BroadcastIP = 0xFFFFFFFF
val BroadcastMac = ubyteArrayOf(0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu).toByteArray() val BroadcastMac = ubyteArrayOf(0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu).toByteArray()
class ARPTable( class ARPTable(
val scope: CoroutineScope,
) { ) {
val routeTable: RouteTable = RouteTable() val routeTable: RouteTable = RouteTable()
val content = ConcurrentHashMap<Int, ARPInfo>() val content = ConcurrentHashMap<Int, ARPInfo>()
suspend fun agingARP() { suspend fun agingARP() {
CoroutineScope(Dispatchers.Default).async { // CoroutineScope(Dispatchers.Default).async {
scope.async {
while(true) { while(true) {
delay(20_000) delay(20_000)
val now = System.currentTimeMillis()/1000 val now = System.currentTimeMillis()/1000

5
app/src/main/java/com/jihe/punchnet/sdlan/network/helper.kt
View File

@ -15,11 +15,13 @@ import com.jihe.punchnet.sdlan.utils.macToString
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async import kotlinx.coroutines.async
import kotlinx.coroutines.launch
import java.nio.ByteBuffer import java.nio.ByteBuffer
import java.nio.ByteOrder import java.nio.ByteOrder
suspend fun handlePacketData( suspend fun handlePacketData(
scope: CoroutineScope,
node: Node, node: Node,
body: ByteBuffer, body: ByteBuffer,
senderSock: SDLanSock senderSock: SDLanSock
@ -36,7 +38,8 @@ suspend fun handlePacketData(
TerminalLogger.debugf { TerminalLogger.debugf {
"[P2P] Rx data from ${senderSock}" "[P2P] Rx data from ${senderSock}"
} }
CoroutineScope(Dispatchers.Default).async { // CoroutineScope(Dispatchers.Default).async {
scope.async {
checkPeerRegistrationNeeded( checkPeerRegistrationNeeded(
node, node,
false, false,

3
app/src/main/java/com/jihe/punchnet/sdlan/network/iface.kt
View File

@ -11,6 +11,7 @@ import com.jihe.punchnet.sdlan.utils.EthHdr
import com.jihe.punchnet.sdlan.utils.ipToString import com.jihe.punchnet.sdlan.utils.ipToString
import com.jihe.punchnet.sdlan.utils.isMultiBroadcast import com.jihe.punchnet.sdlan.utils.isMultiBroadcast
import com.jihe.punchnet.sdlan.utils.macToString import com.jihe.punchnet.sdlan.utils.macToString
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext import kotlinx.coroutines.withContext
import java.io.BufferedReader import java.io.BufferedReader
@ -216,7 +217,7 @@ interface IfaceTun: Iface {
class IfaceMock: IfaceTun { class IfaceMock: IfaceTun {
val deviceName = "dev0" val deviceName = "dev0"
override val arpTable = ARPTable() override val arpTable = ARPTable(CoroutineScope(Dispatchers.IO))
override val arpWaitList = ARPWaitList() override val arpWaitList = ARPWaitList()
val sock = Socket(Inet4Address.getByName("127.0.0.1"), 1234) val sock = Socket(Inet4Address.getByName("127.0.0.1"), 1234)

86
app/src/main/java/com/jihe/punchnet/sdlan/network/libs.kt
View File

@ -26,6 +26,7 @@ import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.SendChannel import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.selects.select import kotlinx.coroutines.selects.select
import kotlinx.coroutines.withContext import kotlinx.coroutines.withContext
import java.io.DataInputStream import java.io.DataInputStream
@ -40,7 +41,7 @@ import java.util.concurrent.atomic.AtomicLong
import kotlin.io.path.pathString import kotlin.io.path.pathString
import kotlin.system.exitProcess import kotlin.system.exitProcess
suspend fun onMessage(data: SDLanTCP) { suspend fun onMessage(scope: CoroutineScope, data: SDLanTCP) {
val node = Node.getInstance() val node = Node.getInstance()
TerminalLogger.debugf {"message received"} TerminalLogger.debugf {"message received"}
@ -70,10 +71,16 @@ suspend fun onMessage(data: SDLanTCP) {
// println("got aes key: ${aeskey.toList()}, length is ${aeskey.size}") // println("got aes key: ${aeskey.toList()}, length is ${aeskey.size}")
node.sendStunRequest() node.sendStunRequest()
scope.async {
node.probeNatType()
TerminalLogger.debugf { "nat type is ${node.nat_type}"}
}
/*
CoroutineScope(Dispatchers.Default).async { CoroutineScope(Dispatchers.Default).async {
node.probeNatType() node.probeNatType()
TerminalLogger.debugf { "nat type is ${node.nat_type}"} TerminalLogger.debugf { "nat type is ${node.nat_type}"}
} }
*/
} }
PacketType.RegisterSuperNAK -> { PacketType.RegisterSuperNAK -> {
println("21") println("21")
@ -126,7 +133,7 @@ suspend fun onMessage(data: SDLanTCP) {
} }
suspend fun run_sdlan(iface: Iface, argument: Arguments, routeinfo: PunchnetServiceArgument?) { suspend fun run_sdlan(scope: CoroutineScope, iface: Iface, argument: Arguments, routeinfo: PunchnetServiceArgument?) {
UniqueNodeID.setBaseDir(argument.baseDir) UniqueNodeID.setBaseDir(argument.baseDir)
val edgeUUID = UniqueNodeID.getUUID() val edgeUUID = UniqueNodeID.getUUID()
val config = parseConfig(edgeUUID, argument) val config = parseConfig(edgeUUID, argument)
@ -140,7 +147,7 @@ suspend fun run_sdlan(iface: Iface, argument: Arguments, routeinfo: PunchnetServ
val toSocket = Channel<ByteArray>(100) val toSocket = Channel<ByteArray>(100)
val start_stop_channel = Channel<StartStopChanInfo>(100) val start_stop_channel = Channel<StartStopChanInfo>(100)
initEdge(iface, argument.token, config, toSocket, start_stop_channel) initEdge(scope, iface, argument.token, config, toSocket, start_stop_channel)
val tcp = argument.tcp.split(":") val tcp = argument.tcp.split(":")
@ -191,6 +198,27 @@ suspend fun run_sdlan(iface: Iface, argument: Arguments, routeinfo: PunchnetServ
} }
*/ */
scope.launch {
while(true) {
val data = node.iface?.recv()
Log.d("SDLAN", "async receive data from iface: ${data?.size} bytes")
if (data == null) {
//delay(1000)
println("got data is null")
continue
}
if (data.isEmpty()) {
println("got data size 0")
// delay(1000)
continue
}
Log.d("SDLAN", "handle data form device starts")
node.iface?.handleDataFromDevice(node, data)
Log.d("SDLAN", "handle data form device stops")
}
}
/*
CoroutineScope((Dispatchers.IO)).async { CoroutineScope((Dispatchers.IO)).async {
while(true) { while(true) {
val data = node.iface?.recv() val data = node.iface?.recv()
@ -210,9 +238,12 @@ suspend fun run_sdlan(iface: Iface, argument: Arguments, routeinfo: PunchnetServ
Log.d("SDLAN", "handle data form device stops") Log.d("SDLAN", "handle data form device stops")
} }
} }
*/
CoroutineScope(Dispatchers.IO).async { // CoroutineScope(Dispatchers.IO).async {
scope.launch {
initTCPConn( initTCPConn(
scope,
tcp[0], tcp[1].toInt(), tcp[0], tcp[1].toInt(),
start_stop_channel, start_stop_channel,
AtomicLong(now), AtomicLong(now),
@ -231,7 +262,7 @@ suspend fun run_sdlan(iface: Iface, argument: Arguments, routeinfo: PunchnetServ
start_stop_channel.send(StartStopChanInfo(StartStopFlag.IsStart, null)) start_stop_channel.send(StartStopChanInfo(StartStopFlag.IsStart, null))
val cancel = Channel<Boolean>(100) val cancel = Channel<Boolean>(100)
runEdgeLoop(node, cancel) runEdgeLoop(scope, node, cancel)
while(true) { while(true) {
TerminalLogger.debugf {"ping to sn"} TerminalLogger.debugf {"ping to sn"}
@ -240,24 +271,24 @@ suspend fun run_sdlan(iface: Iface, argument: Arguments, routeinfo: PunchnetServ
} }
} }
suspend fun runEdgeLoop(node: Node, cancel: ReceiveChannel<Boolean>) { suspend fun runEdgeLoop(scope: CoroutineScope, node: Node, cancel: ReceiveChannel<Boolean>) {
node.ping_to_sn() node.ping_to_sn()
CoroutineScope(Dispatchers.Default).async { scope.launch {
loopSocketV4(node, node.udpSockV4, cancel) loopSocketV4(scope, node, node.udpSockV4, cancel)
} }
} }
suspend fun loopSocketV4(node: Node, sock: SDLanSocket, cancel: ReceiveChannel<Boolean>) { suspend fun loopSocketV4(scope: CoroutineScope, node: Node, sock: SDLanSocket, cancel: ReceiveChannel<Boolean>) {
val job_stun_request = CoroutineScope(Dispatchers.Default).async { val job_stun_request = scope.async {
while(true) { while(true) {
delay(10_000) delay(10_000)
node.sendStunRequest() node.sendStunRequest()
} }
} }
val job_handle_packet = CoroutineScope(Dispatchers.Default).async { val job_handle_packet = scope.async {
while(true) { while(true) {
readAndParsePacket(node, sock) readAndParsePacket(scope, node, sock)
} }
} }
@ -276,7 +307,7 @@ suspend fun loopSocketV4(node: Node, sock: SDLanSocket, cancel: ReceiveChannel<B
job_handle_packet.cancelAndJoin() job_handle_packet.cancelAndJoin()
} }
suspend fun readAndParsePacket(node: Node, sock: SDLanSocket) { suspend fun readAndParsePacket(scope: CoroutineScope, node: Node, sock: SDLanSocket) {
val packet = sock.receive() val packet = sock.receive()
val from = packet.socketAddress val from = packet.socketAddress
@ -286,10 +317,10 @@ suspend fun readAndParsePacket(node: Node, sock: SDLanSocket) {
} }
val data = ByteBuffer.wrap(packet.data, 0, packet.length) val data = ByteBuffer.wrap(packet.data, 0, packet.length)
// val data = packet.data.toByteString(0, packet.length) // val data = packet.data.toByteString(0, packet.length)
handleAPacket(node, from, data) handleAPacket(scope, node, from, data)
} }
suspend fun handleAPacket(node: Node, from: SocketAddress, data: ByteBuffer) { suspend fun handleAPacket(scope: CoroutineScope, node: Node, from: SocketAddress, data: ByteBuffer) {
val pktType = PacketType.fromValue(data.get().toUByte()) val pktType = PacketType.fromValue(data.get().toUByte())
if (pktType == null) { if (pktType == null) {
TerminalLogger.errorf { "invalid packet type" } TerminalLogger.errorf { "invalid packet type" }
@ -307,7 +338,7 @@ suspend fun handleAPacket(node: Node, from: SocketAddress, data: ByteBuffer) {
if (from is InetSocketAddress) { if (from is InetSocketAddress) {
TerminalLogger.debugf {"got data"} TerminalLogger.debugf {"got data"}
val sock = SDLanSock(IPFamily.IPV4, from.port, from.address.address) val sock = SDLanSock(IPFamily.IPV4, from.port, from.address.address)
handlePacketData(node, data, sock) handlePacketData(scope, node, data, sock)
} }
} }
PacketType.StunProbeReply -> { PacketType.StunProbeReply -> {
@ -347,18 +378,18 @@ suspend fun handleAPacket(node: Node, from: SocketAddress, data: ByteBuffer) {
} }
fun initEdge(iface: Iface, token: String, config: NodeConfig, toSocket: SendChannel<ByteArray>, startStopChannel: SendChannel<StartStopChanInfo>) { fun initEdge(scope: CoroutineScope, iface: Iface, token: String, config: NodeConfig, toSocket: SendChannel<ByteArray>, startStopChannel: SendChannel<StartStopChanInfo>) {
val rsa = RSA.getRSA() val rsa = RSA.getRSA()
val pathname = Paths.get(config.baseDir, RSAConfig.BASE_DIR).pathString val pathname = Paths.get(config.baseDir, RSAConfig.BASE_DIR).pathString
Log.d("DIR", "pathname = $pathname") Log.d("DIR", "pathname = $pathname")
rsa.generateKeyPair(pathname) rsa.generateKeyPair(pathname)
//rsa.generateKeyPair(Path.of(config.baseDir, RSAConfig.BASE_DIR).name) //rsa.generateKeyPair(Path.of(config.baseDir, RSAConfig.BASE_DIR).name)
val sockV4 = SDLanSocket("0.0.0.0", config.localPort) val sockV4 = SDLanSocket(scope, "0.0.0.0", config.localPort)
var sockMulticast: SDLanSocket? = null var sockMulticast: SDLanSocket? = null
if (!config.dropMulticast) { if (!config.dropMulticast) {
sockMulticast = SDLanSocket(SDLanMulticastConfig.MULTICAST_V4.toIPV4String(), SDLanMulticastConfig.MULTICAST_PORT) sockMulticast = SDLanSocket(scope, SDLanMulticastConfig.MULTICAST_V4.toIPV4String(), SDLanMulticastConfig.MULTICAST_PORT)
} }
Node.initialize( Node.initialize(
@ -376,6 +407,7 @@ fun initEdge(iface: Iface, token: String, config: NodeConfig, toSocket: SendChan
} }
suspend fun initTCPConn( suspend fun initTCPConn(
scope: CoroutineScope,
tcpHost: String, tcpHost: String,
tcpPort: Int, tcpPort: Int,
start_stop: Channel<StartStopChanInfo>, start_stop: Channel<StartStopChanInfo>,
@ -383,7 +415,7 @@ suspend fun initTCPConn(
connected: AtomicBoolean, connected: AtomicBoolean,
toSocket: ReceiveChannel<ByteArray>, toSocket: ReceiveChannel<ByteArray>,
onConnected: suspend (stream: Socket, pktID: Int?)->Unit, onConnected: suspend (stream: Socket, pktID: Int?)->Unit,
onMessage: suspend (SDLanTCP)->Unit, onMessage: suspend (CoroutineScope, SDLanTCP)->Unit,
onDisconnected: suspend ()->Unit, onDisconnected: suspend ()->Unit,
connectingChan: SendChannel<ConnectingState>? connectingChan: SendChannel<ConnectingState>?
) { ) {
@ -426,7 +458,8 @@ suspend fun initTCPConn(
val outIP = ByteBuffer.wrap(socket.localAddress.address).getInt() val outIP = ByteBuffer.wrap(socket.localAddress.address).getInt()
node.outerIPV4.set(outIP) node.outerIPV4.set(outIP)
val job_read_packet = CoroutineScope(Dispatchers.IO).async { // val job_read_packet = CoroutineScope(Dispatchers.IO).async {
val job_read_packet = scope.async {
val input = DataInputStream(socket.getInputStream()) val input = DataInputStream(socket.getInputStream())
try { try {
println("job read packet starts") println("job read packet starts")
@ -436,7 +469,7 @@ suspend fun initTCPConn(
TerminalLogger.errorf {"tcp Packet is null"} TerminalLogger.errorf {"tcp Packet is null"}
break break
} }
onMessage(tcpPacket) onMessage(scope, tcpPacket)
} }
} finally { } finally {
TerminalLogger.errorf {"input closing"} TerminalLogger.errorf {"input closing"}
@ -444,7 +477,8 @@ suspend fun initTCPConn(
} }
} }
val job_write_to_packet = CoroutineScope(Dispatchers.IO).async { // val job_write_to_packet = CoroutineScope(Dispatchers.IO).async {
val job_write_to_packet = scope.async {
val output = DataOutputStream(socket.getOutputStream()) val output = DataOutputStream(socket.getOutputStream())
try { try {
TerminalLogger.debugf {"job write to packet starts"} TerminalLogger.debugf {"job write to packet starts"}
@ -464,7 +498,8 @@ suspend fun initTCPConn(
} }
} }
val job_check_pong = CoroutineScope(Dispatchers.IO).async { // val job_check_pong = CoroutineScope(Dispatchers.IO).async {
val job_check_pong = scope.async {
println("job check pong starts") println("job check pong starts")
while(true) { while(true) {
delay(10_000) delay(10_000)
@ -476,7 +511,8 @@ suspend fun initTCPConn(
} }
} }
val job_check_stop = CoroutineScope(Dispatchers.IO).async { // val job_check_stop = CoroutineScope(Dispatchers.IO).async {
val job_check_stop = scope.async {
println("job check stop starts") println("job check stop starts")
while(true) { while(true) {
try { try {

4
app/src/main/java/com/jihe/punchnet/sdlan/network/socket.kt
View File

@ -14,7 +14,7 @@ import java.net.DatagramSocket
import java.net.Inet4Address import java.net.Inet4Address
import java.net.SocketAddress import java.net.SocketAddress
class SDLanSocket(val addr: String, val port: Int, val reuseAddress: Boolean = false) { class SDLanSocket(val scope: CoroutineScope, val addr: String, val port: Int, val reuseAddress: Boolean = false) {
private val connection: DatagramSocket = run { private val connection: DatagramSocket = run {
val sock = DatagramSocket(port, Inet4Address.getByName(addr)) val sock = DatagramSocket(port, Inet4Address.getByName(addr))
if (reuseAddress) { if (reuseAddress) {
@ -27,7 +27,7 @@ class SDLanSocket(val addr: String, val port: Int, val reuseAddress: Boolean = f
fun loop(): ReceiveChannel<DatagramPacket> { fun loop(): ReceiveChannel<DatagramPacket> {
val channel = Channel<DatagramPacket>(100) val channel = Channel<DatagramPacket>(100)
job = CoroutineScope(Dispatchers.Default).launch { scope.launch {
while (true) { while (true) {
val msg = receive() val msg = receive()
channel.send(msg) channel.send(msg)