TeaSpeak/Teaspeak-Server
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

A lot of query reworking

Browse Source
This commit is contained in:
WolverinDEV 2021-01-28 20:59:15 +01:00
parent 9a5aa1d42b
commit 902b1f3511
58 changed files with 1615 additions and 1012 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
git-teaspeak

@ -1 +1 @@
Subproject commit 2e7730459d6941a97f7bc2b9b50b678ab195a41d
Subproject commit 8e7874872ebb2c19dda702bc68372da5eea77e99

6
server/CMakeLists.txt
View File

@ -57,7 +57,6 @@ set(SERVER_SOURCE_FILES
src/server/VoiceServer.cpp
src/server/POWHandler.cpp
src/client/voice/VoiceClientConnection.cpp
#src/client/ConnectedClientCommandHandler.cpp
src/client/command_handler/channel.cpp
src/client/command_handler/client.cpp
src/client/command_handler/server.cpp
@ -71,8 +70,6 @@ set(SERVER_SOURCE_FILES
src/Group.cpp
src/manager/BanManager.cpp
src/client/InternalClient.cpp
#src/weblist/WeblistClient.cpp
#src/weblist/WebList.cpp
src/client/DataClient.cpp
src/server/QueryServer.cpp
@ -150,7 +147,8 @@ set(SERVER_SOURCE_FILES
src/client/voice/PacketDecoder.cpp
src/client/voice/PacketEncoder.cpp
src/client/voice/ServerCommandExecutor.cpp
src/client/shared/ServerCommandExecutor.cpp
src/client/shared/RawCommand.cpp
src/client/voice/PingHandler.cpp
src/client/voice/CryptSetupHandler.cpp
src/client/shared/WhisperHandler.cpp

4
server/helpers/permgen.cpp
View File

@ -135,7 +135,7 @@ int main(int argc, char** argv) {
group.target = line == "2" ? TARGET_QUERY : TARGET_SERVER;
read_line(file, line);
auto data = "perms " + line;
ts::Command group_parms = ts::Command::parse(pipes::buffer_view(data.data(), data.length()));
ts::Command group_parms = ts::Command::parse(data);
map<permission::PermissionType, permission::update::UpdatePermission> grantMapping;
for (int index = 0; index < group_parms.bulkCount(); index++) {
@ -184,7 +184,7 @@ int main(int argc, char** argv) {
group.target = TARGET_CHANNEL;
read_line(file, line);
auto data = "perms " + line;
ts::Command group_parms = ts::Command::parse(pipes::buffer_view(data.data(), data.length()));
ts::Command group_parms = ts::Command::parse(data);
map<permission::PermissionType, permission::update::UpdatePermission> grantMapping;
for (int index = 0; index < group_parms.bulkCount(); index++) {

33
server/src/Configuration.cpp
View File

@ -92,6 +92,7 @@ bool config::voice::allow_session_reinitialize;
std::string config::query::motd;
std::string config::query::newlineCharacter;
size_t config::query::max_line_buffer;
int config::query::sslMode;
std::string config::query::ssl::certFile;
std::string config::query::ssl::keyFile;
@ -121,8 +122,7 @@ std::string config::messages::timeout::packet_resend_failed;
std::string config::messages::timeout::connection_reinitialized;
size_t config::threads::ticking;
size_t config::threads::voice::execute_limit;
size_t config::threads::voice::execute_per_server;
size_t config::threads::command_execute;
size_t config::threads::voice::events_per_server;
size_t config::threads::voice::io_min;
size_t config::threads::voice::io_per_server;
@ -1121,13 +1121,18 @@ std::deque<std::shared_ptr<EntryBinding>> config::create_bindings() {
{
BIND_GROUP(query);
{
CREATE_BINDING("nl_char", 0);
CREATE_BINDING("nl_char", FLAG_RELOADABLE);
BIND_STRING(config::query::newlineCharacter, "\n");
ADD_DESCRIPTION("Change the query newline character");
ADD_NOTE("NOTE: TS3 - Compatible bots may require \"\n\r\"");
}
{
CREATE_BINDING("motd", 0);
CREATE_BINDING("max_line_buffer", FLAG_RELOADABLE);
BIND_INTEGRAL(config::query::max_line_buffer, 1024 * 1024, 1024 * 8, 1024 * 1024 * 512);
ADD_DESCRIPTION("Max number of characters one query command could contain.");
}
{
CREATE_BINDING("motd", FLAG_RELOADABLE);
BIND_STRING(config::query::motd, "TeaSpeak\nWelcome on the TeaSpeak ServerQuery interface.\n");
ADD_DESCRIPTION("The query welcome message");
@ -1143,7 +1148,7 @@ std::deque<std::shared_ptr<EntryBinding>> config::create_bindings() {
ADD_DESCRIPTION("Available modes:");
ADD_DESCRIPTION(" 0: Disabled");
ADD_DESCRIPTION(" 1: Enabled (Enforced encryption)");
ADD_DESCRIPTION(" 2: Hybrid (Prefer encryption but fallback when it isnt available)");
ADD_DESCRIPTION(" 2: Hybrid (Prefer encryption but fallback when it isn't available)");
}
{
BIND_GROUP(ssl);
@ -1825,6 +1830,12 @@ std::deque<std::shared_ptr<EntryBinding>> config::create_bindings() {
ADD_DESCRIPTION("Thread pool size for the ticking task of a VirtualServer");
ADD_SENSITIVE();
}
{
CREATE_BINDING("command_execute", 0);
BIND_INTEGRAL(config::threads::command_execute, 4, 1, 128);
ADD_DESCRIPTION("Command executors");
ADD_SENSITIVE();
}
{
BIND_GROUP(voice)
{
@ -1833,18 +1844,6 @@ std::deque<std::shared_ptr<EntryBinding>> config::create_bindings() {
ADD_DESCRIPTION("Kernel events per server");
ADD_SENSITIVE();
}
{
CREATE_BINDING("execute_per_server", 0);
BIND_INTEGRAL(config::threads::voice::execute_per_server, 2, 1, 128);
ADD_DESCRIPTION("Threads per server for command executing");
ADD_SENSITIVE();
}
{
CREATE_BINDING("execute_limit", 0);
BIND_INTEGRAL(config::threads::voice::execute_limit, 5, 1, 1024);
ADD_DESCRIPTION("Max number of threads for command handling threads within the instance");
ADD_SENSITIVE();
}
{
CREATE_BINDING("io_min", 0);
BIND_INTEGRAL(config::threads::voice::io_min, 2, 1, 1024);

5
server/src/Configuration.h
View File

@ -157,6 +157,7 @@ namespace ts::config {
namespace query {
extern std::string motd;
extern std::string newlineCharacter;
extern size_t max_line_buffer;
extern int sslMode;
namespace ssl {
@ -230,11 +231,9 @@ namespace ts::config {
namespace threads {
extern size_t ticking;
extern size_t command_execute;
namespace voice {
extern size_t execute_per_server;
extern size_t execute_limit;
extern size_t events_per_server;
extern size_t io_min;
extern size_t io_per_server;

10
server/src/InstanceHandler.cpp
View File

@ -25,8 +25,8 @@
#define _POSIX_SOURCE
#endif
#include <unistd.h>
#include <files/FileServer.h>
#include "./manager/ActionLogger.h"
#include "./client/shared/ServerCommandExecutor.h"
#undef _POSIX_SOURCE
@ -251,11 +251,15 @@ inline vector<string> split_hosts(const std::string& message, char delimiter) {
}
bool InstanceHandler::startInstance() {
if (this->active)
if (this->active) {
return false;
}
active = true;
string errorMessage;
this->server_command_executor_ = std::make_shared<ServerCommandExecutor>(ts::config::threads::command_execute);
this->web_list->enabled = ts::config::server::enable_teamspeak_weblist;
this->permission_mapper = make_shared<permission::PermissionNameMapper>();
@ -412,6 +416,7 @@ void InstanceHandler::stopInstance() {
this->activeCon.notify_all();
}
this->web_list->enabled = false;
this->server_command_executor_->shutdown();
threads::MutexLock lock_tick(this->lock_tick);
this->scheduler()->cancelTask(INSTANCE_TICK_NAME);
@ -447,6 +452,7 @@ void InstanceHandler::stopInstance() {
this->web_event_loop = nullptr;
this->license_service_->shutdown();
this->server_command_executor_ = nullptr;
}
void InstanceHandler::tickInstance() {

6
server/src/InstanceHandler.h
View File

@ -31,6 +31,8 @@ namespace ts {
class ActionLogger;
}
class ServerCommandExecutor;
class InstanceHandler {
public:
explicit InstanceHandler(SqlDataManager*);
@ -85,6 +87,8 @@ namespace ts {
std::shared_ptr<permission::PermissionNameMapper> getPermissionMapper() { return this->permission_mapper; }
std::shared_ptr<ts::event::EventExecutor> getConversationIo() { return this->conversation_io; }
[[nodiscard]] inline auto server_command_executor() { return this->server_command_executor_; }
permission::v2::PermissionFlaggedValue calculate_permission(
permission::PermissionType,
ClientDbId,
@ -129,6 +133,8 @@ namespace ts {
ts::Properties* _properties = nullptr;
std::shared_ptr<ServerCommandExecutor> server_command_executor_{};
std::shared_ptr<ts::event::EventExecutor> conversation_io = nullptr;
std::shared_ptr<webio::LoopManager> web_event_loop = nullptr;
std::shared_ptr<weblist::WebListManager> web_list = nullptr;

2
server/src/SignalHandler.cpp
View File

@ -81,7 +81,7 @@ bool ts::syssignal::setup() {
//We cant listen for this signal if stdin ist a atty
SIG(SIGINT, &ts::syssignal::handleStopSignal);
}
SIG(SIGABRT, &ts::syssignal::handleAbortSignal);
//SIG(SIGABRT, &ts::syssignal::handleAbortSignal);
std::set_terminate(ts::syssignal::handleTerminate);
return true;

6
server/src/TS3ServerClientManager.cpp
View File

@ -377,14 +377,16 @@ void VirtualServer::client_move(
std::unique_lock<std::shared_mutex> &server_channel_write_lock) {
TIMING_START(timings);
if(server_channel_write_lock.owns_lock())
if(server_channel_write_lock.owns_lock()) {
server_channel_write_lock.unlock();
}
lock_guard client_command_lock(target->command_lock);
server_channel_write_lock.lock();
TIMING_STEP(timings, "chan tree l");
if(target->currentChannel == target_channel)
if(target->currentChannel == target_channel) {
return;
}
/* first step: resolve the target channel / or fix missing */
auto s_target_channel = dynamic_pointer_cast<ServerChannel>(target_channel);

2
server/src/TS3ServerHeartbeat.cpp
View File

@ -126,7 +126,7 @@ void VirtualServer::executeServerTick() {
if(cl->floodPoints > flood_decrease)
cl->floodPoints -= flood_decrease;
else cl->floodPoints = 0;
cl->tick(tick_client_end);
cl->tick_server(tick_client_end);
auto voice = dynamic_pointer_cast<SpeakingClient>(cl);
if(flag_update_spoken && voice)
this->spoken_time += voice->takeSpokenTime();

2
server/src/VirtualServer.cpp
View File

@ -577,7 +577,7 @@ void VirtualServer::stop(const std::string& reason, bool disconnect_query) {
if(disconnect_query) {
auto qc = dynamic_pointer_cast<QueryClient>(cl);
qc->disconnect_from_virtual_server();
qc->disconnect_from_virtual_server("server disconnect");
}
} else if(cl->getType() == CLIENT_MUSIC) {
cl->disconnect("");

15
server/src/VirtualServerManager.cpp
View File

@ -15,7 +15,6 @@ using namespace ts::server;
VirtualServerManager::VirtualServerManager(InstanceHandler* handle) : handle(handle) {
this->puzzles = new udp::PuzzleManager{};
this->handshakeTickers = new threads::Scheduler(1, "handshake ticker");
this->execute_loop = new event::EventExecutor("executor #");
//this->join_loop = new event::EventExecutor("joiner #");
this->_ioManager = new io::VoiceIOManager();
@ -40,11 +39,6 @@ VirtualServerManager::~VirtualServerManager() {
delete this->puzzles;
this->puzzles = nullptr;
if(this->execute_loop)
this->execute_loop->shutdown();
delete this->execute_loop;
this->execute_loop = nullptr;
if(this->join_loop)
this->join_loop->shutdown();
delete this->join_loop;
@ -62,8 +56,6 @@ VirtualServerManager::~VirtualServerManager() {
}
bool VirtualServerManager::initialize(bool autostart) {
this->execute_loop->initialize(1);
this->state = State::STARTING;
logMessage(LOG_INSTANCE, "Generating server puzzles...");
auto start = system_clock::now();
@ -162,7 +154,6 @@ bool VirtualServerManager::initialize(bool autostart) {
(float) server_count / (time / 1024 == 0 ? 1 : time / 1024)
);
this->handle->databaseHelper()->clearStartupCache(0);
this->adjust_executor_threads();
{
this->acknowledge.executor = std::thread([&]{
@ -370,7 +361,6 @@ shared_ptr<VirtualServer> VirtualServerManager::create_server(std::string hosts,
threads::MutexLock l(this->instanceLock);
this->instances.push_back(server);
}
this->adjust_executor_threads();
return server;
}
@ -388,7 +378,6 @@ bool VirtualServerManager::deleteServer(shared_ptr<VirtualServer> server) {
return s == server;
}), this->instances.end());
}
this->adjust_executor_threads();
if(server->getState() != ServerState::OFFLINE)
server->stop("server deleted", true);
@ -397,7 +386,7 @@ bool VirtualServerManager::deleteServer(shared_ptr<VirtualServer> server) {
cl->close_connection(chrono::system_clock::now());
} else if(cl->getType() == CLIENT_QUERY){
auto qc = dynamic_pointer_cast<QueryClient>(cl);
qc->disconnect_from_virtual_server();
qc->disconnect_from_virtual_server("server delete");
} else if(cl->getType() == CLIENT_MUSIC) {
cl->disconnect("");
cl->currentChannel = nullptr;
@ -473,8 +462,6 @@ void VirtualServerManager::shutdownAll(const std::string& msg) {
for(const auto &server : this->serverInstances()){
if(server->running()) server->stop(msg, true);
}
this->execute_loop->shutdown();
}
void VirtualServerManager::tickHandshakeClients() {

10
server/src/VirtualServerManager.h
View File

@ -71,16 +71,7 @@ namespace ts::server {
udp::PuzzleManager* rsaPuzzles() { return this->puzzles; }
event::EventExecutor* get_join_loop() { return this->join_loop; }
event::EventExecutor* get_executor_loop() { return this->execute_loop; }
inline void adjust_executor_threads() {
std::unique_lock instance_lock(this->instanceLock);
auto instance_count = this->instances.size();
instance_lock.unlock();
auto threads = std::min(config::threads::voice::execute_per_server * instance_count, config::threads::voice::execute_limit);
this->execute_loop->threads(threads);
}
io::VoiceIOManager* ioManager(){ return this->_ioManager; }
/* This must be recursive */
@ -94,7 +85,6 @@ namespace ts::server {
std::deque<std::shared_ptr<VirtualServer>> instances;
udp::PuzzleManager* puzzles{nullptr};
event::EventExecutor* execute_loop = nullptr;
event::EventExecutor* join_loop = nullptr;
threads::Scheduler* handshakeTickers = nullptr;
io::VoiceIOManager* _ioManager = nullptr;

6
server/src/client/ConnectedClient.cpp
View File

@ -38,7 +38,9 @@ ConnectedClient::~ConnectedClient() {
bool ConnectedClient::loadDataForCurrentServer() {
auto result = DataClient::loadDataForCurrentServer();
if(!result) return false;
if(!result) {
return false;
}
this->properties()[property::CONNECTION_CLIENT_IP] = this->getLoggingPeerIp();
return true;
@ -720,7 +722,7 @@ void ConnectedClient::sendChannelDescription(const std::shared_ptr<BasicChannel>
this->sendCommand(cmd, true);
}
void ConnectedClient::tick(const std::chrono::system_clock::time_point &time) {
void ConnectedClient::tick_server(const std::chrono::system_clock::time_point &time) {
ALARM_TIMER(A1, "ConnectedClient::tick", milliseconds(2));
if(this->state == ConnectionState::CONNECTED) {
if(this->requireNeededPermissionResend)

13
server/src/client/ConnectedClient.h
View File

@ -240,7 +240,16 @@ namespace ts {
virtual bool notifyConversationMessageDelete(const ChannelId /* conversation id */, const std::chrono::system_clock::time_point& /* begin timestamp */, const std::chrono::system_clock::time_point& /* begin end */, ClientDbId /* client id */, size_t /* messages */);
/* this method should be callable from everywhere; the method is non blocking! */
/**
* Close the network connection.
*
* Note:
* This method could be called from any thread with any locks in hold.
* It's not blocking.
*
* @param timeout The timestamp when to drop the client if not all data has been send.
* @returns `true` if the connection has been closed and `false` if the connection is already closed.
*/
virtual bool close_connection(const std::chrono::system_clock::time_point &timeout) = 0;
/* this method should be callable from everywhere; the method is non blocking! */
@ -382,7 +391,7 @@ namespace ts {
bool loadDataForCurrentServer() override;
virtual void tick(const std::chrono::system_clock::time_point &time);
virtual void tick_server(const std::chrono::system_clock::time_point &time);
//Locked by everything who has something todo with command handling
threads::Mutex command_lock; /* Note: This mutex must be recursive! */
std::vector<std::function<void()>> postCommandHandler;

4
server/src/client/InternalClient.cpp
View File

@ -34,8 +34,8 @@ bool InternalClient::close_connection(const std::chrono::system_clock::time_poin
return true;
}
void InternalClient::tick(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick(time);
void InternalClient::tick_server(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick_server(time);
}
bool InternalClient::disconnect(const std::string &reason) {

2
server/src/client/InternalClient.h
View File

@ -19,6 +19,6 @@ namespace ts::server {
bool close_connection(const std::chrono::system_clock::time_point& timeout = std::chrono::system_clock::time_point()) override;
bool disconnect(const std::string &reason) override;
protected:
void tick(const std::chrono::system_clock::time_point &time) override;
void tick_server(const std::chrono::system_clock::time_point &time) override;
};
}

4
server/src/client/SpeakingClient.cpp
View File

@ -574,8 +574,8 @@ void SpeakingClient::updateSpeak(bool only_update, const std::chrono::system_clo
}
}
void SpeakingClient::tick(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick(time);
void SpeakingClient::tick_server(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick_server(time);
ALARM_TIMER(A1, "SpeakingClient::tick", milliseconds(2));
this->updateSpeak(true, time);

2
server/src/client/SpeakingClient.h
View File

@ -51,7 +51,7 @@ namespace ts::server {
[[nodiscard]] inline whisper::WhisperHandler& whisper_handler() { return this->whisper_handler_; }
protected:
void tick(const std::chrono::system_clock::time_point &time) override;
void tick_server(const std::chrono::system_clock::time_point &time) override;
public:
void updateChannelClientProperties(bool channel_lock, bool notify) override;

12
server/src/client/command_handler/channel.cpp
View File

@ -526,7 +526,7 @@ command_result ConnectedClient::handleCommandChannelGroupAddPerm(Command &cmd) {
if (!channelGroup || channelGroup->target() != GROUPTARGET_CHANNEL) return command_result{error::parameter_invalid, "invalid channel group id"};
ACTION_REQUIRES_GROUP_PERMISSION(channelGroup, permission::i_channel_group_needed_modify_power, permission::i_channel_group_modify_power, true);
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if (!pparser.validate(this->ref(), 0)) {
return pparser.build_command_result();
}
@ -581,7 +581,7 @@ command_result ConnectedClient::handleCommandChannelGroupDelPerm(Command &cmd) {
if (!channelGroup || channelGroup->target() != GROUPTARGET_CHANNEL) return command_result{error::parameter_invalid, "invalid channel group id"};
ACTION_REQUIRES_GROUP_PERMISSION(channelGroup, permission::i_channel_group_needed_modify_power, permission::i_channel_group_modify_power, true);
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if (!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -2142,7 +2142,7 @@ command_result ConnectedClient::handleCommandChannelAddPerm(Command &cmd) {
ACTION_REQUIRES_CHANNEL_PERMISSION(channel, permission::i_channel_needed_permission_modify_power, permission::i_channel_permission_modify_power, true);
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if (!pparser.validate(this->ref(), channel->channelId())) {
return pparser.build_command_result();
}
@ -2219,7 +2219,7 @@ command_result ConnectedClient::handleCommandChannelDelPerm(Command &cmd) {
ACTION_REQUIRES_CHANNEL_PERMISSION(channel, permission::i_channel_needed_permission_modify_power, permission::i_channel_permission_modify_power, true);
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if (!pparser.validate(this->ref(), channel->channelId()))
return pparser.build_command_result();
@ -2332,7 +2332,7 @@ command_result ConnectedClient::handleCommandChannelClientDelPerm(Command &cmd)
ACTION_REQUIRES_PERMISSION(permission::i_client_permission_modify_power, required_permissions, channel_id);
}
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if (!pparser.validate(this->ref(), channel->channelId()))
return pparser.build_command_result();
@ -2400,7 +2400,7 @@ command_result ConnectedClient::handleCommandChannelClientAddPerm(Command &cmd)
auto required_permissions = this->server->calculate_permission(permission::i_client_needed_permission_modify_power, cmd["cldbid"], ClientType::CLIENT_TEAMSPEAK, channel_id);
ACTION_REQUIRES_PERMISSION(permission::i_client_permission_modify_power, required_permissions, channel_id);
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if (!pparser.validate(this->ref(), channel->channelId()))
return pparser.build_command_result();

4
server/src/client/command_handler/client.cpp
View File

@ -953,7 +953,7 @@ command_result ConnectedClient::handleCommandClientAddPerm(Command &cmd) {
ACTION_REQUIRES_GLOBAL_PERMISSION(permission::i_client_permission_modify_power, this->server->calculate_permission(permission::i_client_needed_permission_modify_power, cldbid, ClientType::CLIENT_TEAMSPEAK, 0));
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -997,7 +997,7 @@ command_result ConnectedClient::handleCommandClientDelPerm(Command &cmd) {
auto mgr = serverInstance->databaseHelper()->loadClientPermissionManager(this->server, cldbid);
ACTION_REQUIRES_GLOBAL_PERMISSION(permission::i_client_permission_modify_power, this->server->calculate_permission(permission::i_client_needed_permission_modify_power, cldbid, ClientType::CLIENT_TEAMSPEAK, 0));
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();

40
server/src/client/command_handler/file.cpp
View File

@ -7,7 +7,6 @@
#include <bitset>
#include <algorithm>
#include <openssl/sha.h>
#include "../../build.h"
#include "../ConnectedClient.h"
#include "../InternalClient.h"
#include "../../server/VoiceServer.h"
@ -24,13 +23,10 @@
#include <files/FileServer.h>
#include <log/LogUtils.h>
#include <misc/sassert.h>
#include <misc/base64.h>
#include <misc/hex.h>
#include <misc/rnd.h>
#include <misc/strobf.h>
#include <bbcode/bbcodes.h>
using namespace std::chrono;
using namespace std;
@ -525,16 +521,18 @@ command_result ConnectedClient::handleCommandFTGetFileInfo(ts::Command &cmd) {
}
}
if(!info_response->wait_for(kFileAPITimeout))
if(!info_response->wait_for(kFileAPITimeout)) {
return command_result{error::file_api_timeout};
}
if(!info_response->succeeded()) {
debugMessage(this->getServerId(), "{} Failed to execute file info query: {} / {}", CLIENT_STR_LOG_PREFIX, (int) info_response->error().error_type, info_response->error().error_message);
switch(info_response->error().error_type) {
case ErrorType::UNKNOWN: {
auto error_detail = std::to_string((int) info_response->error().error_type);
if(!info_response->error().error_message.empty())
if(!info_response->error().error_message.empty()) {
error_detail += "/" + info_response->error().error_message;
}
return command_result{error::vs_critical, error_detail};
}
}
@ -547,8 +545,9 @@ command_result ConnectedClient::handleCommandFTGetFileInfo(ts::Command &cmd) {
ts::command_builder notify_file_info{this->notify_response_command("notifyfileinfo")};
size_t notify_index{0};
for(const auto& file : file_status.file_info) {
while(response.response(bulk_index).error_code() != error::ok)
while(response.response(bulk_index).error_code() != error::ok) {
bulk_index++;
}
using Status = file::filesystem::FileInfoResponse::StatusType;
switch (file.status) {
@ -601,16 +600,19 @@ command_result ConnectedClient::handleCommandFTGetFileInfo(ts::Command &cmd) {
notify_index = 0;
}
}
if(notify_index > 0)
if(notify_index > 0) {
this->sendCommand(notify_file_info);
}
if(as_list && this->getExternalType() == ClientType::CLIENT_TEAMSPEAK) {
ts::command_builder notify{this->notify_response_command("notifyfileinfofinished")};
notify.put_unchecked(0, "return_code", cmd["return_code"].string());
this->sendCommand(notify);
}
while(response.length() > bulk_index && response.response(bulk_index).type() == command_result_type::error && response.response(bulk_index).error_code() != error::ok)
while(response.length() > bulk_index && response.response(bulk_index).type() == command_result_type::error && response.response(bulk_index).error_code() != error::ok) {
bulk_index++;
}
assert(bulk_index == cmd.bulkCount());
return command_result{std::move(response)};
@ -978,11 +980,14 @@ command_result ConnectedClient::handleCommandFTList(ts::Command &cmd) {
ACTION_REQUIRES_PERMISSION(permission::b_ft_transfer_list, 1, 0);
auto virtual_file_server = file::server()->find_virtual_server(this->getServerId());
if(!virtual_file_server) return command_result{error::file_virtual_server_not_registered};
if(!virtual_file_server) {
return command_result{error::file_virtual_server_not_registered};
}
auto list_response = file::server()->file_transfer().list_transfer(); //FIXME: Only for the appropriate servers!
if(!list_response->wait_for(kFileAPITimeout))
if(!list_response->wait_for(kFileAPITimeout)) {
return command_result{error::file_api_timeout};
}
if(!list_response->succeeded()) {
using ErrorType = file::transfer::TransferListError;
@ -998,8 +1003,9 @@ command_result ConnectedClient::handleCommandFTList(ts::Command &cmd) {
const auto& transfers = list_response->response();
if(transfers.empty())
if(transfers.empty()) {
return command_result{error::database_empty_result};
}
ts::command_builder notify{this->notify_response_command("notifyftlist")};
size_t bulk_index{0};
@ -1036,11 +1042,14 @@ command_result ConnectedClient::handleCommandFTStop(ts::Command &cmd) {
CMD_CHK_AND_INC_FLOOD_POINTS(25);
auto virtual_file_server = file::server()->find_virtual_server(this->getServerId());
if(!virtual_file_server) return command_result{error::file_virtual_server_not_registered};
if(!virtual_file_server) {
return command_result{error::file_virtual_server_not_registered};
}
auto stop_response = file::server()->file_transfer().stop_transfer(virtual_file_server, cmd["serverftfid"], false);
if(!stop_response->wait_for(kFileAPITimeout))
if(!stop_response->wait_for(kFileAPITimeout)) {
return command_result{error::file_api_timeout};
}
if(!stop_response->succeeded()) {
using ErrorType = file::transfer::TransferActionError::Type;
@ -1051,8 +1060,9 @@ command_result ConnectedClient::handleCommandFTStop(ts::Command &cmd) {
case ErrorType::UNKNOWN: {
auto error_detail = std::to_string((int) stop_response->error().error_type);
if(!stop_response->error().error_message.empty())
if(!stop_response->error().error_message.empty()) {
error_detail += "/" + stop_response->error().error_message;
}
return command_result{error::vs_critical, error_detail};
}
}

8
server/src/client/command_handler/music.cpp
View File

@ -559,7 +559,7 @@ command_result ConnectedClient::handleCommandPlaylistAddPerm(ts::Command &cmd) {
if(auto perr = playlist->client_has_permissions(this->ref(), permission::i_playlist_needed_permission_modify_power, permission::i_playlist_permission_modify_power); perr)
return command_result{perr};
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -589,7 +589,7 @@ command_result ConnectedClient::handleCommandPlaylistDelPerm(ts::Command &cmd) {
if(auto perr = playlist->client_has_permissions(this->ref(), permission::i_playlist_needed_permission_modify_power, permission::i_playlist_permission_modify_power); perr)
return command_result{perr};
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -721,7 +721,7 @@ command_result ConnectedClient::handleCommandPlaylistClientAddPerm(ts::Command &
if(auto perr = playlist->client_has_permissions(this->ref(), permission::i_playlist_needed_permission_modify_power, permission::i_playlist_permission_modify_power); perr)
return command_result{perr};
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if(!pparser.validate(this->ref(), this->getClientDatabaseId()))
return pparser.build_command_result();
@ -755,7 +755,7 @@ command_result ConnectedClient::handleCommandPlaylistClientDelPerm(ts::Command &
return command_result{perr};
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if(!pparser.validate(this->ref(), this->getClientDatabaseId()))
return pparser.build_command_result();

8
server/src/client/command_handler/server.cpp
View File

@ -919,7 +919,7 @@ command_result ConnectedClient::handleCommandServerGroupAddPerm(Command &cmd) {
}
*/
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -985,7 +985,7 @@ command_result ConnectedClient::handleCommandServerGroupDelPerm(Command &cmd) {
}
*/
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -1059,7 +1059,7 @@ command_result ConnectedClient::handleCommandServerGroupAutoAddPerm(ts::Command&
if(groups.empty())
return command_result{error::ok};
command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<true> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();
@ -1138,7 +1138,7 @@ command_result ConnectedClient::handleCommandServerGroupAutoDelPerm(ts::Command&
if(groups.empty()) return command_result{error::ok};
command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
ts::command::bulk_parser::PermissionBulksParser<false> pparser{cmd};
if(!pparser.validate(this->ref(), 0))
return pparser.build_command_result();

2
server/src/client/music/MusicClient.h
View File

@ -95,7 +95,7 @@ namespace ts::server {
void broadcast_text_message(const std::string &message);
void tick(const std::chrono::system_clock::time_point &time) override;
void tick_server(const std::chrono::system_clock::time_point &time) override;
std::chrono::system_clock::time_point next_music_tick;
void execute_music_tick(const std::shared_ptr<ts::music::PlayableSong>&);

4
server/src/client/music/MusicClientPlayer.cpp
View File

@ -138,8 +138,8 @@ void MusicClient::replay_song(const shared_ptr<music::PlayableSong> &entry, cons
this->notifySongChange(entry);
}
void MusicClient::tick(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick(time);
void MusicClient::tick_server(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick_server(time);
if(this->_player_state == ReplayState::SLEEPING)
this->handle_event_song_dry();

741
server/src/client/query/QueryClient.cpp
View File

@ -2,9 +2,6 @@
#include <src/server/QueryServer.h>
#include "QueryClient.h"
#include <netinet/tcp.h>
#include <src/Configuration.h>
#include <log/LogUtils.h>
#include <misc/memtracker.h>
#include "src/InstanceHandler.h"
#include <pipes/errors.h>
#include <misc/std_unique_ptr.h>
@ -13,14 +10,34 @@ using namespace std;
using namespace std::chrono;
using namespace ts;
using namespace ts::server;
using namespace ts::server::query;
#if defined(TCP_CORK) && !defined(TCP_NOPUSH)
#define TCP_NOPUSH TCP_CORK
#endif
//#define DEBUG_TRAFFIC
NetworkBuffer* NetworkBuffer::allocate(size_t length) {
auto result = (NetworkBuffer*) malloc(length + sizeof(NetworkBuffer));
new (result) NetworkBuffer{};
result->length = length;
result->ref_count++;
return result;
}
QueryClient::QueryClient(QueryServer* handle, int sockfd) : ConnectedClient(handle->sql, nullptr), handle(handle), clientFd(sockfd) {
NetworkBuffer* NetworkBuffer::ref() {
this->ref_count++;
return this;
}
void NetworkBuffer::unref() {
if(--this->ref_count == 0) {
this->NetworkBuffer::~NetworkBuffer();
::free(this);
}
}
QueryClient::QueryClient(QueryServer* handle, int sockfd) : ConnectedClient(handle->sql, nullptr), handle(handle), client_file_descriptor(sockfd) {
memtrack::allocated<QueryClient>(this);
int enabled = 1;
int disabled = 0;
@ -28,30 +45,46 @@ QueryClient::QueryClient(QueryServer* handle, int sockfd) : ConnectedClient(hand
if(setsockopt(sockfd, IPPROTO_TCP, TCP_NOPUSH, &disabled, sizeof disabled) < 0) {
logError(this->getServerId(), "[Query] Could not disable nopush for {} ({}/{})", CLIENT_STR_LOG_PREFIX_(this), errno, strerror(errno));
}
if(setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &enabled, sizeof enabled) < 0) {
logError(this->getServerId(), "[Query] Could not disable no delay for {} ({}/{})", CLIENT_STR_LOG_PREFIX_(this), errno, strerror(errno));
}
this->readEvent = event_new(this->handle->eventLoop, this->clientFd, EV_READ | EV_PERSIST, [](int a, short b, void* c){ ((QueryClient*) c)->handleMessageRead(a, b, c); }, this);
this->writeEvent = event_new(this->handle->eventLoop, this->clientFd, EV_WRITE, [](int a, short b, void* c){ ((QueryClient*) c)->handleMessageWrite(a, b, c); }, this);
this->state = ConnectionState::CONNECTED;
connectedTimestamp = system_clock::now();
this->resetEventMask();
}
void QueryClient::applySelfLock(const std::shared_ptr<ts::server::QueryClient> &cl) {
this->_this = cl;
void QueryClient::initialize_self_reference(const std::shared_ptr<QueryClient> &reference) {
this->_this = reference;
this->command_queue = std::make_unique<ServerCommandQueue>(
serverInstance->server_command_executor(),
std::make_unique<QueryClientCommandHandler>(reference)
);
this->event_read = event_new(this->handle->event_io_loop, this->client_file_descriptor, EV_READ | EV_PERSIST, [](int a, short b, void* c){
((QueryClient *) c)->handle_event_read(a, b, c); }, this);
this->event_write = event_new(this->handle->event_io_loop, this->client_file_descriptor, EV_WRITE, [](int a, short b, void* c){
((QueryClient *) c)->handle_event_write(a, b, c); }, this);
}
QueryClient::~QueryClient() {
memtrack::freed<QueryClient>(this);
// if(this->closeLock.tryLock() != 0)
// logCritical("Query manager deleted, but is still in usage! (closeLock)");
// if(this->bufferLock.tryLock() != 0)
// logCritical("Query manager deleted, but is still in usage! (bufferLock)");
if(this->line_buffer) {
free(this->line_buffer);
this->line_buffer = nullptr;
}
this->ssl_handler.finalize();
while(this->write_buffer_head) {
auto buffer = std::exchange(this->write_buffer_head, this->write_buffer_head->next_buffer);
buffer->unref();
}
this->write_buffer_tail = nullptr;
memtrack::freed<QueryClient>(this);
}
void QueryClient::preInitialize() {
@ -62,7 +95,6 @@ void QueryClient::preInitialize() {
DatabaseHelper::assignDatabaseId(this->sql, this->getServerId(), _this.lock());
if(ts::config::query::sslMode == 0) {
this->connectionType = ConnectionType::PLAIN;
this->postInitialize();
@ -70,11 +102,11 @@ void QueryClient::preInitialize() {
}
void QueryClient::postInitialize() {
lock_guard<recursive_mutex> lock(this->lock_packet_handle); /* we dont want to handle anything while we're initializing */
lock_guard command_lock(this->command_lock);
this->connectTimestamp = system_clock::now();
this->properties()[property::CLIENT_LASTCONNECTED] = duration_cast<seconds>(this->connectTimestamp.time_since_epoch()).count();
if(ts::config::query::sslMode == 1 && this->connectionType != ConnectionType::SSL_ENCRIPTED) {
if(ts::config::query::sslMode == 1 && this->connectionType != ConnectionType::SSL_ENCRYPTED) {
command_result error{error::failed_connection_initialisation, "Please use a SSL encryption!"};
this->notifyError(error);
error.release_data();
@ -82,7 +114,7 @@ void QueryClient::postInitialize() {
return;
}
writeMessage(config::query::motd);
send_message(config::query::motd);
assert(this->handle);
if(this->handle->ip_whitelist) {
@ -107,275 +139,283 @@ void QueryClient::postInitialize() {
debugMessage(LOG_QUERY, "Got new query client from {}. Whitelisted: {}", this->getLoggingPeerIp(), this->whitelisted);
if(!this->whitelisted) {
threads::MutexLock lock(this->handle->loginLock);
if(this->handle->queryBann.count(this->getPeerIp()) > 0) {
auto ban = this->handle->queryBann[this->getPeerIp()];
std::lock_guard connect_lock{this->handle->connected_clients_mutex};
auto address = this->getPeerIp();
if(this->handle->client_connect_bans.count(address) > 0) {
auto ban = this->handle->client_connect_bans[address];
Command cmd("error");
auto err = findError("server_connect_banned");
cmd["id"] = err.errorId;
cmd["msg"] = err.message;
cmd["extra_msg"] = "you may retry in " + to_string(duration_cast<seconds>(ban - system_clock::now()).count()) + " seconds";
this->sendCommand(cmd);
this->disconnect("");
this->close_connection(std::chrono::system_clock::now() + std::chrono::seconds{1});
}
}
this->update_cached_permissions();
}
void QueryClient::writeMessage(const std::string& message) {
if(this->state == ConnectionState::DISCONNECTED || !this->handle) return;
void QueryClient::send_message(const std::string_view& message) {
if(this->state == ConnectionState::DISCONNECTED || !this->handle) {
return;
}
if(this->connectionType == ConnectionType::PLAIN) this->writeRawMessage(message);
else if(this->connectionType == ConnectionType::SSL_ENCRIPTED) this->ssl_handler.send(pipes::buffer_view{(void*) message.data(), message.length()});
else logCritical(LOG_GENERAL, "Invalid query connection type to write to!");
if(this->connectionType == ConnectionType::PLAIN) {
this->enqueue_write_buffer(message);
} else if(this->connectionType == ConnectionType::SSL_ENCRYPTED) {
this->ssl_handler.send(pipes::buffer_view{(void*) message.data(), message.length()});
} else {
logCritical(LOG_GENERAL, "Invalid query connection type to write to!");
}
}
bool QueryClient::disconnect(const std::string &reason) {
if(!reason.empty()) {
Command cmd("disconnect");
cmd["reason"] = reason;
this->sendCommand(cmd);
ts::command_builder notify{"disconnect"};
notify.put_unchecked(0, "reason", reason);
this->sendCommand(notify, false);
}
return this->close_connection(system_clock::now() + seconds(1));
}
bool QueryClient::close_connection(const std::chrono::system_clock::time_point& flushTimeout) {
auto ownLock = dynamic_pointer_cast<QueryClient>(_this.lock());
if(!ownLock) return false;
bool QueryClient::close_connection(const std::chrono::system_clock::time_point& flush_timeout_) {
this->flush_timeout = flush_timeout_;
unique_lock<std::recursive_mutex> handleLock(this->lock_packet_handle);
unique_lock<threads::Mutex> lock(this->closeLock);
bool flushing = flushTimeout.time_since_epoch().count() != 0;
if(this->state == ConnectionState::DISCONNECTED || (flushing && this->state == ConnectionState::DISCONNECTING)) return false;
this->state = flushing ? ConnectionState::DISCONNECTING : ConnectionState::DISCONNECTED;
if(this->readEvent) { //Attention dont trigger this within the read thread!
event_del_block(this->readEvent);
event_free(this->readEvent);
this->readEvent = nullptr;
}
if(this->server){
bool should_flush = std::chrono::system_clock::now() < flush_timeout;
{
unique_lock channel_lock(this->server->channel_tree_lock);
this->server->unregisterClient(_this.lock(), "disconnected", channel_lock);
}
this->server->groups->disableCache(this->getClientDatabaseId());
this->server = nullptr;
std::lock_guard network_lock{this->network_mutex};
if(this->event_read) {
event_del_noblock(this->event_read);
}
if(flushing){
this->flushThread = new threads::Thread(THREAD_SAVE_OPERATIONS | THREAD_EXECUTE_LATER, [ownLock, flushTimeout](){
while(ownLock->state == ConnectionState::DISCONNECTING && flushTimeout > system_clock::now()){
{
std::lock_guard buffer_lock(ownLock->buffer_lock);
if(ownLock->readQueue.empty() && ownLock->writeQueue.empty()) break;
if(!should_flush && this->event_write) {
event_del_noblock(this->event_write);
}
usleep(10 * 1000);
}
if(ownLock->state == ConnectionState::DISCONNECTING) ownLock->disconnectFinal();
});
flushThread->name("Flush thread QC").execute();
if(should_flush) {
this->handle->enqueue_query_disconnect(dynamic_pointer_cast<QueryClient>(this->ref()));
} else {
threads::MutexLock l1(this->flushThreadLock);
handleLock.unlock();
lock.unlock();
if(this->flushThread){
threads::NegatedMutexLock l(this->closeLock);
this->flushThread->join();
}
disconnectFinal();
this->handle->enqueue_query_connection_close(dynamic_pointer_cast<QueryClient>(this->ref()));
}
return true;
}
void QueryClient::disconnectFinal() {
lock_guard<recursive_mutex> lock_tick(this->lock_query_tick);
lock_guard<recursive_mutex> lock_handle(this->lock_packet_handle);
threads::MutexLock lock_close(this->closeLock);
std::unique_lock buffer_lock(this->buffer_lock, try_to_lock);
void QueryClient::execute_final_disconnect() {
assert(!this->server);
if(final_disconnected) {
logError(LOG_QUERY, "Tried to disconnect a client twice!");
return;
}
final_disconnected = true;
this->state = ConnectionState::DISCONNECTED;
{
threads::MutexTryLock l(this->flushThreadLock);
if(!!l) {
if(this->flushThread) {
this->flushThread->detach();
delete this->flushThread; //Release the captured this lock
this->flushThread = nullptr;
std::unique_lock network_lock{this->network_mutex};
auto event_read_ = std::exchange(this->event_read, nullptr);
auto event_write_ = std::exchange(this->event_write, nullptr);
network_lock.unlock();
if(event_read_) {
event_del_block(event_read_);
event_free(event_read_);
}
if(event_write_) {
event_del_block(event_write_);
event_free(event_write_);
}
}
if(this->writeEvent) {
event_del_block(this->writeEvent);
event_free(this->writeEvent);
this->writeEvent = nullptr;
}
if(this->readEvent) {
event_del_block(this->readEvent);
event_free(this->readEvent);
this->readEvent = nullptr;
}
if(this->clientFd > 0) {
if(shutdown(this->clientFd, SHUT_RDWR) < 0)
if(this->client_file_descriptor > 0) {
if(shutdown(this->client_file_descriptor, SHUT_RDWR) < 0) {
debugMessage(LOG_QUERY, "Could not shutdown query client socket! {} ({})", errno, strerror(errno));
if(close(this->clientFd) < 0)
}
if(close(this->client_file_descriptor) < 0) {
debugMessage(LOG_QUERY, "Failed to close the query client socket! {} ({})", errno, strerror(errno));
this->clientFd = -1;
}
if(this->server) {
this->client_file_descriptor = -1;
}
}
void QueryClient::enqueue_write_buffer(const std::string_view &message) {
auto buffer = NetworkBuffer::allocate(message.length());
memcpy(buffer->data(), message.data(), message.length());
{
unique_lock channel_lock(this->server->channel_tree_lock);
this->server->unregisterClient(_this.lock(), "disconnected", channel_lock);
}
this->server->groups->disableCache(this->getClientDatabaseId());
this->server = nullptr;
}
std::lock_guard buffer_lock{this->network_mutex};
if(this->event_write) {
*this->write_buffer_tail = buffer;
this->write_buffer_tail = &buffer->next_buffer;
this->readQueue.clear();
this->writeQueue.clear();
if(this->handle)
this->handle->unregisterConnection(dynamic_pointer_cast<QueryClient>(_this.lock()));
}
void QueryClient::writeRawMessage(const std::string &message) {
{
std::lock_guard lock(this->buffer_lock);
this->writeQueue.push_back(message);
}
if(this->writeEvent) event_add(this->writeEvent, nullptr);
}
void QueryClient::handleMessageWrite(int fd, short, void *) {
auto ownLock = _this.lock();
std::unique_lock buffer_lock(this->buffer_lock, try_to_lock);
if(this->state == ConnectionState::DISCONNECTED) return;
if(!buffer_lock.owns_lock()) {
if(this->writeEvent)
event_add(this->writeEvent, nullptr);
event_add(this->event_write, nullptr);
return;
}
int writes = 0;
string buffer;
while(writes < 10 && !this->writeQueue.empty()) {
if(buffer.empty()) {
buffer = std::move(this->writeQueue.front());
this->writeQueue.pop_front();
}
auto length = send(fd, buffer.data(), buffer.length(), MSG_NOSIGNAL);
#ifdef DEBUG_TRAFFIC
debugMessage("Write " + to_string(buffer.length()));
hexDump((void *) buffer.data(), buffer.length());
#endif
/* We don't have a network write event. Drop the buffer. */
buffer->unref();
}
void QueryClient::handle_event_write(int fd, short, void *) {
NetworkBuffer* write_buffer{nullptr};
{
std::lock_guard buffer_lock{this->network_mutex};
if(this->write_buffer_head) {
write_buffer = this->write_buffer_head->ref();
}
}
while(write_buffer) {
auto length = send(fd, (const char*) write_buffer->data() + write_buffer->bytes_written, write_buffer->length - write_buffer->bytes_written, MSG_NOSIGNAL);
if(length == -1) {
write_buffer->unref();
if (errno == EINTR || errno == EAGAIN) {
if(this->writeEvent)
event_add(this->writeEvent, nullptr);
return;
}
else {
logError(LOG_QUERY, "{} Failed to write message: {} ({} => {})", CLIENT_STR_LOG_PREFIX, length, errno, strerror(errno));
threads::Thread([=](){ ownLock->close_connection(chrono::system_clock::now() + chrono::seconds{5}); }).detach();
return;
std::lock_guard event_lock{this->network_mutex};
if(this->event_write) {
event_add(this->event_write, nullptr);
}
} else {
if(buffer.length() == length)
buffer = "";
else
buffer = buffer.substr(length);
}
writes++;
}
if(!buffer.empty())
this->writeQueue.push_front(buffer);
logError(LOG_QUERY, "{} Failed to send message ({}/{}). Closing connection.", CLIENT_STR_LOG_PREFIX, errno, strerror(errno));
this->close_connection(std::chrono::system_clock::time_point{});
if(!this->writeQueue.empty() && this->writeEvent)
event_add(this->writeEvent, nullptr);
{
std::unique_lock event_lock{this->network_mutex};
auto event_write_ = std::exchange(this->event_write, nullptr);
event_lock.unlock();
if(event_write_) {
event_del_noblock(event_write_);
event_free(event_write_);
}
}
/* the "this" ptr might be dangling now since we can't join the write event any more! */
}
void QueryClient::handleMessageRead(int fd, short, void *) {
auto ownLock = dynamic_pointer_cast<QueryClient>(_this.lock());
if(!ownLock) {
logCritical(LOG_QUERY, "Could not get own lock!");
return;
}
string buffer(1024, 0);
write_buffer->bytes_written += length;
assert(write_buffer->bytes_written <= write_buffer->length);
auto length = read(fd, (void*) buffer.data(), buffer.length());
if(write_buffer->bytes_written == write_buffer->length) {
/*
* Even though we might free the buffer here we could still use the pointer for comparison.
* If the buffer is still the head buffer it should not have been deallocated since
* the queue itself holds a reference.
*/
write_buffer->unref();
/* Buffer must be freed, but we don't want do that while holding the lock */
NetworkBuffer* cleanup_buffer{nullptr};
/* Buffer finished, sending next one */
{
std::lock_guard buffer_lock{this->network_mutex};
if(this->write_buffer_head == write_buffer) {
/* Buffer successfully send. Sending the next one. */
cleanup_buffer = this->write_buffer_head;
this->write_buffer_head = this->write_buffer_head->next_buffer;
if(this->write_buffer_head) {
/* we've a next buffer */
write_buffer = this->write_buffer_head->ref();
} else {
assert(this->write_buffer_tail == &write_buffer->next_buffer);
write_buffer = nullptr;
this->write_buffer_tail = &this->write_buffer_head;
}
} else if(this->write_buffer_head) {
/* Our buffer got dropped (who knows why). Just send the next one. */
write_buffer = this->write_buffer_head->ref();
} else {
/*
* Nothing more to write.
*/
write_buffer = nullptr;
}
}
if(cleanup_buffer) {
cleanup_buffer->unref();
}
}
}
/* This state should only be reached when no more messages are pending to write */
assert(!write_buffer);
if(this->state == ConnectionState::DISCONNECTING) {
this->handle->enqueue_query_connection_close(dynamic_pointer_cast<QueryClient>(this->ref()));
}
}
void QueryClient::handle_event_read(int fd, short, void *) {
static const size_t kReadBufferLength = 1024 * 8;
uint8_t buffer[kReadBufferLength];
auto length = read(fd, buffer, kReadBufferLength);
if(length <= 0){
if(errno == EINTR || errno == EAGAIN)
;//event_add(this->readEvent, nullptr);
else if(length == 0 && errno == 0) {
if(errno == EINTR || errno == EAGAIN) {
/* Nothing to read */
return;
}
if(length == 0 && errno == 0) {
logMessage(LOG_QUERY, "{} Connection closed. Client disconnected.", CLIENT_STR_LOG_PREFIX);
event_del_noblock(this->readEvent);
std::thread([ownLock]{
ownLock->close_connection();
}).detach();
} else {
logError(LOG_QUERY, "{} Failed to read! Code: {} errno: {} message: {}", CLIENT_STR_LOG_PREFIX, length, errno, strerror(errno));
event_del_noblock(this->readEvent);
threads::Thread(THREAD_SAVE_OPERATIONS, [ownLock](){ ownLock->close_connection(); }).detach();
logMessage(LOG_QUERY, "{} Failed to received message ({}/{}). Closing connection.", CLIENT_STR_LOG_PREFIX, errno, strerror(errno));
}
this->close_connection(std::chrono::system_clock::time_point{});
{
std::unique_lock network_lock{this->network_mutex};
auto event_read_ = std::exchange(this->event_read, nullptr);
network_lock.unlock();
if(event_read_) {
event_del_noblock(event_read_);
event_free(event_read_);
}
}
/* the "this" ptr might be dangling now since we can't join the read event any more! */
return;
}
buffer.resize(length);
{
std::lock_guard buffer_lock(this->buffer_lock);
if(this->state == ConnectionState::DISCONNECTED)
this->handle_message_read(std::string_view{(const char *) buffer, (size_t) length});
}
inline bool is_ssl_handshake_header(const std::string_view& buffer) {
if(buffer.length() < 0x05) return false; //Header too small!
if(buffer[0] != 0x16) return false; //recordType=handshake
if(buffer[1] < 1 || buffer[1] > 3) return false; //SSL version
if(buffer[2] < 1 || buffer[2] > 3) return false; //TLS version
return true;
}
void QueryClient::handle_message_read(const std::string_view &message) {
if(this->state >= ConnectionState::DISCONNECTING) {
/* We don't need to handle any messages. */
return;
this->readQueue.push_back(std::move(buffer));
#ifdef DEBUG_TRAFFIC
debugMessage("Read " + to_string(buffer.length()));
hexDump((void *) buffer.data(), buffer.length());
#endif
}
if(this->handle)
this->handle->executePool()->execute([ownLock]() {
int counter = 0;
while(ownLock->tickIOMessageProgress() && counter++ < 15);
});
}
switch (this->connectionType) {
case ConnectionType::PLAIN:
this->handle_decoded_message(message);
break;
bool QueryClient::tickIOMessageProgress() {
lock_guard<recursive_mutex> lock(this->lock_packet_handle);
if(!this->handle || this->state == ConnectionState::DISCONNECTED || this->state == ConnectionState::DISCONNECTING) return false;
case ConnectionType::SSL_ENCRYPTED:
this->ssl_handler.process_incoming_data(pipes::buffer_view{message.data(), message.length()});
break;
string message;
bool next = false;
{
std::lock_guard buffer_lock(this->buffer_lock);
if(this->readQueue.empty()) return false;
message = std::move(this->readQueue.front());
this->readQueue.pop_front();
next |= this->readQueue.empty();
}
if(this->connectionType == ConnectionType::PLAIN) {
int count = 0;
while(this->handleMessage(pipes::buffer_view{(void*) message.data(), message.length()}) && count++ < 15) message = "";
next |= count == 15;
} else if(this->connectionType == ConnectionType::SSL_ENCRIPTED) {
this->ssl_handler.process_incoming_data(pipes::buffer_view{(void*) message.data(), message.length()});
} else if(this->connectionType == ConnectionType::UNKNOWN) {
if(config::query::sslMode != 0 && pipes::SSL::isSSLHeader(message)) {
case ConnectionType::UNKNOWN: {
if(config::query::sslMode != 0 && is_ssl_handshake_header(message)) {
this->initializeSSL();
/*
@ -391,30 +431,126 @@ bool QueryClient::tickIOMessageProgress() {
* \x05 -> length (4)
*/
//this->writeRawMessage(string("\x16\x03\x01\x00\x05\x00\x00\x00\x00\x00", 10));
this->ssl_handler.process_incoming_data(pipes::buffer_view{message.data(), message.length()});
} else {
this->connectionType = ConnectionType::PLAIN;
this->postInitialize();
}
next = true;
{
std::lock_guard buffer_lock(this->buffer_lock);
this->readQueue.push_front(std::move(message));
this->handle_decoded_message(message);
}
}
}
return next;
}
extern InstanceHandler* serverInstance;
inline size_t line_buffer_size(size_t target_size) {
return target_size;
}
void QueryClient::handle_decoded_message(const std::string_view &message) {
if(this->line_buffer_length + message.length() > this->line_buffer_capacity) {
this->line_buffer_capacity = line_buffer_size(this->line_buffer_length + message.length());
auto new_buffer = (char*) malloc(this->line_buffer_capacity);
memcpy(new_buffer, this->line_buffer, this->line_buffer_length);
free(this->line_buffer);
this->line_buffer = new_buffer;
}
memcpy(this->line_buffer + this->line_buffer_length, message.data(), message.length());
this->line_buffer_length += message.length();
/*
* Now we're analyzing the line buffer.
* Note: Telnet commands will be executed as empty (idle commands)
*/
size_t command_start_index{0}, command_end_index, command_start_next;
for(; this->line_buffer_scan_offset < this->line_buffer_length; this->line_buffer_scan_offset++) {
if(this->line_buffer[this->line_buffer_scan_offset] == '\n') {
command_end_index = this->line_buffer_scan_offset;
command_start_next = this->line_buffer_scan_offset + 1;
} else if((uint8_t) this->line_buffer[this->line_buffer_scan_offset] == 255) {
if(this->line_buffer_scan_offset + 3 > this->line_buffer_length) {
/* We don't have enough space to fill the telnet command so we use that as the new scan offset */
command_end_index = this->line_buffer_scan_offset;
command_start_next = this->line_buffer_scan_offset;
if(command_start_next == command_end_index) {
/* We've no prepended data so we're waiting for the tcp command. Loop finished. */
break;
}
} else {
command_end_index = this->line_buffer_scan_offset;
command_start_next = this->line_buffer_scan_offset + 3;
logTrace(LOG_QUERY, "[{}:{}] Received telnet command, code = {}, option = {}",
this->getLoggingPeerIp(), this->getPeerPort(),
(uint8_t) this->line_buffer[this->line_buffer_scan_offset + 1],
(uint8_t) this->line_buffer[this->line_buffer_scan_offset + 2]
);
}
} else {
continue;
}
/* No need to check for the upper bounds since there will be \n or 255 before the end of the line */
while(this->line_buffer[command_start_index] == '\r') {
command_start_index++;
}
while(command_end_index > command_start_index + 1 && this->line_buffer[command_end_index - 1] == '\r') {
command_end_index--;
}
std::string_view command_view{this->line_buffer + command_start_index, command_end_index - command_start_index};
logTrace(0, "Found command: '{}'", command_view);
this->command_queue->enqueue_command_string(command_view);
command_start_index = command_start_next;
if(this->line_buffer_scan_offset + 1 < command_start_next) {
this->line_buffer_scan_offset = command_start_next - 1;
}
}
if(command_start_index > 0) {
if(command_start_index == this->line_buffer_length) {
this->line_buffer_length = 0;
this->line_buffer_scan_offset = 0;
} else {
assert(this->line_buffer_length > command_start_index);
assert(this->line_buffer_scan_offset > command_start_index);
memcpy(this->line_buffer, this->line_buffer + command_start_index, this->line_buffer_length - command_start_index);
this->line_buffer_length -= command_start_index;
this->line_buffer_scan_offset -= command_start_index;
}
}
if(this->line_buffer_length > ts::config::query::max_line_buffer) {
this->line_buffer_length = 0; /* Buffer will be truncated later */
logWarning(LOG_QUERY, "[{}] Client exceeded max query line buffer size. Disconnecting client.");
this->disconnect("line buffer length exceeded");
}
/* Shrink if possible */
if(this->line_buffer_capacity > 8 * 1024 && this->line_buffer_length < 8 * 1024) {
this->line_buffer_capacity = 8 * 1024;
auto new_buffer = (char*) malloc(this->line_buffer_capacity);
memcpy(new_buffer, this->line_buffer, this->line_buffer_length);
free(this->line_buffer);
this->line_buffer = new_buffer;
}
}
void QueryClient::initializeSSL() {
this->connectionType = ConnectionType::SSL_ENCRIPTED;
this->connectionType = ConnectionType::SSL_ENCRYPTED;
this->ssl_handler.direct_process(pipes::PROCESS_DIRECTION_OUT, true);
this->ssl_handler.direct_process(pipes::PROCESS_DIRECTION_IN, true);
this->ssl_handler.callback_data(std::bind(&QueryClient::handleMessage, this, placeholders::_1));
this->ssl_handler.callback_write(std::bind(&QueryClient::writeRawMessage, this, placeholders::_1));
this->ssl_handler.callback_data([&](const pipes::buffer_view& buffer) {
this->handle_decoded_message(std::string_view{buffer.data_ptr<char>(), buffer.length()});
});
this->ssl_handler.callback_write([&](const pipes::buffer_view& buffer) {
this->enqueue_write_buffer(std::string_view{buffer.data_ptr<char>(), buffer.length()});
});
this->ssl_handler.callback_initialized = std::bind(&QueryClient::postInitialize, this);
this->ssl_handler.callback_error([&](int code, const std::string& message) {
@ -439,126 +575,23 @@ void QueryClient::initializeSSL() {
}
}
bool QueryClient::handleMessage(const pipes::buffer_view& message) {
{
threads::MutexLock l(this->closeLock);
if(this->state == ConnectionState::DISCONNECTED)
return false;
}
#ifdef DEBUG_TRAFFIC
debugMessage("Handling message " + to_string(message.length()));
hexDump((void *) message.data(), message.length());
#endif
string command;
{
this->lineBuffer += message.string();
int length = 2;
auto pos = this->lineBuffer.find("\r\n");
if(pos == string::npos) pos = this->lineBuffer.find("\n\r");
if(pos == string::npos) {
length = 1;
pos = this->lineBuffer.find('\n');
}
if(pos != string::npos){
command = this->lineBuffer.substr(0, pos);
if(this->lineBuffer.size() > pos + length)
this->lineBuffer = this->lineBuffer.substr(pos + length);
else
this->lineBuffer.clear();
}
if(pos == string::npos) return false;
}
if(command.empty() || command.find_first_not_of(' ') == string::npos) { //Empty command
logTrace(LOG_QUERY, "[{}:{}] Got query idle command (Empty command or spaces)", this->getLoggingPeerIp(), this->getPeerPort());
CMD_RESET_IDLE; //if idle time over 5 min than connection drop
return true;
}
if(auto non_escape{command.find_first_not_of('\r')}; non_escape == std::string::npos) {
logTrace(LOG_QUERY, "[{}:{}] Got query idle command (\\r)", this->getLoggingPeerIp(), this->getPeerPort());
CMD_RESET_IDLE; //if idle time over 5 min than connection drop
return true;
} else {
command = command.substr(non_escape);
}
if(auto non_escape{command.find_first_not_of('\n')}; non_escape == std::string::npos) {
logTrace(LOG_QUERY, "[{}:{}] Got query idle command (\\n)", this->getLoggingPeerIp(), this->getPeerPort());
CMD_RESET_IDLE; //if idle time over 5 min than connection drop
return true;
} else {
command = command.substr(non_escape);
}
if((uint8_t) command[0] == 255) {
string commands{};
/* we got a telnet command here */
while(command.size() >= 2 && (uint8_t) command[0] == 255) {
uint8_t code = command[1];
uint8_t option = command[2];
if(!commands.empty())
commands += ", ";
commands += to_string(code) + ":" + to_string(option);
command = command.substr(3);
}
logTrace(LOG_QUERY, "[{}:{}] Received telnet command(s): {}. Ignoring it.",this->getLoggingPeerIp(), this->getPeerPort(), commands);
CMD_RESET_IDLE;
if(command.empty())
return true;
}
unique_ptr<Command> cmd;
command_result error{};
try {
cmd = make_unique<Command>(Command::parse(pipes::buffer_view{(void*) command.data(), command.length()}, true, !ts::config::server::strict_ut8_mode));
} catch(std::invalid_argument& ex) {
logTrace(LOG_QUERY, "[{}:{}] Failed to parse command (invalid argument): {}", this->getLoggingPeerIp(), this->getPeerPort(), command);
error.reset(command_result{error::parameter_convert});
goto handle_error;
} catch(std::exception& ex) {
logTrace(LOG_QUERY, "[{}:{}] Failed to parse command (exception: {}): {}", this->getLoggingPeerIp(), this->getPeerPort(), ex.what(), command);
error.reset(command_result{error::vs_critical, std::string{ex.what()}});
goto handle_error;
}
try {
this->handleCommandFull(*cmd);
} catch(std::exception& ex) {
error.reset(command_result{error::vs_critical, std::string{ex.what()}});
goto handle_error;
}
return true;
handle_error:
this->notifyError(error);
error.release_data();
return false;
}
void QueryClient::sendCommand(const ts::Command &command, bool) {
auto cmd = command.build();
writeMessage(cmd + config::query::newlineCharacter);
send_message(cmd + config::query::newlineCharacter);
logTrace(LOG_QUERY, "Send command {}", cmd);
}
void QueryClient::sendCommand(const ts::command_builder &command, bool) {
writeMessage(command.build() + config::query::newlineCharacter);
send_message(command.build() + config::query::newlineCharacter);
logTrace(LOG_QUERY, "Send command {}", command.build());
}
void QueryClient::tick(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick(time);
void QueryClient::tick_server(const std::chrono::system_clock::time_point &time) {
ConnectedClient::tick_server(time);
}
void QueryClient::queryTick() {
lock_guard<recursive_mutex> lock_tick(this->lock_query_tick);
/* FIXME: TODO: Forbit this while beeing in finalDisconnect! */
void QueryClient::tick_query() {
if(this->idleTimestamp.time_since_epoch().count() > 0 && system_clock::now() - this->idleTimestamp > minutes(5)){
debugMessage(LOG_QUERY, "Dropping client " + this->getLoggingPeerIp() + "|" + this->getDisplayName() + ". (Timeout)");
this->close_connection(system_clock::now() + seconds(1));
@ -568,37 +601,39 @@ void QueryClient::queryTick() {
this->connectionType = ConnectionType::PLAIN;
this->postInitialize();
}
}
bool QueryClient::notifyChannelSubscribed(const deque<shared_ptr<BasicChannel>> &) {
return false;
if(this->flush_timeout.time_since_epoch().count() > 0 && std::chrono::system_clock::now() > this->flush_timeout) {
this->handle->enqueue_query_connection_close(dynamic_pointer_cast<QueryClient>(this->ref()));
}
bool QueryClient::notifyChannelUnsubscribed(const deque<shared_ptr<BasicChannel>> &){
return false;
}
bool QueryClient::ignoresFlood() {
return this->whitelisted || ConnectedClient::ignoresFlood();
}
void QueryClient::disconnect_from_virtual_server() {
threads::MutexLock lock(this->command_lock);
void QueryClient::disconnect_from_virtual_server(const std::string& reason) {
std::lock_guard command_lock{this->command_lock};
auto server_locked = this->server;
if(server_locked) {
//unregister manager from old server
auto old_server = std::exchange(this->server, nullptr);
if(old_server) {
{
unique_lock tree_lock(this->server->channel_tree_lock);
if(this->currentChannel)
this->server->client_move(this->ref(), nullptr, nullptr, "", ViewReasonId::VREASON_USER_ACTION, false, tree_lock);
this->server->unregisterClient(_this.lock(), "server switch", tree_lock);
std::unique_lock tree_lock(old_server->channel_tree_lock);
if(this->currentChannel) {
old_server->client_move(this->ref(), nullptr, nullptr, "", ViewReasonId::VREASON_USER_ACTION, false, tree_lock);
}
server_locked->groups->disableCache(this->getClientDatabaseId());
old_server->unregisterClient(_this.lock(), reason, tree_lock);
}
{
std::lock_guard channel_lock{this->channel_lock};
this->channels->reset();
this->currentChannel = nullptr;
this->server = nullptr;
}
old_server->groups->disableCache(this->getClientDatabaseId());
this->loadDataForCurrentServer();
}
serverInstance->getGroupManager()->enableCache(this->getClientDatabaseId());
}

110
server/src/client/query/QueryClient.h
View File

@ -5,87 +5,116 @@
#include <protocol/buffers.h>
#include <pipes/ssl.h>
#include "misc/queue.h"
#include "../shared/ServerCommandExecutor.h"
namespace ts::server {
class QueryServer;
class QueryAccount;
class QueryClientCommandHandler;
namespace query {
struct NetworkBuffer {
static NetworkBuffer* allocate(size_t /* length */);
size_t length;
size_t bytes_written{0};
NetworkBuffer* next_buffer{nullptr};
std::atomic_int16_t ref_count{};
[[nodiscard]] inline const void* data() const { return (const char*) this + sizeof(NetworkBuffer); }
[[nodiscard]] inline void* data() { return (char*) this + sizeof(NetworkBuffer); }
[[nodiscard]] NetworkBuffer* ref();
void unref();
};
}
class QueryClient : public ConnectedClient {
friend class QueryServer;
friend class QueryClientCommandHandler;
using NetworkBuffer = query::NetworkBuffer;
enum ConnectionType {
PLAIN,
SSL_ENCRIPTED,
SSL_ENCRYPTED,
UNKNOWN
};
public:
QueryClient(QueryServer*, int sockfd);
QueryClient(QueryServer* /* server handle */, int /* file descriptor */);
~QueryClient() override;
void writeMessage(const std::string&);
void sendCommand(const ts::Command &command, bool low = false) override;
void sendCommand(const ts::command_builder &command, bool low) override;
bool disconnect(const std::string &reason) override;
bool close_connection(const std::chrono::system_clock::time_point& timeout = std::chrono::system_clock::time_point()) override;
void disconnectFinal();
bool close_connection(const std::chrono::system_clock::time_point& flush_timeout) override;
bool eventActive(QueryEventGroup, QueryEventSpecifier);
void toggleEvent(QueryEventGroup, QueryEventSpecifier, bool);
void resetEventMask();
bool ignoresFlood() override;
void disconnect_from_virtual_server();
void disconnect_from_virtual_server(const std::string& /* reason */);
inline std::shared_ptr<QueryAccount> getQueryAccount() { return this->query_account; }
protected:
void initialize_self_reference(const std::shared_ptr<QueryClient> &reference);
void preInitialize();
void postInitialize();
void tick(const std::chrono::system_clock::time_point &time) override;
void queryTick();
protected:
void initializeSSL();
void postInitialize();
bool handleMessage(const pipes::buffer_view&);
bool tickIOMessageProgress();
/* Will be called by the query server */
void execute_final_disconnect();
void handleMessageRead(int, short, void*);
void handleMessageWrite(int, short, void*);
void writeRawMessage(const std::string&);
/* the ticking method will only be called when connected to a server */
void tick_server(const std::chrono::system_clock::time_point &time) override;
void tick_query();
void applySelfLock(const std::shared_ptr<QueryClient> &cl);
/* Methods will be called within the io loop (single thread) */
void handle_event_read(int, short, void*);
void handle_message_read(const std::string_view& /* message */);
void handle_decoded_message(const std::string_view& /* message */);
/* Methods will be called within the io loop (single thread) */
void handle_event_write(int, short, void*);
void send_message(const std::string_view&);
void enqueue_write_buffer(const std::string_view& /* message */);
private:
QueryServer* handle;
ConnectionType connectionType = ConnectionType::UNKNOWN;
ConnectionType connectionType{ConnectionType::UNKNOWN};
bool whitelisted = false;
int clientFd = -1;
bool whitelisted{false};
int client_file_descriptor{-1};
::event* readEvent = nullptr;
::event* writeEvent = nullptr;
threads::Mutex closeLock;
spin_mutex network_mutex{};
::event* event_read{nullptr};
::event* event_write{nullptr};
/* locked by network_mutex */
NetworkBuffer* write_buffer_head{nullptr};
NetworkBuffer** write_buffer_tail{&this->write_buffer_head};
/* pipes::SSL internally thread save */
pipes::SSL ssl_handler;
std::mutex buffer_lock;
std::deque<std::string> writeQueue;
std::deque<std::string> readQueue;
std::chrono::system_clock::time_point flush_timeout{};
threads::Mutex flushThreadLock;
threads::Thread* flushThread = nullptr;
bool final_disconnected = false;
/* The line buffer must only be accessed within the io event loop! */
char* line_buffer{nullptr};
size_t line_buffer_length{0};
size_t line_buffer_capacity{0};
size_t line_buffer_scan_offset{0};
/* thread save to access */
std::unique_ptr<ServerCommandQueue> command_queue{};
std::string lineBuffer;
std::chrono::time_point<std::chrono::system_clock> connectedTimestamp;
uint16_t eventMask[QueryEventGroup::QEVENTGROUP_MAX];
std::recursive_mutex lock_packet_handle;
std::recursive_mutex lock_query_tick;
std::shared_ptr<QueryAccount> query_account;
protected:
command_result handleCommand(Command &command) override;
@ -181,4 +210,15 @@ namespace ts::server {
command_result handleCommandSetCompressionMode(Command&);
};
class QueryClientCommandHandler : public ts::server::ServerCommandHandler {
public:
explicit QueryClientCommandHandler(const std::shared_ptr<QueryClient>& /* client */);
protected:
bool handle_command(const std::string_view &) override;
private:
std::weak_ptr<QueryClient> client_ref;
};
}

66
server/src/client/query/QueryClientCommands.cpp
View File

@ -21,8 +21,56 @@ using namespace std::chrono;
using namespace ts;
using namespace ts::server;
constexpr unsigned int string_hash(const char* str, int h = 0) {
return !str[h] ? 5381 : (string_hash(str, h + 1U) * 33U) ^ str[h];
QueryClientCommandHandler::QueryClientCommandHandler(const std::shared_ptr<QueryClient> &client) : client_ref{client} {}
bool QueryClientCommandHandler::handle_command(const std::string_view &command) {
auto client = this->client_ref.lock();
if(!client) {
return false;
}
if(command.empty()) {
logTrace(LOG_QUERY, "[{}:{}] Got query idle command.", client->getLoggingPeerIp(), client->getPeerPort());
client->resetIdleTime();
return true;
}
unique_ptr<Command> cmd;
command_result error{};
try {
cmd = make_unique<Command>(Command::parse(command, true, !ts::config::server::strict_ut8_mode));
} catch(std::invalid_argument& ex) {
logTrace(LOG_QUERY, "[{}:{}] Failed to parse command (invalid argument): {}", client->getLoggingPeerIp(), client->getPeerPort(), command);
error.reset(command_result{error::parameter_convert});
goto handle_error;
} catch(std::exception& ex) {
logTrace(LOG_QUERY, "[{}:{}] Failed to parse command (exception: {}): {}", client->getLoggingPeerIp(), client->getPeerPort(), ex.what(), command);
error.reset(command_result{error::vs_critical, std::string{ex.what()}});
goto handle_error;
}
try {
std::lock_guard execute_lock{client->command_lock};
if(client->state >= ConnectionState::DISCONNECTING) {
return false;
}
client->handleCommandFull(*cmd);
} catch(std::exception& ex) {
error.reset(command_result{error::vs_critical, std::string{ex.what()}});
goto handle_error;
}
return true;
handle_error:
client->notifyError(error);
error.release_data();
return true;
}
constexpr unsigned int string_hash(const char* str, unsigned int h = 0) {
return !str[h] ? 5381 : (string_hash(str, h + 1U) * 33U) ^ (unsigned int) str[h];
}
command_result QueryClient::handleCommand(Command& cmd) {
@ -108,7 +156,9 @@ command_result QueryClient::handleCommand(Command& cmd) {
#else
auto cmd_str = cmd.build();
ts::command_parser parser{cmd_str};
if(!parser.parse(true)) return command_result{error::vs_critical};
if(!parser.parse(true)) {
return command_result{error::vs_critical};
}
return this->handleCommandServerSnapshotDeployNew(parser);
#endif
@ -154,7 +204,7 @@ command_result QueryClient::handleCommandLogin(Command& cmd) {
auto account = _account ? serverInstance->getQueryServer()->load_password(_account) : nullptr;
{
threads::MutexLock lock(this->handle->loginLock);
std::lock_guard connect_lock{this->handle->client_connect_mutex};
if(!account) {
serverInstance->action_logger()->query_authenticate_logger.log_query_authenticate(this->getServerId(), std::dynamic_pointer_cast<QueryClient>(this->ref()), username, log::QueryAuthenticateResult::UNKNOWN_USER);
@ -163,9 +213,9 @@ command_result QueryClient::handleCommandLogin(Command& cmd) {
if (account->password != password) {
if(!this->whitelisted) {
this->handle->loginAttempts[this->getPeerIp()]++;
if(this->handle->loginAttempts[this->getPeerIp()] > 3) {
this->handle->queryBann[this->getPeerIp()] = system_clock::now() + seconds(serverInstance->properties()[property::SERVERINSTANCE_SERVERQUERY_BAN_TIME].as<uint64_t>()); //TODO configurable | Disconnect all others?
this->handle->client_connect_count[this->getPeerIp()]++;
if(this->handle->client_connect_count[this->getPeerIp()] > 3) {
this->handle->client_connect_bans[this->getPeerIp()] = system_clock::now() + seconds(serverInstance->properties()[property::SERVERINSTANCE_SERVERQUERY_BAN_TIME].as<uint64_t>()); //TODO configurable | Disconnect all others?
this->postCommandHandler.emplace_back([&](){
this->close_connection(system_clock::now() + seconds(1));
});
@ -334,7 +384,7 @@ command_result QueryClient::handleCommandServerSelect(Command &cmd) {
}
}
this->disconnect_from_virtual_server();
this->disconnect_from_virtual_server("server switch");
this->resetEventMask();
//register at current server

8
server/src/client/query/QueryClientNotify.cpp
View File

@ -225,3 +225,11 @@ bool QueryClient::notifyMusicPlayerSongChange(const std::shared_ptr<MusicClient>
CHK_EVENT(QEVENTGROUP_MUSIC, QEVENTSPECIFIER_MUSIC_PLAYER);
return ConnectedClient::notifyMusicPlayerSongChange(bot, newEntry);
}
bool QueryClient::notifyChannelSubscribed(const deque<shared_ptr<BasicChannel>> &) {
return false;
}
bool QueryClient::notifyChannelUnsubscribed(const deque<shared_ptr<BasicChannel>> &){
return false;
}

19
server/src/client/shared/RawCommand.cpp Normal file
View File

@ -0,0 +1,19 @@
//
// Created by WolverinDEV on 28/01/2021.
//
#include "RawCommand.h"
using namespace ts::server::command;
ReassembledCommand *ReassembledCommand::allocate(size_t size) {
auto instance = (ReassembledCommand*) malloc(sizeof(ReassembledCommand) + size);
instance->length_ = size;
instance->capacity_ = size;
instance->next_command = nullptr;
return instance;
}
void ReassembledCommand::free(ReassembledCommand *command) {
::free(command);
}

48
server/src/client/shared/RawCommand.h Normal file
View File

@ -0,0 +1,48 @@
#pragma once
#include <cstdint>
#include <string_view>
#include <pipes/buffer.h>
namespace ts::server::command {
struct CommandFragment {
uint16_t packet_id{0};
uint16_t packet_generation{0};
uint8_t packet_flags{0};
uint32_t payload_length : 24;
pipes::buffer payload{};
CommandFragment() : payload_length{0} { }
CommandFragment(uint16_t packetId, uint16_t packetGeneration, uint8_t packetFlags, uint32_t payloadLength, pipes::buffer payload)
: packet_id{packetId}, packet_generation{packetGeneration}, packet_flags{packetFlags}, payload_length{payloadLength}, payload{std::move(payload)} {}
CommandFragment& operator=(const CommandFragment&) = default;
CommandFragment(const CommandFragment& other) = default;
CommandFragment(CommandFragment&&) = default;
};
static_assert(sizeof(CommandFragment) == 8 + sizeof(pipes::buffer));
struct ReassembledCommand {
public:
static ReassembledCommand* allocate(size_t /* command length */);
static void free(ReassembledCommand* /* command */);
[[nodiscard]] inline size_t length() const { return this->length_; }
inline void set_length(size_t length) { assert(this->capacity_ >= length); this->length_ = length; }
[[nodiscard]] inline size_t capacity() const { return this->capacity_; }
[[nodiscard]] inline const char* command() const { return (const char*) this + sizeof(ReassembledCommand); }
[[nodiscard]] inline char* command() { return (char*) this + sizeof(ReassembledCommand); }
[[nodiscard]] inline std::string_view command_view() const { return std::string_view{this->command(), this->length()}; }
mutable ReassembledCommand* next_command; /* nullptr by default */
private:
explicit ReassembledCommand() = default;
size_t capacity_;
size_t length_;
};
}

272
server/src/client/shared/ServerCommandExecutor.cpp Normal file
View File

@ -0,0 +1,272 @@
//
// Created by WolverinDEV on 29/07/2020.
//
#include <utility>
#include <ThreadPool/ThreadHelper.h>
#include "./ServerCommandExecutor.h"
#include "src/client/voice/PacketDecoder.h"
#include "src/client/voice/VoiceClientConnection.h"
using namespace ts;
using namespace ts::server;
using namespace ts::server::command;
namespace ts::server {
struct ServerCommandQueueInner {
spin_mutex pending_commands_lock{};
command::ReassembledCommand* pending_commands_head{nullptr};
command::ReassembledCommand** pending_commands_tail{&pending_commands_head};
bool has_command_handling_scheduled{false};
~ServerCommandQueueInner() {
auto head = this->pending_commands_head;
while(head) {
auto cmd = head->next_command;
ReassembledCommand::free(head);
head = cmd;
}
}
void reset() {
std::unique_lock pc_lock{this->pending_commands_lock};
auto head = std::exchange(this->pending_commands_head, nullptr);
this->pending_commands_tail = &this->pending_commands_head;
this->has_command_handling_scheduled = false;
pc_lock.unlock();
while(head) {
auto cmd = head->next_command;
ReassembledCommand::free(head);
head = cmd;
}
}
/**
* @param command The target command to enqueue.
* Ownership will be taken.
* @returns `true` if command handling has already been schedules and `false` if not
*/
bool enqueue(ReassembledCommand *command){
std::lock_guard pc_lock{this->pending_commands_lock};
*this->pending_commands_tail = command;
this->pending_commands_tail = &command->next_command;
return std::exchange(this->has_command_handling_scheduled, true);
}
ReassembledCommand* pop_command(bool& more_pending) {
std::lock_guard pc_lock{this->pending_commands_lock};
auto result = this->pending_commands_head;
if(!result) {
more_pending = false;
this->has_command_handling_scheduled = false;
} else if(result->next_command) {
more_pending = true;
this->pending_commands_head = result->next_command;
} else {
/* We assume true here since we might get new commands while the handler itself is still handling our current result. */
more_pending = true;
this->pending_commands_head = nullptr;
this->pending_commands_tail = &this->pending_commands_head;
}
return result;
}
};
}
bool ServerCommandHandler::execute_handling() {
bool more_pending;
std::unique_ptr<ReassembledCommand, void(*)(ReassembledCommand*)> pending_command{nullptr, ReassembledCommand::free};
while(true) {
pending_command.reset(this->inner->pop_command(more_pending));
if(!pending_command) {
break;
}
try {
auto result = this->handle_command(std::string_view{pending_command->command(), pending_command->length()});
if(!result) {
/* flush all commands */
this->inner->reset();
break;
}
} catch (std::exception& ex) {
logCritical(LOG_GENERAL, "Exception reached command execution root! {}",ex.what());
}
break; /* Maybe handle more than one command? Maybe some kind of time limit? */
}
return more_pending;
}
ServerCommandQueue::ServerCommandQueue(std::shared_ptr<ServerCommandExecutor> executor, std::unique_ptr<ServerCommandHandler> command_handler) :
executor{std::move(executor)},
command_handler{command_handler.release()} {
assert(this->command_handler);
this->inner = std::make_shared<ServerCommandQueueInner>();
this->command_handler->inner = this->inner;
}
ServerCommandQueue::~ServerCommandQueue() = default;
void ServerCommandQueue::reset() {
this->inner->reset();
}
void ServerCommandQueue::enqueue_command_string(const std::string_view &buffer) {
auto command = ReassembledCommand::allocate(buffer.length());
memcpy(command->command(), buffer.data(), command->length());
this->enqueue_command_execution(command);
}
void ServerCommandQueue::enqueue_command_execution(ReassembledCommand *command) {
assert(!command->next_command);
bool command_handling_scheduled = this->inner->enqueue(command);
if(!command_handling_scheduled) {
this->executor->enqueue_handler(this->command_handler);
}
}
#if 0
void ServerCommandQueue::execute_handle_command_packets(const std::chrono::system_clock::time_point& /* scheduled */) {
if(!this->client->getServer() || this->client->connectionState() >= ConnectionState::DISCONNECTING) {
return;
}
std::unique_ptr<ReassembledCommand, void(*)(ReassembledCommand*)> pending_command{nullptr, ReassembledCommand::free};
while(true) {
{
std::lock_guard pc_lock{this->pending_commands_lock};
pending_command.reset(this->pending_commands_head);
if(!pending_command) {
this->has_command_handling_scheduled = false;
return;
} else if(pending_command->next_command) {
this->pending_commands_head = pending_command->next_command;
} else {
this->pending_commands_head = nullptr;
this->pending_commands_tail = &this->pending_commands_head;
}
}
auto startTime = std::chrono::system_clock::now();
try {
this->client->handlePacketCommand(pipes::buffer_view{pending_command->command(), pending_command->length()});
} catch (std::exception& ex) {
logCritical(this->client->getServerId(), "{} Exception reached command execution root! {}", CLIENT_STR_LOG_PREFIX_(this->client), ex.what());
}
auto end = std::chrono::system_clock::now();
if(end - startTime > std::chrono::milliseconds(10)) {
logError(this->client->getServerId(),
"{} Handling of command packet needs more than 10ms ({}ms)",
CLIENT_STR_LOG_PREFIX_(this->client),
duration_cast<std::chrono::milliseconds>(end - startTime).count()
);
}
break; /* Maybe handle more than one command? Maybe some kind of time limit? */
}
auto voice_server = this->client->getVoiceServer();
if(voice_server) {
voice_server->schedule_command_handling(client);
}
}
#endif
ServerCommandExecutor::ServerCommandExecutor(size_t threads) : thread_count_{threads} {
this->threads_.reserve(threads);
for(size_t index{0}; index < threads; index++) {
auto& thread = this->threads_.emplace_back(&ServerCommandExecutor::thread_entry_point, this);
threads::name(thread, "cmd executor " + std::to_string(index + 1));
}
}
ServerCommandExecutor::~ServerCommandExecutor() {
this->shutdown();
}
void ServerCommandExecutor::shutdown() {
{
std::lock_guard handler_lock{this->handler_mutex};
this->handler_shutdown = true;
this->handler_notify.notify_all();
}
for(auto& thread : this->threads_) {
threads::save_join(thread, true);
}
}
void ServerCommandExecutor::enqueue_handler(const std::shared_ptr<ServerCommandHandler> &handler) {
std::lock_guard handler_lock{this->handler_mutex};
if(handler->next_handler) {
/* handler already scheduled */
return;
}
if(handler->executing) {
/* handler currently gets executed */
return;
}
if(this->handler_tail == &handler->next_handler) {
/* handler already schedules (current head) */
return;
}
*this->handler_tail = handler;
this->handler_tail = &handler->next_handler;
this->handler_notify.notify_one();
}
void ServerCommandExecutor::thread_entry_point(void *ptr_this) {
reinterpret_cast<ServerCommandExecutor*>(ptr_this)->executor();
}
void ServerCommandExecutor::executor() {
std::unique_lock handler_lock{this->handler_mutex};
while(!this->handler_shutdown) {
this->handler_notify.wait(handler_lock, [&]{
return this->handler_shutdown || this->handler_head != nullptr;
});
if(this->handler_shutdown) {
break;
}
if(!this->handler_head) {
continue;
}
auto executor = this->handler_head;
if(executor->next_handler) {
this->handler_head = executor->next_handler;
} else {
this->handler_head = nullptr;
this->handler_tail = &this->handler_head;
}
executor->executing = true;
handler_lock.unlock();
auto reschedule = executor->execute_handling();
handler_lock.lock();
executor->executing = false;
if(reschedule && !executor->next_handler && this->handler_tail != &executor->next_handler) {
*this->handler_tail = executor;
this->handler_tail = &executor->next_handler;
/* No need to notify anybody since we'll be executing him again */
}
}
}

89
server/src/client/shared/ServerCommandExecutor.h Normal file
View File

@ -0,0 +1,89 @@
#pragma once
#include <misc/spin_mutex.h>
#include <pipes/buffer.h>
#include <EventLoop.h>
namespace ts::server {
class VoiceClient;
}
namespace ts::server {
namespace command {
struct ReassembledCommand;
}
class ServerCommandExecutor;
class ServerCommandQueue;
struct ServerCommandQueueInner;
class ServerCommandHandler {
friend class ServerCommandQueue;
friend class ServerCommandExecutor;
public:
ServerCommandHandler() = default;
protected:
/**
* Handle a command.
* @returns `false` if all other commands should be dropped and no further command handling should be done.
* `true` on success.
*/
virtual bool handle_command(const std::string_view& /* raw command */) = 0;
private:
std::shared_ptr<ServerCommandQueueInner> inner{nullptr};
/* locked by ServerCommandExecutor::handler_mutex */
std::shared_ptr<ServerCommandHandler> next_handler{nullptr};
bool executing{false};
/**
* @returns `true` if more commands need to be handled and `false` if all commands have been handled.
*/
bool execute_handling();
};
class ServerCommandQueue {
public:
explicit ServerCommandQueue(std::shared_ptr<ServerCommandExecutor> /* executor */, std::unique_ptr<ServerCommandHandler> /* command handler */);
~ServerCommandQueue();
void reset();
void enqueue_command_string(const std::string_view& /* payload */);
/* Attention: The method will take ownership of the command */
void enqueue_command_execution(command::ReassembledCommand*);
private:
std::shared_ptr<ServerCommandExecutor> executor{};
std::shared_ptr<ServerCommandHandler> command_handler{};
std::shared_ptr<ServerCommandQueueInner> inner;
};
class ServerCommandExecutor {
friend class ServerCommandQueue;
public:
explicit ServerCommandExecutor(size_t /* threads */);
~ServerCommandExecutor();
[[nodiscard]] inline auto thread_count() const { return this->thread_count_; }
void shutdown();
protected:
void enqueue_handler(const std::shared_ptr<ServerCommandHandler>& /* handler */);
private:
size_t thread_count_;
std::vector<std::thread> threads_{};
std::mutex handler_mutex{};
std::condition_variable handler_notify{};
std::shared_ptr<ServerCommandHandler> handler_head{nullptr};
std::shared_ptr<ServerCommandHandler>* handler_tail{&this->handler_head};
bool handler_shutdown{false};
static void thread_entry_point(void* /* this ptr */);
void executor();
};
}

12
server/src/client/voice/PacketDecoder.cpp
View File

@ -16,18 +16,6 @@ using namespace ts::protocol;
using namespace ts::connection;
using namespace ts::server::server::udp;
ReassembledCommand *ReassembledCommand::allocate(size_t size) {
auto instance = (ReassembledCommand*) malloc(sizeof(ReassembledCommand) + size);
instance->length_ = size;
instance->capacity_ = size;
instance->next_command = nullptr;
return instance;
}
void ReassembledCommand::free(ReassembledCommand *command) {
::free(command);
}
PacketDecoder::PacketDecoder(ts::connection::CryptHandler *crypt_handler)
: crypt_handler_{crypt_handler} {
memtrack::allocated<PacketDecoder>(this);

45
server/src/client/voice/PacketDecoder.h
View File

@ -6,6 +6,7 @@
#include <protocol/Packet.h>
#include <protocol/generation.h>
#include <protocol/ringbuffer.h>
#include "../shared/RawCommand.h"
namespace ts::connection {
class CryptHandler;
@ -18,47 +19,6 @@ namespace ts::stats {
}
namespace ts::server::server::udp {
struct CommandFragment {
uint16_t packet_id{0};
uint16_t packet_generation{0};
uint8_t packet_flags{0};
uint32_t payload_length : 24;
pipes::buffer payload{};
CommandFragment() { this->payload_length = 0; }
CommandFragment(uint16_t packetId, uint16_t packetGeneration, uint8_t packetFlags, uint32_t payloadLength, pipes::buffer payload)
: packet_id{packetId}, packet_generation{packetGeneration}, packet_flags{packetFlags}, payload_length{payloadLength}, payload{std::move(payload)} {}
CommandFragment& operator=(const CommandFragment&) = default;
CommandFragment(const CommandFragment& other) = default;
CommandFragment(CommandFragment&&) = default;
};
static_assert(sizeof(CommandFragment) == 8 + sizeof(pipes::buffer));
struct ReassembledCommand {
public:
static ReassembledCommand* allocate(size_t /* command length */);
static void free(ReassembledCommand* /* command */);
[[nodiscard]] inline size_t length() const { return this->length_; }
inline void set_length(size_t length) { assert(this->capacity_ >= length); this->length_ = length; }
[[nodiscard]] inline size_t capacity() const { return this->capacity_; }
[[nodiscard]] inline const char* command() const { return (const char*) this + sizeof(ReassembledCommand); }
[[nodiscard]] inline char* command() { return (char*) this + sizeof(ReassembledCommand); }
[[nodiscard]] inline std::string_view command_view() const { return std::string_view{this->command(), this->length()}; }
mutable ReassembledCommand* next_command; /* nullptr by default */
private:
explicit ReassembledCommand() = default;
size_t capacity_;
size_t length_;
};
enum struct PacketProcessResult {
SUCCESS,
UNKNOWN_ERROR,
@ -91,6 +51,9 @@ namespace ts::stats {
};
class PacketDecoder {
using CommandFragment = command::CommandFragment;
using ReassembledCommand = command::ReassembledCommand;
typedef protocol::FullPacketRingBuffer<CommandFragment, 32, CommandFragment> command_fragment_buffer_t;
typedef std::array<command_fragment_buffer_t, 2> command_packet_reassembler;
public:

100
server/src/client/voice/ServerCommandExecutor.cpp
View File

@ -1,100 +0,0 @@
//
// Created by WolverinDEV on 29/07/2020.
//
#include "./ServerCommandExecutor.h"
#include "./PacketDecoder.h"
#include "./VoiceClientConnection.h"
using namespace ts;
using namespace ts::server::server::udp;
ServerCommandExecutor::ServerCommandExecutor(VoiceClient *client) : client{client} {}
ServerCommandExecutor::~ServerCommandExecutor() {
this->reset();
}
void ServerCommandExecutor::reset() {
std::unique_lock pc_lock{this->pending_commands_lock};
auto head = std::exchange(this->pending_commands_head, nullptr);
this->pending_commands_tail = &this->pending_commands_head;
pc_lock.unlock();
while(head) {
auto cmd = head->next_command;
ReassembledCommand::free(head);
head = cmd;
}
}
void ServerCommandExecutor::force_insert_command(const pipes::buffer_view &buffer) {
auto command = ReassembledCommand::allocate(buffer.length());
memcpy(command->command(), buffer.data_ptr(), command->length());
this->enqueue_command_execution(command);
}
void ServerCommandExecutor::enqueue_command_execution(ReassembledCommand *command) {
assert(!command->next_command);
bool command_handling_scheduled;
{
std::lock_guard pc_lock{this->pending_commands_lock};
*this->pending_commands_tail = command;
this->pending_commands_tail = &command->next_command;
command_handling_scheduled = std::exchange(this->has_command_handling_scheduled, true);
}
if(!command_handling_scheduled) {
auto voice_server = this->client->getVoiceServer();
if(voice_server) {
voice_server->schedule_command_handling(&*client);
}
}
}
void ServerCommandExecutor::execute_handle_command_packets(const std::chrono::system_clock::time_point& /* scheduled */) {
if(!this->client->getServer() || this->client->connectionState() >= ConnectionState::DISCONNECTING) {
return;
}
std::unique_ptr<ReassembledCommand, void(*)(ReassembledCommand*)> pending_command{nullptr, ReassembledCommand::free};
while(true) {
{
std::lock_guard pc_lock{this->pending_commands_lock};
pending_command.reset(this->pending_commands_head);
if(!pending_command) {
this->has_command_handling_scheduled = false;
return;
} else if(pending_command->next_command) {
this->pending_commands_head = pending_command->next_command;
} else {
this->pending_commands_head = nullptr;
this->pending_commands_tail = &this->pending_commands_head;
}
}
auto startTime = std::chrono::system_clock::now();
try {
this->client->handlePacketCommand(pipes::buffer_view{pending_command->command(), pending_command->length()});
} catch (std::exception& ex) {
logCritical(this->client->getServerId(), "{} Exception reached root tree! {}", CLIENT_STR_LOG_PREFIX_(this->client), ex.what());
}
auto end = std::chrono::system_clock::now();
if(end - startTime > std::chrono::milliseconds(10)) {
logError(this->client->getServerId(),
"{} Handling of command packet needs more than 10ms ({}ms)",
CLIENT_STR_LOG_PREFIX_(this->client),
duration_cast<std::chrono::milliseconds>(end - startTime).count()
);
}
break; /* Maybe handle more than one command? Maybe some kind of time limit? */
}
auto voice_server = this->client->getVoiceServer();
if(voice_server) {
voice_server->schedule_command_handling(client);
}
}

33
server/src/client/voice/ServerCommandExecutor.h
View File

@ -1,33 +0,0 @@
#pragma once
#include <misc/spin_mutex.h>
#include <pipes/buffer.h>
namespace ts::server {
class VoiceClient;
}
namespace ts::server::server::udp {
struct ReassembledCommand;
class ServerCommandExecutor {
public:
explicit ServerCommandExecutor(VoiceClient*);
~ServerCommandExecutor();
void reset();
void force_insert_command(const pipes::buffer_view& /* payload */);
void enqueue_command_execution(ReassembledCommand*); /* Attention: The method will take ownership of the command */
void execute_handle_command_packets(const std::chrono::system_clock::time_point& /* scheduled */);
private:
VoiceClient* client;
spin_mutex pending_commands_lock{};
ReassembledCommand* pending_commands_head{nullptr};
ReassembledCommand** pending_commands_tail{&pending_commands_head};
bool has_command_handling_scheduled{false}; /* locked by pending_commands_lock */
};
struct ReassembledCommand;
}

19
server/src/client/voice/VoiceClient.cpp
View File

@ -19,9 +19,10 @@ using namespace ts::protocol;
constexpr static auto kMaxWhisperClientNameLength{30};
constexpr static auto kWhisperClientUniqueIdLength{28}; /* base64 encoded SHA1 hash */
constexpr static auto kWhisperMaxHeaderLength{2 + 2 + 1 + 2 + kWhisperClientUniqueIdLength + 1 + kMaxWhisperClientNameLength};
VoiceClient::VoiceClient(const std::shared_ptr<VoiceServer>& server, const sockaddr_storage* address) : SpeakingClient(server->server->sql, server->server), voice_server(server) {
VoiceClient::VoiceClient(const std::shared_ptr<VoiceServer>& server, const sockaddr_storage* address) :
SpeakingClient(server->server->sql, server->server),
voice_server(server) {
assert(address);
memtrack::allocated<VoiceClient>(this);
memcpy(&this->remote_address, address, sizeof(sockaddr_storage));
@ -30,7 +31,11 @@ VoiceClient::VoiceClient(const std::shared_ptr<VoiceServer>& server, const socka
}
void VoiceClient::initialize() {
this->event_handle_packet = make_shared<event::ProxiedEventEntry<VoiceClient>>(dynamic_pointer_cast<VoiceClient>(this->ref()), &VoiceClient::execute_handle_packet);
auto ref_self = dynamic_pointer_cast<VoiceClient>(this->ref());
this->server_command_queue_ = std::make_unique<ServerCommandQueue>(
serverInstance->server_command_executor(),
std::make_unique<VoiceClientCommandHandler>(ref_self)
);
this->properties()[property::CLIENT_TYPE] = ClientType::CLIENT_TEAMSPEAK;
this->properties()[property::CLIENT_TYPE_EXACT] = ClientType::CLIENT_TEAMSPEAK;
@ -62,8 +67,8 @@ void VoiceClient::sendCommand0(const std::string_view& cmd, bool low, std::uniqu
#endif
}
void VoiceClient::tick(const std::chrono::system_clock::time_point &time) {
SpeakingClient::tick(time);
void VoiceClient::tick_server(const std::chrono::system_clock::time_point &time) {
SpeakingClient::tick_server(time);
{
ALARM_TIMER(A1, "VoiceClient::tick", milliseconds(3));
if(this->state == ConnectionState::CONNECTED) {
@ -259,10 +264,6 @@ void VoiceClient::finalDisconnect() {
}
}
void VoiceClient::execute_handle_packet(const std::chrono::system_clock::time_point &time) {
this->server_command_executor_.execute_handle_command_packets(time);
}
void VoiceClient::send_voice_packet(const pipes::buffer_view &voice_buffer, const SpeakingClient::VoicePacketFlags &flags) {
PacketFlag::PacketFlags packet_flags{PacketFlag::None};
packet_flags |= flags.encrypted ? 0U : PacketFlag::Unencrypted;

27
server/src/client/voice/VoiceClient.h
View File

@ -15,7 +15,7 @@
#include "VoiceClientConnection.h"
#include "src/server/PrecomputedPuzzles.h"
#include "../../lincense/TeamSpeakLicense.h"
#include "./ServerCommandExecutor.h"
#include "src/client/shared/ServerCommandExecutor.h"
//#define LOG_INCOMPING_PACKET_FRAGMENTS
//#define LOG_AUTO_ACK_AUTORESPONSE
@ -43,6 +43,7 @@ namespace ts {
}
class VirtualServer;
class VoiceClientCommandHandler;
class VoiceClient : public SpeakingClient {
friend class VirtualServer;
@ -53,7 +54,9 @@ namespace ts {
friend class io::IOServerHandler;
friend class server::udp::ServerCommandExecutor;
friend class server::udp::CryptSetupHandler;
using ServerCommandExecutor = ts::server::server::udp::ServerCommandExecutor;
friend class VoiceClientCommandHandler;
using ServerCommandExecutor = ts::server::ServerCommandQueue;
public:
VoiceClient(const std::shared_ptr<VoiceServer>& server, const sockaddr_storage*);
~VoiceClient() override;
@ -76,7 +79,7 @@ namespace ts {
[[nodiscard]] float current_packet_loss() const;
[[nodiscard]] inline auto& server_command_executor() { return this->server_command_executor_; }
[[nodiscard]] const auto& server_command_queue() { return this->server_command_queue_; }
private:
connection::VoiceClientConnection* connection;
@ -84,10 +87,10 @@ namespace ts {
std::shared_ptr<VoiceServer> voice_server;
void initialize();
virtual void tick(const std::chrono::system_clock::time_point &time) override;
virtual void tick_server(const std::chrono::system_clock::time_point &time) override;
/* Attention these handle callbacks are not thread save! */
void handlePacketCommand(const pipes::buffer_view&);
void handlePacketCommand(const std::string_view&);
public:
void send_voice_packet(const pipes::buffer_view &packet, const VoicePacketFlags &flags) override;
void send_voice(const std::shared_ptr<SpeakingClient>& /* source client */, uint16_t /* seq no */, uint8_t /* codec */, const void* /* payload */, size_t /* payload length */);
@ -114,10 +117,18 @@ namespace ts {
//Locked by finalDisconnect, disconnect and close connection
std::shared_ptr<std::thread> flushing_thread;
ServerCommandExecutor server_command_executor_{this};
std::unique_ptr<ServerCommandQueue> server_command_queue_{};
};
std::shared_ptr<event::ProxiedEventEntry<VoiceClient>> event_handle_packet;
void execute_handle_packet(const std::chrono::system_clock::time_point& /* scheduled */);
class VoiceClientCommandHandler : public ts::server::ServerCommandHandler {
public:
explicit VoiceClientCommandHandler(const std::shared_ptr<VoiceClient>& /* client */);
protected:
bool handle_command(const std::string_view &) override;
private:
std::weak_ptr<VoiceClient> client_ref;
};
}
}

13
server/src/client/voice/VoiceClientCommandHandler.cpp
View File

@ -14,7 +14,18 @@ using namespace ts::server;
using namespace ts::protocol;
using namespace ts;
void VoiceClient::handlePacketCommand(const pipes::buffer_view& command_string) {
VoiceClientCommandHandler::VoiceClientCommandHandler(const std::shared_ptr<VoiceClient> &client) : client_ref{client} {}
bool VoiceClientCommandHandler::handle_command(const std::string_view &command_string) {
auto client = this->client_ref.lock();
if(!client) {
return false;
}
client->handlePacketCommand(command_string);
return true;
}
void VoiceClient::handlePacketCommand(const std::string_view& command_string) {
std::unique_ptr<Command> command;
command_result result{};
try {

4
server/src/client/voice/VoiceClientConnection.h
View File

@ -3,7 +3,7 @@
#include "./PacketDecoder.h"
#include "./PacketEncoder.h"
#include "./PacketStatistics.h"
#include "./ServerCommandExecutor.h"
#include "src/client/shared/ServerCommandExecutor.h"
#include "CryptSetupHandler.h"
#include "PingHandler.h"
#include "VoiceClient.h"
@ -49,7 +49,7 @@ namespace ts {
using PacketEncoder = server::server::udp::PacketEncoder;
using PingHandler = server::server::udp::PingHandler;
using CryptSetupHandler = server::server::udp::CryptSetupHandler;
using ReassembledCommand = server::server::udp::ReassembledCommand;
using ReassembledCommand = server::command::ReassembledCommand;
using StatisticsCategory = stats::ConnectionStatistics::category;
public:

5
server/src/client/voice/VoiceClientConnectionPacketHandler.cpp
View File

@ -97,9 +97,11 @@ void VoiceClientConnection::handlePacketCommand(ReassembledCommand* command) {
break;
case CommandHandleResult::CONSUME_COMMAND:
ReassembledCommand::free(command);
return;
case CommandHandleResult::CLOSE_CONNECTION:
ReassembledCommand::free(command);
auto client = this->getCurrentClient();
assert(client); /* FIXME! */
client->close_connection(std::chrono::system_clock::time_point{});
@ -109,9 +111,10 @@ void VoiceClientConnection::handlePacketCommand(ReassembledCommand* command) {
auto client = this->getCurrentClient();
if(!client) {
ReassembledCommand::free(command);
/* TODO! */
return;
}
client->server_command_executor().enqueue_command_execution(command);
client->server_command_queue()->enqueue_command_execution(command);
}

56
server/src/client/web/WSWebClient.cpp
View File

@ -1,4 +1,5 @@
#include "WebClient.h"
#include "../shared/RawCommand.h"
#include <log/LogUtils.h>
#include <src/server/VoiceServer.h>
#include <src/InstanceHandler.h>
@ -83,38 +84,29 @@ void WebClient::handleMessageRead(int fd, short, void *) {
return;
}
auto pbuffer = buffer::allocate_buffer((size_t) length);
pbuffer.write(buffer, length);
auto command = command::ReassembledCommand::allocate((size_t) length);
memcpy(command->command(), buffer, (size_t) length);
{
lock_guard lock(this->queue_mutex);
this->queue_read.push_back(std::move(pbuffer));
}
this->registerMessageProcess();
this->command_queue->enqueue_command_execution(command);
}
void WebClient::enqueue_raw_packet(const pipes::buffer_view &msg) {
auto buffer = msg.owns_buffer() ? msg.own_buffer() : msg.own_buffer(); /* TODO: Use buffer::allocate_buffer(...) */
auto buffer = msg.own_buffer(); /* TODO: Use buffer::allocate_buffer(...) */
{
lock_guard queue_lock(this->queue_mutex);
this->queue_write.push_back(buffer);
}
{
lock_guard lock(this->event_mutex);
if(this->writeEvent)
if(this->writeEvent) {
event_add(this->writeEvent, nullptr);
}
}
this->connectionStatistics->logOutgoingPacket(stats::ConnectionStatistics::category::COMMAND, buffer.length());
}
void WebClient::registerMessageProcess() {
auto weakLock = this->_this;
if(serverInstance->getVoiceServerManager()->getState() == VirtualServerManager::STARTED)
serverInstance->getVoiceServerManager()->get_executor_loop()->schedule(this->event_handle_packet);
}
inline bool is_ssl_handshake_header(const pipes::buffer_view& buffer) {
inline bool is_ssl_handshake_header(const std::string_view& buffer) {
if(buffer.length() < 0x05) return false; //Header too small!
if(buffer[0] != 0x16) return false; //recordType=handshake
@ -125,36 +117,26 @@ inline bool is_ssl_handshake_header(const pipes::buffer_view& buffer) {
return true;
}
void WebClient::processNextMessage(const std::chrono::system_clock::time_point& /* scheduled */) {
bool WebClient::process_next_message(const std::string_view &buffer) {
lock_guard execute_lock(this->execute_mutex);
if(this->state != ConnectionState::INIT_HIGH && this->state != ConnectionState::INIT_LOW && this->state != ConnectionState::CONNECTED)
return;
unique_lock buffer_lock(this->queue_mutex);
if(this->queue_read.empty())
return;
auto buffer = this->queue_read.front();
this->queue_read.pop_front();
bool has_next = !this->queue_read.empty();
buffer_lock.unlock();
if(this->state != ConnectionState::INIT_HIGH && this->state != ConnectionState::INIT_LOW && this->state != ConnectionState::CONNECTED) {
return false;
}
this->connectionStatistics->logIncomingPacket(stats::ConnectionStatistics::category::COMMAND, buffer.length());
if(!this->ssl_detected) {
this->ssl_detected = true;
this->ssl_encrypted = is_ssl_handshake_header(buffer);
if(this->ssl_encrypted)
if(this->ssl_encrypted) {
logMessage(this->getServerId(), "[{}] Using encrypted basic connection.", CLIENT_STR_LOG_PREFIX_(this));
else
} else {
logMessage(this->getServerId(), "[{}] Using unencrypted basic connection.", CLIENT_STR_LOG_PREFIX_(this));
}
}
if(this->ssl_encrypted) {
this->ssl_handler.process_incoming_data(buffer);
this->ssl_handler.process_incoming_data(pipes::buffer_view{buffer.data(), buffer.length()});
} else {
this->ws_handler.process_incoming_data(buffer);
}
if(has_next) {
this->registerMessageProcess();
this->ws_handler.process_incoming_data(pipes::buffer_view{buffer.data(), buffer.length()});
}
return true;
}

26
server/src/client/web/WebClient.cpp
View File

@ -23,7 +23,10 @@ using namespace ts;
using namespace ts::server;
using namespace ts::protocol;
WebClient::WebClient(WebControlServer* server, int fd) : SpeakingClient(server->getTS()->getSql(), server->getTS()), handle(server), whisper_handler_{this} {
WebClient::WebClient(WebControlServer* server, int fd) :
SpeakingClient(server->getTS()->getSql(), server->getTS()),
handle{server},
whisper_handler_{this} {
memtrack::allocated<WebClient>(this);
assert(server->getTS());
@ -34,7 +37,8 @@ WebClient::WebClient(WebControlServer* server, int fd) : SpeakingClient(server->
}
void WebClient::initialize() {
this->event_handle_packet = make_shared<event::ProxiedEventEntry<WebClient>>(dynamic_pointer_cast<WebClient>(this->ref()), &WebClient::processNextMessage);
auto ref_this = dynamic_pointer_cast<WebClient>(this->ref());
this->command_queue = std::make_unique<ServerCommandQueue>(serverInstance->server_command_executor(), std::make_unique<WebClientCommandHandler>(ref_this));
int enabled = 1;
int disabled = 0;
@ -176,7 +180,7 @@ void WebClient::sendCommand(const ts::command_builder &command, bool low) {
this->sendJson(value);
} else {
auto data = command.build();
Command parsed_command = Command::parse(pipes::buffer_view{data.data(), data.length()}, true, false);
Command parsed_command = Command::parse(data, true, false);
this->sendCommand(parsed_command, low);
}
}
@ -221,7 +225,6 @@ bool WebClient::close_connection(const std::chrono::system_clock::time_point& ti
{
lock_guard lock(self_lock->queue_mutex);
flag_flushed &= self_lock->queue_read.empty();
flag_flushed &= self_lock->queue_write.empty();
}
@ -274,8 +277,8 @@ command_result WebClient::handleCommand(Command &command) {
return SpeakingClient::handleCommand(command);
}
void WebClient::tick(const std::chrono::system_clock::time_point& point) {
SpeakingClient::tick(point);
void WebClient::tick_server(const std::chrono::system_clock::time_point& point) {
SpeakingClient::tick_server(point);
if(this->ping.last_request + seconds(1) < point) {
if(this->ping.last_response > this->ping.last_request || this->ping.last_response + this->ping.timeout < point) {
@ -420,6 +423,17 @@ void WebClient::disconnectFinal() {
this->handle->unregisterConnection(static_pointer_cast<WebClient>(self_lock));
}
WebClientCommandHandler::WebClientCommandHandler(const std::shared_ptr<WebClient> &client) : client_ref{client} {}
bool WebClientCommandHandler::handle_command(const std::string_view &command) {
auto client = this->client_ref.lock();
if(!client) {
return false;
}
return client->process_next_message(command);
}
Json::CharReaderBuilder json_reader_builder = []() noexcept {
Json::CharReaderBuilder reader_builder;

24
server/src/client/web/WebClient.h
View File

@ -11,12 +11,15 @@
#include <json/json.h>
#include <EventLoop.h>
#include "../shared/WhisperHandler.h"
#include "../shared/ServerCommandExecutor.h"
namespace ts::server {
class WebControlServer;
class WebClientCommandHandler;
class WebClient : public SpeakingClient {
friend class WebControlServer;
friend class WebClientCommandHandler;
public:
WebClient(WebControlServer*, int socketFd);
~WebClient() override;
@ -34,7 +37,7 @@ namespace ts::server {
[[nodiscard]] inline std::chrono::nanoseconds client_ping_layer_7() const { return this->js_ping.value; }
protected:
void tick(const std::chrono::system_clock::time_point&) override; /* Every 500ms */
void tick_server(const std::chrono::system_clock::time_point&) override; /* Every 500ms */
void applySelfLock(const std::shared_ptr<WebClient> &cl){ _this = cl; }
private:
@ -52,8 +55,6 @@ namespace ts::server {
::event* readEvent;
::event* writeEvent;
std::shared_ptr<event::ProxiedEventEntry<WebClient>> event_handle_packet;
struct {
uint8_t current_id{0};
std::chrono::system_clock::time_point last_request;
@ -73,7 +74,7 @@ namespace ts::server {
} js_ping;
std::mutex queue_mutex;
std::deque<pipes::buffer> queue_read;
std::unique_ptr<ServerCommandQueue> command_queue{};
std::deque<pipes::buffer> queue_write;
threads::Mutex execute_mutex; /* needs to be recursive! */
@ -88,8 +89,8 @@ namespace ts::server {
void handleMessageWrite(int, short, void*);
void enqueue_raw_packet(const pipes::buffer_view& /* buffer */);
void processNextMessage(const std::chrono::system_clock::time_point& /* scheduled */);
void registerMessageProcess();
/* TODO: Put the message processing part into the IO loop and not into command processing! */
bool process_next_message(const std::string_view& buffer);
//WS events
void onWSConnected();
@ -110,5 +111,16 @@ namespace ts::server {
command_result handleCommandWhisperSessionInitialize(Command &command);
command_result handleCommandWhisperSessionReset(Command &command);
};
class WebClientCommandHandler : public ts::server::ServerCommandHandler {
public:
explicit WebClientCommandHandler(const std::shared_ptr<WebClient>& /* client */);
protected:
bool handle_command(const std::string_view &) override;
private:
std::weak_ptr<WebClient> client_ref;
};
}
#endif

36
server/src/manager/IpListManager.cpp
View File

@ -11,36 +11,43 @@
using namespace std;
using namespace ts;
IpListManager::IpListManager(const std::string& file, const std::deque<std::string>& def) : file(file), default_entries(def) { }
IpListManager::IpListManager(std::string file, const std::deque<std::string>& def) : file(std::move(file)), default_entries(def) { }
bool file_exists(const std::string& name) {
struct stat buffer{};
return (stat (name.c_str(), &buffer) == 0);
}
inline string strip(std::string message) {
while(!message.empty()) {
if(message[0] == ' ')
if(message[0] == ' ') {
message = message.substr(1);
else if(message[message.length() - 1] == ' ')
} else if(message[message.length() - 1] == ' ') {
message = message.substr(0, message.length() - 1);
else break;
} else {
break;
}
}
return message;
}
bool IpListManager::reload(std::string& error) {
if(!file_exists(this->file)) {
ofstream os(this->file);
ofstream os{this->file};
if(!os) {
error = "Could not create default file!";
return false;
}
for(const auto& entry : this->default_entries)
for(const auto& entry : this->default_entries) {
os << entry << endl;
}
os.flush();
os.close();
}
ifstream stream(this->file);
ifstream stream{this->file};
if(!stream) {
error = "Failed to read file!";
return false;
@ -51,22 +58,27 @@ bool IpListManager::reload(std::string& error) {
while(getline(stream, line)) {
line_number++;
line = strip(line);
if(line.empty() || line[0] == '#') continue;
if(line.empty() || line[0] == '#') {
continue;
}
IPEntry result{};
if(!this->parse_entry(result, line))
if(!this->parse_entry(result, line)) {
logError(0, "Failed to parse ip entry at line {} of file {}. Line: '{}'", line_number, this->file, line);
else
} else {
this->entries.push_back(result);
}
}
return true;
}
bool IpListManager::contains(const sockaddr_storage &address) {
for(const auto& entry : this->entries)
if(net::address_equal_ranged(address, entry.address, entry.range))
for(const auto& entry : this->entries) {
if(net::address_equal_ranged(address, entry.address, entry.range)) {
return true;
}
}
return false;
}

2
server/src/manager/IpListManager.h
View File

@ -11,7 +11,7 @@ namespace ts {
uint8_t range; /* [0;32] or [0;128] */
};
public:
IpListManager(const std::string&, const std::deque<std::string>&);
IpListManager(std::string , const std::deque<std::string>&);
bool reload(std::string&);

2
server/src/server/POWHandler.cpp
View File

@ -270,7 +270,7 @@ void POWHandler::handle_puzzle_solve(const std::shared_ptr<ts::server::POWHandle
auto voice_client = this->register_verified_client(client);
if(voice_client) {
auto rcommand = server::udp::ReassembledCommand::allocate(command.length());
auto rcommand = command::ReassembledCommand::allocate(command.length());
memcpy(rcommand->command(), command.data_ptr(), rcommand->length());
voice_client->connection->handlePacketCommand(rcommand);
client->state = LowHandshakeState::COMPLETED;

467
server/src/server/QueryServer.cpp
View File

@ -5,14 +5,12 @@
#include "QueryServer.h"
#include <algorithm>
#include <netinet/tcp.h>
#include <poll.h>
#include <src/VirtualServer.h>
#include <src/client/query/QueryClient.h>
#include <src/client/InternalClient.h>
#include <misc/rnd.h>
#include <src/InstanceHandler.h>
#include <log/LogUtils.h>
#include <src/client/ConnectedClient.h>
#include <ThreadPool/ThreadHelper.h>
using namespace std;
using namespace std::chrono;
@ -24,22 +22,19 @@ using namespace ts::server;
#endif
QueryServer::QueryServer(sql::SqlManager* db) : sql(db) {
this->_executePool = new threads::ThreadPool(4, "EXEC Query");
}
QueryServer::~QueryServer() {
stop();
this->_executePool->shutdown();
delete this->_executePool;
}
void QueryServer::unregisterConnection(const shared_ptr<QueryClient> &client) {
{
lock_guard lock(this->connected_clients_lock);
auto found = std::find(this->connectedClients.begin(), this->connectedClients.end(), client);
if(found != this->connectedClients.end())
this->connectedClients.erase(found);
lock_guard lock(this->connected_clients_mutex);
auto found = std::find(this->connected_clients.begin(), this->connected_clients.end(), client);
if(found != this->connected_clients.end())
this->connected_clients.erase(found);
else
logError(LOG_QUERY, "Attempted to unregister an invalid connection!");
}
@ -52,7 +47,7 @@ void QueryServer::unregisterConnection(const shared_ptr<QueryClient> &client) {
/* client->handle = nullptr; */
}
bool QueryServer::start(const deque<shared_ptr<QueryServer::Binding>> &bindings, std::string &error) {
bool QueryServer::start(const deque<shared_ptr<QueryServer::Binding>> &bindings_, std::string &error) {
if(this->active) {
error = "already started";
return false;
@ -61,40 +56,49 @@ bool QueryServer::start(const deque<shared_ptr<QueryServer::Binding>> &bindings,
/* load ip black/whitelist */
{
ip_blacklist.reset(new IpListManager("query_ip_blacklist.txt", {"#A new line separated address blacklist", "#", "#For example if we dont want google:", "8.8.8.8"}));
ip_whitelist.reset(new IpListManager("query_ip_whitelist.txt", {"#A new line separated address whitelist", "#Every ip have no flood and login attempt limit!", "127.0.0.1/8", "::1"}));
string error;
if(!this->ip_blacklist->reload(error)) logError(LOG_QUERY, "Failed to load query blacklist: {}", error);
if(!this->ip_whitelist->reload(error)) logError(LOG_QUERY, "Failed to load query whitelist: {}", error);
ip_blacklist = std::make_unique<IpListManager>("query_ip_blacklist.txt", std::deque<std::string>{"#A new line separated address blacklist", "#", "#For example if we dont want google:", "8.8.8.8"});
ip_whitelist = std::make_unique<IpListManager>("query_ip_whitelist.txt", std::deque<std::string>{"#A new line separated address whitelist", "#Every ip have no flood and login attempt limit!", "127.0.0.1/8", "::1"});
if(!this->ip_blacklist->reload(error)) {
logError(LOG_QUERY, "Failed to load query blacklist: {}", error);
}
if(!this->ip_whitelist->reload(error)) {
logError(LOG_QUERY, "Failed to load query whitelist: {}", error);
}
error.clear();
}
/* reserve backup file descriptor in case that the max file descriptors have been reached */
{
this->server_reserve_fd = dup(1);
if(this->server_reserve_fd < 0)
if(this->server_reserve_fd < 0) {
logWarning(LOG_QUERY, "Failed to reserve a backup accept file descriptor. ({} | {})", errno, strerror(errno));
}
}
/* setup event bases */
{
this->eventLoop = event_base_new();
this->ioThread = new threads::Thread(THREAD_SAVE_OPERATIONS | THREAD_EXECUTE_LATER, [&]{
this->event_io_loop = event_base_new();
this->event_io_thread = std::thread{[&]{
while(this->active) {
debugMessage(LOG_QUERY, "Entering event loop ({})", (void*) this->eventLoop);
event_base_loop(this->eventLoop, EVLOOP_NO_EXIT_ON_EMPTY);
debugMessage(LOG_QUERY, "Entering event loop ({})", (void*) this->event_io_loop);
event_base_loop(this->event_io_loop, EVLOOP_NO_EXIT_ON_EMPTY);
if(this->active) {
debugMessage(LOG_QUERY, "Event loop exited ({}). No active events. Sleeping 1 seconds", (void*) this->eventLoop);
debugMessage(LOG_QUERY, "Event loop exited ({}). No active events. Sleeping 1 seconds", (void*) this->event_io_loop);
this_thread::sleep_for(seconds(1));
} else {
debugMessage(LOG_QUERY, "Event loop exited ({})", (void*) this->eventLoop);
debugMessage(LOG_QUERY, "Event loop exited ({})", (void*) this->event_io_loop);
}
}
});
this->ioThread->name("EVENT Query").execute();
}};
threads::name(this->event_io_thread, "query io");
}
for(auto& binding : bindings) {
binding->file_descriptor = socket(binding->address.ss_family, SOCK_STREAM | SOCK_NONBLOCK, 0);
for(auto& binding : bindings_) {
binding->file_descriptor = socket(binding->address.ss_family, (unsigned) SOCK_STREAM | (unsigned) SOCK_NONBLOCK, 0);
if(binding->file_descriptor < 0) {
logError(LOG_QUERY, "Failed to bind server to {}. (Failed to create socket: {} | {})", binding->as_string(), errno, strerror(errno));
continue;
@ -102,16 +106,23 @@ bool QueryServer::start(const deque<shared_ptr<QueryServer::Binding>> &bindings,
int enable = 1, disabled = 0;
if (setsockopt(binding->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int)) < 0)
if (setsockopt(binding->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int)) < 0) {
logWarning(LOG_QUERY, "Failed to activate SO_REUSEADDR for binding {} ({} | {})", binding->as_string(), errno, strerror(errno));
if(setsockopt(binding->file_descriptor, IPPROTO_TCP, TCP_NOPUSH, &disabled, sizeof disabled) < 0)
}
if(setsockopt(binding->file_descriptor, IPPROTO_TCP, TCP_NOPUSH, &disabled, sizeof disabled) < 0) {
logWarning(LOG_QUERY, "Failed to deactivate TCP_NOPUSH for binding {} ({} | {})", binding->as_string(), errno, strerror(errno));
}
if(binding->address.ss_family == AF_INET6) {
if(setsockopt(binding->file_descriptor, IPPROTO_IPV6, IPV6_V6ONLY, &enable, sizeof(int)) < 0)
if(setsockopt(binding->file_descriptor, IPPROTO_IPV6, IPV6_V6ONLY, &enable, sizeof(int)) < 0) {
logWarning(LOG_QUERY, "Failed to activate IPV6_V6ONLY for IPv6 binding {} ({} | {})", binding->as_string(), errno, strerror(errno));
}
if(fcntl(binding->file_descriptor, F_SETFD, FD_CLOEXEC) < 0)
}
if(fcntl(binding->file_descriptor, F_SETFD, FD_CLOEXEC) < 0) {
logWarning(LOG_QUERY, "Failed to set flag FD_CLOEXEC for binding {} ({} | {})", binding->as_string(), errno, strerror(errno));
}
if (bind(binding->file_descriptor, (struct sockaddr *) &binding->address, sizeof(binding->address)) < 0) {
@ -126,7 +137,7 @@ bool QueryServer::start(const deque<shared_ptr<QueryServer::Binding>> &bindings,
continue;
}
binding->event_accept = event_new(this->eventLoop, binding->file_descriptor, EV_READ | EV_PERSIST, [](int a, short b, void* c){ ((QueryServer *) c)->on_client_receive(a, b, c); }, this);
binding->event_accept = event_new(this->event_io_loop, binding->file_descriptor, EV_READ | EV_PERSIST, [](int a, short b, void* c){ ((QueryServer *) c)->on_client_receive(a, b, c); }, this);
event_add(binding->event_accept, nullptr);
this->bindings.push_back(binding);
}
@ -137,80 +148,105 @@ bool QueryServer::start(const deque<shared_ptr<QueryServer::Binding>> &bindings,
return false;
}
this->tickingId = serverInstance->scheduler()->schedule("query", bind(&QueryServer::tick, this), seconds(1));
this->tick_active = true;
this->tick_thread = std::thread{[&]{ this->tick_executor(); }};
threads::name(this->tick_thread, "query tick");
return true;
}
void QueryServer::stop() {
if(!this->running())
if(!this->active) {
return;
active = false;
serverInstance->scheduler()->cancelTask("query");
this->connected_clients_lock.lock();
auto connected_clients = this->connectedClients;
this->connected_clients_lock.unlock();
Command cmd("serverstop");
cmd["stopped"] = true;
for(const auto &client : connected_clients){
client->sendCommand(cmd);
client->disconnect("server stopped");
}
{
auto now = system_clock::now();
while(!this->connectedClients.empty()) {
if(now + seconds(5) < system_clock::now()) {
logError(LOG_QUERY, "Failed to disconnect all clients!");
break;
}
threads::self::sleep_for(milliseconds(100));
}
}
for(auto thread : this->threads){
if(thread->state() == threads::ThreadState::RUNNING)
thread->join();
delete thread;
}
this->threads.clear();
this->active = false;
/* 1. Shutdown all bindings so we don't get any new queries */
for(auto& binding : this->bindings) {
if(binding->event_accept) {
event_del_block(binding->event_accept);
event_free(binding->event_accept);
binding->event_accept = nullptr;
}
if(binding->file_descriptor > 0) {
if(shutdown(binding->file_descriptor, SHUT_RDWR) < 0)
logWarning(LOG_QUERY, "Failed to shutdown socket for binding {} ({} | {}).", binding->as_string(), errno, strerror(errno));
if(close(binding->file_descriptor) < 0)
/* Shutdown not needed since we're not connected. A shutdown would result in "Transport endpoint is not connected". */
if(close(binding->file_descriptor) < 0) {
logError(LOG_QUERY, "Failed to close socket for binding {} ({} | {}).", binding->as_string(), errno, strerror(errno));
}
binding->file_descriptor = -1;
}
}
this->bindings.clear();
if(this->eventLoop)
event_base_loopexit(this->eventLoop, nullptr);
if(this->ioThread) {
if(this->ioThread->join(seconds(3)) != 0) {
logCritical(LOG_QUERY, "Failed to terminate event loop!");
this->ioThread->detach();
}
}
delete this->ioThread;
this->ioThread = nullptr;
/* 2. Disconnect all connected query clients */
{
ts::command_builder notify{"serverstop"};
notify.put_unchecked(0, "stopped", "1");
if(this->eventLoop) {
event_base_free(this->eventLoop);
this->eventLoop = nullptr;
std::lock_guard client_lock{this->connected_clients_mutex};
for(const auto &client : this->connected_clients) {
client->sendCommand(notify, false);
client->disconnect("server stopped");
/*
* Shortcircuiting the disconnect since we don't want the full "server leave" disconnect.
* We only wan't to prevent the client form receiving any more notifications.
*/
this->execute_query_disconnect(client, true);
}
}
/* Await all clients to disconnect within 5 seconds. */
{
std::unique_lock client_lock{this->connected_clients_mutex};
this->connected_client_disconnected_notify.wait_for(client_lock, std::chrono::seconds{5}, [&]{
return this->connected_clients.empty();
});
}
/* 3. Shutdown the query event loop (to finish of client disconnects as well) */
{
std::lock_guard tick_lock{this->tick_mutex};
this->tick_active = false;
this->tick_notify.notify_all();
}
threads::save_join(this->tick_thread, true);
/*
* 4. Force disconnect pending clients.
*/
{
std::unique_lock client_lock{this->connected_clients_mutex};
auto connected_clients_ = std::move(this->connected_clients);
client_lock.unlock();
if(!connected_clients_.empty()) {
logWarning(LOG_QUERY, "Failed to normally disconnect {} query clients. Closing connection.", connected_clients_.size());
for(const auto& client : this->connected_clients) {
this->execute_query_connection_close(client, false);
}
}
}
/* 5. Shutdown the io event loop */
if(this->event_io_loop) {
event_base_loopexit(this->event_io_loop, nullptr);
}
threads::save_join(this->event_io_thread, false);
if(this->event_io_loop) {
event_base_free(this->event_io_loop);
this->event_io_loop = nullptr;
}
/* 6. Cleanup the servers reserve file descriptor */
if(this->server_reserve_fd > 0) {
if(close(this->server_reserve_fd) < 0)
if(close(this->server_reserve_fd) < 0) {
logError(LOG_QUERY, "Failed to close backup file descriptor ({} | {})", errno, strerror(errno));
}
}
this->server_reserve_fd = -1;
}
@ -221,7 +257,7 @@ inline std::string logging_address(const sockaddr_storage& address) {
}
inline void send_direct_disconnect(const sockaddr_storage& address, int file_descriptor, const char* message, size_t message_length) {
auto _non_block = [&]{
auto enable_non_block = [&]{
int flags = fcntl(file_descriptor, F_GETFL, 0);
if (flags == -1) {
debugMessage(LOG_QUERY, "[{}] Failed to set socket to nonblock. Flag query failed ({} | {})", logging_address(address), errno, strerror(errno));
@ -234,20 +270,22 @@ inline void send_direct_disconnect(const sockaddr_storage& address, int file_des
return;
}
};
_non_block();
enable_non_block();
{
struct timeval timeout{};
timeout.tv_sec = 5;
timeout.tv_usec = 0;
if (setsockopt (file_descriptor, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout)) < 0)
if (setsockopt (file_descriptor, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout)) < 0) {
debugMessage(LOG_QUERY, "[{}] Failed to set the send timeout on socket", logging_address(address));
}
}
bool broken_pipe = false;
auto _send = [&](const char* data, size_t length) {
if(broken_pipe)
auto send_ = [&](const char* data, size_t length) {
if(broken_pipe) {
return;
}
size_t written_bytes = 0;
while(written_bytes < length) {
@ -263,15 +301,15 @@ inline void send_direct_disconnect(const sockaddr_storage& address, int file_des
};
/* we could ignore errors here */
_send(config::query::motd.data(), config::query::motd.size());
_send(message, message_length);
send_(config::query::motd.data(), config::query::motd.size());
send_(message, message_length);
/* "flush" with the last new line and then close */
int flag = 1;
if(setsockopt(file_descriptor, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)) < 0) {
debugMessage(LOG_QUERY, "[{}] Failed to enabled TCP no delay to flush the direct query disconnect socket ({} | {}).", logging_address(address), errno, strerror(errno));
}
_send(config::query::newlineCharacter.data(), config::query::newlineCharacter.size());
send_(config::query::newlineCharacter.data(), config::query::newlineCharacter.size());
if(shutdown(file_descriptor, SHUT_RDWR) < 0) {
debugMessage(LOG_QUERY, "[{}] Failed to shutdown socket ({} | {}).", logging_address(address), errno, strerror(errno));
@ -284,23 +322,25 @@ inline void send_direct_disconnect(const sockaddr_storage& address, int file_des
//dummyfdflood
//dummyfdflood clear
void QueryServer::on_client_receive(int _server_file_descriptor, short ev, void *arg) {
void QueryServer::on_client_receive(int server_file_descriptor, short, void *) {
sockaddr_storage remote_address{};
memset(&remote_address, 0, sizeof(sockaddr_in));
socklen_t address_length = sizeof(remote_address);
int file_descriptor = accept(_server_file_descriptor, (struct sockaddr *) &remote_address, &address_length);
if (file_descriptor < 0) {
if(errno == EAGAIN)
int client_file_descriptor = accept(server_file_descriptor, (struct sockaddr *) &remote_address, &address_length);
if (client_file_descriptor < 0) {
if(errno == EAGAIN) {
return;
}
if(errno == EMFILE || errno == ENFILE) {
if(errno == EMFILE)
if(errno == EMFILE) {
logError(LOG_QUERY, "Server ran out file descriptors. Please increase the process file descriptor limit or decrease the instance variable 'serverinstance_query_max_connections'");
else
} else {
logError(LOG_QUERY, "Server ran out file descriptors. Please increase the process and system-wide file descriptor limit or decrease the instance variable 'serverinstance_query_max_connections'");
}
bool tmp_close_success = false;
bool tmp_close_success{false};
{
lock_guard reserve_fd_lock(server_reserve_fd_lock);
if(this->server_reserve_fd > 0) {
@ -314,31 +354,34 @@ void QueryServer::on_client_receive(int _server_file_descriptor, short ev, void
this->server_reserve_fd = 0;
errno = 0;
file_descriptor = accept(_server_file_descriptor, (struct sockaddr *) &remote_address, &address_length);
if(file_descriptor < 0) {
if(errno == EMFILE || errno == ENFILE)
client_file_descriptor = accept(server_file_descriptor, (struct sockaddr *) &remote_address, &address_length);
if(client_file_descriptor < 0) {
if(errno == EMFILE || errno == ENFILE) {
debugMessage(LOG_QUERY, "[{}] Even with freeing the reserved descriptor accept failed. Attempting to reclaim reserved file descriptor", logging_address(remote_address));
else if(errno == EAGAIN);
else {
} else if(errno == EAGAIN) {
/* Nothing to do */
} else {
debugMessage(LOG_QUERY, "[{}] Failed to accept client with reserved file descriptor. ({} | {})", logging_address(remote_address), errno, strerror(errno));
}
this->server_reserve_fd = dup(1);
if(this->server_reserve_fd < 0)
if(this->server_reserve_fd < 0) {
debugMessage(LOG_QUERY, "[{}] Failed to reclaim reserved file descriptor. Future clients cant be accepted!", logging_address(remote_address));
else
} else {
tmp_close_success = true;
}
return;
}
debugMessage(LOG_QUERY, "[{}] Successfully accepted client via reserved descriptor (fd: {}). Initializing socket and sending MOTD and disconnect.", logging_address(remote_address), file_descriptor);
debugMessage(LOG_QUERY, "[{}] Successfully accepted client via reserved descriptor (fd: {}). Initializing socket and sending MOTD and disconnect.", logging_address(remote_address), client_file_descriptor);
static auto resource_limit_error = R"(error id=57344 msg=query\sserver\sresource\slimit\sreached extra_msg=file\sdescriptor\slimit\sexceeded)";
send_direct_disconnect(remote_address, file_descriptor, resource_limit_error, strlen(resource_limit_error));
send_direct_disconnect(remote_address, client_file_descriptor, resource_limit_error, strlen(resource_limit_error));
this->server_reserve_fd = dup(1);
if(this->server_reserve_fd < 0)
if(this->server_reserve_fd < 0) {
debugMessage(LOG_QUERY, "Failed to reclaim reserved file descriptor. Future clients cant be accepted!");
else
} else {
tmp_close_success = true;
}
logMessage(LOG_QUERY, "[{}] Dropping new query connection attempt because of too many open file descriptors.", logging_address(remote_address));
};
_();
@ -347,8 +390,9 @@ void QueryServer::on_client_receive(int _server_file_descriptor, short ev, void
if(!tmp_close_success) {
debugMessage(LOG_QUERY, "Sleeping two seconds because we're currently having no resources for this user. (Removing the accept event)");
for(auto& binding : this->bindings)
for(auto& binding : this->bindings) {
event_del_noblock(binding->event_accept);
}
accept_event_deleted = system_clock::now();
return;
}
@ -357,21 +401,22 @@ void QueryServer::on_client_receive(int _server_file_descriptor, short ev, void
logMessage(LOG_QUERY, "Got an error while accepting a new client. (errno: {}, message: {})", errno, strerror(errno));
return;
}
{
unique_lock lock(this->connected_clients_lock);
unique_lock lock{this->connected_clients_mutex};
auto max_connections = serverInstance->properties()[property::SERVERINSTANCE_QUERY_MAX_CONNECTIONS].as<size_t>();
if(max_connections > 0 && max_connections <= this->connectedClients.size()) {
if(max_connections > 0 && max_connections <= this->connected_clients.size()) {
lock.unlock();
logMessage(LOG_QUERY, "[{}] Dropping new query connection attempt because of too many connected query clients.", logging_address(remote_address));
static auto query_server_full = R"(error id=4611 msg=max\sclients\sreached)";
send_direct_disconnect(remote_address, file_descriptor, query_server_full, strlen(query_server_full));
send_direct_disconnect(remote_address, client_file_descriptor, query_server_full, strlen(query_server_full));
return;
}
auto max_ip_connections = serverInstance->properties()[property::SERVERINSTANCE_QUERY_MAX_CONNECTIONS_PER_IP].as<size_t>();
if(max_ip_connections > 0) {
size_t connection_count = 0;
for(auto& client : this->connectedClients) {
for(auto& client : this->connected_clients) {
if(net::address_equal(client->remote_address, remote_address))
connection_count++;
}
@ -380,23 +425,24 @@ void QueryServer::on_client_receive(int _server_file_descriptor, short ev, void
lock.unlock();
logMessage(LOG_QUERY, "[{}] Dropping new query connection attempt because of too many simultaneously connected session from this ip.", logging_address(remote_address));
static auto query_server_full = R"(error id=4610 msg=too\smany\ssimultaneously\sconnected\ssessions)";//
send_direct_disconnect(remote_address, file_descriptor, query_server_full, strlen(query_server_full));
send_direct_disconnect(remote_address, client_file_descriptor, query_server_full, strlen(query_server_full));
return;
}
}
}
shared_ptr<QueryClient> client = std::make_shared<QueryClient>(this, file_descriptor);
client->applySelfLock(client);
auto client = std::make_shared<QueryClient>(this, client_file_descriptor);
memcpy(&client->remote_address, &remote_address, sizeof(remote_address));
client->initialize_self_reference(client);
{
lock_guard lock(this->connected_clients_lock);
this->connectedClients.push_back(client);
lock_guard lock(this->connected_clients_mutex);
this->connected_clients.push_back(client);
}
client->preInitialize();
if(client->readEvent) {
event_add(client->readEvent, nullptr);
if(client->event_read) {
event_add(client->event_read, nullptr);
}
logMessage(LOG_QUERY, "Got new client from {}", client->getLoggingPeerIp() + ":" + to_string(client->getPeerPort()));
}
@ -551,43 +597,182 @@ bool QueryServer::change_query_password(const std::shared_ptr<ts::server::QueryA
return true;
}
void QueryServer::tick() {
decltype(this->connectedClients) clCopy;
void QueryServer::tick_clients() {
decltype(this->connected_clients) connected_clients_;
{
lock_guard lock(this->connected_clients_lock);
clCopy = this->connectedClients;
lock_guard lock(this->connected_clients_mutex);
connected_clients_ = this->connected_clients;
}
for(const auto& cl : clCopy) cl->queryTick();
for(const auto& cl : connected_clients_) {
cl->tick_query();
}
{
threads::MutexLock lock(this->loginLock);
std::lock_guard connect_lock{this->client_connect_mutex};
vector<std::string> qbanErase;
for(auto& elm : this->queryBann) {
if(elm.second < system_clock::now())
qbanErase.push_back(elm.first);
std::vector<std::string> erase_bans{};
erase_bans.reserve(32);
for(auto& elm : this->client_connect_bans) {
if(elm.second < system_clock::now()) {
erase_bans.push_back(elm.first);
}
}
for(const auto& ip : qbanErase)
this->queryBann.erase(ip);
if(system_clock::now() - seconds(5) < lastDecrease) {
this->lastDecrease = system_clock::now();
for(const auto& ip : erase_bans) {
this->client_connect_bans.erase(ip);
}
vector<std::string> lattempErase;
for(auto& elm : this->loginAttempts) {
if(elm.second == 0)
lattempErase.push_back(elm.first);
else
if(system_clock::now() - seconds(5) < client_connect_last_decrease) {
this->client_connect_last_decrease = system_clock::now();
std::vector<std::string> erase_attempts{};
for(auto& elm : this->client_connect_count) {
if(elm.second == 0) {
erase_attempts.push_back(elm.first);
} else {
elm.second--;
}
for(const auto& ip : qbanErase)
this->loginAttempts.erase(ip);
}
for(const auto& ip : erase_bans) {
this->client_connect_count.erase(ip);
}
}
}
if(this->accept_event_deleted.time_since_epoch().count() != 0 && accept_event_deleted + seconds(5) < system_clock::now()) {
debugMessage(LOG_QUERY, "Readding accept event and try again if we have enough resources again.");
for(auto& binding : this->bindings)
for(auto& binding : this->bindings) {
event_add(binding->event_accept, nullptr);
}
accept_event_deleted = system_clock::time_point{};
}
}
void QueryServer::tick_executor() {
bool tick_clients;
while(this->tick_active) {
std::unique_lock tick_lock{this->tick_mutex};
this->tick_notify.wait_until(tick_lock, this->tick_next_client_timestamp, [&]{
return !this->tick_active || !this->tick_pending_disconnects.empty() || !this->tick_pending_connection_close.empty();
});
auto current_timestamp = std::chrono::system_clock::now();
if(current_timestamp > this->tick_next_client_timestamp) {
this->tick_next_client_timestamp = current_timestamp;
tick_clients = true;
} else {
tick_clients = false;
}
auto pending_disconnects = std::move(this->tick_pending_disconnects);
auto pending_closes = std::move(this->tick_pending_connection_close);
if(!this->tick_active) {
if(this->tick_pending_connection_close.empty() && this->tick_pending_connection_close.empty()) {
/* We're done with our work */
break;
}
}
tick_lock.unlock();
if(tick_clients) {
this->tick_clients();
}
for(const auto& pending_disconnect : pending_disconnects) {
auto client = pending_disconnect.lock();
if(!client) {
continue;
}
this->execute_query_disconnect(client, false);
}
for(const auto& pending_close : pending_closes) {
auto client = pending_close.lock();
if(!client) {
continue;
}
this->execute_query_connection_close(client, true);
}
}
}
void QueryServer::enqueue_query_disconnect(const std::shared_ptr<QueryClient> &client) {
std::lock_guard lock{this->tick_mutex};
if(!this->tick_active) {
logCritical(LOG_GENERAL, "Tried to close a query connection without an active query event loop.");
return;
}
this->tick_pending_disconnects.push_back(client);
this->tick_notify.notify_one();
}
void QueryServer::enqueue_query_connection_close(const std::shared_ptr<QueryClient> &client) {
std::lock_guard lock{this->tick_mutex};
if(!this->tick_active) {
logCritical(LOG_GENERAL, "Tried to close a query connection without an active query event loop.");
return;
}
this->tick_pending_connection_close.push_back(client);
this->tick_notify.notify_one();
}
void QueryServer::execute_query_disconnect(const std::shared_ptr<QueryClient> &client, bool shutdown_disconnect) {
{
std::lock_guard state_lock{client->state_lock};
if(client->state >= ConnectionState::DISCONNECTING) {
/* client will already be disconnected */
return;
}
client->state = ConnectionState::DISCONNECTING;
}
if(!shutdown_disconnect) {
client->disconnect_from_virtual_server("");
}
{
std::lock_guard network_lock{client->network_mutex};
if(client->event_write) {
event_add(client->event_write, nullptr);
}
}
}
void QueryServer::execute_query_connection_close(const std::shared_ptr<QueryClient> &client, bool warn_unknown_client) {
{
std::lock_guard state_lock{client->state_lock};
if(client->state == ConnectionState::DISCONNECTED) {
/* client has already been disconnected */
return;
}
client->state = ConnectionState::DISCONNECTED;
}
client->disconnect_from_virtual_server("");
client->execute_final_disconnect();
{
std::lock_guard client_lock{this->connected_clients_mutex};
auto index = std::find(this->connected_clients.begin(), this->connected_clients.end(), client);
if(index == this->connected_clients.end()) {
if(warn_unknown_client) {
logWarning(LOG_QUERY, "Closed the connection of an unknown/unregistered query.");
}
return;
}
this->connected_clients.erase(index);
this->connected_client_disconnected_notify.notify_all();
}
}

47
server/src/server/QueryServer.h
View File

@ -78,39 +78,50 @@ namespace ts {
bool rename_query_account(const std::shared_ptr<QueryAccount>& /* account */, const std::string& /* new name */);
bool change_query_password(const std::shared_ptr<QueryAccount>& /* account */, const std::string& /* new password */);
threads::ThreadPool* executePool() { return this->_executePool; }
private:
sql::SqlManager* sql;
bool active = false;
bool active{false};
std::deque<std::shared_ptr<Binding>> bindings;
std::vector<threads::Thread*> threads;
std::mutex server_reserve_fd_lock;
int server_reserve_fd = -1; /* -1 = unset | 0 = in use | > 0 ready to use */
std::mutex server_reserve_fd_lock{};
int server_reserve_fd{-1}; /* -1 = unset | 0 = in use | > 0 ready to use */
std::unique_ptr<IpListManager> ip_whitelist;
std::unique_ptr<IpListManager> ip_blacklist;
//IO stuff
event_base* eventLoop = nullptr;
event_base* event_io_loop{nullptr};
std::thread event_io_thread{};
std::chrono::system_clock::time_point accept_event_deleted;
threads::ThreadPool* _executePool = nullptr;
std::mutex connected_clients_mutex{};
std::deque<std::shared_ptr<QueryClient>> connected_clients{};
std::condition_variable connected_client_disconnected_notify{};
std::mutex connected_clients_lock;
std::deque<std::shared_ptr<QueryClient>> connectedClients;
std::mutex client_connect_mutex{};
std::chrono::system_clock::time_point client_connect_last_decrease{};
std::map<std::string, uint32_t> client_connect_count{};
std::map<std::string, std::chrono::system_clock::time_point> client_connect_bans{};
threads::Mutex loginLock;
std::chrono::system_clock::time_point lastDecrease;
std::map<std::string, uint32_t> loginAttempts;
std::map<std::string, std::chrono::system_clock::time_point> queryBann;
bool tick_active{false};
std::mutex tick_mutex{};
std::thread tick_thread{};
std::condition_variable tick_notify{};
std::deque<std::weak_ptr<QueryClient>> tick_pending_disconnects{};
std::deque<std::weak_ptr<QueryClient>> tick_pending_connection_close{};
std::chrono::system_clock::time_point tick_next_client_timestamp{};
threads::Thread* ioThread = nullptr;
threads::SchedulingTask tickingId = nullptr;
void on_client_receive(int fd, short ev, void *arg);
void tick();
void on_client_receive(int server_file_descriptor, short ev, void *arg);
void tick_clients();
void tick_executor();
void enqueue_query_disconnect(const std::shared_ptr<QueryClient>& /* client */);
void execute_query_disconnect(const std::shared_ptr<QueryClient>& /* client */, bool /* shutdown disconnect */);
void enqueue_query_connection_close(const std::shared_ptr<QueryClient>& /* client */);
void execute_query_connection_close(const std::shared_ptr<QueryClient>& /* client */, bool /* warn on unknown client */);
};
}
}

2
server/src/server/VoiceIOManager.cpp
View File

@ -113,7 +113,7 @@ void VoiceIOManager::shutdownGlobally() {
//TODO also reduce thread pool!
void VoiceIOManager::adjustExecutors(size_t size) {
lock_guard<mutex> l(this->executorLock);
size_t targetThreads = size * config::threads::voice::execute_per_server;
size_t targetThreads = size * config::threads::voice::events_per_server;
if(targetThreads > config::threads::voice::io_limit)
targetThreads = config::threads::voice::io_limit;
if(targetThreads < config::threads::voice::io_min)

15
server/src/server/VoiceServer.cpp
View File

@ -120,17 +120,6 @@ void VoiceServer::triggerWrite(const std::shared_ptr<VoiceClient>& client) {
}
}
void VoiceServer::schedule_command_handling(const ts::server::VoiceClient *client) {
auto vmanager = serverInstance->getVoiceServerManager();
if(!vmanager)
return;
auto evloop = vmanager->get_executor_loop();
if(!evloop)
return;
evloop->schedule(client->event_handle_packet);
}
void VoiceServer::tickHandshakingClients() {
this->pow_handler->execute_tick();
@ -141,7 +130,7 @@ void VoiceServer::tickHandshakingClients() {
}
for(const auto& client : connections)
if(client->state == ConnectionState::INIT_HIGH || client->state == ConnectionState::INIT_LOW)
client->tick(system_clock::now());
client->tick_server(system_clock::now());
}
void VoiceServer::execute_resend(const std::chrono::system_clock::time_point &now, std::chrono::system_clock::time_point &next) {
@ -317,7 +306,7 @@ void VoiceServer::handleMessageRead(int fd, short events, void *_event_handle) {
auto new_address = net::to_string(remote_address);
auto command = "dummy_ipchange old_ip=" + old_address + " new_ip=" + new_address;
client->server_command_executor().force_insert_command(pipes::buffer_view{command.data(), command.length()});
client->server_command_queue()->enqueue_command_string(command);
memcpy(&client->remote_address, &remote_address, sizeof(remote_address));
udp::DatagramPacket::extract_info(message, client->connection->remote_address_info_);
}

1
server/src/server/VoiceServer.h
View File

@ -73,7 +73,6 @@ namespace ts {
std::deque<std::shared_ptr<VoiceClient>> activeConnections;
public:
void triggerWrite(const std::shared_ptr<VoiceClient> &);
void schedule_command_handling(VoiceClient const *client);
void tickHandshakingClients();
void execute_resend(const std::chrono::system_clock::time_point& /* now */, std::chrono::system_clock::time_point& /* next resend */);

1
server/src/snapshots/deploy.cpp
View File

@ -73,7 +73,6 @@ VirtualServerManager::SnapshotDeployResult VirtualServerManager::deploy_snapshot
threads::MutexLock l(this->instanceLock);
this->instances.push_back(server);
}
this->adjust_executor_threads();
if(!server->start(error)) {
logWarning(server->getServerId(), "Failed to auto start server after snapshot deployment: {}", error);

42
server/src/weblist/WebListManager.cpp
View File

@ -1,5 +1,6 @@
#include <algorithm>
#include <thread>
#include <ThreadPool/ThreadHelper.h>
#include "src/VirtualServer.h"
#include "log/LogUtils.h"
#include "TeamSpeakWebClient.h"
@ -13,33 +14,29 @@ using namespace ts::weblist;
WebListManager::WebListManager() {
this->event_base = event_base_new();
this->event_base_dispatch = std::thread([&](){
system_clock::time_point start;
this->event_base_dispatch = std::thread([&]{
while(this->event_base) {
::event_base_dispatch(this->event_base);
if(event_base_got_break(this->event_base)) return;
::event_base_loop(this->event_base, EVLOOP_NO_EXIT_ON_EMPTY);
{
std::unique_lock entry_lock{this->entry_lock};
this->entry_cv.wait(entry_lock);
if(!this->event_base) return;
if(this->event_base) {
logWarning(LOG_GENERAL, "WebList report event loop exited without terminating. Rescheduling....");
std::this_thread::sleep_for(std::chrono::seconds{1});
} else {
break;
}
}
});
}
WebListManager::~WebListManager() {
auto base = this->event_base;
{
std::unique_lock entry_lock{this->entry_lock};
this->event_base = nullptr;
this->entry_cv.notify_all();
entry_lock.unlock();
auto event_base_ = std::exchange(this->event_base, nullptr);
if(event_base_) {
event_base_loopbreak(event_base_);
}
event_base_loopbreak(base);
this->event_base_dispatch.join();
event_base_free(base);
threads::save_join(this->event_base_dispatch, true);
if(event_base_) {
event_base_free(event_base_);
}
}
void WebListManager::enable_report(const std::shared_ptr<ts::server::VirtualServer> &server) {
@ -55,8 +52,6 @@ void WebListManager::enable_report(const std::shared_ptr<ts::server::VirtualServ
entry->scheduled_request = system_clock::now();
entry->fail_count = 0;
this->entries.push_back(entry);
this->entry_cv.notify_all();
}
}
@ -82,7 +77,6 @@ void WebListManager::disable_report(const std::shared_ptr<ts::server::VirtualSer
if(it != this->entries.end())
this->entries.erase(it);
if(copied_entry->current_request) {
this->entry_cv.notify_all();
lock.unlock();
copied_entry->current_request->abort_sync();
}
@ -142,12 +136,6 @@ void WebListManager::tick() {
auto ref_request = entry->current_request;
ref_request->report();
}
{
/* lets run the event loop */
unique_lock lock(this->entry_lock);
this->entry_cv.notify_all();
}
}
}
}

1
server/src/weblist/WebListManager.h
View File

@ -43,7 +43,6 @@ namespace ts {
std::thread event_base_dispatch{};
std::mutex entry_lock{};
std::condition_variable entry_cv{};
std::deque<std::shared_ptr<Entry>> entries{};
};
}

2
shared

@ -1 +1 @@
Subproject commit eb77a7fefb454469ed4f0b495959923048a2b768
Subproject commit 0cd49a6a2b88c171f8fcf73dc0cab9da44338b74