From 96edbb2040637d9d1d1940a3f73fecf378a41dd0 Mon Sep 17 00:00:00 2001
From: sergeych <real.sergeych@gmail.com>
Date: Wed, 15 Nov 2023 02:11:52 +0300
Subject: [PATCH] fixes, tcp/ip optimizations

---
 .idea/GitLink.xml                             |  6 ++
 .idea/gradle.xml                              |  1 +
 .../kiloparsec/adapter/InetProxyDevice.kt     | 11 +++
 .../kiloparsec/adapter/ProxyDevice.kt         |  2 +-
 .../kiloparsec/adapter/asyncSocketToDevice.kt | 70 +++++++++++--------
 .../kiloparsec/adapters/NetworkTest.kt        | 57 +++++++++------
 6 files changed, 97 insertions(+), 50 deletions(-)
 create mode 100644 .idea/GitLink.xml
 create mode 100644 src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/InetProxyDevice.kt

diff --git a/.idea/GitLink.xml b/.idea/GitLink.xml
new file mode 100644
index 0000000..5143819
--- /dev/null
+++ b/.idea/GitLink.xml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="uk.co.ben_gibson.git.link.SettingsState">
+    <option name="host" value="e0f86390-1091-4871-8aeb-f534fbc99cf0" />
+  </component>
+</project>
\ No newline at end of file
diff --git a/.idea/gradle.xml b/.idea/gradle.xml
index f9163b4..ce1c62c 100644
--- a/.idea/gradle.xml
+++ b/.idea/gradle.xml
@@ -1,5 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
+  <component name="GradleMigrationSettings" migrationVersion="1" />
   <component name="GradleSettings">
     <option name="linkedExternalProjectsSettings">
       <GradleProjectSettings>
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/InetProxyDevice.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/InetProxyDevice.kt
new file mode 100644
index 0000000..524f96d
--- /dev/null
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/InetProxyDevice.kt
@@ -0,0 +1,11 @@
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.channels.Channel
+
+@Suppress("unused")
+class InetProxyDevice(
+    inputChannel: Channel<UByteArray?>,
+    outputChannel: Channel<UByteArray>,
+    val remoteAddress: NetworkAddress,
+    onclose: ()->Unit = {}
+) : ProxyDevice(inputChannel, outputChannel, onclose)
\ No newline at end of file
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
index 914bc11..b4f7fe9 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
@@ -5,7 +5,7 @@ import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.SendChannel
 import net.sergeych.kiloparsec.Transport
 
-class ProxyDevice(
+open class ProxyDevice(
     inputChannel: Channel<UByteArray?>,
     outputChannel: Channel<UByteArray>,
     private val onClose: ()->Unit = {}): Transport.Device {
diff --git a/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/asyncSocketToDevice.kt b/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/asyncSocketToDevice.kt
index b9c1e2a..1963e05 100644
--- a/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/asyncSocketToDevice.kt
+++ b/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/asyncSocketToDevice.kt
@@ -3,13 +3,14 @@ package net.sergeych.kiloparsec.adapter
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
-import kotlinx.coroutines.channels.ClosedSendChannelException
+import net.sergeych.bintools.toDump
 import net.sergeych.crypto.encodeVarUnsigned
 import net.sergeych.crypto.readVarUnsigned
 import net.sergeych.kiloparsec.Transport
 import net.sergeych.mp_logger.LogTag
 import net.sergeych.mp_logger.warning
 import net.sergeych.mp_tools.globalLaunch
+import java.net.InetSocketAddress
 import java.net.StandardSocketOptions.TCP_NODELAY
 import java.nio.ByteBuffer
 import java.nio.channels.AsynchronousSocketChannel
@@ -18,52 +19,73 @@ import kotlin.coroutines.suspendCoroutine
 
 private val log = LogTag("ASTD")
 
+/**
+ * Prepend block with its size, varint-encoded
+ */
+private fun encode(block: UByteArray): ByteArray
+    = (encodeVarUnsigned(block.size.toUInt()) + block).toByteArray()
+
 /**
  * Convert asynchronous socket to a [Transport.Device] using non-blocking nio,
  * in a coroutine-effective manner. Note that it runs coroutines to read/write
  * to the socket in a global scope.These are closed when transport is closed
  *  or the socket is closed, for example, by network failure.
  */
-suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): Transport.Device {
-    val deferredDevice = CompletableDeferred<Transport.Device>()
+suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): InetProxyDevice {
+    val deferredDevice = CompletableDeferred<InetProxyDevice>()
     globalLaunch {
         coroutineScope {
             fun stop() {
                 cancel()
             }
             socket.setOption(TCP_NODELAY, true)
+            // socket input is to be parsed for blocks, so we receive bytes
+            // and decode them to blocks
             val input = Channel<UByte>(1024)
-            val output = Channel<UByte>(1024)
-            // copy from socket to input
+            // copy from socket to input:
             launch {
-                val inb = ByteBuffer.allocate(1024)
+                val data = ByteArray(1024)
+                val inb = ByteBuffer.wrap(data)
                 while (isActive) {
+                    inb.position(0)
                     val size: Int = suspendCoroutine { continuation ->
                         socket.read(inb, continuation, IntCompletionHandler)
                     }
                     if (size < 0) stop()
-                    else for (i in 0..<size) input.send(inb[i].toUByte().also { print(it.toInt().toChar()) })
+                    else {
+                        println("recvd:\n${data.sliceArray(0..<size).toDump()}\n------------------")
+                        for (i in 0..<size) input.send(data[i].toUByte())
+                    }
                 }
             }
-            // copy from output tp socket
+            // output is blocks, so we sent blocks:
+            val outputBlocks = Channel<UByteArray>()
+            // copy from output to socket:
             launch {
-                val outBuff = ArrayList<Byte>(1024)
                 try {
                     while (isActive) {
-                        outBuff.clear()
-                        outBuff.add(output.receive().toByte())
-                        while (!output.isEmpty)
-                            outBuff.add(output.receive().toByte())
-                        suspendCoroutine { continuation ->
-                            socket.write(ByteBuffer.wrap(outBuff.toByteArray()), continuation, IntCompletionHandler)
+                        // wait for the first block to send
+                        var data = encode(outputBlocks.receive())
+                        // if there are more, take them all (NO_DELAY optimization)
+                        while (!outputBlocks.isEmpty)
+                            data += encode(outputBlocks.receive())
+                        // now send the aggregate:
+                        val outBuff = ByteBuffer.wrap(data)
+                        val cnt = suspendCoroutine { continuation ->
+                            socket.write(outBuff, continuation, IntCompletionHandler)
+                        }
+                        // be sure it was all sent
+                        if( outBuff.position() != data.size || cnt != data.size) {
+                            throw RuntimeException("PArtial write!")
                         }
                     }
                 } catch (_: ClosedReceiveChannelException) {
                     stop()
                 }
             }
-            // pump blocks from socket output to device input
+            // transport device copes with blocks:
             val inputBlocks = Channel<UByteArray?>()
+            // decode blocks from a byte channel read from the socket:
             launch {
                 try {
                     while (isActive) {
@@ -84,20 +106,10 @@ suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): Transport.De
                     stop()
                 }
             }
-            val outputBlocks = Channel<UByteArray>()
-            launch {
-                try {
-                    while (isActive) {
-                        val block = outputBlocks.receive()
-                        output.sendAll(encodeVarUnsigned(block.size.toUInt()))
-                        output.sendAll(block)
-                    }
-                } catch (_: ClosedSendChannelException) {
-                    stop()
-                }
-            }
+//            SocketAddress.
+            val addr = socket.remoteAddress as InetSocketAddress
             deferredDevice.complete(
-                ProxyDevice(inputBlocks, outputBlocks) { stop() }
+                InetProxyDevice(inputBlocks, outputBlocks, JvmNetworkAddress(addr.address,addr.port)) { stop() }
             )
         }
         globalLaunch { socket.close() }
diff --git a/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapters/NetworkTest.kt b/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapters/NetworkTest.kt
index d68e1a4..ce2e751 100644
--- a/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapters/NetworkTest.kt
+++ b/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapters/NetworkTest.kt
@@ -37,31 +37,48 @@ class NetworkTest {
             val serverFlow = acceptTcpDevice(17171)
             val j = launch {
                 serverFlow.collect { device ->
-                    device.output.send("Hello, world!".encodeToUByteArray())
-                    device.output.send("Great".encodeToUByteArray())
-                    while(true) {
-                        val x = device.input.receive()?.decodeFromUByteArray() ?: break
-                        if( x== "Goodbye" ) break
-                        if( x == "die") {
-                            cancel()
-                            break
+                    launch {
+                        println("connected!")
+                        device.output.send("Hello, world!".encodeToUByteArray())
+                        device.output.send("Great".encodeToUByteArray())
+                        while (true) {
+                            val x = device.input.receive()?.decodeFromUByteArray() ?: break
+                            if (x == "Goodbye") break
+                            if (x == "die") {
+                                println("collector get poisoned pill")
+                                cancel()
+                                break
+                            }
+                            println("ignoring unexpected input: $x")
                         }
-                        println("ignoring unexpected input: $x")
                     }
                 }
             }
             yield()
-            var s = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
-            assertEquals("Hello, world!", s.input.receive()!!.decodeFromUByteArray())
-            assertEquals("Great", s.input.receive()!!.decodeFromUByteArray())
-            s.output.send("Goodbye".encodeToUByteArray())
-            s.close()
-
-            s = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
-            assertEquals("Hello, world!", s.input.receive()!!.decodeFromUByteArray())
-            assertEquals("Great", s.input.receive()!!.decodeFromUByteArray())
-//            s.output.send("die".encodeToUByteArray())
-            s.close()
+            run {
+                println("x0")
+                val s = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
+                println("x1")
+                assertEquals("Hello, world!", s.input.receive()!!.decodeFromUByteArray())
+                println("x2")
+                assertEquals("Great", s.input.receive()!!.decodeFromUByteArray())
+                println("x3")
+                s.output.send("Goodbye".encodeToUByteArray())
+                println("pre1")
+                s.close()
+                println("pre2")
+            }
+            val s1 = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
+            println("conn-0-1")
+            assertEquals("Hello, world!", s1.input.receive()!!.decodeFromUByteArray())
+            println("conn-0-2")
+            assertEquals("Great", s1.input.receive()!!.decodeFromUByteArray())
+            println("1")
+            s1.output.send("die".encodeToUByteArray())
+            println("2")
+            delay(200)
+            s1.close()
+            println("3 -- the -- end")
             j.cancelAndJoin()
         }
     }