From 9b04f3f92cd1436b1354868404a1d51e05843285 Mon Sep 17 00:00:00 2001 From: ch3rn1k Date: Sun, 2 Aug 2026 12:52:37 +0300 Subject: [PATCH] feat: perf --- src/RustServerMetrics/Config/ConfigData.cs | 7 +- .../Utility/MetricsTimeStorage.cs | 23 +- src/RustServerMetrics/MetricsLogger.cs | 159 ++++++---- src/RustServerMetrics/MetricsSendBuffer.cs | 210 +++++++++++++ src/RustServerMetrics/MetricsUploadWorker.cs | 289 ++++++++++++++++++ src/RustServerMetrics/ReportUploader.cs | 192 ++++++------ .../RustServerMetrics.csproj | 9 +- .../RustServerMetricsLoader.cs | 2 +- 8 files changed, 725 insertions(+), 166 deletions(-) create mode 100644 src/RustServerMetrics/MetricsSendBuffer.cs create mode 100644 src/RustServerMetrics/MetricsUploadWorker.cs diff --git a/src/RustServerMetrics/Config/ConfigData.cs b/src/RustServerMetrics/Config/ConfigData.cs index 7ab11a9..19548b5 100644 --- a/src/RustServerMetrics/Config/ConfigData.cs +++ b/src/RustServerMetrics/Config/ConfigData.cs @@ -19,7 +19,7 @@ class ConfigData #endregion [JsonProperty(PropertyName = "Enabled")] - public bool Enabled; + public bool Enabled = false; [JsonProperty(PropertyName = "Influx Database Url")] public string DatabaseUrl = DefaultInfluxDbUrl; @@ -37,11 +37,14 @@ class ConfigData public string ServerTag = DefaultServerTag; [JsonProperty(PropertyName = "Debug Logging")] - public bool DebugLogging; + public bool DebugLogging = false; [JsonProperty(PropertyName = "Amount of metrics to submit in each request")] public ushort BatchSize = 1000; [JsonProperty(PropertyName = "Gather Player Averages (Client FPS, Client Latency, Player FPS, Player Memory, Player Latency, Player Packet Loss)")] public bool GatherPlayerMetrics = true; + + [JsonProperty(PropertyName = "Compress submitted metrics with gzip")] + public bool CompressRequests = true; } \ No newline at end of file diff --git a/src/RustServerMetrics/HarmonyPatches/Utility/MetricsTimeStorage.cs b/src/RustServerMetrics/HarmonyPatches/Utility/MetricsTimeStorage.cs index 93b7807..9740e52 100644 --- a/src/RustServerMetrics/HarmonyPatches/Utility/MetricsTimeStorage.cs +++ b/src/RustServerMetrics/HarmonyPatches/Utility/MetricsTimeStorage.cs @@ -6,22 +6,27 @@ namespace RustServerMetrics.HarmonyPatches.Utility; public class MetricsTimeStorage(string metricKey, Action stringBuilderSerializer) { - private readonly Dictionary _dict = new (); - + private sealed class Accumulator + { + public double Duration; + } + + private readonly Dictionary _dict = new (); + private readonly StringBuilder _sb = new(); public void LogTime(TKey key, double milliseconds) { if (!MetricsLogger.IsReady) return; - - if (!_dict.TryGetValue(key, out var currentDuration)) + + if (_dict.TryGetValue(key, out var accumulator)) { - _dict.Add(key, milliseconds); + accumulator.Duration += milliseconds; return; } - - _dict[key] = currentDuration + milliseconds; + + _dict.Add(key, new Accumulator { Duration = milliseconds }); } public void SerializeToStringBuilder() @@ -44,10 +49,10 @@ public void SerializeToStringBuilder() stringBuilderSerializer.Invoke(_sb, item.Key); _sb.Append("\" duration="); - _sb.Append((float)item.Value); + _sb.Append((float)item.Value.Duration); _sb.Append(" "); _sb.Append(epochNow); - instance.AddToSendBuffer(_sb.ToString()); + instance.AddToSendBuffer(_sb); } _dict.Clear(); diff --git a/src/RustServerMetrics/MetricsLogger.cs b/src/RustServerMetrics/MetricsLogger.cs index c2faa92..114e7d4 100644 --- a/src/RustServerMetrics/MetricsLogger.cs +++ b/src/RustServerMetrics/MetricsLogger.cs @@ -17,9 +17,12 @@ public class MetricsLogger : SingletonComponent { private const string ConfigurationPath = "HarmonyMods_Data/ServerMetrics/Configuration.json"; private readonly StringBuilder _stringBuilder = new(); - private readonly Dictionary _playerStatsActions = new(); private readonly Dictionary _perfReportDelayCounter = new(); + private const int PlayerStatsBucketCount = 10; + private const float PlayerStatsBucketInterval = 1f / PlayerStatsBucketCount; + private int _playerStatsBucket; + private class NetworkUpdateData { public int Count; @@ -33,17 +36,39 @@ public NetworkUpdateData(int count, long bytes) } } - private readonly Dictionary _networkUpdates = Enum.GetValues(typeof(Message.Type)) - .Cast() - .Distinct() - .ToDictionary(x => x, - _ => new NetworkUpdateData(0, 0)); + private static readonly Message.Type[] MessageTypes = Enum.GetValues(typeof(Message.Type)) + .Cast() + .Distinct() + .ToArray(); - private static readonly IReadOnlyDictionary MessageTypeNames = Enum.GetValues(typeof(Message.Type)) - .Cast() - .Distinct() - .ToDictionary(x => x, - x => x.ToString()); + private static readonly int MessageTypeSlotOffset = MessageTypes.Min(x => (int)x); + private static readonly int MessageTypeSlotCount = MessageTypes.Max(x => (int)x) - MessageTypeSlotOffset + 1; + + private static readonly string[] MessageTypeNames = BuildMessageTypeNames(); + + private readonly NetworkUpdateData[] _networkUpdates = BuildNetworkUpdates(); + + private static string[] BuildMessageTypeNames() + { + var names = new string[MessageTypeSlotCount]; + foreach (var messageType in MessageTypes) + { + names[(int)messageType - MessageTypeSlotOffset] = messageType.ToString(); + } + + return names; + } + + private static NetworkUpdateData[] BuildNetworkUpdates() + { + var networkUpdates = new NetworkUpdateData[MessageTypeSlotCount]; + foreach (var messageType in MessageTypes) + { + networkUpdates[(int)messageType - MessageTypeSlotOffset] = new NetworkUpdateData(0, 0); + } + + return networkUpdates; + } public readonly MetricsTimeStorage ServerInvokes = new("invoke_execution", LogMethodInfo); public readonly MetricsTimeStorage ServerRpcCalls = new("rpc_calls", LogMethodName); @@ -139,6 +164,7 @@ public override void Awake() public void StartLoggingMetrics() { InvokeRepeating(LogNetworkUpdates, UnityEngine.Random.Range(0.25f, 0.75f), 0.5f); + InvokeRepeating(GatherPlayerStatsBucket, UnityEngine.Random.Range(0.5f, 1.5f), PlayerStatsBucketInterval); InvokeRepeating(ServerInvokes.SerializeToStringBuilder, UnityEngine.Random.Range(0f, 1f), 1f); InvokeRepeating(ServerRpcCalls.SerializeToStringBuilder, UnityEngine.Random.Range(0f, 1f), 1f); @@ -155,20 +181,13 @@ internal void OnPlayerInit(BasePlayer player) { if (!Ready) return; if (!Configuration.GatherPlayerMetrics) return; - var action = new Action(() => GatherPlayerSecondStats(player)); - if (_playerStatsActions.TryGetValue(player.userID, out var existingAction)) - player.CancelInvoke(existingAction); - _playerStatsActions[player.userID] = action; - player.InvokeRepeating(action, UnityEngine.Random.Range(0.5f, 1.5f), 1f); + + _perfReportDelayCounter[player.userID] = (uint)UnityEngine.Random.Range(0, 5); } internal void OnPlayerDisconnected(BasePlayer player) { if (!Ready) return; - if (!Configuration.GatherPlayerMetrics) return; - if (_playerStatsActions.TryGetValue(player.userID, out var action)) - player.CancelInvoke(action); - _playerStatsActions.Remove(player.userID); _perfReportDelayCounter.Remove(player.userID); } @@ -189,7 +208,18 @@ internal void OnNetWriteSend(NetWrite write, SendInfo sendInfo) return; } - var data = _networkUpdates[_lastMessageType]; + var slot = (int)_lastMessageType - MessageTypeSlotOffset; + if ((uint)slot >= (uint)_networkUpdates.Length) + { + return; + } + + var data = _networkUpdates[slot]; + if (data == null) + { + return; + } + if (sendInfo.connection != null) { data.Count++; @@ -199,7 +229,7 @@ internal void OnNetWriteSend(NetWrite write, SendInfo sendInfo) { var count = sendInfo.connections.Count; data.Count += count; - data.Bytes += write.Length * count; + data.Bytes += (long)write.Length * count; } } @@ -240,8 +270,28 @@ internal bool OnClientPerformanceReport(ProtoBuf.PerformanceReport clientPerform return true; } + private void GatherPlayerStatsBucket() + { + if (!Ready) return; + if (!Configuration.GatherPlayerMetrics) return; + + var players = BasePlayer.activePlayerList; + var bucket = _playerStatsBucket; + _playerStatsBucket = bucket + 1 < PlayerStatsBucketCount ? bucket + 1 : 0; + + for (var i = bucket; i < players.Count; i += PlayerStatsBucketCount) + { + var player = players[i]; + if (player == null) continue; + + GatherPlayerSecondStats(player); + } + } + private void GatherPlayerSecondStats(BasePlayer player) { + if (player.net?.connection == null) return; + if (!player.IsReceivingSnapshot) { _perfReportDelayCounter.TryGetValue(player.userID, out var perfReportCounter); @@ -259,11 +309,12 @@ private void GatherPlayerSecondStats(BasePlayer player) UploadPacket("connection_latency", player, (builder, basePlayer) => { var ip = basePlayer.net.connection.ipaddress; + var portSeparator = ip.LastIndexOf(':'); builder.Append(",steamid="); builder.Append(basePlayer.UserIDString); builder.Append(",ip="); - builder.Append(ip[..ip.LastIndexOf(':')]); + builder.Append(ip, 0, portSeparator < 0 ? ip.Length : portSeparator); builder.Append(" ping="); builder.Append(Net.sv.GetAveragePing(basePlayer.net.connection)); builder.Append("i,packet_loss="); @@ -274,7 +325,7 @@ private void GatherPlayerSecondStats(BasePlayer player) private void LogNetworkUpdates() { - if (_networkUpdates.Count < 1) return; + if (_networkUpdates.Length < 1) return; var serverTag = Configuration.ServerTag; var epochNow = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); _stringBuilder.Clear(); @@ -282,12 +333,21 @@ private void LogNetworkUpdates() _stringBuilder.Append(serverTag); _stringBuilder.Append(" "); - var enumerator = _networkUpdates.GetEnumerator(); - if (enumerator.MoveNext()) + var isFirstField = true; + for (var slot = 0; slot < _networkUpdates.Length; slot++) { - var networkUpdate = enumerator.Current; - var key = MessageTypeNames[networkUpdate.Key]; - var value = networkUpdate.Value; + var value = _networkUpdates[slot]; + if (value == null) continue; + + var key = MessageTypeNames[slot]; + + if (!isFirstField) + { + _stringBuilder.Append(","); + } + + isFirstField = false; + // Count first named {type} _stringBuilder.Append(key); _stringBuilder.Append("="); @@ -303,35 +363,11 @@ private void LogNetworkUpdates() _stringBuilder.Append(value.Bytes); _stringBuilder.Append("i"); value.Bytes = 0; - - while (enumerator.MoveNext()) - { - networkUpdate = enumerator.Current; - key = MessageTypeNames[networkUpdate.Key]; - value = networkUpdate.Value; - - // Count first named {type} - _stringBuilder.Append(","); - _stringBuilder.Append(key); - _stringBuilder.Append("="); - _stringBuilder.Append(value.Count); - _stringBuilder.Append("i"); - value.Count = 0; - - // Bytes second named as "{type}_bytes" - _stringBuilder.Append(","); - _stringBuilder.Append(key); - _stringBuilder.Append("_bytes"); - _stringBuilder.Append("="); - _stringBuilder.Append(value.Bytes); - _stringBuilder.Append("i"); - value.Bytes = 0; - } } _stringBuilder.Append(" "); _stringBuilder.Append(epochNow); - _reportUploader.AddToSendBuffer(_stringBuilder.ToString()); + _reportUploader.AddToSendBuffer(_stringBuilder); } internal void OnPerformanceReportGenerated() @@ -438,7 +474,7 @@ private void LogPerformanceReport(Performance.Tick current, string epochNow, str _stringBuilder.Append("i "); _stringBuilder.Append(epochNow); - _reportUploader.AddToSendBuffer(_stringBuilder.ToString()); + _reportUploader.AddToSendBuffer(_stringBuilder); } @@ -458,11 +494,13 @@ public void UploadPacket(string id, T data, Action serializ _stringBuilder.Append(" "); _stringBuilder.Append(epochNow); - AddToSendBuffer(_stringBuilder.ToString()); + AddToSendBuffer(_stringBuilder); } public void AddToSendBuffer(string toString) => _reportUploader.AddToSendBuffer(toString); + public void AddToSendBuffer(StringBuilder payload) => _reportUploader.AddToSendBuffer(payload); + private long GetMemoryUsage(Performance.Tick performanceTick) { if (performanceTick.memoryUsageSystem > 0) @@ -550,6 +588,8 @@ private void StatusCommand(ConsoleSystem.Arg arg) _stringBuilder.AppendLine("Report Uploader:"); _stringBuilder.Append("\tRunning: "); _stringBuilder.Append(_reportUploader.IsRunning); _stringBuilder.AppendLine(); _stringBuilder.Append("\tIn Buffer: "); _stringBuilder.Append(_reportUploader.BufferSize); _stringBuilder.AppendLine(); + _stringBuilder.Append("\tIn Flight: "); _stringBuilder.Append(_reportUploader.PendingBatches); _stringBuilder.AppendLine(); + _stringBuilder.Append("\tDropped (total): "); _stringBuilder.Append(_reportUploader.TotalDroppedReports); _stringBuilder.AppendLine(); arg.ReplyWith(_stringBuilder.ToString()); } @@ -568,12 +608,7 @@ private void ReloadCfgCommand(ConsoleSystem.Arg arg) CancelInvoke(invoke.action); } - foreach (var player in _playerStatsActions) - { - var basePlayer = BasePlayer.FindByID(player.Key); - if (basePlayer == null) continue; - basePlayer.CancelInvoke(player.Value); - } + _perfReportDelayCounter.Clear(); _reportUploader.Stop(); if (!Configuration.Enabled) diff --git a/src/RustServerMetrics/MetricsSendBuffer.cs b/src/RustServerMetrics/MetricsSendBuffer.cs new file mode 100644 index 0000000..aad7a13 --- /dev/null +++ b/src/RustServerMetrics/MetricsSendBuffer.cs @@ -0,0 +1,210 @@ +using System; +using System.Text; + +namespace RustServerMetrics; + +internal sealed class MetricsSendBuffer +{ + private const int InitialCapacity = 1 << 16; + private const int InitialLineCapacity = 1 << 10; + + private readonly int _maxCapacity; + + private char[] _chars = new char[InitialCapacity]; + private char[] _batch = new char[InitialCapacity]; + + private int[] _lineLengths = new int[InitialLineCapacity]; + private int _lineHead; + private int _lineCount; + + private int _head; + private int _count; + private long _totalDropped; + + public MetricsSendBuffer(int maxCapacity) + { + if (maxCapacity < InitialCapacity) + { + throw new ArgumentOutOfRangeException(nameof(maxCapacity)); + } + + _maxCapacity = maxCapacity; + } + + public int LineCount => _lineCount; + public int CharCount => _count; + public long TotalDropped => _totalDropped; + + public int LastBatchLineCount { get; private set; } + + public void AddDropped(int lines) => _totalDropped += lines; + + public void Append(string line) + { + if (string.IsNullOrEmpty(line)) return; + if (!TryReserve(line.Length + 1)) return; + + for (var i = 0; i < line.Length; i++) + { + _chars[WriteIndex(i)] = line[i]; + } + + Commit(line.Length); + } + + public void Append(StringBuilder line) + { + if (line == null || line.Length == 0) return; + if (!TryReserve(line.Length + 1)) return; + + var writeStart = WriteIndex(0); + var untilEnd = Math.Min(line.Length, _chars.Length - writeStart); + line.CopyTo(0, _chars, writeStart, untilEnd); + + if (untilEnd < line.Length) + { + line.CopyTo(untilEnd, _chars, 0, line.Length - untilEnd); + } + + Commit(line.Length); + } + + public char[] TakeBatch(int maxLines, int maxChars, out int charCount) + { + charCount = 0; + var lines = 0; + + while (lines < maxLines && lines < _lineCount) + { + var lineLength = _lineLengths[(_lineHead + lines) % _lineLengths.Length]; + if (lines > 0 && charCount + lineLength > maxChars) break; + + charCount += lineLength; + lines++; + } + + LastBatchLineCount = lines; + + if (charCount == 0) return _batch; + + if (_batch.Length < charCount) + { + _batch = new char[charCount]; + } + + var untilEnd = Math.Min(charCount, _chars.Length - _head); + Array.Copy(_chars, _head, _batch, 0, untilEnd); + + if (untilEnd < charCount) + { + Array.Copy(_chars, 0, _batch, untilEnd, charCount - untilEnd); + } + + _head = (_head + charCount) % _chars.Length; + _count -= charCount; + _lineHead = (_lineHead + lines) % _lineLengths.Length; + _lineCount -= lines; + + return _batch; + } + + public void Clear() + { + _head = 0; + _count = 0; + _lineHead = 0; + _lineCount = 0; + } + + private int WriteIndex(int offset) => (_head + _count + offset) % _chars.Length; + + private void Commit(int lineLength) + { + _chars[WriteIndex(lineLength)] = '\n'; + _count += lineLength + 1; + + if (_lineCount == _lineLengths.Length) + { + GrowLineLengths(); + } + + _lineLengths[(_lineHead + _lineCount) % _lineLengths.Length] = lineLength + 1; + _lineCount++; + } + + private bool TryReserve(int required) + { + if (required > _maxCapacity) + { + _totalDropped++; + return false; + } + + if (required > _chars.Length - _count) + { + Grow(required); + } + + while (required > _chars.Length - _count) + { + DropOldestLine(); + } + + return true; + } + + private void Grow(int required) + { + var capacity = _chars.Length; + while (capacity < _maxCapacity && required > capacity - _count) + { + capacity = Math.Min(capacity * 2, _maxCapacity); + } + + if (capacity == _chars.Length) return; + + var grown = new char[capacity]; + var untilEnd = Math.Min(_count, _chars.Length - _head); + Array.Copy(_chars, _head, grown, 0, untilEnd); + + if (untilEnd < _count) + { + Array.Copy(_chars, 0, grown, untilEnd, _count - untilEnd); + } + + _chars = grown; + _head = 0; + } + + private void GrowLineLengths() + { + var grown = new int[_lineLengths.Length * 2]; + var untilEnd = Math.Min(_lineCount, _lineLengths.Length - _lineHead); + Array.Copy(_lineLengths, _lineHead, grown, 0, untilEnd); + + if (untilEnd < _lineCount) + { + Array.Copy(_lineLengths, 0, grown, untilEnd, _lineCount - untilEnd); + } + + _lineLengths = grown; + _lineHead = 0; + } + + private void DropOldestLine() + { + if (_lineCount == 0) + { + _head = 0; + _count = 0; + return; + } + + var lineLength = _lineLengths[_lineHead]; + _lineHead = (_lineHead + 1) % _lineLengths.Length; + _lineCount--; + _head = (_head + lineLength) % _chars.Length; + _count -= lineLength; + _totalDropped++; + } +} diff --git a/src/RustServerMetrics/MetricsUploadWorker.cs b/src/RustServerMetrics/MetricsUploadWorker.cs new file mode 100644 index 0000000..9eab5ca --- /dev/null +++ b/src/RustServerMetrics/MetricsUploadWorker.cs @@ -0,0 +1,289 @@ +using System; +using System.IO; +using System.IO.Compression; +using System.Net.Http; +using System.Net.Http.Headers; +using System.Threading; + +namespace RustServerMetrics; + +internal sealed class MetricsUploadWorker : IDisposable +{ + private const int MaxQueuedBatches = 4; + private const int MaxAttempts = 3; + private const int IdleWaitMilliseconds = 250; + private const int RetryDelayMilliseconds = 200; + private const int ShutdownJoinMilliseconds = 1000; + + private sealed class Batch + { + public Uri Uri; + public byte[] Body; + public int Lines; + public bool Compress; + public bool CaptureResponse; + } + + private readonly object _sync = new(); + private readonly Batch[] _queue = new Batch[MaxQueuedBatches]; + private readonly AutoResetEvent _signal = new(false); + private readonly Thread _thread; + private readonly int _timeoutSeconds; + + private int _head; + private int _count; + private int _inFlight; + private volatile bool _running = true; + + private HttpClient _client; + private MemoryStream _compressionBuffer; + + private long _droppedLines; + private int _networkFailures; + private int _httpFailures; + private string _lastNetworkError; + private string _lastHttpError; + private string _lastResponseBody; + + public MetricsUploadWorker(int timeoutSeconds) + { + _timeoutSeconds = timeoutSeconds; + _thread = new Thread(Run) + { + IsBackground = true, + Name = "ServerMetrics uploader" + }; + + _thread.Start(); + } + + public long DroppedLines => Interlocked.Read(ref _droppedLines); + + public int PendingBatches + { + get { lock (_sync) { return _count + _inFlight; } } + } + + public bool HasRoom + { + get { lock (_sync) { return _running && _count < MaxQueuedBatches; } } + } + + public bool TryEnqueue(Uri uri, byte[] body, int lines, bool compress, bool captureResponse) + { + lock (_sync) + { + if (!_running || _count == MaxQueuedBatches) return false; + + _queue[(_head + _count) % MaxQueuedBatches] = new Batch + { + Uri = uri, + Body = body, + Lines = lines, + Compress = compress, + CaptureResponse = captureResponse + }; + + _count++; + } + + _signal.Set(); + return true; + } + + public int TakeNetworkFailures(out string lastError) + { + lock (_sync) + { + var failures = _networkFailures; + lastError = _lastNetworkError; + _networkFailures = 0; + _lastNetworkError = null; + return failures; + } + } + + public int TakeHttpFailures(out string lastError, out string lastResponseBody) + { + lock (_sync) + { + var failures = _httpFailures; + lastError = _lastHttpError; + lastResponseBody = _lastResponseBody; + _httpFailures = 0; + _lastHttpError = null; + _lastResponseBody = null; + return failures; + } + } + + public void DropQueued() + { + lock (_sync) + { + while (_count > 0) + { + var batch = _queue[_head]; + _queue[_head] = null; + _head = (_head + 1) % MaxQueuedBatches; + _count--; + Interlocked.Add(ref _droppedLines, batch.Lines); + } + } + } + + public void Dispose() + { + _running = false; + _signal.Set(); + _thread.Join(ShutdownJoinMilliseconds); + + _client?.Dispose(); + _client = null; + } + + private void Run() + { + try + { + Loop(); + } + catch + { + // + } + } + + private void Loop() + { + while (_running) + { + Batch batch = null; + + lock (_sync) + { + if (_count > 0) + { + batch = _queue[_head]; + _queue[_head] = null; + _head = (_head + 1) % MaxQueuedBatches; + _count--; + _inFlight++; + } + } + + if (batch == null) + { + _signal.WaitOne(IdleWaitMilliseconds); + continue; + } + + try + { + Send(batch); + } + catch (Exception e) + { + RecordNetworkFailure(Describe(e), batch.Lines); + } + finally + { + lock (_sync) { _inFlight--; } + } + } + } + + private void Send(Batch batch) + { + var body = batch.Body; + var length = body.Length; + var compressed = false; + + if (batch.Compress) + { + length = Compress(body, out body); + compressed = true; + } + + string lastError = null; + + for (var attempt = 1; attempt <= MaxAttempts; attempt++) + { + try + { + using var content = new ByteArrayContent(body, 0, length); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + if (compressed) + { + content.Headers.ContentEncoding.Add("gzip"); + } + + using var response = EnsureClient().PostAsync(batch.Uri, content).GetAwaiter().GetResult(); + if (response.IsSuccessStatusCode) return; + + var responseBody = batch.CaptureResponse + ? response.Content.ReadAsStringAsync().GetAwaiter().GetResult() + : null; + + RecordHttpFailure($"{(int)response.StatusCode} {response.ReasonPhrase}", responseBody, batch.Lines); + return; + } + catch (Exception e) + { + lastError = Describe(e); + if (!_running || attempt == MaxAttempts) break; + Thread.Sleep(RetryDelayMilliseconds * attempt); + } + } + + RecordNetworkFailure(lastError ?? "request failed", batch.Lines); + } + + private int Compress(byte[] body, out byte[] compressed) + { + _compressionBuffer ??= new MemoryStream(1 << 16); + _compressionBuffer.SetLength(0); + + using (var gzip = new GZipStream(_compressionBuffer, CompressionLevel.Fastest, true)) + { + gzip.Write(body, 0, body.Length); + } + + compressed = _compressionBuffer.GetBuffer(); + return (int)_compressionBuffer.Length; + } + + private HttpClient EnsureClient() + { + if (_client != null) return _client; + + _client = new HttpClient { Timeout = TimeSpan.FromSeconds(_timeoutSeconds) }; + _client.DefaultRequestHeaders.ExpectContinue = false; + return _client; + } + + private void RecordNetworkFailure(string error, int lines) + { + Interlocked.Add(ref _droppedLines, lines); + + lock (_sync) + { + _networkFailures++; + _lastNetworkError = error; + } + } + + private void RecordHttpFailure(string error, string responseBody, int lines) + { + Interlocked.Add(ref _droppedLines, lines); + + lock (_sync) + { + _httpFailures++; + _lastHttpError = error; + if (responseBody != null) _lastResponseBody = responseBody; + } + } + + private static string Describe(Exception e) => e.GetBaseException().Message; +} diff --git a/src/RustServerMetrics/ReportUploader.cs b/src/RustServerMetrics/ReportUploader.cs index 677cbe2..548653d 100644 --- a/src/RustServerMetrics/ReportUploader.cs +++ b/src/RustServerMetrics/ReportUploader.cs @@ -1,29 +1,28 @@ using System; -using System.Collections; -using System.Collections.Generic; using System.Text; using UnityEngine; -using UnityEngine.Networking; namespace RustServerMetrics; internal class ReportUploader : MonoBehaviour { - private const int SendBufferCapacity = 100000; + private const int SendBufferCapacity = 8 * 1024 * 1024; + + private const int MaxBatchCharacters = 60000; + + private const int RequestTimeoutSeconds = 15; + + private const float FlushInterval = 1f; private readonly Action _notifySubsequentNetworkFailuresAction; private readonly Action _notifySubsequentHttpFailuresAction; - private readonly Queue _sendBuffer = new(SendBufferCapacity); - private readonly StringBuilder _payloadBuilder = new(); + private readonly MetricsSendBuffer _sendBuffer = new(SendBufferCapacity); - private bool _isRunning; - private ushort _attempt; - private byte[] _data; - private Uri _uri; + private MetricsUploadWorker _worker; private MetricsLogger _metricsLogger; - - private char[] _charBuffer = new char[8192 * 4]; + private bool _isRunning; + private float _nextFlush; private bool _throttleNetworkErrorMessages; private uint _accumulatedNetworkErrors; @@ -39,9 +38,11 @@ private ushort BatchSize return configVal < 1000 ? (ushort)1000 : configVal; } } - + public bool IsRunning => _isRunning; - public int BufferSize => _sendBuffer.Count; + public int BufferSize => _sendBuffer.LineCount; + public int PendingBatches => _worker?.PendingBatches ?? 0; + public long TotalDroppedReports => _sendBuffer.TotalDropped + (_worker?.DroppedLines ?? 0); public ReportUploader() { @@ -66,98 +67,106 @@ private void Awake() public void AddToSendBuffer(string payload) { - if (_sendBuffer.Count == SendBufferCapacity) + _sendBuffer.Append(payload); + _isRunning = true; + } + + public void AddToSendBuffer(StringBuilder payload) + { + _sendBuffer.Append(payload); + _isRunning = true; + } + + private void Update() + { + if (_metricsLogger == null) { - _sendBuffer.Dequeue(); + Stop(); + return; } - _sendBuffer.Enqueue(payload); + if (_worker != null) + { + DrainFailures(); + } - if (!_isRunning) + if (_isRunning) { - StartCoroutine(SendBufferLoop()); + PumpBatches(); } } - private IEnumerator SendBufferLoop() + private void PumpBatches() { - _isRunning = true; - yield return null; + var queuedLines = _sendBuffer.LineCount; + if (queuedLines == 0) return; + + var batchSize = BatchSize; + + if (queuedLines < batchSize && Time.realtimeSinceStartup < _nextFlush) return; + _nextFlush = Time.realtimeSinceStartup + FlushInterval; - while (_sendBuffer.Count > 0 && _isRunning) + var worker = EnsureWorker(); + var compress = _metricsLogger.Configuration?.CompressRequests ?? true; + var captureResponse = _metricsLogger.Configuration?.DebugLogging == true; + + while (_sendBuffer.LineCount > 0 && worker.HasRoom) { - var amountToTake = Mathf.Min(_sendBuffer.Count, BatchSize); - for (var i = 0; i < amountToTake; i++) - { - _payloadBuilder.Append(_sendBuffer.Dequeue()); - _payloadBuilder.Append("\n"); - } - _attempt = 0; - - // more GC friendly GetBytes implementation - if (_payloadBuilder.Length > _charBuffer.Length) - { - _charBuffer = new char[_payloadBuilder.Length + 1024]; - } - - _payloadBuilder.CopyTo(0, _charBuffer, 0, _payloadBuilder.Length); - _data = Encoding.UTF8.GetBytes(_charBuffer, 0, _payloadBuilder.Length); - - _uri = _metricsLogger.BaseUri; - _payloadBuilder.Clear(); - yield return SendRequest(); + var batch = _sendBuffer.TakeBatch(batchSize, MaxBatchCharacters, out var characterCount); + var lines = _sendBuffer.LastBatchLineCount; + var data = Encoding.UTF8.GetBytes(batch, 0, characterCount); + + if (worker.TryEnqueue(_metricsLogger.BaseUri, data, lines, compress, captureResponse)) continue; + + _sendBuffer.AddDropped(lines); + break; } - _isRunning = false; } - private IEnumerator SendRequest() + private MetricsUploadWorker EnsureWorker() => _worker ??= new MetricsUploadWorker(RequestTimeoutSeconds); + + private void DrainFailures() { - var request = new UnityWebRequest(_uri, UnityWebRequest.kHttpVerbPOST) + var networkFailures = _worker.TakeNetworkFailures(out var networkError); + if (networkFailures > 0) { - uploadHandler = new UploadHandlerRaw(_data), - downloadHandler = new DownloadHandlerBuffer(), - timeout = 15, - useHttpContinue = true, - redirectLimit = 5 - }; - yield return request.SendWebRequest(); - - if (request.isNetworkError) + ReportNetworkFailures(networkFailures, networkError); + } + + var httpFailures = _worker.TakeHttpFailures(out var httpError, out var responseBody); + if (httpFailures > 0) + { + ReportHttpFailures(httpFailures, httpError, responseBody); + } + } + + private void ReportNetworkFailures(int failures, string error) + { + if (_throttleNetworkErrorMessages) { - if (_attempt >= 2) - { - if (_throttleNetworkErrorMessages) - { - _accumulatedNetworkErrors += 1; - } - else - { - Debug.LogError($"Two consecutive network failures occurred while submitting a batch of metrics"); - InvokeHandler.Invoke(this, _notifySubsequentNetworkFailuresAction, 5); - _throttleNetworkErrorMessages = true; - } - yield break; - } - - _attempt++; - yield return SendRequest(); - yield break; + _accumulatedNetworkErrors += (uint)failures; + return; } - if (request.isHttpError) + Debug.LogError($"Consecutive network failures occurred while submitting a batch of metrics: {error}"); + InvokeHandler.Invoke(this, _notifySubsequentNetworkFailuresAction, 5); + _throttleNetworkErrorMessages = true; + _accumulatedNetworkErrors += (uint)(failures - 1); + } + + private void ReportHttpFailures(int failures, string error, string responseBody) + { + if (_throttleHttpErrorMessages) { - if (_throttleHttpErrorMessages) - { - _accumulatedHttpErrors += 1; - } - else - { - Debug.LogError($"A HTTP error occurred while submitting batch of metrics: {request.error}"); - if (_metricsLogger.Configuration?.DebugLogging == true) Debug.LogError(request.downloadHandler.text); - InvokeHandler.Invoke(this, _notifySubsequentHttpFailuresAction, 5); - _throttleHttpErrorMessages = true; - } + _accumulatedHttpErrors += (uint)failures; + return; } + + Debug.LogError($"A HTTP error occurred while submitting batch of metrics: {error}"); + if (responseBody != null) Debug.LogError(responseBody); + InvokeHandler.Invoke(this, _notifySubsequentHttpFailuresAction, 5); + _throttleHttpErrorMessages = true; + _accumulatedHttpErrors += (uint)(failures - 1); } void NotifySubsequentNetworkFailures() @@ -181,9 +190,18 @@ void OnDestroy() Stop(); } + private void DisposeWorker() + { + var worker = _worker; + _worker = null; + worker?.Dispose(); + } + public void Stop() { _isRunning = false; - StopAllCoroutines(); + + _worker?.DropQueued(); + DisposeWorker(); } -} \ No newline at end of file +} diff --git a/src/RustServerMetrics/RustServerMetrics.csproj b/src/RustServerMetrics/RustServerMetrics.csproj index e188d89..c4af743 100644 --- a/src/RustServerMetrics/RustServerMetrics.csproj +++ b/src/RustServerMetrics/RustServerMetrics.csproj @@ -4,10 +4,11 @@ RustServerMetrics {3C3A47A5-709A-42BB-B2BF-DB1FA0FEE316} net48 - 14 + 13 Library disable ..\..\dependencies\rust;$(AssemblySearchPaths) + MSB3277 @@ -20,7 +21,7 @@ prompt AnyCPU - + bin\Debug\ TRACE;DEBUG @@ -31,7 +32,7 @@ prompt AnyCPU - + @@ -106,7 +107,6 @@ - @@ -125,7 +125,6 @@ - diff --git a/src/RustServerMetrics/RustServerMetricsLoader.cs b/src/RustServerMetrics/RustServerMetricsLoader.cs index 12d76d0..54e5b24 100644 --- a/src/RustServerMetrics/RustServerMetricsLoader.cs +++ b/src/RustServerMetrics/RustServerMetricsLoader.cs @@ -40,7 +40,7 @@ public void OnUnloaded(OnHarmonyModUnloadedArgs args) if (MetricsLogger.Instance != null) { - Object.DestroyImmediate(MetricsLogger.Instance); + Object.DestroyImmediate(MetricsLogger.Instance.gameObject); } }