diff options
Diffstat (limited to 'platform-paper')
8 files changed, 199 insertions, 17 deletions
diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt index d51a254..3041759 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/LunaticChat.kt @@ -44,6 +44,10 @@ class LunaticChat : // Read by commands that must not block the tick thread. lateinit var pluginScope: PluginCoroutineScope private set + + /** Serializes each player's outgoing messages so they arrive in the order they were sent. */ + lateinit var deliveryQueue: PerPlayerWorkQueue + private set private var updateChecker: UpdateChecker? = null private val updateAvailable = AtomicBoolean(false) @@ -64,6 +68,7 @@ class LunaticChat : // Initialize plugin coroutine scope pluginScope = PluginCoroutineScope(logger) + deliveryQueue = PerPlayerWorkQueue(pluginScope.scope) // Initialize all services serviceInitializer = diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt new file mode 100644 index 0000000..392eea8 --- /dev/null +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueue.kt @@ -0,0 +1,54 @@ +package dev.m1sk9.lunaticChat.paper + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.SendChannel +import kotlinx.coroutines.launch +import java.util.UUID +import java.util.concurrent.ConcurrentHashMap + +/** + * Runs work submitted for a player one item at a time, in the order it was submitted. + * + * Launching a coroutine per message would let a fast one overtake a slow one - a cached romaji + * conversion finishing ahead of an uncached one sent before it - so a player's messages could + * appear out of order to themselves and to their recipient. A queue per player keeps each player's + * messages ordered while still letting different players proceed independently. + * + * [submit] must be called from one thread per player (the server's command thread), since that is + * what makes "the order it was submitted" well defined. + */ +class PerPlayerWorkQueue( + private val scope: CoroutineScope, +) { + private val queues = ConcurrentHashMap<UUID, SendChannel<suspend () -> Unit>>() + + /** + * Queues [work] behind anything already pending for [playerId]. + */ + fun submit( + playerId: UUID, + work: suspend () -> Unit, + ) { + queues.computeIfAbsent(playerId) { startWorker() }.trySend(work) + } + + /** + * Drops the player's queue once they can no longer send anything. Work already queued still + * runs; without this the map and its worker coroutines would grow for the life of the server. + */ + fun release(playerId: UUID) { + queues.remove(playerId)?.close() + } + + private fun startWorker(): SendChannel<suspend () -> Unit> { + // Unlimited so that submit never suspends or drops work on the caller's thread. + val channel = Channel<suspend () -> Unit>(Channel.UNLIMITED) + scope.launch { + for (work in channel) { + work() + } + } + return channel + } +} diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PluginCoroutineScope.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PluginCoroutineScope.kt index f67d262..22ff0dc 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PluginCoroutineScope.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/PluginCoroutineScope.kt @@ -1,9 +1,11 @@ package dev.m1sk9.lunaticChat.paper +import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel +import java.util.logging.Level import java.util.logging.Logger /** @@ -28,7 +30,16 @@ class PluginCoroutineScope( private val logger: Logger, ) { private val job = SupervisorJob() - val scope = CoroutineScope(Dispatchers.Default + job) + + // Without this, a coroutine that throws reports to the JVM default handler and never reaches + // the plugin's log - and callers that dispatch work and return immediately have no other way + // to learn that it failed. + private val errorHandler = + CoroutineExceptionHandler { _, throwable -> + logger.log(Level.SEVERE, "Unhandled exception in a plugin coroutine", throwable) + } + + val scope = CoroutineScope(Dispatchers.Default + job + errorHandler) /** * Cancels all coroutines in this scope. diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt index c530ce9..8f676b3 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/ReplyCommand.kt @@ -5,6 +5,7 @@ import com.mojang.brigadier.builder.LiteralArgumentBuilder import dev.m1sk9.lunaticChat.engine.command.CommandResult import dev.m1sk9.lunaticChat.engine.permission.LunaticChatPermissionNode import dev.m1sk9.lunaticChat.paper.LunaticChat +import dev.m1sk9.lunaticChat.paper.PerPlayerWorkQueue import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler import dev.m1sk9.lunaticChat.paper.chat.handler.ReplyTarget import dev.m1sk9.lunaticChat.paper.command.annotation.Command @@ -16,8 +17,6 @@ import dev.m1sk9.lunaticChat.paper.i18n.LanguageManager import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager import io.papermc.paper.command.brigadier.CommandSourceStack import io.papermc.paper.command.brigadier.Commands -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.launch import org.bukkit.Bukkit @Command( @@ -32,9 +31,10 @@ class ReplyCommand( private val dmHandler: DirectMessageHandler, override val languageManager: LanguageManager, private val crossServerDirectMessageManager: CrossServerDirectMessageManager? = null, - // Delivery is dispatched here rather than run inline: romaji conversion can reach the Google - // IME API, and a command executor runs on the tick thread. - private val scope: CoroutineScope = plugin.pluginScope.scope, + // Delivery is queued rather than run inline: romaji conversion can reach the Google IME API, + // and a command executor runs on the tick thread. Queueing per sender keeps their messages in + // the order they typed them. + private val deliveryQueue: PerPlayerWorkQueue = plugin.deliveryQueue, ) : LunaticCommand(plugin) { override val description: String get() = languageManager.getMessage("commandDescription.reply") @@ -70,14 +70,16 @@ class ReplyCommand( val recipient = Bukkit.getPlayer(target.uuid) ?: return fail("directMessage.replyTargetNotFound") - scope.launch { dmHandler.sendDirectMessage(sender, recipient, message) } + deliveryQueue.submit(sender.uniqueId) { dmHandler.sendDirectMessage(sender, recipient, message) } CommandResult.Success } is ReplyTarget.Remote -> { val manager = crossServerDirectMessageManager ?: return fail("directMessage.replyTargetNotFound") - scope.launch { manager.sendCrossServerMessage(sender, target.playerName, target.serverName, message) } + deliveryQueue.submit(sender.uniqueId) { + manager.sendCrossServerMessage(sender, target.playerName, target.serverName, message) + } CommandResult.Success } } diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt index bd98f22..f19afed 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommand.kt @@ -7,6 +7,7 @@ import com.mojang.brigadier.suggestion.SuggestionsBuilder import dev.m1sk9.lunaticChat.engine.command.CommandResult import dev.m1sk9.lunaticChat.engine.permission.LunaticChatPermissionNode import dev.m1sk9.lunaticChat.paper.LunaticChat +import dev.m1sk9.lunaticChat.paper.PerPlayerWorkQueue import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler import dev.m1sk9.lunaticChat.paper.command.annotation.Command import dev.m1sk9.lunaticChat.paper.command.annotation.Permission @@ -18,8 +19,6 @@ import dev.m1sk9.lunaticChat.paper.velocity.CrossServerDirectMessageManager import dev.m1sk9.lunaticChat.paper.velocity.RemotePlayerRegistry import io.papermc.paper.command.brigadier.CommandSourceStack import io.papermc.paper.command.brigadier.Commands -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.launch import org.bukkit.Bukkit import org.bukkit.entity.Player import java.util.concurrent.CompletableFuture @@ -39,9 +38,10 @@ class TellCommand( private val crossServerDirectMessageManager: CrossServerDirectMessageManager? = null, private val remotePlayerRegistry: RemotePlayerRegistry? = null, private val localServerName: String = "", - // Delivery is dispatched here rather than run inline: romaji conversion can reach the Google - // IME API, and a command executor runs on the tick thread. - private val scope: CoroutineScope = plugin.pluginScope.scope, + // Delivery is queued rather than run inline: romaji conversion can reach the Google IME API, + // and a command executor runs on the tick thread. Queueing per sender keeps their messages in + // the order they typed them. + private val deliveryQueue: PerPlayerWorkQueue = plugin.deliveryQueue, ) : LunaticCommand(plugin) { override val description: String get() = languageManager.getMessage("commandDescription.tell") @@ -104,7 +104,7 @@ class TellCommand( ) { return fail("directMessage.yourself") } - scope.launch { manager.sendCrossServerMessage(sender, name, server, message) } + deliveryQueue.submit(sender.uniqueId) { manager.sendCrossServerMessage(sender, name, server, message) } return CommandResult.Success } @@ -116,7 +116,7 @@ class TellCommand( return fail("directMessage.yourself") } - scope.launch { directMessageHandler.sendDirectMessage(sender, recipient, message) } + deliveryQueue.submit(sender.uniqueId) { directMessageHandler.sendDirectMessage(sender, recipient, message) } return CommandResult.Success } diff --git a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/listener/PlayerPresenceListener.kt b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/listener/PlayerPresenceListener.kt index 6418f6f..1e3fde9 100644 --- a/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/listener/PlayerPresenceListener.kt +++ b/platform-paper/src/main/kotlin/dev/m1sk9/lunaticChat/paper/listener/PlayerPresenceListener.kt @@ -96,5 +96,8 @@ class PlayerPresenceListener( // 3. Trigger async save of player settings playerSettingsManager.queueSave() + + // 4. Drop their delivery queue; anything already queued still runs + lunaticChat.deliveryQueue.release(playerId) } } diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt new file mode 100644 index 0000000..17341ca --- /dev/null +++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/PerPlayerWorkQueueTest.kt @@ -0,0 +1,106 @@ +package dev.m1sk9.lunaticChat.paper + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import java.util.UUID +import java.util.concurrent.ConcurrentLinkedQueue +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class PerPlayerWorkQueueTest { + private val alice = UUID.fromString("00000001-0000-0000-0000-000000000000") + private val bob = UUID.fromString("00000002-0000-0000-0000-000000000000") + + private suspend fun awaitSize( + completed: ConcurrentLinkedQueue<String>, + expected: Int, + ) = withTimeout(5_000) { + while (completed.size < expected) { + delay(5) + } + } + + @Test + fun `a player's work runs in submission order even when later work is faster`() = + runBlocking { + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val completed = ConcurrentLinkedQueue<String>() + + // The regression this guards: a cached conversion overtaking an uncached one sent + // before it, so the player sees their messages out of order. + queue.submit(alice) { + delay(200) + completed.add("slow-first") + } + queue.submit(alice) { completed.add("fast-second") } + + awaitSize(completed, 2) + assertEquals(listOf("slow-first", "fast-second"), completed.toList()) + } + + @Test + fun `one player's slow work does not hold up another player`() = + runBlocking { + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val completed = ConcurrentLinkedQueue<String>() + + queue.submit(alice) { + delay(1_000) + completed.add("alice") + } + queue.submit(bob) { completed.add("bob") } + + awaitSize(completed, 1) + assertEquals(listOf("bob"), completed.toList()) + } + + @Test + fun `submit does not block the caller`() { + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val started = ConcurrentLinkedQueue<String>() + + queue.submit(alice) { + delay(2_000) + started.add("done") + } + + // submit returns without waiting for the work; the point of queueing off the tick thread. + assertTrue(started.isEmpty()) + } + + @Test + fun `work already queued still runs after the player is released`() = + runBlocking { + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val completed = ConcurrentLinkedQueue<String>() + + queue.submit(alice) { + delay(50) + completed.add("in-flight") + } + queue.release(alice) + + awaitSize(completed, 1) + assertEquals(listOf("in-flight"), completed.toList()) + } + + @Test + fun `a released player gets a fresh queue if they come back`() = + runBlocking { + val queue = PerPlayerWorkQueue(CoroutineScope(Dispatchers.Default)) + val completed = ConcurrentLinkedQueue<String>() + + queue.submit(alice) { completed.add("before") } + awaitSize(completed, 1) + queue.release(alice) + + queue.submit(alice) { completed.add("after") } + + awaitSize(completed, 2) + assertEquals(listOf("before", "after"), completed.toList()) + } +} diff --git a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt index cea9d6f..36b6ff7 100644 --- a/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt +++ b/platform-paper/src/test/kotlin/dev/m1sk9/lunaticChat/paper/command/impl/TellCommandTest.kt @@ -2,6 +2,7 @@ package dev.m1sk9.lunaticChat.paper.command.impl import dev.m1sk9.lunaticChat.engine.command.CommandResult import dev.m1sk9.lunaticChat.paper.LunaticChat +import dev.m1sk9.lunaticChat.paper.PerPlayerWorkQueue import dev.m1sk9.lunaticChat.paper.TestUtils import dev.m1sk9.lunaticChat.paper.chat.handler.DirectMessageHandler import dev.m1sk9.lunaticChat.paper.command.core.CommandContext @@ -20,7 +21,7 @@ class TellCommandTest { private fun createCommand( crossServerManager: CrossServerDirectMessageManager? = null, localServerName: String = "lobby", - // Unconfined runs the dispatched delivery inline, so a test can assert on it right after + // Unconfined runs the queued delivery inline, so a test can assert on it right after // execute() returns. scope: CoroutineScope = CoroutineScope(Dispatchers.Unconfined), ): TellDeps { @@ -42,7 +43,7 @@ class TellCommandTest { crossServerManager, null, localServerName, - scope, + PerPlayerWorkQueue(scope), ) return TellDeps(command, ctx, dmHandler, crossServerManager, sender) } |