Fixed the logic to detect broken ForkJoinPool instance and added a corresponding test

Fixes #432, #288

Author: elizarov

core/kotlinx-coroutines-core/src/CommonPool.kt

@@ -58,7 +58,7 @@ object CommonPool : CoroutineDispatcher() {
         // Reflection on ForkJoinPool class so that it works on JDK 6 (which is absent there)
         val fjpClass = Try { Class.forName("java.util.concurrent.ForkJoinPool") }
             ?: return createPlainPool() // Fallback to plain thread pool
-        // Try to use commonPool unless parallelism was explicitly specified or int debug privatePool mode
+        // Try to use commonPool unless parallelism was explicitly specified or in debug privatePool mode
         if (!usePrivatePool && requestedParallelism < 0) {
             Try { fjpClass.getMethod("commonPool")?.invoke(null) as? ExecutorService }
                 ?.takeIf { isGoodCommonPool(fjpClass, it) }
@@ -74,8 +74,13 @@ object CommonPool : CoroutineDispatcher() {
     /**
      * Checks that this ForkJoinPool's parallelism is at least one to avoid pathological bugs.
      */
-    private fun isGoodCommonPool(fjpClass: Class<*>, executor: ExecutorService): Boolean {
-        val actual = Try { fjpClass.getMethod("getParallelism").invoke(executor) as? Int }
+    internal fun isGoodCommonPool(fjpClass: Class<*>, executor: ExecutorService): Boolean {
+        // We cannot use getParallelism, since it lies to us (always returns at least 1)
+        // So we submit a task and check that getPoolSize is at least one after that
+        // A broken FJP (that is configured for 0 parallelism) would not execute the task and
+        // would report its pool size as zero.
+        executor.submit {}
+        val actual = Try { fjpClass.getMethod("getPoolSize").invoke(executor) as? Int }
             ?: return false
         return actual >= 1
     }

core/kotlinx-coroutines-core/test/CommonPoolTest.kt

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+import kotlinx.coroutines.experimental.*
+import org.junit.*
+import org.junit.Assert.*
+import java.lang.reflect.*
+import java.util.concurrent.*
+
+class CommonPoolTest {
+    private inline fun <T> Try(block: () -> T) = try { block() } catch (e: Throwable) { null }
+
+    @Test
+    fun testIsGoodCommonPool() {
+        // Test only on JDKs that has all we need
+        val fjpClass = Try { Class.forName("java.util.concurrent.ForkJoinPool") } ?: return
+        val wtfClass = Try { Class.forName("java.util.concurrent.ForkJoinPool${'$'}ForkJoinWorkerThreadFactory") } ?: return
+        val dwtfClass = Try { Class.forName("java.util.concurrent.ForkJoinPool${'$'}DefaultForkJoinWorkerThreadFactory") } ?: return
+        // We need private constructor to create "broken" FJP instance
+        val fjpCtor = Try { fjpClass.getDeclaredConstructor(
+            Int::class.java,
+            wtfClass,
+            Thread.UncaughtExceptionHandler::class.java,
+            Int::class.java,
+            String::class.java
+        ) } ?: return
+        fjpCtor.isAccessible = true
+        val dwtfCtor = Try { dwtfClass.getDeclaredConstructor() } ?: return
+        dwtfCtor.isAccessible = true
+        // Create bad pool
+        val fjp0: ExecutorService = createFJP(0, fjpCtor, dwtfCtor) ?: return
+        assertFalse(CommonPool.isGoodCommonPool(fjpClass, fjp0))
+        fjp0.shutdown()
+        // Create good pool
+        val fjp1: ExecutorService = createFJP(1, fjpCtor, dwtfCtor) ?: return
+        assertTrue(CommonPool.isGoodCommonPool(fjpClass, fjp1))
+        fjp1.shutdown()
+        println("CommonPool.isGoodCommonPool test passed")
+    }
+
+    fun createFJP(
+        parallelism: Int,
+        fjpCtor: Constructor<out Any>,
+        dwtfCtor: Constructor<out Any>
+    ): ExecutorService? = Try {
+        fjpCtor.newInstance(
+            parallelism,
+            dwtfCtor.newInstance(),
+            Thread.getDefaultUncaughtExceptionHandler(),
+            0,
+            "Worker"
+        )
+    } as? ExecutorService
+}