mirror of
https://github.com/DarkflameUniverse/DarkflameServer.git
synced 2026-10-13 00:03:45 +00:00
feat(net): count every server's packets and send a traffic report every 5 seconds
TrafficStats keeps packets and bytes in and out per second and the busiest packet and game message types. dServer counts at its send and receive calls and adds RakNet's connection statistics (datagrams, resends, ping); the report goes to master as SERVER_TRAFFIC (appended), which passes it to the dashboard. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Aaron Kimbrell
@@ -4,6 +4,7 @@ set(DCOMMON_SOURCES
|
||||
"BinaryIO.cpp"
|
||||
"dConfig.cpp"
|
||||
"Diagnostics.cpp"
|
||||
"TrafficStats.cpp"
|
||||
"Locale.cpp"
|
||||
"Logger.cpp"
|
||||
"Game.cpp"
|
||||
|
||||
276
dCommon/TrafficStats.cpp
Normal file
276
dCommon/TrafficStats.cpp
Normal file
@@ -0,0 +1,276 @@
|
||||
#include "TrafficStats.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <cmath>
|
||||
#include <ctime>
|
||||
|
||||
#include "MessageIdentifiers.h"
|
||||
#include "ServiceType.h"
|
||||
#include "MessageType/Client.h"
|
||||
#include "MessageType/World.h"
|
||||
|
||||
namespace TrafficStats {
|
||||
namespace {
|
||||
constexpr double FIRST_BOUND = 100.0; // microseconds
|
||||
|
||||
const std::array<uint64_t, Histogram::BUCKETS>& Bounds() {
|
||||
static const auto bounds = [] {
|
||||
std::array<uint64_t, Histogram::BUCKETS> out{};
|
||||
for (size_t i = 0; i + 1 < Histogram::BUCKETS; i++) out[i] = static_cast<uint64_t>(std::llround(FIRST_BOUND * std::exp2(static_cast<double>(i) / 3.0)));
|
||||
out[Histogram::BUCKETS - 1] = UINT64_MAX;
|
||||
return out;
|
||||
}();
|
||||
return bounds;
|
||||
}
|
||||
|
||||
template<size_t N>
|
||||
void AddArrays(std::array<uint64_t, N>& to, const std::array<uint64_t, N>& from) {
|
||||
for (size_t i = 0; i < N; i++) to[i] += from[i];
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t Histogram::UpperBound(size_t bucket) {
|
||||
return Bounds()[std::min(bucket, BUCKETS - 1)];
|
||||
}
|
||||
|
||||
size_t Histogram::BucketFor(uint64_t microseconds) {
|
||||
const auto& bounds = Bounds();
|
||||
return static_cast<size_t>(std::lower_bound(bounds.begin(), bounds.end(), microseconds) - bounds.begin());
|
||||
}
|
||||
|
||||
void Histogram::Add(uint64_t microseconds, uint32_t count) {
|
||||
m_Counts[BucketFor(microseconds)] += count;
|
||||
m_Count += count;
|
||||
m_Sum += microseconds * count;
|
||||
}
|
||||
|
||||
void Histogram::AddBucket(size_t bucket, uint32_t count) {
|
||||
if (bucket >= BUCKETS) return;
|
||||
m_Counts[bucket] += count;
|
||||
m_Count += count;
|
||||
}
|
||||
|
||||
void Histogram::Merge(const Histogram& other) {
|
||||
for (size_t i = 0; i < BUCKETS; i++) m_Counts[i] += other.m_Counts[i];
|
||||
m_Count += other.m_Count;
|
||||
m_Sum += other.m_Sum;
|
||||
}
|
||||
|
||||
uint64_t Histogram::Percentile(double fraction) const {
|
||||
if (m_Count == 0) return 0;
|
||||
fraction = std::clamp(fraction, 0.0, 1.0);
|
||||
// The rank of the value wanted, 1-based: the smallest value is rank 1
|
||||
const double rank = std::max(1.0, std::ceil(fraction * static_cast<double>(m_Count)));
|
||||
uint64_t seen = 0;
|
||||
for (size_t i = 0; i < BUCKETS; i++) {
|
||||
if (m_Counts[i] == 0) continue;
|
||||
if (static_cast<double>(seen + m_Counts[i]) >= rank) {
|
||||
const double lower = i == 0 ? 0.0 : static_cast<double>(UpperBound(i - 1));
|
||||
// The overflow bucket has no upper bound: report its lower one
|
||||
if (i == BUCKETS - 1) return static_cast<uint64_t>(lower);
|
||||
const double upper = static_cast<double>(UpperBound(i));
|
||||
const double within = (rank - static_cast<double>(seen)) / static_cast<double>(m_Counts[i]);
|
||||
return static_cast<uint64_t>(std::llround(lower + (upper - lower) * within));
|
||||
}
|
||||
seen += m_Counts[i];
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
std::vector<std::pair<uint8_t, uint32_t>> Histogram::Sparse() const {
|
||||
std::vector<std::pair<uint8_t, uint32_t>> out;
|
||||
for (size_t i = 0; i < BUCKETS; i++) {
|
||||
if (m_Counts[i]) out.emplace_back(static_cast<uint8_t>(i), m_Counts[i]);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
Histogram Histogram::FromSparse(const std::vector<std::pair<uint8_t, uint32_t>>& sparse, uint64_t sum) {
|
||||
Histogram out;
|
||||
for (const auto& [bucket, count] : sparse) out.AddBucket(bucket, count);
|
||||
out.m_Sum = sum;
|
||||
return out;
|
||||
}
|
||||
|
||||
size_t StatusClass(uint16_t status) {
|
||||
if (status >= 100 && status < 500) return status / 100 - 1;
|
||||
return 4;
|
||||
}
|
||||
|
||||
uint64_t MessageKey::Packed() const {
|
||||
return (static_cast<uint64_t>(outbound) << 63) | (static_cast<uint64_t>(service & 0x7FFF) << 48) |
|
||||
(static_cast<uint64_t>(packet) << 16) | gameMessage;
|
||||
}
|
||||
|
||||
MessageKey MessageKey::Unpack(uint64_t packed) {
|
||||
MessageKey key;
|
||||
key.outbound = (packed >> 63) != 0;
|
||||
key.service = static_cast<uint16_t>((packed >> 48) & 0x7FFF);
|
||||
if (key.service == (RAKNET & 0x7FFF)) key.service = RAKNET;
|
||||
key.packet = static_cast<uint32_t>(packed >> 16);
|
||||
key.gameMessage = static_cast<uint16_t>(packed);
|
||||
return key;
|
||||
}
|
||||
|
||||
MessageKey KeyOf(const uint8_t* data, size_t length, bool outbound) {
|
||||
MessageKey key;
|
||||
key.outbound = outbound;
|
||||
if (!data || length == 0) {
|
||||
key.service = MessageKey::RAKNET;
|
||||
return key;
|
||||
}
|
||||
// LU packets: ID_USER_PACKET_ENUM, uint16 service, uint32 packet ID, one padding byte
|
||||
if (data[0] != ID_USER_PACKET_ENUM || length < 8) {
|
||||
key.service = MessageKey::RAKNET;
|
||||
key.packet = data[0];
|
||||
return key;
|
||||
}
|
||||
key.service = static_cast<uint16_t>(data[1] | (data[2] << 8));
|
||||
key.packet = static_cast<uint32_t>(data[3]) | (static_cast<uint32_t>(data[4]) << 8) | (static_cast<uint32_t>(data[5]) << 16) | (static_cast<uint32_t>(data[6]) << 24);
|
||||
// Game messages: the header, the target object (8 bytes), then the uint16 game message ID
|
||||
const bool gameMessage = (key.service == static_cast<uint16_t>(ServiceType::WORLD) && key.packet == static_cast<uint32_t>(MessageType::World::GAME_MSG)) ||
|
||||
(key.service == static_cast<uint16_t>(ServiceType::CLIENT) && key.packet == static_cast<uint32_t>(MessageType::Client::GAME_MSG));
|
||||
if (gameMessage && length >= 18) key.gameMessage = static_cast<uint16_t>(data[16] | (data[17] << 8));
|
||||
return key;
|
||||
}
|
||||
|
||||
void Second::Merge(const Second& other) {
|
||||
packetsIn += other.packetsIn;
|
||||
packetsOut += other.packetsOut;
|
||||
bytesIn += other.bytesIn;
|
||||
bytesOut += other.bytesOut;
|
||||
httpRequests += other.httpRequests;
|
||||
AddArrays(httpStatus, other.httpStatus);
|
||||
httpBytesOut += other.httpBytesOut;
|
||||
httpLatency.Merge(other.httpLatency);
|
||||
}
|
||||
|
||||
void RouteStats::Merge(const RouteStats& other) {
|
||||
count += other.count;
|
||||
AddArrays(status, other.status);
|
||||
bytesOut += other.bytesOut;
|
||||
latency.Merge(other.latency);
|
||||
}
|
||||
|
||||
void Recorder::Packet(int64_t now, const MessageKey& key, uint64_t bytes, uint32_t fanout) {
|
||||
if (fanout == 0) return;
|
||||
std::lock_guard lock(m_Mutex);
|
||||
auto& second = SecondAt(now);
|
||||
if (key.outbound) {
|
||||
second.packetsOut += fanout;
|
||||
second.bytesOut += bytes * fanout;
|
||||
} else {
|
||||
second.packetsIn += fanout;
|
||||
second.bytesIn += bytes * fanout;
|
||||
}
|
||||
auto& message = m_Messages[key.Packed()];
|
||||
message.key = key;
|
||||
message.count += fanout;
|
||||
message.bytes += bytes * fanout;
|
||||
}
|
||||
|
||||
void Recorder::Http(int64_t now, const std::string& route, uint16_t status, uint64_t microseconds, uint64_t bytesOut) {
|
||||
std::lock_guard lock(m_Mutex);
|
||||
auto& second = SecondAt(now);
|
||||
second.httpRequests++;
|
||||
second.httpStatus[StatusClass(status)]++;
|
||||
second.httpBytesOut += bytesOut;
|
||||
second.httpLatency.Add(microseconds);
|
||||
|
||||
auto it = m_Routes.find(route);
|
||||
if (it == m_Routes.end()) {
|
||||
const bool full = m_Routes.size() >= MAX_ROUTES;
|
||||
it = m_Routes.try_emplace(full ? std::string("other") : route).first;
|
||||
it->second.route = it->first;
|
||||
}
|
||||
it->second.count++;
|
||||
it->second.status[StatusClass(status)]++;
|
||||
it->second.bytesOut += bytesOut;
|
||||
it->second.latency.Add(microseconds);
|
||||
}
|
||||
|
||||
Second& Recorder::SecondAt(int64_t now) {
|
||||
// Nearly every packet falls in the same second as the one before
|
||||
if (m_Current && m_CurrentTime == now) return *m_Current;
|
||||
auto& second = m_Seconds[now];
|
||||
second.time = now;
|
||||
m_Current = &second;
|
||||
m_CurrentTime = now;
|
||||
return second;
|
||||
}
|
||||
|
||||
void Recorder::SetGauge(const std::string& name, std::function<double()> source) {
|
||||
std::lock_guard lock(m_Mutex);
|
||||
for (auto& gauge : m_Gauges) {
|
||||
if (gauge.first == name) {
|
||||
gauge.second = std::move(source);
|
||||
return;
|
||||
}
|
||||
}
|
||||
m_Gauges.emplace_back(name, std::move(source));
|
||||
}
|
||||
|
||||
bool Recorder::Due(int64_t now, int64_t interval) {
|
||||
std::lock_guard lock(m_Mutex);
|
||||
if (m_LastTake == 0) {
|
||||
m_LastTake = now;
|
||||
return false;
|
||||
}
|
||||
return now - m_LastTake >= interval;
|
||||
}
|
||||
|
||||
Report Recorder::Take(int64_t now) {
|
||||
Report report;
|
||||
std::vector<std::pair<std::string, std::function<double()>>> gauges;
|
||||
{
|
||||
std::lock_guard lock(m_Mutex);
|
||||
m_LastTake = now;
|
||||
// Fill every second from the last report to the one before now; after a long silence only the last MAX_GAP
|
||||
int64_t from = m_LastReported ? m_LastReported + 1 : (m_Seconds.empty() ? now : std::min(m_Seconds.begin()->first, now - 1));
|
||||
from = std::max(from, now - MAX_GAP);
|
||||
for (int64_t t = from; t < now; t++) {
|
||||
const auto it = m_Seconds.find(t);
|
||||
if (it != m_Seconds.end()) report.seconds.push_back(std::move(it->second));
|
||||
else report.seconds.push_back(Second{ .time = t });
|
||||
}
|
||||
m_Seconds.erase(m_Seconds.begin(), m_Seconds.lower_bound(now));
|
||||
m_Current = nullptr;
|
||||
if (now - 1 > m_LastReported) m_LastReported = now - 1;
|
||||
|
||||
std::vector<MessageCount> messages;
|
||||
messages.reserve(m_Messages.size());
|
||||
for (auto& [_, count] : m_Messages) messages.push_back(count);
|
||||
m_Messages.clear();
|
||||
report.messages = Top(messages, TOP_MESSAGES);
|
||||
|
||||
for (auto& [_, route] : m_Routes) report.routes.push_back(std::move(route));
|
||||
m_Routes.clear();
|
||||
gauges = m_Gauges;
|
||||
}
|
||||
for (const auto& [name, source] : gauges) report.gauges.emplace_back(name, source ? source() : 0.0);
|
||||
return report;
|
||||
}
|
||||
|
||||
std::vector<MessageCount> Top(const std::vector<MessageCount>& counts, size_t limit) {
|
||||
std::vector<MessageCount> out;
|
||||
for (const bool outbound : { false, true }) {
|
||||
std::vector<MessageCount> direction;
|
||||
for (const auto& count : counts) if (count.key.outbound == outbound) direction.push_back(count);
|
||||
std::sort(direction.begin(), direction.end(), [](const MessageCount& a, const MessageCount& b) {
|
||||
return a.count != b.count ? a.count > b.count : a.key.Packed() < b.key.Packed();
|
||||
});
|
||||
if (direction.size() > limit) direction.resize(limit);
|
||||
out.insert(out.end(), direction.begin(), direction.end());
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
Recorder& Local() {
|
||||
static Recorder recorder;
|
||||
return recorder;
|
||||
}
|
||||
|
||||
int64_t Now() {
|
||||
return static_cast<int64_t>(std::time(nullptr));
|
||||
}
|
||||
}
|
||||
173
dCommon/TrafficStats.h
Normal file
173
dCommon/TrafficStats.h
Normal file
@@ -0,0 +1,173 @@
|
||||
#pragma once
|
||||
|
||||
#include <array>
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
#include <map>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <unordered_map>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
/**
|
||||
* Traffic diagnostics every server keeps about itself: packets and bytes in and out per second, the packet and game
|
||||
* message types they were, and (servers with a web server) HTTP requests with their latency. Counting is a few
|
||||
* additions under an uncontended lock; every few seconds dServer takes a Report and ships it to the dashboard through
|
||||
* master (see docs/Dashboard.md, "Traffic diagnostics").
|
||||
*
|
||||
* Seconds are Unix seconds, so reports from different servers line up.
|
||||
*/
|
||||
namespace TrafficStats {
|
||||
/**
|
||||
* Latency histogram with fixed, geometric buckets (three per doubling, from 100 microseconds), so histograms from
|
||||
* different seconds and servers add up and percentiles of a minute or an hour come out right.
|
||||
*/
|
||||
class Histogram {
|
||||
public:
|
||||
static constexpr size_t BUCKETS = 58; // the last one holds everything over ~41 s
|
||||
|
||||
// Largest value (microseconds) bucket i holds; the last bucket has no limit (UINT64_MAX)
|
||||
static uint64_t UpperBound(size_t bucket);
|
||||
static size_t BucketFor(uint64_t microseconds);
|
||||
|
||||
void Add(uint64_t microseconds, uint32_t count = 1);
|
||||
void Merge(const Histogram& other);
|
||||
void AddBucket(size_t bucket, uint32_t count);
|
||||
|
||||
uint64_t Count() const { return m_Count; }
|
||||
uint64_t Sum() const { return m_Sum; } // microseconds, all values together
|
||||
void SetSum(uint64_t sum) { m_Sum = sum; }
|
||||
bool Empty() const { return m_Count == 0; }
|
||||
uint32_t At(size_t bucket) const { return m_Counts[bucket]; }
|
||||
|
||||
// The value below which `fraction` (0..1) of the values are, interpolated within its bucket; 0 when empty
|
||||
uint64_t Percentile(double fraction) const;
|
||||
|
||||
// Non-empty buckets as (bucket, count), for the wire and for keeping many histograms small
|
||||
std::vector<std::pair<uint8_t, uint32_t>> Sparse() const;
|
||||
static Histogram FromSparse(const std::vector<std::pair<uint8_t, uint32_t>>& sparse, uint64_t sum);
|
||||
|
||||
private:
|
||||
std::array<uint32_t, BUCKETS> m_Counts{};
|
||||
uint64_t m_Count{};
|
||||
uint64_t m_Sum{};
|
||||
};
|
||||
|
||||
// HTTP status classes 1xx..5xx as index 0..4 (anything else counts as 5xx)
|
||||
size_t StatusClass(uint16_t status);
|
||||
|
||||
// What a packet was: its LU service and packet ID, and for game messages the game message ID
|
||||
struct MessageKey {
|
||||
bool outbound{};
|
||||
uint16_t service{}; // ServiceType; RAKNET for RakNet's own messages (no LU header)
|
||||
uint32_t packet{}; // LU packet ID, or the RakNet message ID
|
||||
uint16_t gameMessage{}; // game messages only
|
||||
|
||||
static constexpr uint16_t RAKNET = 0xFFFF;
|
||||
|
||||
uint64_t Packed() const;
|
||||
static MessageKey Unpack(uint64_t packed);
|
||||
bool operator==(const MessageKey&) const = default;
|
||||
};
|
||||
|
||||
// Reads the key from a packet's first bytes (never reads past `length`)
|
||||
MessageKey KeyOf(const uint8_t* data, size_t length, bool outbound);
|
||||
|
||||
struct MessageCount {
|
||||
MessageKey key;
|
||||
uint64_t count{};
|
||||
uint64_t bytes{};
|
||||
};
|
||||
|
||||
// One second of traffic
|
||||
struct Second {
|
||||
int64_t time{};
|
||||
uint64_t packetsIn{};
|
||||
uint64_t packetsOut{};
|
||||
uint64_t bytesIn{};
|
||||
uint64_t bytesOut{};
|
||||
uint64_t httpRequests{};
|
||||
std::array<uint64_t, 5> httpStatus{}; // by StatusClass
|
||||
uint64_t httpBytesOut{};
|
||||
Histogram httpLatency;
|
||||
|
||||
void Merge(const Second& other); // adds the counts (keeps this one's time)
|
||||
bool Idle() const { return packetsIn == 0 && packetsOut == 0 && httpRequests == 0; }
|
||||
};
|
||||
|
||||
struct RouteStats {
|
||||
std::string route; // "GET /api/players/:id"
|
||||
uint64_t count{};
|
||||
std::array<uint64_t, 5> status{};
|
||||
uint64_t bytesOut{};
|
||||
Histogram latency;
|
||||
|
||||
void Merge(const RouteStats& other);
|
||||
};
|
||||
|
||||
// RakNet's view of the connections (all datagrams, acknowledgements and resends included), over the report
|
||||
struct Link {
|
||||
uint32_t connections{};
|
||||
uint64_t datagramsSent{};
|
||||
uint64_t datagramsReceived{};
|
||||
uint64_t bytesSent{};
|
||||
uint64_t bytesReceived{};
|
||||
uint64_t resends{};
|
||||
uint32_t resendQueue{}; // messages waiting to be resent now
|
||||
uint32_t averagePingMs{}; // over the connections
|
||||
};
|
||||
|
||||
struct Report {
|
||||
std::vector<Second> seconds; // oldest first, one per second without gaps
|
||||
std::vector<MessageCount> messages; // the busiest types per direction
|
||||
std::vector<RouteStats> routes;
|
||||
Link link;
|
||||
std::vector<std::pair<std::string, double>> gauges; // e.g. workers_busy
|
||||
};
|
||||
|
||||
class Recorder {
|
||||
public:
|
||||
static constexpr size_t TOP_MESSAGES = 24; // per direction, per report
|
||||
static constexpr size_t MAX_ROUTES = 64; // distinct routes per report; more count as "other"
|
||||
static constexpr int64_t MAX_GAP = 120; // seconds of silence a report fills in at most
|
||||
|
||||
// `fanout`: how many connections a broadcast went to
|
||||
void Packet(int64_t now, const MessageKey& key, uint64_t bytes, uint32_t fanout = 1);
|
||||
void Http(int64_t now, const std::string& route, uint16_t status, uint64_t microseconds, uint64_t bytesOut);
|
||||
|
||||
// Evaluated when a report is taken (on the thread that takes it)
|
||||
void SetGauge(const std::string& name, std::function<double()> source);
|
||||
|
||||
/**
|
||||
* The seconds before `now` not reported yet (silent ones as zeros, at most MAX_GAP of them), the busiest
|
||||
* message types and the routes since the last report. The current second stays for the next one.
|
||||
*/
|
||||
Report Take(int64_t now);
|
||||
|
||||
// Whether `interval` seconds passed since the last Take (the first call only starts the clock)
|
||||
bool Due(int64_t now, int64_t interval);
|
||||
|
||||
private:
|
||||
Second& SecondAt(int64_t now); // under m_Mutex
|
||||
|
||||
std::mutex m_Mutex;
|
||||
std::map<int64_t, Second> m_Seconds;
|
||||
Second* m_Current{}; // m_Seconds[m_CurrentTime] (map entries stay put)
|
||||
int64_t m_CurrentTime{};
|
||||
int64_t m_LastReported{}; // last second a report covered
|
||||
std::unordered_map<uint64_t, MessageCount> m_Messages;
|
||||
std::map<std::string, RouteStats> m_Routes;
|
||||
std::vector<std::pair<std::string, std::function<double()>>> m_Gauges;
|
||||
int64_t m_LastTake{};
|
||||
};
|
||||
|
||||
// The busiest `limit` message types of each direction, busiest first
|
||||
std::vector<MessageCount> Top(const std::vector<MessageCount>& counts, size_t limit);
|
||||
|
||||
// This process's recorder (dServer counts packets into it, the web server requests)
|
||||
Recorder& Local();
|
||||
|
||||
int64_t Now(); // Unix seconds
|
||||
}
|
||||
@@ -63,5 +63,8 @@ namespace MessageType {
|
||||
MIGRATE_STATUS,
|
||||
// Source world -> master -> target world: what a moved player had that isn't in their saved character
|
||||
MIGRATE_PLAYER_STATE,
|
||||
|
||||
// Any server -> master -> dashboard: traffic counters of the last few seconds (see ServerTraffic.h)
|
||||
SERVER_TRAFFIC,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user