azalea/kotlin-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor daemon periodic check and shutdown logic

Browse Source 
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).
This commit is contained in:
Ilya Chernikov
2017-06-23 19:00:23 +02:00
parent 2c35675fd6
commit 8542e878f7
3 changed files with 173 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compiler/daemon/src/org/jetbrains/kotlin/daemon/CompileServiceImpl.kt
+170 -128
View File
@@ -114,8 +114,6 @@ class CompileServiceImpl(
// wrapped in a class to encapsulate alive check logic
private class ClientOrSessionProxy<out T: Any>(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<ClientOrSessionProxy<Any>> = hashSetOf()
private val clientsLock = ReentrantReadWriteLock()
private val clientProxies: MutableSet<ClientOrSessionProxy<Any>> = 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<String> = 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<T> Iterable<T>.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<Nothing> = ifAlive {
synchronized(state.clientProxies) {
state.clientProxies.add(ClientOrSessionProxy(aliveFlagPath))
log.info("Registered a client alive file: $aliveFlagPath")
}
override fun registerClient(aliveFlagPath: String?): CompileService.CallResult<Nothing> = ifAlive(minAliveness = Aliveness.Alive) {
state.addClient(aliveFlagPath)
log.info("Registered a client alive file: $aliveFlagPath")
CompileService.CallResult.Ok()
}
override fun getClients(): CompileService.CallResult<List<String>> = 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<Long> =
ifAlive { CompileService.CallResult.Good(usedMemory(withGC = true)) }
override fun shutdown(): CompileService.CallResult<Nothing> = ifAliveExclusive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) {
override fun shutdown(): CompileService.CallResult<Nothing> = ifAliveExclusive(minAliveness = Aliveness.LastSession) {
shutdownWithDelay()
CompileService.CallResult.Ok()
}
override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult<Boolean> = ifAlive(minAliveness = Aliveness.LastSession, ignoreCompilerChanged = true) {
when {
!graceful -> {
override fun scheduleShutdown(graceful: Boolean): CompileService.CallResult<Boolean> = 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<DaemonWithMetadata, DaemonJVMOptions>(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<R> ifAlive(minAliveness: Aliveness = Aliveness.Alive,
ignoreCompilerChanged: Boolean = false,
body: () -> CompileService.CallResult<R>
): CompileService.CallResult<R> =
rwlock.read {
ifAliveChecksImpl(minAliveness, ignoreCompilerChanged, body)
}
private inline fun<R> ifAlive(minAliveness: Aliveness = Aliveness.LastSession, body: () -> CompileService.CallResult<R>): CompileService.CallResult<R> = rwlock.read {
ifAliveChecksImpl(minAliveness, body)
}
private fun<R> ifAliveExclusive(minAliveness: Aliveness = Aliveness.Alive,
ignoreCompilerChanged: Boolean = false,
body: () -> CompileService.CallResult<R>
): CompileService.CallResult<R> =
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<R> ifAliveChecksImpl(minAliveness: Aliveness = Aliveness.Alive, ignoreCompilerChanged: Boolean = false, body: () -> CompileService.CallResult<R>): CompileService.CallResult<R> {
private inline fun<R> ifAliveExclusive(minAliveness: Aliveness = Aliveness.LastSession, body: () -> CompileService.CallResult<R>): CompileService.CallResult<R> = 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<R> ifAliveChecksImpl(minAliveness: Aliveness = Aliveness.LastSession, body: () -> CompileService.CallResult<R>): CompileService.CallResult<R> {
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 -> {
compiler/daemon/src/org/jetbrains/kotlin/daemon/KotlinCompileDaemon.kt
+1
View File
@@ -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() {
compiler/tests/org/jetbrains/kotlin/daemon/CompilerDaemonTest.kt
+2 -2
View File
@@ -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")
}
}