worker initialization robustness + query processing

6 files changed, 297 insertions, 258 deletions

diff --git a/src/connections.cc b/src/connections.cc
index ecd8d4e..91fde30 100644
--- a/src/connections.cc
+++ b/src/connections.cc
@@ -244,6 +244,12 @@ void CheckConnectionInput(TConnection *Connection, int Events){
 	}
 }
 
+void ProcessQuery(TConnection *Connection){
+	ASSERT(Connection->Query != NULL);
+	QueryEnqueue(Connection->Query);
+	Connection->State = CONNECTION_RESPONSE;
+}
+
 void SendQueryResponse(TConnection *Connection){
 	ASSERT(Connection->Query != NULL);
 	if(Connection->State != CONNECTION_RESPONSE){
@@ -322,7 +328,7 @@ void CheckConnectionQueryRequest(TConnection *Connection){
 						Connection->RemoteAddress, LoginData);
 				Connection->Authorized = true;
 				Connection->ApplicationType = APPLICATION_TYPE_GAME;
-				QueryEnqueue(Query);
+				ProcessQuery(Connection);
 			}else{
 				// TODO(fusion): This should probably be a PANIC?
 				LOG_ERR("Rejecting connection %s: unable to rewrite login query..."
@@ -377,7 +383,7 @@ void CheckConnectionQueryRequest(TConnection *Connection){
 				|| QueryType == QUERY_EXCLUDE_FROM_AUCTIONS
 				|| QueryType == QUERY_CANCEL_HOUSE_TRANSFER
 				|| QueryType == QUERY_LOAD_WORLD_CONFIG){
-			QueryEnqueue(Connection->Query);
+			ProcessQuery(Connection);
 		}else{
 			LOG_ERR("Unknown GAME query %d from %s",
 					QueryType, Connection->RemoteAddress);
@@ -385,7 +391,7 @@ void CheckConnectionQueryRequest(TConnection *Connection){
 		}
 	}else if(Connection->ApplicationType == APPLICATION_TYPE_LOGIN){
 		if(QueryType == QUERY_LOGIN_ACCOUNT){
-			QueryEnqueue(Connection->Query);
+			ProcessQuery(Connection);
 		}else{
 			LOG_ERR("Unknown LOGIN query %d from %s",
 					QueryType, Connection->RemoteAddress);
@@ -400,7 +406,7 @@ void CheckConnectionQueryRequest(TConnection *Connection){
 				|| QueryType == QUERY_GET_WORLDS
 				|| QueryType == QUERY_GET_ONLINE_CHARACTERS
 				|| QueryType == QUERY_GET_KILL_STATISTICS){
-			QueryEnqueue(Connection->Query);
+			ProcessQuery(Connection);
 		}else{
 			LOG_ERR("Unknown WEB query %d from %s",
 					QueryType, Connection->RemoteAddress);
@@ -555,11 +561,6 @@ bool InitConnections(void){
 	ASSERT(g_Listener == -1);
 	ASSERT(g_Connections == NULL);
 
-	LOG("Query manager port: %d", g_Config.QueryManagerPort);
-	LOG("Query buffer size: %d", g_Config.QueryBufferSize);
-	LOG("Max connections: %d", g_Config.MaxConnections);
-	LOG("Max connection idle time: %dms", g_Config.MaxConnectionIdleTime);
-
 	g_Listener = ListenerBind((uint16)g_Config.QueryManagerPort);
 	if(g_Listener == -1){
 		LOG_ERR("Failed to bind listener");
diff --git a/src/database_sqlite.cc b/src/database_sqlite.cc
index 37e4e09..1ced471 100644
--- a/src/database_sqlite.cc
+++ b/src/database_sqlite.cc
@@ -2373,9 +2373,6 @@ bool CheckDatabaseSchema(void){
 }
 
 bool InitDatabase(void){
-	LOG("Database file: \"%s\"", g_Config.DatabaseFile);
-	LOG("Max cached statements: %d", g_Config.MaxCachedStatements);
-
 	int Flags = SQLITE_OPEN_READWRITE
 			| SQLITE_OPEN_CREATE
 			| SQLITE_OPEN_NOMUTEX;
@@ -2421,12 +2418,18 @@ void ExitDatabase(void){
 	}
 }
 
-TDatabase *OpenDatabase(void){
+TDatabase *DatabaseOpen(void){
+	//TODO
+	//return NULL;
+	return (TDatabase*)1;
+}
+
+void DatabaseClose(TDatabase *Database){
 	//TODO
-	return NULL;
 }
 
-void CloseDatabase(TDatabase *Database){
+bool DatabaseCheckpoint(TDatabase *Database){
 	//TODO
+	return false;
 }
 
diff --git a/src/hostcache.cc b/src/hostcache.cc
index 55c8663..5cbc2fa 100644
--- a/src/hostcache.cc
+++ b/src/hostcache.cc
@@ -18,8 +18,6 @@ static THostCacheEntry *g_CachedHostNames;
 
 bool InitHostCache(void){
 	ASSERT(g_CachedHostNames == NULL);
-	LOG("Max cached host names: %d", g_Config.MaxCachedHostNames);
-	LOG("Host name expire time: %dms", g_Config.HostNameExpireTime);
 	g_CachedHostNames = (THostCacheEntry*)calloc(
 			g_Config.MaxCachedHostNames, sizeof(THostCacheEntry));
 	return true;
diff --git a/src/query.cc b/src/query.cc
index 9d21833..35b9aca 100644
--- a/src/query.cc
+++ b/src/query.cc
@@ -2,6 +2,12 @@
 
 #include <pthread.h>
 
+enum : int {
+	WORKER_STATUS_SPAWNING = 0,
+	WORKER_STATUS_ACTIVE,
+	WORKER_STATUS_DONE,
+};
+
 struct TQueryQueue{
 	pthread_mutex_t Mutex;
 	pthread_cond_t WorkAvailable;
@@ -14,14 +20,129 @@ struct TQueryQueue{
 
 struct TWorker{
 	int WorkerID;
-	AtomicInt Running;
+	AtomicInt Status;
+	AtomicInt Stop;
 	pthread_t Thread;
 };
 
-static int g_NumWorkers;
 static TWorker *g_Workers;
 static TQueryQueue *g_QueryQueue;
 
+// Query Request
+//==============================================================================
+TWriteBuffer QueryBeginRequest(TQuery *Query, int QueryType){
+	Query->Request = TReadBuffer{};
+	TWriteBuffer WriteBuffer = TWriteBuffer(Query->Buffer, Query->BufferSize);
+	WriteBuffer.Write8((uint8)QueryType);
+	return WriteBuffer;
+}
+
+bool QueryFinishRequest(TQuery *Query, TWriteBuffer WriteBuffer){
+	ASSERT(WriteBuffer.Buffer   == Query->Buffer
+		&& WriteBuffer.Size     == Query->BufferSize
+		&& WriteBuffer.Position >= 1);
+	bool Result = !WriteBuffer.Overflowed();
+	if(Result){
+		Query->Request = TReadBuffer(WriteBuffer.Buffer, WriteBuffer.Position);
+	}
+	return Result;
+}
+
+bool QueryInternalResolveWorld(TQuery *Query, const char *World){
+	TWriteBuffer WriteBuffer = QueryBeginRequest(Query, QUERY_INTERNAL_RESOLVE_WORLD);
+	WriteBuffer.WriteString(World);
+	return QueryFinishRequest(Query, WriteBuffer);
+}
+
+// Query Response
+//==============================================================================
+TWriteBuffer *QueryBeginResponse(TQuery *Query, int Status){
+	Query->QueryStatus = 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);
+}
+
+
+// Query Processing
+//==============================================================================
+static bool ProcessInternalResolveWorld(TDatabase *Database, TQuery *Query);
+static bool ProcessCheckAccountPassword(TDatabase *Database, TQuery *Query);
+static bool ProcessLoginAccount(TDatabase *Database, TQuery *Query);
+static bool ProcessLoginAdmin(TDatabase *Database, TQuery *Query);
+static bool ProcessLoginGame(TDatabase *Database, TQuery *Query);
+static bool ProcessLogoutGame(TDatabase *Database, TQuery *Query);
+static bool ProcessSetNamelock(TDatabase *Database, TQuery *Query);
+static bool ProcessBanishAccount(TDatabase *Database, TQuery *Query);
+static bool ProcessSetNotation(TDatabase *Database, TQuery *Query);
+static bool ProcessReportStatement(TDatabase *Database, TQuery *Query);
+static bool ProcessBanishIpAddress(TDatabase *Database, TQuery *Query);
+static bool ProcessLogCharacterDeath(TDatabase *Database, TQuery *Query);
+static bool ProcessAddBuddy(TDatabase *Database, TQuery *Query);
+static bool ProcessRemoveBuddy(TDatabase *Database, TQuery *Query);
+static bool ProcessDecrementIsOnline(TDatabase *Database, TQuery *Query);
+static bool ProcessFinishAuctions(TDatabase *Database, TQuery *Query);
+static bool ProcessTransferHouses(TDatabase *Database, TQuery *Query);
+static bool ProcessEvictFreeAccounts(TDatabase *Database, TQuery *Query);
+static bool ProcessEvictDeletedCharacters(TDatabase *Database, TQuery *Query);
+static bool ProcessEvictExGuildleaders(TDatabase *Database, TQuery *Query);
+static bool ProcessInsertHouseOwner(TDatabase *Database, TQuery *Query);
+static bool ProcessUpdateHouseOwner(TDatabase *Database, TQuery *Query);
+static bool ProcessDeleteHouseOwner(TDatabase *Database, TQuery *Query);
+static bool ProcessGetHouseOwners(TDatabase *Database, TQuery *Query);
+static bool ProcessGetAuctions(TDatabase *Database, TQuery *Query);
+static bool ProcessStartAuction(TDatabase *Database, TQuery *Query);
+static bool ProcessInsertHouses(TDatabase *Database, TQuery *Query);
+static bool ProcessClearIsOnline(TDatabase *Database, TQuery *Query);
+static bool ProcessCreatePlayerlist(TDatabase *Database, TQuery *Query);
+static bool ProcessLogKilledCreatures(TDatabase *Database, TQuery *Query);
+static bool ProcessLoadPlayers(TDatabase *Database, TQuery *Query);
+static bool ProcessExcludeFromAuctions(TDatabase *Database, TQuery *Query);
+static bool ProcessCancelHouseTransfer(TDatabase *Database, TQuery *Query);
+static bool ProcessLoadWorldConfig(TDatabase *Database, TQuery *Query);
+static bool ProcessCreateAccount(TDatabase *Database, TQuery *Query);
+static bool ProcessCreateCharacter(TDatabase *Database, TQuery *Query);
+static bool ProcessGetAccountSummary(TDatabase *Database, TQuery *Query);
+static bool ProcessGetCharacterProfile(TDatabase *Database, TQuery *Query);
+static bool ProcessGetWorlds(TDatabase *Database, TQuery *Query);
+static bool ProcessGetOnlineCharacters(TDatabase *Database, TQuery *Query);
+static bool ProcessGetKillStatistics(TDatabase *Database, TQuery *Query);
+
 // Query Queue and Workers
 //==============================================================================
 TQuery *QueryNew(void){
@@ -82,19 +203,19 @@ void QueryEnqueue(TQuery *Query){
 	pthread_mutex_unlock(&g_QueryQueue->Mutex);
 }
 
-TQuery *QueryDequeue(AtomicInt *Running){
+TQuery *QueryDequeue(AtomicInt *Stop){
 	ASSERT(g_QueryQueue != NULL);
-	ASSERT(Running != NULL);
+	ASSERT(Stop != NULL);
 
 	TQuery *Query = NULL;
 	pthread_mutex_lock(&g_QueryQueue->Mutex);
 	uint32 NumQueries = g_QueryQueue->WritePos - g_QueryQueue->ReadPos;
-	while(NumQueries == 0 && AtomicLoad(Running)){
+	while(NumQueries == 0 && !AtomicLoad(Stop)){
 		pthread_cond_wait(&g_QueryQueue->WorkAvailable, &g_QueryQueue->Mutex);
 		NumQueries = g_QueryQueue->WritePos - g_QueryQueue->ReadPos;
 	}
 
-	if(NumQueries > 0 && AtomicLoad(Running)){
+	if(NumQueries > 0 && !AtomicLoad(Stop)){
 		uint32 MaxQueries = g_QueryQueue->MaxQueries;
 		if(NumQueries == MaxQueries){
 			pthread_cond_signal(&g_QueryQueue->RoomAvailable);
@@ -110,74 +231,88 @@ TQuery *QueryDequeue(AtomicInt *Running){
 static void *WorkerThread(void *Data){
 	ASSERT(Data != NULL);
 	TWorker *Worker = (TWorker*)Data;
-	if(!AtomicLoad(&Worker->Running)){
-		LOG_WARN("%d: Exiting on entry...", Worker->WorkerID);
+	if(AtomicLoad(&Worker->Stop)){
+		LOG_WARN("Worker#%d Stopping on entry...", Worker->WorkerID);
+		AtomicStore(&Worker->Status, WORKER_STATUS_DONE);
 		return NULL;
 	}
 
-	TDatabase *Database = OpenDatabase();
+	TDatabase *Database = DatabaseOpen();
 	if(Database == NULL){
-		LOG_ERR("%d: Failed to connect to database", Worker->WorkerID);
+		LOG_ERR("Worker#%d Failed to connect to database", Worker->WorkerID);
+		AtomicStore(&Worker->Status, WORKER_STATUS_DONE);
 		return NULL;
 	}
 
-	LOG("%d: Running...", Worker->WorkerID);
-	while(TQuery *Query = QueryDequeue(&Worker->Running)){
+	LOG("Worker#%d ACTIVE...", Worker->WorkerID);
+	AtomicStore(&Worker->Status, WORKER_STATUS_ACTIVE);
+	while(TQuery *Query = QueryDequeue(&Worker->Stop)){
+		bool (*ProcessQuery)(TDatabase*, TQuery*) = NULL;
 		Query->QueryType = Query->Request.Read8();
-		//bool (*ProcessQuery)(TDatabase*, TQuery*) = NULL;
-		switch(Query->QueryType){
-			case QUERY_INTERNAL_RESOLVE_WORLD:		ProcessInternalResolveWorld(Database, Query); break;
-			case QUERY_CHECK_ACCOUNT_PASSWORD:		ProcessCheckAccountPassword(Database, Query); break;
-			case QUERY_LOGIN_ACCOUNT:				ProcessLoginAccount(Database, Query); break;
-			case QUERY_LOGIN_ADMIN:					ProcessLoginAdmin(Database, Query); break;
-			case QUERY_LOGIN_GAME:					ProcessLoginGame(Database, Query); break;
-			case QUERY_LOGOUT_GAME:					ProcessLogoutGame(Database, Query); break;
-			case QUERY_SET_NAMELOCK:				ProcessSetNamelock(Database, Query); break;
-			case QUERY_BANISH_ACCOUNT:				ProcessBanishAccount(Database, Query); break;
-			case QUERY_SET_NOTATION:				ProcessSetNotation(Database, Query); break;
-			case QUERY_REPORT_STATEMENT:			ProcessReportStatement(Database, Query); break;
-			case QUERY_BANISH_IP_ADDRESS:			ProcessBanishIpAddress(Database, Query); break;
-			case QUERY_LOG_CHARACTER_DEATH:			ProcessLogCharacterDeath(Database, Query); break;
-			case QUERY_ADD_BUDDY:					ProcessAddBuddy(Database, Query); break;
-			case QUERY_REMOVE_BUDDY:				ProcessRemoveBuddy(Database, Query); break;
-			case QUERY_DECREMENT_IS_ONLINE:			ProcessDecrementIsOnline(Database, Query); break;
-			case QUERY_FINISH_AUCTIONS:				ProcessFinishAuctions(Database, Query); break;
-			case QUERY_TRANSFER_HOUSES:				ProcessTransferHouses(Database, Query); break;
-			case QUERY_EVICT_FREE_ACCOUNTS:			ProcessEvictFreeAccounts(Database, Query); break;
-			case QUERY_EVICT_DELETED_CHARACTERS:	ProcessEvictDeletedCharacters(Database, Query); break;
-			case QUERY_EVICT_EX_GUILDLEADERS:		ProcessEvictExGuildleaders(Database, Query); break;
-			case QUERY_INSERT_HOUSE_OWNER:			ProcessInsertHouseOwner(Database, Query); break;
-			case QUERY_UPDATE_HOUSE_OWNER:			ProcessUpdateHouseOwner(Database, Query); break;
-			case QUERY_DELETE_HOUSE_OWNER:			ProcessDeleteHouseOwner(Database, Query); break;
-			case QUERY_GET_HOUSE_OWNERS:			ProcessGetHouseOwners(Database, Query); break;
-			case QUERY_GET_AUCTIONS:				ProcessGetAuctions(Database, Query); break;
-			case QUERY_START_AUCTION:				ProcessStartAuction(Database, Query); break;
-			case QUERY_INSERT_HOUSES:				ProcessInsertHouses(Database, Query); break;
-			case QUERY_CLEAR_IS_ONLINE:				ProcessClearIsOnline(Database, Query); break;
-			case QUERY_CREATE_PLAYERLIST:			ProcessCreatePlayerlist(Database, Query); break;
-			case QUERY_LOG_KILLED_CREATURES:		ProcessLogKilledCreatures(Database, Query); break;
-			case QUERY_LOAD_PLAYERS:				ProcessLoadPlayers(Database, Query); break;
-			case QUERY_EXCLUDE_FROM_AUCTIONS:		ProcessExcludeFromAuctions(Database, Query); break;
-			case QUERY_CANCEL_HOUSE_TRANSFER:		ProcessCancelHouseTransfer(Database, Query); break;
-			case QUERY_LOAD_WORLD_CONFIG:			ProcessLoadWorldConfig(Database, Query); break;
-			case QUERY_CREATE_ACCOUNT:				ProcessCreateAccount(Database, Query); break;
-			case QUERY_CREATE_CHARACTER:			ProcessCreateCharacter(Database, Query); break;
-			case QUERY_GET_ACCOUNT_SUMMARY:			ProcessGetAccountSummary(Database, Query); break;
-			case QUERY_GET_CHARACTER_PROFILE:		ProcessGetCharacterProfile(Database, Query); break;
-			case QUERY_GET_WORLDS:					ProcessGetWorlds(Database, Query); break;
-			case QUERY_GET_ONLINE_CHARACTERS:		ProcessGetOnlineCharacters(Database, Query); break;
-			case QUERY_GET_KILL_STATISTICS:			ProcessGetKillStatistics(Database, Query); break;
-			default:{
-				QueryFailed(Query);
-				break;
+		/*switch(Query->QueryType){
+			case QUERY_INTERNAL_RESOLVE_WORLD:		ProcessQuery = ProcessInternalResolveWorld; break;
+			case QUERY_CHECK_ACCOUNT_PASSWORD:		ProcessQuery = ProcessCheckAccountPassword; break;
+			case QUERY_LOGIN_ACCOUNT:				ProcessQuery = ProcessLoginAccount; break;
+			case QUERY_LOGIN_ADMIN:					ProcessQuery = ProcessLoginAdmin; break;
+			case QUERY_LOGIN_GAME:					ProcessQuery = ProcessLoginGame; break;
+			case QUERY_LOGOUT_GAME:					ProcessQuery = ProcessLogoutGame; break;
+			case QUERY_SET_NAMELOCK:				ProcessQuery = ProcessSetNamelock; break;
+			case QUERY_BANISH_ACCOUNT:				ProcessQuery = ProcessBanishAccount; break;
+			case QUERY_SET_NOTATION:				ProcessQuery = ProcessSetNotation; break;
+			case QUERY_REPORT_STATEMENT:			ProcessQuery = ProcessReportStatement; break;
+			case QUERY_BANISH_IP_ADDRESS:			ProcessQuery = ProcessBanishIpAddress; break;
+			case QUERY_LOG_CHARACTER_DEATH:			ProcessQuery = ProcessLogCharacterDeath; break;
+			case QUERY_ADD_BUDDY:					ProcessQuery = ProcessAddBuddy; break;
+			case QUERY_REMOVE_BUDDY:				ProcessQuery = ProcessRemoveBuddy; break;
+			case QUERY_DECREMENT_IS_ONLINE:			ProcessQuery = ProcessDecrementIsOnline; break;
+			case QUERY_FINISH_AUCTIONS:				ProcessQuery = ProcessFinishAuctions; break;
+			case QUERY_TRANSFER_HOUSES:				ProcessQuery = ProcessTransferHouses; break;
+			case QUERY_EVICT_FREE_ACCOUNTS:			ProcessQuery = ProcessEvictFreeAccounts; break;
+			case QUERY_EVICT_DELETED_CHARACTERS:	ProcessQuery = ProcessEvictDeletedCharacters; break;
+			case QUERY_EVICT_EX_GUILDLEADERS:		ProcessQuery = ProcessEvictExGuildleaders; break;
+			case QUERY_INSERT_HOUSE_OWNER:			ProcessQuery = ProcessInsertHouseOwner; break;
+			case QUERY_UPDATE_HOUSE_OWNER:			ProcessQuery = ProcessUpdateHouseOwner; break;
+			case QUERY_DELETE_HOUSE_OWNER:			ProcessQuery = ProcessDeleteHouseOwner; break;
+			case QUERY_GET_HOUSE_OWNERS:			ProcessQuery = ProcessGetHouseOwners; break;
+			case QUERY_GET_AUCTIONS:				ProcessQuery = ProcessGetAuctions; break;
+			case QUERY_START_AUCTION:				ProcessQuery = ProcessStartAuction; break;
+			case QUERY_INSERT_HOUSES:				ProcessQuery = ProcessInsertHouses; break;
+			case QUERY_CLEAR_IS_ONLINE:				ProcessQuery = ProcessClearIsOnline; break;
+			case QUERY_CREATE_PLAYERLIST:			ProcessQuery = ProcessCreatePlayerlist; break;
+			case QUERY_LOG_KILLED_CREATURES:		ProcessQuery = ProcessLogKilledCreatures; break;
+			case QUERY_LOAD_PLAYERS:				ProcessQuery = ProcessLoadPlayers; break;
+			case QUERY_EXCLUDE_FROM_AUCTIONS:		ProcessQuery = ProcessExcludeFromAuctions; break;
+			case QUERY_CANCEL_HOUSE_TRANSFER:		ProcessQuery = ProcessCancelHouseTransfer; break;
+			case QUERY_LOAD_WORLD_CONFIG:			ProcessQuery = ProcessLoadWorldConfig; break;
+			case QUERY_CREATE_ACCOUNT:				ProcessQuery = ProcessCreateAccount; break;
+			case QUERY_CREATE_CHARACTER:			ProcessQuery = ProcessCreateCharacter; break;
+			case QUERY_GET_ACCOUNT_SUMMARY:			ProcessQuery = ProcessGetAccountSummary; break;
+			case QUERY_GET_CHARACTER_PROFILE:		ProcessQuery = ProcessGetCharacterProfile; break;
+			case QUERY_GET_WORLDS:					ProcessQuery = ProcessGetWorlds; break;
+			case QUERY_GET_ONLINE_CHARACTERS:		ProcessQuery = ProcessGetOnlineCharacters; break;
+			case QUERY_GET_KILL_STATISTICS:			ProcessQuery = ProcessGetKillStatistics; break;
+		}*/
+
+		if(ProcessQuery != NULL && DatabaseCheckpoint(Database)){
+			// NOTE(fusion): A minimum of 1 attempt is ASSUMED.
+			int Attempts = g_Config.QueryMaxAttempts;
+			while(!ProcessQuery(Database, Query)){
+				if(Attempts <= 0 || !DatabaseCheckpoint(Database)){
+					QueryFailed(Query);
+					break;
+				}
+				Attempts -= 1;
 			}
+		}else{
+			QueryFailed(Query);
 		}
 
 		QueryDone(Query);
 	}
 
-	LOG("%d: Exiting...", Worker->WorkerID);
-	CloseDatabase(Database);
+	LOG("Worker#%d DONE...", Worker->WorkerID);
+	DatabaseClose(Database);
+	AtomicStore(&Worker->Status, WORKER_STATUS_DONE);
 	return NULL;
 }
 
@@ -196,43 +331,74 @@ bool InitQuery(void){
 	g_QueryQueue->MaxQueries = 2 * g_Config.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){
+	int NumWorkers = g_Config.QueryWorkerThreads;
+	g_Workers = (TWorker*)calloc(NumWorkers, sizeof(TWorker));
+	for(int i = 0; i < NumWorkers; i += 1){
 		TWorker *Worker = &g_Workers[i];
 		Worker->WorkerID = i;
-		AtomicStore(&Worker->Running, 1);
+		AtomicStore(&Worker->Status, WORKER_STATUS_SPAWNING);
+		AtomicStore(&Worker->Stop, 0);
 		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;
 		}
 	}
 
+	while(true){
+		int NumWorkersSpawning = 0;
+		int NumWorkersActive = 0;
+		int NumWorkersDone = 0;
+		for(int i = 0; i < NumWorkers; i += 1){
+			int Status = AtomicLoad(&g_Workers[i].Status);
+			if(Status == WORKER_STATUS_SPAWNING){
+				NumWorkersSpawning += 1;
+			}else if(Status == WORKER_STATUS_ACTIVE){
+				NumWorkersActive += 1;
+			}else if(Status == WORKER_STATUS_DONE){
+				NumWorkersDone += 1;
+			}
+		}
+
+		if(NumWorkersSpawning > 0){
+			LOG("Waiting on worker threads... (SPAWNING=%d, ACTIVE=%d, DONE=%d)",
+					NumWorkersSpawning, NumWorkersActive, NumWorkersDone);
+			SleepMS(500);
+			continue;
+		}
+
+		if(NumWorkersDone > 0){
+			LOG_ERR("%d worker threads failed to initialize", NumWorkersDone);
+			return false;
+		}
+
+		ASSERT(NumWorkersActive == NumWorkers);
+		break;
+	}
+
 	return true;
 }
 
 void ExitQuery(void){
 	if(g_Workers != NULL){
-		for(int i = 0; i < g_NumWorkers; i += 1){
-			AtomicStore(&g_Workers[i].Running, 0);
-		}
+		ASSERT(g_QueryQueue != NULL);
 
-		if(g_QueryQueue != NULL){
-			pthread_cond_broadcast(&g_QueryQueue->WorkAvailable);
+		for(int i = 0; i < g_Config.QueryWorkerThreads; i += 1){
+			AtomicStore(&g_Workers[i].Stop, 1);
 		}
 
-		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_cond_broadcast(&g_QueryQueue->WorkAvailable);
+		for(int i = 0; i < g_Config.QueryWorkerThreads; i += 1){
+			// IMPORTANT(fusion): There is no "invalid" pthread handle so this
+			// is non-standard behaviour. Nevertheless the game server uses it
+			// and seems to be consistent on Linux, which is what matters at
+			// the end of the day.
+			if(g_Workers[i].Thread != 0){
 				pthread_join(g_Workers[i].Thread, NULL);
 			}
 		}
 
-		g_NumWorkers = 0;
 		free(g_Workers);
 	}
 
@@ -254,120 +420,6 @@ void ExitQuery(void){
 	}
 }
 
-// Query Request
-//==============================================================================
-TWriteBuffer QueryBeginRequest(TQuery *Query, int QueryType){
-	Query->Request = TReadBuffer{};
-	TWriteBuffer WriteBuffer = TWriteBuffer(Query->Buffer, Query->BufferSize);
-	WriteBuffer.Write8((uint8)QueryType);
-	return WriteBuffer;
-}
-
-bool QueryFinishRequest(TQuery *Query, TWriteBuffer WriteBuffer){
-	ASSERT(WriteBuffer.Buffer   == Query->Buffer
-		&& WriteBuffer.Size     == Query->BufferSize
-		&& WriteBuffer.Position >= 1);
-	bool Result = !WriteBuffer.Overflowed();
-	if(Result){
-		Query->Request = TReadBuffer(WriteBuffer.Buffer, WriteBuffer.Position);
-	}
-	return Result;
-}
-
-bool QueryInternalResolveWorld(TQuery *Query, const char *World){
-	TWriteBuffer WriteBuffer = QueryBeginRequest(Query, QUERY_INTERNAL_RESOLVE_WORLD);
-	WriteBuffer.WriteString(World);
-	return QueryFinishRequest(Query, WriteBuffer);
-}
-
-// Query Response
-//==============================================================================
-TWriteBuffer *QueryBeginResponse(TQuery *Query, int Status){
-	Query->QueryStatus = 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);
-}
-
-// Query Processing
-//==============================================================================
-void ProcessInternalResolveWorld(TDatabase *Database, TQuery *Query);
-void ProcessCheckAccountPassword(TDatabase *Database, TQuery *Query);
-void ProcessLoginAccount(TDatabase *Database, TQuery *Query);
-void ProcessLoginAdmin(TDatabase *Database, TQuery *Query);
-void ProcessLoginGame(TDatabase *Database, TQuery *Query);
-void ProcessLogoutGame(TDatabase *Database, TQuery *Query);
-void ProcessSetNamelock(TDatabase *Database, TQuery *Query);
-void ProcessBanishAccount(TDatabase *Database, TQuery *Query);
-void ProcessSetNotation(TDatabase *Database, TQuery *Query);
-void ProcessReportStatement(TDatabase *Database, TQuery *Query);
-void ProcessBanishIpAddress(TDatabase *Database, TQuery *Query);
-void ProcessLogCharacterDeath(TDatabase *Database, TQuery *Query);
-void ProcessAddBuddy(TDatabase *Database, TQuery *Query);
-void ProcessRemoveBuddy(TDatabase *Database, TQuery *Query);
-void ProcessDecrementIsOnline(TDatabase *Database, TQuery *Query);
-void ProcessFinishAuctions(TDatabase *Database, TQuery *Query);
-void ProcessTransferHouses(TDatabase *Database, TQuery *Query);
-void ProcessEvictFreeAccounts(TDatabase *Database, TQuery *Query);
-void ProcessEvictDeletedCharacters(TDatabase *Database, TQuery *Query);
-void ProcessEvictExGuildleaders(TDatabase *Database, TQuery *Query);
-void ProcessInsertHouseOwner(TDatabase *Database, TQuery *Query);
-void ProcessUpdateHouseOwner(TDatabase *Database, TQuery *Query);
-void ProcessDeleteHouseOwner(TDatabase *Database, TQuery *Query);
-void ProcessGetHouseOwners(TDatabase *Database, TQuery *Query);
-void ProcessGetAuctions(TDatabase *Database, TQuery *Query);
-void ProcessStartAuction(TDatabase *Database, TQuery *Query);
-void ProcessInsertHouses(TDatabase *Database, TQuery *Query);
-void ProcessClearIsOnline(TDatabase *Database, TQuery *Query);
-void ProcessCreatePlayerlist(TDatabase *Database, TQuery *Query);
-void ProcessLogKilledCreatures(TDatabase *Database, TQuery *Query);
-void ProcessLoadPlayers(TDatabase *Database, TQuery *Query);
-void ProcessExcludeFromAuctions(TDatabase *Database, TQuery *Query);
-void ProcessCancelHouseTransfer(TDatabase *Database, TQuery *Query);
-void ProcessLoadWorldConfig(TDatabase *Database, TQuery *Query);
-void ProcessCreateAccount(TDatabase *Database, TQuery *Query);
-void ProcessCreateCharacter(TDatabase *Database, TQuery *Query);
-void ProcessGetAccountSummary(TDatabase *Database, TQuery *Query);
-void ProcessGetCharacterProfile(TDatabase *Database, TQuery *Query);
-void ProcessGetWorlds(TDatabase *Database, TQuery *Query);
-void ProcessGetOnlineCharacters(TDatabase *Database, TQuery *Query);
-void ProcessGetKillStatistics(TDatabase *Database, TQuery *Query);
-
 // TODO(fusion): These are the old query processing functions. The new ones will
 // be very similar, except that we want a way to tell whether there was a database
 // failure, such as a connection reset, so we can automatically retry them up to
diff --git a/src/querymanager.cc b/src/querymanager.cc
index 0088d66..84d3625 100644
--- a/src/querymanager.cc
+++ b/src/querymanager.cc
@@ -382,9 +382,13 @@ bool ReadConfig(const char *FileName, TConfig *Config){
 			ReadIntegerConfig(&Config->QueryManagerPort, Val);
 		}else if(StringEqCI(Key, "QueryManagerPassword")){
 			ReadStringBufConfig(Config->QueryManagerPassword, Val);
+		}else if(StringEqCI(Key, "QueryWorkerThreads")){
+			ReadIntegerConfig(&Config->QueryWorkerThreads, Val);
 		}else if(StringEqCI(Key, "QueryBufferSize")
 				|| StringEqCI(Key, "MaxConnectionPacketSize")){
 			ReadSizeConfig(&Config->QueryBufferSize, Val);
+		}else if(StringEqCI(Key, "QueryMaxAttempts")){
+			ReadIntegerConfig(&Config->QueryMaxAttempts, Val);
 		}else if(StringEqCI(Key, "MaxConnections")){
 			ReadIntegerConfig(&Config->MaxConnections, Val);
 		}else if(StringEqCI(Key, "MaxConnectionIdleTime")){
@@ -430,7 +434,7 @@ int main(int argc, const char **argv){
 
 	// HostCache Config
 	g_Config.MaxCachedHostNames      = 100;
-	g_Config.HostNameExpireTime   = 30 * 60 * 1000; // milliseconds
+	g_Config.HostNameExpireTime      = 30 * 60 * 1000; // milliseconds
 
 	// Database Config
 	g_Config.MaxCachedStatements     = 100;
@@ -446,7 +450,9 @@ int main(int argc, const char **argv){
 	g_Config.UpdateRate              = 20;
 	g_Config.QueryManagerPort        = 7174;
 	StringBufCopy(g_Config.QueryManagerPassword, "");
+	g_Config.QueryWorkerThreads      = 1;
 	g_Config.QueryBufferSize         = (int)MB(1);
+	g_Config.QueryMaxAttempts        = 2;
 	g_Config.MaxConnections          = 50;
 	g_Config.MaxConnectionIdleTime   = 60 * 1000;      // milliseconds
 
@@ -455,6 +461,24 @@ int main(int argc, const char **argv){
 		return EXIT_FAILURE;
 	}
 
+	// NOTE(fusion): Print config values for debugging purposes.
+	LOG("Max cached host names:    %d",     g_Config.MaxCachedHostNames);
+	LOG("Host name expire time:    %dms",   g_Config.HostNameExpireTime);
+	LOG("Max cached statements:    %d",     g_Config.MaxCachedStatements);
+	LOG("Database file:            \"%s\"", g_Config.DatabaseFile);
+	LOG("Database host:            \"%s\"", g_Config.DatabaseHost);
+	LOG("Database port:            %d",     g_Config.DatabasePort);
+	LOG("Database user:            \"%s\"", g_Config.DatabaseUser);
+	LOG("Database name:            \"%s\"", g_Config.DatabaseName);
+	LOG("Database TLS:             %s",     g_Config.DatabaseTLS ? "true" : "false");
+	LOG("Query manager port:       %d",     g_Config.QueryManagerPort);
+	LOG("Query worker threads:     %d",     g_Config.QueryWorkerThreads);
+	LOG("Query buffer size:        %d",     g_Config.QueryBufferSize);
+	LOG("Query max attempts:       %d",     g_Config.QueryMaxAttempts);
+	LOG("Max connections:          %d",     g_Config.MaxConnections);
+	LOG("Max connection idle time: %dms",   g_Config.MaxConnectionIdleTime);
+
+
 	if(!CheckSHA256()){
 		return EXIT_FAILURE;
 	}
diff --git a/src/querymanager.hh b/src/querymanager.hh
index 9cfcbce..43cd6bc 100644
--- a/src/querymanager.hh
+++ b/src/querymanager.hh
@@ -94,7 +94,9 @@ struct TConfig{
 	int  UpdateRate;
 	int  QueryManagerPort;
 	char QueryManagerPassword[30];
+	int  QueryWorkerThreads;
 	int  QueryBufferSize;
+	int  QueryMaxAttempts;
 	int  MaxConnections;
 	int  MaxConnectionIdleTime;
 };
@@ -860,8 +862,9 @@ bool CheckDatabaseSchema(void);
 bool InitDatabase(void);
 void ExitDatabase(void);
 
-TDatabase *OpenDatabase(void);
-void CloseDatabase(TDatabase *Database);
+TDatabase *DatabaseOpen(void);
+void DatabaseClose(TDatabase *Database);
+bool DatabaseCheckpoint(TDatabase *Database);
 
 // query.cc
 //==============================================================================
@@ -927,14 +930,6 @@ struct TQuery{
 	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);
-
 TWriteBuffer QueryBeginRequest(TQuery *Query, int QueryType);
 bool QueryFinishRequest(TQuery *Query, TWriteBuffer WriteBuffer);
 bool QueryInternalResolveWorld(TQuery *Query, const char *World);
@@ -945,47 +940,13 @@ void QueryOk(TQuery *Query);
 void QueryError(TQuery *Query, int ErrorCode);
 void QueryFailed(TQuery *Query);
 
-void ProcessInternalResolveWorld(TDatabase *Database, TQuery *Query);
-void ProcessCheckAccountPassword(TDatabase *Database, TQuery *Query);
-void ProcessLoginAccount(TDatabase *Database, TQuery *Query);
-void ProcessLoginAdmin(TDatabase *Database, TQuery *Query);
-void ProcessLoginGame(TDatabase *Database, TQuery *Query);
-void ProcessLogoutGame(TDatabase *Database, TQuery *Query);
-void ProcessSetNamelock(TDatabase *Database, TQuery *Query);
-void ProcessBanishAccount(TDatabase *Database, TQuery *Query);
-void ProcessSetNotation(TDatabase *Database, TQuery *Query);
-void ProcessReportStatement(TDatabase *Database, TQuery *Query);
-void ProcessBanishIpAddress(TDatabase *Database, TQuery *Query);
-void ProcessLogCharacterDeath(TDatabase *Database, TQuery *Query);
-void ProcessAddBuddy(TDatabase *Database, TQuery *Query);
-void ProcessRemoveBuddy(TDatabase *Database, TQuery *Query);
-void ProcessDecrementIsOnline(TDatabase *Database, TQuery *Query);
-void ProcessFinishAuctions(TDatabase *Database, TQuery *Query);
-void ProcessTransferHouses(TDatabase *Database, TQuery *Query);
-void ProcessEvictFreeAccounts(TDatabase *Database, TQuery *Query);
-void ProcessEvictDeletedCharacters(TDatabase *Database, TQuery *Query);
-void ProcessEvictExGuildleaders(TDatabase *Database, TQuery *Query);
-void ProcessInsertHouseOwner(TDatabase *Database, TQuery *Query);
-void ProcessUpdateHouseOwner(TDatabase *Database, TQuery *Query);
-void ProcessDeleteHouseOwner(TDatabase *Database, TQuery *Query);
-void ProcessGetHouseOwners(TDatabase *Database, TQuery *Query);
-void ProcessGetAuctions(TDatabase *Database, TQuery *Query);
-void ProcessStartAuction(TDatabase *Database, TQuery *Query);
-void ProcessInsertHouses(TDatabase *Database, TQuery *Query);
-void ProcessClearIsOnline(TDatabase *Database, TQuery *Query);
-void ProcessCreatePlayerlist(TDatabase *Database, TQuery *Query);
-void ProcessLogKilledCreatures(TDatabase *Database, TQuery *Query);
-void ProcessLoadPlayers(TDatabase *Database, TQuery *Query);
-void ProcessExcludeFromAuctions(TDatabase *Database, TQuery *Query);
-void ProcessCancelHouseTransfer(TDatabase *Database, TQuery *Query);
-void ProcessLoadWorldConfig(TDatabase *Database, TQuery *Query);
-void ProcessCreateAccount(TDatabase *Database, TQuery *Query);
-void ProcessCreateCharacter(TDatabase *Database, TQuery *Query);
-void ProcessGetAccountSummary(TDatabase *Database, TQuery *Query);
-void ProcessGetCharacterProfile(TDatabase *Database, TQuery *Query);
-void ProcessGetWorlds(TDatabase *Database, TQuery *Query);
-void ProcessGetOnlineCharacters(TDatabase *Database, TQuery *Query);
-void ProcessGetKillStatistics(TDatabase *Database, TQuery *Query);
+TQuery *QueryNew(void);
+void QueryDone(TQuery *Query);
+int QueryRefCount(TQuery *Query);
+void QueryEnqueue(TQuery *Query);
+TQuery *QueryDequeue(AtomicInt *Stop);
+bool InitQuery(void);
+void ExitQuery(void);
 
 // connections.cc
 //==============================================================================