source-engine/utils/vmpi/vmpi_distribute_work.cpp

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

#include <windows.h>
#include "vmpi.h"
#include "vmpi_distribute_work.h"
#include "tier0/platform.h"
#include "tier0/dbg.h"
#include "utlvector.h"
#include "utllinkedlist.h"
#include "vmpi_dispatch.h"
#include "pacifier.h"
#include "vstdlib/random.h"
#include "mathlib/mathlib.h"
#include "threadhelpers.h"
#include "threads.h"
#include "tier1/strtools.h"
#include "tier1/utlmap.h"
#include "tier1/smartptr.h"
#include "tier0/icommandline.h"
#include "cmdlib.h"
#include "vmpi_distribute_tracker.h"
#include "vmpi_distribute_work_internal.h"


// To catch some bugs with 32-bit vs 64-bit and etc.
#pragma warning( default : 4244 )
#pragma warning( default : 4305 )
#pragma warning( default : 4267 )
#pragma warning( default : 4311 )
#pragma warning( default : 4312 )


const int MAX_DW_CALLS = 255;
extern bool g_bSetThreadPriorities;


// Subpacket IDs owned by DistributeWork.
#define DW_SUBPACKETID_MASTER_READY		0
#define DW_SUBPACKETID_WORKER_READY		1
#define DW_SUBPACKETID_MASTER_FINISHED	2
#define DW_SUBPACKETID_WU_RESULTS		4
#define DW_SUBPACKETID_WU_STARTED		6	// A worker telling the master it has started processing a work unit.
// NOTE VMPI_DISTRIBUTE_WORK_EXTRA_SUBPACKET_BASE is where the IWorkUnitDistributorX classes start their subpackets.


IWorkUnitDistributorCallbacks *g_pDistributeWorkCallbacks = NULL;


static CDSInfo g_DSInfo;
static unsigned short g_iCurDSInfo = (unsigned short)-1;	// This is incremented each time DistributeWork is called.
static int g_iMasterFinishedDistributeWorkCall = -1;		// The worker stores this to know which DistributeWork() calls the master has finished.
static int g_iMasterReadyForDistributeWorkCall = -1;

// This is only valid if we're a worker and if the worker currently has threads chewing on work units.
static CDSInfo *g_pCurWorkerThreadsInfo = NULL;

static CUtlVector<uint64> g_wuCountByProcess;
static uint64 g_totalWUCountByProcess[512];

static uint64 g_nWUs;				// How many work units there were this time around.
static uint64 g_nCompletedWUs;		// How many work units completed.
static uint64 g_nDuplicatedWUs;	// How many times a worker sent results for a work unit that was already completed.

// Set to true if Error() is called and we want to exit early. vrad and vvis check for this in their
// thread functions, so the workers quit early when the master is done rather than finishing up
// potentially time-consuming work units they're working on.
bool g_bVMPIEarlyExit = false;

static bool g_bMasterDistributingWork = false;

static IWorkUnitDistributorWorker *g_pCurDistributorWorker = NULL;
static IWorkUnitDistributorMaster *g_pCurDistributorMaster = NULL;

// For the stats database.
WUIndexType g_ThreadWUs[4] = { ~0ull, ~0ull, ~0ull, ~0ull };

class CMasterWorkUnitCompletedList
{
public:
	CUtlVector<WUIndexType> m_CompletedWUs;
};
static CCriticalSectionData<CMasterWorkUnitCompletedList> g_MasterWorkUnitCompletedList;


int SortByWUCount( const void *elem1, const void *elem2 )
{
	uint64 a = g_wuCountByProcess[ *((const int*)elem1) ];
	uint64 b = g_wuCountByProcess[ *((const int*)elem2) ];
	if ( a < b )
		return 1;
	else if ( a == b )
		return 0;
	else	   
		return -1;
}


void PrepareDistributeWorkHeader( MessageBuffer *pBuf, unsigned char cSubpacketID )
{
	char cPacketID[2] = { g_DSInfo.m_cPacketID, cSubpacketID };
	pBuf->write( cPacketID, 2 );
	pBuf->write( &g_iCurDSInfo, sizeof( g_iCurDSInfo ) );
}


void ShowMPIStats( 
	double flTimeSpent,
	unsigned long nBytesSent, 
	unsigned long nBytesReceived, 
	unsigned long nMessagesSent, 
	unsigned long nMessagesReceived )
{
	double flKSent = (nBytesSent + 511) / 1024;
	double flKRecv = (nBytesReceived + 511) / 1024;

	bool bShowOutput = VMPI_IsParamUsed( mpi_ShowDistributeWorkStats );

	bool bOldSuppress = g_bSuppressPrintfOutput;
	g_bSuppressPrintfOutput = !bShowOutput;

		Msg( "\n\n--------------------------------------------------------------\n");
		Msg( "Total Time       : %.2f\n", flTimeSpent );
		Msg( "Total Bytes Sent : %dk (%.2fk/sec, %d messages)\n", (int)flKSent, flKSent / flTimeSpent, nMessagesSent );
		Msg( "Total Bytes Recv : %dk (%.2fk/sec, %d messages)\n", (int)flKRecv, flKRecv / flTimeSpent, nMessagesReceived );
		if ( g_bMPIMaster )
		{
			Msg( "Duplicated WUs   : %I64u (%.1f%%)\n", g_nDuplicatedWUs, (float)g_nDuplicatedWUs * 100.0f / g_nWUs );

			Msg( "\nWU count by proc:\n" );

			int nProcs = VMPI_GetCurrentNumberOfConnections();
			
			CUtlVector<int> sortedProcs;
			sortedProcs.SetSize( nProcs );
			for ( int i=0; i < nProcs; i++ )
				sortedProcs[i] = i;

			qsort( sortedProcs.Base(), nProcs, sizeof( int ), SortByWUCount );

			for ( int i=0; i < nProcs; i++ )
			{
				const char *pMachineName = VMPI_GetMachineName( sortedProcs[i] );
				Msg( "%s", pMachineName );
				
				char formatStr[512];
				Q_snprintf( formatStr, sizeof( formatStr ), "%%%ds %I64u\n", 30 - strlen( pMachineName ), g_wuCountByProcess[ sortedProcs[i] ] );
				Msg( formatStr, ":" );
			}
		}
		Msg( "--------------------------------------------------------------\n\n ");

	g_bSuppressPrintfOutput = bOldSuppress;
}


void VMPI_DistributeWork_DisconnectHandler( int procID, const char *pReason )
{
	if ( g_bMasterDistributingWork )
	{
		// Show the disconnect in the database but not on the screen.
		bool bOldSuppress = g_bSuppressPrintfOutput;
		g_bSuppressPrintfOutput = true;
		Msg( "VMPI_DistributeWork_DisconnectHandler( %d )\n", procID );
		g_bSuppressPrintfOutput = bOldSuppress;
	
		// Redistribute the WUs from this guy's partition to another worker.
		g_pCurDistributorMaster->DisconnectHandler( procID );
	}
}


uint64 VMPI_GetNumWorkUnitsCompleted( int iProc )
{
	Assert( iProc >= 0 && iProc <= ARRAYSIZE( g_totalWUCountByProcess ) );
	return g_totalWUCountByProcess[iProc];
}


void HandleWorkUnitCompleted( CDSInfo *pInfo, int iSource, WUIndexType iWorkUnit, MessageBuffer *pBuf )
{
	VMPITracker_WorkUnitCompleted( ( int ) iWorkUnit, iSource );
	
	if ( g_pCurDistributorMaster->HandleWorkUnitResults( iWorkUnit ) )
	{
		if ( g_iVMPIVerboseLevel >= 1 )
			Msg( "-" );

		++ g_nCompletedWUs;
		++ g_wuCountByProcess[iSource];
		++ g_totalWUCountByProcess[iSource];

		// Let the master process the incoming WU data.
		if ( pBuf )
		{
			pInfo->m_MasterInfo.m_ReceiveFn( iWorkUnit, pBuf, iSource );
		}

		UpdatePacifier( float( g_nCompletedWUs ) / pInfo->m_nWorkUnits );
	}
	else
	{
		// Ignore it if we already got the results for this work unit.
		++ g_nDuplicatedWUs;
		if ( g_iVMPIVerboseLevel >= 1 )
			Msg( "*" );
	}
}


bool DistributeWorkDispatch( MessageBuffer *pBuf, int iSource, int iPacketID )
{											  
	unsigned short iCurDistributeWorkCall = *((unsigned short*)&pBuf->data[2]);
	if ( iCurDistributeWorkCall >= MAX_DW_CALLS )
		Error( "Got an invalid DistributeWork packet (id: %d, sub: %d) (iCurDW: %d).", pBuf->data[0], pBuf->data[1], iCurDistributeWorkCall );

	CDSInfo *pInfo = &g_DSInfo;
		
	pBuf->setOffset( 4 );
	
	switch ( pBuf->data[1] )
	{
		case DW_SUBPACKETID_MASTER_READY:
		{
			g_iMasterReadyForDistributeWorkCall = iCurDistributeWorkCall;
			return true;
		}

		case DW_SUBPACKETID_WORKER_READY:
		{
			if ( iCurDistributeWorkCall > g_iCurDSInfo || !g_bMPIMaster )
				Error( "State incorrect on master for DW_SUBPACKETID_WORKER_READY packet from %s.", VMPI_GetMachineName( iSource ) );

			if ( iCurDistributeWorkCall == g_iCurDSInfo )
			{
				// Ok, give this guy some WUs.
				if ( g_pCurDistributorMaster )
					g_pCurDistributorMaster->OnWorkerReady( iSource );
			}

			return true;
		}

		case DW_SUBPACKETID_MASTER_FINISHED:
		{
			g_iMasterFinishedDistributeWorkCall = iCurDistributeWorkCall;
			return true;
		}
		
		// Worker sends this to tell the master it has started on a work unit.
		case DW_SUBPACKETID_WU_STARTED:
		{
			if ( iCurDistributeWorkCall != g_iCurDSInfo )
				return true;

			WUIndexType iWU;
			pBuf->read( &iWU, sizeof( iWU ) );
			VMPITracker_WorkUnitStarted( ( int ) iWU, iSource );
			return true;
		}
		

		case DW_SUBPACKETID_WU_RESULTS:
		{
			// We only care about work results for the iteration we're in.
			if ( iCurDistributeWorkCall != g_iCurDSInfo )
				return true;

			WUIndexType iWorkUnit;
			pBuf->read( &iWorkUnit, sizeof( iWorkUnit ) );
			if ( iWorkUnit >= pInfo->m_nWorkUnits )
			{
				Error( "DistributeWork: got an invalid work unit index (%I64u for WU count of %I64u).", iWorkUnit, pInfo->m_nWorkUnits );
			}

			HandleWorkUnitCompleted( pInfo, iSource, iWorkUnit, pBuf );
			return true;
		}

		default:
		{
			if ( g_pCurDistributorMaster )
				return g_pCurDistributorMaster->HandlePacket( pBuf, iSource, iCurDistributeWorkCall != g_iCurDSInfo );
			else if ( g_pCurDistributorWorker )
				return g_pCurDistributorWorker->HandlePacket( pBuf, iSource, iCurDistributeWorkCall != g_iCurDSInfo );
			else
				return false;
		}
	}
}


EWorkUnitDistributor VMPI_GetActiveWorkUnitDistributor()
{
	if ( VMPI_IsParamUsed( mpi_UseSDKDistributor ) )
	{
		Msg( "Found %s.\n", VMPI_GetParamString( mpi_UseSDKDistributor ) );
		return k_eWorkUnitDistributor_SDK;
	}
	else if ( VMPI_IsParamUsed( mpi_UseDefaultDistributor ) )
	{
		Msg( "Found %s.\n", VMPI_GetParamString( mpi_UseDefaultDistributor ) );
		return k_eWorkUnitDistributor_Default;
	}
	else
	{
		if ( VMPI_IsSDKMode() )
			return k_eWorkUnitDistributor_SDK;
		else
			return k_eWorkUnitDistributor_Default;
	}
}


void PreDistributeWorkSync( CDSInfo *pInfo )
{
	if ( g_bMPIMaster )
	{
		// Send a message telling all the workers we're ready to go on this DistributeWork call.
		MessageBuffer mb;
		PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_MASTER_READY );
		VMPI_SendData( mb.data, mb.getLen(), VMPI_PERSISTENT );
	}
	else
	{
		if ( g_iVMPIVerboseLevel >= 1 )
			Msg( "PreDistributeWorkSync: waiting for master\n" );

		// Wait for the master's message saying it's ready to go.
		while ( g_iMasterReadyForDistributeWorkCall < g_iCurDSInfo )
		{
			VMPI_DispatchNextMessage();
		}

		if ( g_iVMPIVerboseLevel >= 1 )
			Msg( "PreDistributeWorkSync: master ready\n" );

		// Now tell the master we're ready.
		MessageBuffer mb;
		PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_WORKER_READY );
		VMPI_SendData( mb.data, mb.getLen(), VMPI_MASTER_ID );
	}
}


void DistributeWork_Master( CDSInfo *pInfo, ProcessWorkUnitFn processFn, ReceiveWorkUnitFn receiveFn )
{
	pInfo->m_WorkerInfo.m_pProcessFn = processFn;
	pInfo->m_MasterInfo.m_ReceiveFn = receiveFn;

	VMPITracker_Start( (int) pInfo->m_nWorkUnits );

	g_bMasterDistributingWork = true;
	g_pCurDistributorMaster->DistributeWork_Master( pInfo );
	g_bMasterDistributingWork = false;

	VMPITracker_End();
		
	// Tell all workers to move on.
	MessageBuffer mb;
	PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_MASTER_FINISHED );
	VMPI_SendData( mb.data, mb.getLen(), VMPI_PERSISTENT );

	// Clear the master's local completed work unit list.
	CMasterWorkUnitCompletedList *pList = g_MasterWorkUnitCompletedList.Lock();
	pList->m_CompletedWUs.RemoveAll();	
	g_MasterWorkUnitCompletedList.Unlock();
}


void NotifyLocalMasterCompletedWorkUnit( WUIndexType iWorkUnit )
{
	CMasterWorkUnitCompletedList *pList = g_MasterWorkUnitCompletedList.Lock();
	pList->m_CompletedWUs.AddToTail( iWorkUnit );
	g_MasterWorkUnitCompletedList.Unlock();
}

void CheckLocalMasterCompletedWorkUnits()
{
	CMasterWorkUnitCompletedList *pList = g_MasterWorkUnitCompletedList.Lock();
	
	for ( int i=0; i < pList->m_CompletedWUs.Count(); i++ )
	{
		HandleWorkUnitCompleted( &g_DSInfo, 0, pList->m_CompletedWUs[i], NULL );
	}
	pList->m_CompletedWUs.RemoveAll();	
		
	g_MasterWorkUnitCompletedList.Unlock();
}


void TellMasterThatWorkerStartedAWorkUnit( MessageBuffer &mb, CDSInfo *pInfo, WUIndexType iWU )
{
	mb.setLen( 0 );
	PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_WU_STARTED );
	mb.write( &iWU, sizeof( iWU ) );
	VMPI_SendData( mb.data, mb.getLen(), VMPI_MASTER_ID, k_eVMPISendFlags_GroupPackets );
}


void VMPI_WorkerThread( int iThread, void *pUserData )
{
	CDSInfo *pInfo = (CDSInfo*)pUserData;
	CWorkerInfo *pWorkerInfo = &pInfo->m_WorkerInfo;

	
	// Get our index for running work units
	uint64 idxRunningWorkUnit = (uint64) iThread;
	{
		CCriticalSectionLock csLock( &pWorkerInfo->m_WorkUnitsRunningCS );
		csLock.Lock();
		pWorkerInfo->m_WorkUnitsRunning.ExpandWindow( idxRunningWorkUnit, ~0ull );
		csLock.Unlock();
	}


	MessageBuffer mb;
	PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_WU_RESULTS );

	MessageBuffer mbStartedWorkUnit;	// Special messagebuffer used to tell the master when we started a work unit.

	while ( g_iMasterFinishedDistributeWorkCall < g_iCurDSInfo && !g_bVMPIEarlyExit )
	{
		WUIndexType iWU;

		// Quit out when there are no more work units.
		if ( !g_pCurDistributorWorker->GetNextWorkUnit( &iWU ) )
		{
			// Wait until there are some WUs to do. This should probably use event handles.
			VMPI_Sleep( 10 );
			continue;
		}
			
		CCriticalSectionLock csLock( &pWorkerInfo->m_WorkUnitsRunningCS );
		csLock.Lock();

			// Check if this WU is not running
			WUIndexType const *pBegin = &pWorkerInfo->m_WorkUnitsRunning.Get( 0ull ), *pEnd = pBegin + pWorkerInfo->m_WorkUnitsRunning.PastVisibleIndex();
			WUIndexType const *pRunningWu = GenericFind( pBegin, pEnd, iWU );
			if ( pRunningWu != pEnd )
				continue;

			// We are running it
			pWorkerInfo->m_WorkUnitsRunning.Get( idxRunningWorkUnit ) = iWU;
	
		csLock.Unlock();


		// Process this WU and send the results to the master.
		mb.setLen( 4 );
		mb.write( &iWU, sizeof( iWU ) );

		// Set the current WU for the stats database.
		if ( iThread >= 0 && iThread < 4 )
		{
			g_ThreadWUs[iThread] = iWU;
		}

		// Tell the master we're starting on this WU.
		TellMasterThatWorkerStartedAWorkUnit( mbStartedWorkUnit, pInfo, iWU );
		

		pWorkerInfo->m_pProcessFn( iThread, iWU, &mb );
		g_pCurDistributorWorker->NoteLocalWorkUnitCompleted( iWU );
		
		VMPI_SendData( mb.data, mb.getLen(), VMPI_MASTER_ID, /*k_eVMPISendFlags_GroupPackets*/0 );

		// Flush grouped packets every once in a while.
		//VMPI_FlushGroupedPackets( 1000 );
	}

	if ( g_iVMPIVerboseLevel >= 1 )
		Msg( "Worker thread exiting.\n" );
}


void DistributeWork_Worker( CDSInfo *pInfo, ProcessWorkUnitFn processFn )
{
	if ( g_iVMPIVerboseLevel >= 1 )
		Msg( "VMPI_DistributeWork call %d started.\n", g_iCurDSInfo+1 );

	CWorkerInfo *pWorkerInfo = &pInfo->m_WorkerInfo;
	pWorkerInfo->m_pProcessFn = processFn;

	g_pCurWorkerThreadsInfo = pInfo;
	g_pCurDistributorWorker->Init( pInfo );

	// Start a couple threads to do the work.
	RunThreads_Start( VMPI_WorkerThread, pInfo, g_bSetThreadPriorities ? k_eRunThreadsPriority_Idle : k_eRunThreadsPriority_UseGlobalState );
	if ( g_iVMPIVerboseLevel >= 1 )
		Msg( "RunThreads_Start finished successfully.\n" );

	if ( VMPI_IsSDKMode() )
	{
		Msg( "\n" );
		while ( g_iMasterFinishedDistributeWorkCall < g_iCurDSInfo )
		{
			VMPI_DispatchNextMessage( 300 );
			
			Msg( "\rThreads status: " );
			for ( int i=0; i < ARRAYSIZE( g_ThreadWUs ); i++ )
			{
				if ( g_ThreadWUs[i] != ~0ull )
					Msg( "%d: WU %5d  ", i, (int)g_ThreadWUs[i] );
			}

			VMPI_FlushGroupedPackets();
		}
		Msg( "\n" );
	}
	else
	{
		while ( g_iMasterFinishedDistributeWorkCall < g_iCurDSInfo )
		{
			VMPI_DispatchNextMessage();
		}
	}
	

	// Close the threads.
	g_pCurWorkerThreadsInfo = NULL;
	RunThreads_End();

	if ( g_iVMPIVerboseLevel >= 1 )
		Msg( "VMPI_DistributeWork call %d finished.\n", g_iCurDSInfo+1 );
}


// This is called by VMPI_Finalize in case it's shutting down due to an Error() call.
// In this case, it's important that the worker threads here are shut down before VMPI shuts
// down its sockets.
void DistributeWork_Cancel()
{
	if ( g_pCurWorkerThreadsInfo )
	{
		Msg( "\nDistributeWork_Cancel saves the day!\n" );
		g_pCurWorkerThreadsInfo->m_bMasterFinished = true;
		g_bVMPIEarlyExit = true;
		RunThreads_End();
	}
}


// Returns time it took to finish the work.
double DistributeWork( 
	uint64 nWorkUnits,				// how many work units to dole out
	char cPacketID,
	ProcessWorkUnitFn processFn,	// workers implement this to process a work unit and send results back
	ReceiveWorkUnitFn receiveFn		// the master implements this to receive a work unit
	)
{
	++g_iCurDSInfo;

	if ( g_iCurDSInfo == 0 )
	{
		// Register our disconnect handler so we can deal with it if clients bail out.
		if ( g_bMPIMaster )
		{
			VMPI_AddDisconnectHandler( VMPI_DistributeWork_DisconnectHandler );
		}
	}
	else if ( g_iCurDSInfo >= MAX_DW_CALLS )
	{
		Error( "DistributeWork: called more than %d times.\n", MAX_DW_CALLS );
	}

	CDSInfo *pInfo = &g_DSInfo;
	
	pInfo->m_cPacketID = cPacketID;
	pInfo->m_nWorkUnits = nWorkUnits;

	// Make all the workers wait until the master is ready.
	PreDistributeWorkSync( pInfo );

	g_nWUs = nWorkUnits;
	g_nCompletedWUs = 0ull;
	g_nDuplicatedWUs = 0ull;

	// Setup stats info.
	double flMPIStartTime = Plat_FloatTime();
	g_wuCountByProcess.SetCount( 512 );
	memset( g_wuCountByProcess.Base(), 0, sizeof( int ) * g_wuCountByProcess.Count() );
	
	unsigned long nBytesSentStart = g_nBytesSent;
	unsigned long nBytesReceivedStart = g_nBytesReceived;
	unsigned long nMessagesSentStart = g_nMessagesSent;
	unsigned long nMessagesReceivedStart = g_nMessagesReceived;

	EWorkUnitDistributor eWorkUnitDistributor = VMPI_GetActiveWorkUnitDistributor();
	if ( g_bMPIMaster )
	{
		Assert( !g_pCurDistributorMaster );
		g_pCurDistributorMaster = ( eWorkUnitDistributor == k_eWorkUnitDistributor_SDK ? CreateWUDistributor_SDKMaster() : CreateWUDistributor_DefaultMaster() );
		
		DistributeWork_Master( pInfo, processFn, receiveFn );
		
		g_pCurDistributorMaster->Release();
		g_pCurDistributorMaster = NULL;
	}
	else 
	{
		Assert( !g_pCurDistributorWorker );
		g_pCurDistributorWorker = ( eWorkUnitDistributor == k_eWorkUnitDistributor_SDK ? CreateWUDistributor_SDKWorker() : CreateWUDistributor_DefaultWorker() );

		DistributeWork_Worker( pInfo, processFn );
		
		g_pCurDistributorWorker->Release();
		g_pCurDistributorWorker = NULL;
	}

	double flTimeSpent = Plat_FloatTime() - flMPIStartTime;
	ShowMPIStats( 
		flTimeSpent,
		g_nBytesSent - nBytesSentStart,
		g_nBytesReceived - nBytesReceivedStart,
		g_nMessagesSent - nMessagesSentStart,
		g_nMessagesReceived - nMessagesReceivedStart
		);

	// Mark that the threads aren't working on anything at the moment.
	for ( int i=0; i < ARRAYSIZE( g_ThreadWUs ); i++ )
		g_ThreadWUs[i] = ~0ull;
	
	return flTimeSpent;
}