Fixes after review
This commit is contained in:
Ilya Chernikov
+13
-11
@@ -54,26 +54,26 @@ fun walkDaemons(registryDir: File,
|
||||
val portExtractor = makePortFromRunFilenameExtractor(classPathDigest)
|
||||
return registryDir.walk()
|
||||
.map { Pair(it, portExtractor(it.name)) }
|
||||
.filter { it.second != null && filter(it.first, it.second!!) }
|
||||
.mapNotNull { fileWithPort ->
|
||||
assert(fileWithPort.second!! in 1..(MAX_PORT_NUMBER - 1))
|
||||
val relativeAge = fileToCompareTimestamp.lastModified() - fileWithPort.first.lastModified()
|
||||
report(DaemonReportCategory.DEBUG, "found daemon on port ${fileWithPort.second} ($relativeAge ms old), trying to connect")
|
||||
val daemon = tryConnectToDaemon(fileWithPort.second!!, report)
|
||||
.filter { (file, port) -> port != null && filter(file, port) }
|
||||
.mapNotNull { (file, port) ->
|
||||
assert(port!! in 1..(MAX_PORT_NUMBER - 1))
|
||||
val relativeAge = fileToCompareTimestamp.lastModified() - file.lastModified()
|
||||
report(DaemonReportCategory.DEBUG, "found daemon on port $port ($relativeAge ms old), trying to connect")
|
||||
val daemon = tryConnectToDaemon(port, report)
|
||||
// cleaning orphaned file; note: daemon should shut itself down if it detects that the run file is deleted
|
||||
if (daemon == null) {
|
||||
if (relativeAge - ORPHANED_RUN_FILE_AGE_THRESHOLD_MS <= 0) {
|
||||
report(DaemonReportCategory.DEBUG, "found fresh run file '${fileWithPort.first.absolutePath}' ($relativeAge ms old), but no daemon, ignoring it")
|
||||
report(DaemonReportCategory.DEBUG, "found fresh run file '${file.absolutePath}' ($relativeAge ms old), but no daemon, ignoring it")
|
||||
}
|
||||
else {
|
||||
report(DaemonReportCategory.DEBUG, "found seemingly orphaned run file '${fileWithPort.first.absolutePath}' ($relativeAge ms old), deleting it")
|
||||
if (!fileWithPort.first.delete()) {
|
||||
report(DaemonReportCategory.INFO, "WARNING: unable to delete seemingly orphaned file '${fileWithPort.first.absolutePath}', cleanup recommended")
|
||||
report(DaemonReportCategory.DEBUG, "found seemingly orphaned run file '${file.absolutePath}' ($relativeAge ms old), deleting it")
|
||||
if (!file.delete()) {
|
||||
report(DaemonReportCategory.INFO, "WARNING: unable to delete seemingly orphaned file '${file.absolutePath}', cleanup recommended")
|
||||
}
|
||||
}
|
||||
}
|
||||
try {
|
||||
daemon?.let { DaemonWithMetadata(it, fileWithPort.first, it.getDaemonJVMOptions().get()) }
|
||||
daemon?.let { DaemonWithMetadata(it, file, it.getDaemonJVMOptions().get()) }
|
||||
}
|
||||
catch (e: Exception) {
|
||||
report(DaemonReportCategory.INFO, "ERROR: unable to retrieve daemon JVM options, assuming daemon is dead: ${e.message}")
|
||||
@@ -120,3 +120,5 @@ class FileAgeComparator : Comparator<File> {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const val LOG_PREFIX_ASSUMING_OTHER_DAEMONS_HAVE = "Assuming other daemons have"
|
||||
|
||||
@@ -599,6 +599,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 {
|
||||
@@ -608,35 +609,47 @@ class CompileServiceImpl(
|
||||
.thenBy(FileAgeComparator()) { it.runFile }
|
||||
aliveWithOpts.maxWith(comparator)?.let { bestDaemonWithMetadata ->
|
||||
val fattestOpts = bestDaemonWithMetadata.jvmOptions
|
||||
// second part of the condition means that we prefer other daemon if is "equal" to the current one
|
||||
if (fattestOpts memorywiseFitsInto daemonJVMOptions && FileAgeComparator().compare(bestDaemonWithMetadata.runFile, runFile) < 0 ) {
|
||||
// all others are smaller that me, take overs' clients and shut them down
|
||||
log.info("Assuming other daemons have lower prio, taking clients from them and schedule them to shutdown: my runfile: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
|
||||
aliveWithOpts.forEach {
|
||||
log.info("$LOG_PREFIX_ASSUMING_OTHER_DAEMONS_HAVE lower prio, taking clients from them and schedule them to shutdown: my runfile: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
|
||||
aliveWithOpts.forEach { (daemon, runFile, _) ->
|
||||
try {
|
||||
it.daemon.getClients().check { it.isGood }?.let {
|
||||
it.get().forEach { registerClient(it) }
|
||||
daemon.getClients().takeIf { it.isGood }?.let {
|
||||
it.get().forEach { clientAliveFile -> registerClient(clientAliveFile) }
|
||||
}
|
||||
it.daemon.scheduleShutdown(true)
|
||||
daemon.scheduleShutdown(true)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
log.info("Cannot connect to a daemon, assuming dying ('${it.runFile.canonicalPath}'): ${e.message}")
|
||||
log.info("Cannot connect to a daemon, assuming dying ('${runFile.canonicalPath}'): ${e.message}")
|
||||
}
|
||||
}
|
||||
}
|
||||
// TODO: seems that the second part of condition is incorrect, reconsider:
|
||||
// the comment by @tsvtkv from review:
|
||||
// Algorithm in plain english:
|
||||
// (1) If the best daemon fits into me and the best daemon is younger than me, then I take over all other daemons clients.
|
||||
// (2) If I fit into the best daemon and the best daemon is older than me, then I give my clients to that daemon.
|
||||
//
|
||||
// For example:
|
||||
//
|
||||
// daemon A starts with params: maxMem=100, codeCache=50
|
||||
// daemon B starts with params: maxMem=200, codeCache=50
|
||||
// daemon C starts with params: maxMem=150, codeCache=100
|
||||
// A performs election: (1) is false because neither B nor C does not fit into A, (2) is false because both B and C are younger than A.
|
||||
// B performs election: (1) is false because neither A nor C does not fit into B, (2) is false because B does not fit into neither A nor C.
|
||||
// C performs election: (1) is false because B is better than A and B does not fit into C, (2) is false C does not fit into neither A nor B.
|
||||
// Result: all daemons are alive and well.
|
||||
else if (daemonJVMOptions memorywiseFitsInto fattestOpts && FileAgeComparator().compare(bestDaemonWithMetadata.runFile, runFile) > 0) {
|
||||
// there is at least one bigger, handover my clients to it and shutdown
|
||||
log.info("Assuming other daemons have higher prio, handover clients to it and schedule shutdown: my runfile: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
|
||||
scheduleShutdown(true)
|
||||
aliveWithOpts.first().daemon.let { fattest ->
|
||||
getClients().check { it.isGood }?.let {
|
||||
it.get().forEach { fattest.registerClient(it) }
|
||||
}
|
||||
log.info("$LOG_PREFIX_ASSUMING_OTHER_DAEMONS_HAVE higher prio, handover clients to it and schedule shutdown: my runfile: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
|
||||
getClients().takeIf { it.isGood }?.let {
|
||||
it.get().forEach { bestDaemonWithMetadata.daemon.registerClient(it) }
|
||||
}
|
||||
scheduleShutdown(true)
|
||||
}
|
||||
else {
|
||||
// undecided, do nothing
|
||||
log.info("Assuming other daemon(s) have equal prio, continue: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
|
||||
log.info("$LOG_PREFIX_ASSUMING_OTHER_DAEMONS_HAVE equal prio, continue: ${runFile.name} (${runFile.lastModified()}) vs best other runfile: ${bestDaemonWithMetadata.runFile.name} (${bestDaemonWithMetadata.runFile.lastModified()})")
|
||||
// TODO: implement some behaviour here, e.g.:
|
||||
// - 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)
|
||||
|
||||
@@ -73,16 +73,6 @@ class CompilerDaemonTest : KotlinIntegrationTestBase() {
|
||||
|
||||
private fun run(logName: String, vararg args: String): Int = runJava(getTestBaseDir(), logName, *args)
|
||||
|
||||
override fun setUp() {
|
||||
super.setUp()
|
||||
System.setProperty("COMPILE_DAEMON_VERBOSE_REPORT_PROPERTY", "true")
|
||||
}
|
||||
|
||||
override fun tearDown() {
|
||||
System.clearProperty("COMPILE_DAEMON_VERBOSE_REPORT_PROPERTY")
|
||||
super.tearDown()
|
||||
}
|
||||
|
||||
fun testHelloApp() {
|
||||
withFlagFile(getTestName(true), ".alive") { flagFile ->
|
||||
val daemonOptions = DaemonOptions(runFilesPath = File(tmpdir, getTestName(true)).absolutePath,
|
||||
@@ -415,12 +405,11 @@ class CompilerDaemonTest : KotlinIntegrationTestBase() {
|
||||
|
||||
val PARALLEL_THREADS_TO_START = 10
|
||||
|
||||
val startLatch = CountDownLatch(1)
|
||||
val doneLatch = CountDownLatch(PARALLEL_THREADS_TO_START)
|
||||
|
||||
val resultCodes = arrayOfNulls<Int>(PARALLEL_THREADS_TO_START)
|
||||
val outStreams = Array(PARALLEL_THREADS_TO_START, { ByteArrayOutputStream() })
|
||||
val logFiles = kotlin.arrayOfNulls<File>(PARALLEL_THREADS_TO_START)
|
||||
val logFiles = arrayOfNulls<File>(PARALLEL_THREADS_TO_START)
|
||||
|
||||
fun connectThread(threadNo: Int) = thread(name = "daemonConnect$threadNo") {
|
||||
try {
|
||||
@@ -450,18 +439,19 @@ class CompilerDaemonTest : KotlinIntegrationTestBase() {
|
||||
|
||||
(1..PARALLEL_THREADS_TO_START).forEach { connectThread(it - 1) }
|
||||
|
||||
val succeeded = doneLatch.await(PARALLEL_WAIT_TIMEOUT_S * 2, TimeUnit.SECONDS)
|
||||
val succeeded = doneLatch.await(PARALLEL_WAIT_TIMEOUT_S, TimeUnit.SECONDS)
|
||||
assertTrue("parallel daemons start failed to complete in $PARALLEL_WAIT_TIMEOUT_S s, ${doneLatch.count} unfinished threads", succeeded)
|
||||
|
||||
Thread.sleep(100) // Wait for processes to finish and close log files
|
||||
|
||||
val electionLogs = logFiles.map { it to it?.readLines()?.find { it.contains("Assuming other daemons have") } }
|
||||
val electionLogs = logFiles.map { it to it?.readLines()?.find { it.contains(LOG_PREFIX_ASSUMING_OTHER_DAEMONS_HAVE) } }
|
||||
|
||||
assertTrue("No daemon elected: \n${electionLogs.joinToString("\n")}", electionLogs.any { it.second?.let { it.contains("lower prio") || it.contains("equal prio") } ?: false })
|
||||
assertTrue("No daemon elected: \n${electionLogs.joinToString("\n")}",
|
||||
electionLogs.any { (_, electionLine) -> electionLine != null && (electionLine.contains("lower prio") || electionLine.contains("equal prio")) })
|
||||
|
||||
electionLogs.forEach { it.first?.delete() }
|
||||
electionLogs.forEach { (logFile, _) -> logFile?.delete() }
|
||||
|
||||
(2..PARALLEL_THREADS_TO_START).forEach {
|
||||
(1..PARALLEL_THREADS_TO_START).forEach {
|
||||
assertEquals("Compilation on thread $it failed:\n${outStreams[it - 1]}", 0, resultCodes[it - 1])
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user