RP412/MUNGA/SPOOLER.cpp

#include "munga.h"
#pragma hdrstop

#include "spooler.h"
#include "filestrm.h"

//#############################################################################
//##############################   SpoolFile   ################################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolFile::SpoolFile(
	void *stream_start,
	size_t stream_size,
	size_t initial_offset
):
	MemoryStream(stream_start, stream_size, initial_offset)
{
	spoolState = Empty;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolFile::SpoolFile(SpoolFile& spool):
	MemoryStream(
		spool.streamStart,
		spool.streamSize,
		spool.GetBytesUsed()
	)
{
	spoolState = spool.spoolState;
	Verify(spoolState == Playing);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolFile::~SpoolFile()
{
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	SpoolFile::SaveAs(const char *file_name)
{
	FileStream output(file_name, True);
	size_t length = GetBytesUsed();
	output << length;
	Rewind();
	output.WriteBytes(GetPointer(), length);
	output.Close();
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	SpoolFile::Read(const char *file_name)
{
	FileStream input(file_name);
	if (!input.IsFileOpened())
	{
		spoolState = Empty;
		return;
	}

	//
	//-------------------------------
	// Figure out how big the file is
	//-------------------------------
	//
	size_t spool_size = 0;
	input >> spool_size;
	if (spool_size > GetBytesRemaining() || !spool_size)
	{
		spoolState = Empty;
	}

	//
	//-----------
	// Read it in
	//-----------
	//
	else
	{
		input.ReadBytes(streamStart, spool_size);
		streamSize = spool_size;
		Rewind();
		spoolState = Stored;
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	SpoolFile::SpoolPacket(NetworkPacket *packet)
{
	Check(this);
	Check_Pointer(packet);

	//
	//-------------------------------------------------------------------------
	// Check to see if we are out of spool space.  For now, if it happens, just
	// die
	//-------------------------------------------------------------------------
	//
	int length =
		packet->messageData.messageLength + sizeof(NetworkPacketHeader);
	int remaining = GetBytesRemaining();
	if (remaining < length)
	{
		DEBUG_STREAM << "Error: Spool file ran out of memory!\n" << std::flush;
		PostQuitMessage(AbortExitCodeID);
	}

	//
	//-----------------------------------------------------
	// Copy into the memory stream, and advance the pointer
	//-----------------------------------------------------
	//
	Mem_Copy(GetPointer(), packet, length, remaining);
	AdvancePointer(length);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
NetworkPacket*
	SpoolFile::NextPacket()
{
	Check(this);

	//
	//--------------------------------------------------------------------------
	// Check to see if anything is left in the spool.  If only a partial message
	// remains, die from an error
	//-------------------------------------------------------------------------
	//
	if (!GetBytesRemaining())
	{
		return NULL;
	}
	NetworkPacket *packet = (NetworkPacket*)GetPointer();
	int length =
		packet->messageData.messageLength + sizeof(NetworkPacketHeader);
	if (GetBytesRemaining() < length)
	{
		DEBUG_STREAM << "Error: Partial packet in spool file!\n" << std::flush;
		PostQuitMessage(AbortExitCodeID);
		return NULL;
	}

	//
	//-----------------------------------------------------
	// Copy into the memory stream, and advance the pointer
	//-----------------------------------------------------
	//
	AdvancePointer(length);
	return (NetworkPacket*)GetPointer();
}

//#############################################################################
//##############################  SpoolRecorder  ##############################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
// How much memory to give a recording, in megabytes.
//
// The review build's SPOOL_SIZE is 6MB, which was a sensible arcade
// number and is a silly desktop one: a full grid sends on the order of
// 17KB a second, so six megabytes is about six minutes and a long race
// would hit the end of it. A hundred megabytes is roughly an hour and a
// half and costs nothing on any machine that can run this.
//
static size_t
	RecordSizeBytes()
{
	static size_t cached = 0;

	if (cached == 0)
	{
		const char *setting = getenv("RP412RECORDSIZE");
		int megabytes = (setting != NULL) ? atoi(setting) : 100;

		if (megabytes < 1)		{ megabytes = 1; }
		if (megabytes > 512)	{ megabytes = 512; }
		cached = (size_t) megabytes * 1024 * 1024;
	}
	return cached;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolRecorder::SpoolRecorder():
	buffer(NULL),
	spool(NULL),
	bufferSize(0),
	armed(False),
	full(False),
	packetsRecorded(0)
{
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolRecorder::~SpoolRecorder()
{
	Disarm();
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
Logical
	SpoolRecorder::Arm()
{
	Check(this);

	if (armed)
	{
		return True;
	}

	bufferSize = RecordSizeBytes();
	buffer = new char[bufferSize];
	if (buffer == NULL)
	{
		DEBUG_STREAM << "Record: could not reserve "
			<< (bufferSize / (1024 * 1024))
			<< "MB - recording disabled for this race\n" << std::flush;
		bufferSize = 0;
		return False;
	}

	spool = new SpoolFile(buffer, bufferSize);
	spool->spoolState = SpoolFile::Spooling;
	armed = True;
	full = False;
	packetsRecorded = 0;

	DEBUG_STREAM << "Record: armed, " << (bufferSize / (1024 * 1024))
		<< "MB (RP412RECORDSIZE)\n" << std::flush;
	return True;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	SpoolRecorder::Record(const NetworkPacket *packet)
{
	Check(this);

	if (!armed || full || packet == NULL)
	{
		return;
	}
	Check_Pointer(spool);

	int length =
		packet->messageData.messageLength + sizeof(NetworkPacketHeader);

	//
	// Ask before writing. SpoolFile::SpoolPacket answers a full buffer by
	// quitting the process, which would end the race this is recording -
	// so the recorder never lets it get that far, and stops instead.
	//
	if ((int) spool->GetBytesRemaining() < length)
	{
		full = True;
		DEBUG_STREAM << "Record: buffer full after " << packetsRecorded
			<< " packets - the race continues, the recording stops here."
			<< " Raise RP412RECORDSIZE to keep more.\n" << std::flush;
		return;
	}

	//
	// Where the copy is about to land, taken BEFORE the write so the
	// timestamp can be applied to the copy afterwards. The live packet is
	// never modified: its timeStamp is the sender's sampling moment, which
	// Simulation::ReadUpdateRecord hands to RP412NETCLOCK and from there to
	// the dead reckoner's projection. Restamping it in place - which is
	// what the review spooler does, harmlessly, having nothing else to
	// serve - would put arrival jitter straight into where remote pods are
	// drawn.
	//
	NetworkPacket *copy = (NetworkPacket*) spool->GetPointer();

	spool->SpoolPacket((NetworkPacket*) packet);

	//
	// Playback paces from these, and packets from different senders carry
	// different clock origins, so the spool needs them all on one clock:
	// ours, at the moment of arrival.
	//
	copy->timeStamp = Now();
	++packetsRecorded;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	SpoolRecorder::Save()
{
	Check(this);

	//
	// Never armed means not one packet arrived all race.
	//
	// Recording captures the packets this station RECEIVES, and a race
	// with no other machines in it sends and receives nothing at all -
	// L4NetworkManager::ExclusiveBroadcast walks the remote hosts, and a
	// solo race has none. So there is nothing to keep, and the reason is
	// worth saying rather than leaving an empty folder to be puzzled over.
	//
	// The same gap is why a RACER's recording is not the whole race: its
	// own pod is simulated locally and never arrives as a packet. A Live
	// Cam races nothing, so every pod reaches it over the wire, which
	// makes it the only station that hears the lot.
	//
	if (!armed || spool == NULL)
	{
		DEBUG_STREAM << "Record: nothing to write - no packets were received"
			<< " this race. Recording keeps what arrives over the network,"
			<< " so a single-player race has nothing to keep.\n" << std::flush;
		return;
	}
	if (packetsRecorded == 0)
	{
		DEBUG_STREAM << "Record: nothing captured, no file written\n"
			<< std::flush;
		return;
	}

	//
	// The review build writes here and so does this, so one folder holds
	// every spool however it was made and the playback build finds them
	// all in the same place.
	//
	CreateDirectoryA("SPOOLS", NULL);

	struct tm newtime;
	__int64 ltime;

	_time64(&ltime);
	_gmtime64_s(&newtime, &ltime);

	char filename[MAX_PATH];

	sprintf(
		filename,
		"SPOOLS\\%.4i_%.2i_%.2i_%.2i%.2i%.2i.spl",
		newtime.tm_year + 1900, newtime.tm_mon + 1, newtime.tm_mday,
		newtime.tm_hour, newtime.tm_min, newtime.tm_sec
	);

	//
	// Read the size BEFORE saving: SaveAs rewinds the stream when it is
	// done, so asking afterwards reports nothing written at all - which
	// is what the first recording's log said, next to a 2.8MB file.
	//
	size_t written = spool->GetBytesUsed();

	spool->SaveAs(filename);
	CopyFileA(filename, "last.spl", FALSE);

	DEBUG_STREAM << "Record: wrote " << filename << " - "
		<< packetsRecorded << " packets, " << (written / 1024) << "KB"
		<< (full ? " (truncated - buffer filled)" : "")
		<< "\n" << std::flush;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	SpoolRecorder::Disarm()
{
	Check(this);

	if (spool != NULL)
	{
		delete spool;
		spool = NULL;
	}
	if (buffer != NULL)
	{
		delete [] buffer;
		buffer = NULL;
	}
	bufferSize = 0;
	armed = False;
	full = False;
	packetsRecorded = 0;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolRecorder *
	SpoolRecorder_Get()
{
	static SpoolRecorder recorder;

	return &recorder;
}

//#############################################################################
//#################    MissionReviewApplicationManager    #####################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
MissionReviewApplicationManager::MissionReviewApplicationManager(
	HINSTANCE hInstance,
	HWND hWnd,
	Scalar frame_rate,
	int spool_count,
	size_t spool_size
):
	ApplicationManager(hInstance, hWnd, frame_rate)
{
	//
	//--------------------------------------------------------
	// Allocate the space for the spool files and buffer table
	//--------------------------------------------------------
	//
	spoolCount = spool_count;
	spoolSize = spool_size;
	spoolBuffers = new char* [spoolCount];
	Register_Pointer(spoolBuffers);
	spoolFiles = new SpoolFile* [spoolCount];
	Register_Pointer(spoolFiles);

	//
	//--------------------------------------------------------
	// Allocate each spool buffer, and initial the spool files to none
	//--------------------------------------------------------
	//
	for (int i=0; i<spoolCount; ++i)
	{
		spoolBuffers[i] = new char[spoolSize];
		Register_Pointer(spoolBuffers[i]);
		spoolFiles[i] = new SpoolFile(spoolBuffers[i], spoolSize);
		Register_Object(spoolFiles[i]);
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
MissionReviewApplicationManager::~MissionReviewApplicationManager()
{
	for (int i=0; i<spoolCount; ++i)
	{
		if (spoolFiles[i])
		{
			Unregister_Object(spoolFiles[i]);
			delete spoolFiles[i];
		}
		Unregister_Pointer(spoolBuffers[i]);
		delete spoolBuffers[i];
	}
	Unregister_Pointer(spoolFiles);
	delete spoolFiles;
	Unregister_Pointer(spoolBuffers);
	delete spoolBuffers;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolFile*
	MissionReviewApplicationManager::GetEmptySpoolFile()
{
	for (int i=0; i<spoolCount; ++i)
	{
		if (spoolFiles[i]->spoolState == SpoolFile::Empty)
		{
			spoolFiles[i]->spoolState = SpoolFile::Spooling;
			return spoolFiles[i];
		}
	}

	Fail("No spool files available!");
	return NULL;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
SpoolFile*
	MissionReviewApplicationManager::GetStoredSpoolFile(CString spoolFileName)
{
	for (int i=0; i<spoolCount; ++i)
	{
		if (spoolFiles[i]->spoolState == SpoolFile::Empty)
		{
			spoolFiles[i]->Read(spoolFileName);
			if (spoolFiles[i]->spoolState == SpoolFile::Stored)
			{
				spoolFiles[i]->spoolState = SpoolFile::Playing;
				return spoolFiles[i];
			}
			return NULL;
		}
	}
	return NULL;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	MissionReviewApplicationManager::StoreSpoolFile(SpoolFile *spool_file)
{
	for (int i=0; i<spoolCount; ++i)
	{
		if (spoolFiles[i] == spool_file)
		{
			Verify(spoolFiles[i]->spoolState == SpoolFile::Spooling);
			spoolFiles[i]->spoolState = SpoolFile::Stored;
			
			struct tm newtime;
			__int64 ltime;

			// get local time
			_time64(&ltime);

			// get UTC time
			_gmtime64_s(&newtime, &ltime);

			char filename[2056];
			sprintf(filename, "SPOOLS\\%.4i_%.2i_%.2i_%.2i%.2i%.2i.spl", newtime.tm_year + 1900, newtime.tm_mon + 1, newtime.tm_mday, newtime.tm_hour, newtime.tm_min, newtime.tm_sec);

			spoolFiles[i]->SaveAs(filename);

			// now copy the new file over to last.spl
			CopyFileA(filename, "last.spl", FALSE);
			
			spoolFiles[i]->Rewind();
			return;
		}
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	MissionReviewApplicationManager::ReleaseSpoolFile(SpoolFile *spool_file)
{
	for (int i=0; i<spoolCount; ++i)
	{
		if (spoolFiles[i] == spool_file)
		{
			Verify(spoolFiles[i]->spoolState != SpoolFile::Spooling);
			spoolFiles[i]->spoolState = SpoolFile::Empty;
			spoolFiles[i]->Rewind();
			return;
		}
	}
}