支持监听优先级, 事件传递拦截 (#279)

* 支持监听优先级, 事件传递拦截

* Fix test

* 并发

* 优先级&并发

* Test

* Fix unused

* To GlobalEventListeners

* Add tests

* intercept with subscribeAlways

* test listener.complete()

* Add functions

* Fix test and add new test

* Test concurrent listening

* Test concurrent listening

* update broadcast

* Fix Boom
Co-authored-by: Him188 <Him188@mamoe.net>

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/event/Event.kt

@@ -35,9 +35,11 @@ import kotlin.jvm.JvmSynthetic
  */
 interface Event {
 
-    @Deprecated("""
+    @Deprecated(
+        """
         Don't implement Event but extend AbstractEvent instead.
-    """, level = DeprecationLevel.HIDDEN) // so Kotlin class won't be compiled.
+    """, level = DeprecationLevel.HIDDEN
+    ) // so Kotlin class won't be compiled.
     @Suppress("WRONG_MODIFIER_CONTAINING_DECLARATION", "PropertyName")
     @get:JvmSynthetic // so Java user won't see it
     internal val DoNotImplementThisClassButExtendAbstractEvent: Nothing
@@ -54,10 +56,9 @@ abstract class AbstractEvent : Event {
     final override val DoNotImplementThisClassButExtendAbstractEvent: Nothing
         get() = throw Error("Shouldn't be reached")
 
-    private val _intercepted = atomic(false)
+    private var _intercepted = false
     private val _cancelled = atomic(false)
 
-
     /**
      * 事件是否已被拦截.
      *
@@ -65,7 +66,7 @@ abstract class AbstractEvent : Event {
      */
     @SinceMirai("1.0.0")
     val isIntercepted: Boolean
-        get() = _intercepted.value
+        get() = _intercepted
 
     /**
      * 拦截这个事件.
@@ -73,7 +74,7 @@ abstract class AbstractEvent : Event {
      */
     @SinceMirai("1.0.0")
     fun intercept() {
-        _intercepted.value = true
+        _intercepted = true
     }
 
 
@@ -145,5 +146,6 @@ interface BroadcastControllable : Event {
 @Deprecated(
     "use AbstractEvent and implement CancellableEvent",
     level = DeprecationLevel.ERROR,
-    replaceWith = ReplaceWith("AbstractEvent", "net.mamoe.mirai.event.AbstractEvent"))
+    replaceWith = ReplaceWith("AbstractEvent", "net.mamoe.mirai.event.AbstractEvent")
+)
 abstract class AbstractCancellableEvent : AbstractEvent(), CancellableEvent

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/event/internal/InternalEventListeners.kt

@@ -13,40 +13,41 @@ package net.mamoe.mirai.event.internal
 import kotlinx.coroutines.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
-import net.mamoe.mirai.event.Event
-import net.mamoe.mirai.event.EventDisabled
-import net.mamoe.mirai.event.Listener
-import net.mamoe.mirai.event.ListeningStatus
-import net.mamoe.mirai.utils.LockFreeLinkedList
-import net.mamoe.mirai.utils.MiraiInternalAPI
-import net.mamoe.mirai.utils.MiraiLogger
-import net.mamoe.mirai.utils.isRemoved
+import net.mamoe.mirai.event.*
+import net.mamoe.mirai.utils.*
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.coroutineContext
 import kotlin.jvm.JvmField
 import kotlin.reflect.KClass
 
+
 @PublishedApi
 internal fun <L : Listener<E>, E : Event> KClass<out E>.subscribeInternal(listener: L): L {
-    with(this.listeners()) {
-        addLast(listener)
+    with(GlobalEventListeners[listener.priority]) {
+        @Suppress("UNCHECKED_CAST")
+        val node = ListenerNode(listener as Listener<Event>, this@subscribeInternal)
+        @OptIn(MiraiInternalAPI::class)
+        addLast(node)
         listener.invokeOnCompletion {
-            this.remove(listener)
+            @OptIn(MiraiInternalAPI::class)
+            this.remove(node)
         }
     }
     return listener
 }
 
+
 @PublishedApi
 @Suppress("FunctionName")
 internal fun <E : Event> CoroutineScope.Handler(
     coroutineContext: CoroutineContext,
     concurrencyKind: Listener.ConcurrencyKind,
+    priority: Listener.EventPriority = Listener.EventPriority.NORMAL,
     handler: suspend (E) -> ListeningStatus
 ): Handler<E> {
     @OptIn(ExperimentalCoroutinesApi::class) // don't remove
     val context = this.newCoroutineContext(coroutineContext)
-    return Handler(context[Job], context, handler, concurrencyKind)
+    return Handler(context[Job], context, handler, concurrencyKind, priority)
 }
 
 /**
@@ -58,15 +59,17 @@ internal class Handler<in E : Event>
     parentJob: Job?,
     subscriberContext: CoroutineContext,
     @JvmField val handler: suspend (E) -> ListeningStatus,
-    override val concurrencyKind: Listener.ConcurrencyKind
+    override val concurrencyKind: Listener.ConcurrencyKind,
+    override val priority: Listener.EventPriority
 ) : Listener<E>, CompletableJob by SupervisorJob(parentJob) { // avoid being cancelled on handling event
 
     private val subscriberContext: CoroutineContext = subscriberContext + this // override Job.
 
+
     @MiraiInternalAPI
     val lock: Mutex? = when (concurrencyKind) {
-        Listener.ConcurrencyKind.CONCURRENT -> null
         Listener.ConcurrencyKind.LOCKED -> Mutex()
+        else -> null
     }
 
     @Suppress("unused")
@@ -95,14 +98,12 @@ internal class Handler<in E : Event>
     }
 }
 
-/**
- * 这个事件类的监听器 list
- */
-internal fun <E : Event> KClass<out E>.listeners(): EventListeners<E> = EventListenerManager.get(this)
-
-internal expect class EventListeners<E : Event>(clazz: KClass<E>) : LockFreeLinkedList<Listener<E>> {
-    @Suppress("UNCHECKED_CAST", "UNSUPPORTED", "NO_REFLECTION_IN_CLASS_PATH")
-    val supertypes: Set<KClass<out Event>>
+internal class ListenerNode(
+    val listener: Listener<Event>,
+    val owner: KClass<out Event>
+)
+internal expect object GlobalEventListeners {
+    operator fun get(priority: Listener.EventPriority): LockFreeLinkedList<ListenerNode>
 }
 
 internal expect class MiraiAtomicBoolean(initial: Boolean) {
@@ -112,63 +113,56 @@ internal expect class MiraiAtomicBoolean(initial: Boolean) {
     var value: Boolean
 }
 
-/**
- * 管理每个事件 class 的 [EventListeners].
- * [EventListeners] 是 lazy 的: 它们只会在被需要的时候才创建和存储.
- */
-internal object EventListenerManager {
-    private data class Registry<E : Event>(val clazz: KClass<E>, val listeners: EventListeners<E>)
-
-    private val registries = LockFreeLinkedList<Registry<*>>()
-
-    // 不要用 atomicfu. 在 publish 后会出现 VerifyError
-    private val lock: MiraiAtomicBoolean = MiraiAtomicBoolean(false)
-
-    @OptIn(MiraiInternalAPI::class)
-    @Suppress("UNCHECKED_CAST", "BooleanLiteralArgument")
-    internal tailrec fun <E : Event> get(clazz: KClass<out E>): EventListeners<E> {
-        registries.forEach {
-            if (it.clazz == clazz) {
-                return it.listeners as EventListeners<E>
-            }
-        }
-        if (lock.compareAndSet(false, true)) {
-            val registry = Registry(clazz as KClass<E>, EventListeners(clazz))
-            registries.addLast(registry)
-            lock.value = false
-            return registry.listeners
-        }
-        return get(clazz)
-    }
-}
 
 // inline: NO extra Continuation
 @Suppress("UNCHECKED_CAST")
 internal suspend inline fun Event.broadcastInternal() = coroutineScope {
     if (EventDisabled) return@coroutineScope
-
-    val listeners = this@broadcastInternal::class.listeners()
-    callAndRemoveIfRequired(this@broadcastInternal, listeners)
-    listeners.supertypes.forEach {
-        callAndRemoveIfRequired(this@broadcastInternal, it.listeners())
-    }
+    callAndRemoveIfRequired(this@broadcastInternal as? AbstractEvent ?: error("Events must extends AbstractEvent"))
 }
 
 @OptIn(MiraiInternalAPI::class)
-private fun <E : Event> CoroutineScope.callAndRemoveIfRequired(event: E, listeners: EventListeners<E>) {
-    // atomic foreach
-    listeners.forEachNode { node ->
-        launch {
-            val listener = node.nodeValue
-            if (listener.concurrencyKind == Listener.ConcurrencyKind.LOCKED) {
-                (listener as Handler).lock!!.withLock {
-                    if (!node.isRemoved() && listener.onEvent(event) == ListeningStatus.STOPPED) {
-                        listeners.remove(listener) // atomic remove
-                    }
+private suspend fun <E : AbstractEvent> callAndRemoveIfRequired(
+    event: E
+) {
+    coroutineScope {
+        for (p in Listener.EventPriority.values()) {
+            GlobalEventListeners[p].forEachNode { eventNode ->
+                if (event.isIntercepted) {
+                    return@coroutineScope
                 }
-            } else {
-                if (!node.isRemoved() && listener.onEvent(event) == ListeningStatus.STOPPED) {
-                    listeners.remove(listener) // atomic remove
+                val node = eventNode.nodeValue
+                if (!node.owner.isInstance(event)) return@forEachNode
+                val listener = node.listener
+                when (listener.concurrencyKind) {
+                    Listener.ConcurrencyKind.LOCKED -> {
+                        (listener as Handler).lock!!.withLock {
+                            kotlin.runCatching {
+                                when (listener.onEvent(event)) {
+                                    ListeningStatus.STOPPED -> {
+                                        removeNode(eventNode)
+                                    }
+                                    else -> {
+                                    }
+                                }
+                            }.onFailure {
+                                // TODO("Exception catching")
+                            }
+                        }
+                    }
+                    Listener.ConcurrencyKind.CONCURRENT -> {
+                        kotlin.runCatching {
+                            when (listener.onEvent(event)) {
+                                ListeningStatus.STOPPED -> {
+                                    removeNode(eventNode)
+                                }
+                                else -> {
+                                }
+                            }
+                        }.onFailure {
+                            // TODO("Exception catching")
+                        }
+                    }
                 }
             }
         }

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/utils/LockFreeLinkedList.kt

@@ -156,6 +156,18 @@ open class LockFreeLinkedList<E> {
         }
     }
 
+    open fun tryInsertAfter(node: LockFreeLinkedListNode<E>, newValue: E): Boolean {
+        if (node == tail) {
+            error("Cannot insert value after tail")
+        }
+        if (node.isRemoved()) {
+            return false
+        }
+        val next = node.nextNodeRef.value
+        val newNode = newValue.asNode(next)
+        return node.nextNodeRef.compareAndSet(next, newNode)
+    }
+
     /**
      * 先把元素建立好链表, 再加入到 list.
      */
@@ -329,7 +341,8 @@ open class LockFreeLinkedList<E> {
         }
     }
 
-    inline fun forEachNode(block: (LockFreeLinkedListNode<E>) -> Unit) {
+    inline fun forEachNode(block: LockFreeLinkedList<E>.(LockFreeLinkedListNode<E>) -> Unit) {
+        // Copy from forEach
         var node: LockFreeLinkedListNode<E> = head
         while (true) {
             if (node === tail) return
@@ -358,28 +371,41 @@ open class LockFreeLinkedList<E> {
     @Suppress("unused")
     open fun removeAll(elements: Collection<E>): Boolean = elements.all { remove(it) }
 
-    /*
-
-
-    private fun removeNode(node: Node<E>): Boolean {
+    @Suppress("DuplicatedCode")
+    open fun removeNode(node: LockFreeLinkedListNode<E>): Boolean {
         if (node == tail) {
             return false
         }
         while (true) {
             val before = head.iterateBeforeFirst { it === node }
             val toRemove = before.nextNode
-            val next = toRemove.nextNode
-            if (toRemove == tail) { // This
-                return true
+            if (toRemove === tail) {
+                return false
+            }
+            if (toRemove.isRemoved()) {
+                continue
             }
-            toRemove.nodeValue = null // logically remove first, then all the operations will recognize this node invalid
+            @Suppress("BooleanLiteralArgument") // false positive
+            if (!toRemove.removed.compareAndSet(false, true)) {
+                // logically remove: all the operations will recognize this node invalid
+                continue
+            }
+
 
-            if (before.nextNodeRef.compareAndSet(toRemove, next)) { // physically remove: try to fix the link
+            // physically remove: try to fix the link
+            var next: LockFreeLinkedListNode<E> = toRemove.nextNode
+            while (next !== tail && next.isRemoved()) {
+                next = next.nextNode
+            }
+            if (before.nextNodeRef.compareAndSet(toRemove, next)) {
                 return true
             }
         }
     }
 
+    /*
+
+
     fun removeAt(index: Int): E {
         require(index >= 0) { "index must be >= 0" }
         val nodeBeforeIndex = head.iterateValidNodeNTimes(index)

mirai-core/src/jvmMain/kotlin/net/mamoe/mirai/event/internal/MiraiAtomicBoolean.kt

+/*
+ * Copyright 2020 Mamoe Technologies and contributors.
+ *
+ * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ * Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ * https://github.com/mamoe/mirai/blob/master/LICENSE
+ */
+
 @file:OptIn(MiraiInternalAPI::class)
 
 package net.mamoe.mirai.event.internal
@@ -5,7 +14,10 @@ package net.mamoe.mirai.event.internal
 import net.mamoe.mirai.event.Event
 import net.mamoe.mirai.event.Listener
 import net.mamoe.mirai.utils.LockFreeLinkedList
+import net.mamoe.mirai.utils.LockFreeLinkedListNode
+import net.mamoe.mirai.utils.isRemoved
 import net.mamoe.mirai.utils.MiraiInternalAPI
+import java.util.*
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.reflect.KClass
 
@@ -24,14 +36,30 @@ internal actual class MiraiAtomicBoolean actual constructor(initial: Boolean) {
         }
 }
 
+internal actual object GlobalEventListeners {
+    private val map: Map<Listener.EventPriority, LockFreeLinkedList<ListenerNode>>
+
+    init {
+        val map = EnumMap<Listener.EventPriority, LockFreeLinkedList<ListenerNode>>(Listener.EventPriority::class.java)
+        Listener.EventPriority.values().forEach {
+            map[it] = LockFreeLinkedList()
+        }
+        this.map = map
+    }
+
+    actual operator fun get(priority: Listener.EventPriority): LockFreeLinkedList<ListenerNode> = map[priority]!!
+
+}
+
+/*
 internal actual class EventListeners<E : Event> actual constructor(clazz: KClass<E>) :
     LockFreeLinkedList<Listener<E>>() {
     @Suppress("UNCHECKED_CAST", "UNSUPPORTED", "NO_REFLECTION_IN_CLASS_PATH")
     actual val supertypes: Set<KClass<out Event>> by lazy {
         val supertypes = mutableSetOf<KClass<out Event>>()
 
-        fun addSupertypes(clazz: KClass<out Event>) {
-            clazz.supertypes.forEach {
+        fun addSupertypes(klass: KClass<out Event>) {
+            klass.supertypes.forEach {
                 val classifier = it.classifier as? KClass<out Event>
                 if (classifier != null) {
                     supertypes.add(classifier)
@@ -43,4 +71,6 @@ internal actual class EventListeners<E : Event> actual constructor(clazz: KClass
 
         supertypes
     }
-}
\ No newline at end of file
+
+}
+ */
\ No newline at end of file

mirai-core/src/jvmTest/kotlin/net/mamoe/mirai/event/EventTests.kt

@@ -9,9 +9,14 @@
 
 package net.mamoe.mirai.event
 
-import kotlinx.coroutines.CompletableJob
-import kotlinx.coroutines.GlobalScope
+import kotlinx.atomicfu.AtomicInt
+import kotlinx.atomicfu.atomic
+import kotlinx.coroutines.*
+import net.mamoe.mirai.utils.StepUtil
 import net.mamoe.mirai.utils.internal.runBlocking
+import java.util.concurrent.Executor
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.concurrent.thread
 import kotlin.test.Test
 import kotlin.test.assertTrue
 
@@ -43,6 +48,96 @@ class EventTests {
         }
     }
 
+    @Test
+    fun `test concurrent listening`() {
+        var listeners = 0
+        val counter = AtomicInteger(0)
+        for (p in Listener.EventPriority.values()) {
+            repeat(2333) {
+                listeners++
+                GlobalScope.subscribeAlways<ParentEvent> {
+                    counter.getAndIncrement()
+                }
+            }
+        }
+        kotlinx.coroutines.runBlocking {
+            ParentEvent().broadcast()
+            delay(5000L) // ?
+        }
+        val called = counter.get()
+        println("Registered $listeners listeners and $called called")
+        if (listeners != called) {
+            throw IllegalStateException("Registered $listeners listeners but only $called called")
+        }
+    }
+
+    @Test
+    fun `test concurrent listening 3`() {
+        runBlocking {
+            val called = AtomicInteger()
+            val registered = AtomicInteger()
+            coroutineScope {
+                println("Step 0")
+                for (priority in Listener.EventPriority.values()) {
+                    launch {
+                        repeat(5000) {
+                            registered.getAndIncrement()
+                            GlobalScope.subscribeAlways<ParentEvent>(
+                                priority = priority
+                            ) {
+                                called.getAndIncrement()
+                            }
+                        }
+                        println("Registeterd $priority")
+                    }
+                }
+                println("Step 1")
+            }
+            println("Step 2")
+            ParentEvent().broadcast()
+            println("Step 3")
+            check(called.get() == registered.get())
+            println("Done")
+            println("Called ${called.get()}, registered ${registered.get()}")
+        }
+    }
+
+    @Test
+    fun `test concurrent listening 2`() {
+        val registered = AtomicInteger()
+        val called = AtomicInteger()
+        val threads = mutableListOf<Thread>()
+        repeat(50) {
+            threads.add(thread {
+                repeat(444) {
+                    registered.getAndIncrement()
+                    GlobalScope.launch {
+                        subscribeAlways<ParentEvent> {
+                            called.getAndIncrement()
+                        }
+                    }
+                }
+            })
+        }
+        Thread.sleep(5000L)// Wait all thread started.
+        threads.forEach {
+            it.join() // Wait all finished
+        }
+        println("All listeners registered")
+        val postCount = 3
+        kotlinx.coroutines.runBlocking {
+            repeat(postCount) {
+                ParentEvent().broadcast()
+            }
+            delay(5000L)
+        }
+        val calledCount = called.get()
+        val shouldCalled = registered.get() * postCount
+        println("Should call $shouldCalled times and $called called")
+        if (shouldCalled != calledCount) {
+            throw IllegalStateException("?")
+        }
+    }
 
     open class ParentEvent : Event, AbstractEvent() {
         var triggered = false
@@ -76,4 +171,111 @@ class EventTests {
             job.complete()
         }
     }
+
+    open class PriorityTestEvent : AbstractEvent() {}
+
+    fun singleThreaded(step: StepUtil, invoke: suspend CoroutineScope.() -> Unit) {
+        // runBlocking 会完全堵死, 没法退出
+        val scope = CoroutineScope(Executor { it.run() }.asCoroutineDispatcher())
+        val job = scope.launch {
+            invoke(scope)
+        }
+        kotlinx.coroutines.runBlocking {
+            job.join()
+        }
+        step.throws()
+    }
+
+    @Test
+    fun `test handler remvoe`() {
+        val step = StepUtil()
+        singleThreaded(step) {
+            subscribe<Event> {
+                step.step(0)
+                ListeningStatus.STOPPED
+            }
+            ParentEvent().broadcast()
+            ParentEvent().broadcast()
+        }
+    }
+
+    /*
+    @Test
+    fun `test boom`() {
+        val step = StepUtil()
+        singleThreaded(step) {
+            step.step(0)
+            step.step(0)
+        }
+    }
+    */
+
+    @Test
+    fun `test intercept with always`() {
+        val step = StepUtil()
+        singleThreaded(step) {
+            subscribeAlways<ParentEvent> {
+                step.step(0)
+                intercept()
+            }
+            subscribe<Event> {
+                step.step(-1, "Boom")
+                ListeningStatus.LISTENING
+            }
+            ParentEvent().broadcast()
+        }
+    }
+
+    @Test
+    fun `test intercept`() {
+        val step = StepUtil()
+        singleThreaded(step) {
+            subscribeAlways<AbstractEvent> {
+                step.step(0)
+                intercept()
+            }
+            subscribe<Event> {
+                step.step(-1, "Boom")
+                ListeningStatus.LISTENING
+            }
+            ParentEvent().broadcast()
+        }
+    }
+
+    @Test
+    fun `test listener complete`() {
+        val step = StepUtil()
+        singleThreaded(step) {
+            val listener = subscribeAlways<ParentEvent> {
+                step.step(0, "boom!")
+            }
+            ParentEvent().broadcast()
+            listener.complete()
+            ParentEvent().broadcast()
+        }
+    }
+
+    @Test
+    fun `test event priority`() {
+        val step = StepUtil()
+        singleThreaded(step) {
+            subscribe<PriorityTestEvent> {
+                step.step(1)
+                ListeningStatus.LISTENING
+            }
+            subscribe<PriorityTestEvent>(priority = Listener.EventPriority.HIGH) {
+                step.step(0)
+                ListeningStatus.LISTENING
+            }
+            subscribe<PriorityTestEvent>(priority = Listener.EventPriority.LOW) {
+                step.step(3)
+                ListeningStatus.LISTENING
+            }
+            subscribe<PriorityTestEvent> {
+                step.step(2)
+                ListeningStatus.LISTENING
+            }
+            PriorityTestEvent().broadcast()
+        }
+    }
 }
\ No newline at end of file

mirai-core/src/jvmTest/kotlin/net/mamoe/mirai/utils/StepUtil.kt

+package net.mamoe.mirai.utils
+
+import kotlinx.atomicfu.atomic
+import java.util.concurrent.ConcurrentLinkedDeque
+
+class StepUtil {
+    val step = atomic(0)
+    val exceptions = ConcurrentLinkedDeque<Throwable>()
+    fun step(step: Int, message: String = "Wrong step") {
+        println("Invoking step $step")
+        if (step != this.step.getAndIncrement()) {
+            throw IllegalStateException(message).also { exceptions.add(it) }
+        }
+    }
+
+    fun throws() {
+        if (exceptions.isEmpty()) return
+        val root = exceptions.poll()!!
+        while (true) {
+            root.addSuppressed(exceptions.poll() ?: throw root)
+        }
+    }
+}
\ No newline at end of file