switch to blocking poll + fix config, signals, time, non-game login

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 <fcntl.h>
 #	include <netinet/in.h>
 #	include <poll.h>
+#	include <sys/eventfd.h>
 #	include <sys/socket.h>
 #	include <unistd.h>
 #	include <time.h>
@@ -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);