From 895444f1a80db83d74ab75ff8981ff2c8ebe8e22 Mon Sep 17 00:00:00 2001 From: vddcore <573729-vddcore@users.noreply.gitlab.com> Date: Sat, 16 Jul 2022 07:36:44 +0000 Subject: [PATCH] Refactored pulse running logic, made it easier to follow and debug Made the pulse runner use thread-safe collections instead of synchronizing on a thread-unsafe list --- Server/src/main/kotlin/gui/ServerMonitor.kt | 12 +- Server/src/main/kotlin/rs09/Server.kt | 48 ++++---- .../src/main/kotlin/rs09/ServerConstants.kt | 3 + .../game/system/config/ServerConfigParser.kt | 3 +- .../kotlin/rs09/game/world/PulseRunner.kt | 116 ++++++------------ .../kotlin/rs09/worker/MajorUpdateWorker.kt | 48 ++------ Server/src/test/kotlin/TestUtils.kt | 4 +- .../src/test/kotlin/core/PathfinderTests.kt | 7 +- 8 files changed, 88 insertions(+), 153 deletions(-) diff --git a/Server/src/main/kotlin/gui/ServerMonitor.kt b/Server/src/main/kotlin/gui/ServerMonitor.kt index b289ccd8a..061ed1837 100644 --- a/Server/src/main/kotlin/gui/ServerMonitor.kt +++ b/Server/src/main/kotlin/gui/ServerMonitor.kt @@ -1,12 +1,10 @@ package gui -import rs09.game.system.SystemLogger import rs09.game.world.GameWorld import java.awt.BorderLayout import java.awt.Component import java.awt.Dimension import javax.swing.* -import javax.swing.border.Border object ServerMonitor : JFrame("Server Monitor"){ @@ -134,7 +132,7 @@ object ServerMonitor : JFrame("Server Monitor"){ fun open() { isVisible = true - Thread() { + Thread { while (true) { SwingUtilities.invokeLater { val removeList = ArrayList() @@ -159,8 +157,12 @@ object ServerMonitor : JFrame("Server Monitor"){ removeList.add(event) } - pulseList.text = GameWorld.Pulser.TASKS.toArray().filterNotNull().map { it.javaClass.simpleName + "\n" } - .filter { it.replace("\n", "").isNotEmpty() }.sorted().joinToString("") + pulseList.text = GameWorld.Pulser.currentPulses + .map { it.javaClass.simpleName + "\n" } + .filter { it.replace("\n", "").isNotEmpty() } + .sorted() + .joinToString("") + eventQueue.removeAll(removeList) repaint() } diff --git a/Server/src/main/kotlin/rs09/Server.kt b/Server/src/main/kotlin/rs09/Server.kt index c5b9c352b..57d3bb213 100644 --- a/Server/src/main/kotlin/rs09/Server.kt +++ b/Server/src/main/kotlin/rs09/Server.kt @@ -3,16 +3,9 @@ package rs09 import core.game.system.SystemManager import core.game.system.SystemState import core.game.system.mysql.SQLManager -import core.gui.ConsoleFrame import core.net.NioReactor -import core.net.amsc.WorldCommunicator import core.tools.TimeStamp -import kotlinx.coroutines.GlobalScope -import kotlinx.coroutines.delay -import kotlinx.coroutines.launch -import rs09.game.content.global.GlobalKillCounter; -import rs09.game.ge.GEAutoStock -import rs09.game.ge.GEDB +import kotlinx.coroutines.* import rs09.game.system.SystemLogger import rs09.game.system.config.ServerConfigParser import rs09.game.world.GameWorld @@ -97,22 +90,33 @@ object Server { } } } - GlobalScope.launch { - delay(20000) - while(running){ - if(System.currentTimeMillis() - lastHeartbeat > 7200 && running){ - SystemLogger.logErr("Triggering reboot due to heartbeat timeout") - SystemLogger.logErr("Creating thread dump...") - val dump = threadDump(true, true) - FileWriter("latestdump.txt").use { - it.write(dump) - it.flush() - it.close() + + if (ServerConstants.WATCHDOG_ENABLED) { + GlobalScope.launch { + delay(20000) + while (running) { + if (System.currentTimeMillis() - lastHeartbeat > 7200 && running) { + SystemLogger.logErr("Triggering reboot due to heartbeat timeout") + SystemLogger.logErr("Creating thread dump...") + val dump = threadDump(true, true) + + withContext(Dispatchers.IO) { + FileWriter("latestdump.txt").use { + + if (dump != null) { + it.write(dump) + } + + it.flush() + it.close() + } + } + + if (!SystemManager.isTerminated()) + exitProcess(0) } - if(!SystemManager.isTerminated()) - exitProcess(0) + delay(625) } - delay(625) } } } diff --git a/Server/src/main/kotlin/rs09/ServerConstants.kt b/Server/src/main/kotlin/rs09/ServerConstants.kt index 493a38223..b0ab26f70 100644 --- a/Server/src/main/kotlin/rs09/ServerConstants.kt +++ b/Server/src/main/kotlin/rs09/ServerConstants.kt @@ -134,6 +134,9 @@ class ServerConstants { @JvmField var RULES_AND_INFO_ENABLED = true + @JvmField + var WATCHDOG_ENABLED = true + //location names for the ::to command. val TELEPORT_DESTINATIONS = arrayOf( arrayOf(Location.create(2974, 4383, 2), "corp", "corporal", "corporeal"), diff --git a/Server/src/main/kotlin/rs09/game/system/config/ServerConfigParser.kt b/Server/src/main/kotlin/rs09/game/system/config/ServerConfigParser.kt index 5ab5ba58b..ab551a544 100644 --- a/Server/src/main/kotlin/rs09/game/system/config/ServerConfigParser.kt +++ b/Server/src/main/kotlin/rs09/game/system/config/ServerConfigParser.kt @@ -2,9 +2,7 @@ package rs09.game.system.config import com.moandjiezana.toml.Toml import core.game.world.map.Location -import core.tools.StringUtils import core.tools.mysql.Database -import rs09.JSONUtils.Companion.parsePath import rs09.ServerConstants import rs09.game.system.SystemLogger import rs09.game.world.GameSettings @@ -121,6 +119,7 @@ object ServerConfigParser { ServerConstants.BANK_BOOTH_NOTE_ENABLED = data.getBoolean("world.bank_booth_note_enabled", true) ServerConstants.BANK_BOOTH_NOTE_UIM = data.getBoolean("world.bank_booth_note_uim", true) ServerConstants.DISCORD_GE_WEBHOOK = data.getString("server.discord_webhook", "") + ServerConstants.WATCHDOG_ENABLED = data.getBoolean("server.watchdog_enabled", true) } diff --git a/Server/src/main/kotlin/rs09/game/world/PulseRunner.kt b/Server/src/main/kotlin/rs09/game/world/PulseRunner.kt index 67c00794f..242cfd1e6 100644 --- a/Server/src/main/kotlin/rs09/game/world/PulseRunner.kt +++ b/Server/src/main/kotlin/rs09/game/world/PulseRunner.kt @@ -1,99 +1,61 @@ package rs09.game.world import core.game.system.task.Pulse -import java.util.* - -/** new way of running pulses that multithreads based on core count automatically, should improve performance drastically. - * @author ceik - * @author angle - */ +import rs09.game.ai.general.GeneralBotCreator +import rs09.game.system.SystemLogger +import java.util.concurrent.LinkedBlockingQueue class PulseRunner { - val TASKS = ArrayList() + private val pulses = LinkedBlockingQueue() - fun submit(pulse: Pulse){ - synchronized(this.TASKS) { - TASKS.add(pulse) - } - } -} -/* -class PulseeRunner { - val MAXIMUM_NUM_THREADS = 4 - val TARGET_PULSES_PER_THREAD = 100 - val ThreadPool = Executors.newFixedThreadPool(MAXIMUM_NUM_THREADS - 1) as ThreadPoolExecutor - val TASKS: MutableList = ArrayList() - var cores = Runtime.getRuntime().availableProcessors() - var EXECUTOR = Executors.newSingleThreadScheduledExecutor() - var lastPulse: Pulse? = null - fun init(tickTimeMS: Int) { - EXECUTOR.scheduleAtFixedRate(Runner(), 1200, tickTimeMS.toLong(), TimeUnit.MILLISECONDS) - } + val currentPulses: Array get() = pulses.toTypedArray() fun submit(pulse: Pulse) { - TASKS.add(pulse) + pulses.add(pulse) } - inner class Runner : Runnable { - override fun run() { - val currTime = System.nanoTime() - var pulses: MutableList? = null - pulses = ArrayList(TASKS) - val pulseArray: Array = pulses.toTypedArray() - */ -/*var numThreads = 1 + pulseArray.size / TARGET_PULSES_PER_THREAD - if (numThreads > MAXIMUM_NUM_THREADS) numThreads = MAXIMUM_NUM_THREADS - val nowTime = System.nanoTime() - // Execute all the tasks not run on the first core - for (i in 1 until numThreads) { - val pulsesLengthStart = floor(pulseArray.size / numThreads * i.toDouble()).toInt() - var pulsesLengthEnd = floor(pulseArray.size / numThreads * (i + 1).toDouble()).toInt() - if (i + 1 == numThreads) pulsesLengthEnd = pulseArray.size - ThreadPool.execute(PulseThread(pulsesLengthStart, pulsesLengthEnd, pulseArray)) - } + fun updateAll() { + val pulseCount = pulses.size + for (i in 0..pulseCount) { + val pulse = pulses.take() - // Execute the first core tasks all together just as before - val pulsesLengthStart = floor(pulseArray.size / numThreads.toDouble()).toInt()*//* - - - for (element in pulseArray) { - val pulse = element as Pulse? ?: continue + val elapsedTime = measure { try { - if (TASKS.contains(pulse)) { - lastPulse = pulse - if (pulse.update()) { - TASKS.remove(pulse) - } + if (!pulse.update()) { + pulses.add(pulse) } - } catch (t: Throwable) { - t.printStackTrace() - pulse.stop() - TASKS.remove(pulse) + } catch (e: Exception) { + SystemLogger.logErr("Pulse execution error. Stack trace below.") + e.printStackTrace() } } - pulses.clear() + + notifyIfTooLong(pulse, elapsedTime) } } - inner class PulseThread(var threadStart: Int, var threadFinish: Int, var pulseArray: Array) : Runnable { - override fun run() { - for (i in threadStart until threadFinish) { - val pulse = pulseArray[i] as Pulse? ?: continue - try { - if (TASKS.contains(pulse)) { - lastPulse = pulse - if (pulse.update()) { - TASKS.remove(pulse) - } - } - } catch (t: Throwable) { - t.printStackTrace() - pulse.stop() - TASKS.remove(pulse) - } + private fun measure(logic: () -> Unit): Long { + val startTime = System.currentTimeMillis() + + logic() + + return System.currentTimeMillis() - startTime + } + + private fun notifyIfTooLong(pulse: Pulse, elapsedTime: Long) { + if (elapsedTime >= 100) { + if (pulse is GeneralBotCreator.BotScriptPulse) { + SystemLogger.logWarn("CRITICALLY long bot-script tick - ${pulse.botScript.javaClass.name} took $elapsedTime ms") + } else { + SystemLogger.logWarn("CRITICALLY long running pulse - ${pulse.javaClass.name} took $elapsedTime ms") + } + } else if (elapsedTime >= 30) { + if (pulse is GeneralBotCreator.BotScriptPulse) { + SystemLogger.logWarn("Long bot-script tick - ${pulse.botScript.javaClass.name} took $elapsedTime ms") + } else { + SystemLogger.logWarn("Long running pulse - ${pulse.javaClass.name} took $elapsedTime ms") } } - } -}*/ +} \ No newline at end of file diff --git a/Server/src/main/kotlin/rs09/worker/MajorUpdateWorker.kt b/Server/src/main/kotlin/rs09/worker/MajorUpdateWorker.kt index 6a3728878..40469cf87 100644 --- a/Server/src/main/kotlin/rs09/worker/MajorUpdateWorker.kt +++ b/Server/src/main/kotlin/rs09/worker/MajorUpdateWorker.kt @@ -6,7 +6,6 @@ import core.plugin.CorePluginTypes.Managers import rs09.Server import rs09.ServerConstants import rs09.ServerStore -import rs09.game.ai.general.GeneralBotCreator import rs09.game.system.SystemLogger import rs09.game.world.GameWorld import rs09.game.world.repository.Repository @@ -30,13 +29,13 @@ class MajorUpdateWorker { Thread.currentThread().name = "Major Update Worker" started = true Thread.sleep(600L) - while(running){ + while (running) { val start = System.currentTimeMillis() Server.heartbeat() handleTickActions() - for (player in Repository.players.filter {!it.isArtificial}) { + for (player in Repository.players.filter { !it.isArtificial }) { if (System.currentTimeMillis() - player.session.lastPing > 20000L) { player?.details?.session?.disconnect() player?.session?.lastPing = Long.MAX_VALUE @@ -46,14 +45,14 @@ class MajorUpdateWorker { } //Handle daily restart if enabled - if(sdf.format(Date()).toInt() == 0){ + if (sdf.format(Date()).toInt() == 0) { - if(GameWorld.checkDay() == 1) {//monday + if (GameWorld.checkDay() == 1) {//monday ServerStore.clearWeeklyEntries() } ServerStore.clearDailyEntries() - if(ServerConstants.DAILY_RESTART ) { + if (ServerConstants.DAILY_RESTART) { Repository.sendNews(colorize("%RSERVER GOING DOWN FOR DAILY RESTART IN 5 MINUTES!")) ServerConstants.DAILY_RESTART = false submitWorldPulse(object : Pulse(100) { @@ -79,40 +78,11 @@ class MajorUpdateWorker { SystemLogger.logInfo("Update worker stopped.") } - fun handleTickActions() { - val rmlist = ArrayList() - val list = ArrayList(GameWorld.Pulser.TASKS) - //run our pulses - for (pulse in list) { - val b = System.currentTimeMillis() - if (pulse == null || pulse.update()) rmlist.add(pulse) - - val time = System.currentTimeMillis() - b - - if (time >= 100) { - if (pulse is GeneralBotCreator.BotScriptPulse) { - SystemLogger.logWarn("CRITICALLY Long Botscript Tick: ${pulse.botScript.javaClass.name} - $time ms") - } else { - SystemLogger.logWarn("CRITICALLY long running pulse: ${pulse.javaClass.name} - $time ms") - } - } else if (time >= 30) { - if (pulse is GeneralBotCreator.BotScriptPulse) { - SystemLogger.logWarn("Long Botscript Tick: ${pulse.botScript.javaClass.name} - $time ms") - } else { - SystemLogger.logWarn("Long Running Pulse: ${pulse.javaClass.name} - $time ms") - } - } + fun handleTickActions(skipPulseUpdate: Boolean = false) { + if (!skipPulseUpdate) { + GameWorld.Pulser.updateAll() } - //remove all null or finished pulses from the list - rmlist.forEach { - synchronized(GameWorld.Pulser.TASKS) { - GameWorld.Pulser.TASKS.remove(it) - } - } - - rmlist.clear() - //perform our update sequence where we write masks, etc try { sequence.start() sequence.run() @@ -130,7 +100,7 @@ class MajorUpdateWorker { } fun start() { - if(!started){ + if (!started) { running = true worker.start() } diff --git a/Server/src/test/kotlin/TestUtils.kt b/Server/src/test/kotlin/TestUtils.kt index 05f11ebd5..ec337bc91 100644 --- a/Server/src/test/kotlin/TestUtils.kt +++ b/Server/src/test/kotlin/TestUtils.kt @@ -55,10 +55,10 @@ object TestUtils { } } - fun advanceTicks(amount: Int) { + fun advanceTicks(amount: Int, skipPulseUpdates: Boolean = true) { SystemLogger.logInfo("Advancing ticks by $amount.") for(i in 0 until amount) { - GameWorld.majorUpdateWorker.handleTickActions() + GameWorld.majorUpdateWorker.handleTickActions(skipPulseUpdates) } } } diff --git a/Server/src/test/kotlin/core/PathfinderTests.kt b/Server/src/test/kotlin/core/PathfinderTests.kt index 519902a66..7262903eb 100644 --- a/Server/src/test/kotlin/core/PathfinderTests.kt +++ b/Server/src/test/kotlin/core/PathfinderTests.kt @@ -1,9 +1,6 @@ package core import TestUtils -import api.replaceScenery -import core.cache.def.impl.SceneryDefinition -import core.game.interaction.MovementPulse import core.game.node.scenery.Scenery import core.game.world.map.Location import core.game.world.map.RegionManager @@ -12,8 +9,6 @@ import org.junit.jupiter.api.Test import rs09.game.interaction.InteractionListener import rs09.game.interaction.InteractionListeners import rs09.game.node.entity.skill.gather.GatheringSkillOptionListeners -import rs09.game.system.SystemLogger -import rs09.game.world.GameWorld class PathfinderTests { companion object {init {TestUtils.preTestSetup(); GatheringSkillOptionListeners().defineListeners() }} @@ -47,7 +42,7 @@ class PathfinderTests { p.init() Assertions.assertEquals(true, InteractionListeners.run(1307, InteractionListener.SCENERY, "chop-down", p, dest!!)) - TestUtils.advanceTicks(20) + TestUtils.advanceTicks(20, false) Assertions.assertEquals(Location.create(2722, 3475, 0), p.location) } } \ No newline at end of file