Skip to content

Commit 7755edb

Browse files
qwwdfsaddkhalanskyjb
qwwdfsad
and
dkhalanskyjb
authored
Mutex.onLock deprecation (#2850)
* Actually test 'onLock' and the corresponding concurrency and cancellation of Reactive's onSend * Update benchmarks * Non-linearizable implementation of PublisherCoroutine.onSend that isn't using Mutex.onLock * Deprecate Mutex.onLock Fixes #2794 Co-authored-by: Dmitry Khalanskiy <[email protected]>
1 parent dfc4821 commit 7755edb

File tree

15 files changed

+166
-28
lines changed

15 files changed

+166
-28
lines changed

‎benchmarks/build.gradle.kts

+7-6
Original file line numberDiff line numberDiff line change
@@ -54,13 +54,14 @@ tasks.named<Jar>("jmhJar") {
5454
}
5555

5656
dependencies {
57-
compile("org.openjdk.jmh:jmh-core:1.26")
58-
compile("io.projectreactor:reactor-core:${version("reactor")}")
59-
compile("io.reactivex.rxjava2:rxjava:2.1.9")
60-
compile("com.github.akarnokd:rxjava2-extensions:0.20.8")
57+
implementation("org.openjdk.jmh:jmh-core:1.26")
58+
implementation("io.projectreactor:reactor-core:${version("reactor")}")
59+
implementation("io.reactivex.rxjava2:rxjava:2.1.9")
60+
implementation("com.github.akarnokd:rxjava2-extensions:0.20.8")
6161

62-
compile("com.typesafe.akka:akka-actor_2.12:2.5.0")
63-
compile(project(":kotlinx-coroutines-core"))
62+
implementation("com.typesafe.akka:akka-actor_2.12:2.5.0")
63+
implementation(project(":kotlinx-coroutines-core"))
64+
implementation(project(":kotlinx-coroutines-reactive"))
6465

6566
// add jmh dependency on main
6667
"jmhImplementation"(sourceSets.main.get().runtimeClasspath)

‎kotlinx-coroutines-core/README.md

-3
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ helper function. [NonCancellable] job object is provided to suppress cancellatio
5757
| [SendChannel][kotlinx.coroutines.channels.SendChannel] | [send][kotlinx.coroutines.channels.SendChannel.send] | [onSend][kotlinx.coroutines.channels.SendChannel.onSend] | [trySend][kotlinx.coroutines.channels.SendChannel.trySend]
5858
| [ReceiveChannel][kotlinx.coroutines.channels.ReceiveChannel] | [receive][kotlinx.coroutines.channels.ReceiveChannel.receive] | [onReceive][kotlinx.coroutines.channels.ReceiveChannel.onReceive] | [tryReceive][kotlinx.coroutines.channels.ReceiveChannel.tryReceive]
5959
| [ReceiveChannel][kotlinx.coroutines.channels.ReceiveChannel] | [receiveCatching][kotlinx.coroutines.channels.receiveCatching] | [onReceiveCatching][kotlinx.coroutines.channels.onReceiveCatching] | [tryReceive][kotlinx.coroutines.channels.ReceiveChannel.tryReceive]
60-
| [Mutex][kotlinx.coroutines.sync.Mutex] | [lock][kotlinx.coroutines.sync.Mutex.lock] | [onLock][kotlinx.coroutines.sync.Mutex.onLock] | [tryLock][kotlinx.coroutines.sync.Mutex.tryLock]
6160
| none | [delay][kotlinx.coroutines.delay] | [onTimeout][kotlinx.coroutines.selects.SelectBuilder.onTimeout] | none
6261

6362
# Package kotlinx.coroutines
@@ -121,8 +120,6 @@ Obsolete and deprecated module to test coroutines. Replaced with `kotlinx-corout
121120

122121
[kotlinx.coroutines.sync.Mutex]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/index.html
123122
[kotlinx.coroutines.sync.Mutex.lock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/lock.html
124-
[kotlinx.coroutines.sync.Mutex.onLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/on-lock.html
125-
[kotlinx.coroutines.sync.Mutex.tryLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/try-lock.html
126123

127124
<!--- INDEX kotlinx.coroutines.channels -->
128125

‎kotlinx-coroutines-core/common/README.md

-2
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,6 @@ helper function. [NonCancellable] job object is provided to suppress cancellatio
6060
| [SendChannel][kotlinx.coroutines.channels.SendChannel] | [send][kotlinx.coroutines.channels.SendChannel.send] | [onSend][kotlinx.coroutines.channels.SendChannel.onSend] | [trySend][kotlinx.coroutines.channels.SendChannel.trySend]
6161
| [ReceiveChannel][kotlinx.coroutines.channels.ReceiveChannel] | [receive][kotlinx.coroutines.channels.ReceiveChannel.receive] | [onReceive][kotlinx.coroutines.channels.ReceiveChannel.onReceive] | [tryReceive][kotlinx.coroutines.channels.ReceiveChannel.tryReceive]
6262
| [ReceiveChannel][kotlinx.coroutines.channels.ReceiveChannel] | [receiveCatching][kotlinx.coroutines.channels.ReceiveChannel.receiveCatching] | [onReceiveCatching][kotlinx.coroutines.channels.ReceiveChannel.onReceiveCatching] | [tryReceive][kotlinx.coroutines.channels.ReceiveChannel.tryReceive]
63-
| [Mutex][kotlinx.coroutines.sync.Mutex] | [lock][kotlinx.coroutines.sync.Mutex.lock] | [onLock][kotlinx.coroutines.sync.Mutex.onLock] | [tryLock][kotlinx.coroutines.sync.Mutex.tryLock]
6463
| none | [delay] | [onTimeout][kotlinx.coroutines.selects.SelectBuilder.onTimeout] | none
6564

6665
This module provides debugging facilities for coroutines (run JVM with `-ea` or `-Dkotlinx.coroutines.debug` options)
@@ -131,7 +130,6 @@ Low-level primitives for finer-grained control of coroutines.
131130

132131
[kotlinx.coroutines.sync.Mutex]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/index.html
133132
[kotlinx.coroutines.sync.Mutex.lock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/lock.html
134-
[kotlinx.coroutines.sync.Mutex.onLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/on-lock.html
135133
[kotlinx.coroutines.sync.Mutex.tryLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/try-lock.html
136134

137135
<!--- INDEX kotlinx.coroutines.channels -->

‎kotlinx-coroutines-core/common/src/channels/Channel.kt

-1
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,6 @@ public interface SendChannel<in E> {
6464
*/
6565
public val onSend: SelectClause2<E, SendChannel<E>>
6666

67-
6867
/**
6968
* Immediately adds the specified [element] to this channel, if this doesn't violate its capacity restrictions,
7069
* and returns the successful result. Otherwise, returns failed or closed result.

‎kotlinx-coroutines-core/common/src/selects/Select.kt

-1
Original file line numberDiff line numberDiff line change
@@ -186,7 +186,6 @@ public interface SelectInstance<in R> {
186186
* | [SendChannel] | [send][SendChannel.send] | [onSend][SendChannel.onSend]
187187
* | [ReceiveChannel] | [receive][ReceiveChannel.receive] | [onReceive][ReceiveChannel.onReceive]
188188
* | [ReceiveChannel] | [receiveCatching][ReceiveChannel.receiveCatching] | [onReceiveCatching][ReceiveChannel.onReceiveCatching]
189-
* | [Mutex] | [lock][Mutex.lock] | [onLock][Mutex.onLock]
190189
* | none | [delay] | [onTimeout][SelectBuilder.onTimeout]
191190
*
192191
* This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this

‎kotlinx-coroutines-core/common/src/sync/Mutex.kt

+4-5
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,7 @@ public interface Mutex {
5252
* Note that this function does not check for cancellation when it is not suspended.
5353
* Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
5454
*
55-
* This function can be used in [select] invocation with [onLock] clause.
56-
* Use [tryLock] to try acquire lock without waiting.
55+
* Use [tryLock] to try acquiring a lock without waiting.
5756
*
5857
* This function is fair; suspended callers are resumed in first-in-first-out order.
5958
*
@@ -63,10 +62,10 @@ public interface Mutex {
6362
public suspend fun lock(owner: Any? = null)
6463

6564
/**
66-
* Clause for [select] expression of [lock] suspending function that selects when the mutex is locked.
67-
* Additional parameter for the clause in the `owner` (see [lock]) and when the clause is selected
68-
* the reference to this mutex is passed into the corresponding block.
65+
* Deprecated for removal without built-in replacement.
6966
*/
67+
@Deprecated(level = DeprecationLevel.WARNING, message = "Mutex.onLock deprecated without replacement. " +
68+
"For additional details please refer to #2794")
7069
public val onLock: SelectClause2<Any?, Mutex>
7170

7271
/**

‎kotlinx-coroutines-core/jvm/test/channels/RandevouzChannelStressTest.kt renamed to ‎kotlinx-coroutines-core/jvm/test/channels/RendezvousChannelStressTest.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ package kotlinx.coroutines.channels
77
import kotlinx.coroutines.*
88
import org.junit.*
99

10-
class RandevouzChannelStressTest : TestBase() {
10+
class RendezvousChannelStressTest : TestBase() {
1111

1212
@Test
1313
fun testStress() = runTest {

‎reactive/kotlinx-coroutines-reactive/src/Publish.kt

+13-1
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package kotlinx.coroutines.reactive
66
import kotlinx.atomicfu.*
77
import kotlinx.coroutines.*
88
import kotlinx.coroutines.channels.*
9+
import kotlinx.coroutines.intrinsics.*
910
import kotlinx.coroutines.selects.*
1011
import kotlinx.coroutines.sync.*
1112
import org.reactivestreams.*
@@ -104,10 +105,21 @@ public class PublisherCoroutine<in T>(
104105
// registerSelectSend
105106
@Suppress("PARAMETER_NAME_CHANGED_ON_OVERRIDE")
106107
override fun <R> registerSelectClause2(select: SelectInstance<R>, element: T, block: suspend (SendChannel<T>) -> R) {
107-
mutex.onLock.registerSelectClause2(select, null) {
108+
val clause = suspend {
108109
doLockedNext(element)?.let { throw it }
109110
block(this)
110111
}
112+
113+
launch(start = CoroutineStart.UNDISPATCHED) {
114+
mutex.lock()
115+
// Already selected -- bail out
116+
if (!select.trySelect()) {
117+
mutex.unlock()
118+
return@launch
119+
}
120+
121+
clause.startCoroutineCancellable(select.completion)
122+
}
111123
}
112124

113125
/*

‎reactive/kotlinx-coroutines-reactive/test/PublishTest.kt

+37-1
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,13 @@
55
package kotlinx.coroutines.reactive
66

77
import kotlinx.coroutines.*
8+
import kotlinx.coroutines.CancellationException
89
import kotlinx.coroutines.channels.*
10+
import kotlinx.coroutines.flow.*
11+
import kotlinx.coroutines.sync.*
912
import org.junit.Test
1013
import org.reactivestreams.*
14+
import java.util.concurrent.*
1115
import kotlin.test.*
1216

1317
class PublishTest : TestBase() {
@@ -284,4 +288,36 @@ class PublishTest : TestBase() {
284288
}
285289
assertEquals("OK", publisher.awaitFirstOrNull())
286290
}
287-
}
291+
292+
@Test
293+
fun testOnSendCancelled() = runTest {
294+
val latch = CountDownLatch(1)
295+
val published = publish(Dispatchers.Default) {
296+
expect(2)
297+
// Collector is ready
298+
send(1)
299+
try {
300+
send(2)
301+
expectUnreached()
302+
} catch (e: CancellationException) {
303+
// publisher cancellation is async
304+
latch.countDown()
305+
throw e
306+
}
307+
}
308+
309+
expect(1)
310+
val collectorLatch = Mutex(true)
311+
val job = launch {
312+
published.asFlow().buffer(0).collect {
313+
collectorLatch.unlock()
314+
hang { expect(4) }
315+
}
316+
}
317+
collectorLatch.lock()
318+
expect(3)
319+
job.cancelAndJoin()
320+
latch.await()
321+
finish(5)
322+
}
323+
}

‎reactive/kotlinx-coroutines-reactive/test/PublisherMultiTest.kt

+24-1
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package kotlinx.coroutines.reactive
66

77
import kotlinx.coroutines.*
8+
import kotlinx.coroutines.selects.*
89
import org.junit.Test
910
import kotlin.test.*
1011

@@ -16,7 +17,7 @@ class PublisherMultiTest : TestBase() {
1617
// concurrent emitters (many coroutines)
1718
val jobs = List(n) {
1819
// launch
19-
launch {
20+
launch(Dispatchers.Default) {
2021
send(it)
2122
}
2223
}
@@ -28,4 +29,26 @@ class PublisherMultiTest : TestBase() {
2829
}
2930
assertEquals(n, resultSet.size)
3031
}
32+
33+
@Test
34+
fun testConcurrentStressOnSend() = runBlocking {
35+
val n = 10_000 * stressTestMultiplier
36+
val observable = publish<Int> {
37+
// concurrent emitters (many coroutines)
38+
val jobs = List(n) {
39+
// launch
40+
launch(Dispatchers.Default) {
41+
select<Unit> {
42+
onSend(it) {}
43+
}
44+
}
45+
}
46+
jobs.forEach { it.join() }
47+
}
48+
val resultSet = mutableSetOf<Int>()
49+
observable.collect {
50+
assertTrue(resultSet.add(it))
51+
}
52+
assertEquals(n, resultSet.size)
53+
}
3154
}

‎reactive/kotlinx-coroutines-rx2/src/RxObservable.kt

+14-1
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import kotlinx.atomicfu.*
1010
import kotlinx.coroutines.*
1111
import kotlinx.coroutines.channels.*
1212
import kotlinx.coroutines.internal.*
13+
import kotlinx.coroutines.intrinsics.*
1314
import kotlinx.coroutines.selects.*
1415
import kotlinx.coroutines.sync.*
1516
import kotlin.coroutines.*
@@ -95,10 +96,22 @@ private class RxObservableCoroutine<T : Any>(
9596
element: T,
9697
block: suspend (SendChannel<T>) -> R
9798
) {
98-
mutex.onLock.registerSelectClause2(select, null) {
99+
val clause = suspend {
99100
doLockedNext(element)?.let { throw it }
100101
block(this)
101102
}
103+
104+
// This is the default replacement proposed in onLock replacement
105+
launch(start = CoroutineStart.UNDISPATCHED) {
106+
mutex.lock()
107+
// Already selected -- bail out
108+
if (!select.trySelect()) {
109+
mutex.unlock()
110+
return@launch
111+
}
112+
113+
clause.startCoroutineCancellable(select.completion)
114+
}
102115
}
103116

104117
// assert: mutex.isLocked()

‎reactive/kotlinx-coroutines-rx2/test/ObservableCompletionStressTest.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import kotlin.coroutines.*
1212
class ObservableCompletionStressTest : TestBase() {
1313
private val N_REPEATS = 10_000 * stressTestMultiplier
1414

15-
private fun CoroutineScope.range(context: CoroutineContext, start: Int, count: Int) = rxObservable(context) {
15+
private fun range(context: CoroutineContext, start: Int, count: Int) = rxObservable(context) {
1616
for (x in start until start + count) send(x)
1717
}
1818

@@ -33,4 +33,4 @@ class ObservableCompletionStressTest : TestBase() {
3333
}
3434
}
3535
}
36-
}
36+
}

‎reactive/kotlinx-coroutines-rx2/test/ObservableMultiTest.kt

+25-1
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package kotlinx.coroutines.rx2
66

77
import io.reactivex.*
88
import kotlinx.coroutines.*
9+
import kotlinx.coroutines.selects.*
910
import org.junit.Test
1011
import java.io.*
1112
import kotlin.test.*
@@ -47,6 +48,29 @@ class ObservableMultiTest : TestBase() {
4748
}
4849
}
4950

51+
@Test
52+
fun testConcurrentStressOnSend() {
53+
val n = 10_000 * stressTestMultiplier
54+
val observable = rxObservable<Int> {
55+
newCoroutineContext(coroutineContext)
56+
// concurrent emitters (many coroutines)
57+
val jobs = List(n) {
58+
// launch
59+
launch(Dispatchers.Default) {
60+
val i = it
61+
select<Unit> {
62+
onSend(i) {}
63+
}
64+
}
65+
}
66+
jobs.forEach { it.join() }
67+
}
68+
checkSingleValue(observable.toList()) { list ->
69+
assertEquals(n, list.size)
70+
assertEquals((0 until n).toList(), list.sorted())
71+
}
72+
}
73+
5074
@Test
5175
fun testIteratorResendUnconfined() {
5276
val n = 10_000 * stressTestMultiplier
@@ -88,4 +112,4 @@ class ObservableMultiTest : TestBase() {
88112
assertEquals("OK", it)
89113
}
90114
}
91-
}
115+
}

‎reactive/kotlinx-coroutines-rx3/src/RxObservable.kt

+14-1
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import kotlinx.coroutines.selects.*
1313
import kotlinx.coroutines.sync.*
1414
import kotlin.coroutines.*
1515
import kotlinx.coroutines.internal.*
16+
import kotlinx.coroutines.intrinsics.*
1617

1718
/**
1819
* Creates cold [observable][Observable] that will run a given [block] in a coroutine.
@@ -95,10 +96,22 @@ private class RxObservableCoroutine<T : Any>(
9596
element: T,
9697
block: suspend (SendChannel<T>) -> R
9798
) {
98-
mutex.onLock.registerSelectClause2(select, null) {
99+
val clause = suspend {
99100
doLockedNext(element)?.let { throw it }
100101
block(this)
101102
}
103+
104+
// This is the default replacement proposed in onLock replacement
105+
launch(start = CoroutineStart.UNDISPATCHED) {
106+
mutex.lock()
107+
// Already selected -- bail out
108+
if (!select.trySelect()) {
109+
mutex.unlock()
110+
return@launch
111+
}
112+
113+
clause.startCoroutineCancellable(select.completion)
114+
}
102115
}
103116

104117
// assert: mutex.isLocked()

0 commit comments

Comments
 (0)