Introduce new Kotlin Daemon without RMI abstraction

2019-02-28 17:30:35 +03:00
parent a3f718733d
commit ced973b707
107 changed files with 8866 additions and 523 deletions
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+/*
+ * Copyright 2000-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+import com.sun.javafx.scene.CameraHelper.project
+
+plugins {
+    java
+    kotlin("jvm")
+    id("jps-compatible")
+}
+
+jvmTarget = "1.6"
+
+val ktorExcludesForDaemon : List<Pair<String, String>> by rootProject.extra
+
+dependencies {
+    compile(project(":core:descriptors"))
+    compile(project(":core:descriptors.jvm"))
+    compile(project(":compiler:util"))
+    compile(project(":compiler:cli-common"))
+    compileOnly(project(":compiler:daemon-common"))
+    compile(kotlinStdlib())
+    compileOnly(project(":js:js.frontend"))
+    compileOnly(intellijCoreDep()) { includeJars("intellij-core") }
+    compileOnly(intellijDep()) { includeIntellijCoreJarDependencies(project) }
+    compile(commonDep("org.jetbrains.kotlinx", "kotlinx-coroutines-jdk8")) {
+        isTransitive = false
+    }
+    compile(commonDep("io.ktor", "ktor-network")) {
+        ktorExcludesForDaemon.forEach { (group, module) ->
+            exclude(group = group, module = module)
+        }
+    }
+}
+
+sourceSets {
+    "main" { projectDefault() }
+    "test" {}
+}
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import java.util.concurrent.TimeUnit
+
+val RMI_WRAPPER_PORTS_RANGE_START: Int = 13001
+val RMI_WRAPPER_PORTS_RANGE_END: Int = 14000
+
+val REPL_SERVER_PORTS_RANGE_START: Int = 14001
+val REPL_SERVER_PORTS_RANGE_END: Int = 15000
+
+
+val CALLBACK_SERVER_PORTS_RANGE_START: Int = 15001
+val CALLBACK_SERVER_PORTS_RANGE_END: Int = 16000
+
+val RESULTS_SERVER_PORTS_RANGE_START: Int = 16001
+val RESULTS_SERVER_PORTS_RANGE_END: Int = 17000
+
+val COMPILER_DAEMON_CLASS_FQN_EXPERIMENTAL: String = "org.jetbrains.kotlin.daemon.experimental.KotlinCompileDaemon"
+
+val FIRST_HANDSHAKE_BYTE_TOKEN = byteArrayOf(1, 2, 3, 4)
+val AUTH_TIMEOUT_IN_MILLISECONDS = 200L
+
+val DAEMON_PERIODIC_CHECK_INTERVAL_MS = 1000L
+val DAEMON_PERIODIC_SELDOM_CHECK_INTERVAL_MS = 60000L
+
+val KEEPALIVE_PERIOD = 2000L
+val KEEPALIVE_PERIOD_SERVER = 4000L
@@ -0,0 +1,153 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+
+import kotlinx.coroutines.*
+import org.jetbrains.kotlin.daemon.common.*
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.runWithTimeout
+import org.jetbrains.kotlin.daemon.common.DaemonReportCategory
+import org.jetbrains.kotlin.daemon.common.makePortFromRunFilenameExtractor
+import java.io.File
+import java.rmi.registry.LocateRegistry
+import java.util.logging.Logger
+
+/*
+1) walkDaemonsAsync = walkDaemons + some async calls inside (also some used classes changed *** -> ***Async)
+2) tryConnectToDaemonBySockets / tryConnectToDaemonByRMI
+
+*/
+
+internal val MAX_PORT_NUMBER = 0xffff
+
+private const val ORPHANED_RUN_FILE_AGE_THRESHOLD_MS = 1000000L
+
+data class DaemonWithMetadataAsync(val daemon: CompileServiceAsync, val runFile: File, val jvmOptions: DaemonJVMOptions)
+
+val log = Logger.getLogger("client utils")
+
+// TODO: replace mapNotNull in walkDaemonsAsync with this method.
+private suspend fun <T, R : Any> List<T>.mapNotNullAsync(transform: suspend (T) -> R?): List<R> =
+    this
+        .map { GlobalScope.async { transform(it) } }
+        .mapNotNull { it.await() } // await for completion of the last action
+
+
+// TODO: write metadata into discovery file to speed up selection
+// TODO: consider using compiler jar signature (checksum) as a CompilerID (plus java version, plus ???) instead of classpath checksum
+//    would allow to use same compiler from taken different locations
+//    reqs: check that plugins (or anything els) should not be part of the CP
+suspend fun walkDaemonsAsync(
+    registryDir: File,
+    compilerId: CompilerId,
+    fileToCompareTimestamp: File,
+    filter: (File, Int) -> Boolean = { _, _ -> true },
+    report: (DaemonReportCategory, String) -> Unit = { _, _ -> },
+    useRMI: Boolean = true,
+    useSockets: Boolean = true
+): List<DaemonWithMetadataAsync> { // TODO: replace with Deferred<List<DaemonWithMetadataAsync>> and use mapNotNullAsync to speed this up
+    val classPathDigest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest().toHexString()
+    val portExtractor = makePortFromRunFilenameExtractor(classPathDigest)
+    return registryDir.walk().toList() // list, since walk returns Sequence and Sequence.map{...} is not inline => coroutines dont work
+        .map { Pair(it, portExtractor(it.name)) }
+        .filter { (file, port) -> port != null && filter(file, port) }
+        .mapNotNull { (file, port) ->
+            // all actions process concurrently
+            assert(port!! in 1..(MAX_PORT_NUMBER - 1))
+            val relativeAge = fileToCompareTimestamp.lastModified() - file.lastModified()
+            val daemon = tryConnectToDaemonAsync(port, report, file, useRMI, useSockets)
+            // cleaning orphaned file; note: daemon should shut itself down if it detects that the runServer file is deleted
+            if (daemon == null) {
+                if (relativeAge - ORPHANED_RUN_FILE_AGE_THRESHOLD_MS <= 0) {
+                    report(
+                        DaemonReportCategory.DEBUG,
+                        "found fresh runServer file '${file.absolutePath}' ($relativeAge ms old), but no daemon, ignoring it"
+                    )
+                } else {
+                    report(
+                        DaemonReportCategory.DEBUG,
+                        "found seemingly orphaned runServer file '${file.absolutePath}' ($relativeAge ms old), deleting it"
+                    )
+                    if (!file.delete()) {
+                        report(
+                            DaemonReportCategory.INFO,
+                            "WARNING: unable to delete seemingly orphaned file '${file.absolutePath}', cleanup recommended"
+                        )
+                    }
+                }
+            }
+            try {
+                daemon?.let {
+                    DaemonWithMetadataAsync(it, file, it.getDaemonJVMOptions().get())
+                }
+            } catch (e: Exception) {
+                report(
+                    DaemonReportCategory.INFO,
+                    "ERROR: unable to retrieve daemon JVM options, assuming daemon is dead: ${e.message}"
+                )
+                null
+            }
+        }
+}
+
+private inline fun tryConnectToDaemonByRMI(port: Int, report: (DaemonReportCategory, String) -> Unit): CompileServiceAsync? {
+    try {
+        log.info("tryConnectToDaemonByRMI(port = $port)")
+        val daemon = runBlocking {
+            runWithTimeout(2 * DAEMON_PERIODIC_CHECK_INTERVAL_MS) {
+                LocateRegistry.getRegistry(
+                    LoopbackNetworkInterface.loopbackInetAddressName,
+                    port,
+                    LoopbackNetworkInterface.clientLoopbackSocketFactoryRMI
+                )?.lookup(COMPILER_SERVICE_RMI_NAME)
+            }
+        }
+        when (daemon) {
+            null -> report(DaemonReportCategory.INFO, "daemon not found")
+            is CompileService -> return daemon.toClient()
+            else -> report(DaemonReportCategory.INFO, "Unable to cast compiler service, actual class received: ${daemon::class.java.name}")
+        }
+    } catch (e: Throwable) {
+        report(DaemonReportCategory.INFO, "cannot connect to registry: " + (e.cause?.message ?: e.message ?: "unknown error"))
+    }
+    return null
+}
+
+private suspend fun tryConnectToDaemonBySockets(
+    port: Int,
+    file: File,
+    report: (DaemonReportCategory, String) -> Unit
+): CompileServiceClientSide? {
+    return CompileServiceClientSideImpl(
+        port,
+        LoopbackNetworkInterface.loopbackInetAddressName,
+        file
+    ).let { daemon ->
+        try {
+            log.info("tryConnectToDaemonBySockets(port = $port)")
+            log.info("daemon($port) = $daemon")
+            log.info("daemon($port) connecting to server...")
+            daemon.connectToServer()
+            log.info("OK - daemon($port) connected to server!!!")
+            daemon
+        } catch (e: Throwable) {
+            report(DaemonReportCategory.INFO, "kcannot find or connect to socket")
+            daemon.close()
+            null
+        }
+    }
+}
+
+private suspend fun tryConnectToDaemonAsync(
+    port: Int,
+    report: (DaemonReportCategory, String) -> Unit,
+    file: File,
+    useRMI: Boolean = true,
+    useSockets: Boolean = true
+): CompileServiceAsync? =
+    useSockets.takeIf { it }?.let { tryConnectToDaemonBySockets(port, file, report) }
+        ?: (useRMI.takeIf { it }?.let { tryConnectToDaemonByRMI(port, report) })
+
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.CompilationResultsAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClient
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server
+import java.io.Serializable
+
+interface CompilationResultsServerSide : CompilationResultsAsync, Server<CompilationResultsServerSide> {
+    class AddMessage(
+        val compilationResultCategory: Int,
+        val value: Serializable
+    ) : Server.Message<CompilationResultsServerSide>() {
+        override suspend fun processImpl(server: CompilationResultsServerSide, sendReply: (Any?) -> Unit) {
+            server.add(compilationResultCategory, value)
+        }
+    }
+}
+
+interface CompilationResultsClientSide : CompilationResultsAsync, Client<CompilationResultsServerSide>
+
+class CompilationResultsClientSideImpl(val socketPort: Int) : CompilationResultsClientSide,
+    Client<CompilationResultsServerSide> by DefaultClient(socketPort) {
+
+    override val clientSide: CompilationResultsAsync
+        get() = this
+
+    override suspend fun add(compilationResultCategory: Int, value: Serializable) {
+        sendMessage(CompilationResultsServerSide.AddMessage(compilationResultCategory, value))
+    }
+
+    // TODO: consider connecting to server in init-block
+}
+
+enum class CompilationResultCategory(val code: Int) {
+    IC_COMPILE_ITERATION(0)
+}
@@ -0,0 +1,15 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+@file:Suppress("EXPERIMENTAL_FEATURE_WARNING")
+
+package org.jetbrains.kotlin.daemon.common
+
+import org.jetbrains.kotlin.daemon.common.experimental.CompileServiceServerSide
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.*
+
+interface CompileServiceClientSide : CompileServiceAsync, Client<CompileServiceServerSide> {
+    override val serverPort: Int
+}
@@ -0,0 +1,393 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import kotlinx.coroutines.*
+import org.jetbrains.kotlin.cli.common.repl.ReplCheckResult
+import org.jetbrains.kotlin.cli.common.repl.ReplCodeLine
+import org.jetbrains.kotlin.cli.common.repl.ReplCompileResult
+import org.jetbrains.kotlin.daemon.common.*
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.*
+import org.jetbrains.kotlin.daemon.common.CompileService
+import org.jetbrains.kotlin.daemon.common.CompilerServicesFacadeBase
+import java.io.File
+import java.util.logging.Logger
+
+class CompileServiceClientSideImpl(
+    override val serverPort: Int,
+    val serverHost: String,
+    val serverFile: File
+) : CompileServiceClientSide,
+    Client<CompileServiceServerSide> by object : DefaultAuthorizableClient<CompileServiceServerSide>(
+        serverPort,
+        serverHost
+    ) {
+
+        private fun nowMillieconds() = System.currentTimeMillis()
+
+        @Volatile
+        private var lastUsedMilliSeconds: Long = nowMillieconds()
+
+        private fun millisecondsSinceLastUsed() = nowMillieconds() - lastUsedMilliSeconds
+
+        private fun keepAliveSuccess() = millisecondsSinceLastUsed() < KEEPALIVE_PERIOD
+
+        override suspend fun authorizeOnServer(serverOutputChannel: ByteWriteChannelWrapper): Boolean =
+            runWithTimeout {
+                val signature = serverFile.inputStream().use(::readTokenKeyPairAndSign)
+                sendSignature(serverOutputChannel, signature)
+                true
+            } ?: false
+
+        override suspend fun clientHandshake(input: ByteReadChannelWrapper, output: ByteWriteChannelWrapper, log: Logger): Boolean {
+            return trySendHandshakeMessage(output, log) && tryAcquireHandshakeMessage(input, log)
+        }
+
+        override fun startKeepAlives() {
+            val keepAliveMessage = Server.KeepAliveMessage<CompileServiceServerSide>()
+            GlobalScope.async(newSingleThreadContext("keepAliveThread")) {
+                delay(KEEPALIVE_PERIOD * 4)
+                while (true) {
+                    delay(KEEPALIVE_PERIOD)
+                    while (keepAliveSuccess()) {
+                        delay(KEEPALIVE_PERIOD - millisecondsSinceLastUsed())
+                    }
+                    runWithTimeout(timeout = KEEPALIVE_PERIOD / 2) {
+                        val id = sendMessage(keepAliveMessage)
+                        readMessage<Server.KeepAliveAcknowledgement<*>>(id)
+                    } ?: if (!keepAliveSuccess()) readActor.send(StopAllRequests()).also {
+                    }
+                }
+            }
+        }
+
+        override fun delayKeepAlives() {
+            lastUsedMilliSeconds = nowMillieconds()
+        }
+
+    } {
+    override suspend fun classesFqNamesByFiles(sessionId: Int, sourceFiles: Set<File>): CompileService.CallResult<Set<String>> {
+        val id = sendMessage(ClassesFqNamesByFilesMessage(sessionId, sourceFiles))
+        return readMessage(id)
+    }
+
+    override suspend fun compile(
+        sessionId: Int,
+        compilerArguments: Array<out String>,
+        compilationOptions: CompilationOptions,
+        servicesFacade: CompilerServicesFacadeBaseAsync,
+        compilationResults: CompilationResultsAsync?
+    ): CompileService.CallResult<Int> {
+        val id = sendMessage(CompileMessage(
+            sessionId,
+            compilerArguments,
+            compilationOptions,
+            servicesFacade,
+            compilationResults
+        ))
+        return readMessage(id)
+    }
+
+    override suspend fun leaseReplSession(
+        aliveFlagPath: String?,
+        compilerArguments: Array<out String>,
+        compilationOptions: CompilationOptions,
+        servicesFacade: CompilerServicesFacadeBaseAsync,
+        templateClasspath: List<File>,
+        templateClassName: String
+    ): CompileService.CallResult<Int> {
+        val id = sendMessage(
+            LeaseReplSessionMessage(
+                aliveFlagPath,
+                compilerArguments,
+                compilationOptions,
+                servicesFacade,
+                templateClasspath,
+                templateClassName
+            )
+        )
+        return readMessage(id)
+    }
+
+    // CompileService methods:
+
+    override suspend fun checkCompilerId(expectedCompilerId: CompilerId): Boolean {
+        val id = sendMessage(
+            CheckCompilerIdMessage(
+                expectedCompilerId
+            )
+        )
+        return readMessage(id)
+    }
+
+    override suspend fun getUsedMemory(): CompileService.CallResult<Long> {
+        val id = sendMessage(GetUsedMemoryMessage())
+        return readMessage(id)
+    }
+
+
+    override suspend fun getDaemonOptions(): CompileService.CallResult<DaemonOptions> {
+        val id = sendMessage(GetDaemonOptionsMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun getDaemonInfo(): CompileService.CallResult<String> {
+        val id = sendMessage(GetDaemonInfoMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun getDaemonJVMOptions(): CompileService.CallResult<DaemonJVMOptions> {
+        val id = sendMessage(GetDaemonJVMOptionsMessage())
+        val res = readMessage<CompileService.CallResult<DaemonJVMOptions>>(id)
+        return res
+    }
+
+    override suspend fun registerClient(aliveFlagPath: String?): CompileService.CallResult<Nothing> {
+        val id = sendMessage(RegisterClientMessage(aliveFlagPath))
+        return readMessage(id)
+    }
+
+    override suspend fun getClients(): CompileService.CallResult<List<String>> {
+        val id = sendMessage(GetClientsMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun leaseCompileSession(aliveFlagPath: String?): CompileService.CallResult<Int> {
+        val id = sendMessage(
+            LeaseCompileSessionMessage(
+                aliveFlagPath
+            )
+        )
+        return readMessage(id)
+    }
+
+    override suspend fun releaseCompileSession(sessionId: Int): CompileService.CallResult<Nothing> {
+        val id = sendMessage(
+            ReleaseCompileSessionMessage(
+                sessionId
+            )
+        )
+        return readMessage(id)
+    }
+
+    override suspend fun shutdown(): CompileService.CallResult<Nothing> {
+        val id = sendMessage(ShutdownMessage())
+        val res = readMessage<CompileService.CallResult<Nothing>>(id)
+        return res
+    }
+
+    override suspend fun scheduleShutdown(graceful: Boolean): CompileService.CallResult<Boolean> {
+        val id = sendMessage(ScheduleShutdownMessage(graceful))
+        return readMessage(id)
+    }
+
+    override suspend fun clearJarCache() {
+        val id = sendMessage(ClearJarCacheMessage())
+    }
+
+    override suspend fun releaseReplSession(sessionId: Int): CompileService.CallResult<Nothing> {
+        val id = sendMessage(ReleaseReplSessionMessage(sessionId))
+        return readMessage(id)
+    }
+
+    override suspend fun replCreateState(sessionId: Int): CompileService.CallResult<ReplStateFacadeAsync> {
+        val id = sendMessage(ReplCreateStateMessage(sessionId))
+        return readMessage(id)
+    }
+
+    override suspend fun replCheck(
+        sessionId: Int,
+        replStateId: Int,
+        codeLine: ReplCodeLine
+    ): CompileService.CallResult<ReplCheckResult> {
+        val id = sendMessage(
+            ReplCheckMessage(
+                sessionId,
+                replStateId,
+                codeLine
+            )
+        )
+        return readMessage(id)
+    }
+
+    override suspend fun replCompile(
+        sessionId: Int,
+        replStateId: Int,
+        codeLine: ReplCodeLine
+    ): CompileService.CallResult<ReplCompileResult> {
+        val id = sendMessage(
+            ReplCompileMessage(
+                sessionId,
+                replStateId,
+                codeLine
+            )
+        )
+        return readMessage(id)
+    }
+
+    // Query messages:
+
+    class CheckCompilerIdMessage(val expectedCompilerId: CompilerId) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.checkCompilerId(expectedCompilerId))
+    }
+
+    class GetUsedMemoryMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getUsedMemory())
+    }
+
+    class GetDaemonOptionsMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getDaemonOptions())
+    }
+
+    class GetDaemonJVMOptionsMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getDaemonJVMOptions())
+    }
+
+    class GetDaemonInfoMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getDaemonInfo())
+    }
+
+    class RegisterClientMessage(val aliveFlagPath: String?) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.registerClient(aliveFlagPath))
+    }
+
+
+    class GetClientsMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getClients())
+    }
+
+    class LeaseCompileSessionMessage(val aliveFlagPath: String?) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.leaseCompileSession(aliveFlagPath))
+    }
+
+    class ReleaseCompileSessionMessage(val sessionId: Int) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.releaseCompileSession(sessionId))
+    }
+
+    class ShutdownMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.shutdown())
+    }
+
+    class ScheduleShutdownMessage(val graceful: Boolean) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.scheduleShutdown(graceful))
+    }
+
+    class CompileMessage(
+        val sessionId: Int,
+        val compilerArguments: Array<out String>,
+        val compilationOptions: CompilationOptions,
+        val servicesFacade: CompilerServicesFacadeBaseAsync,
+        val compilationResults: CompilationResultsAsync?
+    ) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(
+                server.compile(
+                    sessionId,
+                    compilerArguments,
+                    compilationOptions,
+                    servicesFacade,
+                    compilationResults
+                )
+            )
+    }
+
+    class ClassesFqNamesByFilesMessage(
+        val sessionId: Int,
+        val sourceFiles: Set<File>
+    ) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(
+                server.classesFqNamesByFiles(sessionId, sourceFiles)
+            )
+    }
+
+    class ClearJarCacheMessage : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            server.clearJarCache()
+    }
+
+    class LeaseReplSessionMessage(
+        val aliveFlagPath: String?,
+        val compilerArguments: Array<out String>,
+        val compilationOptions: CompilationOptions,
+        val servicesFacade: CompilerServicesFacadeBaseAsync,
+        val templateClasspath: List<File>,
+        val templateClassName: String
+    ) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(
+                server.leaseReplSession(
+                    aliveFlagPath,
+                    compilerArguments,
+                    compilationOptions,
+                    servicesFacade,
+                    templateClasspath,
+                    templateClassName
+                )
+            )
+    }
+
+    class ReleaseReplSessionMessage(val sessionId: Int) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.releaseReplSession(sessionId))
+    }
+
+    class LeaseReplSession_Short_Message(
+        val aliveFlagPath: String?,
+        val compilerArguments: Array<out String>,
+        val compilationOptions: CompilationOptions,
+        val servicesFacade: CompilerServicesFacadeBase,
+        val templateClasspath: List<File>,
+        val templateClassName: String
+    ) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(
+                server.leaseReplSession(
+                    aliveFlagPath,
+                    compilerArguments,
+                    compilationOptions,
+                    servicesFacade.toClient(),
+                    templateClasspath,
+                    templateClassName
+                )
+            )
+    }
+
+    class ReplCreateStateMessage(val sessionId: Int) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.replCreateState(sessionId))
+    }
+
+    class ReplCheckMessage(
+        val sessionId: Int,
+        val replStateId: Int,
+        val codeLine: ReplCodeLine
+    ) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.replCheck(sessionId, replStateId, codeLine))
+    }
+
+    class ReplCompileMessage(
+        val sessionId: Int,
+        val replStateId: Int,
+        val codeLine: ReplCodeLine
+    ) : Server.Message<CompileServiceServerSide>() {
+        override suspend fun processImpl(server: CompileServiceServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.replCompile(sessionId, replStateId, codeLine))
+    }
+
+}
@@ -0,0 +1,234 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import kotlinx.coroutines.runBlocking
+import org.jetbrains.kotlin.cli.common.repl.ReplCodeLine
+import org.jetbrains.kotlin.daemon.common.*
+import org.jetbrains.kotlin.daemon.common.LoopbackNetworkInterface
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClientRMIWrapper
+import org.jetbrains.kotlin.daemon.common.*
+import java.io.File
+import java.io.Serializable
+import java.rmi.NoSuchObjectException
+import java.rmi.server.UnicastRemoteObject
+import java.util.*
+import java.util.logging.Logger
+
+class CompileServiceRMIWrapper(val server: CompileServiceServerSide, daemonOptions: DaemonOptions, compilerId: CompilerId) :
+    CompileService {
+
+    override fun classesFqNamesByFiles(sessionId: Int, sourceFiles: Set<File>) = runBlocking {
+        server.classesFqNamesByFiles(sessionId, sourceFiles)
+    }
+
+    val log = Logger.getLogger("CompileServiceRMIWrapper")
+
+    private fun deprecated(): Nothing = TODO("NEVER USE DEPRECATED METHODS, PLEASE!") // prints this todo message
+
+    override fun checkCompilerId(expectedCompilerId: CompilerId) = runBlocking {
+        server.checkCompilerId(expectedCompilerId)
+    }
+
+    override fun getUsedMemory() = runBlocking {
+        server.getUsedMemory()
+    }
+
+    override fun getDaemonOptions() = runBlocking {
+        server.getDaemonOptions()
+    }
+
+    override fun getDaemonInfo() = runBlocking {
+        server.getDaemonInfo()
+    }
+
+    override fun getDaemonJVMOptions() = runBlocking {
+        server.getDaemonJVMOptions()
+    }
+
+    override fun registerClient(aliveFlagPath: String?) = runBlocking {
+        server.registerClient(aliveFlagPath)
+    }
+
+    override fun getClients() = runBlocking {
+        server.getClients()
+    }
+
+    override fun leaseCompileSession(aliveFlagPath: String?) = runBlocking {
+        server.leaseCompileSession(aliveFlagPath)
+    }
+
+    override fun releaseCompileSession(sessionId: Int) = runBlocking {
+        server.releaseCompileSession(sessionId)
+    }
+
+    override fun shutdown() = runBlocking {
+        server.shutdown()
+    }
+
+    override fun scheduleShutdown(graceful: Boolean) = runBlocking {
+        server.scheduleShutdown(graceful)
+    }
+
+    override fun remoteCompile(
+        sessionId: Int,
+        targetPlatform: CompileService.TargetPlatform,
+        args: Array<out String>,
+        servicesFacade: CompilerCallbackServicesFacade,
+        compilerOutputStream: RemoteOutputStream,
+        outputFormat: CompileService.OutputFormat,
+        serviceOutputStream: RemoteOutputStream,
+        operationsTracer: RemoteOperationsTracer?
+    ) = deprecated()
+
+    override fun remoteIncrementalCompile(
+        sessionId: Int,
+        targetPlatform: CompileService.TargetPlatform,
+        args: Array<out String>,
+        servicesFacade: CompilerCallbackServicesFacade,
+        compilerOutputStream: RemoteOutputStream,
+        compilerOutputFormat: CompileService.OutputFormat,
+        serviceOutputStream: RemoteOutputStream,
+        operationsTracer: RemoteOperationsTracer?
+    ) = deprecated()
+
+    override fun compile(
+        sessionId: Int,
+        compilerArguments: Array<out String>,
+        compilationOptions: CompilationOptions,
+        servicesFacade: CompilerServicesFacadeBase,
+        compilationResults: CompilationResults?
+    ) = runBlocking {
+        server.compile(
+            sessionId,
+            compilerArguments,
+            compilationOptions,
+            servicesFacade.toClient(),
+            compilationResults?.toClient() ?: object : CompilationResultsClientSide,
+                Client<CompilationResultsServerSide> by DefaultClientRMIWrapper() {
+
+                override val clientSide: CompilationResultsAsync
+                    get() = this
+
+                override suspend fun add(compilationResultCategory: Int, value: Serializable) {}
+            }
+        )
+    }
+
+    override fun clearJarCache() = runBlocking {
+        server.clearJarCache()
+    }
+
+    override fun leaseReplSession(
+        aliveFlagPath: String?,
+        targetPlatform: CompileService.TargetPlatform,
+        servicesFacade: CompilerCallbackServicesFacade,
+        templateClasspath: List<File>,
+        templateClassName: String,
+        scriptArgs: Array<out Any?>?,
+        scriptArgsTypes: Array<out Class<out Any>>?,
+        compilerMessagesOutputStream: RemoteOutputStream,
+        evalOutputStream: RemoteOutputStream?,
+        evalErrorStream: RemoteOutputStream?,
+        evalInputStream: RemoteInputStream?,
+        operationsTracer: RemoteOperationsTracer?
+    ) = deprecated()
+
+    override fun releaseReplSession(sessionId: Int) = runBlocking {
+        server.releaseReplSession(sessionId)
+    }
+
+    override fun remoteReplLineCheck(sessionId: Int, codeLine: ReplCodeLine) = deprecated()
+
+    override fun remoteReplLineCompile(
+        sessionId: Int,
+        codeLine: ReplCodeLine,
+        history: List<ReplCodeLine>?
+    ) = deprecated()
+
+    override fun remoteReplLineEval(
+        sessionId: Int,
+        codeLine: ReplCodeLine,
+        history: List<ReplCodeLine>?
+    ) = deprecated()
+
+    override fun leaseReplSession(
+        aliveFlagPath: String?,
+        compilerArguments: Array<out String>,
+        compilationOptions: CompilationOptions,
+        servicesFacade: CompilerServicesFacadeBase,
+        templateClasspath: List<File>,
+        templateClassName: String
+    ) = runBlocking {
+        server.leaseReplSession(
+            aliveFlagPath,
+            compilerArguments,
+            compilationOptions,
+            servicesFacade.toClient(),
+            templateClasspath,
+            templateClassName
+        )
+    }
+
+    override fun replCreateState(sessionId: Int) = runBlocking {
+        server.replCreateState(sessionId).toRMI()
+    }
+
+    override fun replCheck(sessionId: Int, replStateId: Int, codeLine: ReplCodeLine) = runBlocking {
+        server.replCheck(sessionId, replStateId, codeLine)
+    }
+
+    override fun replCompile(sessionId: Int, replStateId: Int, codeLine: ReplCodeLine) = runBlocking {
+        server.replCompile(sessionId, replStateId, codeLine)
+    }
+
+    init {
+        try {
+            // cleanup for the case of incorrect restart and many other situations
+            UnicastRemoteObject.unexportObject(this, false)
+        } catch (e: NoSuchObjectException) {
+            // ignoring if object already exported
+        }
+
+        val (registry, port) = findPortAndCreateRegistry(
+            COMPILE_DAEMON_FIND_PORT_ATTEMPTS,
+            RMI_WRAPPER_PORTS_RANGE_START,
+            RMI_WRAPPER_PORTS_RANGE_END
+        )
+
+        val stub = UnicastRemoteObject.exportObject(
+            this,
+            port,
+            LoopbackNetworkInterface.clientLoopbackSocketFactory,
+            LoopbackNetworkInterface.serverLoopbackSocketFactory
+        ) as CompileService
+
+        registry.rebind(COMPILER_SERVICE_RMI_NAME, stub)
+
+        // create file :
+        val runFileDir = File(daemonOptions.runFilesPathOrDefault)
+        runFileDir.mkdirs()
+        val runFile = File(
+            runFileDir,
+            makeRunFilenameString(
+                timestamp = "%tFT%<tH-%<tM-%<tS.%<tLZ".format(Calendar.getInstance(TimeZone.getTimeZone("Z"))),
+                digest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest().toHexString(),
+                port = port.toString()
+            )
+        )
+        try {
+            if (!runFile.createNewFile()) throw Exception("createNewFile returned false")
+        } catch (e: Throwable) {
+            throw IllegalStateException("Unable to create runServer file '${runFile.absolutePath}'", e)
+        }
+        runFile.deleteOnExit()
+    }
+
+}
+
+fun CompileServiceServerSide.toRMIServer(daemonOptions: DaemonOptions, compilerId: CompilerId) =
+    CompileServiceRMIWrapper(this, daemonOptions, compilerId)
@@ -0,0 +1,13 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.CompileServiceAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server
+
+interface CompileServiceServerSide : CompileServiceAsync, Server<CompileServiceServerSide> {
+    override val serverPort: Int
+}
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.CompilerCallbackServicesFacadeAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClient
+import org.jetbrains.kotlin.incremental.components.LookupInfo
+import org.jetbrains.kotlin.load.kotlin.incremental.components.JvmPackagePartProto
+import org.jetbrains.kotlin.modules.TargetId
+import java.io.Serializable
+
+
+interface CompilerCallbackServicesFacadeClientSide : CompilerCallbackServicesFacadeAsync, Client<CompilerServicesFacadeBaseServerSide>, CompilerServicesFacadeBaseClientSide
+
+@Suppress("UNCHECKED_CAST")
+class CompilerCallbackServicesFacadeClientSideImpl(serverPort: Int) : CompilerCallbackServicesFacadeClientSide,
+    Client<CompilerServicesFacadeBaseServerSide> by DefaultClient(serverPort) {
+
+    override suspend fun hasIncrementalCaches(): Boolean {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.HasIncrementalCachesMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun hasLookupTracker(): Boolean {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.HasLookupTrackerMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun hasCompilationCanceledStatus(): Boolean {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.HasCompilationCanceledStatusMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun incrementalCache_getObsoletePackageParts(target: TargetId): Collection<String> {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_getObsoletePackagePartsMessage(target))
+        return readMessage(id)
+    }
+
+    override suspend fun incrementalCache_getObsoleteMultifileClassFacades(target: TargetId): Collection<String> {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_getObsoleteMultifileClassFacadesMessage(target))
+        return readMessage(id)
+    }
+
+    override suspend fun incrementalCache_getPackagePartData(target: TargetId, partInternalName: String): JvmPackagePartProto? {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_getPackagePartDataMessage(target, partInternalName))
+        return readMessage(id)
+    }
+
+    override suspend fun incrementalCache_getModuleMappingData(target: TargetId): ByteArray? {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_getModuleMappingDataMessage(target))
+        return readMessage(id)
+    }
+
+    override suspend fun incrementalCache_registerInline(target: TargetId, fromPath: String, jvmSignature: String, toPath: String) {
+        sendNoReplyMessage(
+            CompilerCallbackServicesFacadeServerSide.IncrementalCache_registerInlineMessage(
+                target,
+                fromPath,
+                jvmSignature,
+                toPath
+            )
+        )
+    }
+
+    override suspend fun incrementalCache_getClassFilePath(target: TargetId, internalClassName: String): String {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_getClassFilePathMessage(target, internalClassName))
+        return readMessage(id)
+    }
+
+    override suspend fun incrementalCache_close(target: TargetId) =
+        sendNoReplyMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_closeMessage(target))
+
+    override suspend fun incrementalCache_getMultifileFacadeParts(target: TargetId, internalName: String): Collection<String>? {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.IncrementalCache_getMultifileFacadePartsMessage(target, internalName))
+        return readMessage(id)
+    }
+
+    override suspend fun lookupTracker_requiresPosition(): Boolean {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.LookupTracker_requiresPositionMessage())
+        return readMessage(id)
+    }
+
+    override fun lookupTracker_record(lookups: Collection<LookupInfo>) =
+        sendNoReplyMessage(CompilerCallbackServicesFacadeServerSide.LookupTracker_recordMessage(lookups))
+
+    override suspend fun lookupTracker_isDoNothing(): Boolean {
+        val id = sendMessage(CompilerCallbackServicesFacadeServerSide.LookupTracker_isDoNothingMessage())
+        return readMessage(id)
+    }
+
+    override suspend fun compilationCanceledStatus_checkCanceled(): Void? {
+        sendNoReplyMessage(CompilerCallbackServicesFacadeServerSide.CompilationCanceledStatus_checkCanceledMessage())
+        return null
+    }
+
+    override suspend fun report(category: Int, severity: Int, message: String?, attachment: Serializable?) {
+        sendNoReplyMessage(CompilerServicesFacadeBaseServerSide.ReportMessage(category, severity, message, attachment))
+    }
+
+}
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.CompilerCallbackServicesFacadeAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server.Message
+import org.jetbrains.kotlin.incremental.components.LookupInfo
+import org.jetbrains.kotlin.modules.TargetId
+
+interface CompilerCallbackServicesFacadeServerSide : CompilerCallbackServicesFacadeAsync, CompilerServicesFacadeBaseServerSide {
+
+    class HasIncrementalCachesMessage : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.hasIncrementalCaches())
+    }
+
+    class HasLookupTrackerMessage : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.hasLookupTracker())
+    }
+
+    class HasCompilationCanceledStatusMessage : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.hasCompilationCanceledStatus())
+    }
+
+    // ----------------------------------------------------
+    // IncrementalCache
+    class IncrementalCache_getObsoletePackagePartsMessage(val target: TargetId) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.incrementalCache_getObsoletePackageParts(target))
+    }
+
+    class IncrementalCache_getObsoleteMultifileClassFacadesMessage(val target: TargetId) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.incrementalCache_getObsoleteMultifileClassFacades(target))
+    }
+
+    class IncrementalCache_getPackagePartDataMessage(val target: TargetId, val partInternalName: String) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.incrementalCache_getPackagePartData(target, partInternalName))
+    }
+
+    class IncrementalCache_getModuleMappingDataMessage(val target: TargetId) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.incrementalCache_getModuleMappingData(target))
+    }
+
+    class IncrementalCache_registerInlineMessage(
+        val target: TargetId,
+        val fromPath: String,
+        val jvmSignature: String,
+        val toPath: String
+    ) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            server.incrementalCache_registerInline(target, fromPath, jvmSignature, toPath)
+    }
+
+    class IncrementalCache_getClassFilePathMessage(val target: TargetId, val internalClassName: String) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.incrementalCache_getClassFilePath(target, internalClassName))
+    }
+
+    class IncrementalCache_closeMessage(val target: TargetId) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            server.incrementalCache_close(target)
+    }
+
+    class IncrementalCache_getMultifileFacadePartsMessage(val target: TargetId, val internalName: String) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.incrementalCache_getMultifileFacadeParts(target, internalName))
+    }
+
+    // ----------------------------------------------------
+    // LookupTracker
+
+    class LookupTracker_requiresPositionMessage : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) {
+            server.lookupTracker_requiresPosition()
+        }
+    }
+
+    class LookupTracker_recordMessage(val lookups: Collection<LookupInfo>) : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.lookupTracker_record(lookups))
+    }
+
+    class LookupTracker_isDoNothingMessage : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.lookupTracker_isDoNothing())
+    }
+
+    // ----------------------------------------------------
+    // CompilationCanceledStatus
+    class CompilationCanceledStatus_checkCanceledMessage : Message<CompilerCallbackServicesFacadeServerSide>() {
+        override suspend fun processImpl(server: CompilerCallbackServicesFacadeServerSide, sendReply: (Any?) -> Unit) {
+            server.compilationCanceledStatus_checkCanceled()
+        }
+    }
+
+}
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.CompilerServicesFacadeBaseAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClient
+import java.io.Serializable
+
+interface CompilerServicesFacadeBaseClientSide : CompilerServicesFacadeBaseAsync, Client<CompilerServicesFacadeBaseServerSide>
+
+class CompilerServicesFacadeBaseClientSideImpl(val serverPort: Int) :
+    CompilerServicesFacadeBaseClientSide,
+    Client<CompilerServicesFacadeBaseServerSide> by DefaultClient(serverPort) {
+
+    // TODO: consider invoking connectToServer() in init block
+
+    override suspend fun report(category: Int, severity: Int, message: String?, attachment: Serializable?) {
+        log.info("client $serverPort - fun report")
+        sendNoReplyMessage(
+            CompilerServicesFacadeBaseServerSide.ReportMessage(
+                category, severity, message, attachment
+            )
+        )
+    }
+
+}
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.CompilerServicesFacadeBaseAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server
+import java.io.Serializable
+
+interface CompilerServicesFacadeBaseServerSide : CompilerServicesFacadeBaseAsync, Server<CompilerServicesFacadeBaseServerSide> {
+
+    class ReportMessage(
+        val category: Int,
+        val severity: Int,
+        val message: String?,
+        val attachment: Serializable?
+    ) : Server.Message<CompilerServicesFacadeBaseServerSide>() {
+
+        override suspend fun processImpl(server: CompilerServicesFacadeBaseServerSide, sendReply: (Any?) -> Unit) {
+            server.report(category, severity, message, attachment)
+        }
+
+    }
+}
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.IncrementalCompilerServicesFacade
+import org.jetbrains.kotlin.daemon.common.SimpleDirtyData
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClientRMIWrapper
+import java.io.File
+import java.io.Serializable
+
+class IncrementalCompilerServicesFacadeAsyncWrapper(
+    val rmiImpl: IncrementalCompilerServicesFacade
+) : IncrementalCompilerServicesFacadeClientSide, Client<CompilerServicesFacadeBaseServerSide> by DefaultClientRMIWrapper() {
+
+    override suspend fun report(category: Int, severity: Int, message: String?, attachment: Serializable?) =
+        rmiImpl.report(category, severity, message, attachment)
+
+}
+
+fun IncrementalCompilerServicesFacade.toClient() = IncrementalCompilerServicesFacadeAsyncWrapper(this)
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.IncrementalCompilerServicesFacadeAsync
+import org.jetbrains.kotlin.daemon.common.SimpleDirtyData
+import org.jetbrains.kotlin.daemon.common.experimental.IncrementalCompilerServicesFacadeServerSide.*
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClient
+import java.io.File
+import java.io.Serializable
+
+interface IncrementalCompilerServicesFacadeClientSide : IncrementalCompilerServicesFacadeAsync, CompilerServicesFacadeBaseClientSide
+
+class IncrementalCompilerServicesFacadeClientSideImpl(val serverPort: Int) :
+    IncrementalCompilerServicesFacadeClientSide,
+    Client<CompilerServicesFacadeBaseServerSide> by DefaultClient(serverPort) {
+
+    override suspend fun report(category: Int, severity: Int, message: String?, attachment: Serializable?) {
+        sendNoReplyMessage(CompilerServicesFacadeBaseServerSide.ReportMessage(category, severity, message, attachment))
+    }
+
+}
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import kotlinx.coroutines.runBlocking
+import org.jetbrains.kotlin.daemon.common.IncrementalCompilerServicesFacade
+import org.jetbrains.kotlin.daemon.common.SimpleDirtyData
+import java.io.File
+import java.io.Serializable
+
+class IncrementalCompilerServicesFacadeRMIWrapper(val clientSide: IncrementalCompilerServicesFacadeClientSide) :
+    IncrementalCompilerServicesFacade, Serializable {
+
+    override fun report(category: Int, severity: Int, message: String?, attachment: Serializable?) = runBlocking {
+        clientSide.report(category, severity, message, attachment)
+    }
+
+    // TODO: consider connecting to server right here in init-block
+}
+
+fun IncrementalCompilerServicesFacadeClientSide.toRMI() =
+    if (this is IncrementalCompilerServicesFacadeAsyncWrapper) this.rmiImpl
+    else IncrementalCompilerServicesFacadeRMIWrapper(this)
@@ -0,0 +1,14 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.IncrementalCompilerServicesFacadeAsync
+import org.jetbrains.kotlin.daemon.common.SimpleDirtyData
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server
+import java.io.File
+
+
+interface IncrementalCompilerServicesFacadeServerSide : IncrementalCompilerServicesFacadeAsync, CompilerServicesFacadeBaseServerSide
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import io.ktor.network.selector.ActorSelectorManager
+import io.ktor.network.sockets.aSocket
+import kotlinx.coroutines.*
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.ServerSocketWrapper
+import org.jetbrains.kotlin.daemon.common.*
+import java.io.IOException
+import java.io.Serializable
+import java.net.*
+import java.rmi.server.RMIClientSocketFactory
+import java.rmi.server.RMIServerSocketFactory
+import java.util.*
+
+
+// copyed from original(org.jetbrains.kotlin.daemon.common.NetworkUtils) TODO
+// unique part :
+// - AbstractClientLoopbackSocketFactory / ServerLoopbackSocketFactoryRMI / ServerLoopbackSocketFactoryKtor - Ktor-Sockets instead of java sockets
+// - findPortAndCreateSocket
+// TODO: get rid of copy-paste here
+
+object LoopbackNetworkInterface {
+
+    const val IPV4_LOOPBACK_INET_ADDRESS = "127.0.0.1"
+    const val IPV6_LOOPBACK_INET_ADDRESS = "::1"
+
+    // size of the requests queue for daemon services, so far seems that we don't need any big numbers here
+    // but if we'll start getting "connection refused" errors, that could be the first place to try to fix it
+    val SERVER_SOCKET_BACKLOG_SIZE by lazy {
+        System.getProperty(DAEMON_RMI_SOCKET_BACKLOG_SIZE_PROPERTY)?.toIntOrNull() ?: DEFAULT_SERVER_SOCKET_BACKLOG_SIZE
+    }
+    val SOCKET_CONNECT_ATTEMPTS by lazy {
+        System.getProperty(DAEMON_RMI_SOCKET_CONNECT_ATTEMPTS_PROPERTY)?.toIntOrNull() ?: DEFAULT_SOCKET_CONNECT_ATTEMPTS
+    }
+    val SOCKET_CONNECT_INTERVAL_MS by lazy {
+        System.getProperty(DAEMON_RMI_SOCKET_CONNECT_INTERVAL_PROPERTY)?.toLongOrNull() ?: DEFAULT_SOCKET_CONNECT_INTERVAL_MS
+    }
+
+    val serverLoopbackSocketFactoryRMI by lazy { ServerLoopbackSocketFactoryRMI() }
+    val clientLoopbackSocketFactoryRMI by lazy { ClientLoopbackSocketFactoryRMI() }
+
+    val serverLoopbackSocketFactoryKtor by lazy { ServerLoopbackSocketFactoryKtor() }
+    val clientLoopbackSocketFactoryKtor by lazy { ClientLoopbackSocketFactoryKtor() }
+
+    // TODO switch to InetAddress.getLoopbackAddress on java 7+
+    val loopbackInetAddressName by lazy {
+        try {
+            if (InetAddress.getByName(null) is Inet6Address) IPV6_LOOPBACK_INET_ADDRESS else IPV4_LOOPBACK_INET_ADDRESS
+        } catch (e: IOException) {
+            // getLocalHost may fail for unknown reasons in some situations, the fallback is to assume IPv4 for now
+            // TODO consider some other ways to detect default to IPv6 addresses in this case
+            IPV4_LOOPBACK_INET_ADDRESS
+        }
+    }
+
+    // base socket factories by default don't implement equals properly (see e.g. http://stackoverflow.com/questions/21555710/rmi-and-jmx-socket-factories)
+    // so implementing it in derived classes using the fact that they are singletons
+
+    class ServerLoopbackSocketFactoryRMI : RMIServerSocketFactory, Serializable {
+        override fun equals(other: Any?): Boolean = other === this || super.equals(other)
+        override fun hashCode(): Int = super.hashCode()
+
+        @Throws(IOException::class)
+        override fun createServerSocket(port: Int): java.net.ServerSocket =
+            ServerSocket(port, SERVER_SOCKET_BACKLOG_SIZE, InetAddress.getByName(null))
+    }
+
+    val selectorMgr = ActorSelectorManager(Dispatchers.IO)
+
+    class ServerLoopbackSocketFactoryKtor : Serializable {
+        override fun equals(other: Any?): Boolean = other === this || super.equals(other)
+        override fun hashCode(): Int = super.hashCode()
+
+        @Throws(IOException::class)
+        fun createServerSocket(port: Int) =
+            aSocket(selectorMgr)
+                .tcp()
+                .bind(InetSocketAddress(InetAddress.getByName(null), port)) // TODO : NO BACKLOG SIZE CHANGE =(
+    }
+
+    abstract class AbstractClientLoopbackSocketFactory<SocketType> : Serializable {
+        override fun equals(other: Any?): Boolean = other === this || super.equals(other)
+        override fun hashCode(): Int = super.hashCode()
+
+        abstract protected fun socketCreate(host: String, port: Int): SocketType
+
+        @Throws(IOException::class)
+        fun createSocket(host: String, port: Int): SocketType {
+            var attemptsLeft = SOCKET_CONNECT_ATTEMPTS
+            while (true) {
+                try {
+                    return socketCreate(host, port)
+                } catch (e: ConnectException) {
+                    if (--attemptsLeft <= 0) throw e
+                }
+                Thread.sleep(SOCKET_CONNECT_INTERVAL_MS)
+            }
+        }
+    }
+
+    class ClientLoopbackSocketFactoryRMI : AbstractClientLoopbackSocketFactory<java.net.Socket>(), RMIClientSocketFactory {
+        override fun socketCreate(host: String, port: Int): Socket = Socket(InetAddress.getByName(null), port)
+    }
+
+    class ClientLoopbackSocketFactoryKtor : AbstractClientLoopbackSocketFactory<io.ktor.network.sockets.Socket>() {
+        override fun socketCreate(host: String, port: Int): io.ktor.network.sockets.Socket =
+            runBlocking { aSocket(selectorMgr).tcp().connect(InetSocketAddress(host, port)) }
+    }
+
+}
+
+
+private val portSelectionRng = Random()
+
+fun findPortForSocket(attempts: Int, portRangeStart: Int, portRangeEnd: Int): ServerSocketWrapper {
+    var i = 0
+    var lastException: Exception? = null
+
+    while (i++ < attempts) {
+        val port = portSelectionRng.nextInt(portRangeEnd - portRangeStart) + portRangeStart
+        try {
+            return ServerSocketWrapper(
+                port,
+                LoopbackNetworkInterface.serverLoopbackSocketFactoryKtor.createServerSocket(port)
+            )
+        } catch (e: Exception) {
+            // assuming that the socketPort is already taken
+            lastException = e
+        }
+    }
+    throw IllegalStateException("Cannot find free socketPort in $attempts attempts", lastException)
+}
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+interface RemoteOutputStreamAsync {
+
+    /** closeStream() name is chosen since Clients are AutoClosable now
+     * and Client-implementations of RemoteOutputStreamAsync have conflict of 'close' name **/
+    suspend fun closeStream()
+
+    suspend fun write(data: ByteArray, offset: Int, length: Int)
+
+    suspend fun write(dataByte: Int)
+}
+
+interface RemoteInputStreamAsync {
+
+    /** closeStream() name is chosen since Clients are AutoClosable now
+     * and Client-implementations of RemoteInputStreamAsync have conflict of 'close' name **/
+    suspend fun closeStream()
+
+    suspend fun read(length: Int): ByteArray
+
+    suspend fun read(): Int
+
+}
@@ -0,0 +1,12 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+
+interface RemoteOutputStreamAsyncClientSide : RemoteOutputStreamAsync, Client<RemoteOutputStreamAsyncServerSide>
+
+interface RemoteInputStreamClientSide : RemoteInputStreamAsync, Client<RemoteInputStreamServerSide>
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.ByteWriteChannelWrapper
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server
+
+
+interface RemoteOutputStreamAsyncServerSide : RemoteOutputStreamAsync, Server<RemoteOutputStreamAsyncServerSide> {
+    // Query messages:
+    class CloseMessage : Server.Message<RemoteOutputStreamAsyncServerSide>() {
+        override suspend fun processImpl(server: RemoteOutputStreamAsyncServerSide, sendReply: (Any?) -> Unit) =
+            server.closeStream()
+    }
+
+    class WriteMessage(val data: ByteArray, val offset: Int = -1, val length: Int = -1) :
+        Server.Message<RemoteOutputStreamAsyncServerSide>() {
+        override suspend fun processImpl(server: RemoteOutputStreamAsyncServerSide, sendReply: (Any?) -> Unit) =
+            server.write(data, offset, length)
+    }
+
+    class WriteIntMessage(val dataByte: Int) : Server.Message<RemoteOutputStreamAsyncServerSide>() {
+        override suspend fun processImpl(server: RemoteOutputStreamAsyncServerSide, sendReply: (Any?) -> Unit) =
+            server.write(dataByte)
+    }
+}
+
+
+interface RemoteInputStreamServerSide : RemoteInputStreamAsync, Server<RemoteInputStreamServerSide> {
+    // Query messages:
+    class CloseMessage : Server.Message<RemoteInputStreamServerSide>() {
+        override suspend fun processImpl(server: RemoteInputStreamServerSide, sendReply: (Any?) -> Unit) =
+            server.closeStream()
+    }
+
+    class ReadMessage(val length: Int = -1) : Server.Message<RemoteInputStreamServerSide>() {
+        override suspend fun processImpl(server: RemoteInputStreamServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(if (length == -1) server.read() else server.read(length))
+    }
+}
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.RemoteInputStream
+import org.jetbrains.kotlin.daemon.common.RemoteOutputStream
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.DefaultClientRMIWrapper
+
+class RemoteOutputStreamAsyncWrapper(val rmiOutput: RemoteOutputStream) : RemoteOutputStreamAsyncClientSide,
+    Client<RemoteOutputStreamAsyncServerSide> by DefaultClientRMIWrapper() {
+
+    override suspend fun closeStream() =
+        rmiOutput.close()
+
+    override suspend fun write(data: ByteArray, offset: Int, length: Int) =
+        rmiOutput.write(data, offset, length)
+
+    override suspend fun write(dataByte: Int) =
+        rmiOutput.write(dataByte)
+
+}
+
+class RemoteInputStreamAsyncWrapper(private val rmiInput: RemoteInputStream) : RemoteInputStreamClientSide,
+    Client<RemoteInputStreamServerSide> by DefaultClientRMIWrapper() {
+
+    override suspend fun closeStream() =
+        rmiInput.close()
+
+    override suspend fun read() =
+        rmiInput.read()
+
+    override suspend fun read(length: Int) =
+        rmiInput.read(length)
+}
+
+fun RemoteOutputStream.toClient() = RemoteOutputStreamAsyncWrapper(this)
+fun RemoteInputStream.toClient() = RemoteInputStreamAsyncWrapper(this)
@@ -0,0 +1,11 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.ReplStateFacadeAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Client
+
+interface ReplStateFacadeClientSide: ReplStateFacadeAsync, Client<ReplStateFacadeServerSide>
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.cli.common.repl.ILineId
+import org.jetbrains.kotlin.daemon.common.ReplStateFacadeAsync
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.ByteWriteChannelWrapper
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server
+
+interface ReplStateFacadeServerSide: ReplStateFacadeAsync, Server<ReplStateFacadeServerSide> {
+
+    // Query messages:
+    class GetIdMessage : Server.Message<ReplStateFacadeServerSide>() {
+        override suspend fun processImpl(server: ReplStateFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getId())
+    }
+
+    class GetHistorySizeMessage : Server.Message<ReplStateFacadeServerSide>() {
+        override suspend fun processImpl(server: ReplStateFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.getHistorySize())
+    }
+
+    class HistoryGetMessage(val index: Int) : Server.Message<ReplStateFacadeServerSide>() {
+        override suspend fun processImpl(server: ReplStateFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.historyGet(index))
+    }
+
+    class HistoryResetMessage : Server.Message<ReplStateFacadeServerSide>() {
+        override suspend fun processImpl(server: ReplStateFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.historyReset())
+    }
+
+    class HistoryResetToMessage(val id: ILineId) : Server.Message<ReplStateFacadeServerSide>() {
+        override suspend fun processImpl(server: ReplStateFacadeServerSide, sendReply: (Any?) -> Unit) =
+            sendReply(server.historyResetTo(id))
+    }
+}
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+@file:Suppress("CAST_NEVER_SUCCEEDS")
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.ByteReadChannelWrapper
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.ByteWriteChannelWrapper
+import java.io.FileInputStream
+import java.io.FileOutputStream
+import java.io.ObjectInputStream
+import java.io.ObjectOutputStream
+import java.security.*
+import java.security.spec.InvalidKeySpecException
+import java.security.spec.X509EncodedKeySpec
+
+
+const val SECURITY_TOKEN_SIZE = 128
+
+private val secureRandom = SecureRandom.getInstance("SHA1PRNG", "SUN");
+private val pairGenerator = KeyPairGenerator.getInstance("DSA", "SUN")
+
+private fun generateSecurityToken(): ByteArray {
+    val tokenBuffer = ByteArray(SECURITY_TOKEN_SIZE)
+    secureRandom.nextBytes(tokenBuffer)
+    return tokenBuffer
+}
+
+data class SecurityData(val privateKey: PrivateKey, val publicKey: PublicKey, val token: ByteArray)
+fun generateKeysAndToken() = pairGenerator.generateKeyPair().let {
+    SecurityData(it.private, it.public, generateSecurityToken())
+}
+
+private fun FileInputStream.readBytesFixedLength(n: Int): ByteArray {
+    val buffer = ByteArray(n)
+    var bytesRead = 0
+    while (bytesRead != n) {
+        bytesRead += this.read(buffer, bytesRead, n - bytesRead)
+    }
+    return buffer
+}
+
+// server part :
+fun sendTokenKeyPair(output: FileOutputStream, token: ByteArray, privateKey: PrivateKey) {
+    output.write(token)
+    ObjectOutputStream(output).use {
+        it.writeObject(privateKey)
+    }
+}
+
+private fun instantiateDsa() = Signature.getInstance("SHA1withDSA", "SUN")
+
+suspend fun getSignatureAndVerify(input: ByteReadChannelWrapper, expectedToken: ByteArray, publicKey: PublicKey): Boolean {
+    val signature = input.nextBytes()
+    val dsa = instantiateDsa()
+    dsa.initVerify(publicKey)
+    dsa.update(expectedToken, 0, SECURITY_TOKEN_SIZE)
+    val verified = dsa.verify(signature)
+    log.fine("verified : $verified")
+    return verified
+}
+
+
+// client part :
+fun readTokenKeyPairAndSign(input: FileInputStream): ByteArray {
+    val token = input.readBytesFixedLength(SECURITY_TOKEN_SIZE)
+    val privateKey = ObjectInputStream(input).use(ObjectInputStream::readObject) as PrivateKey
+    val dsa = instantiateDsa()
+    dsa.initSign(privateKey)
+    dsa.update(token, 0, SECURITY_TOKEN_SIZE)
+    return dsa.sign()
+}
+
+suspend fun sendSignature(output: ByteWriteChannelWrapper, signature: ByteArray) = output.writeBytesAndLength(signature.size, signature)
@@ -0,0 +1,21 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.daemon.common.experimental
+
+import org.jetbrains.kotlin.daemon.common.COMPILE_DAEMON_FIND_PORT_ATTEMPTS
+import org.jetbrains.kotlin.daemon.common.experimental.*
+
+fun findCallbackServerSocket() = findPortForSocket(
+    COMPILE_DAEMON_FIND_PORT_ATTEMPTS,
+    CALLBACK_SERVER_PORTS_RANGE_START,
+    CALLBACK_SERVER_PORTS_RANGE_END
+)
+
+fun findReplServerSocket() = findPortForSocket(
+    COMPILE_DAEMON_FIND_PORT_ATTEMPTS,
+    REPL_SERVER_PORTS_RANGE_START,
+    REPL_SERVER_PORTS_RANGE_END
+)
@@ -0,0 +1,260 @@
+package org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure
+
+import io.ktor.network.sockets.Socket
+import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.*
+import org.jetbrains.kotlin.daemon.common.experimental.LoopbackNetworkInterface
+import sun.net.ConnectionResetException
+import java.beans.Transient
+import java.io.IOException
+import java.io.ObjectInputStream
+import java.io.ObjectOutputStream
+import java.io.Serializable
+import java.util.logging.Logger
+import org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure.Server.*
+
+interface Client<ServerType : ServerBase> : Serializable, AutoCloseable {
+
+    @Throws(Exception::class)
+    suspend fun connectToServer()
+
+    suspend fun sendMessage(msg: AnyMessage<out ServerType>): Int // returns message unique id
+    fun sendNoReplyMessage(msg: AnyMessage<out ServerType>)
+    suspend fun <T> readMessage(id: Int): T
+
+}
+
+@Suppress("UNCHECKED_CAST")
+abstract class DefaultAuthorizableClient<ServerType : ServerBase>(
+    val serverPort: Int,
+    val serverHost: String = LoopbackNetworkInterface.loopbackInetAddressName
+) : Client<ServerType> {
+
+    val log: Logger
+        @Transient get() = Logger.getLogger("default client($serverPort)")//.also { it.setUseParentHandlers(false); }
+
+    @kotlin.jvm.Transient
+    lateinit var input: ByteReadChannelWrapper
+
+    @kotlin.jvm.Transient
+    lateinit var output: ByteWriteChannelWrapper
+
+    @kotlin.jvm.Transient
+    private var socket: Socket? = null
+
+    abstract suspend fun authorizeOnServer(serverOutputChannel: ByteWriteChannelWrapper): Boolean
+    abstract suspend fun clientHandshake(input: ByteReadChannelWrapper, output: ByteWriteChannelWrapper, log: Logger): Boolean
+
+    /*
+     The purpose of ths function is the following : a client starts sending keep-alive requests to the server.
+     It sends a request every K seconds and awaits a response after K/2 seconds.
+     In case of not receiving the response from the server we stop waiting for other responses.
+     This resolves the case when the server is down but the client is still waiting for some calculations.
+     Keep-alives run in a separate thread on client and on server, hence we can assume that the calculations on server
+     shouldn't delay keep-alive responses too much
+    */
+    abstract fun startKeepAlives()
+
+    /*
+     `delayKeepAlives` resolves another issue. Imagine that a server and a client have too many short requests/responses,
+     say, 1000 requests/responses and 0.001 seconds of latency. Then we can miss keep-alive response because of a socket workload,
+     or, say, if a scheduler decides not to schedule on a keep-alive thread.
+     In this case we say that any response can stand for a keep-alive message, as well. delayKeepAlives() serves this purpose.
+    */
+    abstract fun delayKeepAlives()
+
+    override fun close() {
+        socket?.close()
+    }
+
+    class MessageReply<T : Any>(val messageId: Int, val reply: T?) : Serializable
+
+    protected interface ReadActorQuery
+    protected data class ExpectReplyQuery(val messageId: Int, val result: CompletableDeferred<MessageReply<*>>) : ReadActorQuery
+    protected class ReceiveReplyQuery(val reply: MessageReply<*>) : ReadActorQuery
+
+    protected interface WriteActorQuery
+    protected data class SendNoreplyMessageQuery(val message: AnyMessage<*>) : WriteActorQuery
+    protected data class SendMessageQuery(val message: AnyMessage<*>, val messageId: CompletableDeferred<Any>) : WriteActorQuery
+
+    protected class StopAllRequests : ReadActorQuery, WriteActorQuery
+
+    @kotlin.jvm.Transient
+    protected lateinit var readActor: SendChannel<ReadActorQuery>
+
+    @kotlin.jvm.Transient
+    private lateinit var writeActor: SendChannel<WriteActorQuery>
+
+    override suspend fun sendMessage(msg: AnyMessage<out ServerType>): Int {
+        val id = CompletableDeferred<Any>()
+        writeActor.send(SendMessageQuery(msg, id))
+        val idVal = id.await()
+        if (idVal is IOException) {
+            throw idVal
+        }
+        return idVal as Int
+    }
+
+    override fun sendNoReplyMessage(msg: AnyMessage<out ServerType>) {
+        writeActor.offer(SendNoreplyMessageQuery(msg))
+    }
+
+    override suspend fun <T> readMessage(id: Int): T {
+        val result = CompletableDeferred<MessageReply<*>>()
+        try {
+            readActor.send(ExpectReplyQuery(id, result))
+        } catch (e: ClosedSendChannelException) {
+            throw IOException("failed to read message (channel was closed)")
+        }
+        val actualResult = result.await().reply
+        if (actualResult is IOException) {
+            throw actualResult
+        }
+        return actualResult as T
+    }
+
+    override suspend fun connectToServer() {
+
+        writeActor = GlobalScope.actor(capacity = Channel.UNLIMITED) {
+            var firstFreeMessageId = 0
+            consumeEach { query ->
+                when (query) {
+                    is SendMessageQuery -> {
+                        val id = firstFreeMessageId++
+                        try {
+                            output.writeObject(query.message.withId(id))
+                            query.messageId.complete(id)
+                        } catch (e: IOException) {
+                            query.messageId.complete(e)
+                        }
+                    }
+                    is SendNoreplyMessageQuery -> {
+                        output.writeObject(query.message.withId(-1))
+                    }
+                    is StopAllRequests -> {
+                        channel.close()
+                    }
+                }
+            }
+        }
+
+        class NextObjectQuery
+
+        val nextObjectQuery = NextObjectQuery()
+        val objectReaderActor = GlobalScope.actor<NextObjectQuery>(capacity = Channel.UNLIMITED) {
+            consumeEach {
+                try {
+                    val reply = input.nextObject()
+                    when (reply) {
+                        is ServerDownMessage<*> -> throw IOException("connection closed by server")
+                        !is MessageReply<*> -> throw IOException("contrafact message (expected MessageReply<*>)")
+                        else -> readActor.send(ReceiveReplyQuery(reply))
+                    }
+                } catch (e: IOException) {
+                    readActor.send(StopAllRequests())
+                }
+            }
+        }
+
+        readActor = GlobalScope.actor(capacity = Channel.UNLIMITED) {
+            val receivedMessages = hashMapOf<Int, MessageReply<*>>()
+            val expectedMessages = hashMapOf<Int, ExpectReplyQuery>()
+
+            fun broadcastIOException(e: IOException) {
+                channel.close()
+                expectedMessages.forEach { id, deferred ->
+                    deferred.result.complete(MessageReply(id, e))
+                }
+                expectedMessages.clear()
+                receivedMessages.clear()
+            }
+
+            consumeEach { query ->
+                when (query) {
+                    is ExpectReplyQuery -> {
+                        receivedMessages[query.messageId]?.also { reply ->
+                            query.result.complete(reply)
+                        } ?: expectedMessages.put(query.messageId, query).also {
+                            objectReaderActor.send(nextObjectQuery)
+                        }
+                    }
+                    is ReceiveReplyQuery -> {
+                        val reply = query.reply
+                        expectedMessages[reply.messageId]?.also { expectedMsg ->
+                            expectedMsg.result.complete(reply)
+                        } ?: receivedMessages.put(reply.messageId, reply).also {
+                            objectReaderActor.send(nextObjectQuery)
+                        }
+                        delayKeepAlives()
+                    }
+                    is StopAllRequests -> {
+                        broadcastIOException(IOException("KeepAlive failed"))
+                        writeActor.send(StopAllRequests())
+                    }
+                }
+            }
+        }
+
+
+
+        try {
+            socket = LoopbackNetworkInterface.clientLoopbackSocketFactoryKtor.createSocket(
+                serverHost,
+                serverPort
+            )
+        } catch (e: Throwable) {
+            close()
+            throw e
+        }
+        socket?.openIO(log)?.also {
+            input = it.input
+            output = it.output
+            if (!clientHandshake(input, output, log)) {
+                throw ConnectionResetException("failed to establish connection with server (handshake failed)")
+            }
+            if (!authorizeOnServer(output)) {
+                throw ConnectionResetException("failed to establish connection with server (authorization failed)")
+            }
+        }
+
+        startKeepAlives()
+
+    }
+
+    @Throws(ClassNotFoundException::class, IOException::class)
+    private fun readObject(aInputStream: ObjectInputStream) {
+        aInputStream.defaultReadObject()
+        println("connecting...")
+        runBlocking { connectToServer() }
+        println("connectED")
+    }
+
+    @Throws(IOException::class)
+    private fun writeObject(aOutputStream: ObjectOutputStream) {
+        aOutputStream.defaultWriteObject()
+    }
+
+}
+
+class DefaultClient<ServerType : ServerBase>(
+    serverPort: Int,
+    serverHost: String = LoopbackNetworkInterface.loopbackInetAddressName
+) : DefaultAuthorizableClient<ServerType>(serverPort, serverHost) {
+    override suspend fun clientHandshake(input: ByteReadChannelWrapper, output: ByteWriteChannelWrapper, log: Logger) = true
+    override suspend fun authorizeOnServer(output: ByteWriteChannelWrapper): Boolean = true
+    override fun startKeepAlives() {}
+    override fun delayKeepAlives() {}
+}
+
+class DefaultClientRMIWrapper<ServerType : ServerBase> : Client<ServerType> {
+
+    override suspend fun connectToServer() {}
+    override suspend fun sendMessage(msg: AnyMessage<out ServerType>) =
+        throw UnsupportedOperationException("sendMessage is not supported for RMI wrappers")
+
+    override fun sendNoReplyMessage(msg: AnyMessage<out ServerType>) =
+        throw UnsupportedOperationException("sendMessage is not supported for RMI wrappers")
+
+    override suspend fun <T> readMessage(id: Int) = throw UnsupportedOperationException("readMessage is not supported for RMI wrappers")
+    override fun close() {}
+}
@@ -0,0 +1,208 @@
+package org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure
+
+import io.ktor.network.sockets.ServerSocket
+import io.ktor.network.sockets.Socket
+import kotlinx.coroutines.*
+import org.jetbrains.kotlin.daemon.common.experimental.*
+import java.io.Serializable
+import java.util.concurrent.TimeUnit
+import java.util.logging.Logger
+
+/*
+ * Copyright 2000-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+data class ServerSocketWrapper(val port: Int, val socket: ServerSocket)
+
+interface ServerBase
+
+@Suppress("UNCHECKED_CAST")
+interface Server<out T : ServerBase> : ServerBase {
+
+    val serverSocketWithPort: ServerSocketWrapper
+    val serverPort: Int
+        get() = serverSocketWithPort.port
+
+    private val log: Logger
+        get() = Logger.getLogger("default server($serverPort)")
+
+    enum class State {
+        WORKING, CLOSED, ERROR, DOWNING, UNVERIFIED
+    }
+
+    fun processMessage(msg: AnyMessage<in T>, output: ByteWriteChannelWrapper): State =
+        when (msg) {
+            is Message<in T> -> State.WORKING.also {
+                msg.process(this as T, output)
+            }
+            is EndConnectionMessage<in T> -> {
+                State.CLOSED
+            }
+            is ServerDownMessage<in T> -> State.CLOSED
+            else -> State.ERROR
+        }
+
+    // TODO: replace GlobalScope here and below with smth. more explicit
+    fun attachClient(client: Socket): Deferred<State> = GlobalScope.async {
+        val (input, output) = client.openIO(log)
+        if (!serverHandshake(input, output, log)) {
+            return@async State.UNVERIFIED
+        }
+        if (!checkClientCanReadFile(input)) {
+            return@async State.UNVERIFIED
+        }
+        clients[client] = ClientInfo(client, input, output)
+        var finalState = State.WORKING
+        val keepAliveAcknowledgement = KeepAliveAcknowledgement<T>()
+        loop@
+        while (true) {
+            val message = input.nextObject()
+            when (message) {
+                is ServerDownMessage<*> -> {
+                    shutdownClient(client)
+                    break@loop
+                }
+                is KeepAliveMessage<*> -> State.WORKING.also {
+                    output.writeObject(
+                        DefaultAuthorizableClient.MessageReply(
+                            message.messageId!!,
+                            keepAliveAcknowledgement
+                        )
+                    )
+                }
+                !is AnyMessage<*> -> {
+                    finalState = State.ERROR
+                    break@loop
+                }
+                else -> {
+                    val state = processMessage(message as AnyMessage<T>, output)
+                    when (state) {
+                        State.WORKING -> continue@loop
+                        State.ERROR -> {
+                            finalState = State.ERROR
+                            break@loop
+                        }
+                        else -> {
+                            finalState = state
+                            break@loop
+                        }
+                    }
+                }
+            }
+        }
+        finalState
+    }
+
+    abstract class AnyMessage<ServerType : ServerBase> : Serializable {
+        var messageId: Int? = null
+        fun withId(id: Int): AnyMessage<ServerType> {
+            messageId = id
+            return this
+        }
+    }
+
+    abstract class Message<ServerType : ServerBase> : AnyMessage<ServerType>() {
+        fun process(server: ServerType, output: ByteWriteChannelWrapper) = GlobalScope.async {
+            log.fine("$server starts processing ${this@Message}")
+            processImpl(server, {
+                log.fine("$server finished processing ${this@Message}, sending output")
+                GlobalScope.async {
+                    log.fine("$server starts sending ${this@Message} to output")
+                    output.writeObject(DefaultAuthorizableClient.MessageReply(messageId ?: -1, it))
+                    log.fine("$server finished sending ${this@Message} to output")
+                }
+            })
+        }
+
+        abstract suspend fun processImpl(server: ServerType, sendReply: (Any?) -> Unit)
+    }
+
+    class EndConnectionMessage<ServerType : ServerBase> : AnyMessage<ServerType>()
+
+    class KeepAliveAcknowledgement<ServerType : ServerBase> : AnyMessage<ServerType>()
+
+    class KeepAliveMessage<ServerType : ServerBase> : AnyMessage<ServerType>()
+
+    class ServerDownMessage<ServerType : ServerBase> : AnyMessage<ServerType>()
+
+    data class ClientInfo(val socket: Socket, val input: ByteReadChannelWrapper, val output: ByteWriteChannelWrapper)
+
+    val clients: HashMap<Socket, ClientInfo>
+
+    private fun dealWithClient(client: Socket) = GlobalScope.async {
+        val state = attachClient(client).await()
+        when (state) {
+            State.CLOSED, State.UNVERIFIED -> shutdownClient(client)
+            State.DOWNING -> shutdownServer()
+            else -> shutdownClient(client)
+        }
+    }
+
+    fun runServer(): Deferred<Unit> {
+        val serverSocket = serverSocketWithPort.socket
+        return GlobalScope.async {
+            serverSocket.use {
+                while (true) {
+                    dealWithClient(serverSocket.accept())
+                }
+            }
+        }
+    }
+
+    fun shutdownServer() {
+        clients.forEach { socket, info ->
+            runBlockingWithTimeout {
+                info.output.writeObject(ServerDownMessage<T>())
+                info.output.close()
+            }
+            socket.close()
+        }
+        clients.clear()
+        serverSocketWithPort.socket.close()
+    }
+
+    private fun shutdownClient(client: Socket) {
+        clients.remove(client)
+        client.close()
+    }
+
+    /*
+        This function writes some message in the server file, and awaits the confirmation from the client that it has read the message
+        correctly. The purpose here is to check whether the client can actually access file system and read file contents.
+    */
+    suspend fun checkClientCanReadFile(clientInputChannel: ByteReadChannelWrapper): Boolean = true
+
+    suspend fun serverHandshake(input: ByteReadChannelWrapper, output: ByteWriteChannelWrapper, log: Logger) = true
+
+}
+
+fun <T> runBlockingWithTimeout(timeout: Long = AUTH_TIMEOUT_IN_MILLISECONDS, block: suspend () -> T) =
+    runBlocking { runWithTimeout(timeout = timeout) { block() } }
+
+//@Throws(TimeoutException::class)
+suspend fun <T> runWithTimeout(
+    timeout: Long = AUTH_TIMEOUT_IN_MILLISECONDS,
+    unit: TimeUnit = TimeUnit.MILLISECONDS,
+    block: suspend CoroutineScope.() -> T
+): T? = withTimeoutOrNull(unit.toMillis(timeout)) { block() }
+
+//@Throws(ConnectionResetException::class)
+suspend fun tryAcquireHandshakeMessage(input: ByteReadChannelWrapper, log: Logger): Boolean {
+    val bytes = runWithTimeout {
+        input.nextBytes()
+    } ?: return false
+    if (bytes.zip(FIRST_HANDSHAKE_BYTE_TOKEN).any { it.first != it.second }) {
+        return false
+    }
+    return true
+}
+
+
+//@Throws(ConnectionResetException::class)
+suspend fun trySendHandshakeMessage(output: ByteWriteChannelWrapper, log: Logger): Boolean {
+    runWithTimeout {
+        output.writeBytesAndLength(FIRST_HANDSHAKE_BYTE_TOKEN.size, FIRST_HANDSHAKE_BYTE_TOKEN)
+    } ?: return false
+    return true
+}
@@ -0,0 +1,217 @@
+package org.jetbrains.kotlin.daemon.common.experimental.socketInfrastructure
+
+import io.ktor.network.sockets.Socket
+import io.ktor.network.sockets.openReadChannel
+import io.ktor.network.sockets.openWriteChannel
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.actor
+import kotlinx.coroutines.channels.consumeEach
+import kotlinx.coroutines.io.*
+import kotlinx.io.core.readBytes
+import java.io.*
+import java.nio.ByteBuffer
+import java.util.logging.Logger
+
+private val DEFAULT_BYTE_ARRAY = byteArrayOf(0, 0, 0, 0)
+
+class ByteReadChannelWrapper(readChannel: ByteReadChannel, private val log: Logger) {
+
+    private interface ReadQuery
+
+    private open class BytesQuery(val bytes: CompletableDeferred<ByteArray>) : ReadQuery
+
+    private class SerObjectQuery(val obj: CompletableDeferred<Any?>) : ReadQuery
+
+    suspend fun readLength(readChannel: ByteReadChannel) =
+        if (readChannel.isClosedForRead)
+            null
+        else
+            try {
+                readChannel.readPacket(4).readBytes()
+            } catch (e: Exception) {
+                log.fine("failed to read message length, ${e.message}")
+                null
+            }
+
+    suspend fun readPacket(length: Int, readChannel: ByteReadChannel) =
+        try {
+            readChannel.readPacket(
+                length
+            ).readBytes()
+        } catch (e: Exception) {
+            log.fine("failed to read packet (${e.message})")
+            null
+        }
+
+    // TODO : replace GlobalScope with something more explicit here and below.
+    private val readActor = GlobalScope.actor<ReadQuery>(capacity = Channel.UNLIMITED) {
+        consumeEach { message ->
+            if (!readChannel.isClosedForRead) {
+                readLength(readChannel)?.let { messageLength ->
+                    when (message) {
+                        is BytesQuery -> message.bytes.complete(
+                            readChannel.readPacket(
+                                getLength(messageLength)
+                            ).readBytes()
+                        )
+
+                        is SerObjectQuery -> message.obj.complete(
+                            getObject(
+                                getLength(messageLength),
+                                { len -> readPacket(len, readChannel) }
+                            )
+                        )
+
+                        else -> {
+                        }
+                    }
+                }
+            } else {
+                log.fine("read chanel closed " + log.name)
+            }
+        }
+    }
+
+    private fun getLength(packet: ByteArray): Int {
+        val (b1, b2, b3, b4) = packet.map(Byte::toInt)
+        return (0xFF and b1 shl 24 or (0xFF and b2 shl 16) or
+                (0xFF and b3 shl 8) or (0xFF and b4)).also { log.fine("   $it") }
+    }
+
+    /** first reads <t>length</t> token (4 bytes) and then -- reads <t>length</t> bytes.
+     * after deafault timeout returns <tt>DEFAULT_BYTE_ARRAY</tt> */
+    suspend fun nextBytes(): ByteArray = runWithTimeout {
+        val expectedBytes = CompletableDeferred<ByteArray>()
+        readActor.send(BytesQuery(expectedBytes))
+        expectedBytes.await()
+    } ?: DEFAULT_BYTE_ARRAY
+
+    private suspend fun getObject(length: Int, readPacket: suspend (Int) -> ByteArray?): Any? =
+        if (length >= 0) {
+            readPacket(length)?.let { bytes ->
+                ObjectInputStream(
+                    ByteArrayInputStream(bytes)
+                ).use {
+                    it.readObject()
+                }
+            }
+        } else { // optimize for long strings!
+            readPacket(-length)?.let { bytes ->
+                String(
+                    ByteArrayInputStream(
+                        bytes
+                    ).readBytes()
+                )
+            }
+        }
+
+    /** first reads <t>length</t> token (4 bytes), then reads <t>length</t> bytes and returns deserialized object */
+    suspend fun nextObject(): Any? {
+        val obj = CompletableDeferred<Any?>()
+        readActor.send(SerObjectQuery(obj))
+        val result = obj.await()
+        if (result is Server.ServerDownMessage<*>) {
+            throw IOException("connection closed by server")
+        }
+        return result
+    }
+
+}
+
+class ByteWriteChannelWrapper(writeChannel: ByteWriteChannel, private val log: Logger) {
+
+    private interface WriteActorQuery
+
+    private open class ByteData(val bytes: ByteArray) : WriteActorQuery {
+        open fun toByteArray(): ByteArray = bytes
+    }
+
+    private class ObjectWithLength(val lengthBytes: ByteArray, bytes: ByteArray) : ByteData(bytes) {
+        override fun toByteArray() = lengthBytes + bytes
+    }
+
+    private class CloseMessage : WriteActorQuery
+
+    private suspend fun tryWrite(b: ByteArray, writeChannel: ByteWriteChannel) {
+        if (!writeChannel.isClosedForWrite) {
+            try {
+                writeChannel.writeFully(b)
+            } catch (e: Exception) {
+                log.fine("failed to print message, ${e.message}")
+            }
+        } else {
+            log.fine("closed chanel (write)")
+        }
+    }
+
+    private val writeActor = GlobalScope.actor<WriteActorQuery>(capacity = Channel.UNLIMITED) {
+        consumeEach { message ->
+            if (!writeChannel.isClosedForWrite) {
+                when (message) {
+                    is CloseMessage -> {
+                        log.fine("${log.name} closing chanel...")
+                        writeChannel.close()
+                    }
+                    is ByteData -> {
+                        tryWrite(message.toByteArray(), writeChannel)
+                        if (!writeChannel.isClosedForWrite) {
+                            try {
+                                writeChannel.flush()
+                            } catch (e: Exception) {
+                                log.fine("failed to flush byte write chanel")
+                            }
+                        }
+                    }
+                }
+            } else {
+                log.fine("${log.name} write chanel closed")
+            }
+        }
+    }
+
+    suspend fun writeBytesAndLength(length: Int, bytes: ByteArray) {
+        writeActor.send(
+            ObjectWithLength(
+                getLengthBytes(length),
+                bytes
+            )
+        )
+    }
+
+    private suspend fun writeObjectImpl(obj: Any?) =
+        ByteArrayOutputStream().use { bos ->
+            ObjectOutputStream(bos).use { objOut ->
+                objOut.writeObject(obj)
+                objOut.flush()
+                val bytes = bos.toByteArray()
+                writeBytesAndLength(bytes.size, bytes)
+            }
+        }
+
+    private suspend fun writeString(s: String) = writeBytesAndLength(-s.length, s.toByteArray())
+
+    fun getLengthBytes(length: Int) =
+        ByteBuffer
+            .allocate(4)
+            .putInt(length)
+            .array()
+
+    suspend fun writeObject(obj: Any?) {
+        if (obj is String) writeString(obj)
+        else writeObjectImpl(obj)
+    }
+
+    suspend fun close() = writeActor.send(CloseMessage())
+
+}
+
+fun ByteReadChannel.toWrapper(log: Logger) = ByteReadChannelWrapper(this, log)
+fun ByteWriteChannel.toWrapper(log: Logger) = ByteWriteChannelWrapper(this, log)
+
+fun Socket.openAndWrapReadChannel(log: Logger) = this.openReadChannel().toWrapper(log)
+fun Socket.openAndWrapWriteChannel(log: Logger) = this.openWriteChannel().toWrapper(log)
+
+data class IOPair(val input: ByteReadChannelWrapper, val output: ByteWriteChannelWrapper)
+
+fun Socket.openIO(log: Logger) = IOPair(this.openAndWrapReadChannel(log), this.openAndWrapWriteChannel(log))