[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.

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

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"

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"

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<R>(private val executionTimeout: Duration) : AbstractRunner<R>() {
@@ -21,39 +23,51 @@ internal abstract class AbstractLocalProcessRunner<R>(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<R>(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) }
+        }
+    }
+}

native/native.tests/tests/org/jetbrains/kotlin/konan/blackboxtest/support/util/IO.kt

@@ -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)
-    )
-}