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

Using one global event loop for the query and web client

Browse Source
This commit is contained in:
WolverinDEV
2021-04-15 12:54:52 +02:00
parent 6e2e005ed7
commit abeeae4ed5
30 changed files with 529 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/client/SpeakingClient.cpp
+2 -1
View File
@@ -119,7 +119,8 @@ command_result SpeakingClient::applyClientInitParameters(Command &cmd) {
this->properties()[property::CLIENT_NICKNAME_PHONETIC] = name;
} else if(key == "client_version" || key == "client_platform") {
auto value = cmd[key].string();
if(value.length() > 64) {
if(value.length() > 512) {
/* The web client uses the full browser string which might be a bit longer */
return command_result{error::client_hacked};
}
server/src/client/command_handler/misc.cpp
+4 -4
View File
@@ -3353,10 +3353,10 @@ enum struct FeatureSupportMode {
DEPRECATED
};
#define REGISTER_FEATURE(name, support, version) \
notify.put_unchecked(index, "name", name); \
notify.put_unchecked(index, "support", (int) support); \
notify.put_unchecked(index, "version", version); \
#define REGISTER_FEATURE(name, support, version) \
notify.put_unchecked(index, "name", name); \
notify.put_unchecked(index, "support", (int) support); \
notify.put_unchecked(index, "version", version); \
index++
command_result ConnectedClient::handleCommandListFeatureSupport(ts::Command &cmd) {
server/src/client/query/QueryClient.cpp
+3 -2
View File
@@ -7,6 +7,7 @@
#include <misc/std_unique_ptr.h>
#include <log/LogUtils.h>
#include "../../groups/GroupAssignmentManager.h"
#include "../../server/GlobalNetworkEvents.h"
using namespace std;
using namespace std::chrono;
@@ -66,8 +67,8 @@ void QueryClient::initialize_weak_reference(const std::shared_ptr<ConnectedClien
std::make_unique<QueryClientCommandHandler>(dynamic_pointer_cast<QueryClient>(self))
);
this->event_read = event_new(this->handle->event_io_loop, this->client_file_descriptor, EV_READ | EV_PERSIST, QueryClient::handle_event_read, this);
this->event_write = event_new(this->handle->event_io_loop, this->client_file_descriptor, EV_WRITE, QueryClient::handle_event_write, this);
this->event_read = serverInstance->network_event_loop()->allocate_event(this->client_file_descriptor, EV_READ | EV_PERSIST, QueryClient::handle_event_read, this, nullptr);
this->event_write = serverInstance->network_event_loop()->allocate_event(this->client_file_descriptor, EV_WRITE, QueryClient::handle_event_write, this, nullptr);
}
QueryClient::~QueryClient() {
server/src/client/voice/VoiceClientCommandHandler.cpp
+4 -2
View File
@@ -72,10 +72,12 @@ command_result VoiceClient::handleCommand(ts::Command &command) {
if(!this->voice_server) return command_result{error::server_unbound};
if(this->state == ConnectionState::INIT_HIGH && this->handshake.state == HandshakeState::SUCCEEDED) {
if(command.command() == "clientinit")
if(command.command() == "clientinit") {
return this->handleCommandClientInit(command);
} else if(command.command() == "clientdisconnect")
}
} else if(command.command() == "clientdisconnect") {
return this->handleCommandClientDisconnect(command);
}
return SpeakingClient::handleCommand(command);
}
server/src/client/voice/VoiceClientConnection.cpp
+4 -11
View File
@@ -1,4 +1,3 @@
#include <algorithm>
#include <log/LogUtils.h>
#include <misc/memtracker.h>
#include <protocol/Packet.h>
@@ -6,7 +5,6 @@
#include "../../server/VoiceServer.h"
#include "./VoiceClientConnection.h"
#include "./VoiceClient.h"
//#define LOG_AUTO_ACK_AUTORESPONSE
@@ -76,16 +74,11 @@ void VoiceClientConnection::triggerWrite() {
}
}
void VoiceClientConnection::handle_incoming_datagram(const pipes::buffer_view& buffer) {
ClientPacketParser packet_parser{buffer};
if(!packet_parser.valid()) {
return;
}
void VoiceClientConnection::handle_incoming_datagram(protocol::ClientPacketParser& packet_parser) {
#ifndef CONNECTION_NO_STATISTICS
if(this->current_client) {
auto stats = this->current_client->connectionStatistics;
stats->logIncomingPacket(stats::ConnectionStatistics::category::from_type(packet_parser.type()), buffer.length() + 96); /* 96 for the UDP packet overhead */
stats->logIncomingPacket(stats::ConnectionStatistics::category::from_type(packet_parser.type()), packet_parser.buffer().length() + 96); /* 96 for the UDP packet overhead */
}
this->packet_statistics().received_packet((protocol::PacketType) packet_parser.type(), packet_parser.full_packet_id());
#endif
@@ -197,8 +190,8 @@ void VoiceClientConnection::callback_command_decoded(void *ptr_this, Reassembled
connection->handlePacketCommand(std::exchange(command, nullptr));
}
bool VoiceClientConnection::verify_encryption(const pipes::buffer_view &buffer /* incl. mac etc */) {
return this->packet_decoder_.verify_encryption_client_packet(buffer);
bool VoiceClientConnection::verify_encryption(const protocol::ClientPacketParser& packet) {
return this->packet_decoder_.verify_encryption_client_packet(packet);
}
std::shared_ptr<ts::server::VoiceClient> VoiceClientConnection::getCurrentClient() {
server/src/client/voice/VoiceClientConnection.h
+2 -2
View File
@@ -81,8 +81,8 @@ namespace ts {
[[nodiscard]] inline auto& ping_handler() { return this->ping_handler_; }
[[nodiscard]] inline auto& crypt_setup_handler() { return this->crypt_setup_handler_; }
protected:
void handle_incoming_datagram(const pipes::buffer_view &buffer);
bool verify_encryption(const pipes::buffer_view& /* full packet */);
void handle_incoming_datagram(protocol::ClientPacketParser& /* packet */);
bool verify_encryption(const protocol::ClientPacketParser& /* packet */);
void triggerWrite();
private:
server/src/client/web/WSWebClient.cpp
+31 -29
View File
@@ -10,59 +10,61 @@ using namespace ts;
using namespace ts::server;
using namespace ts::protocol;
void WebClient::handleMessageWrite(int fd, short, void *) {
auto self_lock = this->ref();
void WebClient::handleMessageWrite(int fd, short, void *ptr_client) {
auto client = dynamic_pointer_cast<WebClient>(((WebClient*) ptr_client)->ref());
assert(client);
unique_lock buffer_lock(this->queue_mutex);
if(this->queue_write.empty()) return;
unique_lock buffer_lock(client->queue_mutex);
if(client->queue_write.empty()) return;
auto buffer = this->queue_write[0];
this->queue_write.pop_front();
auto buffer = client->queue_write[0];
client->queue_write.pop_front();
auto written = send(fd, buffer.data_ptr(), buffer.length(), MSG_NOSIGNAL | MSG_DONTWAIT);
if(written == -1) {
buffer_lock.unlock();
if (errno == EINTR || errno == EAGAIN) {
lock_guard event_lock(this->event_mutex);
if(this->writeEvent)
event_add(this->writeEvent, nullptr);
lock_guard event_lock(client->event_mutex);
if(client->writeEvent)
event_add(client->writeEvent, nullptr);
return;
} else {
//new ServerConnection(globalClient).startConnection({ host: "localhost", port: 9987}, new HandshakeHandler(profiles.default_profile(), "test"))
{
std::lock_guard event_lock{this->event_mutex};
if(this->writeEvent) {
event_del_noblock(this->writeEvent);
event_free(this->writeEvent);
this->writeEvent = nullptr;
std::lock_guard event_lock{client->event_mutex};
if(client->writeEvent) {
event_del_noblock(client->writeEvent);
event_free(client->writeEvent);
client->writeEvent = nullptr;
}
}
debugMessage(this->getServerId(), "[{}] Failed to write message (length {}, errno {}, message {}) Disconnecting client.", CLIENT_STR_LOG_PREFIX, written, errno, strerror(errno));
debugMessage(client->getServerId(), "[{}] Failed to write message (length {}, errno {}, message {}) Disconnecting client.", client->getLoggingPrefix(), written, errno, strerror(errno));
}
this->close_connection(system_clock::now()); /* close connection in a new thread */
client->close_connection(system_clock::now()); /* close connection in a new thread */
return;
}
if(written < buffer.length()) {
buffer = buffer.range((size_t) written); /* get the overhead */
this->queue_write.push_front(buffer);
client->queue_write.push_front(buffer);
}
if(this->queue_write.empty())
if(client->queue_write.empty())
return;
/* reschedule new write */
buffer_lock.unlock();
lock_guard event_lock(this->event_mutex);
if(this->writeEvent)
event_add(this->writeEvent, nullptr);
lock_guard event_lock(client->event_mutex);
if(client->writeEvent)
event_add(client->writeEvent, nullptr);
}
void WebClient::handleMessageRead(int fd, short, void *) {
auto self_lock = this->ref();
void WebClient::handleMessageRead(int fd, short, void *ptr_client) {
auto client = dynamic_pointer_cast<WebClient>(((WebClient*) ptr_client)->ref());
assert(client);
size_t buffer_length = 1024 * 4;
uint8_t buffer[buffer_length];
@@ -72,14 +74,14 @@ void WebClient::handleMessageRead(int fd, short, void *) {
if(errno == EINTR || errno == EAGAIN)
;
else {
debugMessage(this->getServerId(), "[{}] Failed to read message (length {}, errno {}, message: {}). Closing connection.", CLIENT_STR_LOG_PREFIX, length, errno, strerror(errno));
debugMessage(client->getServerId(), "[{}] Failed to read message (length {}, errno {}, message: {}). Closing connection.", client->getLoggingPrefix(), length, errno, strerror(errno));
{
lock_guard lock(this->event_mutex);
if(this->readEvent)
event_del_noblock(this->readEvent);
lock_guard lock(client->event_mutex);
if(client->readEvent)
event_del_noblock(client->readEvent);
}
self_lock->close_connection(system_clock::now()); /* direct close, but from another thread */
client->close_connection(system_clock::now()); /* direct close, but from another thread */
}
return;
}
@@ -87,7 +89,7 @@ void WebClient::handleMessageRead(int fd, short, void *) {
auto command = command::ReassembledCommand::allocate((size_t) length);
memcpy(command->command(), buffer, (size_t) length);
this->command_queue->enqueue_command_execution(command);
client->command_queue->enqueue_command_execution(command);
}
void WebClient::enqueue_raw_packet(const pipes::buffer_view &msg) {
server/src/client/web/WebClient.cpp
+13 -10
View File
@@ -12,6 +12,7 @@
#include <misc/std_unique_ptr.h>
#include <src/client/SpeakingClient.h>
#include "../../manager/ActionLogger.h"
#include "../../server/GlobalNetworkEvents.h"
#if defined(TCP_CORK) && !defined(TCP_NOPUSH)
#define TCP_NOPUSH TCP_CORK
@@ -43,14 +44,12 @@ void WebClient::initialize() {
int enabled = 1;
int disabled = 0;
setsockopt(this->file_descriptor, SOL_SOCKET, SO_KEEPALIVE, &enabled, sizeof(enabled));
if(setsockopt(this->file_descriptor, IPPROTO_TCP, TCP_NOPUSH, &disabled, sizeof disabled) < 0)
if(setsockopt(this->file_descriptor, IPPROTO_TCP, TCP_NOPUSH, &disabled, sizeof disabled) < 0) {
logError(this->getServerId(), "{} Cant disable nopush! system error: {} => {}", CLIENT_STR_LOG_PREFIX, errno, strerror(errno));
}
auto event_loop = serverInstance->getWebIoLoop()->next_loop();
this->readEvent = event_new(event_loop->loop, this->file_descriptor, EV_READ|EV_PERSIST, [](int a, short b, void* c){ ((WebClient*) c)->handleMessageRead(a, b, c); }, this);
this->writeEvent = event_new(event_loop->loop, this->file_descriptor, EV_WRITE, [](int a, short b, void* c){ ((WebClient*) c)->handleMessageWrite(a, b, c); }, this);
this->readEvent = serverInstance->network_event_loop()->allocate_event(this->file_descriptor, EV_READ | EV_PERSIST, WebClient::handleMessageRead, this, nullptr);
this->writeEvent = serverInstance->network_event_loop()->allocate_event(this->file_descriptor, EV_WRITE, WebClient::handleMessageWrite, this, nullptr);
{
this->ws_handler.direct_process(pipes::PROCESS_DIRECTION_IN, true);
@@ -281,8 +280,9 @@ command_result WebClient::handleCommand(Command &command) {
if(this->connectionState() == ConnectionState::INIT_HIGH && this->handshake.state == HandshakeState::SUCCEEDED){
if(command.command() == "clientinit") {
auto result = this->handleCommandClientInit(command);
if(result.has_error())
if(result.has_error()) {
this->close_connection(system_clock::now() + seconds(1));
}
return result;
}
}
@@ -400,10 +400,11 @@ void WebClient::disconnectFinal() {
lock_guard lock(this->execute_mutex);
}
if(this->flush_thread.get_id() == this_thread::get_id())
if(this->flush_thread.get_id() == this_thread::get_id()) {
this->flush_thread.detach();
else
} else {
assert(!this->flush_thread.joinable()); /* shall be already joined via closeConnection(...)*/
}
{
::event *event_read, *event_write;
@@ -434,6 +435,7 @@ void WebClient::disconnectFinal() {
this->file_descriptor = -1;
}
this->state = ConnectionState::DISCONNECTED;
this->processLeave();
/* We do not finalize here since we might still try to send some data */
@@ -551,8 +553,9 @@ bool WebClient::disconnect(const std::string &reason) {
}
command_result WebClient::handleCommandClientInit(Command &command) {
if(!config::server::clients::teaweb)
if(!config::server::clients::teaweb) {
return command_result{error::client_type_is_not_allowed, config::server::clients::teaweb_not_allowed_message};
}
return SpeakingClient::handleCommandClientInit(command);
}
server/src/client/web/WebClient.h
+2 -2
View File
@@ -83,8 +83,8 @@ namespace ts::server {
private:
void initialize();
void handleMessageRead(int, short, void*);
void handleMessageWrite(int, short, void*);
static void handleMessageRead(int, short, void*);
static void handleMessageWrite(int, short, void*);
void enqueue_raw_packet(const pipes::buffer_view& /* buffer */);
/* TODO: Put the message processing part into the IO loop and not into command processing! */