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") } }