From ea4a1df839e849c054534cd543c1d0fc74f54c57 Mon Sep 17 00:00:00 2001 From: Ilya Chernikov Date: Sun, 2 Oct 2016 16:42:27 +0200 Subject: [PATCH] Refactor session accounting in daemon, add REPL to it, write a test --- .../daemon/client/KotlinRemoteReplClient.kt | 7 +- .../kotlin/daemon/CompileServiceImpl.kt | 175 ++++++++++-------- .../kotlin/daemon/CompilerDaemonTest.kt | 42 +++++ 3 files changed, 143 insertions(+), 81 deletions(-) diff --git a/compiler/daemon/daemon-client/src/org/jetbrains/kotlin/daemon/client/KotlinRemoteReplClient.kt b/compiler/daemon/daemon-client/src/org/jetbrains/kotlin/daemon/client/KotlinRemoteReplClient.kt index 73e74ec899d..876232b2e92 100644 --- a/compiler/daemon/daemon-client/src/org/jetbrains/kotlin/daemon/client/KotlinRemoteReplClient.kt +++ b/compiler/daemon/daemon-client/src/org/jetbrains/kotlin/daemon/client/KotlinRemoteReplClient.kt @@ -62,7 +62,12 @@ open class KotlinRemoteReplClientBase( init { Disposer.register(disposable, Disposable { - compileService.releaseReplSession(sessionId) + try { + compileService.releaseReplSession(sessionId) + } + catch (ex: java.rmi.RemoteException) { + // assuming that communication failed and daemon most likely is already down + } }) } diff --git a/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt b/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt index 03324fa9032..d3cb4c6c6d9 100644 --- a/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt +++ b/compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt @@ -151,27 +151,33 @@ class CompileServiceImpl( // RMI-exposed API - override fun getDaemonOptions(): CompileService.CallResult = ifAlive { daemonOptions } + override fun getDaemonOptions(): CompileService.CallResult = ifAlive { + CompileService.CallResult.Good(daemonOptions) + } - override fun getDaemonJVMOptions(): CompileService.CallResult = ifAlive { daemonJVMOptions } + override fun getDaemonJVMOptions(): CompileService.CallResult = ifAlive { + CompileService.CallResult.Good(daemonJVMOptions) + } - override fun registerClient(aliveFlagPath: String?): CompileService.CallResult = ifAlive_Nothing { + override fun registerClient(aliveFlagPath: String?): CompileService.CallResult = ifAlive { synchronized(state.clientProxies) { state.clientProxies.add(ClientOrSessionProxy(aliveFlagPath)) } + CompileService.CallResult.Ok() } override fun getClients(): CompileService.CallResult> = ifAlive { synchronized(state.clientProxies) { - state.clientProxies.mapNotNull { it.aliveFlagPath } + CompileService.CallResult.Good(state.clientProxies.mapNotNull { it.aliveFlagPath }) } } // TODO: consider tying a session to a client and use this info to cleanup override fun leaseCompileSession(aliveFlagPath: String?): CompileService.CallResult = ifAlive(minAliveness = Aliveness.Alive) { - leaseSessionImpl(ClientOrSessionProxy(aliveFlagPath)).apply { - log.info("leased a new session $this, client alive file: $aliveFlagPath") - } + CompileService.CallResult.Good( + leaseSessionImpl(ClientOrSessionProxy(aliveFlagPath)).apply { + log.info("leased a new session $this, client alive file: $aliveFlagPath") + }) } private fun leaseSessionImpl(session: ClientOrSessionProxy): Int { @@ -192,7 +198,7 @@ class CompileServiceImpl( throw IllegalStateException("Invalid state or algorithm error") } - override fun releaseCompileSession(sessionId: Int) = ifAlive_Nothing(minAliveness = Aliveness.LastSession) { + override fun releaseCompileSession(sessionId: Int) = ifAlive(minAliveness = Aliveness.LastSession) { synchronized(state.sessions) { state.sessions[sessionId]?.dispose() state.sessions.remove(sessionId) @@ -205,6 +211,7 @@ class CompileServiceImpl( timer.schedule(0) { periodicAndAfterSessionCheck() } + CompileService.CallResult.Ok() } override fun checkCompilerId(expectedCompilerId: CompilerId): Boolean = @@ -212,27 +219,31 @@ class CompileServiceImpl( (compilerId.compilerClasspath.all { expectedCompilerId.compilerClasspath.contains(it) }) && !classpathWatcher.isChanged - override fun getUsedMemory(): CompileService.CallResult = ifAlive { usedMemory(withGC = true) } + override fun getUsedMemory(): CompileService.CallResult = + ifAlive { CompileService.CallResult.Good(usedMemory(withGC = true)) } - override fun shutdown(): CompileService.CallResult = ifAliveExclusive_Nothing(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { + override fun shutdown(): CompileService.CallResult = ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { shutdownImpl() + CompileService.CallResult.Ok() } override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult = ifAlive(minAliveness = Aliveness.Alive) { - if (!graceful || state.alive.compareAndSet(Aliveness.Alive.ordinal, Aliveness.LastSession.ordinal)) { - timer.schedule(0) { - ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { - if (!graceful || state.sessions.isEmpty()) { - shutdownImpl() - } - else { - log.info("Some sessions are active, waiting for them to finish") + CompileService.CallResult.Good( + if (!graceful || state.alive.compareAndSet(Aliveness.Alive.ordinal, Aliveness.LastSession.ordinal)) { + timer.schedule(0) { + ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { + if (!graceful || state.sessions.isEmpty()) { + shutdownImpl() + } + else { + log.info("Some sessions are active, waiting for them to finish") + } + CompileService.CallResult.Ok() + } } + true } - } - true - } - else false + else false) } override fun remoteCompile(sessionId: Int, @@ -280,7 +291,7 @@ class CompileServiceImpl( evalErrorStream: RemoteOutputStream?, evalInputStream: RemoteInputStream?, operationsTracer: RemoteOperationsTracer? - ): CompileService.CallResult = ifAlive_Raw(minAliveness = Aliveness.Alive) { + ): CompileService.CallResult = ifAlive(minAliveness = Aliveness.Alive) { if (targetPlatform != CompileService.TargetPlatform.JVM) CompileService.CallResult.Error("Sorry, only JVM target platform is supported now") else { @@ -292,25 +303,18 @@ class CompileServiceImpl( } } - private fun withValidRepl(sessionId: Int, body: KotlinJvmReplService.() -> R): CompileService.CallResult = - state.sessions[sessionId]?.let { - (it.data as? KotlinJvmReplService?)?.let { - CompileService.CallResult.Good(it.body()) - } - } ?: CompileService.CallResult.Error("Unknown or invalid session $sessionId") - // TODO: add more checks (e.g. is it a repl session) override fun releaseReplSession(sessionId: Int): CompileService.CallResult = releaseCompileSession(sessionId) override fun remoteReplLineCheck(sessionId: Int, codeLine: ReplCodeLine, history: List): CompileService.CallResult = - ifAlive_Raw(minAliveness = Aliveness.Alive) { + ifAlive(minAliveness = Aliveness.Alive) { withValidRepl(sessionId) { check(codeLine, history) } } override fun remoteReplLineCompile(sessionId: Int, codeLine: ReplCodeLine, history: List): CompileService.CallResult = - ifAlive_Raw(minAliveness = Aliveness.Alive) { + ifAlive(minAliveness = Aliveness.Alive) { withValidRepl(sessionId) { compile(codeLine, history) } @@ -321,7 +325,7 @@ class CompileServiceImpl( codeLine: ReplCodeLine, history: List ): CompileService.CallResult = - ifAlive_Raw(minAliveness = Aliveness.Alive) { + ifAlive(minAliveness = Aliveness.Alive) { withValidRepl(sessionId) { eval(codeLine, history) } @@ -373,7 +377,7 @@ class CompileServiceImpl( private fun periodicAndAfterSessionCheck() { - ifAlive_Nothing(minAliveness = Aliveness.LastSession) { + 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 }, @@ -427,13 +431,14 @@ class CompileServiceImpl( clearJarCache() } } + CompileService.CallResult.Ok() } } private fun initiateElections() { - ifAlive_Nothing { + ifAlive { val aliveWithOpts = walkDaemons(File(daemonOptions.runFilesPathOrDefault), compilerId, filter = { f, p -> p != port }, report = { lvl, msg -> log.info(msg) }) .map { Pair(it, it.getDaemonJVMOptions()) } @@ -465,6 +470,7 @@ class CompileServiceImpl( // - 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() } } @@ -507,25 +513,24 @@ class CompileServiceImpl( operationsTracer: RemoteOperationsTracer?, body: (PrintStream, EventManger, Profiler) -> ExitCode): CompileService.CallResult = ifAlive { - - operationsTracer?.before("compile") - compilationsCounter.incrementAndGet() - val rpcProfiler = if (daemonOptions.reportPerf) WallAndThreadTotalProfiler() else DummyProfiler() - val eventManger = EventMangerImpl() - val compilerMessagesStream = PrintStream(BufferedOutputStream(RemoteOutputStreamClient(compilerMessagesStreamProxy, rpcProfiler), REMOTE_STREAM_BUFFER_SIZE)) - val serviceOutputStream = PrintStream(BufferedOutputStream(RemoteOutputStreamClient(serviceOutputStreamProxy, rpcProfiler), REMOTE_STREAM_BUFFER_SIZE)) - try { - checkedCompile(args, serviceOutputStream, rpcProfiler) { - val res = body(compilerMessagesStream, eventManger, rpcProfiler).code - _lastUsedSeconds = nowSeconds() - res + withValidClientOrSessionProxy(sessionId) { session -> + operationsTracer?.before("compile") + val rpcProfiler = if (daemonOptions.reportPerf) WallAndThreadTotalProfiler() else DummyProfiler() + val eventManger = EventMangerImpl() + val compilerMessagesStream = PrintStream(BufferedOutputStream(RemoteOutputStreamClient(compilerMessagesStreamProxy, rpcProfiler), REMOTE_STREAM_BUFFER_SIZE)) + val serviceOutputStream = PrintStream(BufferedOutputStream(RemoteOutputStreamClient(serviceOutputStreamProxy, rpcProfiler), REMOTE_STREAM_BUFFER_SIZE)) + try { + CompileService.CallResult.Good( + checkedCompile(args, serviceOutputStream, rpcProfiler) { + body(compilerMessagesStream, eventManger, rpcProfiler).code + }) + } + finally { + serviceOutputStream.flush() + compilerMessagesStream.flush() + eventManger.fireCompilationFinished() + operationsTracer?.after("compile") } - } - finally { - serviceOutputStream.flush() - compilerMessagesStream.flush() - eventManger.fireCompilationFinished() - operationsTracer?.after("compile") } } @@ -588,34 +593,21 @@ class CompileServiceImpl( (KotlinCoreEnvironment.applicationEnvironment?.jarFileSystem as? CoreJarFileSystem)?.clearHandlersCache() } - private fun ifAlive(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> R): CompileService.CallResult = rwlock.read { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { CompileService.CallResult.Good(body()) } - } + private fun ifAlive(minAliveness: Aliveness = Aliveness.Alive, + ignoreCompilerChanged: Boolean = false, + body: () -> CompileService.CallResult + ): CompileService.CallResult = + rwlock.read { + ifAliveChecksImpl(minAliveness, ignoreCompilerChanged, body) + } - // TODO: find how to implement it without using unique name for this variant; making name deliberately ugly meanwhile - private fun ifAlive_Nothing(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> Unit): CompileService.CallResult = rwlock.read { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { - body() - CompileService.CallResult.Ok() - } - } - - private fun ifAlive_Raw(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> CompileService.CallResult): CompileService.CallResult = rwlock.read { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { body() } - } - - private fun ifAliveExclusive(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> R): CompileService.CallResult = rwlock.write { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { CompileService.CallResult.Good(body()) } - } - - // see comment to ifAliveNothing - private fun ifAliveExclusive_Nothing(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> Unit): CompileService.CallResult = rwlock.write { - ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { - body() - CompileService.CallResult.Ok() - - } - } + private fun ifAliveExclusive(minAliveness: Aliveness = Aliveness.Alive, + ignoreCompilerChanged: Boolean = false, + body: () -> CompileService.CallResult + ): CompileService.CallResult = + rwlock.write { + ifAliveChecksImpl(minAliveness, ignoreCompilerChanged, body) + } inline private fun ifAliveChecksImpl(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> CompileService.CallResult): CompileService.CallResult = when { @@ -635,4 +627,27 @@ class CompileServiceImpl( } } } + + private inline fun withValidClientOrSessionProxy(sessionId: Int, + body: (ClientOrSessionProxy?) -> CompileService.CallResult + ): CompileService.CallResult { + val session: ClientOrSessionProxy? = + if (sessionId == CompileService.NO_SESSION) null + else state.sessions[sessionId] ?: return CompileService.CallResult.Error("Unknown or invalid session $sessionId") + try { + compilationsCounter.incrementAndGet() + return body(session) + } + finally { + _lastUsedSeconds = nowSeconds() + } + + } + + private inline fun withValidRepl(sessionId: Int, body: KotlinJvmReplService.() -> R): CompileService.CallResult = + withValidClientOrSessionProxy(sessionId) { session -> + (session?.data as? KotlinJvmReplService?)?.let { + CompileService.CallResult.Good(it.body()) + } ?: CompileService.CallResult.Error("Not a REPL session $sessionId") + } } diff --git a/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt b/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt index f111a652315..56fd768903c 100644 --- a/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt +++ b/compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt @@ -567,6 +567,48 @@ class CompilerDaemonTest : KotlinIntegrationTestBase() { } } + fun testDaemonReplAutoshutdownOnIdle() { + withFlagFile(getTestName(true), ".alive") { flagFile -> + val daemonOptions = DaemonOptions(autoshutdownIdleSeconds = 1, autoshutdownUnusedSeconds = 1, runFilesPath = File(tmpdir, getTestName(true)).absolutePath) + KotlinCompilerClient.shutdownCompileService(compilerId, daemonOptions) + + val logFile = createTempFile("kotlin-daemon-test", ".log") + val daemonJVMOptions = + configureDaemonJVMOptions("D$COMPILE_DAEMON_LOG_PATH_PROPERTY=\"${logFile.loggerCompatiblePath}\"", + inheritMemoryLimits = false, inheritAdditionalProperties = false) + + val daemon = KotlinCompilerClient.connectToCompileService(compilerId, flagFile, daemonJVMOptions, daemonOptions, DaemonReportingTargets(out = System.err), autostart = true) + assertNotNull("failed to connect daemon", daemon) + + val disposable = Disposer.newDisposable() + val repl = KotlinRemoteReplCompiler(disposable, daemon!!, null, CompileService.TargetPlatform.JVM, + classpathFromClassloader(), + ScriptWithNoParam::class.qualifiedName!!, + System.err) + + // use repl compiler for >> 1s, making sure that idle/unused timeouts are not firing + for (attempts in 1..10) { + val codeLine1 = ReplCodeLine(1, "3 + 5") + val res1 = repl.compile(codeLine1, emptyList()) + val res1c = res1 as? ReplCompileResult.CompiledClasses + TestCase.assertNotNull("Unexpected compile result: $res1", res1c) + Thread.sleep(200) + } + + // wait up to 4s (more than 1s idle timeout) + for (attempts in 1..20) { + if (logFile.isLogContainsSequence("Idle timeout exceeded 1s")) break + Thread.sleep(200) + } + Disposer.dispose(disposable) + + Thread.sleep(200) + logFile.assertLogContainsSequence("Idle timeout exceeded 1s", + "Shutdown complete") + logFile.delete() + } + } + internal fun withDaemon(body: (CompileService) -> Unit) { withFlagFile(getTestName(true), ".alive") { flagFile -> val daemonOptions = DaemonOptions(runFilesPath = File(tmpdir, getTestName(true)).absolutePath)