Actually fixed startup/shutdown of DefaultExecutor for tests to prevent

thread leak. Includes test-case.

Author: elizarov

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/DefaultExecutor.kt

@@ -43,38 +43,39 @@ internal object DefaultExecutor : EventLoopBase(), Runnable {
     @Volatile
     private var debugStatus: Int = FRESH
 
+    private val isShutdownRequested: Boolean get() {
+        val debugStatus = debugStatus
+        return debugStatus == SHUTDOWN_REQ || debugStatus == SHUTDOWN_ACK
+    }
+
     override fun run() {
         timeSource.registerTimeLoopThread()
         try {
             var shutdownNanos = Long.MAX_VALUE
-            if (notifyStartup()) {
-                runLoop@ while (true) {
-                    Thread.interrupted() // just reset interruption flag
-                    var parkNanos = processNextEvent()
-                    if (parkNanos == Long.MAX_VALUE) {
-                        // nothing to do, initialize shutdown timeout
-                        if (shutdownNanos == Long.MAX_VALUE) {
-                            val now = timeSource.nanoTime()
-                            if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + KEEP_ALIVE_NANOS
-                            val tillShutdown = shutdownNanos - now
-                            if (tillShutdown <= 0) break@runLoop // shut thread down
-                            parkNanos = parkNanos.coerceAtMost(tillShutdown)
-                        } else
-                            parkNanos = parkNanos.coerceAtMost(KEEP_ALIVE_NANOS) // limit wait time anyway
-                    }
-                    if (parkNanos > 0) {
-                        // check if shutdown was requested and bail out in this case
-                        if (debugStatus == SHUTDOWN_REQ) {
-                            acknowledgeShutdown()
-                            break@runLoop
-                        } else {
-                            timeSource.parkNanos(this, parkNanos)
-                        }
-                    }
+            if (!notifyStartup()) return
+            while (true) {
+                Thread.interrupted() // just reset interruption flag
+                var parkNanos = processNextEvent()
+                if (parkNanos == Long.MAX_VALUE) {
+                    // nothing to do, initialize shutdown timeout
+                    if (shutdownNanos == Long.MAX_VALUE) {
+                        val now = timeSource.nanoTime()
+                        if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + KEEP_ALIVE_NANOS
+                        val tillShutdown = shutdownNanos - now
+                        if (tillShutdown <= 0) return // shut thread down
+                        parkNanos = parkNanos.coerceAtMost(tillShutdown)
+                    } else
+                        parkNanos = parkNanos.coerceAtMost(KEEP_ALIVE_NANOS) // limit wait time anyway
+                }
+                if (parkNanos > 0) {
+                    // check if shutdown was requested and bail out in this case
+                    if (isShutdownRequested) return
+                    timeSource.parkNanos(this, parkNanos)
                 }
             }
         } finally {
             _thread = null // this thread is dead
+            acknowledgeShutdownIfNeeded()
             timeSource.unregisterTimeLoopThread()
             // recheck if queues are empty after _thread reference was set to null (!!!)
             if (!isEmpty) thread() // recreate thread if it is needed
@@ -110,32 +111,31 @@ internal object DefaultExecutor : EventLoopBase(), Runnable {
 
     @Synchronized
     private fun notifyStartup(): Boolean {
-        if (debugStatus == SHUTDOWN_REQ) return false
+        if (isShutdownRequested) return false
         debugStatus = ACTIVE
         (this as Object).notifyAll()
         return true
     }
 
     // used for tests
     @Synchronized
-    internal fun shutdown(timeout: Long) {
-        if (_thread != null) {
-            val deadline = System.currentTimeMillis() + timeout
-            if (debugStatus == ACTIVE || debugStatus == FRESH) debugStatus = SHUTDOWN_REQ
-            unpark()
-            // loop while there is anything to do immediately or deadline passes
-            while (debugStatus != SHUTDOWN_ACK && _thread != null) {
-                val remaining = deadline - System.currentTimeMillis()
-                if (remaining <= 0) break
-                (this as Object).wait(timeout)
-            }
+    fun shutdown(timeout: Long) {
+        val deadline = System.currentTimeMillis() + timeout
+        if (!isShutdownRequested) debugStatus = SHUTDOWN_REQ
+        // loop while there is anything to do immediately or deadline passes
+        while (debugStatus != SHUTDOWN_ACK && _thread != null) {
+            _thread?.let { timeSource.unpark(it) } // wake up thread if present
+            val remaining = deadline - System.currentTimeMillis()
+            if (remaining <= 0) break
+            (this as Object).wait(timeout)
         }
         // restore fresh status
         debugStatus = FRESH
     }
 
     @Synchronized
-    private fun acknowledgeShutdown() {
+    private fun acknowledgeShutdownIfNeeded() {
+        if (!isShutdownRequested) return
         debugStatus = SHUTDOWN_ACK
         resetAll() // clear queues
         (this as Object).notifyAll()

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt

@@ -288,13 +288,13 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
         }
 
         final override fun dispose() = synchronized(this) {
-            when (state) {
-                DELAYED -> _delayed.value?.remove(this)
-                RESCHEDULED -> DefaultExecutor.removeDelayedImpl(this)
-                else -> return
+                when (state) {
+                    DELAYED -> _delayed.value?.remove(this)
+                    RESCHEDULED -> DefaultExecutor.removeDelayedImpl(this)
+                    else -> return
+                }
+                state = REMOVED
             }
-            state = REMOVED
-        }
 
         override fun toString(): String = "Delayed[nanos=$nanoTime]"
     }

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import org.junit.*
+
+class TestBaseTest : TestBase() {
+    @Test
+    fun testThreadsShutdown() {
+        val SHUTDOWN_TIMEOUT = 1_000L
+        repeat(1000 * stressTestMultiplier) { i ->
+            CommonPool.usePrivatePool()
+            val threadsBefore = currentThreads()
+            runBlocking {
+                val sub = launch(DefaultDispatcher) {
+                    delay(10000000L)
+                }
+                sub.cancel()
+                sub.join()
+            }
+            CommonPool.shutdown(SHUTDOWN_TIMEOUT)
+            DefaultExecutor.shutdown(SHUTDOWN_TIMEOUT)
+            checkTestThreads(threadsBefore)
+        }
+
+    }
+}