sketch out new query queue/worker for async queries

7 files changed, 372 insertions, 25 deletions

diff --git a/Makefile b/Makefile
index 3a8453d..0ba1f30 100644
--- a/Makefile
+++ b/Makefile
@@ -8,14 +8,29 @@ CFLAGS = -m64 -fno-strict-aliasing -pedantic -Wno-unused-parameter -Wall -Wextra
 CXXFLAGS = $(CFLAGS) --std=c++11
 LFLAGS = -Wl,-t
 
+# IMPORTANT(fusion): Using TABS outside recipes may cause errors in some cases.
+# This is because .RECIPEPREFIX defaults to it and any line starting with it is
+# assumed to introduce another recipe command. If indentation is needed before
+# any target is specified, use SPACES.
+
 DEBUG ?= 0
 ifneq ($(DEBUG), 0)
-	CFLAGS += -g -Og
+  CFLAGS += -g -Og -DENABLE_ASSERTIONS=1
+else
+  CFLAGS += -O2
+endif
+
+DATABASE ?= sqlite
+ifeq ($(DATABASE), sqlite)
+  DATABASEOBJ = $(BUILDDIR)/database_sqlite.obj $(BUILDDIR)/sqlite3.obj
+else ifeq ($(DATABASE), postgres)
+  DATABASEOBJ = $(BUILDDIR)/database_pq.obj
+  LFLAGS += -lpq
 else
-	CFLAGS += -O2
+  $(error Unsupported DATABASE: `$(DATABASE)`. Valid options are `sqlite` or `postgres`)
 endif
 
-$(BUILDDIR)/$(OUTPUTEXE): $(BUILDDIR)/connections.obj $(BUILDDIR)/database.obj $(BUILDDIR)/hostcache.obj $(BUILDDIR)/querymanager.obj $(BUILDDIR)/sha256.obj $(BUILDDIR)/sqlite3.obj
+$(BUILDDIR)/$(OUTPUTEXE): $(BUILDDIR)/connections.obj $(BUILDDIR)/hostcache.obj $(BUILDDIR)/query.obj $(BUILDDIR)/querymanager.obj $(BUILDDIR)/sha256.obj $(DATABASEOBJ)
 	@mkdir -p $(@D)
 	$(CXX) $(CXXFLAGS) -o $@ $^ $(LFLAGS)
 
@@ -23,11 +38,11 @@ $(BUILDDIR)/connections.obj: $(SRCDIR)/connections.cc $(SRCDIR)/querymanager.hh
 	@mkdir -p $(@D)
 	$(CXX) -c $(CXXFLAGS) -o $@ $<
 
-$(BUILDDIR)/database.obj: $(SRCDIR)/database.cc $(SRCDIR)/querymanager.hh
+$(BUILDDIR)/hostcache.obj: $(SRCDIR)/hostcache.cc $(SRCDIR)/querymanager.hh
 	@mkdir -p $(@D)
 	$(CXX) -c $(CXXFLAGS) -o $@ $<
 
-$(BUILDDIR)/hostcache.obj: $(SRCDIR)/hostcache.cc $(SRCDIR)/querymanager.hh
+$(BUILDDIR)/query.obj: $(SRCDIR)/query.cc $(SRCDIR)/querymanager.hh
 	@mkdir -p $(@D)
 	$(CXX) -c $(CXXFLAGS) -o $@ $<
 
@@ -39,6 +54,14 @@ $(BUILDDIR)/sha256.obj: $(SRCDIR)/sha256.cc $(SRCDIR)/querymanager.hh
 	@mkdir -p $(@D)
 	$(CXX) -c $(CXXFLAGS) -o $@ $<
 
+$(BUILDDIR)/database_sqlite.obj: $(SRCDIR)/database_sqlite.cc $(SRCDIR)/querymanager.hh
+	@mkdir -p $(@D)
+	$(CXX) -c $(CXXFLAGS) -o $@ $<
+
+$(BUILDDIR)/database_pq.obj: $(SRCDIR)/database_pq.cc $(SRCDIR)/querymanager.hh
+	@mkdir -p $(@D)
+	$(CXX) -c $(CXXFLAGS) -o $@ $<
+
 $(BUILDDIR)/sqlite3.obj: $(SRCDIR)/sqlite3.c $(SRCDIR)/sqlite3.h
 	@mkdir -p $(@D)
 	$(CC) -c $(CFLAGS) -o $@ $<
diff --git a/src/connections.cc b/src/connections.cc
index 3c92a51..26630cc 100644
--- a/src/connections.cc
+++ b/src/connections.cc
@@ -88,7 +88,7 @@ int ListenerAccept(int Listener, uint32 *OutAddr, uint16 *OutPort){
 		uint32 Addr = ntohl(SocketAddr.sin_addr.s_addr);
 		uint16 Port = ntohs(SocketAddr.sin_port);
 		if(Addr != INADDR_LOOPBACK){
-			LOG_ERR("Rejecting remote connection from %08X:%d.", Addr, Port);
+			LOG_ERR("Rejecting remote connection from %08X:%d", Addr, Port);
 			close(Socket);
 			continue;
 		}
@@ -220,7 +220,7 @@ void CheckConnectionInput(TConnection *Connection, int Events){
 		Connection->RWPosition += BytesRead;
 		if(Connection->RWPosition >= ReadSize){
 			if(Connection->RWSize != 0){
-				Connection->State = CONNECTION_PROCESSING;
+				Connection->State = CONNECTION_PENDING_QUERY;
 				Connection->LastActive = g_MonotonicTimeMS;
 				break;
 			}else if(Connection->RWPosition == 2){
@@ -249,9 +249,15 @@ void CheckConnectionInput(TConnection *Connection, int Events){
 			}
 		}
 	}
+}
 
-	if(Connection->State == CONNECTION_PROCESSING){
-		ProcessConnectionQuery(Connection);
+void CheckConnectionQuery(TConnection *Connection){
+	if(Connection->State == CONNECTION_PENDING_QUERY){
+		//ProcessConnectionQuery(Connection);
+	}else if(Connection->State == CONNECTION_PROCESSING_QUERY
+			&& QueryRefCount(Connection->Query) == 1){
+		//
+		//Connection->State = CONNECTION_WRITING;
 	}
 }
 
@@ -355,6 +361,7 @@ void ProcessConnections(void){
 		TConnection *Connection = &g_Connections[ConnectionIndices[i]];
 		int Events = (int)ConnectionFds[i].revents;
 		CheckConnectionInput(Connection, Events);
+		CheckConnectionQuery(Connection);
 		CheckConnectionOutput(Connection, Events);
 		CheckConnection(Connection, Events);
 	}
@@ -419,7 +426,7 @@ void CompoundBanishment(TBanishmentStatus Status, int *Days, bool *FinalWarning)
 }
 
 TWriteBuffer PrepareResponse(TConnection *Connection, int Status){
-	if(Connection->State != CONNECTION_PROCESSING){
+	if(Connection->State != CONNECTION_PROCESSING_QUERY){
 		LOG_ERR("Connection %s is not processing query (State: %d)",
 				Connection->RemoteAddress, Connection->State);
 		CloseConnection(Connection);
@@ -433,7 +440,7 @@ TWriteBuffer PrepareResponse(TConnection *Connection, int Status){
 }
 
 void SendResponse(TConnection *Connection, TWriteBuffer *WriteBuffer){
-	if(Connection->State != CONNECTION_PROCESSING){
+	if(Connection->State != CONNECTION_PROCESSING_QUERY){
 		LOG_ERR("Connection %s is not processing query (State: %d)",
 				Connection->RemoteAddress, Connection->State);
 		CloseConnection(Connection);
@@ -2068,7 +2075,7 @@ void ProcessConnectionQuery(TConnection *Connection){
 	// the disk.
 
 	TReadBuffer Buffer(Connection->Buffer, Connection->RWSize);
-	uint8 Query = Buffer.Read8();
+	int Query = Buffer.Read8();
 	if(!Connection->Authorized){
 		if(Query == QUERY_LOGIN){
 			ProcessLoginQuery(Connection, &Buffer);
diff --git a/src/database_pq.cc b/src/database_pq.cc
new file mode 100644
index 0000000..69cc45e
--- /dev/null
+++ b/src/database_pq.cc
@@ -0,0 +1,4 @@
+#include "querymanager.hh"
+
+// TODO(fusion):
+
diff --git a/src/database.cc b/src/database_sqlite.cc
index f4ca942..83f5bd9 100644
--- a/src/database.cc
+++ b/src/database_sqlite.cc
@@ -2445,3 +2445,13 @@ void ExitDatabase(void){
 		}
 	}
 }
+
+TDatabase *OpenDatabase(void){
+	//TODO
+	return NULL;
+}
+
+void CloseDatabase(TDatabase *Database){
+	//TODO
+}
+
diff --git a/src/query.cc b/src/query.cc
new file mode 100644
index 0000000..d1042e1
--- /dev/null
+++ b/src/query.cc
@@ -0,0 +1,248 @@
+#include "querymanager.hh"
+
+#include <pthread.h>
+
+struct TQueryQueue{
+	pthread_mutex_t Mutex;
+	pthread_cond_t EmptyCond;
+	pthread_cond_t FullCond;
+	uint32 ReadPos;
+	uint32 WritePos;
+	uint32 MaxQueries;
+	TQuery **Queries;
+};
+
+struct TWorker{
+	int WorkerID;
+	AtomicInt Running;
+	pthread_t Thread;
+};
+
+static int g_NumWorkers;
+static TWorker *g_Workers;
+static TQueryQueue *g_QueryQueue;
+
+// Query Queue and Workers
+//==============================================================================
+TQuery *QueryNew(void){
+	TQuery *Query = (TQuery*)calloc(1, sizeof(TQuery));
+	AtomicStore(&Query->RefCount, 1);
+	Query->BufferSize = g_MaxConnectionPacketSize;
+	Query->Buffer = (uint8*)calloc(1, Query->BufferSize);
+	return Query;
+}
+
+void QueryDone(TQuery *Query){
+	int RefCount = AtomicFetchAdd(&Query->RefCount, -1);
+	ASSERT(RefCount >= 1);
+	if(RefCount == 1){
+		free(Query->Buffer);
+		free(Query);
+	}
+}
+
+int QueryRefCount(TQuery *Query){
+	return AtomicLoad(&Query->RefCount);
+}
+
+void QueryEnqueue(TQuery *Query){
+	ASSERT(g_QueryQueue != NULL);
+	ASSERT(Query != NULL);
+
+	// IMPORTANT(fusion): A query object should be referenced by a connection
+	// and a query queue/worker at most. Anything else, we're gonna have a bad
+	// time.
+	int RefCount = 1;
+	if(!AtomicCompareExchange(&Query->RefCount, &RefCount, 2)){
+		LOG_ERR("Query already have %d references", RefCount);
+		return;
+	}
+
+	pthread_mutex_lock(&g_QueryQueue->Mutex);
+	uint32 NumQueries = g_QueryQueue->WritePos - g_QueryQueue->ReadPos;
+	uint32 MaxQueries = g_QueryQueue->MaxQueries;
+	while(NumQueries >= MaxQueries){
+		LOG_WARN("Execution stalled: queue is full (%u / %u)...",
+				NumQueries, MaxQueries);
+		pthread_cond_wait(&g_QueryQueue->FullCond, &g_QueryQueue->Mutex);
+		NumQueries = g_QueryQueue->WritePos - g_QueryQueue->ReadPos;
+		MaxQueries = g_QueryQueue->MaxQueries;
+	}
+
+	if(NumQueries == 0){
+		pthread_cond_signal(&g_QueryQueue->EmptyCond);
+	}
+
+	g_QueryQueue->Queries[g_QueryQueue->WritePos % MaxQueries] = Query;
+	g_QueryQueue->WritePos += 1;
+	pthread_mutex_unlock(&g_QueryQueue->Mutex);
+}
+
+TQuery *QueryDequeue(AtomicInt *Running){
+	ASSERT(g_QueryQueue != NULL);
+	ASSERT(Running != NULL);
+
+	TQuery *Query = NULL;
+	pthread_mutex_lock(&g_QueryQueue->Mutex);
+	uint32 NumQueries = g_QueryQueue->WritePos - g_QueryQueue->ReadPos;
+	while(NumQueries == 0 && AtomicLoad(Running)){
+		pthread_cond_wait(&g_QueryQueue->EmptyCond, &g_QueryQueue->Mutex);
+		NumQueries = g_QueryQueue->WritePos - g_QueryQueue->ReadPos;
+	}
+
+	if(NumQueries > 0 && AtomicLoad(Running)){
+		uint32 MaxQueries = g_QueryQueue->MaxQueries;
+		if(NumQueries == MaxQueries){
+			pthread_cond_signal(&g_QueryQueue->FullCond);
+		}
+
+		Query = g_QueryQueue->Queries[g_QueryQueue->ReadPos % MaxQueries];
+		g_QueryQueue->ReadPos += 1;
+	}
+	pthread_mutex_unlock(&g_QueryQueue->Mutex);
+	return Query;
+}
+
+static void *WorkerThread(void *Data){
+	ASSERT(Data != NULL);
+	TWorker *Worker = (TWorker*)Data;
+	if(!AtomicLoad(&Worker->Running)){
+		LOG_WARN("%d: Exiting on entry...", Worker->WorkerID);
+		return NULL;
+	}
+
+	TDatabase *Database = OpenDatabase();
+	if(Database == NULL){
+		LOG_ERR("%d: Failed to connect to database", Worker->WorkerID);
+		return NULL;
+	}
+
+	LOG("%d: Running...", Worker->WorkerID);
+	while(TQuery *Query = QueryDequeue(&Worker->Running)){
+		//TODO
+		switch(Query->QueryType){
+			default:{
+				//
+			}
+		}
+
+		QueryDone(Query);
+	}
+
+	LOG("%d: Exiting...", Worker->WorkerID);
+	CloseDatabase(Database);
+	return NULL;
+}
+
+bool InitQuery(void){
+	ASSERT(g_QueryQueue == NULL);
+	ASSERT(g_Workers == NULL);
+
+	// IMPORTANT(fusion): We'd ideally have a single query per connection at any
+	// given time but, in reality, connections could be reset while their queries
+	// are still in a query queue/worker, increasing the maximum number of queries
+	// in flight.
+	g_QueryQueue = (TQueryQueue*)calloc(1, sizeof(TQueryQueue));
+	pthread_mutex_init(&g_QueryQueue->Mutex, NULL);
+	pthread_cond_init(&g_QueryQueue->EmptyCond, NULL);
+	pthread_cond_init(&g_QueryQueue->FullCond, NULL);
+	g_QueryQueue->MaxQueries = 2 * g_MaxConnections;
+	g_QueryQueue->Queries = (TQuery**)calloc(g_QueryQueue->MaxQueries, sizeof(TQuery*));
+
+	g_NumWorkers = 1;
+	g_Workers = (TWorker*)calloc(g_NumWorkers, sizeof(TWorker));
+	for(int i = 0; i < g_NumWorkers; i += 1){
+		TWorker *Worker = &g_Workers[i];
+		Worker->WorkerID = i;
+		AtomicStore(&Worker->Running, 1);
+		int ErrorCode = pthread_create(&Worker->Thread, NULL, WorkerThread, Worker);
+		if(ErrorCode != 0){
+			LOG_ERR("Failed to spawn worker thread %d: (%d) %s",
+					i, ErrorCode, strerrordesc_np(ErrorCode));
+			Worker->WorkerID = -1;
+			return false;
+		}
+	}
+
+	return true;
+}
+
+void ExitQuery(void){
+	if(g_Workers != NULL){
+		for(int i = 0; i < g_NumWorkers; i += 1){
+			AtomicStore(&g_Workers[i].Running, 0);
+		}
+
+		for(int i = 0; i < g_NumWorkers; i += 1){
+			// IMPORTANT(fusion): The `WorkerID` will be set to -1 if we fail
+			// to spawn its thread.
+			if(g_Workers[i].WorkerID != -1){
+				pthread_join(g_Workers[i].Thread, NULL);
+			}
+		}
+
+		g_NumWorkers = 0;
+		free(g_Workers);
+	}
+
+	if(g_QueryQueue != NULL){
+		pthread_mutex_destroy(&g_QueryQueue->Mutex);
+		pthread_cond_destroy(&g_QueryQueue->EmptyCond);
+		pthread_cond_destroy(&g_QueryQueue->FullCond);
+
+		// TODO(fusion): Abort queries instead?
+		uint32 MaxQueries = g_QueryQueue->MaxQueries;
+		for(uint32 ReadPos = g_QueryQueue->ReadPos;
+				ReadPos != g_QueryQueue->WritePos;
+				ReadPos += 1){
+			QueryDone(g_QueryQueue->Queries[ReadPos % MaxQueries]);
+		}
+
+		free(g_QueryQueue->Queries);
+		free(g_QueryQueue);
+	}
+}
+
+// Queries
+//==============================================================================
+TWriteBuffer *QueryBeginResponse(TQuery *Query, int Status){
+	Query->Response = TWriteBuffer(Query->Buffer, Query->BufferSize);
+	Query->Response.Write16(0);
+	Query->Response.Write8((uint8)Status);
+	return &Query->Response;
+}
+
+bool QueryFinishResponse(TQuery *Query){
+	TWriteBuffer *Response = &Query->Response;
+	if(Response->Position <= 2){
+		LOG_ERR("Invalid response size");
+		return false;
+	}
+
+	int PayloadSize = Response->Position - 2;
+	if(PayloadSize < 0xFFFF){
+		Response->Rewrite16(0, (uint16)PayloadSize);
+	}else{
+		Response->Rewrite16(0, 0xFFFF);
+		Response->Insert32(2, (uint32)PayloadSize);
+	}
+
+	return !Response->Overflowed();
+}
+
+void QueryOk(TQuery *Query){
+	QueryBeginResponse(Query, QUERY_STATUS_OK);
+	QueryFinishResponse(Query);
+}
+
+void QueryError(TQuery *Query, int ErrorCode){
+	TWriteBuffer *Response = QueryBeginResponse(Query, QUERY_STATUS_ERROR);
+	Response->Write8((uint8)ErrorCode);
+	QueryFinishResponse(Query);
+}
+
+void QueryFailed(TQuery *Query){
+	QueryBeginResponse(Query, QUERY_STATUS_FAILED);
+	QueryFinishResponse(Query);
+}
+
diff --git a/src/querymanager.cc b/src/querymanager.cc
index 78a9eda..7d613b9 100644
--- a/src/querymanager.cc
+++ b/src/querymanager.cc
@@ -443,10 +443,10 @@ int main(int argc, const char **argv){
 	}
 
 	atexit(ExitHostCache);
-	atexit(ExitDatabase);
+	atexit(ExitQuery);
 	atexit(ExitConnections);
 	if(!InitHostCache()
-			|| !InitDatabase()
+			|| !InitQuery()
 			|| !InitConnections()){
 		return EXIT_FAILURE;
 	}
diff --git a/src/querymanager.hh b/src/querymanager.hh
index c1e48eb..5066e18 100644
--- a/src/querymanager.hh
+++ b/src/querymanager.hh
@@ -60,7 +60,7 @@ typedef size_t usize;
 #endif
 
 #define ASSERT_ALWAYS(expr) if(!(expr)) { TRAP(); }
-#if BUILD_DEBUG
+#if ENABLE_ASSERTIONS
 #	define ASSERT(expr) ASSERT_ALWAYS(expr)
 #else
 #	define ASSERT(expr) ((void)(expr))
@@ -117,6 +117,34 @@ bool ReadSizeConfig(int *Dest, const char *Val);
 bool ReadStringConfig(char *Dest, int DestCapacity, const char *Val);
 bool ReadConfig(const char *FileName);
 
+// AtomicInt
+//==============================================================================
+#if COMPILER_GCC || COMPILER_CLANG
+struct AtomicInt{
+	volatile int Value;
+};
+
+inline int AtomicLoad(AtomicInt *Ptr){
+	return __atomic_load_n(&Ptr->Value, __ATOMIC_SEQ_CST);
+}
+
+inline void AtomicStore(AtomicInt *Ptr, int Value){
+	__atomic_store_n(&Ptr->Value, Value, __ATOMIC_SEQ_CST);
+}
+
+inline int AtomicFetchAdd(AtomicInt *Ptr, int Value){
+	return __atomic_fetch_add(&Ptr->Value, Value, __ATOMIC_SEQ_CST);
+}
+
+inline bool AtomicCompareExchange(AtomicInt *Ptr, int *Expected, int Desired){
+	return __atomic_compare_exchange_n(&Ptr->Value, Expected,
+			Desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
+}
+#else
+// TODO(fusion): On MSVC you'd use Interlocked* builtins.
+#	error "Atomics not implemented for compiler."
+#endif
+
 // Buffer Utility
 //==============================================================================
 inline uint8 BufferRead8(const uint8 *Buffer){
@@ -533,6 +561,33 @@ public:
 	const T *end(void) const { return m_Data + m_Length; }
 };
 
+// hostcache.cc
+//==============================================================================
+bool InitHostCache(void);
+void ExitHostCache(void);
+bool ResolveHostName(const char *HostName, int *OutAddr);
+
+// query.cc
+//==============================================================================
+struct TQuery{
+	AtomicInt RefCount;
+	int QueryType;
+	int WorldID;
+	int BufferSize;
+	uint8 *Buffer;
+	TReadBuffer Request;
+	TWriteBuffer Response;
+};
+
+TQuery *QueryNew(void);
+void QueryDone(TQuery *Query);
+int QueryRefCount(TQuery *Query);
+void QueryEnqueue(TQuery *Query);
+TQuery *QueryDequeue(AtomicInt *Running);
+
+bool InitQuery(void);
+void ExitQuery(void);
+
 // connections.cc
 //==============================================================================
 enum : int {
@@ -592,10 +647,11 @@ enum : int {
 };
 
 enum ConnectionState: int {
-	CONNECTION_FREE			= 0,
-	CONNECTION_READING		= 1,
-	CONNECTION_PROCESSING	= 2,
-	CONNECTION_WRITING		= 3,
+	CONNECTION_FREE					= 0,
+	CONNECTION_READING				= 1,
+	CONNECTION_PENDING_QUERY		= 2,
+	CONNECTION_PROCESSING_QUERY		= 3,
+	CONNECTION_WRITING				= 4,
 };
 
 struct TConnection{
@@ -609,6 +665,7 @@ struct TConnection{
 	int ApplicationType;
 	int WorldID;
 	char RemoteAddress[30];
+	TQuery *Query; // TODO
 };
 
 int ListenerBind(uint16 Port);
@@ -832,7 +889,8 @@ struct TOnlineCharacter{
 	char Profession[30];
 };
 
-// NOTE(fusion): Transaction scope guard.
+struct TDatabase;
+//struct TDatabaseTx;
 struct TransactionScope{
 private:
 	const char *m_Context;
@@ -940,11 +998,8 @@ bool CheckDatabaseSchema(void);
 bool InitDatabase(void);
 void ExitDatabase(void);
 
-// hostcache.cc
-//==============================================================================
-bool InitHostCache(void);
-void ExitHostCache(void);
-bool ResolveHostName(const char *HostName, int *OutAddr);
+TDatabase *OpenDatabase(void);
+void CloseDatabase(TDatabase *Database);
 
 // sha256.cc
 //==============================================================================