Reduce server hitching: dead dispatcher config, per-packet allocation, and unindexed queries

build.sbt

@@ -110,7 +110,18 @@ lazy val server = (project in file("server"))
     inConfig(QuietTest)(Defaults.testTasks),
     packMain := Map("psforever-server" -> "net.psforever.server.Server"),
     packArchivePrefix := "psforever-server",
-    packJvmOpts := Map("psforever-server" -> Seq("-Dstacktrace.app.packages=net.psforever")),
+    // Memory and collector settings for the packaged launcher. Without them the server takes
+    // the JVM defaults, which in a container is a fraction of host RAM. Note the -Xmx in
+    // .jvmopts governs sbt, not this script. G1 with an explicit pause target keeps
+    // collections short, which matters because a stop-the-world pause stalls every connected
+    // session simultaneously.
+    packJvmOpts := Map("psforever-server" -> Seq(
+      "-Dstacktrace.app.packages=net.psforever",
+      "-Xms2g",
+      "-Xmx4g",
+      "-XX:+UseG1GC",
+      "-XX:MaxGCPauseMillis=50"
+    )),
     packExtraClasspath := Map("psforever-server" -> Seq("${PROG_HOME}/config")),
     packResourceDir += ((psforever / baseDirectory).value / "config" -> "config")
   )

docker-compose.yml

@@ -1,6 +1,9 @@
 services:
   loginserver:
-    image: sbtscala/scala-sbt:eclipse-temurin-jammy-8u352-b08_1.8.3_2.13.10
+    # Matches the JDK the Dockerfile builds with, so local runs and released builds share a
+    # collector. Keep this off JDK 8, whose default ParallelGC does stop-the-world full
+    # collections that can run into seconds and stall every connected client at once.
+    image: sbtscala/scala-sbt:eclipse-temurin-focal-11.0.17_8_1.8.2_2.13.10
     volumes:
       - .:/PSF-Loginserver:z
     working_dir: /PSF-Loginserver

server/src/main/resources/db/migration/V018__PerformanceIndexes.sql

@@ -0,0 +1,38 @@
+-- Indexes for foreign keys that are queried on the login and gameplay hot paths.
+--
+-- PostgreSQL does not create an index for a REFERENCES constraint, so each foreign key that
+-- is filtered on needs one declared explicitly or the lookup is a sequential scan.
+--
+-- Deliberately omitted, because an existing constraint already provides a usable btree with
+-- avatar_id as its leading column:
+--   friend, ignored              UNIQUE(avatar_id, char_id)
+--   shortcut                     UNIQUE(avatar_id, slot)
+--   avatarmodepermission         UNIQUE(avatar_id)
+--   savedplayer, savedavatar     avatar_id is the primary key
+-- Adding further indexes there would only cost write throughput.
+
+-- Read for every character on the character select screen, and rewritten whenever a loadout
+-- is saved or deleted. The primary key is id alone, so it does not serve avatar_id lookups.
+CREATE INDEX IF NOT EXISTS "loadout_avatar_id_idx" ON "loadout" ("avatar_id");
+CREATE INDEX IF NOT EXISTS "vehicleloadout_avatar_id_idx" ON "vehicleloadout" ("avatar_id");
+
+-- Read on login and rewritten on every locker change.
+CREATE INDEX IF NOT EXISTS "locker_avatar_id_idx" ON "locker" ("avatar_id");
+
+-- Primary key is (id, avatar_id), so a lookup by avatar_id alone cannot use it.
+CREATE INDEX IF NOT EXISTS "certification_avatar_id_idx" ON "certification" ("avatar_id");
+
+-- Primary key is (name, avatar_id); loading every implant for an avatar does not supply
+-- name and so cannot use the leading column.
+CREATE INDEX IF NOT EXISTS "implant_avatar_id_idx" ON "implant" ("avatar_id");
+
+-- Scanned to build the character list for an account.
+CREATE INDEX IF NOT EXISTS "avatar_account_id_idx" ON "avatar" ("account_id");
+
+-- killactivity is an append-only kill log that grows for the lifetime of the server, and the
+-- campaign KDA query reads it on every single login with
+--   WHERE (killer_id = ? OR victim_id = ?) AND killer_id <> victim_id
+-- Two single-column indexes let the OR resolve as a bitmap union, keeping that query
+-- proportional to one player's kills rather than to every kill the server has recorded.
+CREATE INDEX IF NOT EXISTS "killactivity_killer_id_idx" ON "killactivity" ("killer_id");
+CREATE INDEX IF NOT EXISTS "killactivity_victim_id_idx" ON "killactivity" ("victim_id");

server/src/main/scala/net/psforever/server/Server.scala

@@ -115,7 +115,10 @@ object Server {
     val loginPlan = (ref: ActorRef[MiddlewareActor.Command], info: InetSocketAddress, connectionId: String) => {
       import net.psforever.services.account.IPAddress
       Behaviors.setup[PlanetSidePacket](context => {
-        val actor = context.actorOf(classic.Props(new LoginActor(ref, connectionId, Login.getNewId())), "login")
+        val actor = context.actorOf(
+          classic.Props(new LoginActor(ref, connectionId, Login.getNewId())).withDispatcher("login-session"),
+          "login"
+        )
         actor ! ReceiveIPAddress(new IPAddress(info))
         Behaviors.receiveMessage(message => {
           actor ! message
@@ -127,7 +130,10 @@ object Server {
       Behaviors.setup[PlanetSidePacket](context => {
         val uuid = randomUUID().toString
         val actor =
-          context.actorOf(classic.Props(new SessionActor(ref, connectionId, Session.getNewId())), s"session-$uuid")
+          context.actorOf(
+            classic.Props(new SessionActor(ref, connectionId, Session.getNewId())).withDispatcher("world-session"),
+            s"session-$uuid"
+          )
         Behaviors.receiveMessage(message => {
           actor ! message
           Behaviors.same

src/main/resources/akka.conf

@@ -221,11 +221,11 @@ akka.actor.deployment {
   "/service/cluster/tzdrvs-actor/*" {
     dispatcher = tzdrvs-zone-dispatcher
   }
-  "/service/cluster/tzsdrtr-actor" {
-    dispatcher = tzsdrtr-zone-dispatcher
+  "/service/cluster/tzdrtr-actor" {
+    dispatcher = tzdrtr-zone-dispatcher
   }
-  "/service/cluster/tzsdrtr-actor/*" {
-    dispatcher = tzsdrtr-zone-dispatcher
+  "/service/cluster/tzdrtr-actor/*" {
+    dispatcher = tzdrtr-zone-dispatcher
   }
   "/service/cluster/tzshnc-actor" {
     dispatcher = tzshnc-zone-dispatcher
@@ -246,3 +246,35 @@ akka.actor.deployment {
     dispatcher = tzshvs-zone-dispatcher
   }
 }
+
+# Configuration for the default dispatcher.
+#
+# Anything that does not select a dispatcher explicitly runs here, so it is sized for real
+# server load rather than left at Akka's stock throughput of 5, which produces frequent
+# thread hand-off on the packet path.
+#
+# Note that the path-based assignments above do not reach the network, session or zone
+# actors. Those are spawned through the typed API, which Akka's config-based deployment does
+# not apply to, and they select their dispatchers at their spawn sites instead -- see
+# SocketPane, SocketActor, Server and InterstellarClusterService. The paths named above
+# (/world-udp-endpoint, world-session-router, <id>-actor) do not correspond to actors this
+# server creates.
+akka.actor.default-dispatcher {
+  throughput = 20
+
+  fork-join-executor {
+    parallelism-min = 8
+    parallelism-factor = 1.0
+    parallelism-max = 64
+  }
+}
+
+# The scheduler carries a large number of short-lived tasks: a per-session outbound packet
+# timer, per-player stamina and keep-alive ticks, and per-entity ticks for painboxes,
+# siphons, turrets and proximity terminals. The stock 512 buckets collide heavily at those
+# volumes, and a cancelled task stays in its bucket until its tick comes round, so the tick
+# thread walks long chains.
+akka.scheduler {
+  tick-duration = 10ms
+  ticks-per-wheel = 2048
+}

src/main/resources/application.conf

@@ -63,7 +63,14 @@ database {
   sslmode = prefer
 
   # The maximum number of active connections.
-  maxActiveConnections = 5
+  #
+  # This is the ceiling on concurrent SQL statements for the entire server, not a player
+  # limit -- game clients never touch it. Every session, zone actor and kill report shares
+  # this one pool, and the driver queues excess queries rather than rejecting them, so
+  # saturation shows up purely as latency: logins, loadout applies and terminal purchases
+  # waiting behind each other. Size it against concurrent database work, bearing in mind
+  # character login alone issues about a dozen sequential queries.
+  maxActiveConnections = 32
 }
 
 # Enable non-standard game properties
@@ -553,6 +560,15 @@ network {
     # Affects the base value on the timer
     packet-bundling-delay-multiplier = 1.25
 
+    # How many bundles may be dispatched in a single execution of the bundling process.
+    # Each bundle is capped at the 440 byte MTU, so this multiplied by the bundle size and
+    # divided by the delay above is the effective per-client outbound ceiling.
+    # This governs how quickly a backlog drains -- a zone load, a base capture, or a squad
+    # coming into render range all queue far more than one bundle -- while the delay above
+    # governs how aggressively packets are coalesced into each bundle.
+    # Raise for faster burst delivery, lower to be gentler on constrained connections.
+    packet-bundling-drain-limit = 8
+
     # Pause inbound packet transmission towards the network if the sequence number is out of order
     # Packets are put aside until the sequence is restored, or this timeout passes
     in-reorder-timeout = 50 milliseconds

src/main/resources/dispatchers.conf

@@ -452,7 +452,7 @@ tzdrvs-zone-dispatcher {
   throughput = 50
   throughput-deadline-time = 50ms
 }
-tzsdrtr-zone-dispatcher {
+tzdrtr-zone-dispatcher {
   type = Dispatcher
   executor = "fork-join-executor"
 

src/main/scala/net/psforever/actors/net/MiddlewareActor.scala

@@ -304,6 +304,10 @@ class MiddlewareActor(
     "milliseconds"
   )
 
+  /** How many bundles a single run of the bundling process may dispatch; at least one */
+  private val packetBundlingDrainLimit: Int =
+    math.max(1, Config.app.network.middleware.packetBundlingDrainLimit)
+
   /** Timer that handles the bundling and throttling of outgoing packets and resolves disorganized inbound packets */
   private var packetProcessor: Cancellable = Default.Cancellable
 
@@ -671,7 +675,16 @@ class MiddlewareActor(
     */
   private def processQueue(): Unit = {
     inReorderQueueFunc()
-    processOutQueueBundle()
+    /* Drain up to `packetBundlingDrainLimit` bundles per run. processOutQueueBundle dispatches
+       one bundle per call, so this budget governs how much of a backlog a single run clears,
+       while the bundling delay governs how aggressively packets are coalesced into each
+       bundle. The budget also bounds the loop, which therefore always terminates after a
+       fixed number of iterations even if a queue fails to shrink. */
+    var budget: Int = packetBundlingDrainLimit
+    while (budget > 0 && (outQueueBundled.nonEmpty || outQueue.nonEmpty)) {
+      processOutQueueBundle()
+      budget -= 1
+    }
   }
 
   /**

src/main/scala/net/psforever/actors/net/SocketActor.scala

@@ -142,7 +142,11 @@ class SocketActor(
                   val ref = context.spawn(
                     MiddlewareActor(udpCommandAdapter, remote, nextPlan, connectionId),
                     s"middleware-$connectionId",
-                    ActorTags(s"uuid=$connectionId")
+                    /* crypto, bundling and reordering for one connection, isolated onto the
+                       world-session pool. The dispatcher is selected here at the spawn site
+                       because this is a typed actor, which config-based deployment in
+                       akka.conf does not apply to. */
+                    ActorTags(s"uuid=$connectionId").withDispatcherFromConfig("world-session")
                   )
                   context.watch(ref)
                   packetActors(remote) = ref

src/main/scala/net/psforever/actors/net/SocketPane.scala

@@ -5,7 +5,7 @@ import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
 
 import java.net.{InetAddress, InetSocketAddress}
 import akka.{actor => classic}
-import akka.actor.typed.{ActorRef, Behavior, PostStop}
+import akka.actor.typed.{ActorRef, Behavior, DispatcherSelector, PostStop}
 import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
 import net.psforever.packet.PlanetSidePacket
 
@@ -88,6 +88,13 @@ final case class SocketSetupInfo(
 object SocketPane {
   val SocketPaneKey: ServiceKey[Command] = ServiceKey[SocketPane.Command](id = "socketPane")
 
+  /** Dispatcher for the UDP read loop.
+    * network-listener is a PinnedDispatcher, so each socket actor -- one per bound port --
+    * owns a dedicated thread and is insulated from game logic running on other pools.
+    * The dispatcher is selected here at the spawn site because these are typed actors, which
+    * config-based deployment in akka.conf does not apply to. */
+  private val socketDispatcher: DispatcherSelector = DispatcherSelector.fromConfig("network-listener")
+
   /**
    * Overrode constructor for `SocketPane` entities.
    * Registers the entity with the actor receptionist.
@@ -138,7 +145,7 @@ class SocketPane(
   /** all sockets produced by the socket group information and any later socket creation commands */
   private var socketActors: Array[ActorRef[SocketActor.Command]] = initialPortSetup.flatMap {
       case SocketSetup(_, SocketSetupInfo(address, ports, plan), _) =>
-        ports.map { portNum => context.spawn(SocketActor(new InetSocketAddress(address, portNum), plan), name=s"world-socket-$portNum") }
+        ports.map { portNum => context.spawn(SocketActor(new InetSocketAddress(address, portNum), plan), name=s"world-socket-$portNum", SocketPane.socketDispatcher) }
     }.toArray
   /** load balancing for redirecting newly discovered packet input to different sockets (ports);
    * should be referenced externally to switch sockets;
@@ -169,7 +176,7 @@ class SocketPane(
           val index = socketConfigs.indexWhere { setup => setup.groupId.equals(groupId) }
           val SocketSetup(_, SocketSetupInfo(address, ports, plan), _) = socketConfigs(index)
           socketActors = socketActors :+
-            context.spawn(SocketActor(new InetSocketAddress(address, port), plan), name=s"world-socket-$port")
+            context.spawn(SocketActor(new InetSocketAddress(address, port), plan), name=s"world-socket-$port", SocketPane.socketDispatcher)
           socketConfigs = (socketConfigs.take(index) :+ SocketSetup(groupId, SocketSetupInfo(address, ports :+ port, plan))) ++
             socketConfigs.drop(index + 1)
           socketRotations = (socketRotations.take(index) :+ SocketPanePortRotation(socketRotations(index), port)) ++
@@ -183,7 +190,7 @@ class SocketPane(
 
         case SocketPane.CreateNewSocketGroup(groupId, info @ SocketSetupInfo(address, ports, plan)) =>
           socketActors = socketActors ++
-            ports.map { portNum => context.spawn(SocketActor(new InetSocketAddress(address, portNum), plan), name=s"world-socket-$portNum") }
+            ports.map { portNum => context.spawn(SocketActor(new InetSocketAddress(address, portNum), plan), name=s"world-socket-$portNum", SocketPane.socketDispatcher) }
           socketConfigs = socketConfigs :+
             SocketSetup(groupId, info)
           socketRotations = socketRotations :+

src/main/scala/net/psforever/actors/session/support/ChatOperations.scala

@@ -1487,5 +1487,7 @@ class ChatOperations(
   override protected[session] def stop(): Unit = {
     silenceTimer.cancel()
     chatService ! ChatService.LeaveAllChannels(chatServiceAdapter)
+    /* the scheduler pool is owned by this session, so it is released with the session */
+    scheduler.shutdown()
   }
 }

src/main/scala/net/psforever/objects/guid/TaskWorkflow.scala

@@ -145,11 +145,18 @@ object TaskWorkflow {
         //no subtasks; just execute the main task
         promise.completeWith(task.action())
       case list =>
-        var unassignedCompletion: Boolean = true //shared mutex
+        /* A dedicated object, private to this bundle, guards `unassignedCompletion` so that
+           exactly one subtask completion resolves the promise.
+           The monitor must not be the flag itself: synchronising on a Boolean locks the boxed
+           java.lang.Boolean, and valueOf returns the cached TRUE/FALSE singletons, which are
+           shared JVM-wide and whose identity changes when the flag flips -- neither mutual
+           exclusion between threads nor isolation between bundles. */
+        val completionLock = new Object
+        var unassignedCompletion: Boolean = true
         //wait for subtasks to complete
         list.foreach { result =>
           result.onComplete { _ =>
-            unassignedCompletion.synchronized {
+            completionLock.synchronized {
               if (unassignedCompletion && composedSubs.forall(matchOnFutureCompletion)) {
                 unassignedCompletion = false
                 if (composedSubs.forall(matchOnFutureSuccess)) {

src/main/scala/net/psforever/objects/guid/pool/ExclusivePool.scala

@@ -9,6 +9,16 @@ class ExclusivePool(override val numbers: List[Int]) extends SimplePool(numbers)
   private val pool: Array[Int] = Array.ofDim[Int](numbers.length)
   numbers.indices.foreach(i => { pool(i) = i })
 
+  /* Constant-time views over `numbers`, which is a List: an indexed copy for resolving a
+     selector index to its number, and a reverse lookup for resolving a number back to its
+     index. Both operations are hot -- the configured pools are large (deployables 16000,
+     ammo and kits 13500 each) and unregistration arrives in bulk when a player dies or a
+     vehicle is deconstructed, all serialised through one pool actor.
+     Neither structure is mutated after construction, and SimplePool has already rejected
+     duplicates, so the reverse lookup is unambiguous. */
+  private val numbersByIndex: Array[Int] = numbers.toArray
+  private val indexByNumber: Map[Int, Int] = numbers.zipWithIndex.toMap
+
   override def Count: Int = pool.count(value => value == -1)
 
   override def Selector_=(slctr: NumberSelector): Unit = {
@@ -21,12 +31,11 @@ class ExclusivePool(override val numbers: List[Int]) extends SimplePool(numbers)
     if (index == -1) {
       Failure(new Exception("there are no numbers available in the pool"))
     } else {
-      Success(numbers(index))
+      Success(numbersByIndex(index))
     }
   }
 
   override def Return(number: Int): Boolean = {
-    val index = Numbers.indexOf(number)
-    index != -1 && Selector.Return(index, pool)
+    indexByNumber.get(number).exists(index => Selector.Return(index, pool))
   }
 }

src/main/scala/net/psforever/objects/guid/pool/GenericPool.scala

@@ -82,7 +82,11 @@ object GenericPool {
     */
   def first(list: List[Long], domainSize: Int): Int = {
     if (list.size < domainSize) {
-      val sortedList: List[Long] = 0L +: list.sorted :+ domainSize
+      /* An Array, because the loop below indexes it by position and Array.apply is constant
+         time. That matters here more than elsewhere: this pool is the fallback for any
+         exhausted pool and receives every number in the hub, on the order of tens of
+         thousands, and the scan runs on the pool actor's thread while its mailbox waits. */
+      val sortedList: Array[Long] = (0L +: list.sorted :+ domainSize.toLong).toArray
       var index: Int = 0
       val listLen = sortedList.length - 1
       while(index < listLen && index < domainSize) {
@@ -112,7 +116,8 @@ object GenericPool {
   def rand(list: List[Long], domainSize: Int): Int = {
     if (list.size < domainSize) {
       //get a list of all assigned numbers with an appended min and max
-      val sortedList: List[Long] = -1L +: list.sorted :+ domainSize.toLong
+      //an Array for the same reason as in `first`: the loop below indexes it by position
+      val sortedList: Array[Long] = (-1L +: list.sorted :+ domainSize.toLong).toArray
       //compare the delta between every two entries and find the start of that greatest delta comparison
       var maxDelta: Long = -1
       var maxDeltaIndex  = -1

src/main/scala/net/psforever/objects/serverobject/containable/ContainableBehavior.scala

@@ -2,7 +2,7 @@
 package net.psforever.objects.serverobject.containable
 
 import akka.actor.{Actor, ActorRef}
-import akka.pattern.{AskTimeoutException, ask}
+import akka.pattern.ask
 import akka.util.Timeout
 import net.psforever.objects.equipment.{Equipment, EquipmentSize}
 import net.psforever.objects.inventory.{Container, InventoryItem}
@@ -196,10 +196,11 @@ trait ContainableBehavior {
 
               case _ => ; //TODO what?
             }
-            //always do this
-            moveItemOver
-              .recover { case _: AskTimeoutException => destination.Actor ! ContainableBehavior.Resume() }
-              .onComplete { _ => destination.Actor ! ContainableBehavior.Resume() }
+            /* Exactly one Resume answers the Wait() above, and onComplete alone delivers it
+               for both success and failure -- including an AskTimeoutException. A second
+               Resume would release an unrelated concurrent move's guard and let insertions
+               interleave with a move that is still in flight. */
+            moveItemOver.onComplete { _ => destination.Actor ! ContainableBehavior.Resume() }
           }
         case _ => ;
         //we could not find the item to be moved in the source location; trying to act on old data?