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
2022-07-16 07:36:44 +00:00
parent c7c97ea176
commit 895444f1a8
8 changed files with 88 additions and 153 deletions
@@ -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<GuiEvent>()
@@ -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()
                }
@@ -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)
            }
        }
    }
@@ -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"),
@@ -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)
    }


@@ -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<Pulse>()
+    private val pulses = LinkedBlockingQueue<Pulse>()

-    fun submit(pulse: Pulse){
-        synchronized(this.TASKS) {
-            TASKS.add(pulse)
+    val currentPulses: Array<Pulse> get() = pulses.toTypedArray()
+
+    fun submit(pulse: Pulse) {
+        pulses.add(pulse)
+    }
+
+    fun updateAll() {
+        val pulseCount = pulses.size
+
+        for (i in 0..pulseCount) {
+            val pulse = pulses.take()
+
+            val elapsedTime = measure {
+                try {
+                    if (!pulse.update()) {
+                        pulses.add(pulse)
+                    }
+                } catch (e: Exception) {
+                    SystemLogger.logErr("Pulse execution error. Stack trace below.")
+                    e.printStackTrace()
+                }
+            }
+
+            notifyIfTooLong(pulse, elapsedTime)
+        }
+    }
+
+    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")
+            }
        }
    }
 }
-/*
-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<Pulse> = 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)
-    }
-
-    fun submit(pulse: Pulse) {
-        TASKS.add(pulse)
-    }
-
-    inner class Runner : Runnable {
-        override fun run() {
-            val currTime = System.nanoTime()
-            var pulses: MutableList<Pulse>? = null
-            pulses = ArrayList(TASKS)
-            val pulseArray: Array<Any?> = 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))
-            }
-
-
-            // 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
-                try {
-                    if (TASKS.contains(pulse)) {
-                        lastPulse = pulse
-                        if (pulse.update()) {
-                            TASKS.remove(pulse)
-                        }
-                    }
-                } catch (t: Throwable) {
-                    t.printStackTrace()
-                    pulse.stop()
-                    TASKS.remove(pulse)
-                }
-            }
-            pulses.clear()
-        }
-    }
-
-    inner class PulseThread(var threadStart: Int, var threadFinish: Int, var pulseArray: Array<Any?>) : 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)
-                }
-            }
-        }
-
-    }
-}*/
@@ -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<Pulse>()
-        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()
        }
@@ -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)
        }
    }
 }
@@ -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)
    }
 }