Introduce SegmentQueueSynchronizer abstraction for synchronization primitives and ReadWriteMutex

Author: ndkoval

Diff for: benchmarks/src/jmh/kotlin/benchmarks/SemaphoreBenchmark.kt

@@ -48,7 +48,7 @@ open class SemaphoreBenchmark {
         val semaphore = Semaphore(_3_maxPermits)
         val jobs = ArrayList<Job>(coroutines)
         repeat(coroutines) {
-            jobs += GlobalScope.launch {
+            jobs += GlobalScope.launch(dispatcher) {
                 repeat(n) {
                     semaphore.withPermit {
                         doGeomDistrWork(WORK_INSIDE)
@@ -66,7 +66,7 @@ open class SemaphoreBenchmark {
         val semaphore = Channel<Unit>(_3_maxPermits)
         val jobs = ArrayList<Job>(coroutines)
         repeat(coroutines) {
-            jobs += GlobalScope.launch {
+            jobs += GlobalScope.launch(dispatcher) {
                 repeat(n) {
                     semaphore.send(Unit) // acquire
                     doGeomDistrWork(WORK_INSIDE)
@@ -87,4 +87,4 @@ enum class SemaphoreBenchDispatcherCreator(val create: (parallelism: Int) -> Cor
 
 private const val WORK_INSIDE = 50
 private const val WORK_OUTSIDE = 50
-private const val BATCH_SIZE = 100000
+private const val BATCH_SIZE = 1000000

Diff for: kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -1329,6 +1329,18 @@ public final class kotlinx/coroutines/sync/MutexKt {
 	public static synthetic fun withLock$default (Lkotlinx/coroutines/sync/Mutex;Ljava/lang/Object;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
 }
 
+public abstract interface class kotlinx/coroutines/sync/ReadWriteMutex {
+	public abstract fun getWrite ()Lkotlinx/coroutines/sync/Mutex;
+	public abstract fun readLock (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun readUnlock ()V
+}
+
+public final class kotlinx/coroutines/sync/ReadWriteMutexKt {
+	public static final fun ReadWriteMutex ()Lkotlinx/coroutines/sync/ReadWriteMutex;
+	public static final fun read (Lkotlinx/coroutines/sync/ReadWriteMutex;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun write (Lkotlinx/coroutines/sync/ReadWriteMutex;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
 public abstract interface class kotlinx/coroutines/sync/Semaphore {
 	public abstract fun acquire (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun getAvailablePermits ()I

Diff for: kotlinx-coroutines-core/build.gradle

@@ -266,8 +266,8 @@ task jvmStressTest(type: Test, dependsOn: compileTestKotlinJvm) {
     testLogging.showStandardStreams = true
     systemProperty 'kotlinx.coroutines.scheduler.keep.alive.sec', '100000' // any unpark problem hangs test
     // Adjust internal algorithmic parameters to increase the testing quality instead of performance.
-    systemProperty 'kotlinx.coroutines.semaphore.segmentSize', '1'
-    systemProperty 'kotlinx.coroutines.semaphore.maxSpinCycles', '10'
+    systemProperty 'kotlinx.coroutines.sqs.segmentSize', '1'
+    systemProperty 'kotlinx.coroutines.sqs.maxSpinCycles', '10'
     systemProperty 'kotlinx.coroutines.bufferedChannel.segmentSize', '2'
     systemProperty 'kotlinx.coroutines.bufferedChannel.expandBufferCompletionWaitIterations', '1'
 }
@@ -302,8 +302,8 @@ static void configureJvmForLincheck(task, additional = false) {
                     '--add-exports', 'java.base/jdk.internal.util=ALL-UNNAMED'] // in the model checking mode
     // Adjust internal algorithmic parameters to increase the testing quality instead of performance.
     var segmentSize = additional ? '2' : '1'
-    task.systemProperty 'kotlinx.coroutines.semaphore.segmentSize', segmentSize
-    task.systemProperty 'kotlinx.coroutines.semaphore.maxSpinCycles', '1' // better for the model checking mode
+    task.systemProperty 'kotlinx.coroutines.sqs.segmentSize', segmentSize
+    task.systemProperty 'kotlinx.coroutines.sqs.maxSpinCycles', '1' // better for the model checking mode
     task.systemProperty 'kotlinx.coroutines.bufferedChannel.segmentSize', segmentSize
     task.systemProperty 'kotlinx.coroutines.bufferedChannel.expandBufferCompletionWaitIterations', '1'
 }

Diff for: kotlinx-coroutines-core/common/src/Debug.common.kt

@@ -8,6 +8,7 @@ internal expect val DEBUG: Boolean
 internal expect val Any.hexAddress: String
 internal expect val Any.classSimpleName: String
 internal expect fun assert(value: () -> Boolean)
+internal inline fun assertNot(crossinline value: () -> Boolean) = assert { !value() }
 
 /**
  * Throwable which can be cloned during stacktrace recovery in a class-specific way.