Implement more reliable daemon election, fixes KT-15562

also making shutdown more reliable
2017-02-08 20:49:23 +01:00
parent ebf9e386b0
commit cb7de0e262
5 changed files with 146 additions and 81 deletions
@@ -291,17 +291,23 @@ object KotlinCompilerClient {
    }

    private fun tryFindSuitableDaemonOrNewOpts(registryDir: File, compilerId: CompilerId, daemonJVMOptions: DaemonJVMOptions, report: (DaemonReportCategory, String) -> Unit): Pair<CompileService?, DaemonJVMOptions> {
-        val aliveWithOpts = walkDaemons(registryDir, compilerId, report = report)
-                .map { Pair(it, it.getDaemonJVMOptions()) }
-                .filter { it.second.isGood }
-                .sortedWith(compareByDescending(DaemonJVMOptionsMemoryComparator(), { it.second.get() }))
+        registryDir.mkdirs()
+        val timestampMarker = createTempFile("kotlin-daemon-client-tsmarker", directory = registryDir)
+        val aliveWithMetadata = try {
+            walkDaemons(registryDir, compilerId, timestampMarker, report = report).toList()
+        }
+        finally {
+            timestampMarker.delete()
+        }
+        val comparator = compareByDescending<DaemonWithMetadata, DaemonJVMOptions>(DaemonJVMOptionsMemoryComparator(), { it.jvmOptions })
+                .thenBy(FileAgeComparator()) { it.runFile }
        val optsCopy = daemonJVMOptions.copy()
        // if required options fit into fattest running daemon - return the daemon and required options with memory params set to actual ones in the daemon
-        return aliveWithOpts.firstOrNull()?.check { daemonJVMOptions memorywiseFitsInto it.second.get() }?.let {
-                Pair(it.first, optsCopy.updateMemoryUpperBounds(it.second.get()))
+        return aliveWithMetadata.maxWith(comparator)?.takeIf { daemonJVMOptions memorywiseFitsInto it.jvmOptions }?.let {
+                Pair(it.daemon, optsCopy.updateMemoryUpperBounds(it.jvmOptions))
            }
            // else combine all options from running daemon to get fattest option for a new daemon to run
-            ?: Pair(null, aliveWithOpts.fold(optsCopy, { opts, d -> opts.updateMemoryUpperBounds(d.second.get()) }))
+            ?: Pair(null, aliveWithMetadata.fold(optsCopy, { opts, d -> opts.updateMemoryUpperBounds(d.jvmOptions) }))
    }


@@ -40,30 +40,48 @@ fun makePortFromRunFilenameExtractor(digest: String): (String) -> Int? {
    }
 }

+private const val ORPHANED_RUN_FILE_AGE_THRESHOLD_MS = 1000000L
+
+data class DaemonWithMetadata(val daemon: CompileService, val runFile: File, val jvmOptions: DaemonJVMOptions)

 fun walkDaemons(registryDir: File,
                compilerId: CompilerId,
-                filter: (File, Int) -> Boolean = { f, p -> true },
-                report: (DaemonReportCategory, String) -> Unit = { cat, msg -> }
-): Sequence<CompileService> {
+                fileToCompareTimestamp: File,
+                filter: (File, Int) -> Boolean = { _, _ -> true },
+                report: (DaemonReportCategory, String) -> Unit = { _, _ -> }
+): Sequence<DaemonWithMetadata> {
    val classPathDigest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest().toHexString()
    val portExtractor = makePortFromRunFilenameExtractor(classPathDigest)
    return registryDir.walk()
            .map { Pair(it, portExtractor(it.name)) }
            .filter { it.second != null && filter(it.first, it.second!!) }
-            .mapNotNull {
-                assert(it.second!! > 0 && it.second!! < MAX_PORT_NUMBER)
-                report(DaemonReportCategory.DEBUG, "found daemon on port ${it.second}, trying to connect")
-                val daemon = tryConnectToDaemon(it.second!!, report)
+            .mapNotNull { fileWithPort ->
+                assert(fileWithPort.second!! in 1..(MAX_PORT_NUMBER - 1))
+                val relativeAge = fileToCompareTimestamp.lastModified() - fileWithPort.first.lastModified()
+                report(DaemonReportCategory.DEBUG, "found daemon on port ${fileWithPort.second} ($relativeAge ms old), trying to connect")
+                val daemon = tryConnectToDaemon(fileWithPort.second!!, report)
                // cleaning orphaned file; note: daemon should shut itself down if it detects that the run file is deleted
-                if (daemon == null && !it.first.delete()) {
-                    report(DaemonReportCategory.INFO, "WARNING: unable to delete seemingly orphaned file '${it.first.absolutePath}', cleanup recommended")
+                if (daemon == null) {
+                    if (relativeAge - ORPHANED_RUN_FILE_AGE_THRESHOLD_MS <= 0) {
+                        report(DaemonReportCategory.DEBUG, "found fresh run file '${fileWithPort.first.absolutePath}' ($relativeAge ms old), but no daemon, ignoring it")
+                    }
+                    else {
+                        report(DaemonReportCategory.DEBUG, "found seemingly orphaned run file '${fileWithPort.first.absolutePath}' ($relativeAge ms old), deleting it")
+                        if (!fileWithPort.first.delete()) {
+                            report(DaemonReportCategory.INFO, "WARNING: unable to delete seemingly orphaned file '${fileWithPort.first.absolutePath}', cleanup recommended")
+                        }
+                    }
+                }
+                try {
+                    daemon?.let { DaemonWithMetadata(it, fileWithPort.first, it.getDaemonJVMOptions().get()) }
+                }
+                catch (e: Exception) {
+                    report(DaemonReportCategory.INFO, "ERROR: unable to retrieve daemon JVM options, assuming daemon is dead: ${e.message}")
+                    null
                }
-                daemon
            }
 }

-
 private inline fun tryConnectToDaemon(port: Int, report: (DaemonReportCategory, String) -> Unit): CompileService? {

    try {
@@ -88,3 +106,17 @@ fun makeAutodeletingFlagFile(keyword: String = "compiler-client", baseDir: File?
    flagFile.deleteOnExit()
    return flagFile
 }
+
+// Comparator for reliable choice between daemons
+class FileAgeComparator : Comparator<File> {
+    override fun compare(left: File, right: File): Int {
+        val leftTS = left.lastModified()
+        val rightTS = right.lastModified()
+        return when {
+            leftTS == 0L || rightTS == 0L -> 0 // cannot read any file timestamp, => undecidable
+            leftTS > rightTS -> -1
+            leftTS < rightTS -> 1
+            else -> compareValues(left.canonicalPath, right.canonicalPath)
+        }
+    }
+}
@@ -267,20 +267,19 @@ class CompileServiceImpl(
            ifAlive { CompileService.CallResult.Good(usedMemory(withGC = true)) }

    override fun shutdown(): CompileService.CallResult<Nothing> = ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) {
-        shutdownImpl()
+        shutdownWithDelay()
        CompileService.CallResult.Ok()
    }

    override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult<Boolean> = ifAlive(minAliveness = Aliveness.Alive) {
        CompileService.CallResult.Good(
                if (!graceful || state.alive.compareAndSet(Aliveness.Alive.ordinal, Aliveness.LastSession.ordinal)) {
-                    timer.schedule(0) {
+                    timer.schedule(1) {
                        ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) {
-                            if (!graceful || state.sessions.isEmpty()) {
-                                shutdownImpl()
-                            }
-                            else {
-                                log.info("Some sessions are active, waiting for them to finish")
+                            when {
+                                !graceful -> shutdownImpl()
+                                state.sessions.isEmpty() -> shutdownWithDelay()
+                                else -> log.info("Some sessions are active, waiting for them to finish")
                            }
                            CompileService.CallResult.Ok()
                        }
@@ -529,7 +528,7 @@ class CompileServiceImpl(
        val stub = UnicastRemoteObject.exportObject(this, port, LoopbackNetworkInterface.clientLoopbackSocketFactory, LoopbackNetworkInterface.serverLoopbackSocketFactory) as CompileService
        registry.rebind (COMPILER_SERVICE_RMI_NAME, stub)

-        timer.schedule(0) {
+        timer.schedule(100) {
            initiateElections()
        }
        timer.schedule(delay = DAEMON_PERIODIC_CHECK_INTERVAL_MS, period = DAEMON_PERIODIC_CHECK_INTERVAL_MS) {
@@ -552,7 +551,7 @@ class CompileServiceImpl(
            // 1. check if unused for a timeout - shutdown
            if (shutdownCondition({ daemonOptions.autoshutdownUnusedSeconds != COMPILE_DAEMON_TIMEOUT_INFINITE_S && compilationsCounter.get() == 0 && nowSeconds() - lastUsedSeconds > daemonOptions.autoshutdownUnusedSeconds },
                                  "Unused timeout exceeded ${daemonOptions.autoshutdownUnusedSeconds}s, shutting down")) {
-                shutdown()
+                scheduleShutdown(true)
            }
            else {
                var anyDead = state.sessions.cleanDead()
@@ -561,7 +560,7 @@ class CompileServiceImpl(

                // 3. check if in graceful shutdown state and all sessions are closed
                if (shutdownCondition({ state.alive.get() == Aliveness.LastSession.ordinal && state.sessions.isEmpty() }, "All sessions finished, shutting down")) {
-                    shutdown()
+                    shutdownWithDelay()
                    shuttingDown = true
                }

@@ -587,7 +586,7 @@ class CompileServiceImpl(
                    // TODO: could be too expensive anyway, consider removing this check
                    shutdownCondition({ classpathWatcher.isChanged }, "Compiler changed"))
                {
-                    shutdown()
+                    scheduleShutdown(true)
                    shuttingDown = true
                }

@@ -604,35 +603,44 @@ class CompileServiceImpl(

        ifAlive {

-            val aliveWithOpts = walkDaemons(File(daemonOptions.runFilesPathOrDefault), compilerId, filter = { f, p -> p != port }, report = { lvl, msg -> log.info(msg) })
-                    .map { Pair(it, it.getDaemonJVMOptions()) }
-                    .filter { it.second.isGood }
-                    .sortedWith(compareByDescending(DaemonJVMOptionsMemoryComparator(), { it.second.get() }))
-            if (aliveWithOpts.any()) {
-                val fattestOpts = aliveWithOpts.first().second.get()
+            val aliveWithOpts = walkDaemons(File(daemonOptions.runFilesPathOrDefault), compilerId, runFile, filter = { f, p -> p != port }, report = { _, msg -> log.info(msg) }).toList()
+            val comparator = compareByDescending<DaemonWithMetadata, DaemonJVMOptions>(DaemonJVMOptionsMemoryComparator(), { it.jvmOptions })
+                    .thenBy(FileAgeComparator()) { it.runFile }
+            aliveWithOpts.maxWith(comparator)?.let { bestDaemonWithMetadata ->
+                val fattestOpts = bestDaemonWithMetadata.jvmOptions
                // second part of the condition means that we prefer other daemon if is "equal" to the current one
-                if (fattestOpts memorywiseFitsInto daemonJVMOptions && !(daemonJVMOptions memorywiseFitsInto fattestOpts)) {
+                if (fattestOpts memorywiseFitsInto daemonJVMOptions && FileAgeComparator().compare(bestDaemonWithMetadata.runFile, runFile) < 0 ) {
                    // all others are smaller that me, take overs' clients and shut them down
+                    log.info("Assuming other daemons have lower prio, taking clients from them and schedule them to shutdown: my runfile: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
                    aliveWithOpts.forEach {
-                        it.first.getClients().check { it.isGood }?.let {
-                            it.get().forEach { registerClient(it) }
+                        try {
+                            it.daemon.getClients().check { it.isGood }?.let {
+                                it.get().forEach { registerClient(it) }
+                            }
+                            it.daemon.scheduleShutdown(true)
+                        }
+                        catch (e: Exception) {
+                            log.info("Cannot connect to a daemon, assuming dying ('${it.runFile.canonicalPath}'): ${e.message}")
                        }
-                        it.first.scheduleShutdown(true)
                    }
                }
-                else if (daemonJVMOptions memorywiseFitsInto fattestOpts) {
+                else if (daemonJVMOptions memorywiseFitsInto fattestOpts && FileAgeComparator().compare(bestDaemonWithMetadata.runFile, runFile) > 0) {
                    // there is at least one bigger, handover my clients to it and shutdown
+                    log.info("Assuming other daemons have higher prio, handover clients to it and schedule shutdown: my runfile: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
                    scheduleShutdown(true)
-                    aliveWithOpts.first().first.let { fattest ->
+                    aliveWithOpts.first().daemon.let { fattest ->
                        getClients().check { it.isGood }?.let {
                            it.get().forEach { fattest.registerClient(it) }
                        }
                    }
                }
-                // else - do nothing, all daemons are staying
-                // TODO: implement some behaviour here, e.g.:
-                //   - shutdown/takeover smaller daemon
-                //   - run (or better persuade client to run) a bigger daemon (in fact may be even simple shutdown will do, because of client's daemon choosing logic)
+                else {
+                    // undecided, do nothing
+                    log.info("Assuming other daemon(s) have equal prio, continue: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
+                    // TODO: implement some behaviour here, e.g.:
+                    //   - shutdown/takeover smaller daemon
+                    //   - run (or better persuade client to run) a bigger daemon (in fact may be even simple shutdown will do, because of client's daemon choosing logic)
+                }
            }
            CompileService.CallResult.Ok()
        }
@@ -653,8 +661,11 @@ class CompileServiceImpl(
        timer.schedule(daemonOptions.shutdownDelayMilliseconds) {
            state.delayedShutdownQueued.set(false)
            if (currentCompilationsCount == compilationsCounter.get()) {
-                log.fine("Execute delayed shutdown")
-                shutdown()
+                ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) {
+                    log.fine("Execute delayed shutdown")
+                    shutdownImpl()
+                    CompileService.CallResult.Ok()
+                }
            }
            else {
                log.info("Cancel delayed shutdown due to new client")
@@ -801,7 +812,7 @@ class CompileServiceImpl(
            state.alive.get() < minAliveness.ordinal -> CompileService.CallResult.Dying()
            !ignoreCompilerChanged && classpathWatcher.isChanged -> {
                log.info("Compiler changed, scheduling shutdown")
-                timer.schedule(0) { shutdown() }
+                shutdownWithDelay()
                CompileService.CallResult.Dying()
            }
            else -> {
@@ -30,9 +30,7 @@ import java.net.URLClassLoader
 import java.text.SimpleDateFormat
 import java.util.*
 import java.util.jar.Manifest
-import java.util.logging.Level
-import java.util.logging.LogManager
-import java.util.logging.Logger
+import java.util.logging.*
 import kotlin.concurrent.schedule

 val DAEMON_PERIODIC_CHECK_INTERVAL_MS = 1000L
@@ -55,7 +53,6 @@ class LogStream(name: String) : OutputStream() {
    }
 }

-
 object KotlinCompileDaemon {

    init {
@@ -71,7 +68,7 @@ object KotlinCompileDaemon {
                "java.util.logging.FileHandler.count     = ${if (fileIsGiven) 1 else 3}\n" +
                "java.util.logging.FileHandler.append    = $fileIsGiven\n" +
                "java.util.logging.FileHandler.pattern   = ${if (fileIsGiven) logPath else (logPath + File.separator + "${COMPILE_DAEMON_DEFAULT_FILES_PREFIX}.$logTime.%u%g.log")}\n" +
-                "java.util.logging.SimpleFormatter.format = %1\$tF %1\$tT.%1\$tL [%3\$s] %4\$s: %5\$s\\n\n"
+                "java.util.logging.SimpleFormatter.format = %1\$tF %1\$tT.%1\$tL [%3\$s] %4\$s: %5\$s%n\n"

        LogManager.getLogManager().readConfiguration(cfg.byteInputStream())
    }
@@ -96,6 +93,7 @@ object KotlinCompileDaemon {
        log.info("Kotlin compiler daemon version " + (loadVersionFromResource() ?: "<unknown>"))
        log.info("daemon JVM args: " + ManagementFactory.getRuntimeMXBean().inputArguments.joinToString(" "))
        log.info("daemon args: " + args.joinToString(" "))
+        log.info("daemon process name: " + ManagementFactory.getRuntimeMXBean().name)

        val compilerId = CompilerId()
        val daemonOptions = DaemonOptions()
@@ -159,6 +157,8 @@ object KotlinCompileDaemon {
            if (daemonOptions.runFilesPath.isNotEmpty())
                println(daemonOptions.runFilesPath)

+            log.info("daemon is listening on port: $port")
+
            // this supposed to stop redirected streams reader(s) on the client side and prevent some situations with hanging threads, but doesn't work reliably
            // TODO: implement more reliable scheme
            System.out.close()