beginning of `communication.cc`

1 files changed, 335 insertions, 0 deletions

diff --git a/src/communication.cc b/src/communication.cc
new file mode 100644
index 0000000..dc28566
--- /dev/null
+++ b/src/communication.cc
@@ -0,0 +1,335 @@
+#include "communication.hh"
+#include "config.hh"
+#include "connections.hh"
+#include "containers.hh"
+#include "crypto.hh"
+#include "query.hh"
+#include "threads.hh"
+
+#include "stubs.hh"
+
+#include <signal.h>
+
+#define MAX_COMMUNICATION_THREADS 1100
+#define THREAD_OWN_STACK_SIZE ((int)KB(64))
+
+static int TERMINAL_VERSION[3] = {770, 770, 770};
+static int TCPSocket;
+static ThreadHandle AcceptorThread;
+static pid_t AcceptorThreadPID;
+static int ActiveConnections;
+
+static Semaphore RSAMutex(1);
+static TRSAPrivateKey PrivateKey;
+
+static TQueryManagerConnectionPool QueryManagerConnectionPool(10);
+static int LoadHistory[360];
+static int LoadHistoryPointer;
+static int TotalLoad;
+static int TotalSend;
+static int TotalRecv;
+static uint32 LagEnd;
+static uint32 EarliestFreeAccountAdmissionRound;
+static store<TWaitinglistEntry, 100> Waitinglist;
+static TWaitinglistEntry *WaitinglistHead;
+
+static Semaphore CommunicationThreadMutex(1);
+static bool UseOwnStacks;
+static uint8 CommunicationThreadStacks[MAX_COMMUNICATION_THREADS][THREAD_OWN_STACK_SIZE];
+static pid_t LastUsingCommunicationThread[MAX_COMMUNICATION_THREADS];
+static int FreeCommunicationThreadStacks[MAX_COMMUNICATION_THREADS];
+static int NumberOfFreeCommunicationThreadStacks;
+
+// Communication Thread Stacks
+// =============================================================================
+void GetCommunicationThreadStack(int *StackNumber, void **Stack){
+	*StackNumber = -1;
+	*Stack = NULL;
+
+	if(!UseOwnStacks){
+		error("GetCommunicationThreadStack: Bibliothek unterstützt keine eigenen Stacks.\n");
+		return;
+	}
+
+	CommunicationThreadMutex.down();
+	for(int i = 0; i < NumberOfFreeCommunicationThreadStacks; i += 1){
+		int FreeStack = FreeCommunicationThreadStacks[i];
+		if(LastUsingCommunicationThread[FreeStack] == 0
+				|| kill(LastUsingCommunicationThread[FreeStack], 0) == -1){
+			// NOTE(fusion): A little swap and pop action.
+			NumberOfFreeCommunicationThreadStacks -= 1;
+			FreeCommunicationThreadStacks[i] = FreeCommunicationThreadStacks[NumberOfFreeCommunicationThreadStacks];
+
+			*StackNumber = FreeStack;
+			*Stack = CommunicationThreadStacks[FreeStack];
+			break;
+		}
+	}
+	CommunicationThreadMutex.up();
+}
+
+void AttachCommunicationThreadStack(int StackNumber){
+	LastUsingCommunicationThread[StackNumber] = getpid();
+}
+
+void ReleaseCommunicationThreadStack(int StackNumber){
+	CommunicationThreadMutex.down();
+	FreeCommunicationThreadStacks[NumberOfFreeCommunicationThreadStacks] = StackNumber;
+	NumberOfFreeCommunicationThreadStacks += 1;
+	CommunicationThreadMutex.up();
+}
+
+void InitCommunicationThreadStacks(void){
+	if(UseOwnStacks){
+		memset(CommunicationThreadStacks, 0xAA, sizeof(CommunicationThreadStacks));
+		for(int i = 0; i < MAX_COMMUNICATION_THREADS; i += 1){
+			LastUsingCommunicationThread[i] = 0;
+			FreeCommunicationThreadStacks[i] = i;
+		}
+		NumberOfFreeCommunicationThreadStacks = MAX_COMMUNICATION_THREADS;
+	}
+}
+
+void ExitCommunicationThreadStacks(void){
+	if(UseOwnStacks){
+		if(NumberOfFreeCommunicationThreadStacks != MAX_COMMUNICATION_THREADS){
+			error("FreeCommunicationThreadStacks: Nicht alle Stacks freigegeben.\n");
+		}
+
+		int HighestStackAddress = -1;
+		int LowestStackAddress = THREAD_OWN_STACK_SIZE;
+		for(int i = 0; i < MAX_COMMUNICATION_THREADS; i += 1){
+			for(int Addr = 0; Addr < THREAD_OWN_STACK_SIZE; Addr += 1){
+				if(CommunicationThreadStacks[i][Addr] != 0xAA){
+					if(Addr < LowestStackAddress){
+						LowestStackAddress = Addr;
+					}
+
+					if(Addr > HighestStackAddress){
+						HighestStackAddress = Addr;
+					}
+				}
+			}
+		}
+
+		// NOTE(fusion): It seems we want to track whether the stack size is
+		// too small but I'd argue the method is not very robust.
+		if((HighestStackAddress - LowestStackAddress) > (THREAD_OWN_STACK_SIZE / 2)){
+			error("Maximale Stack-Ausdehnung: %d..%d\n", LowestStackAddress, HighestStackAddress);
+		}
+	}
+}
+
+// Load History
+// =============================================================================
+void InitLoadHistory(void){
+	for(int i = 0; i < NARRAY(LoadHistory); i += 1){
+		LoadHistory[i] = 0;
+	}
+	LoadHistoryPointer = 0;
+	TotalLoad = 0;
+	TotalSend = 0;
+	TotalRecv = 0;
+	LagEnd = 0;
+	EarliestFreeAccountAdmissionRound = 0;
+	InitLog("netload");
+}
+
+bool LagDetected(void){
+	return RoundNr <= LagEnd;
+}
+
+void NetLoad(int Amount, bool Send){
+	CommunicationThreadMutex.down();
+	if(Send){
+		TotalSend += Amount;
+	}else{
+		TotalRecv += Amount;
+	}
+	CommunicationThreadMutex.up();
+}
+
+void NetLoadSummary(void){
+	CommunicationThreadMutex.down();
+	Log("netload", "gesendet:  %d Bytes.\n", TotalSend);
+	Log("netload", "empfangen: %d Bytes.\n", TotalRecv);
+	TotalSend = 0;
+	TotalRecv = 0;
+	CommunicationThreadMutex.up();
+}
+
+void NetLoadCheck(void){
+	static int LastRecv;
+
+	int DeltaRecv = TotalRecv - LastRecv;
+	LastRecv = TotalRecv;
+	if(DeltaRecv < 0){
+		return;
+	}
+
+	int DeltaRecvPerPlayer = 0;
+	int PlayersOnline = GetPlayersOnline();
+	if(PlayersOnline > 0){
+		DeltaRecvPerPlayer = DeltaRecv / PlayersOnline;
+	}
+
+	TotalLoad -= LoadHistory[LoadHistoryPointer];
+	TotalLoad += DeltaRecvPerPlayer;
+	LoadHistory[LoadHistoryPointer] = DeltaRecvPerPlayer;
+	LoadHistoryPointer += 1;
+	if(LoadHistoryPointer >= NARRAY(LoadHistory)){
+		LoadHistoryPointer = 0;
+	}
+
+	// NOTE(fusion): Running this lag check only makes sense if `LoadHistory`
+	// is filled up which won't be the case until this function executes at
+	// least as many times as there are entries in `LoadHistory`.
+	//	Looking at `AdvanceGame`, we see that this function is called every
+	// 10 rounds, giving us the value of `EarliestLagCheckRound`.
+	constexpr uint32 EarliestLagCheckRound = 10 * NARRAY(LoadHistory);
+	if(RoundNr >= EarliestLagCheckRound && PlayersOnline >= 50){
+		int AvgDeltaRecvPerPlayer = (TotalLoad / NARRAY(LoadHistory));
+		if(DeltaRecvPerPlayer < (AvgDeltaRecvPerPlayer / 2)){
+			Log("game", "Lag erkannt!\n");
+			LagEnd = RoundNr + 30;
+
+			// NOTE(fusion): This formula looks weird but when we take `MaxPlayers` 
+			// and `PremiumPlayerBuffer` to be 950 and 150 (taken from the config
+			// file, although their values are now loaded from the database), we
+			// get a line with negative values when the number of players online
+			// is less than 650, and exactly 60 when it is 950. Since the delay
+			// is 60 when `PremiumPlayerBuffer` is zero, I don't think this is a
+			// coincidence.
+			int FreeAccountAdmissionDelay = 60;
+			if(PremiumPlayerBuffer != 0){
+				FreeAccountAdmissionDelay = (PlayersOnline + PremiumPlayerBuffer * 2 - MaxPlayers);
+				FreeAccountAdmissionDelay = (FreeAccountAdmissionDelay * 30) / PremiumPlayerBuffer;
+				if(FreeAccountAdmissionDelay < 0){
+					FreeAccountAdmissionDelay = 0;
+				}
+			}
+
+			uint32 FreeAccountAdmissionRound = RoundNr + (uint32)FreeAccountAdmissionDelay;
+			if(EarliestFreeAccountAdmissionRound < FreeAccountAdmissionRound){
+				EarliestFreeAccountAdmissionRound = FreeAccountAdmissionRound;
+			}
+
+			TConnection *Connection = GetFirstConnection();
+			while(Connection != NULL){
+				if(Connection->Live()){
+					Connection->EmergencyPing();
+				}
+				Connection = GetNextConnection();
+			}
+		}
+	}
+}
+
+// Communication Handling
+// =============================================================================
+bool WriteToSocket(TConnection *Connection, uint8 *Buffer, int Size){
+	// TODO(fusion): I think `Size` refers to the payload but `Buffer` also has
+	// room for writing the packet's length at the beginning and enough room for
+	// padding (supposedly).
+
+	while((Size % 8) != 0){
+		Buffer[Size + 2] = rand_r(&Connection->RandomSeed);
+		Size += 1;
+	}
+
+	for(int i = 0; i < Size; i += 8){
+		Connection->SymmetricKey.encrypt(&Buffer[i + 2]);
+	}
+
+	TWriteBuffer WriteBuffer(Buffer, 2);
+	WriteBuffer.writeWord((uint16)Size);
+
+	int Attempts = 50;
+	int BytesToWrite = Size + 2;
+	uint8 *WritePtr = Buffer;
+	while(BytesToWrite > 0){
+		int ret = (int)write(Connection->GetSocket(), WritePtr, BytesToWrite);
+		if(ret > 0){
+			BytesToWrite -= ret;
+			WritePtr += ret;
+		}else if(ret == 0){
+			// TODO(fusion): Can this even happen?
+			error("WriteToSocket: Fehler %d beim Senden an Socket %d.\n",
+					errno, Connection->GetSocket());
+			return false;
+		}else{
+			if(errno == EINTR){
+				continue;
+			}
+
+			if(errno != EAGAIN || Attempts <= 0){
+				if(errno == ECONNRESET || errno == EPIPE || errno == EAGAIN){
+					Log("game", "Verbindung an Socket %d zusammengebrochen.\n",
+							Connection->GetSocket());
+				}else{
+					error("WriteToSocket: Fehler %d beim Senden an Socket %d.\n",
+							errno, Connection->GetSocket());
+				}
+				return false;
+			}
+
+			DelayThread(0, 100000);
+			Attempts -= 1;
+		}
+	}
+
+	// TODO(fusion): Do we add 50 extra bytes to account for TCP segment headers?
+	// This does make sense If we assume packets are split into ~2.5 segments on
+	// average, with each segment header being 20 bytes.
+	NetLoad(Size + 50, true);
+	return true;
+}
+
+bool SendLoginMessage(TConnection *Connection, int Type, char *Message, int WaitingTime){
+	// TODO(fusion):
+	//	LOGIN_MESSAGE_ERROR = 20
+	//	LOGIN_MESSAGE_? = 21
+	//	LOGIN_MESSAGE_WAITING_LIST = 22
+	if(Type != 20 && Type != 21 && Type != 22){
+		error("SendLoginMessage: Ungültiger Meldungstyp %d.\n", Type);
+		return true;
+	}
+
+	if(Message == NULL){
+		error("SendLoginMessage: Message ist NULL.\n");
+		return true;
+	}
+
+	if(Type == 22 && (WaitingTime < 0 || WaitingTime > UINT8_MAX)){
+		error("SendLoginMessage: Ungültige Wartezeit %d.\n", WaitingTime);
+		return true;
+	}
+
+	if(strlen(Message) > 290){
+		error("SendLoginMessage: Botschaft zu lang (%s).\n", Message);
+		return true;
+	}
+
+	// TODO(fusion): Writing output messages should have more robust helpers
+	// to avoid all sorts of memory bugs but since we're only doing it in two
+	// places from what I've seen, I'm not actually gonna bother (for now at
+	// least).
+
+	// NOTE(fusion): We make sure we leave two extra bytes at the beginning so
+	// `WriteToSocket` can write the packet size. We also make sure that the
+	// remainder of the buffer has a size that is multiple of 8 so `WriteToSocket`
+	// can add any necessary padding for XTEA encryption without overflowing it.
+	uint8 Data[302]; // 2 + 300
+	TWriteBuffer WriteBuffer(Data + 2, sizeof(Data) - 2);
+	WriteBuffer.writeWord(0);
+	WriteBuffer.writeByte((uint8)Type);
+	WriteBuffer.writeString(Message);
+	if(Type == 22){
+		WriteBuffer.writeByte(WaitingTime);
+	}
+
+	int Size = WriteBuffer.Position;
+	WriteBuffer.Position = 0;
+	WriteBuffer.writeWord((uint16)(Size - 2));
+	return WriteToSocket(Connection, Data, Size);
+}