diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf index 86ac54b7f..765c800fe 100644 --- a/src/main/resources/application.conf +++ b/src/main/resources/application.conf @@ -575,6 +575,17 @@ network { # before it is dropped. Can be used as a watchdog for hung server sessions. outbound-grace-time = 1 minute } + + event-caching { + # The minimum amount of time a cached message has to wait before being dispatched. (ms) + # Ideal circumstances are when message traffic is light. + flush-cache-delay = 15 + # The absolute maximum amount of time a cached message has to wait before being dispatched. (ms) + flush-cache-max-delay = 50 + # The numeric difference between light traffic and heavy traffic. + # In this case, the event system's traffic density and the message caching delay are numerically synonymous. + message-traffic-threshold = 12 + } } development { diff --git a/src/main/scala/net/psforever/actors/session/SessionActor.scala b/src/main/scala/net/psforever/actors/session/SessionActor.scala index b34d3965f..8138c56e9 100644 --- a/src/main/scala/net/psforever/actors/session/SessionActor.scala +++ b/src/main/scala/net/psforever/actors/session/SessionActor.scala @@ -392,6 +392,7 @@ class SessionActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], con case unknownStamp => log.error(s"received a message from an unknown event system - reply: $envelope, stamp: $unknownStamp") } + println(s"event-system-rtt: ${System.currentTimeMillis() - envelope.time} ms") } private def handleEnvelopeWithResponseHandler( diff --git a/src/main/scala/net/psforever/services/base/GenericEventServiceWithCacheAndSupport.scala b/src/main/scala/net/psforever/services/base/GenericEventServiceWithCacheAndSupport.scala index 3f926e1ff..fb224b619 100644 --- a/src/main/scala/net/psforever/services/base/GenericEventServiceWithCacheAndSupport.scala +++ b/src/main/scala/net/psforever/services/base/GenericEventServiceWithCacheAndSupport.scala @@ -7,12 +7,13 @@ import net.psforever.services.Service import net.psforever.services.base.envelope.{GenericMessageEnvelope, GenericResponseEnvelope, MessageEnvelope, MessageTransformationBehavior} import net.psforever.services.base.message.EventMessage import net.psforever.types.PlanetSideGUID +import net.psforever.util.Config import scala.collection.concurrent.{Map => CMap} import scala.jdk.CollectionConverters._ import java.util.concurrent.ConcurrentHashMap import scala.concurrent.ExecutionContext.Implicits.global -import scala.concurrent.duration.DurationInt +import scala.concurrent.duration.DurationLong /* Adapted from the rating limiting code in PSForever fork https://github.com/Pinapse/giant with permission @@ -97,8 +98,11 @@ class GenericEventServiceWithCacheAndSupport stamp: EventSystemStamp, eventSupportServices: List[EventServiceSupport] ) extends GenericEventServiceWithSupport(stamp, eventSupportServices) { - private val flushCacheWait: Long = 50L //milliseconds - private var hasCachedMessages: Boolean = false + private val flushCacheDelay: Long = Config.app.network.eventCaching.flushCacheDelay + private val flushCacheMaxDelay: Long = Config.app.network.eventCaching.flushCacheMaxDelay + private var hasCachedMessages: Int = 0 + private var lastCachedMessages: Int = 0 + private val messageThreshold: Long = Config.app.network.eventCaching.messageTrafficThreshold private var nextTimeToFlushCache: Long = 0L private var emergencyFlush: Cancellable = Default.Cancellable @@ -110,6 +114,14 @@ class GenericEventServiceWithCacheAndSupport super.postStop() } + private def adjustedFlushDelay(): Long = { + // flushCacheWait <= t <= emergencyFlushMaxDelay + math.min( + flushCacheDelay + math.max(0, lastCachedMessages - messageThreshold), + flushCacheMaxDelay + ) + } + /** * If there were previously no messages in the cache, * prepare to flush the cache after the intended interval passes, @@ -117,10 +129,11 @@ class GenericEventServiceWithCacheAndSupport * or if a safety timer expires and the cache is flushed in precaution. */ private def tryRetimeFlushCache(): Unit = { - if (!hasCachedMessages) { - hasCachedMessages = true - nextTimeToFlushCache = System.currentTimeMillis() + flushCacheWait - emergencyFlush = context.system.scheduler.scheduleOnce(55 milliseconds, self, FlushCachedMessages) + hasCachedMessages += 1 + if (hasCachedMessages == 1) { + val flushDelay = adjustedFlushDelay() + nextTimeToFlushCache = System.currentTimeMillis() + flushDelay + emergencyFlush = context.system.scheduler.scheduleOnce(delay = (1.25f * flushDelay).toLong milliseconds, self, FlushCachedMessages) } } @@ -145,7 +158,7 @@ class GenericEventServiceWithCacheAndSupport * flush the cache messages to the normal event system bus. */ private def tryFlushCache(): Boolean = { - val willFlush = hasCachedMessages && nextTimeToFlushCache < System.currentTimeMillis() + val willFlush = hasCachedMessages > 0 && nextTimeToFlushCache < System.currentTimeMillis() if (willFlush) { flushCache() } @@ -165,7 +178,8 @@ class GenericEventServiceWithCacheAndSupport map.clear() } } - hasCachedMessages = false + lastCachedMessages = hasCachedMessages + hasCachedMessages = 0 emergencyFlush.cancel() emergencyFlush = Default.Cancellable } diff --git a/src/main/scala/net/psforever/services/base/envelope/AllEnvelopes.scala b/src/main/scala/net/psforever/services/base/envelope/AllEnvelopes.scala index 7e067d5d5..d60860743 100644 --- a/src/main/scala/net/psforever/services/base/envelope/AllEnvelopes.scala +++ b/src/main/scala/net/psforever/services/base/envelope/AllEnvelopes.scala @@ -12,4 +12,6 @@ trait AllEnvelopes { def channel: String /** specific subscriber endpoint to be excluded (the subscriber should filter themselves upon receipt) */ def filter: PlanetSideGUID + + val time: Long = System.currentTimeMillis() } diff --git a/src/main/scala/net/psforever/util/Config.scala b/src/main/scala/net/psforever/util/Config.scala index c814904b6..f32c2adaf 100644 --- a/src/main/scala/net/psforever/util/Config.scala +++ b/src/main/scala/net/psforever/util/Config.scala @@ -131,7 +131,8 @@ case class AntiCheatConfig( case class NetworkConfig( session: SessionConfig, - middleware: MiddlewareConfig + middleware: MiddlewareConfig, + eventCaching: CachedMessagesConfig ) case class MiddlewareConfig( @@ -147,6 +148,12 @@ case class SessionConfig( outboundGraceTime: FiniteDuration ) +case class CachedMessagesConfig( + flushCacheDelay: Long, //milliseconds + flushCacheMaxDelay: Long, //milliseconds + messageTrafficThreshold: Long +) + case class GameConfig( instantAction: InstantActionConfig, amenityAutorepairRate: Float,