Realm queue is now responsible for active session tracking (#293)

All client authentication requests must go via the realm queue, allowing it to better manage player counts and apply back-pressure if required in the future.

Author: Chaosvex

src/realm/ClientContext.cpp

@@ -23,7 +23,8 @@ ClientContext::ClientContext(executor& executor, const ConfigStore& cfg_store, E
 	, account_rpc(account_rpc)
 	, character_rpc(character_rpc)
 	, realm_rpc(realm_rpc)
-	, logger(logger) {}
+	, logger(logger)
+	, active(false) {}
 
 void ClientContext::start_timer(const std::chrono::milliseconds& time) {
 	timer_.expires_after(time);
@@ -60,6 +61,14 @@ bool ClientContext::packet_logging() const {
 	return connection_->packet_logging();
 }
 
+void ClientContext::set_active() {
+	active = true;
+}
+
+bool ClientContext::is_active() const {
+	return active;
+}
+
 /*
  * Helper that decides whether to print the IP address or username
  * and IP address in log outputs, based on whether authentication

src/realm/ClientContext.h

@@ -100,6 +100,10 @@ class ClientContext {
 		connection_->send(packet);
 	}
 
+	// if the client is occupying a slot
+	void set_active();
+	bool is_active() const;
+
 	StateContext state_ctx;
 	std::optional<AccountInfo> account;
 	const ConfigStore& cfg_store;
@@ -109,6 +113,7 @@ class ClientContext {
 	CharacterClient& character_rpc;
 	const RealmService& realm_rpc;
 	log::Logger& logger;
+	bool active;
 
 	friend class ClientHandler;
 };

src/realm/RealmQueue.cpp

@@ -51,9 +51,16 @@ void RealmQueue::update_clients() {
 	dirty_ = false;
 }
 
-void RealmQueue::enqueue(ClientIdent client, int priority) {
+auto RealmQueue::enqueue(const ClientIdent& client, int priority) -> Result {
 	std::lock_guard guard(lock_);
 
+	const auto max_slots = cstore_.config().max_slots;
+
+	if(active_ < max_slots && queue_.empty()) {
+		++active_;
+		return Result::success;
+	}
+
 	if(queue_.empty()) {
 		set_timer();
 	}
@@ -64,6 +71,7 @@ void RealmQueue::enqueue(ClientIdent client, int priority) {
 	// but allows for multiple priority levels without multiple hard-coded queues
 	queue_.sort();
 	dirty_ = true;
+	return Result::queued;
 }
 
 /* 
@@ -75,9 +83,8 @@ void RealmQueue::dequeue(const ClientIdent& client) {
 
 	if(auto it = std::ranges::find(queue_, client, &QueueEntry::client); it != queue_.end()) {
 		queue_.erase(it);
+		dirty_ = true;
 	}
-
-	dirty_ = true;
 }
 
 /* 
@@ -87,6 +94,8 @@ void RealmQueue::dequeue(const ClientIdent& client) {
 void RealmQueue::free_slot() {
 	std::lock_guard guard(lock_);
 
+	--active_;
+
 	if(queue_.empty()) {
 		return;
 	}
@@ -119,6 +128,7 @@ void RealmQueue::shutdown() {
 }
 
 std::size_t RealmQueue::size() const {
+	std::lock_guard guard(lock_);
 	return queue_.size();
 }
 

src/realm/RealmQueue.h

@@ -8,6 +8,7 @@
 
 #pragma once
 
+#include "ConfigStore.h"
 #include <shared/ClientIdent.h>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/steady_timer.hpp>
@@ -45,27 +46,37 @@ class RealmQueue final {
 	boost::asio::steady_timer timer_;
 	std::list<QueueEntry> queue_;
 	EventDispatcher& dispatcher_;
-	std::mutex lock_;
+	mutable std::mutex lock_;
 	std::atomic_bool dirty_;
+	std::size_t active_;
+	const ConfigStore& cstore_;
 
 	bool empty();
 	void update_clients();
 	void set_timer();
 
 public:
 	constexpr static std::size_t npos = 0;
+	
+	enum class Result {
+		success,
+		queued,
+	};
 
 	explicit RealmQueue(boost::asio::io_context& service,
 	                    EventDispatcher& dispatcher,
+	                    ConfigStore& config_store,
 	                    std::chrono::milliseconds frequency = default_frequency)
 		: frequency_(frequency)
 		, timer_(service)
 		, dispatcher_(dispatcher)
-		, dirty_(false) { }
+		, dirty_(false)
+		, active_(0)
+		, cstore_(config_store) { }
 
 	~RealmQueue();
 
-	void enqueue(ClientIdent client, int priority = 0);
+	Result enqueue(const ClientIdent& client, int priority = 0);
 	void dequeue(const ClientIdent& client);
 	std::size_t poll(const ClientIdent& client);
 	void free_slot();

src/realm/Service.cpp

@@ -233,7 +233,7 @@ void Service::initialise(const opts::variables_map& args) try {
 	const auto nsd_port = args["nsd.port"].as<std::uint16_t>();
 
 	ctx->rpc_discovery = std::make_unique<NetworkServiceDiscovery>(*ctx->rpc, nsd_host, nsd_port, logger);
-	ctx->queue = std::make_unique<RealmQueue>(service, *ctx->dispatcher);
+	ctx->queue = std::make_unique<RealmQueue>(service, *ctx->dispatcher, *ctx->config_store);
 
 	// Misc. information
 	const auto max_socks = utility::max_sockets_desc();

src/realm/commands/Ungrouped.cpp

@@ -25,12 +25,27 @@ auto uptime(utility::CommandExecutor& exec, std::chrono::steady_clock::time_poin
 	));
 }
 
+auto queue(utility::CommandExecutor& exec, const RealmQueue& realm_queue, log::Logger& logger) 
+-> std::shared_ptr<commands::Command> {
+	return commands::create("queue")
+		->description("Display authentication queue size")
+		->handler(exec([&](auto&) {
+			LOG_CONSOLE(logger, "Current realm queue size is {}", realm_queue.size());
+		}
+	));
+}
+
 } // unnamed
 
 void add_ungrouped_commands(ServiceContext& context, commands::Command& registry, log::Logger& logger) {
 	auto impl = context.get();
-	auto scoped = registry.scoped_insert(uptime(*impl->cmd_exec, impl->start_time, logger));
-	impl->commands.emplace_back(std::move(scoped));
+	auto& realm_queue = *impl->queue;
+
+	auto command = uptime(*impl->cmd_exec, impl->start_time, logger);
+	impl->commands.emplace_back(registry.scoped_insert(std::move(command)));
+
+	command = queue(*impl->cmd_exec, realm_queue, logger);
+	impl->commands.emplace_back(registry.scoped_insert(std::move(command)));
 }
 
 } // realm, ember

src/realm/states/Authentication.cpp

@@ -39,7 +39,6 @@ void send_auth_challenge(ClientContext& ctx);
 void send_auth_result(ClientContext& ctx, protocol::Result result);
 void handle_authentication(ClientContext& ctx);
 void handle_queue_update(ClientContext& ctx, const QueuePosition& event);
-void handle_queue_success(ClientContext& ctx);
 void auth_success(ClientContext& ctx);
 void auth_queue(ClientContext& ctx);
 void prove_session(ClientContext& ctx, const Botan::BigInt& key);
@@ -197,18 +196,13 @@ void prove_session(ClientContext& ctx, const Botan::BigInt& key) {
 		.username = auth_ctx.packet->username
 	};
 
-	 // todo, allowing for multiple realms to connect to a single world server
-	 // will require an external service to keep track of available slots
-	unsigned int active_players = 0;
-	const auto& config = ctx.cfg_store.config_tls();
-
-	if(active_players < config.max_slots) {
+	ctx.cancel_timer();
+	
+	if(ctx.queue.enqueue(ctx.handler().uuid(), 0) == RealmQueue::Result::success) {
 		auth_success(ctx);
 	} else {
-		auth_queue(ctx);
+		auth_state(ctx, State::in_queue);
 	}
-	
-	++active_players;
 }
 
 void send_auth_challenge(ClientContext& ctx) {
@@ -248,26 +242,14 @@ void send_addon_data(ClientContext& ctx) {
 	ctx.send(response);
 }
 
-void auth_queue(ClientContext& ctx) {
-	LOG_TRACE(ctx.logger, log_func);
-
-	const auto& uuid = ctx.handler().uuid();
-	auto& dispatcher = ctx.dispatcher;
-
-	ctx.queue.enqueue(uuid);
-	ctx.cancel_timer();
-	auth_state(ctx, State::in_queue);
-
-	CLIENT_DEBUG(ctx, "Added to queue, position {}", ctx.queue.poll(uuid));
-}
-
 void auth_success(ClientContext& ctx) {
 	LOG_TRACE(ctx.logger, log_func);
 
 	send_auth_result(ctx, protocol::Result::auth_ok);
 	send_addon_data(ctx);
 	auth_state(ctx, State::success);
 	ctx.state_update(ClientState::cs_character_list);
+	ctx.set_active();
 
 	CLIENT_DEBUG(ctx, "Authenticated as {}", ctx.account->username);
 }
@@ -333,7 +315,7 @@ void handle_event(ClientContext& ctx, const Event& event) {
 			handle_queue_update(ctx, event.as<QueuePosition>());
 			break;
 		case queue_success:
-			handle_queue_success(ctx);
+			auth_success(ctx);
 			break;
 		default:
 			break;

src/realm/states/SessionClose.cpp

@@ -10,11 +10,16 @@
 #include "../ClientContext.h"
 #include "../ClientHandler.h"
 #include "../Events.h"
+#include "../RealmQueue.h"
 
 namespace ember::realm::session_close {
 
 void enter(ClientContext& ctx) {
 	ctx.handler().log_redirect_stop();
+
+	if(ctx.is_active()) {
+		ctx.queue.free_slot();
+	}
 }
 
 void handle_packet(ClientContext& ctx, protocol::ClientOpcode opcode) {