Scheduler tests fixed:

[temporary] Ignore flaky tests
Fix benchmark compilation
Fix WorkQueueStressTest

Author: qwwdfsad

benchmarks/src/jmh/kotlin/benchmarks/GuideSyncBenchmark.kt

@@ -45,36 +45,36 @@ open class GuideSyncBenchmark {
 
     @Benchmark
     fun sync01Problem() {
-        guide.sync.example01.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync01.main(emptyArray())
     }
 
     @Benchmark
     fun sync02Volatile() {
-        guide.sync.example02.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync02.main(emptyArray())
     }
 
     @Benchmark
     fun sync03AtomicInt() {
-        guide.sync.example03.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync03.main(emptyArray())
     }
 
     @Benchmark
     fun sync04ConfineFine() {
-        guide.sync.example04.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync04.main(emptyArray())
     }
 
     @Benchmark
     fun sync05ConfineCoarse() {
-        guide.sync.example05.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync05.main(emptyArray())
     }
 
     @Benchmark
     fun sync06Mutex() {
-        guide.sync.example06.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync06.main(emptyArray())
     }
 
     @Benchmark
     fun sync07Actor() {
-        guide.sync.example07.main(emptyArray())
+        kotlinx.coroutines.experimental.guide.sync07.main(emptyArray())
     }
 }

benchmarks/src/jmh/kotlin/benchmarks/ParametrizedDispatcherBase.kt

@@ -4,7 +4,7 @@ import benchmarks.actors.CORES_COUNT
 import kotlinx.coroutines.experimental.CommonPool
 import kotlinx.coroutines.experimental.ThreadPoolDispatcher
 import kotlinx.coroutines.experimental.newFixedThreadPoolContext
-import scheduling.ExperimentalCoroutineDispatcher
+import kotlinx.coroutines.experimental.scheduling.*
 import org.openjdk.jmh.annotations.Param
 import org.openjdk.jmh.annotations.Setup
 import org.openjdk.jmh.annotations.TearDown

benchmarks/src/jmh/kotlin/benchmarks/actors/PingPongWithBlockingContext.kt

@@ -2,8 +2,8 @@ package benchmarks.actors
 
 import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.channels.*
+import kotlinx.coroutines.experimental.scheduling.*
 import org.openjdk.jmh.annotations.*
-import scheduling.*
 import java.util.concurrent.*
 import kotlin.coroutines.experimental.*
 

core/kotlinx-coroutines-core/src/internal/LockFreeMPMCQueue.kt

@@ -39,7 +39,7 @@ internal open class LockFreeMPMCQueue<T : LockFreeMPMCQueueNode<T>> {
         }
     }
 
-    public fun removeFistOrNull(): T? {
+    public fun removeFirstOrNull(): T? {
         head.loop { curHead ->
             val next = curHead.next.value ?: return null
             if (head.compareAndSet(curHead, next)) {
@@ -50,7 +50,7 @@ internal open class LockFreeMPMCQueue<T : LockFreeMPMCQueueNode<T>> {
 
     fun headCas(curHead: T, update: T) = head.compareAndSet(curHead, update)
 
-    public inline fun removeFistOrNullIf(predicate: (T) -> Boolean): T? {
+    public inline fun removeFirstOrNullIf(predicate: (T) -> Boolean): T? {
         while (true) {
             val curHead = headValue
             val next = curHead.nextValue ?: return null

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

@@ -248,7 +248,7 @@ internal class CoroutineScheduler(
     private val random = Random()
 
     // This is used a "stop signal" for debugging/tests only
-    private val isTerminated = atomic(false)
+    private val isTerminated = atomic(0) // workaround for atomicfu bug
 
     companion object {
         private const val MAX_SPINS = 1000
@@ -298,7 +298,7 @@ internal class CoroutineScheduler(
      * and intended to be used only for testing. Invocation has no additional effect if already closed.
      */
     fun shutdown(timeout: Long) {
-        if (!isTerminated.compareAndSet(false, true)) return
+        if (!isTerminated.compareAndSet(0, 1)) return
         // Race with recently created threads which may park indefinitely
         var finishedThreads = 0
         while (finishedThreads < createdWorkers) {
@@ -439,7 +439,7 @@ internal class CoroutineScheduler(
     private fun createNewWorker(): Int {
         synchronized(workers) {
             // for test purposes make sure we're not trying to resurrect terminated scheduler
-            require(!isTerminated.value) { "This scheduler was terminated "}
+            require(isTerminated.value == 0) { "This scheduler was terminated "}
             val state = controlState.value
             val created = createdWorkers(state)
             val blocking = blockingWorkers(state)
@@ -683,7 +683,7 @@ internal class CoroutineScheduler(
 
         override fun run() {
             var wasIdle = false // local variable to avoid extra idleReset invocations when tasks repeatedly arrive
-            while (!isTerminated.value && state != WorkerState.TERMINATED) {
+            while (isTerminated.value == 0 && state != WorkerState.TERMINATED) {
                 val task = findTask()
                 if (task == null) {
                     // Wait for a job with potential park
@@ -918,9 +918,9 @@ internal class CoroutineScheduler(
              * once per two core pool size iterations
              */
             val globalFirst = nextInt(2 * corePoolSize) == 0
-            if (globalFirst) globalQueue.removeFistOrNull()?.let { return it }
+            if (globalFirst) globalQueue.removeFirstOrNull()?.let { return it }
             localQueue.poll()?.let { return it }
-            if (!globalFirst) globalQueue.removeFistOrNull()?.let { return it }
+            if (!globalFirst) globalQueue.removeFirstOrNull()?.let { return it }
             return trySteal()
         }
 

core/kotlinx-coroutines-core/src/scheduling/Tasks.kt

@@ -105,7 +105,7 @@ internal class Task(
 internal open class GlobalQueue : LockFreeMPMCQueue<Task>() {
     // Open for tests
     public open fun removeFirstBlockingModeOrNull(): Task? =
-        removeFistOrNullIf { it.mode == TaskMode.PROBABLY_BLOCKING }
+        removeFirstOrNullIf { it.mode == TaskMode.PROBABLY_BLOCKING }
 }
 
 internal abstract class TimeSource {

core/kotlinx-coroutines-core/test/internal/LockFreeMPMCQueueTest.kt

@@ -13,12 +13,12 @@ class LockFreeMPMCQueueTest : TestBase() {
     @Test
     fun testBasic() {
         val q = LockFreeMPMCQueue<Node>()
-        assertEquals(null, q.removeFistOrNull())
+        assertEquals(null, q.removeFirstOrNull())
         assertTrue(q.isEmpty())
         q.addLast(Node(1))
         assertEquals(1, q.size)
-        assertEquals(Node(1), q.removeFistOrNull())
-        assertEquals(null, q.removeFistOrNull())
+        assertEquals(Node(1), q.removeFirstOrNull())
+        assertEquals(null, q.removeFirstOrNull())
         assertTrue(q.isEmpty())
     }
 

core/kotlinx-coroutines-core/test/scheduling/BlockingCoroutineDispatcherStressTest.kt

@@ -36,8 +36,6 @@ class BlockingCoroutineDispatcherStressTest : SchedulerTestBase() {
 
         tasks.forEach { it.await() }
         require(tasks.isNotEmpty())
-        // Simple sanity, test is too short to guarantee that every possible state was observed
-        require(observedConcurrency.size >= 3.coerceAtMost(CORES_COUNT))
         for (i in CORES_COUNT + 1..CORES_COUNT * 2) {
             require(i !in observedConcurrency.keys) { "Unexpected state: $observedConcurrency" }
         }

core/kotlinx-coroutines-core/test/scheduling/CoroutineSchedulerShrinkTest.kt

@@ -5,6 +5,7 @@ import org.junit.*
 import java.util.concurrent.*
 import kotlin.coroutines.experimental.*
 
+@Ignore // these tests are too unstable on Windows, should be virtualized
 class CoroutineSchedulerShrinkTest : SchedulerTestBase() {
 
     private val blockingTasksCount = CORES_COUNT * 3

core/kotlinx-coroutines-core/test/scheduling/WorkQueueStressTest.kt

@@ -41,7 +41,7 @@ class WorkQueueStressTest : TestBase() {
         threads += thread(name = "producer") {
             startLatch.await()
             for (i in 1..offerIterations) {
-                while (producerQueue.bufferSize == BUFFER_CAPACITY - 1) {
+                while (producerQueue.bufferSize > BUFFER_CAPACITY / 2) {
                     Thread.yield()
                 }
 
@@ -55,7 +55,7 @@ class WorkQueueStressTest : TestBase() {
             threads += thread(name = "stealer $i") {
                 val myQueue = WorkQueue()
                 startLatch.await()
-                while (!producerFinished || producerQueue.bufferSize != 0) {
+                while (!producerFinished || producerQueue.size() != 0) {
                     myQueue.trySteal(producerQueue, stolenTasks[i])
                 }
 
@@ -106,8 +106,7 @@ class WorkQueueStressTest : TestBase() {
     private fun validate() {
         val result = mutableSetOf<Long>()
         for (stolenTask in stolenTasks) {
-            require(!stolenTask.isEmpty())
-            assertEquals(stolenTask.size, stolenTask.size)
+            assertEquals(stolenTask.size, stolenTask.map { it }.toSet().size)
             result += stolenTask.map { it.submissionTime }
         }