Additional checks to ensure that there are no leaking threads after tests

Author: elizarov

integration/kotlinx-coroutines-guava/src/test/kotlin/kotlinx/coroutines/experimental/guava/ListenableFutureTest.kt

@@ -18,16 +18,23 @@ package kotlinx.coroutines.experimental.guava
 
 import com.google.common.util.concurrent.MoreExecutors
 import com.google.common.util.concurrent.SettableFuture
+import guide.test.ignoreLostThreads
 import kotlinx.coroutines.experimental.*
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsInstanceOf
 import org.junit.Assert.*
+import org.junit.Before
 import org.junit.Test
 import java.util.concurrent.Callable
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.ForkJoinPool
 
 class ListenableFutureTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("ForkJoinPool.commonPool-worker-")
+    }
+
     @Test
     fun testSimpleAwait() {
         val service = MoreExecutors.listeningDecorator(ForkJoinPool.commonPool())

integration/kotlinx-coroutines-jdk8/src/test/kotlin/kotlinx/coroutines/experimental/future/FutureTest.kt

@@ -16,9 +16,11 @@
 
 package kotlinx.coroutines.experimental.future
 
+import guide.test.ignoreLostThreads
 import kotlinx.coroutines.experimental.*
 import org.hamcrest.core.IsEqual
 import org.junit.Assert.*
+import org.junit.Before
 import org.junit.Test
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CompletionStage
@@ -27,6 +29,11 @@ import java.util.concurrent.atomic.AtomicInteger
 import kotlin.coroutines.experimental.CoroutineContext
 
 class FutureTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("ForkJoinPool.commonPool-worker-")
+    }
+
     @Test
     fun testSimpleAwait() {
         val future = future {

kotlinx-coroutines-core/src/test/kotlin/guide/test/TestUtil.kt

@@ -27,6 +27,35 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.locks.LockSupport
 
+private val ignoreLostThreads = mutableSetOf<String>()
+
+fun ignoreLostThreads(vararg s: String) { ignoreLostThreads += s }
+
+fun threadNames(): Set<String> {
+    val arrayOfThreads = Array<Thread?>(Thread.activeCount()) { null }
+    val n = Thread.enumerate(arrayOfThreads)
+    val names = hashSetOf<String>()
+    for (i in 0 until n)
+        names.add(arrayOfThreads[i]!!.name)
+    return names
+}
+
+fun checkTestThreads(threadNamesBefore: Set<String>) {
+    // give threads some time to shutdown
+    val waitTill = System.currentTimeMillis() + 1000L
+    var diff: List<String>
+    do {
+        val threadNamesAfter = threadNames()
+        diff = (threadNamesAfter - threadNamesBefore).filter { name ->
+            ignoreLostThreads.none { prefix -> name.startsWith(prefix) }
+        }
+        if (diff.isEmpty()) break
+    } while (System.currentTimeMillis() <= waitTill)
+    ignoreLostThreads.clear()
+    diff.forEach { println("Lost thread '$it'") }
+    check(diff.isEmpty()) { "Lost ${diff.size} threads" }
+}
+
 fun trackTask(block: Runnable) = timeSource.trackTask(block)
 
 // helper function to dump exception to stdout for ease of debugging failed tests
@@ -53,6 +82,7 @@ fun test(name: String, block: () -> Unit): List<String> = outputException(name)
     resetCoroutineId()
     // shutdown execution with old time source (in case it was working)
     DefaultExecutor.shutdown(SHUTDOWN_TIMEOUT)
+    val threadNamesBefore = threadNames()
     val testTimeSource = TestTimeSource(oldOut)
     timeSource = testTimeSource
     DefaultExecutor.ensureStarted() // should start with new time source
@@ -77,6 +107,7 @@ fun test(name: String, block: () -> Unit): List<String> = outputException(name)
         oldOut.println("--- done")
         System.setOut(oldOut)
         System.setErr(oldErr)
+        checkTestThreads(threadNamesBefore)
     }
     return ByteArrayInputStream(bytes).bufferedReader().readLines()
 }

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/ExecutorsTest.kt

@@ -16,42 +16,10 @@
 
 package kotlinx.coroutines.experimental
 
-import org.junit.After
-import org.junit.Before
 import org.junit.Test
 import java.util.concurrent.Executors
 
-class ExecutorsTest {
-    fun threadNames(): Set<String> {
-        val arrayOfThreads = Array<Thread?>(Thread.activeCount()) { null }
-        val n = Thread.enumerate(arrayOfThreads)
-        val names = hashSetOf<String>()
-        for (i in 0 until n)
-            names.add(arrayOfThreads[i]!!.name)
-        return names
-    }
-
-    lateinit var threadNamesBefore: Set<String>
-
-    @Before
-    fun before() {
-        threadNamesBefore = threadNames()
-    }
-
-    @After
-    fun after() {
-        // give threads some time to shutdown
-        val waitTill = System.currentTimeMillis() + 1000L
-        var diff: Set<String>
-        do {
-            val threadNamesAfter = threadNames()
-            diff = threadNamesAfter - threadNamesBefore
-            if (diff.isEmpty()) break
-        } while (System.currentTimeMillis() <= waitTill)
-        diff.forEach { println("Lost thread '$it'") }
-        check(diff.isEmpty()) { "Lost ${diff.size} threads"}
-    }
-
+class ExecutorsTest : TestBase() {
     fun checkThreadName(prefix: String) {
         val name = Thread.currentThread().name
         check(name.startsWith(prefix)) { "Expected thread name to start with '$prefix', found: '$name'" }

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/TestBase.kt

@@ -16,7 +16,10 @@
 
 package kotlinx.coroutines.experimental
 
+import guide.test.checkTestThreads
+import guide.test.threadNames
 import org.junit.After
+import org.junit.Before
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicReference
@@ -96,9 +99,21 @@ open class TestBase {
         check(!finished.getAndSet(true)) { "Should call 'finish(...)' at most once" }
     }
 
+    private lateinit var threadNamesBefore: Set<String>
+    private val SHUTDOWN_TIMEOUT = 5000L // 5 sec at most to wait
+
+    @Before
+    fun before() {
+        CommonPool.usePrivatePool()
+        threadNamesBefore = threadNames()
+    }
+
     @After
     fun onCompletion() {
         error.get()?.let { throw it }
         check(actionIndex.get() == 0 || finished.get()) { "Expecting that 'finish(...)' was invoked, but it was not" }
+        CommonPool.shutdown(SHUTDOWN_TIMEOUT)
+        DefaultExecutor.shutdown(SHUTDOWN_TIMEOUT)
+        checkTestThreads(threadNamesBefore)
     }
 }

reactive/kotlinx-coroutines-reactor/src/test/kotlin/kotlinx/coroutines/experimental/reactor/MonoTest.kt

@@ -16,25 +16,32 @@
 
 package kotlinx.coroutines.experimental.reactor
 
+import guide.test.ignoreLostThreads
 import kotlinx.coroutines.experimental.CommonPool
-import kotlinx.coroutines.experimental.runBlocking
-import org.junit.Assert.assertEquals
-import org.junit.Test
 import kotlinx.coroutines.experimental.TestBase
 import kotlinx.coroutines.experimental.reactive.awaitFirst
 import kotlinx.coroutines.experimental.reactive.awaitLast
 import kotlinx.coroutines.experimental.reactive.awaitSingle
+import kotlinx.coroutines.experimental.runBlocking
 import kotlinx.coroutines.experimental.yield
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsInstanceOf
 import org.junit.Assert
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Test
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 
 /**
  * Tests emitting single item with [mono].
  */
 class MonoTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("timer-")
+    }
+
     @Test
     fun testBasicSuccess() = runBlocking {
         expect(1)

reactive/kotlinx-coroutines-reactor/src/test/kotlin/kotlinx/coroutines/experimental/reactor/SchedulerTest.kt

@@ -16,17 +16,24 @@
 
 package kotlinx.coroutines.experimental.reactor
 
+import guide.test.ignoreLostThreads
 import kotlinx.coroutines.experimental.TestBase
 import kotlinx.coroutines.experimental.delay
 import kotlinx.coroutines.experimental.run
 import kotlinx.coroutines.experimental.runBlocking
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsNot
 import org.junit.Assert.assertThat
+import org.junit.Before
 import org.junit.Test
 import reactor.core.scheduler.Schedulers
 
 class SchedulerTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("single-")
+    }
+
     @Test
     fun testSingleScheduler(): Unit = runBlocking {
         expect(1)

reactive/kotlinx-coroutines-rx1/src/test/kotlin/kotlinx/coroutines/experimental/rx1/SchedulerTest.kt

@@ -16,14 +16,24 @@
 
 package kotlinx.coroutines.experimental.rx1
 
-import kotlinx.coroutines.experimental.*
+import guide.test.ignoreLostThreads
+import kotlinx.coroutines.experimental.TestBase
+import kotlinx.coroutines.experimental.delay
+import kotlinx.coroutines.experimental.run
+import kotlinx.coroutines.experimental.runBlocking
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsNot
 import org.junit.Assert.assertThat
+import org.junit.Before
 import org.junit.Test
 import rx.schedulers.Schedulers
 
 class SchedulerTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("RxIoScheduler-")
+    }
+
     @Test
     fun testIoScheduler(): Unit = runBlocking {
         expect(1)

reactive/kotlinx-coroutines-rx1/src/test/kotlin/kotlinx/coroutines/experimental/rx1/SingleTest.kt

@@ -16,11 +16,14 @@
 
 package kotlinx.coroutines.experimental.rx1
 
+import guide.test.ignoreLostThreads
 import kotlinx.coroutines.experimental.*
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsInstanceOf
 import org.hamcrest.core.IsNull
-import org.junit.Assert.*
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertThat
+import org.junit.Before
 import org.junit.Test
 import rx.Observable
 import rx.Single
@@ -30,6 +33,11 @@ import java.util.concurrent.TimeUnit
  * Tests emitting single item with [rxSingle].
  */
 class SingleTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("RxComputationScheduler-", "RxIoScheduler-")
+    }
+
     @Test
     fun testBasicSuccess() = runBlocking<Unit> {
         expect(1)

reactive/kotlinx-coroutines-rx2/src/test/kotlin/guide/test/ReactiveTestBase.kt

@@ -28,6 +28,11 @@ open class ReactiveTestBase {
     fun setup() {
         RxJavaPlugins.setIoSchedulerHandler(Handler)
         RxJavaPlugins.setComputationSchedulerHandler(Handler)
+        ignoreLostThreads(
+            "RxComputationThreadPool-",
+            "RxCachedThreadScheduler-",
+            "RxCachedWorkerPoolEvictor-",
+            "RxSchedulerPurge-")
     }
 
     @After