source-engine/replay/sv_sessionblockpublisher.cpp

//========= Copyright Valve Corporation, All rights reserved. ============//
//
//=======================================================================================//

#include "sv_sessionblockpublisher.h"
#include "sv_replaycontext.h"
#include "demofile/demoformat.h"
#include "sv_recordingsession.h"
#include "sv_recordingsessionblock.h"
#include "sv_sessioninfopublisher.h"

// memdbgon must be the last include file in a .cpp file!!!
#include "tier0/memdbgon.h"

//----------------------------------------------------------------------------------------

CSessionBlockPublisher::CSessionBlockPublisher( CServerRecordingSession *pSession,
											    CSessionInfoPublisher *pSessionInfoPublisher )
:	m_pSession( pSession ),
	m_pSessionInfoPublisher( pSessionInfoPublisher )
{
	// Cache the dump interval so it can't be modified during a round - doing so would require
	// an update on all clients.
	extern ConVar replay_block_dump_interval;
	m_nDumpInterval = MAX( MIN_SERVER_DUMP_INTERVAL, replay_block_dump_interval.GetInt() );

	// Write the first block 15 or so seconds from now
	m_flLastBlockWriteTime = g_pEngine->GetHostTime();
}

CSessionBlockPublisher::~CSessionBlockPublisher()
{
}

void CSessionBlockPublisher::PublishAllSynchronous()
{
	while ( !IsDone() )
	{
		Think();
	}
}

void CSessionBlockPublisher::AbortPublish()
{
	FOR_EACH_LL( m_lstPublishingBlocks, it )
	{
		CServerRecordingSessionBlock *pCurBlock = m_lstPublishingBlocks[ it ];
		IFilePublisher *&pPublisher = pCurBlock->m_pPublisher;	// Shorthand

		if ( !pPublisher )
			continue;

		// Already done?
		if ( pPublisher->IsDone() )
			continue;

		pPublisher->AbortAndCleanup();
	}

	// Remove all blocks
	m_lstPublishingBlocks.RemoveAll();
}

void CSessionBlockPublisher::OnStartRecording()
{
}

void CSessionBlockPublisher::OnStopRecord( bool bAborting )
{
	if ( !bAborting )
	{
		// Write one final session block.
		WriteAndPublishSessionBlock();
	}
}

ReplayHandle_t CSessionBlockPublisher::GetSessionHandle() const
{
	return m_pSession->GetHandle();
}

void CSessionBlockPublisher::WriteAndPublishSessionBlock()
{
	// Make sure there is data to write
	uint8 *pSessionBuffer;
	int nSessionBufferSize;
	g_pEngine->GetSessionRecordBuffer( &pSessionBuffer, &nSessionBufferSize );	// This will get called the last client disconnects from the server - but in waiting for players state we won't have a demo buffer
	if ( !pSessionBuffer || nSessionBufferSize == 0 )
		return;

	// Create a new block
	CServerRecordingSessionBlock *pNewBlock = SV_CastBlock( SV_GetRecordingSessionBlockManager()->CreateAndGenerateHandle() );
	if ( !pNewBlock )
	{
		Warning( "Failed to create replay \"%s\"\n", pNewBlock->m_szFullFilename );
		delete pNewBlock;
		return;
	}
	
	if ( m_pSession->m_nServerStartRecordTick < 0 )
	{
		Warning( "Error: Current recording start tick was not properly setup.  Aborting block write.\n" );
		delete pNewBlock;
		return;
	}

	// Figure out what the current block is
	const int iCurrentSessionBlock = m_pSession->GetNumBlocks();

	// Add an entry to the server index with the "writing" status set
	const char *pFullFilename = Replay_va(
		"%s%s_part_%u.%s", SV_GetTmpDir(),
		SV_GetRecordingSessionManager()->GetCurrentSessionName(), iCurrentSessionBlock, BLOCK_FILE_EXTENSION
	);
	V_strcpy( pNewBlock->m_szFullFilename, pFullFilename );
	pNewBlock->m_nWriteStatus = CServerRecordingSessionBlock::WRITESTATUS_INVALID;	// Must be set here to trigger write
	pNewBlock->m_nRemoteStatus = CBaseRecordingSessionBlock::STATUS_WRITING;
	pNewBlock->m_iReconstruction = iCurrentSessionBlock;
	pNewBlock->m_hSession = m_pSession->GetHandle();

	// Match the session's lock - the block will be unlocked once recording has stopped and all publishing is complete.
	pNewBlock->SetLocked( true );

	// Commit the replay to the history manager's list
	SV_GetRecordingSessionBlockManager()->Add( pNewBlock );

	// Also store a pointer to the block in the session - NOTE: session will not attempt to free this pointer
	m_pSession->AddBlock( pNewBlock, false );

	// Cache the block temporarily while the binary block itself writes to disk - NOTE: will not attempt to free
	m_lstPublishingBlocks.AddToTail( pNewBlock );

	// Write the block now
	PublishBlock( pNewBlock );	// pNewBlock->m_nWriteStatus modified here

	IF_REPLAY_DBG( Warning( "%f: (%i) Publishing new block, %s\n", g_pEngine->GetHostTime(), iCurrentSessionBlock, pNewBlock->GetFilename() ) );
}

void CSessionBlockPublisher::GatherBlockData( uint8 *pSessionBuffer, int nSessionBufferSize, CServerRecordingSessionBlock *pBlock, unsigned char **ppSafeBlockData, int *pBlockSize )
{
	const int nHeaderSize = sizeof( demoheader_t );

	int nBlockOffset = 0;
	const int nBlockSize = nSessionBufferSize;
	int nTotalSize = nBlockSize;

	demoheader_t *pHeader = NULL;

	// If this is the first block, pass in a header to be written.  Otherwise, just write the block.
	if ( !pBlock->m_iReconstruction )
	{
		// Setup start tick in the header
		pHeader = g_pEngine->GetReplayDemoHeader();

		// Add header size
		nBlockOffset = nHeaderSize;
		nTotalSize += nHeaderSize;
	}

	// Make a copy of the block
	unsigned char *pBuffer = new unsigned char[ nTotalSize ];
	unsigned char *pBlockCopy = pBuffer + nBlockOffset;

	// Only write the header if necessary
	if ( pHeader )
	{
		demoheader_t littleEndianHeader = *pHeader;
		littleEndianHeader.playback_time = FLT_MAX;
		littleEndianHeader.playback_ticks = INT_MAX;
		littleEndianHeader.playback_frames = INT_MAX;

		// Byteswap
		ByteSwap_demoheader_t( littleEndianHeader );

		// Write header
		V_memcpy( pBuffer, &littleEndianHeader, sizeof( littleEndianHeader ) );
	}

	// Note that pBlockCopy is based on pBuffer, which was allocated with nBlockSize PLUS
	// header size - this will not overflow.
	V_memcpy( pBlockCopy, pSessionBuffer, nBlockSize );

	// Copy to "out" parameters
	*pBlockSize = nTotalSize;
	*ppSafeBlockData = pBuffer;
}

void CSessionBlockPublisher::PublishBlock( CServerRecordingSessionBlock *pBlock )
{
	uint8 *pSessionBuffer;
	int nSessionBufferSize;
	if ( !g_pEngine->GetSessionRecordBuffer( &pSessionBuffer, &nSessionBufferSize ) )
	{
		Warning( "Block publish failed!\n" );
		return;
	}

	unsigned char *pSafeBlockData;
	int nBlockSize;
	GatherBlockData( pSessionBuffer, nSessionBufferSize, pBlock, &pSafeBlockData, &nBlockSize );

	// We've got what we need and can reset the put ptr
	g_pEngine->ResetReplayRecordBuffer();

	AssertMsg( !pBlock->m_pPublisher, "No publisher should exist for this block yet!" );

	// Set status to working
	pBlock->m_nWriteStatus = CServerRecordingSessionBlock::WRITESTATUS_WORKING;

	// Get the number of bytes written
	pBlock->m_uFileSize = nBlockSize;

	// Make sure the main thread doesn't unload the block while it's being published
	pBlock->SetLocked( true );

	// Asynchronously publish to fileserver
	PublishFileParams_t params;
	params.m_pOutFilename = pBlock->m_szFullFilename;
	params.m_pSrcData = pSafeBlockData;
	params.m_nSrcSize = nBlockSize;
	params.m_pCallbackHandler = this;
	params.m_nCompressorType = COMPRESSORTYPE_BZ2;
	params.m_bHash = true;
	params.m_bFreeSrcData = true;
	params.m_bDeleteFile = false;
	params.m_pUserData = pBlock;
	pBlock->m_pPublisher = SV_PublishFile( params );
}

void CSessionBlockPublisher::OnPublishComplete( const IFilePublisher *pPublisher, void *pUserData )
{
	CServerRecordingSessionBlock *pBlock = (CServerRecordingSessionBlock *)pUserData;
	Assert( pBlock );

	// Set block status
	if ( pPublisher->GetStatus() == IFilePublisher::PUBLISHSTATUS_OK )
	{
		pBlock->m_nWriteStatus = CServerRecordingSessionBlock::WRITESTATUS_SUCCESS;
	}
	else
	{
		pBlock->m_nWriteStatus = CServerRecordingSessionBlock::WRITESTATUS_FAILED;

		// Publish failed - handle as needed
		g_pServerReplayContext->OnPublishFailed();
	}

	// Did the block compress OK?
	if ( pPublisher->Compressed() )
	{
		// Cache compressor type
		pBlock->m_nCompressorType = pPublisher->GetCompressorType();

		const int nCompressedSize = pPublisher->GetCompressedSize();
		const float flRatio = (float)pBlock->m_uFileSize / nCompressedSize;
		IF_REPLAY_DBG( Warning( "Block compression ratio: %.3f:1\n", flRatio ) );

		// Update size
		pBlock->m_uUncompressedSize = pBlock->m_uFileSize;
		pBlock->m_uFileSize = nCompressedSize;
	}

	// Get the MD5
	if ( pPublisher->Hashed() )
	{
		pPublisher->GetHash( pBlock->m_aHash );
	}

	// Now that m_nWriteStatus has been set in the block, the session info will be updated
	// accordingly the next time PublishThink() is run.

	// Mark the block as dirty since it was modified
	Assert( pBlock->m_nWriteStatus != CServerRecordingSessionBlock::WRITESTATUS_INVALID );
	SV_GetRecordingSessionBlockManager()->FlagForFlush( pBlock, false );

	IF_REPLAY_DBG( Warning( "Publish complete for block %s\n", pBlock->GetDebugName() ) );
}

void CSessionBlockPublisher::OnPublishAborted( const IFilePublisher *pPublisher )
{
	CServerRecordingSessionBlock *pBlock = FindBlockFromPublisher( pPublisher );

	// Update the block's status
	if ( pBlock )
	{
		pBlock->m_nWriteStatus = CServerRecordingSessionBlock::WRITESTATUS_FAILED;
	}

	g_pServerReplayContext->OnPublishFailed();
}

CServerRecordingSessionBlock *CSessionBlockPublisher::FindBlockFromPublisher( const IFilePublisher *pPublisher )
{
	FOR_EACH_LL( m_lstPublishingBlocks, i )
	{
		CServerRecordingSessionBlock *pCurBlock = m_lstPublishingBlocks[ i ];
		if ( pCurBlock->m_pPublisher == pPublisher )
		{
			return pCurBlock;
		}
	}

	AssertMsg( 0, "Could not find block with the given publisher!" );
	return NULL;
}

void CSessionBlockPublisher::Think()
{
	// NOTE: This member function gets called even if replay is disabled.  This is intentional.

	VPROF_BUDGET( "CSessionBlockPublisher::Think", VPROF_BUDGETGROUP_REPLAY );

	PublishThink();
}

void CSessionBlockPublisher::PublishThink()
{
	AssertMsg( m_pSession->IsLocked(), "The session isn't locked, which means blocks can be being deleted and will probably cause a crash." );

	// Go through all currently publishing blocks and free/think
	FOR_EACH_LL( m_lstPublishingBlocks, it )
	{
		CServerRecordingSessionBlock *pCurBlock = m_lstPublishingBlocks[ it ];
		IFilePublisher *&pPublisher = pCurBlock->m_pPublisher;	// Shorthand

		if ( !pPublisher )
			continue;

		// If the publisher's done, free it
		if ( pPublisher->IsDone() )
		{
			delete pPublisher;
			pPublisher = NULL;
		}
		else
		{
			// Let the publisher think
			pPublisher->Think();
		}
	}

	// Write a new session block out right now?
	float flHostTime = g_pEngine->GetHostTime();
	if ( m_flLastBlockWriteTime != 0.0f &&
		 flHostTime - m_flLastBlockWriteTime >= m_nDumpInterval &&
		 m_pSession->m_bRecording )
	{
		Assert( m_nDumpInterval > 0 );

		// Write it
		WriteAndPublishSessionBlock();

		// Update the time
		m_flLastBlockWriteTime = flHostTime;
	}

	// Check status of any replays that are being written
	bool bUpdateSessionInfo  = false;
	for( int it = m_lstPublishingBlocks.Head(); it != m_lstPublishingBlocks.InvalidIndex(); )
	{
		CServerRecordingSessionBlock *pCurBlock = m_lstPublishingBlocks[ it ];

		// Updated when write status is set to success or failure
		int nPendingRequestStatus = CBaseRecordingSessionBlock::STATUS_INVALID;

		// If set to anything besides InvalidIndex(), it will be removed from the list
		int itRemove = m_lstPublishingBlocks.InvalidIndex();
		bool bWriteBlockInfoToDisk = false;

		switch ( pCurBlock->m_nWriteStatus )
		{
		case CServerRecordingSessionBlock::WRITESTATUS_INVALID:
			AssertMsg( 0, "Why is m_nWriteStatus WRITESTATUS_INVALID here?" );
			break;

		case CServerRecordingSessionBlock::WRITESTATUS_WORKING:	// Do nothing if still writing
			break;

		case CServerRecordingSessionBlock::WRITESTATUS_SUCCESS:
			IF_REPLAY_DBG2( Warning( "  Block %i marked as succeeded.\n", pCurBlock->m_iReconstruction ) );
			pCurBlock->m_nRemoteStatus = CBaseRecordingSessionBlock::STATUS_READYFORDOWNLOAD;
			nPendingRequestStatus = pCurBlock->m_nRemoteStatus;
			bWriteBlockInfoToDisk = true;
			itRemove = it;
			break;

		case CServerRecordingSessionBlock::WRITESTATUS_FAILED:
		default:	// Error?
			IF_REPLAY_DBG2( Warning( "  Block %i marked as failed.\n", pCurBlock->m_iReconstruction ) );
			pCurBlock->m_nRemoteStatus = CBaseRecordingSessionBlock::STATUS_ERROR;
			pCurBlock->m_nHttpError = CBaseRecordingSessionBlock::ERROR_WRITEFAILED;
			nPendingRequestStatus = pCurBlock->m_nRemoteStatus;
			bWriteBlockInfoToDisk = true;
			itRemove = it;

			// TODO: Retry
		}

		if ( bWriteBlockInfoToDisk )
		{
			// Save the master index file
			Assert( pCurBlock->m_nWriteStatus != CServerRecordingSessionBlock::WRITESTATUS_INVALID );
			SV_GetRecordingSessionBlockManager()->FlagForFlush( pCurBlock, false );
		}

		// Find the owning session
		Assert( pCurBlock->m_hSession == m_pSession->GetHandle() );

		// Refresh session info file
		if ( nPendingRequestStatus != CBaseRecordingSessionBlock::STATUS_INVALID )
		{
			// Update it after this loop
			bUpdateSessionInfo = true;
		}

		// Update iterator
		it = m_lstPublishingBlocks.Next( it );

		// Remove?
		if ( itRemove != m_lstPublishingBlocks.InvalidIndex() )
		{
			IF_REPLAY_DBG( Warning( "Removing block %i from publisher\n", pCurBlock->m_iReconstruction ) );
			// Free/clear publisher
			delete pCurBlock->m_pPublisher;
			pCurBlock->m_pPublisher = NULL;

			// Removes from the list but doesn't free, since any pointer here points to a block somewhere
			m_lstPublishingBlocks.Unlink( itRemove );
		}
	}

	// Publish session info file now if it isn't already publishing
	if ( bUpdateSessionInfo )
	{
		m_pSessionInfoPublisher->Publish();
	}
}

bool CSessionBlockPublisher::IsDone() const
{
	return m_lstPublishingBlocks.Count() == 0;
}

#ifdef _DEBUG
void CSessionBlockPublisher::Validate()
{
	FOR_EACH_LL( m_lstPublishingBlocks, i )
	{
		CServerRecordingSessionBlock *pCurBlock = m_lstPublishingBlocks[ i ];
		Assert( pCurBlock->m_nRemoteStatus == CBaseRecordingSessionBlock::STATUS_READYFORDOWNLOAD );
	}
}
#endif

//----------------------------------------------------------------------------------------