Skip to content

Commit 2adf8bc

Browse files
committed
AbstractCoroutine.start is introduced.
1 parent d1842c6 commit 2adf8bc

File tree

20 files changed

+168
-158
lines changed

20 files changed

+168
-158
lines changed

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonCoroutineStart.kt

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,16 @@
1616

1717
package kotlinx.coroutines.experimental
1818

19-
import kotlin.coroutines.experimental.Continuation
19+
import kotlin.coroutines.experimental.*
2020

2121
public expect enum class CoroutineStart {
2222
DEFAULT,
2323
LAZY,
2424
ATOMIC,
2525
UNDISPATCHED;
26+
@Deprecated(message = "Use AbstractCoroutine.start") // todo: make it internal & rename
2627
public operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>)
28+
@Deprecated(message = "Use AbstractCoroutine.start") // todo: make it internal & rename
2729
public operator fun <R, T> invoke(block: suspend R.() -> T, receiver: R, completion: Continuation<T>)
2830
public val isLazy: Boolean
2931
}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/AbstractCoroutine.kt

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package kotlinx.coroutines.experimental
1818

19+
import kotlinx.coroutines.experimental.CoroutineStart.*
20+
import kotlinx.coroutines.experimental.intrinsics.*
1921
import kotlin.coroutines.experimental.*
2022

2123
/**
@@ -46,7 +48,7 @@ public abstract class AbstractCoroutine<in T>(
4648
* Invocation of this function may cause this coroutine to become cancelled if parent is already cancelled,
4749
* in which case it synchronously invokes all the corresponding handlers.
4850
*/
49-
public fun initParentJob() {
51+
internal fun initParentJob() {
5052
initParentJobInternal(parentContext[Job])
5153
}
5254

@@ -120,5 +122,43 @@ public abstract class AbstractCoroutine<in T>(
120122
val coroutineName = context.coroutineName ?: return super.nameString()
121123
return "\"$coroutineName\":${super.nameString()}"
122124
}
125+
126+
/**
127+
* Starts the corresponding block as a coroutine with this coroutine start strategy.
128+
*
129+
* First, this function initializes parent job from the `parentContext` of this coroutine that was passed to it
130+
* during construction. Second, it starts the coroutine based on [start] parameter:
131+
*
132+
* * [DEFAULT] uses [startCoroutineCancellable].
133+
* * [ATOMIC] uses [startCoroutine].
134+
* * [UNDISPATCHED] uses [startCoroutineUndispatched].
135+
* * [LAZY] does nothing.
136+
*
137+
* This function shall be invoked at most once.
138+
*/
139+
public fun start(start: CoroutineStart, block: suspend () -> T) {
140+
initParentJob()
141+
@Suppress("DEPRECATION")
142+
start(block, this)
143+
}
144+
145+
/**
146+
* Starts the corresponding block with receiver as a coroutine with this coroutine start strategy.
147+
*
148+
* First, this function initializes parent job from the `parentContext` of this coroutine that was passed to it
149+
* during construction. Second, it starts the coroutine based on [start] parameter:
150+
*
151+
* * [DEFAULT] uses [startCoroutineCancellable].
152+
* * [ATOMIC] uses [startCoroutine].
153+
* * [UNDISPATCHED] uses [startCoroutineUndispatched].
154+
* * [LAZY] does nothing.
155+
*
156+
* This function shall be invoked at most once.
157+
*/
158+
public fun <R> start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
159+
initParentJob()
160+
@Suppress("DEPRECATION")
161+
start(block, receiver, this)
162+
}
123163
}
124164

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,9 @@
1616

1717
package kotlinx.coroutines.experimental
1818

19-
import java.util.concurrent.locks.LockSupport
19+
import java.util.concurrent.locks.*
2020
import kotlin.coroutines.experimental.*
21-
import kotlin.coroutines.experimental.intrinsics.startCoroutineUninterceptedOrReturn
22-
import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
21+
import kotlin.coroutines.experimental.intrinsics.*
2322

2423
// --------------- basic coroutine builders ---------------
2524

@@ -62,8 +61,7 @@ public actual fun launch(
6261
val coroutine = if (start.isLazy)
6362
LazyStandaloneCoroutine(newContext, block) else
6463
StandaloneCoroutine(newContext, active = true)
65-
coroutine.initParentJob()
66-
start(block, coroutine, coroutine)
64+
coroutine.start(start, coroutine, block)
6765
return coroutine
6866
}
6967

@@ -127,6 +125,7 @@ public actual suspend fun <T> withContext(
127125
delegate = cont,
128126
resumeMode = if (start == CoroutineStart.ATOMIC) MODE_ATOMIC_DEFAULT else MODE_CANCELLABLE)
129127
completion.initParentJobInternal(newContext[Job]) // attach to job
128+
@Suppress("DEPRECATION")
130129
start(block, completion)
131130
completion.getResult()
132131
}
@@ -169,8 +168,7 @@ public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, bl
169168
val eventLoop = if (context[ContinuationInterceptor] == null) BlockingEventLoop(currentThread) else null
170169
val newContext = newCoroutineContext(context + (eventLoop ?: EmptyCoroutineContext))
171170
val coroutine = BlockingCoroutine<T>(newContext, currentThread, privateEventLoop = eventLoop != null)
172-
coroutine.initParentJob()
173-
block.startCoroutine(coroutine, coroutine)
171+
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
174172
return coroutine.joinBlocking()
175173
}
176174

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineStart.kt

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,8 @@
1717
package kotlinx.coroutines.experimental
1818

1919
import kotlinx.coroutines.experimental.CoroutineStart.*
20-
import kotlinx.coroutines.experimental.intrinsics.startCoroutineUndispatched
21-
import kotlin.coroutines.experimental.Continuation
22-
import kotlin.coroutines.experimental.startCoroutine
20+
import kotlinx.coroutines.experimental.intrinsics.*
21+
import kotlin.coroutines.experimental.*
2322

2423
/**
2524
* Defines start option for coroutines builders.
@@ -91,7 +90,10 @@ public actual enum class CoroutineStart {
9190
* * [ATOMIC] uses [startCoroutine].
9291
* * [UNDISPATCHED] uses [startCoroutineUndispatched].
9392
* * [LAZY] does nothing.
93+
*
94+
* @suppress **Deprecated**: Use [AbstractCoroutine.start]
9495
*/
96+
@Deprecated(message = "Use AbstractCoroutine.start") // todo: make it internal & rename
9597
public actual operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>) =
9698
when (this) {
9799
CoroutineStart.DEFAULT -> block.startCoroutineCancellable(completion)
@@ -107,7 +109,10 @@ public actual enum class CoroutineStart {
107109
* * [ATOMIC] uses [startCoroutine].
108110
* * [UNDISPATCHED] uses [startCoroutineUndispatched].
109111
* * [LAZY] does nothing.
112+
*
113+
* @suppress **Deprecated**: Use [AbstractCoroutine.start]
110114
*/
115+
@Deprecated(message = "Use AbstractCoroutine.start") // todo: make it internal & rename
111116
public actual operator fun <R, T> invoke(block: suspend R.() -> T, receiver: R, completion: Continuation<T>) =
112117
when (this) {
113118
CoroutineStart.DEFAULT -> block.startCoroutineCancellable(receiver, completion)

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,7 @@
1717
package kotlinx.coroutines.experimental
1818

1919
import kotlinx.coroutines.experimental.selects.*
20-
import kotlin.coroutines.experimental.ContinuationInterceptor
21-
import kotlin.coroutines.experimental.CoroutineContext
20+
import kotlin.coroutines.experimental.*
2221

2322
/**
2423
* Deferred value is a non-blocking cancellable future.
@@ -167,8 +166,7 @@ public actual fun <T> async(
167166
val coroutine = if (start.isLazy)
168167
LazyDeferredCoroutine(newContext, block) else
169168
DeferredCoroutine<T>(newContext, active = true)
170-
coroutine.initParentJob()
171-
start(block, coroutine, coroutine)
169+
coroutine.start(start, coroutine, block)
172170
return coroutine
173171
}
174172

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt

Lines changed: 6 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,11 @@
1616

1717
package kotlinx.coroutines.experimental
1818

19-
import kotlinx.coroutines.experimental.selects.SelectBuilder
20-
import kotlinx.coroutines.experimental.selects.select
21-
import java.util.concurrent.TimeUnit
22-
import kotlin.coroutines.experimental.Continuation
23-
import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
24-
import kotlin.coroutines.experimental.intrinsics.startCoroutineUninterceptedOrReturn
25-
import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
19+
import kotlinx.coroutines.experimental.intrinsics.*
20+
import kotlinx.coroutines.experimental.selects.*
21+
import java.util.concurrent.*
22+
import kotlin.coroutines.experimental.*
23+
import kotlin.coroutines.experimental.intrinsics.*
2624

2725
/**
2826
* Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
@@ -78,21 +76,9 @@ private fun <U, T: U> setupTimeout(
7876
val cont = coroutine.cont
7977
val context = cont.context
8078
coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine.unit, coroutine))
81-
coroutine.initParentJob()
8279
// restart block using new coroutine with new job,
8380
// however start it as undispatched coroutine, because we are already in the proper context
84-
val result = try {
85-
block.startCoroutineUninterceptedOrReturn(receiver = coroutine, completion = coroutine)
86-
} catch (e: Throwable) {
87-
CompletedExceptionally(e)
88-
}
89-
return when {
90-
result == COROUTINE_SUSPENDED -> COROUTINE_SUSPENDED
91-
coroutine.makeCompletingOnce(result, MODE_IGNORE) -> {
92-
if (result is CompletedExceptionally) throw result.exception else result
93-
}
94-
else -> COROUTINE_SUSPENDED
95-
}
81+
return coroutine.startUndispatchedOrReturn(coroutine, block)
9682
}
9783

9884
/**

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Actor.kt

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,8 @@
1717
package kotlinx.coroutines.experimental.channels
1818

1919
import kotlinx.coroutines.experimental.*
20-
import kotlinx.coroutines.experimental.selects.SelectClause2
21-
import kotlinx.coroutines.experimental.selects.SelectInstance
22-
import kotlin.coroutines.experimental.ContinuationInterceptor
23-
import kotlin.coroutines.experimental.CoroutineContext
20+
import kotlinx.coroutines.experimental.selects.*
21+
import kotlin.coroutines.experimental.*
2422

2523
/**
2624
* Scope for [actor] coroutine builder.
@@ -94,8 +92,7 @@ public fun <E> actor(
9492
val coroutine = if (start.isLazy)
9593
LazyActorCoroutine(newContext, channel, block) else
9694
ActorCoroutine(newContext, channel, active = true)
97-
coroutine.initParentJob()
98-
start(block, coroutine, coroutine)
95+
coroutine.start(start, coroutine, block)
9996
return coroutine
10097
}
10198

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Produce.kt

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,7 @@
1717
package kotlinx.coroutines.experimental.channels
1818

1919
import kotlinx.coroutines.experimental.*
20-
import kotlin.coroutines.experimental.ContinuationInterceptor
21-
import kotlin.coroutines.experimental.CoroutineContext
22-
import kotlin.coroutines.experimental.startCoroutine
20+
import kotlin.coroutines.experimental.*
2321

2422
/**
2523
* Scope for [produce] coroutine builder.
@@ -83,8 +81,7 @@ public fun <E> produce(
8381
val channel = Channel<E>(capacity)
8482
val newContext = newCoroutineContext(context, parent)
8583
val coroutine = ProducerCoroutine(newContext, channel)
86-
coroutine.initParentJob()
87-
block.startCoroutine(coroutine, coroutine)
84+
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
8885
return coroutine
8986
}
9087

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/intrinsics/Undispatched.kt

Lines changed: 48 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,9 @@
1616

1717
package kotlinx.coroutines.experimental.intrinsics
1818

19-
import kotlinx.coroutines.experimental.resumeCancellable
20-
import kotlin.coroutines.experimental.Continuation
21-
import kotlin.coroutines.experimental.createCoroutine
22-
import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
23-
import kotlin.coroutines.experimental.intrinsics.startCoroutineUninterceptedOrReturn
24-
import kotlin.coroutines.experimental.suspendCoroutine
19+
import kotlinx.coroutines.experimental.*
20+
import kotlin.coroutines.experimental.*
21+
import kotlin.coroutines.experimental.intrinsics.*
2522

2623
/**
2724
* Use this function to restart coroutine directly from inside of [suspendCoroutine].
@@ -56,3 +53,48 @@ internal fun <R, T> (suspend (R) -> T).startCoroutineUndispatched(receiver: R, c
5653
if (value !== COROUTINE_SUSPENDED)
5754
completion.resume(value as T)
5855
}
56+
57+
/**
58+
* Starts the corresponding block as a coroutine with this coroutine start strategy.
59+
*
60+
* First, this function initializes parent job from the `parentContext` of this coroutine that was passed to it
61+
* during construction. Second, it starts the coroutine using [startCoroutineUninterceptedOrReturn].
62+
*
63+
* This function shall be invoked at most once.
64+
*
65+
* @suppress **This is unstable API and it is subject to change.**
66+
*/
67+
public fun <T> AbstractCoroutine<T>.startUndispatchedOrReturn(block: suspend () -> T): Any? {
68+
initParentJob()
69+
return undispatchedResult { block.startCoroutineUninterceptedOrReturn(this) }
70+
}
71+
72+
/**
73+
* Starts the corresponding block with receiver as a coroutine with this coroutine start strategy.
74+
*
75+
* First, this function initializes parent job from the `parentContext` of this coroutine that was passed to it
76+
* during construction. Second, it starts the coroutine using [startCoroutineUninterceptedOrReturn].
77+
*
78+
* This function shall be invoked at most once.
79+
*
80+
* @suppress **This is unstable API and it is subject to change.**
81+
*/
82+
public fun <T, R> AbstractCoroutine<T>.startUndispatchedOrReturn(receiver: R, block: suspend R.() -> T): Any? {
83+
initParentJob()
84+
return undispatchedResult { block.startCoroutineUninterceptedOrReturn(receiver, this) }
85+
}
86+
87+
private inline fun <T> AbstractCoroutine<T>.undispatchedResult(startBlock: () -> Any?): Any? {
88+
val result = try {
89+
startBlock()
90+
} catch (e: Throwable) {
91+
CompletedExceptionally(e)
92+
}
93+
return when {
94+
result == COROUTINE_SUSPENDED -> COROUTINE_SUSPENDED
95+
makeCompletingOnce(result, MODE_IGNORE) -> {
96+
if (result is CompletedExceptionally) throw result.exception else result
97+
}
98+
else -> COROUTINE_SUSPENDED
99+
}
100+
}

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ReaderJob.kt

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,7 @@
11
package kotlinx.coroutines.experimental.io
22

3-
import kotlinx.coroutines.experimental.CoroutineScope
4-
import kotlinx.coroutines.experimental.Job
5-
import kotlinx.coroutines.experimental.newCoroutineContext
6-
import kotlin.coroutines.experimental.CoroutineContext
7-
import kotlin.coroutines.experimental.startCoroutine
3+
import kotlinx.coroutines.experimental.*
4+
import kotlin.coroutines.experimental.*
85

96
/**
107
* A coroutine job that is reading from a byte channel
@@ -26,8 +23,7 @@ fun reader(coroutineContext: CoroutineContext,
2623
block: suspend ReaderScope.() -> Unit): ReaderJob {
2724
val newContext = newCoroutineContext(coroutineContext, parent)
2825
val coroutine = ReaderCoroutine(newContext, channel)
29-
coroutine.initParentJob()
30-
block.startCoroutine(coroutine, coroutine)
26+
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
3127
return coroutine
3228
}
3329

0 commit comments

Comments
 (0)