From: Maksym Mamontov Date: Tue, 26 Jul 2022 17:13:50 +0000 (+0300) Subject: Some rscriptd refactoring. X-Git-Url: https://git.stg.codes/stg.git/commitdiff_plain/fca49a45944d877965930a623fff27a84257782c Some rscriptd refactoring. --- diff --git a/projects/rscriptd/listener.cpp b/projects/rscriptd/listener.cpp index 3d8fd48e..b168e5e2 100644 --- a/projects/rscriptd/listener.cpp +++ b/projects/rscriptd/listener.cpp @@ -19,41 +19,40 @@ * Author : Maxim Mamontov */ -#include -#include // readv -#include // for historical versions of BSD -#include -#include -#include +#include "listener.h" +#include "stg/scriptexecuter.h" +#include "stg/locker.h" +#include "stg/common.h" +#include "stg/const.h" + +#include +#include +#include #include #include #include #include -#include -#include -#include "stg/scriptexecuter.h" -#include "stg/locker.h" -#include "stg/common.h" -#include "stg/const.h" -#include "listener.h" +#include +#include // readv +#include // for historical versions of BSD +#include +#include +#include void InitEncrypt(BLOWFISH_CTX * ctx, const std::string & password); -void Decrypt(BLOWFISH_CTX * ctx, char * dst, const char * src, int len8); //----------------------------------------------------------------------------- LISTENER::LISTENER() : WriteServLog(STG::Logger::get()), port(0), - running(false), receiverStopped(true), processorStopped(true), userTimeout(0), listenSocket(0), version("rscriptd listener v.1.2") { -pthread_mutex_init(&mutex, NULL); } //----------------------------------------------------------------------------- void LISTENER::SetPassword(const std::string & p) @@ -66,30 +65,17 @@ InitEncrypt(&ctxS, password); bool LISTENER::Start() { printfd(__FILE__, "LISTENER::Start()\n"); -running = true; if (PrepareNet()) { return true; } -if (receiverStopped) - { - if (pthread_create(&receiverThread, NULL, Run, this)) - { - errorStr = "Cannot create thread."; - return true; - } - } +if (!m_receiverThread.joinable()) + m_receiverThread = std::jthread([this](auto token){ Run(std::move(token)); }); -if (processorStopped) - { - if (pthread_create(&processorThread, NULL, RunProcessor, this)) - { - errorStr = "Cannot create thread."; - return true; - } - } +if (!m_processorThread.joinable()) + m_processorThread = std::jthread([this](auto token){ RunProcessor(std::move(token)); }); errorStr = ""; @@ -98,31 +84,24 @@ return false; //----------------------------------------------------------------------------- bool LISTENER::Stop() { -running = false; +m_receiverThread.request_stop(); +m_processorThread.request_stop(); printfd(__FILE__, "LISTENER::Stop()\n"); -struct timespec ts = {0, 500000000}; -nanosleep(&ts, NULL); +std::this_thread::sleep_for(std::chrono::milliseconds(500)); if (!processorStopped) { //5 seconds to thread stops itself for (int i = 0; i < 25 && !processorStopped; i++) - { - struct timespec ts = {0, 200000000}; - nanosleep(&ts, NULL); - } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); //after 5 seconds waiting thread still running. now killing it if (!processorStopped) { //TODO pthread_cancel() - if (pthread_kill(processorThread, SIGINT)) - { - errorStr = "Cannot kill thread."; - return true; - } + m_processorThread.detach(); printfd(__FILE__, "LISTENER killed Timeouter\n"); } } @@ -131,88 +110,53 @@ if (!receiverStopped) { //5 seconds to thread stops itself for (int i = 0; i < 25 && !receiverStopped; i++) - { - struct timespec ts = {0, 200000000}; - nanosleep(&ts, NULL); - } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); //after 5 seconds waiting thread still running. now killing it if (!receiverStopped) { //TODO pthread_cancel() - if (pthread_kill(receiverThread, SIGINT)) - { - errorStr = "Cannot kill thread."; - return true; - } + m_receiverThread.detach(); printfd(__FILE__, "LISTENER killed Run\n"); } } -pthread_join(receiverThread, NULL); -pthread_join(processorThread, NULL); - -pthread_mutex_destroy(&mutex); +if (receiverStopped) + m_receiverThread.join(); +if (processorStopped) + m_processorThread.join(); FinalizeNet(); -std::for_each(users.begin(), users.end(), DisconnectUser(*this)); +for (const auto& user : users) + Disconnect(user); printfd(__FILE__, "LISTENER::Stoped successfully.\n"); return false; } //----------------------------------------------------------------------------- -void * LISTENER::Run(void * d) -{ -sigset_t signalSet; -sigfillset(&signalSet); -pthread_sigmask(SIG_BLOCK, &signalSet, NULL); - -LISTENER * listener = static_cast(d); - -listener->Runner(); - -return NULL; -} -//----------------------------------------------------------------------------- -void LISTENER::Runner() +void LISTENER::Run(std::stop_token token) { receiverStopped = false; -while (running) - { - RecvPacket(); - } +while (!token.stop_requested()) + RecvPacket(token); receiverStopped = true; } //----------------------------------------------------------------------------- -void * LISTENER::RunProcessor(void * d) -{ -sigset_t signalSet; -sigfillset(&signalSet); -pthread_sigmask(SIG_BLOCK, &signalSet, NULL); - -LISTENER * listener = static_cast(d); - -listener->ProcessorRunner(); - -return NULL; -} -//----------------------------------------------------------------------------- -void LISTENER::ProcessorRunner() +void LISTENER::RunProcessor(std::stop_token token) { processorStopped = false; -while (running) - { - struct timespec ts = {0, 500000000}; - nanosleep(&ts, NULL); +while (!token.stop_requested()) +{ + std::this_thread::sleep_for(std::chrono::milliseconds(500)); if (!pending.empty()) ProcessPending(); ProcessTimeouts(); - } +} processorStopped = true; } @@ -232,7 +176,7 @@ listenAddr.sin_family = AF_INET; listenAddr.sin_port = htons(port); listenAddr.sin_addr.s_addr = inet_addr("0.0.0.0"); -if (bind(listenSocket, (struct sockaddr*)&listenAddr, sizeof(listenAddr)) < 0) +if (bind(listenSocket, reinterpret_cast(&listenAddr), sizeof(listenAddr)) < 0) { errorStr = "LISTENER: Bind failed."; return true; @@ -250,7 +194,7 @@ close(listenSocket); return false; } //----------------------------------------------------------------------------- -bool LISTENER::RecvPacket() +bool LISTENER::RecvPacket(const std::stop_token& token) { struct iovec iov[2]; @@ -267,7 +211,7 @@ while (dataLen < sizeof(buffer)) { if (!WaitPackets(listenSocket)) { - if (!running) + if (token.stop_requested()) return false; continue; } @@ -285,35 +229,37 @@ if (CheckHeader(packetHead)) return true; } -std::string userLogin((char *)packetHead.login); -PendingData data; -data.login = userLogin; -data.ip = ntohl(packetHead.ip); -data.id = ntohl(packetHead.id); +std::string userLogin(reinterpret_cast(packetHead.login)); +PendingData pd; +pd.data.login = userLogin; +pd.data.ip = ntohl(packetHead.ip); +pd.data.id = ntohl(packetHead.id); if (packetHead.packetType == RS_ALIVE_PACKET) { - data.type = PendingData::ALIVE; + pd.type = PendingData::ALIVE; } else if (packetHead.packetType == RS_CONNECT_PACKET) { - data.type = PendingData::CONNECT; - if (GetParams(buffer, data)) + pd.type = PendingData::CONNECT; + if (GetParams(buffer, pd.data)) { return true; } } else if (packetHead.packetType == RS_DISCONNECT_PACKET) { - data.type = PendingData::DISCONNECT; - if (GetParams(buffer, data)) + pd.type = PendingData::DISCONNECT; + if (GetParams(buffer, pd.data)) { return true; } } +else + return true; -STG_LOCKER lock(&mutex); -pending.push_back(data); +std::lock_guard lock(m_mutex); +pending.push_back(pd); return false; } @@ -322,9 +268,9 @@ bool LISTENER::GetParams(char * buffer, UserData & data) { RS::PACKET_TAIL packetTail; -Decrypt(&ctxS, (char *)&packetTail, buffer, sizeof(packetTail) / 8); +DecryptString(&packetTail, buffer, sizeof(packetTail), &ctxS); -if (strncmp((char *)packetTail.magic, RS_ID, RS_MAGIC_LEN)) +if (strncmp(reinterpret_cast(packetTail.magic), RS_ID, RS_MAGIC_LEN)) { printfd(__FILE__, "Invalid crypto magic\n"); return true; @@ -334,7 +280,7 @@ std::ostringstream params; params << "\"" << data.login << "\" " << inet_ntostring(data.ip) << " " << data.id << " " - << (char *)packetTail.params; + << reinterpret_cast(packetTail.params); data.params = params.str(); @@ -343,67 +289,58 @@ return false; //----------------------------------------------------------------------------- void LISTENER::ProcessPending() { -std::list::iterator it(pending.begin()); +auto it = pending.begin(); size_t count = 0; printfd(__FILE__, "Pending: %d\n", pending.size()); while (it != pending.end() && count < 256) { - std::vector::iterator uit( - std::lower_bound( - users.begin(), - users.end(), - it->login) - ); + auto uit = std::lower_bound(users.begin(), users.end(), it->data.login); if (it->type == PendingData::CONNECT) { printfd(__FILE__, "Connect packet\n"); - if (uit == users.end() || uit->login != it->login) + if (uit == users.end() || uit->data.login != it->data.login) { - printfd(__FILE__, "Connect new user '%s'\n", it->login.c_str()); + printfd(__FILE__, "Connect new user '%s'\n", it->data.login.c_str()); // Add new user Connect(*it); - users.insert(uit, AliveData(static_cast(*it))); + users.insert(uit, AliveData(it->data)); } - else if (uit->login == it->login) + else { - printfd(__FILE__, "Update existing user '%s'\n", it->login.c_str()); + printfd(__FILE__, "Update existing user '%s'\n", it->data.login.c_str()); // Update already existing user time(&uit->lastAlive); - uit->params = it->params; - } - else - { - printfd(__FILE__, "Hmmm... Strange connect for '%s'\n", it->login.c_str()); + uit->data.params = it->data.params; } } else if (it->type == PendingData::ALIVE) { printfd(__FILE__, "Alive packet\n"); - if (uit != users.end() && uit->login == it->login) + if (uit != users.end() && uit->data.login == it->data.login) { - printfd(__FILE__, "Alive user '%s'\n", it->login.c_str()); + printfd(__FILE__, "Alive user '%s'\n", it->data.login.c_str()); // Update existing user time(&uit->lastAlive); } else { - printfd(__FILE__, "Alive user '%s' is not found\n", it->login.c_str()); + printfd(__FILE__, "Alive user '%s' is not found\n", it->data.login.c_str()); } } else if (it->type == PendingData::DISCONNECT) { printfd(__FILE__, "Disconnect packet\n"); - if (uit != users.end() && uit->login == it->login.c_str()) + if (uit != users.end() && uit->data.login == it->data.login.c_str()) { - printfd(__FILE__, "Disconnect user '%s'\n", it->login.c_str()); + printfd(__FILE__, "Disconnect user '%s'\n", it->data.login.c_str()); // Disconnect existing user - uit->params = it->params; + uit->data.params = it->data.params; Disconnect(*uit); users.erase(uit); } else { - printfd(__FILE__, "Cannot find user '%s' for disconnect\n", it->login.c_str()); + printfd(__FILE__, "Cannot find user '%s' for disconnect\n", it->data.login.c_str()); } } else @@ -413,40 +350,31 @@ while (it != pending.end() && count < 256) ++it; ++count; } -STG_LOCKER lock(&mutex); +std::lock_guard lock(m_mutex); pending.erase(pending.begin(), it); } //----------------------------------------------------------------------------- void LISTENER::ProcessTimeouts() { -const std::vector::iterator it( - std::stable_partition( - users.begin(), - users.end(), - IsNotTimedOut(userTimeout) - ) - ); +const auto now = time(nullptr); +const auto it = std::stable_partition(users.begin(), users.end(), [this, now](const auto& data){ return difftime(now, data.lastAlive) < userTimeout; }); if (it != users.end()) { printfd(__FILE__, "Total users: %d, users to disconnect: %d\n", users.size(), std::distance(it, users.end())); - std::for_each( - it, - users.end(), - DisconnectUser(*this) - ); + std::for_each(it, users.end(), [this](const auto& user){ Disconnect(user);}); users.erase(it, users.end()); } } //----------------------------------------------------------------------------- -bool LISTENER::Connect(const UserData & data) const +bool LISTENER::Connect(const PendingData & pd) const { -printfd(__FILE__, "Connect %s\n", data.login.c_str()); +printfd(__FILE__, "Connect %s\n", pd.data.login.c_str()); if (access(scriptOnConnect.c_str(), X_OK) == 0) { - if (ScriptExec((scriptOnConnect + " " + data.params).c_str())) + if (ScriptExec((scriptOnConnect + " " + pd.data.params).c_str())) { WriteServLog("Script %s cannot be executed for an unknown reason.", scriptOnConnect.c_str()); return true; @@ -460,12 +388,12 @@ else return false; } //----------------------------------------------------------------------------- -bool LISTENER::Disconnect(const UserData & data) const +bool LISTENER::Disconnect(const AliveData& ad) const { -printfd(__FILE__, "Disconnect %s\n", data.login.c_str()); +printfd(__FILE__, "Disconnect %s\n", ad.data.login.c_str()); if (access(scriptOnDisconnect.c_str(), X_OK) == 0) { - if (ScriptExec((scriptOnDisconnect + " " + data.params).c_str())) + if (ScriptExec((scriptOnDisconnect + " " + ad.data.params).c_str())) { WriteServLog("Script %s cannot be executed for an unknown reson.", scriptOnDisconnect.c_str()); return true; @@ -481,33 +409,18 @@ return false; //----------------------------------------------------------------------------- bool LISTENER::CheckHeader(const RS::PACKET_HEADER & header) const { -if (strncmp((char *)header.magic, RS_ID, RS_MAGIC_LEN)) - { +if (strncmp(reinterpret_cast(header.magic), RS_ID, RS_MAGIC_LEN)) return true; - } -if (strncmp((char *)header.protoVer, "02", RS_PROTO_VER_LEN)) - { +if (strncmp(reinterpret_cast(header.protoVer), "02", RS_PROTO_VER_LEN)) return true; - } return false; } //----------------------------------------------------------------------------- -inline void InitEncrypt(BLOWFISH_CTX * ctx, const std::string & password) { -unsigned char keyL[PASSWD_LEN]; +char keyL[PASSWD_LEN]; memset(keyL, 0, PASSWD_LEN); -strncpy((char *)keyL, password.c_str(), PASSWD_LEN); +strncpy(keyL, password.c_str(), PASSWD_LEN); Blowfish_Init(ctx, keyL, PASSWD_LEN); } //----------------------------------------------------------------------------- -inline -void Decrypt(BLOWFISH_CTX * ctx, char * dst, const char * src, int len8) -{ -if (dst != src) - memcpy(dst, src, len8 * 8); - -for (int i = 0; i < len8; i++) - Blowfish_Decrypt(ctx, (uint32_t *)(dst + i * 8), (uint32_t *)(dst + i * 8 + 4)); -} -//----------------------------------------------------------------------------- diff --git a/projects/rscriptd/listener.h b/projects/rscriptd/listener.h index 7fb71ee3..4d36b87a 100644 --- a/projects/rscriptd/listener.h +++ b/projects/rscriptd/listener.h @@ -19,18 +19,21 @@ * Author : Maxim Mamontov */ -#include +#include "stg/blowfish.h" +#include "stg/rs_packets.h" +#include "stg/logger.h" #include #include #include #include +#include +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wshadow" +#include +#pragma GCC diagnostic pop #include -#include "stg/blowfish.h" -#include "stg/rs_packets.h" -#include "stg/logger.h" - struct UserData { std::string params; @@ -39,31 +42,21 @@ struct UserData uint32_t id; }; -struct PendingData : public UserData +struct PendingData { + UserData data; enum {CONNECT, ALIVE, DISCONNECT} type; }; -struct AliveData : public UserData +struct AliveData { - explicit AliveData(const UserData & data) - : UserData(data), + explicit AliveData(const UserData& ud) + : data(ud), lastAlive(time(NULL)) {}; - bool operator<(const std::string & rvalue) const { return login < rvalue; }; - time_t lastAlive; -}; - -class IsNotTimedOut : public std::unary_function { - public: - explicit IsNotTimedOut(double to) : timeout(to), now(time(NULL)) {} - bool operator()(const AliveData & data) const - { - return difftime(now, data.lastAlive) < timeout; - } - private: - double timeout; - time_t now; + bool operator<(const std::string& rhs) const { return data.login < rhs; }; + UserData data; + time_t lastAlive; }; class LISTENER @@ -87,22 +80,20 @@ public: private: // Threading stuff - static void * Run(void * self); - static void * RunProcessor(void * self); - void Runner(); - void ProcessorRunner(); + void Run(std::stop_token token); + void RunProcessor(std::stop_token token); // Networking stuff bool PrepareNet(); bool FinalizeNet(); - bool RecvPacket(); + bool RecvPacket(const std::stop_token& token); // Parsing stuff bool CheckHeader(const RS::PACKET_HEADER & header) const; bool GetParams(char * buffer, UserData & data); // Processing stuff void ProcessPending(); void ProcessTimeouts(); - bool Disconnect(const UserData & data) const; - bool Connect(const UserData & data) const; + bool Disconnect(const AliveData& data) const; + bool Connect(const PendingData& data) const; BLOWFISH_CTX ctxS; STG::Logger& WriteServLog; @@ -113,32 +104,17 @@ private: std::string password; uint16_t port; - bool running; bool receiverStopped; bool processorStopped; std::vector users; - std::list pending; + std::vector pending; int userTimeout; - pthread_t receiverThread; - pthread_t processorThread; - pthread_mutex_t mutex; + std::jthread m_receiverThread; + std::jthread m_processorThread; + std::mutex m_mutex; int listenSocket; std::string version; - - friend class DisconnectUser; -}; - -class DisconnectUser : public std::unary_function { - public: - explicit DisconnectUser(LISTENER & l) : listener(l) {}; - void operator()(const UserData & data) - { - listener.Disconnect(data); - }; - private: - LISTENER & listener; }; -//----------------------------------------------------------------------------- diff --git a/projects/rscriptd/main.cpp b/projects/rscriptd/main.cpp index e7c54eec..1baca9e8 100644 --- a/projects/rscriptd/main.cpp +++ b/projects/rscriptd/main.cpp @@ -176,11 +176,7 @@ else KillExecuters(); } //----------------------------------------------------------------------------- -#ifdef NO_DAEMON -int ForkAndWait(const std::string &) -#else -int ForkAndWait(const std::string & confDir) -#endif +int ForkAndWait() { #ifndef NO_DAEMON pid_t childPid = fork(); @@ -252,7 +248,7 @@ cfg->ReadInt("UserTimeout", &userTimeout, 60); cfg->ReadString("ScriptOnConnect", &onConnect, "/etc/rscriptd/OnConnect"); cfg->ReadString("ScriptOnDisconnect", &onDisconnect, "/etc/rscriptd/OnDisconnect"); -if (ForkAndWait(confDir) < 0) +if (ForkAndWait() < 0) { auto & WriteServLog = STG::Logger::get(); WriteServLog("Fork error!");