Implementation of multiple daemon instances management

2015-08-31 14:29:29 +02:00
parent f6c6087fe3
commit 406bfba473
5 changed files with 183 additions and 62 deletions
@@ -19,11 +19,8 @@ package org.jetbrains.kotlin.rmi.kotlinr
 import org.jetbrains.kotlin.load.kotlin.incremental.components.IncrementalCache
 import org.jetbrains.kotlin.modules.TargetId
 import org.jetbrains.kotlin.rmi.*
-import java.io.File
-import java.io.OutputStream
-import java.io.PrintStream
+import java.io.*
 import java.rmi.ConnectException
-import java.rmi.Remote
 import java.rmi.registry.LocateRegistry
 import java.util.concurrent.Semaphore
 import java.util.concurrent.TimeUnit
@@ -43,20 +40,39 @@ public class KotlinCompilerClient {
    companion object {

        val DAEMON_DEFAULT_STARTUP_TIMEOUT_MS = 10000L
+        val DAEMON_CONNECT_CYCLE_ATTEMPTS = 3

-        private fun connectToService(compilerId: CompilerId, daemonOptions: DaemonOptions, errStream: PrintStream): CompileService? {
-
-            val compilerObj = connectToDaemon(compilerId, daemonOptions, errStream) ?: return null // no registry - no daemon running
-            return compilerObj as? CompileService ?:
-                throw ClassCastException("Unable to cast compiler service, actual class received: ${compilerObj.javaClass}")
+        private fun tryFindDaemon(registryDir: File, compilerId: CompilerId, errStream: PrintStream): CompileService? {
+            val classPathDigest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest()
+            var port = 0
+            val daemons = registryDir.walk()
+                    .map { Pair(it, makeRunFilenameRegex(digest = classPathDigest, port = "(\\d+)").match(it.name)?.groups?.get(1)?.value?.toInt() ?: 0) }
+                    .filter { it.second != 0 }
+                    .map {
+                        val daemon = tryConnectToDaemon(it.second, errStream)
+                        // cleaning orphaned file; note: daemon should shut itself down if it detects that the run file is deleted
+                        if (daemon == null && !it.first.delete()) {
+                            errStream.println("WARNING: Unable to delete seemingly orphaned file '', cleanup recommended")
+                        }
+                        daemon
+                    }
+                    .filterNotNull()
+                    .toList()
+            return when (daemons.size()) {
+                0 -> null
+                1 -> daemons.first()
+                else -> throw IllegalStateException("Multiple daemons serving the same compiler, reset with the cleanup required")
+                // TODO: consider implementing automatic recovery instead, e.g. getting the youngest or least used daemon and shut down others
+            }
        }

-        private fun connectToDaemon(compilerId: CompilerId, daemonOptions: DaemonOptions, errStream: PrintStream): Remote? {
+        private fun tryConnectToDaemon(port: Int, errStream: PrintStream): CompileService? {
            try {
-                val daemon = LocateRegistry.getRegistry("localhost", daemonOptions.port)
+                val daemon = LocateRegistry.getRegistry("localhost", port)
                        ?.lookup(COMPILER_SERVICE_RMI_NAME)
                if (daemon != null)
-                    return daemon
+                    return daemon as? CompileService ?:
+                       throw ClassCastException("Unable to cast compiler service, actual class received: ${daemon.javaClass}")
                errStream.println("[daemon client] daemon not found")
            }
            catch (e: ConnectException) {
@@ -93,7 +109,7 @@ public class KotlinCompilerClient {
                        daemon.getInputStream()
                              .reader()
                              .forEachLine {
-                                  if (daemonOptions.startEcho.isNotEmpty() && it.contains(daemonOptions.startEcho)) {
+                                  if (daemonOptions.runFilesPath.isNotEmpty() && it.contains(daemonOptions.runFilesPath)) {
                                      isEchoRead.release()
                                      return@forEachLine
                                  }
@@ -102,7 +118,7 @@ public class KotlinCompilerClient {
                    }
            try {
                // trying to wait for process
-                if (daemonOptions.startEcho.isNotEmpty()) {
+                if (daemonOptions.runFilesPath.isNotEmpty()) {
                    val daemonStartupTimeout = System.getProperty(COMPILE_DAEMON_STARTUP_TIMEOUT_PROPERTY)?.let {
                            try {
                                it.toLong()
@@ -140,29 +156,93 @@ public class KotlinCompilerClient {
                   (localId.compilerDigest.isEmpty() || remoteId.compilerDigest.isEmpty() || localId.compilerDigest == remoteId.compilerDigest)
        }

-        public fun connectToCompileService(compilerId: CompilerId, daemonJVMOptions: DaemonJVMOptions, daemonOptions: DaemonOptions, errStream: PrintStream, autostart: Boolean = true, checkId: Boolean = true): CompileService? {
-            val service = connectToService(compilerId, daemonOptions, errStream)
-            if (service != null) {
-                if (!checkId || checkCompilerId(service, compilerId, errStream)) {
-                    errStream.println("[daemon client] found the suitable daemon")
-                    return service
+
+        class FileBasedLock(compilerId: CompilerId, daemonOptions: DaemonOptions) {
+            private val lockFile: File =
+                    File(daemonOptions.runFilesPath,
+                         makeRunFilenameString(ts = "lock",
+                                               digest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest(),
+                                               port = "0"))
+
+            private var locked: Boolean = acquireLockFile(lockFile)
+
+            public fun isLocked(): Boolean = locked
+
+            synchronized public fun release(): Unit {
+                if (locked) {
+                    lock?.release()
+                    channel?.close()
+                    lockFile.delete()
+                    locked = false
                }
-                errStream.println("[daemon client] compiler identity don't match: " + compilerId.mappers.flatMap { it.toArgs("") }.joinToString(" "))
-                if (!autostart) return null;
-                errStream.println("[daemon client] shutdown the daemon")
-                service.shutdown()
-                // TODO: find more reliable way
-                Thread.sleep(1000)
-                errStream.println("[daemon client] daemon shut down correctly, restarting")
-            }
-            else {
-                if (!autostart) return null;
-                else errStream.println("[daemon client] cannot connect to Compile Daemon, trying to start")
            }

-            startDaemon(compilerId, daemonJVMOptions, daemonOptions, errStream)
-            errStream.println("[daemon client] daemon started, trying to connect")
-            return connectToService(compilerId, daemonOptions, errStream)
+            private val channel = if (locked) RandomAccessFile(lockFile, "rw").channel else null
+            private val lock = channel?.lock()
+
+            synchronized private fun acquireLockFile(lockFile: File): Boolean {
+                if (lockFile.createNewFile()) return true
+                try {
+                    // attempt to delete if file is orphaned
+                    if (lockFile.delete() && lockFile.createNewFile())
+                        return true // if orphaned file deleted assuming that the probability of
+                }
+                catch (e: IOException) {
+                    // Ignoring it - assuming it is another client process owning it
+                }
+                var attempts = 0L
+                while (lockFile.exists() && attempts++ * COMPILE_DAEMON_STARTUP_LOCK_TIMEOUT_CHECK_MS < COMPILE_DAEMON_STARTUP_LOCK_TIMEOUT_MS) {
+                    Thread.sleep(COMPILE_DAEMON_STARTUP_LOCK_TIMEOUT_CHECK_MS)
+                }
+                if (lockFile.exists())
+                    throw IOException("Timeout waiting the release of the lock file '${lockFile.absolutePath}")
+
+                return lockFile.createNewFile()
+            }
+
+        }
+
+        public fun connectToCompileService(compilerId: CompilerId, daemonJVMOptions: DaemonJVMOptions, daemonOptions: DaemonOptions, errStream: PrintStream, autostart: Boolean = true, checkId: Boolean = true): CompileService? {
+            var attempts = 0
+            var fileLock: FileBasedLock? = null
+            try {
+                while (attempts++ < DAEMON_CONNECT_CYCLE_ATTEMPTS) {
+                    val service = tryFindDaemon(File(daemonOptions.runFilesPath), compilerId, errStream)
+                    if (service != null) {
+                        if (!checkId || checkCompilerId(service, compilerId, errStream)) {
+                            errStream.println("[daemon client] found the suitable daemon")
+                            return service
+                        }
+                        errStream.println("[daemon client] compiler identity don't match: " + compilerId.mappers.flatMap { it.toArgs("") }.joinToString(" "))
+                        if (!autostart) return null
+                        errStream.println("[daemon client] shutdown the daemon")
+                        service.shutdown()
+                        // TODO: find more reliable way
+                        Thread.sleep(1000)
+                        errStream.println("[daemon client] daemon shut down correctly, restarting")
+                    }
+                    else {
+                        if (!autostart) return null
+                        errStream.println("[daemon client] cannot connect to Compile Daemon, trying to start")
+                    }
+
+                    if (fileLock == null || !fileLock.isLocked()) {
+                        File(daemonOptions.runFilesPath).mkdirs()
+                        fileLock = FileBasedLock(compilerId, daemonOptions)
+                        // need to check the daemons again here, because of possible racing conditions
+                        // note: the algorithm could be simpler if we'll acquire lock right from the beginning, but it may be costly
+                        attempts--
+                    }
+                    else {
+                        startDaemon(compilerId, daemonJVMOptions, daemonOptions, errStream)
+                        errStream.println("[daemon client] daemon started, trying to connect")
+                    }
+                }
+            }
+            finally {
+                fileLock?.release()
+            }
+            return null
        }

        public fun shutdownCompileService(daemonOptions: DaemonOptions): Unit {
@@ -22,22 +22,30 @@ import java.lang.management.ManagementFactory
 import java.security.DigestInputStream
 import java.security.MessageDigest
 import kotlin.reflect.KMutableProperty1
+import kotlin.text.Regex


 public val COMPILER_JAR_NAME: String = "kotlin-compiler.jar"
 public val COMPILER_SERVICE_RMI_NAME: String = "KotlinJvmCompilerService"
 public val COMPILER_DAEMON_CLASS_FQN: String = "org.jetbrains.kotlin.rmi.service.CompileDaemon"
-public val COMPILE_DAEMON_DEFAULT_PORT: Int = 17031
-public val COMPILE_DAEMON_ENABLED_PROPERTY: String ="kotlin.daemon.enabled"
-public val COMPILE_DAEMON_JVM_OPTIONS_PROPERTY: String ="kotlin.daemon.jvm.options"
-public val COMPILE_DAEMON_OPTIONS_PROPERTY: String ="kotlin.daemon.options"
+public val COMPILE_DAEMON_FIND_PORT_ATTEMPTS: Int = 10
+public val COMPILE_DAEMON_PORTS_RANGE_START: Int = 17001
+public val COMPILE_DAEMON_PORTS_RANGE_END: Int = 18000
+public val COMPILE_DAEMON_STARTUP_LOCK_TIMEOUT_MS: Long = 10000L
+public val COMPILE_DAEMON_STARTUP_LOCK_TIMEOUT_CHECK_MS: Long = 100L
+public val COMPILE_DAEMON_ENABLED_PROPERTY: String = "kotlin.daemon.enabled"
+public val COMPILE_DAEMON_JVM_OPTIONS_PROPERTY: String = "kotlin.daemon.jvm.options"
+public val COMPILE_DAEMON_OPTIONS_PROPERTY: String = "kotlin.daemon.options"
+public val COMPILE_DAEMON_CMDLINE_OPTIONS_PREFIX: String = "--daemon-"
 public val COMPILE_DAEMON_STARTUP_TIMEOUT_PROPERTY: String ="kotlin.daemon.startup.timeout"
-public val COMPILE_DAEMON_CMDLINE_OPTIONS_PREFIX: String ="--daemon-"
 public val COMPILE_DAEMON_TIMEOUT_INFINITE_S: Int = 0
 public val COMPILE_DAEMON_MEMORY_THRESHOLD_INFINITE: Long = 0L

 val COMPILER_ID_DIGEST = "MD5"

+public fun makeRunFilenameString(ts: String, digest: String, port: String, esc: String = ""): String = "kotlin-daemon$esc.$ts$esc.$digest$esc.$port$esc.run"
+public fun makeRunFilenameRegex(ts: String = "[0-9-]+", digest: String = "[0-9a-f]+", port: String = "\\d+"): Regex = makeRunFilenameString(ts, digest, port, esc = "\\").toRegex()
+

 open class PropMapper<C, V, P: KMutableProperty1<C, V>>(val dest: C,
                                                        val prop: P,
@@ -167,17 +175,15 @@ public data class DaemonJVMOptions(
 }

 public data class DaemonOptions(
-        public var port: Int = COMPILE_DAEMON_DEFAULT_PORT,
+        public var runFilesPath: String = File(System.getProperty("java.io.tmpdir"), "kotlin_daemon").absolutePath,
        public var autoshutdownMemoryThreshold: Long = COMPILE_DAEMON_MEMORY_THRESHOLD_INFINITE,
-        public var autoshutdownIdleSeconds: Int = COMPILE_DAEMON_TIMEOUT_INFINITE_S,
-        public var startEcho: String = COMPILER_SERVICE_RMI_NAME
+        public var autoshutdownIdleSeconds: Int = COMPILE_DAEMON_TIMEOUT_INFINITE_S
 ) : OptionsGroup {

    override val mappers: List<PropMapper<*, *, *>>
-        get() = listOf( PropMapper(this, ::port, fromString = { it.toInt() }),
+        get() = listOf( PropMapper(this, ::runFilesPath, fromString = { it.trim('"') }),
                        PropMapper(this, ::autoshutdownMemoryThreshold, fromString = { it.toLong() }, skipIf = { it == 0L }),
-                        PropMapper(this, ::autoshutdownIdleSeconds, fromString = { it.toInt() }, skipIf = { it == 0 }),
-                        PropMapper(this, ::startEcho, fromString = { it.trim('"') }))
+                        PropMapper(this, ::autoshutdownIdleSeconds, fromString = { it.toInt() }, skipIf = { it == 0 }))
 }


@@ -205,13 +211,20 @@ fun updateEntryDigest(entry: File, md: MessageDigest) {
    }
 }

+jvmName("getFilesClasspathDigest_Files")
 fun Iterable<File>.getFilesClasspathDigest(): String {
    val md = MessageDigest.getInstance(COMPILER_ID_DIGEST)
    this.forEach { updateEntryDigest(it, md) }
    return md.digest().joinToString("", transform = { "%02x".format(it) })
 }

-fun Iterable<String>.getClasspathDigest(): String = map { File(it) }.getFilesClasspathDigest()
+jvmName("getFilesClasspathDigest_Strings")
+fun Iterable<String>.getFilesClasspathDigest(): String = map { File(it) }.getFilesClasspathDigest()
+
+fun Iterable<String>.distinctStringsDigest(): String =
+    MessageDigest.getInstance(COMPILER_ID_DIGEST)
+            .digest(this.distinct().sort().joinToString("").toByteArray())
+            .joinToString("", transform = { "%02x".format(it) })


 public data class CompilerId(
@@ -227,7 +240,7 @@ public data class CompilerId(
                        StringPropMapper(this, ::compilerVersion))

    public fun updateDigest() {
-        compilerDigest = compilerClasspath.getClasspathDigest()
+        compilerDigest = compilerClasspath.getFilesClasspathDigest()
    }

    companion object {
@@ -17,23 +17,22 @@
 package org.jetbrains.kotlin.rmi.service

 import org.jetbrains.kotlin.cli.jvm.K2JVMCompiler
-import org.jetbrains.kotlin.rmi.COMPILE_DAEMON_CMDLINE_OPTIONS_PREFIX
-import org.jetbrains.kotlin.rmi.CompilerId
-import org.jetbrains.kotlin.rmi.DaemonOptions
-import org.jetbrains.kotlin.rmi.filterExtractProps
+import org.jetbrains.kotlin.rmi.*
 import org.jetbrains.kotlin.service.CompileServiceImpl
+import java.io.File
 import java.io.IOException
 import java.io.OutputStream
 import java.io.PrintStream
 import java.lang.management.ManagementFactory
 import java.net.URLClassLoader
+import java.rmi.RemoteException
 import java.rmi.registry.LocateRegistry
+import java.rmi.registry.Registry
 import java.text.SimpleDateFormat
 import java.util.*
 import java.util.jar.Manifest
 import java.util.logging.LogManager
 import java.util.logging.Logger
-import kotlin.platform.platformStatic

 class LogStream(name: String) : OutputStream() {

@@ -107,7 +106,7 @@ public class CompileDaemon {
            return null
        }

-        platformStatic public fun main(args: Array<String>) {
+        jvmStatic public fun main(args: Array<String>) {

            log.info("Kotlin compiler daemon version " + (loadVersionFromResource() ?: "<unknown>"))
            log.info("daemon JVM args: " + ManagementFactory.getRuntimeMXBean().inputArguments.joinToString(" "))
@@ -132,13 +131,23 @@ public class CompileDaemon {
 //
 //            setDaemonPpermissions(daemonOptions.port)

-            val registry = LocateRegistry.createRegistry(daemonOptions.port);
-            val compiler = K2JVMCompiler()
+            val (registry, port) = createRegistry(COMPILE_DAEMON_FIND_PORT_ATTEMPTS)
+            val runFileDir = File(daemonOptions.runFilesPath)
+            runFileDir.mkdirs()
+            val runFile = File(runFileDir,
+                 makeRunFilenameString(ts = "%tFT%<tRZ".format(Calendar.getInstance(TimeZone.getTimeZone("Z"))),
+                                       digest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest(),
+                                       port = port.toString()))
+            if (!runFile.createNewFile()) {
+                throw IllegalStateException("Unable to create run file '${runFile.absolutePath}'")
+            }
+            runFile.deleteOnExit()

+            val compiler = K2JVMCompiler()
            CompileServiceImpl(registry, compiler, compilerId, daemonOptions)

-            if (daemonOptions.startEcho.isNotEmpty())
-                println(daemonOptions.startEcho)
+            if (daemonOptions.runFilesPath.isNotEmpty())
+                println(runFile.name)

            // this stops redirected streams reader(s) on the client side and prevent some situations with hanging threads
            System.out.close()
@@ -147,5 +156,23 @@ public class CompileDaemon {
            System.setErr(PrintStream(LogStream("stderr")))
            System.setOut(PrintStream(LogStream("stdout")))
        }
+
+        val random = Random()
+
+        private fun createRegistry(attempts: Int) : Pair<Registry, Int> {
+            var i = 0
+            var lastException: RemoteException? = null
+            while (i++ < attempts) {
+                val port = random.nextInt(COMPILE_DAEMON_PORTS_RANGE_END - COMPILE_DAEMON_PORTS_RANGE_START) + COMPILE_DAEMON_PORTS_RANGE_START
+                try {
+                    return Pair(LocateRegistry.createRegistry(port), port)
+                }
+                catch (e: RemoteException) {
+                    // assuming that the port is already taken
+                    lastException = e
+                }
+            }
+            throw IllegalStateException("Cannot find free port in $attempts attempts", lastException)
+        }
    }
 }
@@ -115,6 +115,7 @@ class CompileServiceImpl<Compiler: CLICompiler<*>>(
            log.info("Used memory: $endMem kb (${"%+d".format(endMem - startMem)} kb)")
            return res
        }
+        // TODO: consider possibilities to handle OutOfMemory
        catch (e: Exception) {
            log.info("Error: $e")
            throw e
@@ -26,21 +26,21 @@ import org.jetbrains.kotlin.rmi.kotlinr.KotlinCompilerClient
 import org.jetbrains.kotlin.test.JetTestUtils
 import java.io.ByteArrayOutputStream
 import java.io.File
-
-val KOTLIN_DAEMON_TEST_PORT = 19753
+import java.io.IOException
+import java.util.*

 public class CompilerDaemonTest : KotlinIntegrationTestBase() {

    data class CompilerResults(val resultCode: Int, val out: String)

-    val daemonOptions = DaemonOptions(port = KOTLIN_DAEMON_TEST_PORT)
-    val daemonLaunchingOptions = DaemonJVMOptions()
+    val daemonOptions = DaemonOptions(runFilesPath = tmpdir.absolutePath)
+    val daemonJVMOptions = DaemonJVMOptions()
    val compilerId by lazy { CompilerId.makeCompilerId( File(KotlinIntegrationTestBase.getCompilerLib(), "kotlin-compiler.jar"),
                                                        File("dependencies/bootstrap-compiler/Kotlin/kotlinc/lib/kotlin-runtime.jar"),
                                                        File("dependencies/bootstrap-compiler/Kotlin/kotlinc/lib/kotlin-reflect.jar")) }

    private fun compileOnDaemon(args: Array<out String>): CompilerResults {
-        val daemon = KotlinCompilerClient.connectToCompileService(compilerId, daemonLaunchingOptions, daemonOptions, System.err, autostart = true, checkId = true)
+        val daemon = KotlinCompilerClient.connectToCompileService(compilerId, daemonJVMOptions, daemonOptions, System.err, autostart = true, checkId = true)
        TestCase.assertNotNull("failed to connect daemon", daemon)
        val strm = ByteArrayOutputStream()
        val code = KotlinCompilerClient.compile(daemon!!, args, strm)