diff --git a/compiler/rmi/kotlinr/src/org/jetbrains/kotlin/rmi/kotlinr/KotlinCompilerClient.kt b/compiler/rmi/kotlinr/src/org/jetbrains/kotlin/rmi/kotlinr/KotlinCompilerClient.kt index ed2041ef7b2..091bdfa647a 100644 --- a/compiler/rmi/kotlinr/src/org/jetbrains/kotlin/rmi/kotlinr/KotlinCompilerClient.kt +++ b/compiler/rmi/kotlinr/src/org/jetbrains/kotlin/rmi/kotlinr/KotlinCompilerClient.kt @@ -57,9 +57,9 @@ public object KotlinCompilerClient { val flagFile = System.getProperty(COMPILE_DAEMON_CLIENT_ALIVE_PATH_PROPERTY) ?.let { it.trimQuotes() } - ?.ifOrNull { !isBlank() } + ?.ifOrNull { !it.isBlank() } ?.let { File(it) } - ?.ifOrNull { exists() } + ?.ifOrNull { it.exists() } ?: newFlagFile() return connectToCompileService(compilerId, flagFile, daemonJVMOptions, daemonOptions, reportingTargets, autostart) } @@ -254,14 +254,17 @@ public object KotlinCompilerClient { private fun tryFindSuitableDaemonOrNewOpts(registryDir: File, compilerId: CompilerId, daemonJVMOptions: DaemonJVMOptions, report: (DaemonReportCategory, String) -> Unit): Pair { - val aliveWithOpts = walkDaemons(registryDir, compilerId, report) + val aliveWithOpts = walkDaemons(registryDir, compilerId, report = report) .map { Pair(it, it.getDaemonJVMOptions()) } .filter { it.second.isGood } - // TODO: consider to sort the found daemons by memory settings and take the largest (rather that the first found), but carefully analyze possible situations first + .sortedWith(compareBy(DaemonJVMOptionsMemoryComparator().reversed(), { it.second.get() })) val opts = daemonJVMOptions.copy() - return aliveWithOpts.firstOrNull { daemonJVMOptions memorywiseFitsInto it.second.get() } - ?.let { Pair(it.first, opts.updateMemoryUpperBounds(it.second.get())) } - ?: Pair(null, aliveWithOpts.fold(daemonJVMOptions, { opts, d -> opts.updateMemoryUpperBounds(d.second.get()) })) + // 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()?.ifOrNull { daemonJVMOptions memorywiseFitsInto it.second.get() }?.let { + Pair(it.first, opts.updateMemoryUpperBounds(it.second.get())) + } + // else combine all options from running daemon to get fattest option for a new daemon to run + ?: Pair(null, aliveWithOpts.fold(daemonJVMOptions, { opts, d -> opts.updateMemoryUpperBounds(d.second.get()) })) } @@ -351,4 +354,5 @@ internal fun isProcessAlive(process: Process) = true } -internal fun T.ifOrNull(pred: T.() -> Boolean): T? = if (this.pred()) this else null +internal inline fun T.ifOrNull(pred: (T) -> Boolean): T? = if (pred(this)) this else null + diff --git a/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/ClientUtils.kt b/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/ClientUtils.kt index 25df9acc4c9..9ea20f37d9c 100644 --- a/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/ClientUtils.kt +++ b/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/ClientUtils.kt @@ -40,11 +40,15 @@ fun String.extractPortFromRunFilename(digest: String): Int = ?: 0 -fun walkDaemons(registryDir: File, compilerId: CompilerId, report: (DaemonReportCategory, String) -> Unit): Sequence { +fun walkDaemons(registryDir: File, + compilerId: CompilerId, + filter: (File, Int) -> Boolean = { f,p -> true }, + report: (DaemonReportCategory, String) -> Unit = { cat, msg -> ; } +): Sequence { val classPathDigest = compilerId.compilerClasspath.map { File(it).absolutePath }.distinctStringsDigest().toHexString() return registryDir.walk() .map { Pair(it, it.name.extractPortFromRunFilename(classPathDigest)) } - .filter { it.second != 0 } + .filter { it.second != 0 && filter(it.first, it.second) } .map { assert(it.second > 0 && it.second < MAX_PORT_NUMBER) report(DaemonReportCategory.DEBUG, "found daemon on port ${it.second}, trying to connect") diff --git a/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/CompileService.kt b/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/CompileService.kt index 36d144f5855..00c9afe57ea 100644 --- a/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/CompileService.kt +++ b/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/CompileService.kt @@ -70,14 +70,20 @@ public interface CompileService : Remote { // TODO: consider adding another client alive checking mechanism, e.g. socket/port + @Throws(RemoteException::class) + public fun getClients(): CallResult> + @Throws(RemoteException::class) public fun leaseCompileSession(aliveFlagPath: String?): CallResult @Throws(RemoteException::class) - public fun releaseCompileSession(sessionId: Int): Unit + public fun releaseCompileSession(sessionId: Int): CallResult @Throws(RemoteException::class) - public fun shutdown(): Unit + public fun shutdown(): CallResult + + @Throws(RemoteException::class) + public fun scheduleShutdown(graceful: Boolean): CallResult // TODO: consider adding async version of shutdown and release diff --git a/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/DaemonParams.kt b/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/DaemonParams.kt index 24c482a9b1e..a0874b51003 100644 --- a/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/DaemonParams.kt +++ b/compiler/rmi/rmi-interface/src/org/jetbrains/kotlin/rmi/DaemonParams.kt @@ -21,6 +21,7 @@ import java.io.File import java.io.Serializable import java.lang.management.ManagementFactory import java.security.MessageDigest +import java.util.* import kotlin.reflect.KMutableProperty1 import kotlin.text.RegexOption @@ -224,6 +225,10 @@ public data class DaemonOptions( BoolPropMapper(this, DaemonOptions::reportPerf)) } +// TODO: consider implementing generic approach to it or may be replace getters with ones returning default if necessary +val DaemonOptions.runFilesPathOrDefault: String + get() = if (runFilesPath.isBlank()) COMPILE_DAEMON_DEFAULT_RUN_DIR_PATH else runFilesPath + fun Iterable.distinctStringsDigest(): ByteArray = MessageDigest.getInstance(CLASSPATH_ID_DIGEST) @@ -324,17 +329,31 @@ fun String.memToBytes(): Long? = } -private fun daemonJVMOptionsMemoryProps() = +private val daemonJVMOptionsMemoryProps: List> by lazy { listOf(DaemonJVMOptions::maxMemory, DaemonJVMOptions::maxPermSize, DaemonJVMOptions::reservedCodeCacheSize) - +} infix fun DaemonJVMOptions.memorywiseFitsInto(other: DaemonJVMOptions): Boolean = - daemonJVMOptionsMemoryProps() + daemonJVMOptionsMemoryProps .all { (it.get(this).memToBytes() ?: 0) <= (it.get(other).memToBytes() ?: 0) } +fun compareDaemonJVMOptionsMemory(left: DaemonJVMOptions, right: DaemonJVMOptions): Int { + val props = daemonJVMOptionsMemoryProps.map { Pair(it.get(left).memToBytes() ?: 0, it.get(right).memToBytes() ?: 0) } + return when { + props.all { it.first == it.second } -> 0 + props.all { it.first <= it.second } -> -1 + props.all { it.first >= it.second } -> 1 + else -> 0 + } +} + +class DaemonJVMOptionsMemoryComparator : Comparator { + override fun compare(left: DaemonJVMOptions, right: DaemonJVMOptions): Int = compareDaemonJVMOptionsMemory(left, right) +} + fun DaemonJVMOptions.updateMemoryUpperBounds(other: DaemonJVMOptions): DaemonJVMOptions { - daemonJVMOptionsMemoryProps() + daemonJVMOptionsMemoryProps .forEach { if ((it.get(this).memToBytes() ?: 0) < (it.get(other).memToBytes() ?: 0)) it.set(this, it.get(other)) } return this } diff --git a/compiler/rmi/rmi-server/src/org/jetbrains/kotlin/rmi/service/CompileServiceImpl.kt b/compiler/rmi/rmi-server/src/org/jetbrains/kotlin/rmi/service/CompileServiceImpl.kt index 974f2ff86d3..2e8e64aa1bf 100644 --- a/compiler/rmi/rmi-server/src/org/jetbrains/kotlin/rmi/service/CompileServiceImpl.kt +++ b/compiler/rmi/rmi-server/src/org/jetbrains/kotlin/rmi/service/CompileServiceImpl.kt @@ -67,7 +67,7 @@ class CompileServiceImpl( val compilerId: CompilerId, val daemonOptions: DaemonOptions, val daemonJVMOptions: DaemonJVMOptions, - port: Int, + val port: Int, val timer: Timer, val onShutdown: () -> Unit ) : CompileService { @@ -79,16 +79,27 @@ class CompileServiceImpl( val isAlive: Boolean get() = aliveFlagPath?.let { File(it).exists() } ?: true // assuming that if no file was given, the client is alive } - private val clientProxies: MutableSet = hashSetOf() - private val sessions: MutableMap = hashMapOf() - private val sessionsIdCounter = AtomicInteger(0) + private val compilationsCounter = AtomicInteger(0) private val internalRng = Random() private val classpathWatcher = LazyClasspathWatcher(compilerId.compilerClasspath) - private val compilationsCounter = AtomicInteger(0) - private val shutdownQueued = AtomicBoolean(false) + enum class Aliveness(val value: Int) { + // ordering of values is used in state comparison + Alive(2), LastSession(1), Dying(0) + } + + // TODO: encapsulate operations on state here + private val state = object { + + val clientProxies: MutableSet = hashSetOf() + val sessions: MutableMap = hashMapOf() + + val delayedShutdownQueued = AtomicBoolean(false) + + var alive = AtomicInteger(Aliveness.Alive.value) + } @Volatile private var _lastUsedSeconds = nowSeconds() public val lastUsedSeconds: Long get() = if (rwlock.isWriteLocked || rwlock.readLockCount - rwlock.readHoldCount > 0) nowSeconds() else _lastUsedSeconds @@ -96,12 +107,11 @@ class CompileServiceImpl( val log by lazy { Logger.getLogger("compiler") } private val rwlock = ReentrantReadWriteLock() - private var alive = false private var runFile: File init { - val runFileDir = File(if (daemonOptions.runFilesPath.isBlank()) COMPILE_DAEMON_DEFAULT_RUN_DIR_PATH else daemonOptions.runFilesPath) + val runFileDir = File(daemonOptions.runFilesPathOrDefault) runFileDir.mkdirs() runFile = File(runFileDir, makeRunFilenameString(timestamp = "%tFT% = ifAlive { daemonJVMOptions } override fun registerClient(aliveFlagPath: String?): CompileService.CallResult = ifAlive_Nothing { - synchronized(clientProxies) { - clientProxies.add(ClientOrSessionProxy(aliveFlagPath)) + synchronized(state.clientProxies) { + state.clientProxies.add(ClientOrSessionProxy(aliveFlagPath)) + } + } + + override fun getClients(): CompileService.CallResult> = ifAlive { + synchronized(state.clientProxies) { + state.clientProxies.map { it.aliveFlagPath }.filterNotNull() } } // TODO: consider tying a session to a client and use this info to cleanup - override fun leaseCompileSession(aliveFlagPath: String?): CompileService.CallResult = ifAlive { + override fun leaseCompileSession(aliveFlagPath: String?): CompileService.CallResult = ifAlive(minAliveness = Aliveness.Alive) { // fighting hypothetical integer wrapping var newId = sessionsIdCounter.incrementAndGet() val session = ClientOrSessionProxy(aliveFlagPath) for (attempt in 1..100) { if (newId != CompileService.NO_SESSION) { - synchronized(sessions) { - if (!sessions.containsKey(newId)) { - sessions.put(newId, session) + synchronized(state.sessions) { + if (!state.sessions.containsKey(newId)) { + state.sessions.put(newId, session) return@ifAlive newId } } @@ -147,15 +163,17 @@ class CompileServiceImpl( throw Exception("Invalid state or algorithm error") } - override fun releaseCompileSession(sessionId: Int) { - synchronized(sessions) { - sessions.remove(sessionId) + override fun releaseCompileSession(sessionId: Int) = ifAlive_Nothing(minAliveness = Aliveness.LastSession) { + synchronized(state.sessions) { + state.sessions.remove(sessionId) // TODO: some cleanup goes here - if (sessions.isEmpty()) { + if (state.sessions.isEmpty()) { // TODO: and some goes here } } - periodicAndAfterSessionCheck() + timer.schedule(0) { + periodicAndAfterSessionCheck() + } } override fun checkCompilerId(expectedCompilerId: CompilerId): Boolean = @@ -165,14 +183,25 @@ class CompileServiceImpl( override fun getUsedMemory(): CompileService.CallResult = ifAlive { usedMemory(withGC = true) } - override fun shutdown() { - ifAliveExclusive(ignoreCompilerChanged = true) { - log.info("Shutdown started") - alive = false - UnicastRemoteObject.unexportObject(this, true) - log.info("Shutdown complete") - onShutdown() + override fun shutdown(): CompileService.CallResult = ifAliveExclusive_Nothing(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) { + shutdownImpl() + } + + override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult = ifAlive(minAliveness = Aliveness.Alive) { + if (!graceful || state.alive.compareAndSet(Aliveness.Alive.value, Aliveness.LastSession.value)) { + 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") + } + } + } + true } + else false } override fun remoteCompile(sessionId: Int, @@ -234,8 +263,10 @@ class CompileServiceImpl( val stub = UnicastRemoteObject.exportObject(this, port, LoopbackNetworkInterface.clientLoopbackSocketFactory, LoopbackNetworkInterface.serverLoopbackSocketFactory) as CompileService registry.rebind (COMPILER_SERVICE_RMI_NAME, stub); - alive = true + timer.schedule(0) { + initiateElections() + } timer.schedule(delay = DAEMON_PERIODIC_CHECK_INTERVAL_MS, period = DAEMON_PERIODIC_CHECK_INTERVAL_MS) { try { periodicAndAfterSessionCheck() @@ -248,45 +279,99 @@ class CompileServiceImpl( } } + private fun periodicAndAfterSessionCheck() { - // 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() - } - else { - synchronized(sessions) { - // 2. check if any session hanged - clean - // making copy of the list before calling release - sessions.filterValues { !it.isAlive }.keys.toArrayList() - }.forEach { releaseCompileSession(it) } + ifAlive_Nothing(minAliveness = Aliveness.LastSession) { - // 3. clean dead clients, then check if any left - conditional shutdown (with small delay) - synchronized(clientProxies) { clientProxies.removeAll( clientProxies.filter{ !it.isAlive }) } - if (clientProxies.none() && compilationsCounter.get() > 0 && !shutdownQueued.get()) { - log.info("No more clients left, delayed shutdown in ${daemonOptions.shutdownDelayMilliseconds}ms") - shutdownWithDelay() - } - // 4. 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") || - // 5. discovery file removed - shutdown - shutdownCondition({ !runFile.exists() }, "Run file removed, shutting down") || - // 6. compiler changed (seldom check) - shutdown - // TODO: could be too expensive anyway, consider removing this check - shutdownCondition({ classpathWatcher.isChanged }, "Compiler changed")) - { + // 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() } + else { + synchronized(state.sessions) { + // 2. check if any session hanged - clean + // making copy of the list before calling release + state.sessions.filterValues { !it.isAlive }.keys.toArrayList() + }.forEach { releaseCompileSession(it) } + + // 3. check if in graceful shutdown state and all sessions are closed + if (shutdownCondition({state.alive.get() == Aliveness.LastSession.value && state.sessions.none()}, "All sessions finished, shutting down")) { + shutdown() + } + + // 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 }) } + if (state.clientProxies.none() && 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")) { + shutdown() + } + } } } + + private fun initiateElections() { + + ifAlive_Nothing { + + 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(compareBy(DaemonJVMOptionsMemoryComparator().reversed(), { it.second.get() })) + if (aliveWithOpts.any()) { + val fattestOpts = aliveWithOpts.first().second.get() + // 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)) { + // all others are smaller that me, take overs' clients and shut them down + aliveWithOpts.forEach { + it.first.getClients().ifOrNull { it.isGood }?.let { + it.get().forEach { registerClient(it) } + } + it.first.scheduleShutdown(true) + } + } + else if (daemonJVMOptions memorywiseFitsInto fattestOpts) { + // there is at least one bigger, handover my clients to it and shutdown + scheduleShutdown(true) + aliveWithOpts.first().first.let { fattest -> + getClients().ifOrNull { 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 daemosn + // - 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) + } + } + } + + private fun shutdownImpl() { + log.info("Shutdown started") + state.alive.set(Aliveness.Dying.value) + UnicastRemoteObject.unexportObject(this, true) + log.info("Shutdown complete") + onShutdown() + } + private fun shutdownWithDelay() { - shutdownQueued.set(true) + state.delayedShutdownQueued.set(true) val currentCompilationsCount = compilationsCounter.get() timer.schedule(daemonOptions.shutdownDelayMilliseconds) { - shutdownQueued.set(false) + state.delayedShutdownQueued.set(false) if (currentCompilationsCount == compilationsCounter.get()) { log.fine("Execute delayed shutdown") shutdown() @@ -388,28 +473,37 @@ class CompileServiceImpl( } } - private fun ifAlive(ignoreCompilerChanged: Boolean = false, body: () -> R): CompileService.CallResult = rwlock.read { - ifAliveChecksImpl(ignoreCompilerChanged) { CompileService.CallResult.Good(body()) } + private fun ifAlive(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> R): CompileService.CallResult = rwlock.read { + ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { CompileService.CallResult.Good(body()) } } // TODO: find how to implement it without using unique name for this variant; making name deliberately ugly meanwhile - private fun ifAlive_Nothing(ignoreCompilerChanged: Boolean = false, body: () -> Unit): CompileService.CallResult = rwlock.read { - ifAliveChecksImpl(ignoreCompilerChanged) { + 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 ifAliveExclusive(ignoreCompilerChanged: Boolean = false, body: () -> R): CompileService.CallResult = rwlock.write { - ifAliveChecksImpl(ignoreCompilerChanged) { CompileService.CallResult.Good(body()) } + private fun ifAliveExclusive(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> R): CompileService.CallResult = rwlock.write { + ifAliveChecksImpl(minAliveness, ignoreCompilerChanged) { CompileService.CallResult.Good(body()) } } - inline private fun ifAliveChecksImpl(ignoreCompilerChanged: Boolean = false, body: () -> CompileService.CallResult): CompileService.CallResult = + // 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() + + } + } + + inline private fun ifAliveChecksImpl(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> CompileService.CallResult): CompileService.CallResult = when { - !alive -> CompileService.CallResult.Dying() + state.alive.get() < minAliveness.value -> CompileService.CallResult.Dying() !ignoreCompilerChanged && classpathWatcher.isChanged -> { log.info("Compiler changed, scheduling shutdown") - timer.schedule(1) { shutdown() } + timer.schedule(0) { shutdown() } CompileService.CallResult.Dying() } else -> { @@ -424,3 +518,6 @@ class CompileServiceImpl( } } + +internal inline fun T.ifOrNull(pred: (T) -> Boolean): T? = if (pred(this)) this else null +