fix: thread starvation in DungeonListener from sending room data to WS (

#502) * fix: thread starvation checking size/not empty on ConcurrentLinkedQueue is not constant time and requires either looping or grabbing first element
Skytils · Oct 2, 2024 · 2b0cb7b · 2b0cb7b
1 parent b524118
commit 2b0cb7b
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 8 deletions.
diff --git a/...ain/kotlin/gg/skytils/skytilsmod/features/impl/dungeons/catlas/handlers/DungeonScanner.kt b/...ain/kotlin/gg/skytils/skytilsmod/features/impl/dungeons/catlas/handlers/DungeonScanner.kt
@@ -18,6 +18,8 @@
 
 package gg.skytils.skytilsmod.features.impl.dungeons.catlas.handlers
 
+import gg.essential.universal.UChat
+import gg.skytils.skytilsmod.Skytils
 import gg.skytils.skytilsmod.Skytils.Companion.IO
 import gg.skytils.skytilsmod.Skytils.Companion.mc
 import gg.skytils.skytilsmod.features.impl.dungeons.DungeonFeatures.dungeonFloorNumber
@@ -26,6 +28,7 @@ import gg.skytils.skytilsmod.features.impl.dungeons.catlas.handlers.DungeonScann
 import gg.skytils.skytilsmod.features.impl.dungeons.catlas.utils.ScanUtils
 import gg.skytils.skytilsmod.listeners.DungeonListener
 import gg.skytils.skytilsmod.utils.SBInfo
+import gg.skytils.skytilsmod.utils.printDevMessage
 import gg.skytils.skytilsws.shared.packet.C2SPacketDungeonRoom
 import kotlinx.coroutines.launch
 import net.minecraft.init.Blocks
@@ -80,7 +83,13 @@ object DungeonScanner {
                     DungeonInfo.dungeonList[z * 11 + x] = it
                     if (it is Room && it.data.name != "Unknown") {
                         SBInfo.server?.let { server ->
-                            DungeonListener.outboundRoomQueue.add(C2SPacketDungeonRoom(server, it.data.name, xPos, zPos, x, z, it.core, it.isSeparator))
+                            printDevMessage("Sending room data to channel: ${it.data.name}", "dungeonws")
+                            val result = DungeonListener.outboundRoomQueue.trySend(
+                                C2SPacketDungeonRoom(server, it.data.name, xPos, zPos, x, z, it.core, it.isSeparator)
+                            )
+                            if (result.isFailure) {
+                                UChat.chat("${Skytils.failPrefix} §cFailed to send room data to server. ${result.isClosed}")
+                            }
                         }
                     }
                 }

diff --git a/src/main/kotlin/gg/skytils/skytilsmod/listeners/DungeonListener.kt b/src/main/kotlin/gg/skytils/skytilsmod/listeners/DungeonListener.kt
@@ -32,6 +32,7 @@ import gg.skytils.skytilsmod.core.API
 import gg.skytils.skytilsmod.core.tickTimer
 import gg.skytils.skytilsmod.events.impl.MainReceivePacketEvent
 import gg.skytils.skytilsmod.events.impl.skyblock.DungeonEvent
+import gg.skytils.skytilsmod.events.impl.skyblock.LocationChangeEvent
 import gg.skytils.skytilsmod.features.impl.dungeons.DungeonFeatures
 import gg.skytils.skytilsmod.features.impl.dungeons.DungeonTimer
 import gg.skytils.skytilsmod.features.impl.dungeons.ScoreCalculation
@@ -52,7 +53,10 @@ import gg.skytils.skytilsws.shared.packet.C2SPacketDungeonEnd
 import gg.skytils.skytilsws.shared.packet.C2SPacketDungeonRoom
 import gg.skytils.skytilsws.shared.packet.C2SPacketDungeonRoomSecret
 import gg.skytils.skytilsws.shared.packet.C2SPacketDungeonStart
+import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
+import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import net.hypixel.modapi.packet.impl.clientbound.ClientboundPartyInfoPacket
@@ -64,7 +68,7 @@ import net.minecraftforge.client.event.ClientChatReceivedEvent
 import net.minecraftforge.event.world.WorldEvent
 import net.minecraftforge.fml.common.eventhandler.EventPriority
 import net.minecraftforge.fml.common.eventhandler.SubscribeEvent
-import java.util.concurrent.ConcurrentLinkedQueue
+import kotlin.jvm.optionals.getOrNull
 
 object DungeonListener {
     val team = hashMapOf<String, DungeonTeammate>()
@@ -114,7 +118,7 @@ object DungeonListener {
     private val keyPickupRegex = Regex("§r§e§lRIGHT CLICK §r§7on §r§7.+?§r§7 to open it\\. This key can only be used to open §r§a(?<num>\\d+)§r§7 door!§r")
     private val witherDoorOpenedRegex = Regex("^(?:\\[.+?] )?(?<name>\\w+) opened a WITHER door!$")
     private const val bloodOpenedString = "§r§cThe §r§c§lBLOOD DOOR§r§c has been opened!§r"
-    val outboundRoomQueue = ConcurrentLinkedQueue<C2SPacketDungeonRoom>()
+    var outboundRoomQueue = Channel<C2SPacketDungeonRoom>(UNLIMITED)
     var isSoloDungeon = false
 
     @SubscribeEvent
@@ -125,10 +129,22 @@ object DungeonListener {
         missingPuzzles.clear()
         completedPuzzles.clear()
         teamCached.clear()
-        outboundRoomQueue.clear()
+        printDevMessage("closed room queue world load", "dungeonws")
+        outboundRoomQueue.cancel()
         isSoloDungeon = false
     }
 
+    @SubscribeEvent
+    fun onLocationUpdate(event: LocationChangeEvent) {
+        if (event.packet.mode.getOrNull() == "dungeon") {
+            printDevMessage("closed room queue", "dungeonws")
+            outboundRoomQueue.cancel()
+            outboundRoomQueue = Channel(UNLIMITED) {
+                printDevMessage("failed to deliver $it", "dungeonws")
+            }
+        }
+    }
+
     @SubscribeEvent
     fun onPacket(event: MainReceivePacketEvent<*, *>) {
         if (!Utils.inDungeons) return
@@ -213,19 +229,25 @@ object DungeonListener {
                             }
                             val partyMembers = party.await().members.ifEmpty { setOf(mc.thePlayer.uniqueID) }.mapTo(hashSetOf()) { it.toString() }
                             val entrance = DungeonInfo.uniqueRooms.first { it.mainRoom.data.type == RoomType.ENTRANCE }
-                            async(WSClient.wsClient.coroutineContext) {
+                            printDevMessage("hi", "dungeonws")
+                            launch(IO.coroutineContext) {
                                 WSClient.sendPacketAsync(C2SPacketDungeonStart(
-                                    serverId = SBInfo.server ?: return@async,
+                                    serverId = SBInfo.server ?: return@launch,
                                     floor = DungeonFeatures.dungeonFloor!!,
                                     members = partyMembers,
                                     startTime = DungeonTimer.dungeonStartTime,
                                     entranceLoc = entrance.mainRoom.z * entrance.mainRoom.x
                                 ))
                                 while (DungeonTimer.dungeonStartTime != -1L) {
-                                    while (outboundRoomQueue.isNotEmpty()) {
-                                        val packet = outboundRoomQueue.poll() ?: continue
+                                    for (packet in outboundRoomQueue) {
                                         WSClient.sendPacketAsync(packet)
+                                        printDevMessage(packet.toString(), "dungeonws")
                                     }
+                                    printDevMessage("escaped loop", "dungeonws")
+                                }
+                            }.also {
+                                it.invokeOnCompletion {
+                                    printDevMessage("loop exit $it", "dungeonws")
                                 }
                             }
                         }