BlockHound integration

build.gradle

@@ -8,8 +8,8 @@ apply from: rootProject.file("gradle/experimental.gradle")
 def rootModule = "kotlinx.coroutines"
 def coreModule = "kotlinx-coroutines-core"
 // Not applicable for Kotlin plugin
-def sourceless = ['kotlinx.coroutines', 'site', 'kotlinx-coroutines-bom', 'publication-validator']
-def internal = ['kotlinx.coroutines', 'site', 'benchmarks', 'js-stub', 'stdlib-stubs', 'publication-validator']
+def sourceless = ['kotlinx.coroutines', 'site', 'kotlinx-coroutines-bom', 'integration-testing']
+def internal = ['kotlinx.coroutines', 'site', 'benchmarks', 'js-stub', 'stdlib-stubs', 'integration-testing']
 // Not published
 def unpublished = internal + ['example-frontend-js', 'android-unit-tests']
 

gradle.properties

@@ -14,13 +14,15 @@ knit_version=0.1.3
 html_version=0.6.8
 lincheck_version=2.5.3
 dokka_version=0.9.16-rdev-2-mpp-hacks
-byte_buddy_version=1.9.3
+byte_buddy_version=1.10.7
 reactor_vesion=3.2.5.RELEASE
 reactive_streams_version=1.0.2
 rxjava2_version=2.2.8
 javafx_version=11.0.2
 javafx_plugin_version=0.0.8
 binary_compatibility_validator_version=0.2.2
+blockhound_version=1.0.2.RELEASE
+jna_version=5.5.0
 
 # Android versions
 android_version=4.1.1.4

integration-testing/README.md

@@ -0,0 +1,14 @@
+# Integration tests
+
+This is a supplementary subproject of kotlinx.coroutines that provides
+integration tests.
+
+The tests are the following:
+* `NpmPublicationValidator` tests that version of NPM artifact is correct and that it has neither source nor package dependencies on atomicfu
+  In order for the test to work, one needs to run gradle with `-PdryRun=true`.
+  `-PdryRun` affects `npmPublish` so that it only provides a packed publication
+  and does not in fact attempt to send the build for publication.
+* `MavenPublicationValidator` depends on the published artifacts and tests artifacts binary content and absence of atomicfu in the classpath
+* `DebugAgentTest` checks that the coroutine debugger can be run as a Java agent.
+
+All the available tests can be run with `integration-testing:test`.

integration-testing/build.gradle

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+apply from: rootProject.file("gradle/compile-jvm.gradle")
+
+repositories {
+    mavenLocal()
+    mavenCentral()
+}
+
+sourceSets {
+    npmTest {
+        kotlin
+        compileClasspath += sourceSets.test.runtimeClasspath
+        runtimeClasspath += sourceSets.test.runtimeClasspath
+    }
+    mavenTest {
+        kotlin
+        compileClasspath += sourceSets.test.runtimeClasspath
+        runtimeClasspath += sourceSets.test.runtimeClasspath
+    }
+    debugAgentTest {
+        kotlin
+        compileClasspath += sourceSets.test.runtimeClasspath
+        runtimeClasspath += sourceSets.test.runtimeClasspath
+    }
+}
+
+task npmTest(type: Test) {
+    def sourceSet = sourceSets.npmTest
+    environment "projectRoot", project.rootDir
+    environment "deployVersion", version
+    def dryRunNpm = project.properties['dryRun']
+    def doRun = dryRunNpm == "true" // so that we don't accidentally publish anything, especially before the test
+    onlyIf { doRun }
+    if (doRun) { // `onlyIf` only affects execution of the task, not the dependency subtree
+        dependsOn(project(':').getTasksByName("publishNpm", true))
+    }
+    testClassesDirs = sourceSet.output.classesDirs
+    classpath = sourceSet.runtimeClasspath
+}
+
+task mavenTest(type: Test) {
+    def sourceSet = sourceSets.mavenTest
+    dependsOn(project(':').getTasksByName("publishToMavenLocal", true))
+    dependsOn.remove(project(':').getTasksByName("dokka", true))
+    testClassesDirs = sourceSet.output.classesDirs
+    classpath = sourceSet.runtimeClasspath
+}
+
+task debugAgentTest(type: Test) {
+    def sourceSet = sourceSets.debugAgentTest
+    dependsOn(project(':kotlinx-coroutines-debug').shadowJar)
+    jvmArgs ('-javaagent:' + project(':kotlinx-coroutines-debug').shadowJar.outputs.files.getFiles()[0])
+    testClassesDirs = sourceSet.output.classesDirs
+    classpath = sourceSet.runtimeClasspath
+}
+
+dependencies {
+    testCompile "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
+    testCompile 'junit:junit:4.12'
+    npmTestCompile 'org.apache.commons:commons-compress:1.18'
+    npmTestCompile 'com.google.code.gson:gson:2.8.5'
+    mavenTestRuntimeOnly project(':kotlinx-coroutines-core')
+    mavenTestRuntimeOnly project(':kotlinx-coroutines-android')
+    debugAgentTestCompile project(':kotlinx-coroutines-core')
+    debugAgentTestCompile project(':kotlinx-coroutines-debug')
+}
+
+compileTestKotlin {
+    kotlinOptions.jvmTarget = "1.8"
+}
+
+test {
+    dependsOn([npmTest, mavenTest, debugAgentTest])
+}

integration-testing/src/debugAgentTest/kotlin/DebugAgentTest.kt

@@ -0,0 +1,20 @@
+/*
+ *  Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+import org.junit.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.debug.*
+import java.io.*
+
+class DebugAgentTest {
+
+    @Test
+    fun agentDumpsCoroutines() = runBlocking {
+        val baos = ByteArrayOutputStream()
+        DebugProbes.dumpCoroutines(PrintStream(baos))
+        // if the agent works, then dumps should contain something,
+        // at least the fact that this test is running.
+        Assert.assertTrue(baos.toString().contains("agentDumpsCoroutines"))
+    }
+
+}

publication-validator/src/test/kotlin/kotlinx/coroutines/tools/MavenPublicationValidator.kt → integration-testing/src/mavenTest/kotlin/MavenPublicationValidator.kt

@@ -6,7 +6,6 @@ package kotlinx.coroutines.validator
 
 import org.junit.*
 import org.junit.Assert.assertTrue
-import java.io.*
 import java.util.jar.*
 
 class MavenPublicationValidator {

kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt

@@ -951,3 +951,20 @@ internal class CoroutineScheduler(
         TERMINATED
     }
 }
+
+/**
+ * Checks if the thread is part of a thread pool that supports coroutines.
+ * This function is needed for integration with BlockHound.
+ */
+@Suppress("UNUSED")
+@JvmName("isSchedulerWorker")
+internal fun isSchedulerWorker(thread: Thread) = thread is CoroutineScheduler.Worker
+
+/**
+ * Checks if the thread is running a CPU-bound task.
+ * This function is needed for integration with BlockHound.
+ */
+@Suppress("UNUSED")
+@JvmName("mayNotBlock")
+internal fun mayNotBlock(thread: Thread) = thread is CoroutineScheduler.Worker &&
+    thread.state == CoroutineScheduler.WorkerState.CPU_ACQUIRED

kotlinx-coroutines-debug/README.md

@@ -13,6 +13,10 @@ suspension stacktraces.
 Additionally, it is possible to process the list of such coroutines via [DebugProbes.dumpCoroutinesInfo] or dump isolated parts
 of coroutines hierarchy referenced by a [Job] or [CoroutineScope] instances using  [DebugProbes.printJob] and [DebugProbes.printScope] respectively.
 
+This module also provides an automatic [BlockHound](https://github.com/reactor/BlockHound) integration
+that detects when a blocking operation was called in a coroutine context that prohibits it. In order to use it,
+please follow the BlockHound [quick start guide](https://github.com/reactor/BlockHound/blob/master/docs/quick_start.md).
+
 ### Using in your project
 
 Add `kotlinx-coroutines-debug` to your project test dependencies:

kotlinx-coroutines-debug/api/kotlinx-coroutines-debug.api

@@ -8,6 +8,11 @@ public final class kotlinx/coroutines/debug/CoroutineInfo {
 	public fun toString ()Ljava/lang/String;
 }
 
+public final class kotlinx/coroutines/debug/CoroutinesBlockHoundIntegration : reactor/blockhound/integration/BlockHoundIntegration {
+	public fun <init> ()V
+	public fun applyTo (Lreactor/blockhound/BlockHound$Builder;)V
+}
+
 public final class kotlinx/coroutines/debug/DebugProbes {
 	public static final field INSTANCE Lkotlinx/coroutines/debug/DebugProbes;
 	public final fun dumpCoroutines (Ljava/io/PrintStream;)V

kotlinx-coroutines-debug/build.gradle

@@ -22,6 +22,10 @@ dependencies {
     compileOnly "junit:junit:$junit_version"
     shadowDeps "net.bytebuddy:byte-buddy:$byte_buddy_version"
     shadowDeps "net.bytebuddy:byte-buddy-agent:$byte_buddy_version"
+    compileOnly "io.projectreactor.tools:blockhound:$blockhound_version"
+    testCompile "io.projectreactor.tools:blockhound:$blockhound_version"
+    runtime "net.java.dev.jna:jna:$jna_version"
+    runtime "net.java.dev.jna:jna-platform:$jna_version"
 }
 
 jar {
@@ -35,5 +39,5 @@ shadowJar {
     classifier null
     // Shadow only byte buddy, do not package kotlin stdlib
     configurations = [project.configurations.shadowDeps]
-    relocate 'net.bytebuddy', 'kotlinx.coroutines.repackaged.net.bytebuddy'
+    relocate('net.bytebuddy', 'kotlinx.coroutines.repackaged.net.bytebuddy')
 }

kotlinx-coroutines-debug/resources/META-INF/services/reactor.blockhound.integration.BlockHoundIntegration

@@ -0,0 +1 @@
+kotlinx.coroutines.debug.CoroutinesBlockHoundIntegration

kotlinx-coroutines-debug/src/CoroutinesBlockHoundIntegration.kt

@@ -0,0 +1,16 @@
+@file:Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")
+package kotlinx.coroutines.debug
+
+import reactor.blockhound.BlockHound
+import kotlinx.coroutines.scheduling.*
+import reactor.blockhound.integration.*
+
+@Suppress("UNUSED")
+public class CoroutinesBlockHoundIntegration: BlockHoundIntegration {
+
+    override fun applyTo(builder: BlockHound.Builder) {
+        builder.addDynamicThreadPredicate { isSchedulerWorker(it) }
+        builder.nonBlockingThreadPredicate { p -> p.or { mayNotBlock(it) } }
+    }
+
+}

kotlinx-coroutines-debug/src/internal/DebugProbesImpl.kt

@@ -57,7 +57,7 @@ internal object DebugProbesImpl {
     public fun install(): Unit = coroutineStateLock.write {
         if (++installations > 1) return
 
-        ByteBuddyAgent.install()
+        ByteBuddyAgent.install(ByteBuddyAgent.AttachmentProvider.ForEmulatedAttachment.INSTANCE)
         val cl = Class.forName("kotlin.coroutines.jvm.internal.DebugProbesKt")
         val cl2 = Class.forName("kotlinx.coroutines.debug.DebugProbesKt")
 

kotlinx-coroutines-debug/test/BlockHoundTest.kt

@@ -0,0 +1,73 @@
+package kotlinx.coroutines.debug
+import kotlinx.coroutines.*
+import org.junit.*
+import reactor.blockhound.*
+
+class BlockHoundTest : TestBase() {
+
+    @Before
+    fun init() {
+        BlockHound.install()
+    }
+
+    @Test(expected = BlockingOperationError::class)
+    fun shouldDetectBlockingInDefault() = runTest {
+        withContext(Dispatchers.Default) {
+            Thread.sleep(1)
+        }
+    }
+
+    @Test
+    fun shouldNotDetectBlockingInIO() = runTest {
+        withContext(Dispatchers.IO) {
+            Thread.sleep(1)
+        }
+    }
+
+    @Test
+    fun shouldNotDetectNonblocking() = runTest {
+        withContext(Dispatchers.Default) {
+            val a = 1
+            val b = 2
+            assert(a + b == 3)
+        }
+    }
+
+    @Test
+    fun testReusingThreads() = runTest {
+        val n = 100
+        repeat(n) {
+            async(Dispatchers.IO) {
+                Thread.sleep(1)
+            }
+        }
+        repeat(n) {
+            async(Dispatchers.Default) {
+            }
+        }
+        repeat(n) {
+            async(Dispatchers.IO) {
+                Thread.sleep(1)
+            }
+        }
+    }
+
+    @Test(expected = BlockingOperationError::class)
+    fun testReusingThreadsFailure() = runTest {
+        val n = 100
+        repeat(n) {
+            async(Dispatchers.IO) {
+                Thread.sleep(1)
+            }
+        }
+        async(Dispatchers.Default) {
+            Thread.sleep(1)
+        }
+        repeat(n) {
+            async(Dispatchers.IO) {
+                Thread.sleep(1)
+            }
+        }
+    }
+
+}

kotlinx-coroutines-debug/test/CoroutinesDumpTest.kt

@@ -39,7 +39,7 @@ class CoroutinesDumpTest : DebugTestBase() {
 
     @Test
     fun testRunningCoroutine() = runBlocking {
-        val deferred = async(Dispatchers.Default) {
+        val deferred = async(Dispatchers.IO) {
             activeMethod(shouldSuspend = false)
             assertTrue(true)
         }
@@ -70,7 +70,7 @@ class CoroutinesDumpTest : DebugTestBase() {
 
     @Test
     fun testRunningCoroutineWithSuspensionPoint() = runBlocking {
-        val deferred = async(Dispatchers.Default) {
+        val deferred = async(Dispatchers.IO) {
             activeMethod(shouldSuspend = true)
             yield() // tail-call
         }
@@ -100,7 +100,7 @@ class CoroutinesDumpTest : DebugTestBase() {
 
     @Test
     fun testCreationStackTrace() = runBlocking {
-        val deferred = async(Dispatchers.Default) {
+        val deferred = async(Dispatchers.IO) {
             activeMethod(shouldSuspend = true)
         }
 
@@ -129,7 +129,7 @@ class CoroutinesDumpTest : DebugTestBase() {
 
     @Test
     fun testFinishedCoroutineRemoved() = runBlocking {
-        val deferred = async(Dispatchers.Default) {
+        val deferred = async(Dispatchers.IO) {
             activeMethod(shouldSuspend = true)
         }
 
@@ -149,7 +149,10 @@ class CoroutinesDumpTest : DebugTestBase() {
         if (shouldSuspend) yield()
         notifyCoroutineStarted()
         while (coroutineContext[Job]!!.isActive) {
-            runCatching { Thread.sleep(60_000) }
+            try {
+                Thread.sleep(60_000)
+            } catch (_ : InterruptedException) {
+            }
         }
     }
 

kotlinx-coroutines-debug/test/RunningThreadStackMergeTest.kt

@@ -133,7 +133,7 @@ class RunningThreadStackMergeTest : DebugTestBase() {
     }
 
     private fun CoroutineScope.launchEscapingCoroutineWithoutContext() {
-        launch(Dispatchers.Default) {
+        launch(Dispatchers.IO) {
             suspendingFunctionWithoutContext()
             assertTrue(true)
         }