KtorSocketPerformance.kt

/*
 * Copyright (C) 2021 Anton Malinskiy
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.malinskiy.adam

import com.malinskiy.adam.transport.withDefaultBuffer
import io.ktor.network.selector.*
import io.ktor.network.sockets.*
import io.ktor.utils.io.*
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.runBlocking
import org.junit.AfterClass
import org.junit.BeforeClass
import org.junit.Test
import java.net.ServerSocket
import java.net.Socket
import java.nio.ByteBuffer
import java.util.concurrent.Executors
import kotlin.system.measureTimeMillis

class KtorSocketPerformance {

    companion object {
        private val executor = Executors.newSingleThreadExecutor()
        private lateinit var serverSocket: ServerSocket

        @BeforeClass
        @JvmStatic
        fun setup() {
            serverSocket = ServerSocket(4200)
            executor.submit {
                val buffer = ByteArray(4096)
                while (!Thread.interrupted()) {
                    serverSocket.accept().use { socket ->
                        socket.getInputStream().use { stream ->
                            while (true) {
                                val read = stream.read(buffer)
                                if (read == -1) {
                                    socket.close()
                                    return@use
                                }
                            }
                        }
                    }
                }
            }
        }

        @AfterClass
        @JvmStatic
        fun cleanup() {
            serverSocket.close()
            executor.shutdown()
        }
    }

    @Test
    fun testKtorSocketWrite() {
        measure("Ktor socket write (128MiB)") {
            val buffer = ByteArray(4088)
            val selectorManager = ActorSelectorManager(Dispatchers.IO)
            aSocket(selectorManager).tcp().connect("localhost", 4200).use { socket ->
                val channel = socket.openWriteChannel(autoFlush = true)
                for (i in 1..32768) { //~100M
                    channel.writeFully(buffer, 0, buffer.size)
                }
                channel.close()
            }

            selectorManager.close()
        }
        executor.shutdown()
    }

    @Test
    fun testJvmSocketWrite() {
        measure("JVM socket write (128MiB)") {
            Socket("localhost", 4200).use {
                val buffer = ByteArray(4088)
                it.getOutputStream().use { stream ->
                    for (i in 1..32768) { //~100M
                        stream.write(buffer, 0, buffer.size)
                    }
                }
            }
        }
    }

    private fun measure(name: String, iterations: Int = 20, block: suspend ByteArray.() -> Unit) {
        val results = mutableListOf<Long>()
        runBlocking {
            withDefaultBuffer {
                for (i in 1..iterations) {
                    measureTimeMillis {
                        block(this.array())
                    }.let { results.add(it) }
                }
            }
        }

        println(
            """
            --------------------------------------------------------------------
            $name, ${results.size} iterations
            Avg: ${results.average()}ms Min: ${results.minOrNull()}ms, Max: ${results.maxOrNull()}ms
            Raw: ${results.sorted().joinToString()}
            
        """.trimIndent()
        )
    }
}

@Malinskiy, why do you open write channel with openWriteChannel(autoFlush = true) ?
I guess you should be using autoFlush = false here

Hey @vlsi, this is because autoFlush = false is even slower than autoFlush = true (new measurements on a different configuration, so can't directly compare to the ones in the article, but still can see the relative performance):

--------------------------------------------------------------------
Ktor socket write (autoFlush = false) (128MiB), 20 iterations
Avg: 307.85ms Min: 263ms, Max: 761ms
Raw: 263, 265, 266, 267, 267, 268, 269, 270, 273, 276, 282, 282, 283, 284, 292, 294, 302, 335, 358, 761
--------------------------------------------------------------------
Ktor socket write (128MiB), 20 iterations
Avg: 288.2ms Min: 263ms, Max: 369ms
Raw: 263, 266, 268, 269, 269, 269, 270, 271, 272, 280, 280, 286, 287, 294, 300, 303, 306, 317, 325, 369

Apart from this, I'm not so sure flushing, in this case, is actual flush, see the issue I raised KTOR-1618. The only way to really flush things with ktor is to await the job.