From cc4873e33866ba86561774a7e55449d59367661e Mon Sep 17 00:00:00 2001 From: fusion32 Date: Fri, 10 Oct 2025 02:17:19 -0300 Subject: switch to blocking poll + fix config, signals, time, non-game login --- config.cfg | 20 +++++-- src/connections.cc | 147 ++++++++++++++++++++++++++++++++++++++----------- src/database_sqlite.cc | 4 +- src/hostcache.cc | 5 +- src/query.cc | 1 + src/querymanager.cc | 43 +++++++-------- src/querymanager.hh | 6 +- 7 files changed, 157 insertions(+), 69 deletions(-) diff --git a/config.cfg b/config.cfg index 5634a09..30ca4cd 100644 --- a/config.cfg +++ b/config.cfg @@ -1,15 +1,23 @@ -# Database Config -DatabaseFile = "tibia.db" -MaxCachedStatements = 100 - # HostCache Config MaxCachedHostNames = 100 HostNameExpireTime = 30m +# Database Config +MaxCachedStatements = 100 +DatabaseFile = "tibia.db" +# TODO(fusion): This is not great. Probably switch to database specific options? +#DatabaseHost = "localhost" +#DatabasePort = 5432 +#DatabaseUser = "postgres" +#DatabasePassword = "" +#DatabaseName = "tibia" +#DatabaseTLS = true + # Connection Config -UpdateRate = 20 QueryManagerPort = 7173 QueryManagerPassword = "a6glaf0c" +QueryWorkerThreads = 1 +QueryBufferSize = 1M +QueryMaxAttempts = 3 MaxConnections = 25 MaxConnectionIdleTime = 5m -MaxConnectionPacketSize = 1M diff --git a/src/connections.cc b/src/connections.cc index e22bc6f..b08b52a 100644 --- a/src/connections.cc +++ b/src/connections.cc @@ -8,6 +8,7 @@ # include # include # include +# include # include # include # include @@ -16,6 +17,7 @@ #endif static int g_Listener = -1; +static int g_UpdateEvent = -1; static TConnection *g_Connections; // Connection Handling @@ -139,7 +141,7 @@ TConnection *AssignConnection(int Socket, uint32 Addr, uint16 Port){ Connection = &g_Connections[ConnectionIndex]; Connection->State = CONNECTION_READING; Connection->Socket = Socket; - Connection->LastActive = g_MonotonicTimeMS; + Connection->LastActive = GetMonotonicUptimeMS(); snprintf(Connection->RemoteAddress, sizeof(Connection->RemoteAddress), "%d.%d.%d.%d:%d", @@ -213,7 +215,7 @@ void CheckConnectionInput(TConnection *Connection, int Events){ if(Connection->RWPosition >= ReadSize){ if(Connection->RWSize != 0){ Connection->State = CONNECTION_REQUEST; - Connection->LastActive = g_MonotonicTimeMS; + Connection->LastActive = GetMonotonicUptimeMS(); Connection->Query->Request = TReadBuffer(Buffer, Connection->RWSize); break; }else if(Connection->RWPosition == 2){ @@ -252,8 +254,9 @@ void ProcessQuery(TConnection *Connection){ void SendQueryResponse(TConnection *Connection){ ASSERT(Connection->Query != NULL); - if(Connection->State != CONNECTION_RESPONSE){ - LOG_ERR("Connection %s is not in a RESPONSE state (State: %d)", + if(Connection->State != CONNECTION_REQUEST + && Connection->State != CONNECTION_RESPONSE){ + LOG_ERR("Connection %s is not in a REQUEST/RESPONSE state (State: %d)", Connection->RemoteAddress, Connection->State); CloseConnection(Connection); return; @@ -453,8 +456,16 @@ void CheckConnectionQueryResponse(TConnection *Connection){ } } - void CheckConnectionOutput(TConnection *Connection, int Events){ + // TODO(fusion): We're only polling `POLLOUT` when the connection is WRITING + // meaning that a writes will be delayed at least one cycle after a response + // is available. This could be solved by adding a `CanWrite` boolean to the + // connection struct that is set to false when `write` returns `EAGAIN`, and + // is used to determine whether we should poll `POLLOUT`. That said, it may + // not even make that big of a difference. + // if(!Connection->CanWrite) { Events |= POLLOUT; } + // if((Events & POLLOUT) != 0) { Connection->CanWrite = true; } + // if(errno == EAGAIN) { Connection->CanWrite = false; } if((Events & POLLOUT) == 0 || Connection->Socket == -1){ return; } @@ -500,7 +511,7 @@ void CheckConnection(TConnection *Connection, int Events){ } if(g_Config.MaxConnectionIdleTime > 0){ - int IdleTime = (g_MonotonicTimeMS - Connection->LastActive); + int IdleTime = (GetMonotonicUptimeMS() - Connection->LastActive); if(IdleTime >= g_Config.MaxConnectionIdleTime){ LOG_WARN("Dropping connection %s due to inactivity", Connection->RemoteAddress); @@ -513,8 +524,37 @@ void CheckConnection(TConnection *Connection, int Events){ } } -void ProcessConnections(void){ - // NOTE(fusion): Accept new connections. +void WakeConnections(void){ + if(g_UpdateEvent != -1){ + uint64 One = 1; + int Written = (int)write(g_UpdateEvent, &One, sizeof(One)); + if(Written != sizeof(One)){ + LOG_ERR("Failed to signal update event: (%d) %s", + errno, strerrordesc_np(errno)); + } + } +} + +static void ConsumeUpdateEvent(int Events){ + ASSERT(g_UpdateEvent != -1); + if((Events & POLLIN) == 0){ + return; + } + + uint64 Dummy; + int Read = (int)read(g_UpdateEvent, &Dummy, sizeof(Dummy)); + if(Read != sizeof(Dummy)){ + LOG_ERR("Failed to consume update event: (%d) %s", + errno, strerrordesc_np(errno)); + } +} + +static void AcceptConnections(int Events){ + ASSERT(g_Listener != -1); + if((Events & POLLIN) == 0){ + return; + } + while(true){ uint32 Addr; uint16 Port; @@ -530,50 +570,88 @@ void ProcessConnections(void){ close(Socket); } } +} + +void ProcessConnections(void){ + int NumFds = 0; + int MaxFds = g_Config.MaxConnections + 2; + pollfd *Fds = (pollfd*)alloca(MaxFds * sizeof(pollfd)); + int *ConnectionIndices = (int*)alloca(MaxFds * sizeof(int)); + + if(g_UpdateEvent != -1){ + Fds[NumFds].fd = g_UpdateEvent; + Fds[NumFds].events = POLLIN; + Fds[NumFds].revents = 0; + ConnectionIndices[NumFds] = -1; + NumFds += 1; + } + + if(g_Listener != -1){ + Fds[NumFds].fd = g_Listener; + Fds[NumFds].events = POLLIN; + Fds[NumFds].revents = 0; + ConnectionIndices[NumFds] = -1; + NumFds += 1; + } - // NOTE(fusion): Gather active connections. - int NumConnections = 0; - int *ConnectionIndices = (int*)alloca(g_Config.MaxConnections * sizeof(int)); - pollfd *ConnectionFds = (pollfd*)alloca(g_Config.MaxConnections * sizeof(pollfd)); for(int i = 0; i < g_Config.MaxConnections; i += 1){ if(g_Connections[i].State == CONNECTION_FREE || g_Connections[i].Socket == -1){ continue; } - ConnectionIndices[NumConnections] = i; - ConnectionFds[NumConnections].fd = g_Connections[i].Socket; - ConnectionFds[NumConnections].events = POLLIN | POLLOUT; - ConnectionFds[NumConnections].revents = 0; - NumConnections += 1; - } - - if(NumConnections <= 0){ - return; + Fds[NumFds].fd = g_Connections[i].Socket; + Fds[NumFds].events = POLLIN; + if(g_Connections[i].State == CONNECTION_WRITING){ + Fds[NumFds].events |= POLLOUT; + } + Fds[NumFds].revents = 0; + ConnectionIndices[NumFds] = i; + NumFds += 1; } - // NOTE(fusion): Poll connections. - int NumEvents = poll(ConnectionFds, NumConnections, 0); + ASSERT(NumFds > 0); + int NumEvents = poll(Fds, NumFds, -1); if(NumEvents == -1){ - LOG_ERR("Failed to poll connections: (%d) %s", errno, strerrordesc_np(errno)); + if(errno != ETIMEDOUT){ + LOG_ERR("Failed to poll connections: (%d) %s", + errno, strerrordesc_np(errno)); + } return; } // NOTE(fusion): Process connections. - for(int i = 0; i < NumConnections; i += 1){ - TConnection *Connection = &g_Connections[ConnectionIndices[i]]; - int Events = (int)ConnectionFds[i].revents; - CheckConnectionInput(Connection, Events); - CheckConnectionQueryRequest(Connection); - CheckConnectionQueryResponse(Connection); - CheckConnectionOutput(Connection, Events); - CheckConnection(Connection, Events); + for(int i = 0; i < NumFds; i += 1){ + int Index = ConnectionIndices[i]; + int Events = (int)Fds[i].revents; + if(Index >= 0 && Index < g_Config.MaxConnections){ + TConnection *Connection = &g_Connections[Index]; + CheckConnectionInput(Connection, Events); + CheckConnectionQueryRequest(Connection); + CheckConnectionQueryResponse(Connection); + CheckConnectionOutput(Connection, Events); + CheckConnection(Connection, Events); + }else if(Index == -1 && Fds[i].fd == g_UpdateEvent){ + ConsumeUpdateEvent(Events); + }else if(Index == -1 && Fds[i].fd == g_Listener){ + AcceptConnections(Events); + }else{ + LOG_ERR("Unknown connection index %d", Index); + } } } bool InitConnections(void){ + ASSERT(g_UpdateEvent == -1); ASSERT(g_Listener == -1); ASSERT(g_Connections == NULL); + g_UpdateEvent = eventfd(0, EFD_NONBLOCK); + if(g_UpdateEvent == -1){ + LOG_ERR("Failed to create eventfd: (%d) (%s)", + errno, strerrordesc_np(errno)); + return false; + } + g_Listener = ListenerBind((uint16)g_Config.QueryManagerPort); if(g_Listener == -1){ LOG_ERR("Failed to bind listener"); @@ -590,6 +668,11 @@ bool InitConnections(void){ } void ExitConnections(void){ + if(g_UpdateEvent != -1){ + close(g_UpdateEvent); + g_UpdateEvent = -1; + } + if(g_Listener != -1){ close(g_Listener); g_Listener = -1; diff --git a/src/database_sqlite.cc b/src/database_sqlite.cc index a9b4a58..8a5e526 100644 --- a/src/database_sqlite.cc +++ b/src/database_sqlite.cc @@ -90,7 +90,7 @@ static sqlite3_stmt *PrepareQuery(TDatabase *Database, const char *Text){ ASSERT(EntryText != NULL); if(strcmp(EntryText, Text) == 0){ Stmt = Entry->Stmt; - Entry->LastUsed = g_MonotonicTimeMS; + Entry->LastUsed = GetMonotonicUptimeMS(); break; } } @@ -109,7 +109,7 @@ static sqlite3_stmt *PrepareQuery(TDatabase *Database, const char *Text){ } Entry->Stmt = Stmt; - Entry->LastUsed = g_MonotonicTimeMS; + Entry->LastUsed = GetMonotonicUptimeMS(); Entry->Hash = Hash; }else{ if(sqlite3_stmt_busy(Stmt) != 0){ diff --git a/src/hostcache.cc b/src/hostcache.cc index 5cbc2fa..ef84def 100644 --- a/src/hostcache.cc +++ b/src/hostcache.cc @@ -62,10 +62,11 @@ bool ResolveHostName(const char *HostName, int *OutAddr){ THostCacheEntry *Entry = NULL; int LeastRecentlyUsedIndex = 0; int LeastRecentlyUsedTime = g_CachedHostNames[0].ResolveTime; + int TimeNow = GetMonotonicUptimeMS(); for(int i = 0; i < g_Config.MaxCachedHostNames; i += 1){ THostCacheEntry *Current = &g_CachedHostNames[i]; - if((g_MonotonicTimeMS - Current->ResolveTime) >= g_Config.HostNameExpireTime){ + if((TimeNow - Current->ResolveTime) >= g_Config.HostNameExpireTime){ memset(Current, 0, sizeof(THostCacheEntry)); } @@ -89,7 +90,7 @@ bool ResolveHostName(const char *HostName, int *OutAddr){ (int)strlen(HostName), (int)sizeof(Entry->HostName)); } Entry->Resolved = DoResolveHostName(HostName, &Entry->IPAddress); - Entry->ResolveTime = g_MonotonicTimeMS; + Entry->ResolveTime = TimeNow; } if(Entry && Entry->Resolved){ diff --git a/src/query.cc b/src/query.cc index f03a1db..b980b71 100644 --- a/src/query.cc +++ b/src/query.cc @@ -257,6 +257,7 @@ static void *WorkerThread(void *Data){ } QueryDone(Query); + WakeConnections(); } LOG("Worker#%d: DONE...", Worker->WorkerID); diff --git a/src/querymanager.cc b/src/querymanager.cc index d9f7ab8..6c1fffe 100644 --- a/src/querymanager.cc +++ b/src/querymanager.cc @@ -9,9 +9,9 @@ # error "Operating system not currently supported." #endif -int g_ShutdownSignal = 0; -int g_MonotonicTimeMS = 0; -TConfig g_Config = {}; +AtomicInt g_ShutdownSignal = {}; +int g_StartTimeMS = 0; +TConfig g_Config = {}; void LogAdd(const char *Prefix, const char *Format, ...){ char Entry[4096]; @@ -68,13 +68,17 @@ int64 GetClockMonotonicMS(void){ return (int64)((Counter.QuadPart * 1000) / Frequency.QuadPart); #else struct timespec Time; - clock_gettime(CLOCK_MONOTONIC, &Time); + clock_gettime(CLOCK_MONOTONIC_COARSE, &Time); return ((int64)Time.tv_sec * 1000) + ((int64)Time.tv_nsec / 1000000); #endif } -void SleepMS(int64 DurationMS){ +int GetMonotonicUptimeMS(void){ + return (int)(GetClockMonotonicMS() - g_StartTimeMS); +} + +void SleepMS(int DurationMS){ #if OS_WINDOWS Sleep((DWORD)DurationMS); #else @@ -386,8 +390,6 @@ bool ReadConfig(const char *FileName, TConfig *Config){ ReadStringBufConfig(Config->DatabaseName, Val); }else if(StringEqCI(Key, "DatabaseTLS")){ ReadBooleanConfig(&Config->DatabaseTLS, Val); - }else if(StringEqCI(Key, "UpdateRate")){ - ReadIntegerConfig(&Config->UpdateRate, Val); }else if(StringEqCI(Key, "QueryManagerPort")){ ReadIntegerConfig(&Config->QueryManagerPort, Val); }else if(StringEqCI(Key, "QueryManagerPassword")){ @@ -425,22 +427,22 @@ static bool SigHandler(int SigNr, sighandler_t Handler){ } static void ShutdownHandler(int SigNr){ - g_ShutdownSignal = SigNr; + AtomicStore(&g_ShutdownSignal, SigNr); + WakeConnections(); } int main(int argc, const char **argv){ (void)argc; (void)argv; - g_ShutdownSignal = 0; + AtomicStore(&g_ShutdownSignal, 0); if(!SigHandler(SIGPIPE, SIG_IGN) || !SigHandler(SIGINT, ShutdownHandler) || !SigHandler(SIGTERM, ShutdownHandler)){ return EXIT_FAILURE; } - int64 StartTime = GetClockMonotonicMS(); - g_MonotonicTimeMS = 0; + g_StartTimeMS = GetClockMonotonicMS(); // HostCache Config g_Config.MaxCachedHostNames = 100; @@ -457,7 +459,6 @@ int main(int argc, const char **argv){ g_Config.DatabaseTLS = true; // Connection Config - g_Config.UpdateRate = 20; g_Config.QueryManagerPort = 7174; StringBufCopy(g_Config.QueryManagerPassword, ""); g_Config.QueryWorkerThreads = 1; @@ -488,7 +489,6 @@ int main(int argc, const char **argv){ LOG("Max connections: %d", g_Config.MaxConnections); LOG("Max connection idle time: %dms", g_Config.MaxConnectionIdleTime); - if(!CheckSHA256()){ return EXIT_FAILURE; } @@ -502,21 +502,16 @@ int main(int argc, const char **argv){ return EXIT_FAILURE; } - LOG("Running at %d updates per second...", g_Config.UpdateRate); - int64 UpdateInterval = 1000 / (int64)g_Config.UpdateRate; - while(g_ShutdownSignal == 0){ - int64 UpdateStart = GetClockMonotonicMS(); - g_MonotonicTimeMS = (int)(UpdateStart - StartTime); + LOG("Running..."); + while(AtomicLoad(&g_ShutdownSignal) == 0){ + // NOTE(fusion): `ProcessConnections` will do a blocking `poll` which + // prevents this from being a hot loop, while still being reactive. ProcessConnections(); - int64 UpdateEnd = GetClockMonotonicMS(); - int64 NextUpdate = UpdateStart + UpdateInterval; - if(NextUpdate > UpdateEnd){ - SleepMS(NextUpdate - UpdateEnd); - } } + int ShutdownSignal = AtomicLoad(&g_ShutdownSignal); LOG("Received signal %d (%s), shutting down...", - g_ShutdownSignal, sigdescr_np(g_ShutdownSignal)); + ShutdownSignal, sigdescr_np(ShutdownSignal)); return EXIT_SUCCESS; } diff --git a/src/querymanager.hh b/src/querymanager.hh index 84ce9e5..15234e9 100644 --- a/src/querymanager.hh +++ b/src/querymanager.hh @@ -91,7 +91,6 @@ struct TConfig{ bool DatabaseTLS; // Connection Config - int UpdateRate; int QueryManagerPort; char QueryManagerPassword[30]; int QueryWorkerThreads; @@ -101,7 +100,6 @@ struct TConfig{ int MaxConnectionIdleTime; }; -extern int g_MonotonicTimeMS; extern TConfig g_Config; void LogAdd(const char *Prefix, const char *Format, ...) ATTR_PRINTF(2, 3); @@ -110,7 +108,8 @@ void LogAddVerbose(const char *Prefix, const char *Function, struct tm GetLocalTime(time_t t); int64 GetClockMonotonicMS(void); -void SleepMS(int64 DurationMS); +int GetMonotonicUptimeMS(void); +void SleepMS(int DurationMS); void CryptoRandom(uint8 *Buffer, int Count); int RoundSecondsToDays(int Seconds); @@ -1027,6 +1026,7 @@ void CheckConnectionQueryRequest(TConnection *Connection); void CheckConnectionQueryResponse(TConnection *Connection); void CheckConnectionOutput(TConnection *Connection, int Events); void CheckConnection(TConnection *Connection, int Events); +void WakeConnections(void); void ProcessConnections(void); bool InitConnections(void); void ExitConnections(void); -- cgit v1.2.3