From 8542e878f76bd44b9b80b5f21d13ed027948e55b Mon Sep 17 00:00:00 2001 From: Ilya Chernikov Date: Fri, 23 Jun 2017 19:00:23 +0200 Subject: [PATCH] Refactor daemon periodic check and shutdown logic Cleaner logic and logging, improved stability during shutdown, reduced overhead of the periodic checks, and regular calls, should also eliminate dying daemons scenario on parallel execution (another attempt to do it). --- .../kotlin/daemon/CompileServiceImpl.kt | 298 ++++++++++-------- .../kotlin/daemon/KotlinCompileDaemon.kt | 1 + .../kotlin/daemon/CompilerDaemonTest.kt | 4 +- 3 files changed, 173 insertions(+), 130 deletions(-) diff --git a/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt b/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt index e523bf054be..8acedf8581c 100644 --- a/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt +++ b/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt @@ -114,8 +114,6 @@ class CompileServiceImpl( // wrapped in a class to encapsulate alive check logic private class ClientOrSessionProxy(val aliveFlagPath: String?, val data: T? = null, private var disposable: Disposable? = null) { - val registered = nowSeconds() - val secondsSinceRegistered: Long get() = nowSeconds() - registered val isAlive: Boolean get() = aliveFlagPath?.let { File(it).exists() } ?: true // assuming that if no file was given, the client is alive fun dispose() { disposable?.let { @@ -174,12 +172,50 @@ class CompileServiceImpl( // TODO: encapsulate operations on state here private val state = object { - val clientProxies: MutableSet> = hashSetOf() + private val clientsLock = ReentrantReadWriteLock() + private val clientProxies: MutableSet> = hashSetOf() + val sessions = SessionsContainer() val delayedShutdownQueued = AtomicBoolean(false) var alive = AtomicInteger(Aliveness.Alive.ordinal) + + val aliveClientsCount: Int get() = clientProxies.size + + fun addClient(aliveFlagPath: String?) { + clientsLock.write { + clientProxies.add(ClientOrSessionProxy(aliveFlagPath)) + } + } + + fun getClientsFlagPaths(): List = clientsLock.read { + clientProxies.mapNotNull { it.aliveFlagPath } + } + + fun cleanDeadClients(): Boolean = clientProxies.cleanMatching(clientsLock, { !it.isAlive }, { if (clientProxies.remove(it)) it.dispose() }) + } + + private fun Int.toAlivenessName(): String = + try { + Aliveness.values()[this].name + } + catch (_: Throwable) { + "invalid($this)" + } + + private inline fun Iterable.cleanMatching(lock: ReentrantReadWriteLock, crossinline pred: (T) -> Boolean, crossinline clean: (T) -> Unit): Boolean { + var anyDead = false + lock.read { + val toRemove = filter(pred) + if (toRemove.isNotEmpty()) { + anyDead = true + lock.write { + toRemove.forEach(clean) + } + } + } + return anyDead } @Volatile private var _lastUsedSeconds = nowSeconds() @@ -220,18 +256,14 @@ class CompileServiceImpl( CompileService.CallResult.Good(daemonJVMOptions) } - override fun registerClient(aliveFlagPath: String?): CompileService.CallResult = ifAlive { - synchronized(state.clientProxies) { - state.clientProxies.add(ClientOrSessionProxy(aliveFlagPath)) - log.info("Registered a client alive file: $aliveFlagPath") - } + override fun registerClient(aliveFlagPath: String?): CompileService.CallResult = ifAlive(minAliveness = Aliveness.Alive) { + state.addClient(aliveFlagPath) + log.info("Registered a client alive file: $aliveFlagPath") CompileService.CallResult.Ok() } override fun getClients(): CompileService.CallResult> = ifAlive { - synchronized(state.clientProxies) { - CompileService.CallResult.Good(state.clientProxies.mapNotNull { it.aliveFlagPath }) - } + CompileService.CallResult.Good(state.getClientsFlagPaths()) } // TODO: consider tying a session to a client and use this info to cleanup @@ -266,36 +298,20 @@ class CompileServiceImpl( override fun getUsedMemory(): CompileService.CallResult = ifAlive { CompileService.CallResult.Good(usedMemory(withGC = true)) } - override fun shutdown(): CompileService.CallResult = ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { + override fun shutdown(): CompileService.CallResult = ifAliveExclusive(minAliveness = Aliveness.LastSession) { shutdownWithDelay() CompileService.CallResult.Ok() } - override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult = ifAlive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { - when { - !graceful -> { + override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult = ifAlive(minAliveness = Aliveness.LastSession) { + val res = when { + graceful -> gracefulShutdown(true) + else -> { shutdownWithDelay() - CompileService.CallResult.Good(true) + true } - state.alive.compareAndSet(Aliveness.Alive.ordinal, Aliveness.LastSession.ordinal) -> { - timer.schedule(1) { - ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { - when { - state.sessions.isEmpty() -> shutdownWithDelay() - else -> { - daemonOptions.autoshutdownIdleSeconds = TimeUnit.MILLISECONDS.toSeconds(daemonOptions.forceShutdownTimeoutMilliseconds).toInt() - daemonOptions.autoshutdownUnusedSeconds = daemonOptions.autoshutdownIdleSeconds - log.info("Graceful shutdown signalled; unused/idle timeouts are set to ${daemonOptions.autoshutdownUnusedSeconds}/${daemonOptions.autoshutdownIdleSeconds}s") - log.info("Some sessions are active, waiting for them to finish") - } - } - CompileService.CallResult.Ok() - } - } - CompileService.CallResult.Good(true) - } - else -> CompileService.CallResult.Good(false) } + CompileService.CallResult.Good(res) } override fun remoteCompile(sessionId: Int, @@ -568,72 +584,80 @@ class CompileServiceImpl( registry.rebind (COMPILER_SERVICE_RMI_NAME, stub) timer.schedule(10) { - initiateElections() + exceptionLoggingTimerThread { initiateElections() } } timer.schedule(delay = DAEMON_PERIODIC_CHECK_INTERVAL_MS, period = DAEMON_PERIODIC_CHECK_INTERVAL_MS) { - try { - periodicAndAfterSessionCheck() + exceptionLoggingTimerThread { periodicAndAfterSessionCheck() } + } + timer.schedule(delay = DAEMON_PERIODIC_SELDOM_CHECK_INTERVAL_MS + 100, period = DAEMON_PERIODIC_SELDOM_CHECK_INTERVAL_MS) { + exceptionLoggingTimerThread { periodicSeldomCheck() } + } + } + + private inline fun exceptionLoggingTimerThread(body: () -> Unit) { + try { + body() + } + catch (e: Exception) { + System.err.println("Exception in timer thread: " + e.message) + e.printStackTrace(System.err) + log.log(Level.SEVERE, "Exception in timer thread", e) + } + } + + private fun periodicAndAfterSessionCheck() { + + if (state.delayedShutdownQueued.get()) return + + val anyDead = state.sessions.cleanDead() && state.cleanDeadClients() + + ifAliveUnit(minAliveness = Aliveness.LastSession) { + when { + // check if in graceful shutdown state and all sessions are closed + state.alive.get() == Aliveness.LastSession.ordinal && state.sessions.isEmpty() -> { + log.info("All sessions finished") + shutdownWithDelay() + return + } + state.aliveClientsCount == 0 && compilationsCounter.get() > 0 -> { + log.info("No more clients left") + shutdownWithDelay() + return + } + // discovery file removed - shutdown + !runFile.exists() -> { + log.info("Run file removed") + shutdownWithDelay() + return + } } - catch (e: Exception) { - System.err.println("Exception in timer thread: " + e.message) - e.printStackTrace(System.err) - log.log(Level.SEVERE, "Exception in timer thread", e) + } + + ifAliveUnit(minAliveness = Aliveness.Alive) { + when { + daemonOptions.autoshutdownUnusedSeconds != COMPILE_DAEMON_TIMEOUT_INFINITE_S && compilationsCounter.get() == 0 && nowSeconds() - lastUsedSeconds > daemonOptions.autoshutdownUnusedSeconds -> { + log.info("Unused timeout exceeded ${daemonOptions.autoshutdownUnusedSeconds}s") + gracefulShutdown(false) + } + daemonOptions.autoshutdownIdleSeconds != COMPILE_DAEMON_TIMEOUT_INFINITE_S && nowSeconds() - lastUsedSeconds > daemonOptions.autoshutdownIdleSeconds -> { + log.info("Idle timeout exceeded ${daemonOptions.autoshutdownIdleSeconds}s") + gracefulShutdown(false) + } + anyDead -> { + clearJarCache() + } } } } + private fun periodicSeldomCheck() { + ifAliveUnit(minAliveness = Aliveness.Alive) { - private fun periodicAndAfterSessionCheck() { - - ifAlive(minAliveness = Aliveness.LastSession) { - - // 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")) { - scheduleShutdown(true) + // compiler changed (seldom check) - shutdown + if (classpathWatcher.isChanged) { + log.info("Compiler changed.") + gracefulShutdown(false) } - else { - var anyDead = state.sessions.cleanDead() - var shuttingDown = false - - - // 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")) { - shutdownWithDelay() - shuttingDown = true - } - - // 4. clean dead clients, then check if any left - conditional shutdown (with small delay) - synchronized(state.clientProxies) { - state.clientProxies.removeAll( - state.clientProxies.filter { !it.isAlive }.map { - it.dispose() - anyDead = true - it - }) - } - if (state.clientProxies.isEmpty() && compilationsCounter.get() > 0 && !state.delayedShutdownQueued.get()) { - log.info("No more clients left, delayed shutdown in ${daemonOptions.shutdownDelayMilliseconds}ms") - shutdownWithDelay() - } - // 5. check idle timeout - shutdown - if (shutdownCondition({ daemonOptions.autoshutdownIdleSeconds != COMPILE_DAEMON_TIMEOUT_INFINITE_S && nowSeconds() - lastUsedSeconds > daemonOptions.autoshutdownIdleSeconds }, - "Idle timeout exceeded ${daemonOptions.autoshutdownIdleSeconds}s, shutting down") || - // 6. discovery file removed - shutdown - shutdownCondition({ !runFile.exists() }, "Run file removed, shutting down") || - // 7. compiler changed (seldom check) - shutdown - // TODO: could be too expensive anyway, consider removing this check - shutdownCondition({ classpathWatcher.isChanged }, "Compiler changed")) - { - scheduleShutdown(true) - shuttingDown = true - } - - if (anyDead && !shuttingDown) { - clearJarCache() - } - } - CompileService.CallResult.Ok() } } @@ -641,7 +665,7 @@ class CompileServiceImpl( // TODO: handover should include mechanism for client to switch to a new daemon then previous "handed over responsibilities" and shot down private fun initiateElections() { - ifAlive { + ifAliveUnit { val aliveWithOpts = walkDaemons(File(daemonOptions.runFilesPathOrDefault), compilerId, runFile, filter = { _, p -> p != port }, report = { _, msg -> log.info(msg) }).toList() val comparator = compareByDescending(DaemonJVMOptionsMemoryComparator(), { it.jvmOptions }) @@ -694,11 +718,10 @@ class CompileServiceImpl( // - 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() } } - private fun shutdownImpl() { + private fun shutdownNow() { log.info("Shutdown started") fun Long.mb() = this / (1024 * 1024) with (Runtime.getRuntime()) { @@ -715,13 +738,13 @@ class CompileServiceImpl( private fun shutdownWithDelay() { state.delayedShutdownQueued.set(true) val currentCompilationsCount = compilationsCounter.get() + log.info("Delayed shutdown in ${daemonOptions.shutdownDelayMilliseconds}ms") timer.schedule(daemonOptions.shutdownDelayMilliseconds) { state.delayedShutdownQueued.set(false) if (currentCompilationsCount == compilationsCounter.get()) { - ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { + ifAliveExclusiveUnit(minAliveness = Aliveness.LastSession) { log.fine("Execute delayed shutdown") - shutdownImpl() - CompileService.CallResult.Ok() + shutdownNow() } } else { @@ -730,12 +753,35 @@ class CompileServiceImpl( } } - private inline fun shutdownCondition(check: () -> Boolean, message: String): Boolean { - val res = check() - if (res) { - log.info(message) + private fun gracefulShutdown(onAnotherThread: Boolean): Boolean { + + fun shutdownIfIdle() = when { + state.sessions.isEmpty() -> shutdownWithDelay() + else -> { + daemonOptions.autoshutdownIdleSeconds = TimeUnit.MILLISECONDS.toSeconds(daemonOptions.forceShutdownTimeoutMilliseconds).toInt() + daemonOptions.autoshutdownUnusedSeconds = daemonOptions.autoshutdownIdleSeconds + log.info("Some sessions are active, waiting for them to finish") + log.info("Unused/idle timeouts are set to ${daemonOptions.autoshutdownUnusedSeconds}/${daemonOptions.autoshutdownIdleSeconds}s") + } } - return res + + if (!state.alive.compareAndSet(Aliveness.Alive.ordinal, Aliveness.LastSession.ordinal)) { + log.info("Invalid state for graceful shutdown: ${state.alive.get().toAlivenessName()}") + return false + } + log.info("Graceful shutdown signalled") + + if (!onAnotherThread) { + shutdownIfIdle() + } + else { + timer.schedule(1) { + ifAliveExclusiveUnit(minAliveness = Aliveness.LastSession) { + shutdownIfIdle() + } + } + } + return true } // todo: remove after remoteIncrementalCompile is removed @@ -848,37 +894,33 @@ class CompileServiceImpl( (KotlinCoreEnvironment.applicationEnvironment?.jarFileSystem as? CoreJarFileSystem)?.clearHandlersCache() } - private fun ifAlive(minAliveness: Aliveness = Aliveness.Alive, - ignoreCompilerChanged: Boolean = false, - body: () -> CompileService.CallResult - ): CompileService.CallResult = - rwlock.read { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged, body) - } + private inline fun ifAlive(minAliveness: Aliveness = Aliveness.LastSession, body: () -> CompileService.CallResult): CompileService.CallResult = rwlock.read { + ifAliveChecksImpl(minAliveness, body) + } - private fun ifAliveExclusive(minAliveness: Aliveness = Aliveness.Alive, - ignoreCompilerChanged: Boolean = false, - body: () -> CompileService.CallResult - ): CompileService.CallResult = - rwlock.write { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged, body) - } + private inline fun ifAliveUnit(minAliveness: Aliveness = Aliveness.LastSession, body: () -> Unit): Unit = rwlock.read { + ifAliveChecksImpl(minAliveness) { + body() + CompileService.CallResult.Ok() + } + } - inline private fun ifAliveChecksImpl(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> CompileService.CallResult): CompileService.CallResult { + private inline fun ifAliveExclusive(minAliveness: Aliveness = Aliveness.LastSession, body: () -> CompileService.CallResult): CompileService.CallResult = rwlock.write { + ifAliveChecksImpl(minAliveness, body) + } + + private inline fun ifAliveExclusiveUnit(minAliveness: Aliveness = Aliveness.LastSession, body: () -> Unit): Unit = rwlock.write { + ifAliveChecksImpl(minAliveness) { + body() + CompileService.CallResult.Ok() + } + } + + private inline fun ifAliveChecksImpl(minAliveness: Aliveness = Aliveness.LastSession, body: () -> CompileService.CallResult): CompileService.CallResult { val curState = state.alive.get() return when { curState < minAliveness.ordinal -> { - log.info("Cannot perform operation, requested state: ${minAliveness.name} > actual: ${try { - Aliveness.values()[curState].name - } - catch (_: Throwable) { - "invalid($curState)" - }}") - CompileService.CallResult.Dying() - } - !ignoreCompilerChanged && classpathWatcher.isChanged -> { - log.info("Compiler changed, scheduling shutdown") - shutdownWithDelay() + log.info("Cannot perform operation, requested state: ${minAliveness.name} > actual: ${curState.toAlivenessName()}") CompileService.CallResult.Dying() } else -> { diff --git a/compiler/daemon/src/org/jetbrains/kotlin/daemon/KotlinCompileDaemon.kt b/compiler/daemon/src/org/jetbrains/kotlin/daemon/KotlinCompileDaemon.kt index b33e23f5052..e6da311ab8d 100644 --- a/compiler/daemon/src/org/jetbrains/kotlin/daemon/KotlinCompileDaemon.kt +++ b/compiler/daemon/src/org/jetbrains/kotlin/daemon/KotlinCompileDaemon.kt @@ -35,6 +35,7 @@ import java.util.logging.* import kotlin.concurrent.schedule val DAEMON_PERIODIC_CHECK_INTERVAL_MS = 1000L +val DAEMON_PERIODIC_SELDOM_CHECK_INTERVAL_MS = 60000L class LogStream(name: String) : OutputStream() { diff --git a/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt b/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt index f887acbabdb..2b35f5ad6c2 100644 --- a/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt +++ b/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt @@ -335,7 +335,7 @@ class CompilerDaemonTest : KotlinIntegrationTestBase() { logFile.assertLogContainsSequence("Some sessions are active, waiting for them to finish") - val res = daemon?.getUsedMemory() + val res = daemon?.leaseCompileSession(null) assertEquals("Invalid state", CompileService.CallResult.Dying(), res) @@ -343,7 +343,7 @@ class CompilerDaemonTest : KotlinIntegrationTestBase() { Thread.sleep(100) // allow after session timed action to run - logFile.assertLogContainsSequence("All sessions finished, shutting down", + logFile.assertLogContainsSequence("All sessions finished", "Shutdown started") } }