Skip to content

Commit 65a846b

Browse files
ForteScarletForteScarlet
authored
Merge pull request #832 from simple-robot/pref-jvm-blocking-runner
优化 `BlockingRunner` 内部实现
2 parents 7feb3f8 + c1cba91 commit 65a846b

File tree

2 files changed

+144
-49
lines changed

2 files changed

+144
-49
lines changed

‎simbot-commons/simbot-common-suspend-runner/src/jvmMain/kotlin/love/forte/simbot/suspendrunner/BlockingRunner.kt

Lines changed: 72 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,7 @@ private inline fun initDispatcher(
338338
dispatcher eq DISPATCHER_USE_CUSTOM_PROPERTY_VALUE -> loadCustomBlockingDispatcher(
339339
Thread.currentThread().contextClassLoader
340340
)
341+
341342
else -> null
342343
}
343344

@@ -563,12 +564,14 @@ private var runInNoScopeBlockingStrategy: RunInNoScopeBlockingStrategy = Default
563564

564565
@OptIn(ExperimentalSimbotAPI::class)
565566
private object DefaultRunInNoScopeBlockingStrategy : RunInNoScopeBlockingStrategy {
567+
@kotlin.jvm.Throws(Exception::class)
566568
override fun <T> invoke(context: CoroutineContext, block: suspend () -> T): T {
567569
val runner = SuspendRunner<T>(context)
568570
block.startCoroutine(runner)
569571
return runner.await(SuspendRunner.isWaitTimeoutEnabled)
570572
}
571573

574+
@kotlin.jvm.Throws(Exception::class)
572575
fun <T> invokeWithoutTimeoutLog(context: CoroutineContext, block: suspend () -> T): T {
573576
val runner = SuspendRunner<T>(context)
574577
block.startCoroutine(runner)
@@ -595,31 +598,31 @@ public fun setRunInNoScopeBlockingStrategy(strategy: RunInNoScopeBlockingStrateg
595598
*
596599
* 在默认未提供上下文的情况下，[runInBlocking] 所使用的 [context] 为 [DefaultBlockingContext].
597600
*
598-
* @throws RunInBlockingException [block] 中产生的所有异常（除了 [TimeoutCancellationException]）均会被包装在 [RunInBlockingException] 的 `cause` 中被抛出。
601+
* @throws Exception 原函数可能被抛出的任何异常
602+
* @throws RunInBlockingException 当出现执行 [block] 过程中由于 future 或线程中断等非 [block] 本身产生的异常时被包装为 [RunInBlockingException]
599603
*
600604
* @see DefaultBlockingContext
601605
* @see runBlocking
602606
*/
603607
@OptIn(ExperimentalSimbotAPI::class, InternalSimbotAPI::class)
604-
@Throws(RunInBlockingException::class)
608+
@Throws(Exception::class)
605609
public fun <T> runInBlocking(
606610
context: CoroutineContext = DefaultBlockingContext,
607611
block: suspend CoroutineScope.() -> T,
608-
): T = runCatching {
609-
runInBlockingStrategy(context, block)
610-
}.getOrElse { throw `$RunInBlockingException$`(it) }
612+
): T = runInBlockingStrategy(context, block)
611613

612614
/**
613615
* 如果超时，则抛出 [TimeoutCancellationException].
614616
*
615617
* @throws TimeoutCancellationException 如果超时
616-
* @throws RunInBlockingException [block] 中产生的所有异常（除了 [TimeoutCancellationException]）均会被包装在 [RunInBlockingException] 的 `cause` 中被抛出。
618+
* @throws Exception 原函数可能被抛出的任何异常
619+
* @throws RunInBlockingException 当出现执行 [block] 过程中由于 future 或线程中断等非 [block] 本身产生的异常时被包装为 [RunInBlockingException]
617620
*
618621
* @see runInBlocking
619622
* @see withTimeout
620623
*/
621624
@OptIn(ExperimentalSimbotAPI::class, InternalSimbotAPI::class)
622-
@Throws(RunInBlockingException::class)
625+
@Throws(Exception::class)
623626
public fun <T> runInTimeoutBlocking(
624627
timeout: Long,
625628
context: CoroutineContext = DefaultBlockingContext,
@@ -640,46 +643,41 @@ public fun <T> runInTimeoutBlocking(
640643
*
641644
* 在默认未提供上下文的情况下，[runInBlocking] 所使用的 [context] 为 [DefaultBlockingContext].
642645
*
643-
* @throws RunInBlockingException [block] 中产生的所有异常均会被包装在 [RunInBlockingException] 的 `cause` 中被抛出。
644-
*
646+
* @throws Exception 原函数可能被抛出的任何异常
647+
* @throws RunInBlockingException 当出现执行 [block] 过程中由于 future 或线程中断等非 [block] 本身产生的异常时被包装为 [RunInBlockingException]
645648
* @see DefaultBlockingContext
646649
* @see runBlocking
647650
*/
648651
@OptIn(ExperimentalSimbotAPI::class, InternalSimbotAPI::class)
649-
@Throws(RunInBlockingException::class)
652+
@Throws(Exception::class)
650653
public fun <T> runInNoScopeBlocking(
651654
context: CoroutineContext = DefaultBlockingContext,
652655
block: suspend () -> T,
653-
): T = runCatching {
654-
runInNoScopeBlockingStrategy(context, block)
655-
}.getOrElse { throw `$RunInBlockingException$`(it) }
656+
): T = runInNoScopeBlockingStrategy(context, block)
656657

657658
/**
658659
* @suppress 内部API
659660
*
660-
* @throws RunInBlockingException [block] 中产生的所有异常均会被包装在 [RunInBlockingException] 的 `cause` 中被抛出。
661+
* @throws Exception 原函数可能被抛出的任何异常
662+
* @throws RunInBlockingException 当出现执行 [block] 过程中由于 future 或线程中断等非 [block] 本身产生的异常时被包装为 [RunInBlockingException]
661663
*
662664
* @see runInNoScopeBlocking
663665
* @see DefaultBlockingContext
664666
* @see runBlocking
665667
*/
666668
@OptIn(ExperimentalSimbotAPI::class)
667-
@Throws(RunInBlockingException::class)
669+
@Throws(Exception::class)
668670
@InternalSimbotAPI
669671
public fun <T> runInNoScopeBlockingWithoutTimeoutDebug(
670672
context: CoroutineContext = DefaultBlockingContext,
671673
block: suspend () -> T,
672674
): T {
673-
try {
674-
val strategy = runInNoScopeBlockingStrategy
675-
if (strategy is DefaultRunInNoScopeBlockingStrategy) {
676-
return DefaultRunInNoScopeBlockingStrategy.invokeWithoutTimeoutLog(context, block)
677-
}
678-
679-
return strategy(context, block)
680-
} catch (e: Throwable) {
681-
throw `$RunInBlockingException$`(e)
675+
val strategy = runInNoScopeBlockingStrategy
676+
if (strategy is DefaultRunInNoScopeBlockingStrategy) {
677+
return DefaultRunInNoScopeBlockingStrategy.invokeWithoutTimeoutLog(context, block)
682678
}
679+
680+
return strategy(context, block)
683681
}
684682

685683
/**
@@ -746,7 +744,7 @@ public fun <T> `$$asReserve`(scope: CoroutineScope? = null, block: suspend () ->
746744

747745
@InternalSimbotAPI
748746
@Deprecated("Just used by compiler", level = DeprecationLevel.HIDDEN)
749-
@Throws(RunInBlockingException::class)
747+
@Throws(Exception::class)
750748
public fun <T> `$$runInBlocking`(block: suspend () -> T): T = runInNoScopeBlocking(block = block)
751749

752750

@@ -784,38 +782,51 @@ public fun <T> `$$runInAsyncNullable`(block: suspend () -> T, scope: CoroutineSc
784782

785783
/**
786784
* 使用在 `runBlocking` 或相关函数中，用于将运行其中的函数所抛出的函数捕获并包装。
787-
* 内容函数抛出的真正异常在 [cause] 中。
785+
*
786+
* [RunInBlockingException] 只会包装那些由 future 或者线程中断导致的异常，
787+
* 实际执行的blocking函数所抛出的异常会被原样抛出。
788+
*
789+
* 通常来讲，[cause] 可能是：
790+
* - [CancellationException]
791+
* - [ExecutionException] (概率很低）
792+
* - [InterruptedException]
788793
*/
789794
public sealed class RunInBlockingException protected constructor(cause: Throwable) : RuntimeException(cause)
790795

791-
792796
@Suppress("unused", "ClassName")
793797
private class `$RunInBlockingException$`(cause: Throwable) : RunInBlockingException(cause)
794798

799+
private fun throwRunInBlockingException(cause: Throwable): Nothing =
800+
throw `$RunInBlockingException$`(cause)
801+
795802
// Yes. I am the BlockingRunner.
796803
private class SuspendRunner<T>(override val context: CoroutineContext = EmptyCoroutineContext) : Continuation<T> {
797804
@Suppress("unused")
798805
@Volatile
799-
var s: Int = 0
806+
var s: Int = SIGNAL_NONE
800807
// 1 = resume - success
801808
// 2 = resume - exception
802809
// 3 = suspend
803810

811+
@Volatile
812+
var value: Any? = null
804813
// 1 -> value
805814
// 2 -> ex
806815
// 3 -> CompletableFuture<T>
807-
@Volatile
808-
var value: Any? = null
809816

810817
private object NULL
811818

812819
override fun resumeWith(result: Result<T>) {
820+
// 先变更信号
813821
val resumed =
814822
signalUpdater.compareAndSet(
815823
this,
816824
SIGNAL_NONE,
817825
if (result.isSuccess) SIGNAL_RESUME_SUCCESS else SIGNAL_RESUME_FAILED
818826
)
827+
828+
// 信号变更失败，说明现在信号是 SUSPEND
829+
// 那么 value 就已经有一个 future 值了，或者它应该被初始化为一个 future
819830
if (!resumed) {
820831
// value is a Future.
821832
@Suppress("UNCHECKED_CAST")
@@ -839,12 +850,14 @@ private class SuspendRunner<T>(override val context: CoroutineContext = EmptyCor
839850
}
840851

841852
/**
853+
* @param isWaitTimeoutEnabled 是否输出长时间阻塞警告
854+
*
842855
* @see CompletableFuture.join
843856
* @see CompletableFuture.get
844857
* @throws CancellationException cancellation
845858
* @throws CompletionException completion
846-
* @throws ExecutionException if [ExecutionException.cause] is null
847-
* @throws Exception [ExecutionException.cause]
859+
* @throws InterruptedException
860+
* @throws RunInBlockingException
848861
*/
849862
@Suppress("UNCHECKED_CAST", "ReturnCount", "ThrowsCount")
850863
@Throws(Exception::class)
@@ -875,29 +888,36 @@ private class SuspendRunner<T>(override val context: CoroutineContext = EmptyCor
875888
throw value as Throwable
876889
}
877890

891+
// 不需要检测长时间等待的日志,
892+
// 或者不需要在virtual时输出日志，且当前是virtual thread
878893
if (!isWaitTimeoutEnabled || (!logIfVirtual && Thread.currentThread().isVirtualThread())) {
879894
try {
880895
return future.get()
881896
} catch (cancellation: CancellationException) {
882-
throw cancellation
897+
throwRunInBlockingException(cancellation)
883898
} catch (execution: ExecutionException) {
884-
throw execution.cause ?: execution
899+
// 一般来讲 ExecutionException.cause 不会是 null
900+
throw execution.cause ?: throwRunInBlockingException(execution)
885901
}
886-
// catch (other: Throwable) {
887-
// throw other // CompletionException(other)
888-
// }
902+
// InterruptedException 直接向外传递
889903
}
890904

905+
// 需要输出长时间等待日志
906+
891907
var times = 0
892-
while (!future.isDone && !Thread.currentThread().isInterrupted) {
908+
while (!future.isDone) {
909+
if (Thread.interrupted()) {
910+
throw InterruptedException()
911+
}
912+
893913
if (times > 0) {
894914
val duration = (waitTimeout * times).milliseconds
895915
if (logger.isDebugEnabled) {
896916
val durationString = duration.toString()
897917
logger.warn("Blocking runner has been blocking for at least {}.", durationString)
898-
val e: Throwable = LongTimeBlockingException(durationString)
918+
val e: Throwable = ProlongedBlockingException(durationString)
899919
logger.debug(
900-
"Long time blocking duration at least {}",
920+
"Prolonged blocking duration at least {}",
901921
durationString,
902922
e
903923
)
@@ -916,21 +936,24 @@ private class SuspendRunner<T>(override val context: CoroutineContext = EmptyCor
916936
} catch (ignore: TimeoutException) {
917937
times += 1
918938
} catch (cancellation: CancellationException) {
919-
throw cancellation
939+
throwRunInBlockingException(cancellation)
920940
} catch (execution: ExecutionException) {
921-
throw execution.cause ?: execution
941+
throw execution.cause ?: throwRunInBlockingException(execution)
922942
}
923-
// catch (other: Throwable) {
924-
// throw other // CompletionException(other)
925-
// }
926943
}
927944

928-
// done.
929-
return future.join()
945+
// Is done, but not in the future.get(timeout)
946+
try {
947+
return future.join()
948+
} catch (cancellation: CancellationException) {
949+
throwRunInBlockingException(cancellation)
950+
} catch (execution: CompletionException) {
951+
throwRunInBlockingException(execution)
952+
}
930953
}
931954

932-
// for displaying the stack only
933-
private class LongTimeBlockingException(message: String) : RuntimeException(message)
955+
// Used only to show the stack
956+
private class ProlongedBlockingException(message: String) : RuntimeException(message)
934957

935958
companion object {
936959
private const val SIGNAL_NONE = 0

‎simbot-commons/simbot-common-suspend-runner/src/jvmTest/kotlin/love/forte/simbot/suspendrunner/BlockingRunnerTests.kt

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Copyright (c) 2024. ForteScarlet.
3+
*
4+
* Project https://github.com/simple-robot/simpler-robot
5+
* Email [email protected]
6+
*
7+
* This file is part of the Simple Robot Library (Alias: simple-robot, simbot, etc.).
8+
*
9+
* This program is free software: you can redistribute it and/or modify
10+
* it under the terms of the GNU Lesser General Public License as published by
11+
* the Free Software Foundation, either version 3 of the License, or
12+
* (at your option) any later version.
13+
*
14+
* This program is distributed in the hope that it will be useful,
15+
* but WITHOUT ANY WARRANTY; without even the implied warranty of
16+
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17+
* Lesser GNU General Public License for more details.
18+
*
19+
* You should have received a copy of the Lesser GNU General Public License
20+
* along with this program. If not, see <https://www.gnu.org/licenses/>.
21+
*
22+
*/
23+
24+
package love.forte.simbot.suspendrunner
25+
26+
import kotlinx.coroutines.delay
27+
import kotlinx.coroutines.suspendCancellableCoroutine
28+
import kotlin.coroutines.resume
29+
import kotlin.coroutines.resumeWithException
30+
import kotlin.test.Test
31+
import kotlin.test.assertFails
32+
import kotlin.test.assertIs
33+
34+
35+
/**
36+
*
37+
* @author ForteScarlet
38+
*/
39+
class BlockingRunnerTests {
40+
@Test
41+
fun runBlockingExceptionallyTest() {
42+
runInNoScopeBlocking { runNormally() }
43+
44+
assertIs<RunException>(
45+
assertFails {
46+
runInNoScopeBlocking { runExceptionally1() }
47+
}
48+
)
49+
50+
assertIs<RunException>(
51+
assertFails {
52+
runInNoScopeBlocking { runExceptionally2() }
53+
}
54+
)
55+
}
56+
57+
private suspend fun runNormally() = suspendCancellableCoroutine { continuation ->
58+
continuation.resume(0)
59+
}
60+
61+
private suspend fun runExceptionally1() {
62+
delay(1)
63+
throw RunException()
64+
}
65+
66+
private suspend fun runExceptionally2() = suspendCancellableCoroutine<Int> { continuation ->
67+
continuation.resumeWithException(RunException())
68+
}
69+
70+
71+
private class RunException : RuntimeException()
72+
}

0 commit comments

Comments
 (0)