BotoX
/
hl2_src-leak-2017
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
hl2_src-leak-2017/src/utils/vmpi/vmpi_distribute_work.cpp

629 lines
18 KiB
C++
Raw Permalink Blame History

//========= 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;
}
Reference in New Issue View Git Blame Copy Permalink