BlockHound integration POC

Author: dkhalanskyjb

build.gradle

@@ -60,6 +60,7 @@ buildscript {
     repositories {
         jcenter()
         maven { url "https://kotlin.bintray.com/kotlinx" }
+        maven { url "https://repo.spring.io/snapshot/" }
         maven {
             url "https://kotlin.bintray.com/kotlin-dev"
             credentials {
@@ -153,6 +154,7 @@ allprojects {
         jcenter()
         maven {
             url "https://kotlin.bintray.com/kotlin-dev"
+            url "https://repo.spring.io/snapshot/"
             credentials {
                 username = project.hasProperty('bintrayUser') ? project.property('bintrayUser') : System.getenv('BINTRAY_USER') ?: ""
                 password = project.hasProperty('bintrayApiKey') ? project.property('bintrayApiKey') : System.getenv('BINTRAY_API_KEY') ?: ""

kotlinx-coroutines-debug/build.gradle

@@ -22,6 +22,8 @@ dependencies {
     compileOnly "junit:junit:$junit_version"
     shadowDeps "net.bytebuddy:byte-buddy:$byte_buddy_version"
     shadowDeps "net.bytebuddy:byte-buddy-agent:$byte_buddy_version"
+    compile 'io.projectreactor.tools:blockhound:1.0.1.BUILD-SNAPSHOT'
+    compile "org.jetbrains.kotlin:kotlin-reflect:$kotlin_version"
 }
 
 jar {

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

@@ -0,0 +1,43 @@
+package kotlinx.coroutines.debug.internal
+
+import reactor.blockhound.BlockHound
+import kotlin.reflect.KClass
+import kotlin.reflect.full.*
+
+internal object BlockHoundIntegration {
+
+    init {
+        val cls = Class.forName("kotlinx.coroutines.scheduling.CoroutineScheduler\$Worker").kotlin
+        initializerHelper(cls)
+    }
+
+    private fun <T : Any> initializerHelper(cls: KClass<T>) {
+        val field = cls.declaredMemberProperties.find { it.name == "state" }!!
+        BlockHound.builder()
+                .addDynamicThreadPredicate(cls::isInstance)
+                .nonBlockingThreadPredicate { p ->
+                    p.or { thread ->
+                        val castThread = cls.safeCast(thread)
+                        if (!enabled || castThread == null) {
+                            false
+                        } else {
+                            val state = field(castThread) as Enum<*>
+                            state.name == "CPU_ACQUIRED"
+                        }
+                    }
+                }
+                .install()
+    }
+
+    @Volatile
+    private var enabled = false
+
+    fun install() {
+        enabled = true
+    }
+
+    fun uninstall() {
+        enabled = false
+    }
+
+}

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

@@ -66,6 +66,8 @@ internal object DebugProbesImpl {
             .name(cl.name)
             .make()
             .load(cl.classLoader, ClassReloadingStrategy.fromInstalledAgent())
+
+        BlockHoundIntegration.install()
     }
 
     public fun uninstall(): Unit = coroutineStateLock.write {
@@ -82,6 +84,8 @@ internal object DebugProbesImpl {
             .name(cl.name)
             .make()
             .load(cl.classLoader, ClassReloadingStrategy.fromInstalledAgent())
+
+        BlockHoundIntegration.uninstall()
     }
 
     public fun hierarchyToString(job: Job): String = coroutineStateLock.write {

kotlinx-coroutines-debug/test/BlockHoundTest.kt

@@ -0,0 +1,79 @@
+package kotlinx.coroutines.debug
+import kotlinx.coroutines.*
+import kotlinx.coroutines.debug.internal.BlockHoundIntegration
+import org.junit.*
+import reactor.blockhound.BlockingOperationError
+
+class BlockHoundTest : TestBase() {
+
+    @Before
+    fun init() {
+        BlockHoundIntegration.install()
+    }
+
+    @After
+    fun deinit() {
+        BlockHoundIntegration.uninstall()
+    }
+
+    @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)
+            }
+        }
+    }
+
+}