From 8e98deb3b0ea0edd848b1f5a1f91ba5ccec14060 Mon Sep 17 00:00:00 2001 From: Dmitriy Dolovov Date: Tue, 7 Dec 2021 12:06:43 +0300 Subject: [PATCH] [Native][tests] Read stdout/stderr of test process in separate threads Sometimes, the launched test process can't finish on Windows if stdout and stderr threads are not read completely. The proper approach here would be to spawn two threads for constantly reading stdout & stderr while the process in being executed. --- .../samples/standalone_notr_long_running.kt | 31 +++++++ ...tandalone_notr_long_running_and_verbose.kt | 40 +++++++++ .../samples/standalone_notr_too_verbose.kt | 16 ++++ .../runner/AbstractLocalProcessRunner.kt | 87 +++++++++++++------ .../konan/blackboxtest/support/util/IO.kt | 43 --------- 5 files changed, 146 insertions(+), 71 deletions(-) create mode 100644 native/native.tests/testData/samples/standalone_notr_long_running.kt create mode 100644 native/native.tests/testData/samples/standalone_notr_long_running_and_verbose.kt create mode 100644 native/native.tests/testData/samples/standalone_notr_too_verbose.kt delete mode 100644 native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/util/IO.kt diff --git a/native/native.tests/testData/samples/standalone_notr_long_running.kt b/native/native.tests/testData/samples/standalone_notr_long_running.kt new file mode 100644 index 00000000000..13b02ddad20 --- /dev/null +++ b/native/native.tests/testData/samples/standalone_notr_long_running.kt @@ -0,0 +1,31 @@ +// KIND: STANDALONE_NO_TR + +import kotlin.math.E +import kotlin.math.sqrt +import kotlin.system.getTimeMillis +import kotlin.test.assertTrue + +// Runs for ~60 seconds. Prints a short message to stdout every second. +fun main() { + for (i in 0..60) { + println("Iteration $i") + sleep(1000) + } + println("Done.") +} + +private fun sleep(millis: Int) { + assertTrue(millis > 0) + + val endTimeMillis = getTimeMillis() + millis + do { + // Emulate intensive computations to spend CPU time. + for (i in 1..100) { + for (j in 1..100) { + storage = if (storage.toLong() % 2 == 0L) sqrt(i.toDouble() * j.toDouble()) else E * i / j + } + } + } while (getTimeMillis() < endTimeMillis) +} + +private var storage: Double = 0.0 diff --git a/native/native.tests/testData/samples/standalone_notr_long_running_and_verbose.kt b/native/native.tests/testData/samples/standalone_notr_long_running_and_verbose.kt new file mode 100644 index 00000000000..b550b0d2810 --- /dev/null +++ b/native/native.tests/testData/samples/standalone_notr_long_running_and_verbose.kt @@ -0,0 +1,40 @@ +// KIND: STANDALONE_NO_TR + +import kotlin.math.E +import kotlin.math.sqrt +import kotlin.system.getTimeMillis +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +// Runs for ~60 seconds. Prints 200 thousand bytes to stdout every second. +fun main() { + assertEquals(10, TEN_BYTES_STRING.length) + + for (i in 0..60) { + repeat(200) { print1000Bytes() } + sleep(1000) + } + println("Done.") +} + +private fun sleep(millis: Int) { + assertTrue(millis > 0) + + val endTimeMillis = getTimeMillis() + millis + do { + // Emulate intensive computations to spend CPU time. + for (i in 1..100) { + for (j in 1..100) { + storage = if (storage.toLong() % 2 == 0L) sqrt(i.toDouble() * j.toDouble()) else E * i / j + } + } + } while (getTimeMillis() < endTimeMillis) +} + +private fun print1000Bytes() { + // Print 1000 bytes. + repeat(100) { print(TEN_BYTES_STRING) } +} + +private var storage: Double = 0.0 +private const val TEN_BYTES_STRING = "Hi, test!\n" diff --git a/native/native.tests/testData/samples/standalone_notr_too_verbose.kt b/native/native.tests/testData/samples/standalone_notr_too_verbose.kt new file mode 100644 index 00000000000..05f2ccbafa0 --- /dev/null +++ b/native/native.tests/testData/samples/standalone_notr_too_verbose.kt @@ -0,0 +1,16 @@ +// KIND: STANDALONE_NO_TR + +import kotlin.test.assertEquals + +// Prints 1_000_000 bytes to stdout and exits. +fun main() { + assertEquals(10, TEN_BYTES_STRING.length) + repeat(1000) { print1000Bytes() } +} + +private fun print1000Bytes() { + // Print 1000 bytes. + repeat(100) { print(TEN_BYTES_STRING) } +} + +private const val TEN_BYTES_STRING = "Hi, test!\n" diff --git a/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/runner/AbstractLocalProcessRunner.kt b/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/runner/AbstractLocalProcessRunner.kt index 84735bfb4a5..09da973bdfc 100644 --- a/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/runner/AbstractLocalProcessRunner.kt +++ b/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/runner/AbstractLocalProcessRunner.kt @@ -5,10 +5,12 @@ package org.jetbrains.kotlin.konan.blackboxtest.support.runner +import kotlinx.coroutines.* import org.jetbrains.kotlin.konan.blackboxtest.support.TestExecutable import org.jetbrains.kotlin.konan.blackboxtest.support.runner.AbstractRunner.AbstractRun +import org.jetbrains.kotlin.konan.blackboxtest.support.runner.UnfilteredProcessOutput.Companion.launchReader import org.jetbrains.kotlin.konan.blackboxtest.support.util.TestOutputFilter -import org.jetbrains.kotlin.konan.blackboxtest.support.util.readOutput +import java.io.ByteArrayOutputStream import kotlin.time.* internal abstract class AbstractLocalProcessRunner(private val executionTimeout: Duration) : AbstractRunner() { @@ -21,39 +23,51 @@ internal abstract class AbstractLocalProcessRunner(private val executionTimeo @OptIn(ExperimentalTime::class) final override fun buildRun() = AbstractRun { - val (result, duration) = measureTimedValue { - val process = ProcessBuilder(programArgs).directory(executable.executableFile.parentFile).start() - customizeProcess(process) + runBlocking(Dispatchers.IO) { + val unfilteredOutput = UnfilteredProcessOutput() + val unfilteredOutputReader: Job - val hasFinishedOnTime = process.waitFor( - executionTimeout.toLong(DurationUnit.MILLISECONDS), - DurationUnit.MILLISECONDS.toTimeUnit() - ) + val process: Process + val hasFinishedOnTime: Boolean - process to hasFinishedOnTime - } - val (process, hasFinishedOnTime) = result + val duration = measureTime { + process = ProcessBuilder(programArgs).directory(executable.executableFile.parentFile).start() + customizeProcess(process) - // Don't use blocking read from stdout/stderr on non-finished process. If the process is hanging this would result in hanging test. - val output = process.readOutput(outputFilter, nonBlocking = !hasFinishedOnTime) + unfilteredOutputReader = launchReader(unfilteredOutput, process) - if (hasFinishedOnTime) { - val exitCode: Int = process.exitValue() - - RunResult.Completed(exitCode, duration, output) - } else { - process.destroy() // Initiate destroy of non-finished process. - Thread.sleep(5) // And give it a white to become actually destroyed. - - val exitCode: Int? = try { - // If we are lucky enough, the process is destroyed to this moment. And it's possible to fetch exit code. - process.exitValue() - } catch (_: IllegalThreadStateException) { - // Still not destroyed. Let's go further. - null + hasFinishedOnTime = process.waitFor( + executionTimeout.toLong(DurationUnit.MILLISECONDS), + DurationUnit.MILLISECONDS.toTimeUnit() + ) } - RunResult.TimeoutExceeded(executionTimeout, exitCode, duration, output) + if (hasFinishedOnTime) { + unfilteredOutputReader.join() // Wait until all output streams are drained. + + RunResult.Completed( + exitCode = process.exitValue(), + duration = duration, + processOutput = unfilteredOutput.toProcessOutput(outputFilter) + ) + } else { + val exitCode: Int? = try { // It could happen just by an accident that the process has exited by itself. + val exitCode = process.exitValue() // Fetch exit code. + unfilteredOutputReader.join() // Wait until all streams are drained. + exitCode + } catch (_: IllegalThreadStateException) { // Still not destroyed. + unfilteredOutputReader.cancel() // Cancel it. No need to read streams, actually. + process.destroyForcibly() // kill -9 + null + } + + RunResult.TimeoutExceeded( + timeout = executionTimeout, + exitCode = exitCode, + duration = duration, + processOutput = unfilteredOutput.toProcessOutput(outputFilter) + ) + } } } @@ -69,3 +83,20 @@ internal abstract class AbstractLocalProcessRunner(private val executionTimeo protected abstract fun doHandle(): R } } + +private class UnfilteredProcessOutput { + private val stdOut = ByteArrayOutputStream() + private val stdErr = ByteArrayOutputStream() + + fun toProcessOutput(outputFilter: TestOutputFilter): ProcessOutput = ProcessOutput( + stdOut = outputFilter.filter(stdOut.toString(Charsets.UTF_8)), + stdErr = stdErr.toString(Charsets.UTF_8) + ) + + companion object { + fun CoroutineScope.launchReader(unfilteredOutput: UnfilteredProcessOutput, process: Process): Job = launch { + launch { process.inputStream.copyTo(unfilteredOutput.stdOut) } + launch { process.errorStream.copyTo(unfilteredOutput.stdErr) } + } + } +} diff --git a/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/util/IO.kt b/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/util/IO.kt deleted file mode 100644 index be1896c9123..00000000000 --- a/native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/util/IO.kt +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2010-2021 JetBrains s.r.o. and Kotlin Programming Language contributors. - * Use of this source code is governed by the Apache 2.0 license that can be found in the license/LICENSE.txt file. - */ - -package org.jetbrains.kotlin.konan.blackboxtest.support.util - -import org.jetbrains.kotlin.konan.blackboxtest.support.runner.ProcessOutput -import java.io.ByteArrayOutputStream -import java.io.InputStream - -/** - * Read bytes from the given [InputStream] without blocking. - * - * Note: This function does not guarantee that the whole [InputStream] contents is read. It only - * guarantees that the bytes currently available in [InputStream] are read and no I/O blocks happen. - */ -internal fun InputStream.readBytesNonBlocking(): ByteArray { - val result = ByteArrayOutputStream() - val buffer = ByteArray(128) - - while (true) { - val availableBytes = available() - if (availableBytes == 0) break - - val readBytes = read(buffer) - if (readBytes == 0) break - - result.write(buffer, 0, readBytes) - } - - return result.toByteArray() -} - -internal fun Process.readOutput(outputFilter: TestOutputFilter, nonBlocking: Boolean): ProcessOutput { - val stdOut = if (nonBlocking) inputStream.readBytesNonBlocking() else inputStream.readBytes() - val stdErr = if (nonBlocking) errorStream.readBytesNonBlocking() else errorStream.readBytes() - - return ProcessOutput( - stdOut = outputFilter.filter(stdOut.toString(Charsets.UTF_8)), - stdErr = stdErr.toString(Charsets.UTF_8) - ) -}