Check that worker threads belongs to the same scheduler in dispatchers.shutdown

Fixes #990

Author: qwwdfsad

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

@@ -299,7 +299,7 @@ internal class CoroutineScheduler(
         // atomically set termination flag which is checked when workers are added or removed
         if (!_isTerminated.compareAndSet(0, 1)) return
         // make sure we are not waiting for the current thread
-        val currentWorker = Thread.currentThread() as? Worker
+        val currentWorker = currentWorker()
         // Capture # of created workers that cannot change anymore (mind the synchronized block!)
         val created = synchronized(workers) { createdWorkers }
         // Shutdown all workers with the only exception of the current thread
@@ -481,9 +481,7 @@ internal class CoroutineScheduler(
      * Returns [ADDED], or [NOT_ADDED], or [ADDED_REQUIRES_HELP].
      */
     private fun submitToLocalQueue(task: Task, fair: Boolean): Int {
-        val worker = Thread.currentThread() as? Worker
-            ?: return NOT_ADDED
-        if (worker.scheduler !== this) return NOT_ADDED // different scheduler's worker (!!!)
+        val worker = currentWorker() ?: return NOT_ADDED
 
         /*
          * This worker could have been already terminated from this thread by close/shutdown and it should not
@@ -533,6 +531,11 @@ internal class CoroutineScheduler(
         return ADDED_REQUIRES_HELP
     }
 
+    private fun currentWorker(): Worker? {
+        val worker = Thread.currentThread() as? Worker ?: return null
+        return if (worker.scheduler == this) worker else null
+    }
+
     /**
      * Returns a string identifying the state of this scheduler for nicer debugging.
      * Note that this method is not atomic and represents rough state of pool.

kotlinx-coroutines-core/jvm/test/scheduling/SharingWorkerClassTest.kt

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.scheduling
+
+import kotlinx.coroutines.*
+import org.junit.Test
+import kotlin.test.*
+
+class SharingWorkerClassTest : SchedulerTestBase() {
+    private val threadLocal = ThreadLocal<Int?>()
+
+    @Test
+    fun testSharedThread() = runTest {
+        val dispatcher = ExperimentalCoroutineDispatcher(1, schedulerName = "first")
+        val dispatcher2 = ExperimentalCoroutineDispatcher(1, schedulerName = "second")
+
+        try {
+            withContext(dispatcher) {
+                assertNull(threadLocal.get())
+                threadLocal.set(239)
+                withContext(dispatcher2) {
+                    assertNull(threadLocal.get())
+                    threadLocal.set(42)
+                }
+
+                assertEquals(239, threadLocal.get())
+            }
+        } finally {
+            dispatcher.close()
+            dispatcher2.close()
+        }
+    }
+
+    @Test(timeout = 5000L)
+    fun testProgress() = runTest {
+        // See #990
+        val cores = Runtime.getRuntime().availableProcessors()
+        repeat(cores + 1) {
+            CoroutineScope(Dispatchers.Default).launch {
+                ExperimentalCoroutineDispatcher(1).close()
+            }.join()
+        }
+    }
+}