RP412/RP_L4/RPL4SPOOLSTATS.cpp

#include "..\munga_l4\mungal4.h"
#pragma hdrstop

#include <stdio.h>
#include "..\munga\app.h"
#include "..\munga\network.h"
#include "..\munga\interest.h"
#include "..\munga\entity.h"
#include "..\munga\simulate.h"

//########################################################################
// rpl4opt -spoolstats <file.spl>  -  offline latency analysis of a spool
//
// Every Live Cam / pod recording already contains a complete one-way-
// delay dataset and nothing ever read it that way: the recorder restamps
// each packet's header with LOCAL ARRIVAL time (SPOOLER.cpp), while the
// UpdateRecords inside still carry the SENDER's sim-grid timestamp
// untouched. Per sender,
//
//     sample = arrival - senderStamp = clockOffset + oneWayDelay
//
// and the running minimum of sample converges on the offset from above -
// the same estimator RP412NETCLOCK runs live (SIMULATE.cpp). Subtracting
// it back out gives one-way delay ABOVE THE MINIMUM per packet: jitter
// and queueing, which is what the dead reckoner actually fights. The
// minimum itself (the flat latency floor) is unobservable without a
// synchronized clock and is absorbed into the offset - every figure
// printed here is delay above that floor, not absolute delay.
//
// Also decomposed per entity: gaps on the SENDER timeline (send pacing)
// versus gaps on the ARRIVAL timeline (network + receive-loop jitter).
// The send side of the tick story lives in the first, the delivery side
// in the second, and no live counter can separate them.
//
// This must live in the game exe, not RPL4TOOL: the engine compiles
// /Zp1 and RPL4TOOL deliberately does not, so the tool would misread
// every struct in the file.
//
// Caveats printed with the output:
//  - arrival stamps are quantized by the receive drain (post-render,
//    so roughly frame cadence)
//  - sender stamps sit on the sender's fixed-step grid (20 ms at 50 Hz)
//  - a racer's spool never contains its own pod; a Live Cam hears all
//########################################################################

namespace
{
	enum
	{
		statsMaxHosts = 32,
		statsMaxEntities = 128,
		statsWindow = 128,		// mirrors netClockWindow (SIMULATE.cpp)
		statsBuckets = 16
	};

	//
	// log2-millisecond histogram, same shape as Entity::NetRaceStats -
	// bucket 0 is sub-millisecond, bucket b covers [2^(b-1), 2^b) ms.
	//
	struct Histogram
	{
		unsigned long	buckets[statsBuckets];
		unsigned long	total;
		long			worstMs;

		void
			Count(long ms)
		{
			if (ms > worstMs)
			{
				worstMs = ms;
			}
			unsigned long value = (ms > 0) ? (unsigned long) ms : 0;
			int bucket = 0;
			while (value != 0 && bucket < statsBuckets - 1)
			{
				value >>= 1;
				++bucket;
			}
			buckets[bucket]++;
			total++;
		}

		unsigned long
			PercentileMs(unsigned long percent) const
		{
			if (total == 0)
			{
				return 0;
			}
			unsigned long target = (total * percent + 99) / 100;
			unsigned long seen = 0;
			for (int b = 0; b < statsBuckets; ++b)
			{
				seen += buckets[b];
				if (seen >= target)
				{
					return (b == 0) ? 0 : (1UL << (b - 1));
				}
			}
			return 1UL << (statsBuckets - 2);
		}
	};

	//
	// Per sender host: packet accounting and the netclock replica.
	//
	struct HostStats
	{
		HostID			host;
		Logical			inUse;
		unsigned long	packetCount;
		unsigned long	updateMessageCount;
		unsigned long	byteCount;

		Logical			settled;
		long			offsetTicks;
		long			windowMinTicks;
		int				windowCount;

		Histogram		owdAboveMin;
	};

	//
	// Per entity: one entry per replicated entity seen in the spool,
	// with the sender-vs-arrival gap decomposition.
	//
	struct EntityStats
	{
		Logical			inUse;
		int				entityHost;			// EntityID host part
		int				entityLocal;		// EntityID local part
		HostID			fromHost;
		unsigned long	updateCount;
		long			lastSenderTicks;
		long			lastArrivalTicks;
		Logical			haveLast;
		Histogram		senderGaps;
		Histogram		arrivalGaps;
	};

	HostStats	gHosts[statsMaxHosts];
	EntityStats	gEntities[statsMaxEntities];

	HostStats *
		FindHost(HostID host)
	{
		HostStats *free_slot = NULL;
		for (int i = 0; i < statsMaxHosts; ++i)
		{
			if (gHosts[i].inUse)
			{
				if (gHosts[i].host == host)
				{
					return &gHosts[i];
				}
			}
			else if (free_slot == NULL)
			{
				free_slot = &gHosts[i];
			}
		}
		if (free_slot != NULL)
		{
			memset(free_slot, 0, sizeof(*free_slot));
			free_slot->inUse = True;
			free_slot->host = host;
		}
		return free_slot;
	}

	EntityStats *
		FindEntity(int entity_host, int entity_local, HostID from_host)
	{
		EntityStats *free_slot = NULL;
		for (int i = 0; i < statsMaxEntities; ++i)
		{
			if (gEntities[i].inUse)
			{
				if (gEntities[i].entityHost == entity_host &&
					gEntities[i].entityLocal == entity_local)
				{
					return &gEntities[i];
				}
			}
			else if (free_slot == NULL)
			{
				free_slot = &gEntities[i];
			}
		}
		if (free_slot != NULL)
		{
			memset(free_slot, 0, sizeof(*free_slot));
			free_slot->inUse = True;
			free_slot->entityHost = entity_host;
			free_slot->entityLocal = entity_local;
			free_slot->fromHost = from_host;
		}
		return free_slot;
	}

	//
	// The netclock replica: returns delay-above-minimum for one sample,
	// updating the host's running-minimum offset exactly the way the
	// live estimator does (immediate adopt of a shorter path, window
	// close follows drift).
	//
	long
		OwdAboveMin(HostStats *host, long sample)
	{
		if (!host->settled)
		{
			host->settled = True;
			host->offsetTicks = sample;
			host->windowMinTicks = sample;
			host->windowCount = 0;
			return 0;
		}
		if (sample < host->windowMinTicks)
		{
			host->windowMinTicks = sample;
		}
		if (sample < host->offsetTicks)
		{
			host->offsetTicks = sample;
		}
		if (++host->windowCount >= statsWindow)
		{
			host->offsetTicks = host->windowMinTicks;
			host->windowMinTicks = sample;
			host->windowCount = 0;
		}
		long above = sample - host->offsetTicks;
		return (above > 0) ? above : 0;
	}

	//
	// Is there a plausible packet chain at this offset? The same check
	// playback runs against a wrong egg, run over a few packets in a
	// row - used to find where the host-pair table ends without needing
	// the egg at all.
	//
	Logical
		PacketChainPlausible(
			const unsigned char *buffer,
			size_t size,
			size_t offset)
	{
		int checked = 0;
		while (checked < 5)
		{
			if (offset == size)
			{
				return (checked > 0) ? True : False;
			}
			if (offset + sizeof(NetworkPacket) > size)
			{
				// a truncated tail is normal (the buffer filled mid-write)
				return (checked > 0) ? True : False;
			}
			const NetworkPacket *packet = (const NetworkPacket *) (buffer + offset);
			size_t length = packet->messageData.messageLength;
			if (length < sizeof(Receiver::Message) || length > 65536)
			{
				return False;
			}
			if (offset + length + sizeof(NetworkPacketHeader) > size)
			{
				return (checked > 0) ? True : False;
			}
			offset += length + sizeof(NetworkPacketHeader);
			++checked;
		}
		return True;
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
// Returns a process exit code: 0 analysis written, 1 something wrong.
//
int
	RPL4SpoolStats_Run(const char *path)
{
	printf("SpoolStats: %s\n", path);

	FILE *file = fopen(path, "rb");
	if (file == NULL)
	{
		printf("SpoolStats: cannot open the file\n");
		return 1;
	}
	fseek(file, 0, SEEK_END);
	long file_size = ftell(file);
	fseek(file, 0, SEEK_SET);
	if (file_size <= 0)
	{
		printf("SpoolStats: empty file\n");
		fclose(file);
		return 1;
	}
	unsigned char *buffer = (unsigned char *) malloc((size_t) file_size);
	if (buffer == NULL ||
		fread(buffer, 1, (size_t) file_size, file) != (size_t) file_size)
	{
		printf("SpoolStats: could not read %ld bytes\n", file_size);
		fclose(file);
		free(buffer);
		return 1;
	}
	fclose(file);
	size_t size = (size_t) file_size;

	//
	// Skip the header: ApplicationID + resource major revision, then one
	// (Logical remote, HostID) pair per host named in the egg. The pair
	// count is not recorded, so find it by trying each count and asking
	// whether a plausible packet chain starts where the table would end.
	//
	size_t table_start = sizeof(ApplicationID) + sizeof(int);
	size_t pair_size = sizeof(Logical) + sizeof(HostID);
	size_t packets_start = 0;
	int host_pairs = -1;

	for (int try_count = 0; try_count <= 16; ++try_count)
	{
		size_t offset = table_start + (size_t) try_count * pair_size;
		if (offset > size)
		{
			break;
		}
		//
		// Each pair must read as a Logical 0/1 to be believable.
		//
		Logical pairs_believable = True;
		for (int p = 0; p < try_count; ++p)
		{
			long remote = *(const long *) (buffer + table_start + (size_t) p * pair_size);
			if (remote != 0 && remote != 1)
			{
				pairs_believable = False;
				break;
			}
		}
		if (pairs_believable && PacketChainPlausible(buffer, size, offset))
		{
			packets_start = offset;
			host_pairs = try_count;
			break;
		}
	}
	if (host_pairs < 0)
	{
		printf("SpoolStats: no plausible packet chain found - not a spool?\n");
		free(buffer);
		return 1;
	}
	printf("SpoolStats: header holds %d host pair(s), %lu bytes of packets\n",
		host_pairs, (unsigned long) (size - packets_start));

	memset(gHosts, 0, sizeof(gHosts));
	memset(gEntities, 0, sizeof(gEntities));

	char csv_path[512];
	_snprintf(csv_path, sizeof(csv_path) - 1, "%s.netstats.csv", path);
	csv_path[sizeof(csv_path) - 1] = 0;
	FILE *csv = fopen(csv_path, "w");
	if (csv != NULL)
	{
		fprintf(csv, "fromHost,entityHost,entityLocal,senderMs,arrivalMs,owdAboveMinMs\n");
	}

	//
	// The packet walk.
	//
	size_t offset = packets_start;
	unsigned long packet_total = 0;
	long first_arrival = 0;
	long last_arrival = 0;
	Logical have_arrival = False;

	while (offset + sizeof(NetworkPacket) <= size)
	{
		NetworkPacket *packet = (NetworkPacket *) (buffer + offset);
		size_t length = packet->messageData.messageLength;
		if (length < sizeof(Receiver::Message) || length > 65536 ||
			offset + length + sizeof(NetworkPacketHeader) > size)
		{
			break;		// truncated tail (buffer filled mid-write)
		}

		long arrival = packet->timeStamp.ticks;
		if (!have_arrival)
		{
			first_arrival = arrival;
			have_arrival = True;
		}
		last_arrival = arrival;

		HostStats *host = FindHost(packet->fromHost);
		if (host != NULL)
		{
			host->packetCount++;
			host->byteCount += (unsigned long) (length + sizeof(NetworkPacketHeader));
		}
		++packet_total;

		//
		// Entity update messages carry the sender sim-grid stamps.
		//
		if (host != NULL &&
			packet->clientID == NetworkClient::InterestManagerClientID &&
			packet->messageData.messageID ==
				InterestManager::EntityUpdateReplicantsMessageID &&
			length >= sizeof(Entity::Message) + sizeof(Simulation::UpdateRecord))
		{
			host->updateMessageCount++;

			Entity::Message *entity_message =
				(Entity::Message *) &packet->messageData;

			//
			// EntityID is (hostID, localID), both 32-bit under /Zp1;
			// its members are private and the analyzer only needs the
			// two numbers, so read them as the pair they are on disk.
			//
			int entity_id[2];
			memcpy(entity_id, &entity_message->entityID, sizeof(entity_id));

			//
			// First record's stamp speaks for the message - every record
			// in it was written at the same lastPerformance.
			//
			Simulation::UpdateRecord *record =
				(Simulation::UpdateRecord *)
					((unsigned char *) entity_message + sizeof(Entity::Message));
			long sender_ticks = record->timeStamp.ticks;
			long sample = arrival - sender_ticks;
			long above = OwdAboveMin(host, sample);
			host->owdAboveMin.Count(above);

			EntityStats *entity =
				FindEntity(entity_id[0], entity_id[1], packet->fromHost);
			if (entity != NULL)
			{
				entity->updateCount++;
				if (entity->haveLast)
				{
					entity->senderGaps.Count(sender_ticks - entity->lastSenderTicks);
					entity->arrivalGaps.Count(arrival - entity->lastArrivalTicks);
				}
				entity->lastSenderTicks = sender_ticks;
				entity->lastArrivalTicks = arrival;
				entity->haveLast = True;
			}

			if (csv != NULL)
			{
				fprintf(csv, "%d,%d,%d,%ld,%ld,%ld\n",
					(int) packet->fromHost, entity_id[0], entity_id[1],
					sender_ticks, arrival, above);
			}
		}

		offset += length + sizeof(NetworkPacketHeader);
	}

	//
	// The report.
	//
	printf("SpoolStats: %lu packets spanning %.1f s of arrivals\n",
		packet_total,
		have_arrival ? (last_arrival - first_arrival) / 1000.0f : 0.0f);
	printf("SpoolStats: figures are delay ABOVE the per-host minimum - the\n");
	printf("SpoolStats: flat latency floor is absorbed into the clock offset\n");
	printf("SpoolStats: and is unobservable without a synchronized clock.\n");
	printf("SpoolStats: arrivals are quantized by the receive drain (~frame\n");
	printf("SpoolStats: cadence); sender stamps sit on the 20 ms sim grid.\n");

	int h;
	for (h = 0; h < statsMaxHosts; ++h)
	{
		HostStats *host = &gHosts[h];
		if (!host->inUse)
		{
			continue;
		}
		printf("SpoolStats: host %d: %lu packets %lu B, %lu update msgs, "
			"owd-above-min med/p95 ~%lu/~%lu ms, worst %ld ms, offset %ld ms\n",
			(int) host->host,
			host->packetCount,
			host->byteCount,
			host->updateMessageCount,
			host->owdAboveMin.PercentileMs(50),
			host->owdAboveMin.PercentileMs(95),
			host->owdAboveMin.worstMs,
			host->offsetTicks);
	}
	for (int e = 0; e < statsMaxEntities; ++e)
	{
		EntityStats *entity = &gEntities[e];
		if (!entity->inUse || entity->updateCount < 2)
		{
			continue;
		}
		printf("SpoolStats: entity %d:%d (host %d): %lu updates, "
			"sender gaps med/p95 ~%lu/~%lu ms, arrival gaps med/p95 ~%lu/~%lu ms, "
			"worst %ld/%ld ms\n",
			entity->entityHost, entity->entityLocal, (int) entity->fromHost,
			entity->updateCount,
			entity->senderGaps.PercentileMs(50),
			entity->senderGaps.PercentileMs(95),
			entity->arrivalGaps.PercentileMs(50),
			entity->arrivalGaps.PercentileMs(95),
			entity->senderGaps.worstMs,
			entity->arrivalGaps.worstMs);
	}

	if (csv != NULL)
	{
		fclose(csv);
		printf("SpoolStats: per-update rows written to %s\n", csv_path);
	}
	free(buffer);
	return 0;
}