samp-re/reliability.c

/* vim: set filetype=c ts=8 noexpandtab: */

//#define CALL_ORIGINAL

#define RELIABILITY_PRINT

#ifdef RELIABILITY_PRINT
#define dprintf(X,...) printf(X,__VA_ARGS__)
#else
#define dprintf(X,...)
#endif

#include "common.h"
#include "bitstream.h"
#include "rakpeer.h"
#include "reliability.h"
#include "rangelist_deserialize.h"
#include <stdio.h>

__declspec(naked)
static
void RangeList_Short__Insert(
	void *this,
	short value)
{
	_asm {
		pop eax
		pop ecx
		push eax
		mov eax, 0x45E0C0
		jmp eax
	}
}

static
void SendAckForPacket(
	struct CReliabilityLayer *this,
	struct CInternalPacket *packet)
{
	if (packet->packetReliability != RELIABLE_SEQUENCED ||
		packet->packetReliability != RELIABLE_ORDENED ||
		packet->packetReliability != RELIABLE)
	{
		return;
	}

	this->statistics_acknowlegementsSent++;
	/*acknowlegements is probably an inlined struct*/
	RangeList_Short__Insert(&this->acknowlegements, packet->messageNumber);
}

static
void ResetReceivedPackets(struct CReliabilityLayer *this)
{
	if (!this->resetReceivedPackets) {
		return;
	}

	thiscall1((void*) 0x45B770, &this->hasReceivedPacketQueue,
			DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE);
}

/**
@return 0 if packet should be discarded
*/
static
int HandlePacketHole(
	struct CReliabilityLayer *this,
	int holeSize,
	struct CRaknetTimeNS time)
{
	struct CQueue *queueptr;
	struct CRaknetTimeNS *timeAt, timeTmp;

#define queue this->hasReceivedPacketQueue
	queueptr = &this->hasReceivedPacketQueue;

	if (holeSize == 0) {
		if (QUEUE_SIZE(queue)) {
			QUEUE_POP_IGNOREVALUE(queue);
		}
		this->receivedPacketsBaseIndex++;
		return 1;
	}

	dprintf("packet hole is %d\n", holeSize);

	if (holeSize > 0x8000) {
		/*? underflow apparently*/
		this->statistics_duplicateMessagesReceived++;
		return 0;
	}

	if (holeSize < QUEUE_SIZE(queue)) {
		timeAt = Queue_AtPosition_QWORD(queueptr, holeSize);

		if (timeAt->lo32 == 0 && timeAt->hi32 == 0) {
			/*was already received, so duplicate packet*/
			this->statistics_duplicateMessagesReceived++;
			return 0;
		}

		/*it was a hole that has just been filled*/
		timeAt->lo32 = timeAt->hi32 = 0;
		return 1;
	}

	/*add to queue, with holes in between*/
	while (holeSize > QUEUE_SIZE(queue)) {
		timeTmp = time;
		RaknetTimeNS_Add(&timeTmp, this->timeoutTimeMS * 1000);
		Queue_Push_QWORD(queueptr, &time);
	}
	timeTmp.lo32 = timeTmp.hi32 = 0;
	Queue_Push_QWORD(queueptr, &timeTmp);
	return 1;
}

static
void RemovePacketHolesNotReceivedInTime(
	struct CReliabilityLayer *this,
	struct CRaknetTimeNS time)
{
	struct CQueue *queueptr;
	struct CRaknetTimeNS *expireTime;

#define queue this->hasReceivedPacketQueue
	queueptr = &this->hasReceivedPacketQueue;

	while (QUEUE_SIZE(queue)) {
		expireTime = Queue_AtPosition_QWORD(queueptr, queue.head);
		if (!RaknetTimeNS_IsBigger(time, *expireTime)) {
			return;
		}
		QUEUE_POP_IGNOREVALUE(queue);
		this->receivedPacketsBaseIndex++;
	}
}

static
void CompressMessageHoleQueue(struct CReliabilityLayer *this)
{
#define queue this->hasReceivedPacketQueue

	if (queue.allocationSize <= DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE) {
		return;
	}

	if (QUEUE_SIZE(queue) * 3 <= queue.allocationSize) {
		return;
	}

	/*Queue::Compress*/
	thiscall0((void*) 0x45B6C0, &this->hasReceivedPacketQueue);
}

static
void DiscardPacket(
	struct CReliabilityLayer *this,
	struct CInternalPacket *packet)
{
	j__free(packet->pData);
	thiscall1((void*) 0x44EBF0, &this->internalPacketPool, (int) packet);
}

static
int HandleSequencedPacket(
	struct CReliabilityLayer *this,
	struct CInternalPacket *packet)
{
	char channel;
	short waitingIndex;
	short newIndex;

	if (packet->packetReliability != RELIABLE_SEQUENCED ||
		packet->packetReliability != UNRELIABLE_SEQUENCED)
	{
		return 0;
	}

	channel = packet->orderingChannel;
	if (channel >= NUMBER_OF_ORDERED_STREAMS) {
		DiscardPacket(this, packet);
		return 1;
	}

	waitingIndex = this->waitingForSequencedPacketReadIndex[channel];
	newIndex = packet->orderingIndex;
	/*isOlderSequencedPacked*/
	if (thiscall2((void*) 0x45B3D0, this, newIndex, waitingIndex)) {
		this->statistics_sequencedMessagesOutOfOrder++;
		DiscardPacket(this, packet);
		return 1;
	}
	this->statistics_sequencedMessagesInOrder++;

	/*ignoring split packet code @0x460098*/

	this->waitingForSequencedPacketReadIndex[channel] = newIndex + 1;
	Queue_Push_DWORD(&this->outputQueue, &packet);

	return 1;
}

static
void OutputNowInOrderPackets(
	struct CReliabilityLayer *this,
	struct CCircularLinkedList *orderingList,
	short *waitingForIndex)
{
	struct CInternalPacket *packet;
	struct CListNode_DWORD *node, *root;
	int wasAtLeastOneRemoved;

	wasAtLeastOneRemoved = 1;
	while (wasAtLeastOneRemoved && orderingList->list_size) {
		wasAtLeastOneRemoved = 0;
		root = node = orderingList->position = orderingList->root;
		do {
			packet = node->value;
			if (packet->orderingIndex == *waitingForIndex) {
				dprintf("popped out of order packet\n");
				Queue_Push_DWORD(&this->outputQueue, &packet);
				/*CCircularLinkedList::PopCurrent_DWORD*/
				thiscall0((void*) 0x44E750, orderingList);
				*waitingForIndex++;
				wasAtLeastOneRemoved = 1;
				break;
			}
			node = node->next;
		} while (node != root);
	}
}

static
int HandleOrdenedPacket(
	struct CReliabilityLayer *this,
	struct CInternalPacket *packet)
{
	char channel;
	short newIndex;
	short *waitingForIndex;
	struct CCircularLinkedList *orderingList;

	if (packet->packetReliability != RELIABLE_ORDENED) {
		return 0;
	}

	channel = packet->orderingChannel;
	if (channel >= NUMBER_OF_ORDERED_STREAMS) {
		DiscardPacket(this, packet);
		return 1;
	}

	newIndex = packet->orderingIndex;
	waitingForIndex = &this->waitingForOrderedPacketReadIndex[channel];
	if (newIndex != *waitingForIndex) {
		/*not yet time to handle this one*/
		dprintf("out of order packet\n");
		/*do outOfOrderLimit here*/
		this->statistics_orderedMessagesOutOfOrder++;
		/*ReliabilityLayer__AddToOrderingList*/
		thiscall1((void*) 0x45DEE0, this, (int) packet);
		return 1;
	}
	this->statistics_orderedMessagesInOrder++;

	Queue_Push_DWORD(&this->outputQueue, &packet);
	orderingList = thiscall1((void*) 0x45C4C0, this, channel);
	if (orderingList == NULL) {
		return 1;
	}
	OutputNowInOrderPackets(this, orderingList, waitingForIndex);
	return 1;
}

static
void HandlePacket(
	struct CReliabilityLayer *this,
	struct CInternalPacket *packet,
	struct CRaknetTimeNS time)
{
	short holes;

	SendAckForPacket(this, packet);
	ResetReceivedPackets(this);

	holes = packet->messageNumber - this->receivedPacketsBaseIndex;
	if (!HandlePacketHole(this, holes, time)) {
		return;
	}

	RemovePacketHolesNotReceivedInTime(this, time);
	this->statistics_messagesReceived++;
	CompressMessageHoleQueue(this);

	/*split packets are dropped, skipping 0x45FECE*/

	if (HandleSequencedPacket(this, packet)) {
		return;
	}

	if (HandleOrdenedPacket(this, packet)) {
		return;
	}

	Queue_Push_DWORD(&this->outputQueue, &packet);
}

__declspec(naked)
static
struct CInternalPacket *
__stdcall ReliabilityLayer__CreateInternalPacketFromBitStream(
	struct CReliabilityLayer *this,
	struct CBitStream *bitstream,
	struct CRaknetTimeNS time)
{
	_asm {
		pop eax
		pop ecx
		push eax
		mov eax, 0x45C000
		jmp eax
	}
}

/*ReliabilityLayer__RemovePacketFromResendListAndDeleteOlderReliableSequenced*/
__declspec(naked)
int __stdcall ReliabilityLayer__RemovePacketsConfirmedByAck(
	struct CReliabilityLayer *this,
	short ackMessageNumber,
	struct CRaknetTimeNS time)
{
	_asm {
		pop eax
		pop ecx
		push eax
		mov eax, 0x45EC40
		jmp eax
	}
}

static
void UpdateHistogramAckCount(
	struct CReliabilityLayer *this,
	struct CRaknetTimeNS timeNS,
	unsigned int ackedCounter)
{
	if (!RaknetTimeNS_IsBigger(this->histogramStartTime, timeNS)) {
		return;
	}

	if (ackedCounter == (unsigned int) -1) {
		return;
	}

	if (ackedCounter != this->histogramReceiveMarker) {
		return;
	}

	this->histogramAckCount++;
}

static
void HandleAcksForMessageRange(
	struct CReliabilityLayer *this,
	short minMessageNumber,
	short maxMessageNumber,
	struct CRaknetTimeNS timeNS)
{
	unsigned int ackedCounter;
	short msgNum;

	/*messageholelimit checked here?*/

	for (msgNum = minMessageNumber; msgNum <= maxMessageNumber; msgNum++) {
		ackedCounter = ReliabilityLayer__RemovePacketsConfirmedByAck(
			this,
			msgNum,
			timeNS);

		UpdateHistogramAckCount(this, timeNS, ackedCounter);

		/*resendList is not a pointer but
		inlined struct that hasn't been defined yet*/
		if (BPlusTree__IsEmpty(&this->resendList)) {
			this->lastAckTime.lo32 = 0;
			this->lastAckTime.hi32 = 0;
			continue;
		}

		this->lastAckTime = timeNS;
	}
}

/**
@return 0 if it did not contain valid acks
*/
static
int HandleAcks(
	struct CReliabilityLayer *this,
	struct CBitStream *bitStream,
	struct CRaknetTimeNS timeNS)
{
	struct CRangeList incomingAcksList;
	struct CRangeNode_Short *ranges;
	int amountOfAckRanges, ackRangeIdx;
	int numAcks, pos;
	short minMsg, maxMsg;

	pos = bitStream->readOffset;
	RangeList__ctor(&incomingAcksList);
	if (!RangeList__Deserialize(&incomingAcksList, bitStream)) {
		dprintf("could not deserialize ackslist\n");
		RangeList__dtor(&incomingAcksList);
		return 0;
	}
	dprintf("read acks from %d to %d\n", pos, bitStream->readOffset);

	amountOfAckRanges = incomingAcksList._._.list_size;
	if (amountOfAckRanges == 0) {
		dprintf("ackslist is empty\n");
		RangeList__dtor(&incomingAcksList);
		return 1;
	}

	numAcks = 0;
	ranges = (struct CRangeNode_Short*) incomingAcksList._._.values;
	for (ackRangeIdx = 0; ackRangeIdx < amountOfAckRanges; ackRangeIdx++) {
		minMsg = ranges[ackRangeIdx].minIndex;
		maxMsg = ranges[ackRangeIdx].maxIndex;

		if (minMsg > maxMsg) {
			dprintf("invalid ack range %hd>%hd\n", minMsg, maxMsg);
			continue;
		}

		numAcks += maxMsg - minMsg + 1;
		HandleAcksForMessageRange(this, minMsg, maxMsg, timeNS);

		/*ackslimit checked here*/
	}

	RangeList__dtor(&incomingAcksList);
	return numAcks > 0;
}

/**
seems like a nop
*/
static
void UpdateThreadedMemory(struct CReliabilityLayer *this)
{
	if (this->freeThreadedMemoryOnNextUpdate) {
		this->freeThreadedMemoryOnNextUpdate = 0;
	}
}

static
void AddBitsReceivedStatistic(struct CReliabilityLayer *this, int bits)
{
	unsigned int loValue;

	loValue = this->statistics_bitsReceivedLo32;
	this->statistics_bitsReceivedLo32 += bits;
	if (this->statistics_bitsReceivedLo32 < loValue) {
		this->statistics_bitsReceivedHi32++;
	}
}

#ifdef CALL_ORIGINAL
__declspec(naked)
#endif
int __stdcall ReliabilityLayer__HandleSocketReceiveFromConnectedPlayer(
	struct CReliabilityLayer *this,
	char *buffer,
	int length,
	struct PlayerID playerId,
	void *messageHandlerList,
	int MTUSize,
	int *ptrOutIsPacketFlood)
{
#ifdef CALL_ORIGINAL
	_asm {
		pop eax
		pop ecx
		push eax
		mov eax, 0x45F7E0
		jmp eax
	}
#else
	struct CInternalPacket *packet;
	struct CBitStream bitStream;
	struct CRaknetTimeNS timeNS;
	char hasAcks, wereAcksHandled;
	int returnValue;
	int packetCount;

	dprintf("HandleSocketReceiveFromConnectedPlayerStart\n");

	if (length == 1 || buffer == NULL) {
		dprintf("ignoring length %d buffer %p\n", length, buffer);
		// connection request, ignore
		return 1;
	}

	UpdateThreadedMemory(this);
	AddBitsReceivedStatistic(this, length * 8);

	BitStream__ctor(&bitStream, buffer, length, 0);
	RakNet__GetTimeNS(&timeNS);

	wereAcksHandled = 0;
	BitStream__Read(&bitStream, &hasAcks);
	if (hasAcks) {
		wereAcksHandled = HandleAcks(this, &bitStream, timeNS);
	}
	returnValue = wereAcksHandled;

	packetCount = 0;
	while (packet = ReliabilityLayer__CreateInternalPacketFromBitStream(
				this, &bitStream, timeNS))
	{
		returnValue |= 0x10000000;
		packetCount++;
		HandlePacket(this, packet, timeNS);
	}

	this->receivePacketCount++;
	BitStream__dtor(&bitStream);
	dprintf("HandleSocketReceiveFromConnectedPlayerEnd acks=%d packets=%d\n",
					returnValue & 0xFFFFFFF, packetCount);
	return returnValue;
#endif
}