Speedup linerizability tests, make conflation part of helpClose (fixes nightly stress failure)

Author: qwwdfsad

common/kotlinx-coroutines-core-common/src/channels/AbstractChannel.kt

@@ -257,13 +257,11 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
         if (!closeAdded) {
             val actualClosed = queue.prevNode as Closed<*>
             helpClose(actualClosed)
-            onClosedIdempotent(actualClosed)
             return false
         }
 
         helpClose(closed)
         invokeOnCloseHandler(cause)
-        onClosedIdempotent(closed)
         return true
     }
 
@@ -325,6 +323,8 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
             previous as Receive<E> // type assertion
             previous.resumeReceiveClosed(closed)
         }
+
+        onClosedIdempotent(closed)
     }
 
     /**

common/kotlinx-coroutines-core-common/src/channels/ConflatedChannel.kt

@@ -38,11 +38,17 @@ internal open class ConflatedChannel<E> : AbstractChannel<E>() {
                     val sendResult = sendConflated(element)
                     when (sendResult) {
                         null -> return OFFER_SUCCESS
-                        is Closed<*> -> return sendResult
+                        is Closed<*> -> {
+                            conflatePreviousSendBuffered(sendResult)
+                            return sendResult
+                        }
                     }
                     // otherwise there was receiver in queue, retry super.offerInternal
                 }
-                result is Closed<*> -> return result
+                result is Closed<*> -> {
+                    conflatePreviousSendBuffered(result)
+                    return result
+                }
                 else -> error("Invalid offerInternal result $result")
             }
         }
@@ -58,7 +64,10 @@ internal open class ConflatedChannel<E> : AbstractChannel<E>() {
                 result === ALREADY_SELECTED -> return ALREADY_SELECTED
                 result === OFFER_SUCCESS -> return OFFER_SUCCESS
                 result === OFFER_FAILED -> {} // retry
-                result is Closed<*> -> return result
+                result is Closed<*> -> {
+                    conflatePreviousSendBuffered(result)
+                    return result
+                }
                 else -> error("Invalid result $result")
             }
         }

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

@@ -35,7 +35,7 @@ private val VERBOSE = systemProp("test.verbose", false)
 @Suppress("DEPRECATION")
 public actual open class TestBase actual constructor() {
     /**
-     * Is `true` when nightly stress test is done.
+     * Is `true` when running in a nightly stress test mode.
      */
     public actual val isStressTest = System.getProperty("stressTest") != null
 

core/kotlinx-coroutines-core/test/linearizability/ChannelIsClosedLinearizabilityTest.kt

@@ -51,21 +51,4 @@ class ChannelIsClosedLinearizabilityTest : TestBase() {
 
         LinChecker.check(ChannelIsClosedLinearizabilityTest::class.java, options)
     }
-
-    override fun equals(other: Any?): Boolean {
-        if (this === other) return true
-        if (javaClass != other?.javaClass) return false
-
-        other as ChannelIsClosedLinearizabilityTest
-
-        if (channel != other.channel) return false
-
-        return true
-    }
-
-    override fun hashCode(): Int {
-        return channel.hashCode()
-    }
-
-
 }

core/kotlinx-coroutines-core/test/linearizability/ChannelLinearizabilityTest.kt

@@ -31,18 +31,12 @@ class ChannelLinearizabilityTest : TestBase() {
     @Operation(runOnce = true)
     fun send2(@Param(name = "value") value: Int) = lt.run("send2") { channel.send(value) }
 
-    @Operation(runOnce = true)
-    fun send3(@Param(name = "value") value: Int) = lt.run("send3") { channel.send(value) }
-
     @Operation(runOnce = true)
     fun receive1() = lt.run("receive1") { channel.receive() }
 
     @Operation(runOnce = true)
     fun receive2() = lt.run("receive2") { channel.receive() }
 
-    @Operation(runOnce = true)
-    fun receive3() = lt.run("receive3") { channel.receive() }
-
     @Operation(runOnce = true)
     fun close1() = lt.run("close1") { channel.close(IOException("close1")) }
 
@@ -70,8 +64,8 @@ class ChannelLinearizabilityTest : TestBase() {
     private fun runTest(capacity: Int) {
         ChannelLinearizabilityTest.capacity = capacity
         val options = StressOptions()
-            .iterations(100 * stressTestMultiplierSqrt)
-            .invocationsPerIteration(1000 * stressTestMultiplierSqrt)
+            .iterations(50 * stressTestMultiplierSqrt)
+            .invocationsPerIteration(500 * stressTestMultiplierSqrt)
             .threads(3)
             .verifier(LinVerifier::class.java)
         LinChecker.check(ChannelLinearizabilityTest::class.java, options)