700 lines
20 KiB
C++
700 lines
20 KiB
C++
//========= Copyright Valve Corporation, All rights reserved. ============//
|
|
//
|
|
// Purpose:
|
|
//
|
|
//=============================================================================
|
|
|
|
#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"
|
|
|
|
|
|
|
|
#define DW_SUBPACKETID_SHUFFLE (VMPI_DISTRIBUTE_WORK_EXTRA_SUBPACKET_BASE+0)
|
|
#define DW_SUBPACKETID_REQUEST_SHUFFLE (VMPI_DISTRIBUTE_WORK_EXTRA_SUBPACKET_BASE+1)
|
|
#define DW_SUBPACKETID_WUS_COMPLETED_LIST (VMPI_DISTRIBUTE_WORK_EXTRA_SUBPACKET_BASE+2)
|
|
|
|
|
|
|
|
// This is a pretty simple iterator. Basically, it holds a matrix of numbers.
|
|
// Each row is assigned to a worker, and the worker just walks through his row.
|
|
//
|
|
// When a worker reaches the end of his row, it gets a little trickier.
|
|
// They'll start doing their neighbor's row
|
|
// starting at the back and continue on. At about this time, the master should reshuffle the
|
|
// remaining work units to evenly distribute them amongst the workers.
|
|
class CWorkUnitWalker
|
|
{
|
|
public:
|
|
CWorkUnitWalker()
|
|
{
|
|
m_nWorkUnits = 0;
|
|
}
|
|
|
|
// This is all that's needed for it to start assigning work units.
|
|
void Init( WUIndexType matrixWidth, WUIndexType matrixHeight, WUIndexType nWorkUnits )
|
|
{
|
|
m_nWorkUnits = nWorkUnits;
|
|
m_MatrixWidth = matrixWidth;
|
|
m_MatrixHeight = matrixHeight;
|
|
Assert( m_MatrixWidth * m_MatrixHeight >= nWorkUnits );
|
|
|
|
m_WorkerInfos.RemoveAll();
|
|
m_WorkerInfos.EnsureCount( m_MatrixHeight );
|
|
for ( int i=0; i < m_MatrixHeight; i++ )
|
|
{
|
|
m_WorkerInfos[i].m_iStartWorkUnit = matrixWidth * i;
|
|
m_WorkerInfos[i].m_iWorkUnitOffset = 0;
|
|
}
|
|
}
|
|
|
|
// This is the main function of the shuffler
|
|
bool GetNextWorkUnit( int iWorker, WUIndexType *pWUIndex, bool *bWorkerFinishedHisColumn )
|
|
{
|
|
if ( iWorker < 0 || iWorker >= m_WorkerInfos.Count() )
|
|
{
|
|
Assert( false );
|
|
return false;
|
|
}
|
|
|
|
// If this worker has walked through all the work units, then he's done.
|
|
CWorkerInfo *pWorker = &m_WorkerInfos[iWorker];
|
|
if ( pWorker->m_iWorkUnitOffset >= m_nWorkUnits )
|
|
return false;
|
|
|
|
// If we've gone past the end of our work unit list, then we start at the BACK of the other rows of work units
|
|
// in the hopes that we won't collide with the guy working there. We also should tell the master to reshuffle.
|
|
WUIndexType iWorkUnitOffset = pWorker->m_iWorkUnitOffset;
|
|
if ( iWorkUnitOffset >= m_MatrixWidth )
|
|
{
|
|
WUIndexType xOffset = iWorkUnitOffset % m_MatrixWidth;
|
|
WUIndexType yOffset = iWorkUnitOffset / m_MatrixWidth;
|
|
xOffset = m_MatrixWidth - xOffset - 1;
|
|
iWorkUnitOffset = yOffset * m_MatrixWidth + xOffset;
|
|
*bWorkerFinishedHisColumn = true;
|
|
}
|
|
else
|
|
{
|
|
*bWorkerFinishedHisColumn = false;
|
|
}
|
|
|
|
*pWUIndex = (pWorker->m_iStartWorkUnit + iWorkUnitOffset) % m_nWorkUnits;
|
|
++pWorker->m_iWorkUnitOffset;
|
|
return true;
|
|
}
|
|
|
|
|
|
private:
|
|
class CWorkerInfo
|
|
{
|
|
public:
|
|
WUIndexType m_iStartWorkUnit;
|
|
WUIndexType m_iWorkUnitOffset; // Which work unit in my list of work units am I working on?
|
|
};
|
|
|
|
WUIndexType m_nWorkUnits;
|
|
WUIndexType m_MatrixWidth;
|
|
WUIndexType m_MatrixHeight;
|
|
CUtlVector<CWorkerInfo> m_WorkerInfos;
|
|
};
|
|
|
|
|
|
class IShuffleRequester
|
|
{
|
|
public:
|
|
virtual void RequestShuffle() = 0;
|
|
};
|
|
|
|
|
|
// This is updated every time the master decides to reshuffle.
|
|
// In-between shuffles, you can call NoteWorkUnitCompleted when a work unit is completed
|
|
// and it'll avoid returning that work unit from GetNextWorkUnit again, but it WON'T
|
|
class CShuffledWorkUnitWalker
|
|
{
|
|
public:
|
|
void Init( WUIndexType nWorkUnits, IShuffleRequester *pRequester )
|
|
{
|
|
m_iLastShuffleRequest = 0;
|
|
m_iCurShuffle = 1;
|
|
m_flLastShuffleTime = Plat_FloatTime();
|
|
m_pShuffleRequester = pRequester;
|
|
|
|
int nBytes = PAD_NUMBER( nWorkUnits, 8 ) / 8;
|
|
m_CompletedWUBits.SetSize( nBytes );
|
|
m_LocalCompletedWUBits.SetSize( nBytes );
|
|
for ( WUIndexType i=0; i < m_CompletedWUBits.Count(); i++ )
|
|
m_LocalCompletedWUBits[i] = m_CompletedWUBits[i] = 0;
|
|
|
|
// Setup our list of work units remaining.
|
|
for ( WUIndexType iWU=0; iWU < nWorkUnits; iWU++ )
|
|
{
|
|
// Note: we're making an assumption here that if we add entries to a CUtlLinkedList in ascending order, their indices
|
|
// will be ascending 1-by-1 as well. If that assumption breaks, we can create an extra array here to map WU indices to the linked list indices.
|
|
WUIndexType index = m_WorkUnitsRemaining.AddToTail( iWU );
|
|
if ( index != iWU )
|
|
{
|
|
Error( "CShuffledWorkUnitWalker: assumption on CUtlLinkedList indexing failed.\n" );
|
|
}
|
|
}
|
|
}
|
|
|
|
void Shuffle( int nWorkers )
|
|
{
|
|
if ( nWorkers == 0 )
|
|
return;
|
|
|
|
++m_iCurShuffle;
|
|
m_flLastShuffleTime = Plat_FloatTime();
|
|
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
|
|
m_WorkUnitsMap.RemoveAll();
|
|
m_WorkUnitsMap.EnsureCount( m_WorkUnitsRemaining.Count() );
|
|
|
|
// Here's the shuffle. The CWorkUnitWalker is going to walk each worker through its own group from 0-W,
|
|
// and our job is to interleave it so when worker 0 goes [0,1,2] and worker 1 goes [100,101,102], they're actually
|
|
// doing [0,N,2N] and [1,N+1,2N+1] where N=# of workers.
|
|
|
|
// The grid is RxW long, and R*W is >= nWorkUnits.
|
|
// R = # units per worker = width of the matrix
|
|
// W = # workers = height of the matrix
|
|
WUIndexType matrixHeight = nWorkers;
|
|
WUIndexType matrixWidth = m_WorkUnitsRemaining.Count() / matrixHeight;
|
|
if ( (m_WorkUnitsRemaining.Count() % matrixHeight) != 0 )
|
|
++matrixWidth;
|
|
|
|
Assert( matrixWidth * matrixHeight >= m_WorkUnitsRemaining.Count() );
|
|
|
|
WUIndexType iWorkUnit = 0;
|
|
FOR_EACH_LL( m_WorkUnitsRemaining, i )
|
|
{
|
|
WUIndexType xCoord = iWorkUnit / matrixHeight;
|
|
WUIndexType yCoord = iWorkUnit % matrixHeight;
|
|
Assert( xCoord < matrixWidth );
|
|
Assert( yCoord < matrixHeight );
|
|
|
|
m_WorkUnitsMap[yCoord*matrixWidth+xCoord] = m_WorkUnitsRemaining[i];
|
|
++iWorkUnit;
|
|
}
|
|
|
|
m_Walker.Init( matrixWidth, matrixHeight, m_WorkUnitsRemaining.Count() );
|
|
}
|
|
|
|
// Threadsafe.
|
|
bool Thread_IsWorkUnitCompleted( WUIndexType iWU )
|
|
{
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
|
|
byte val = m_CompletedWUBits[iWU >> 3] & (1 << (iWU & 7));
|
|
return (val != 0);
|
|
}
|
|
|
|
WUIndexType Thread_NumWorkUnitsRemaining()
|
|
{
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
|
|
return m_WorkUnitsRemaining.Count();
|
|
}
|
|
|
|
bool Thread_GetNextWorkUnit( int iWorker, WUIndexType *pWUIndex )
|
|
{
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
|
|
while ( 1 )
|
|
{
|
|
WUIndexType iUnmappedWorkUnit;
|
|
bool bWorkerFinishedHisColumn;
|
|
if ( !m_Walker.GetNextWorkUnit( iWorker, &iUnmappedWorkUnit, &bWorkerFinishedHisColumn ) )
|
|
return false;
|
|
|
|
// If we've done all the work units assigned to us in the last shuffle, then request a reshuffle.
|
|
if ( bWorkerFinishedHisColumn )
|
|
HandleWorkerFinishedColumn();
|
|
|
|
// Check the pending list.
|
|
*pWUIndex = m_WorkUnitsMap[iUnmappedWorkUnit];
|
|
byte bIsCompleted = m_CompletedWUBits[*pWUIndex >> 3] & (1 << (*pWUIndex & 7));
|
|
byte bIsCompletedLocally = m_LocalCompletedWUBits[*pWUIndex >> 3] & (1 << (*pWUIndex & 7));
|
|
if ( !bIsCompleted && !bIsCompletedLocally )
|
|
return true;
|
|
}
|
|
}
|
|
|
|
void HandleWorkerFinishedColumn()
|
|
{
|
|
if ( m_iLastShuffleRequest != m_iCurShuffle )
|
|
{
|
|
double flCurTime = Plat_FloatTime();
|
|
if ( flCurTime - m_flLastShuffleTime > 2.0f )
|
|
{
|
|
m_pShuffleRequester->RequestShuffle();
|
|
m_iLastShuffleRequest = m_iCurShuffle;
|
|
}
|
|
}
|
|
}
|
|
|
|
void Thread_NoteWorkUnitCompleted( WUIndexType iWU )
|
|
{
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
|
|
byte val = m_CompletedWUBits[iWU >> 3] & (1 << (iWU & 7));
|
|
if ( val == 0 )
|
|
{
|
|
m_WorkUnitsRemaining.Remove( iWU );
|
|
m_CompletedWUBits[iWU >> 3] |= (1 << (iWU & 7));
|
|
}
|
|
}
|
|
|
|
void Thread_NoteLocalWorkUnitCompleted( WUIndexType iWU )
|
|
{
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
m_LocalCompletedWUBits[iWU >> 3] |= (1 << (iWU & 7));
|
|
}
|
|
|
|
CRC32_t GetShuffleCRC()
|
|
{
|
|
#ifdef _DEBUG
|
|
static bool bCalcShuffleCRC = true;
|
|
#else
|
|
static bool bCalcShuffleCRC = VMPI_IsParamUsed( mpi_CalcShuffleCRC );
|
|
#endif
|
|
if ( bCalcShuffleCRC )
|
|
{
|
|
CCriticalSectionLock csLock( &m_CS );
|
|
csLock.Lock();
|
|
|
|
CRC32_t ret;
|
|
CRC32_Init( &ret );
|
|
|
|
FOR_EACH_LL( m_WorkUnitsRemaining, i )
|
|
{
|
|
WUIndexType iWorkUnit = m_WorkUnitsRemaining[i];
|
|
CRC32_ProcessBuffer( &ret, &iWorkUnit, sizeof( iWorkUnit ) );
|
|
}
|
|
|
|
for ( int i=0; i < m_WorkUnitsMap.Count(); i++ )
|
|
{
|
|
WUIndexType iWorkUnit = m_WorkUnitsMap[i];
|
|
CRC32_ProcessBuffer( &ret, &iWorkUnit, sizeof( iWorkUnit ) );
|
|
}
|
|
|
|
CRC32_Final( &ret );
|
|
return ret;
|
|
}
|
|
else
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private:
|
|
// These are PENDING WU completions until we call Shuffle() again, at which point we actually reorder the list
|
|
// based on the completed WUs.
|
|
CUtlVector<byte> m_CompletedWUBits; // Bit vector of completed WUs.
|
|
CUtlLinkedList<WUIndexType, WUIndexType> m_WorkUnitsRemaining;
|
|
CUtlVector<WUIndexType> m_WorkUnitsMap; // Maps the 0-N indices in the CWorkUnitWalker to the list of remaining work units.
|
|
|
|
// Helps us avoid some duplicates that happen during shuffling if we've completed some WUs and sent them
|
|
// to the master, but the master hasn't included them in the DW_SUBPACKETID_WUS_COMPLETED_LIST yet.
|
|
CUtlVector<byte> m_LocalCompletedWUBits; // Bit vector of completed WUs.
|
|
|
|
// Used to control how frequently we request a reshuffle.
|
|
unsigned int m_iCurShuffle;
|
|
unsigned int m_iLastShuffleRequest; // The index of the shuffle we last requested a reshuffle on (don't request a reshuffle on the same one).
|
|
double m_flLastShuffleTime;
|
|
IShuffleRequester *m_pShuffleRequester;
|
|
|
|
CWorkUnitWalker m_Walker;
|
|
CCriticalSection m_CS;
|
|
};
|
|
|
|
|
|
|
|
class CDistributor_SDKMaster : public IWorkUnitDistributorMaster, public IShuffleRequester
|
|
{
|
|
public:
|
|
virtual void Release()
|
|
{
|
|
delete this;
|
|
}
|
|
|
|
static void Master_WorkerThread_Static( int iThread, void *pUserData )
|
|
{
|
|
((CDistributor_SDKMaster*)pUserData)->Master_WorkerThread( iThread );
|
|
}
|
|
|
|
void Master_WorkerThread( int iThread )
|
|
{
|
|
while ( m_WorkUnitWalker.Thread_NumWorkUnitsRemaining() > 0 && !g_bVMPIEarlyExit )
|
|
{
|
|
WUIndexType iWU;
|
|
if ( !m_WorkUnitWalker.Thread_GetNextWorkUnit( 0, &iWU ) )
|
|
{
|
|
// Wait until there are some WUs to do.
|
|
VMPI_Sleep( 10 );
|
|
continue;
|
|
}
|
|
|
|
// Do this work unit.
|
|
m_WorkUnitWalker.Thread_NoteLocalWorkUnitCompleted( iWU ); // We do this before it's completed because otherwise if a Shuffle() occurs,
|
|
// the other thread might happen to pickup this work unit and we don't want that.
|
|
m_pInfo->m_WorkerInfo.m_pProcessFn( iThread, iWU, NULL );
|
|
NotifyLocalMasterCompletedWorkUnit( iWU );
|
|
}
|
|
}
|
|
|
|
virtual void DistributeWork_Master( CDSInfo *pInfo )
|
|
{
|
|
m_pInfo = pInfo;
|
|
m_bForceShuffle = false;
|
|
m_bShuffleRequested = false;
|
|
m_flLastShuffleRequestServiceTime = Plat_FloatTime();
|
|
|
|
// Spawn idle-priority worker threads right here.
|
|
m_bUsingMasterLocalThreads = (pInfo->m_WorkerInfo.m_pProcessFn != 0);
|
|
if ( VMPI_IsParamUsed( mpi_NoMasterWorkerThreads ) )
|
|
{
|
|
Msg( "%s found. No worker threads will be created.\n", VMPI_GetParamString( mpi_NoMasterWorkerThreads ) );
|
|
m_bUsingMasterLocalThreads = false;
|
|
}
|
|
m_WorkUnitWalker.Init( pInfo->m_nWorkUnits, this );
|
|
Shuffle();
|
|
|
|
if ( m_bUsingMasterLocalThreads )
|
|
RunThreads_Start( Master_WorkerThread_Static, this, k_eRunThreadsPriority_Idle );
|
|
|
|
uint64 lastShuffleTime = Plat_MSTime();
|
|
while ( m_WorkUnitWalker.Thread_NumWorkUnitsRemaining() > 0 )
|
|
{
|
|
VMPI_DispatchNextMessage( 200 );
|
|
CheckLocalMasterCompletedWorkUnits();
|
|
|
|
VMPITracker_HandleDebugKeypresses();
|
|
if ( g_pDistributeWorkCallbacks && g_pDistributeWorkCallbacks->Update() )
|
|
break;
|
|
|
|
// Reshuffle the work units optimally every certain interval.
|
|
if ( m_bForceShuffle || CheckShuffleRequest() )
|
|
{
|
|
Shuffle();
|
|
lastShuffleTime = Plat_MSTime();
|
|
m_bForceShuffle = false;
|
|
}
|
|
}
|
|
|
|
RunThreads_End();
|
|
}
|
|
|
|
virtual void RequestShuffle()
|
|
{
|
|
m_bShuffleRequested = true;
|
|
}
|
|
|
|
bool CheckShuffleRequest()
|
|
{
|
|
if ( m_bShuffleRequested )
|
|
{
|
|
double flCurTime = Plat_FloatTime();
|
|
if ( flCurTime - m_flLastShuffleRequestServiceTime > 2.0f ) // Only handle shuffle requests every so often.
|
|
{
|
|
m_flLastShuffleRequestServiceTime = flCurTime;
|
|
m_bShuffleRequested = false;
|
|
return true;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
void Shuffle()
|
|
{
|
|
// Build a list of who's working.
|
|
CUtlVector<unsigned short> whosWorking;
|
|
if ( m_bUsingMasterLocalThreads )
|
|
{
|
|
whosWorking.AddToTail( VMPI_MASTER_ID );
|
|
Assert( VMPI_MASTER_ID == 0 );
|
|
}
|
|
|
|
{
|
|
CWorkersReady *pWorkersReady = m_WorkersReadyCS.Lock();
|
|
for ( int i=0; i < pWorkersReady->m_WorkersReady.Count(); i++ )
|
|
{
|
|
int iWorker = pWorkersReady->m_WorkersReady[i];
|
|
if ( VMPI_IsProcConnected( iWorker ) )
|
|
whosWorking.AddToTail( iWorker );
|
|
}
|
|
m_WorkersReadyCS.Unlock();
|
|
}
|
|
|
|
// Before sending the shuffle command, tell any of these active workers about the pending WUs completed.
|
|
CWUsCompleted *pWUsCompleted = m_WUsCompletedCS.Lock();
|
|
|
|
m_WUSCompletedMessageBuffer.setLen( 0 );
|
|
if ( BuildWUsCompletedMessage( pWUsCompleted->m_Pending, m_WUSCompletedMessageBuffer ) > 0 )
|
|
{
|
|
for ( int i=m_bUsingMasterLocalThreads; i < whosWorking.Count(); i++ )
|
|
{
|
|
VMPI_SendData( m_WUSCompletedMessageBuffer.data, m_WUSCompletedMessageBuffer.getLen(), whosWorking[i] );
|
|
}
|
|
}
|
|
pWUsCompleted->m_Completed.AddMultipleToTail( pWUsCompleted->m_Pending.Count(), pWUsCompleted->m_Pending.Base() ); // Add the pending ones to the full list now.
|
|
pWUsCompleted->m_Pending.RemoveAll();
|
|
|
|
m_WUsCompletedCS.Unlock();
|
|
|
|
// Shuffle ourselves.
|
|
m_WorkUnitWalker.Shuffle( whosWorking.Count() );
|
|
|
|
// Send the shuffle command to the workers.
|
|
MessageBuffer mb;
|
|
PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_SHUFFLE );
|
|
|
|
unsigned short nWorkers = whosWorking.Count();
|
|
mb.write( &nWorkers, sizeof( nWorkers ) );
|
|
|
|
CRC32_t shuffleCRC = m_WorkUnitWalker.GetShuffleCRC();
|
|
mb.write( &shuffleCRC, sizeof( shuffleCRC ) );
|
|
|
|
// Now for each worker, assign him an index in the shuffle and send the shuffle command.
|
|
int workerIDPos = mb.getLen();
|
|
unsigned short id = 0;
|
|
mb.write( &id, sizeof( id ) );
|
|
for ( int i=m_bUsingMasterLocalThreads; i < whosWorking.Count(); i++ )
|
|
{
|
|
id = (unsigned short)i;
|
|
mb.update( workerIDPos, &id, sizeof( id ) );
|
|
VMPI_SendData( mb.data, mb.getLen(), whosWorking[i] );
|
|
}
|
|
}
|
|
|
|
int BuildWUsCompletedMessage( CUtlVector<WUIndexType> &wusCompleted, MessageBuffer &mb )
|
|
{
|
|
PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_WUS_COMPLETED_LIST );
|
|
m_pInfo->WriteWUIndex( wusCompleted.Count(), &mb );
|
|
for ( int i=0; i < wusCompleted.Count(); i++ )
|
|
{
|
|
m_pInfo->WriteWUIndex( wusCompleted[i], &mb );
|
|
}
|
|
return wusCompleted.Count();
|
|
}
|
|
|
|
virtual void OnWorkerReady( int iSource )
|
|
{
|
|
CWorkersReady *pWorkersReady = m_WorkersReadyCS.Lock();
|
|
if ( pWorkersReady->m_WorkersReady.Find( iSource ) == -1 )
|
|
{
|
|
pWorkersReady->m_WorkersReady.AddToTail( iSource );
|
|
|
|
// Get this guy up to speed on which WUs are done.
|
|
{
|
|
CWUsCompleted *pWUsCompleted = m_WUsCompletedCS.Lock();
|
|
m_WUSCompletedMessageBuffer.setLen( 0 );
|
|
BuildWUsCompletedMessage( pWUsCompleted->m_Completed, m_WUSCompletedMessageBuffer );
|
|
m_WUsCompletedCS.Unlock();
|
|
}
|
|
|
|
VMPI_SendData( m_WUSCompletedMessageBuffer.data, m_WUSCompletedMessageBuffer.getLen(), iSource );
|
|
m_bForceShuffle = true;
|
|
}
|
|
m_WorkersReadyCS.Unlock();
|
|
}
|
|
|
|
virtual bool HandleWorkUnitResults( WUIndexType iWorkUnit )
|
|
{
|
|
return Thread_HandleWorkUnitResults( iWorkUnit );
|
|
}
|
|
|
|
bool Thread_HandleWorkUnitResults( WUIndexType iWorkUnit )
|
|
{
|
|
if ( m_WorkUnitWalker.Thread_IsWorkUnitCompleted( iWorkUnit ) )
|
|
{
|
|
return false;
|
|
}
|
|
else
|
|
{
|
|
m_WorkUnitWalker.Thread_NoteWorkUnitCompleted( iWorkUnit );
|
|
|
|
// We need the lock on here because our own worker threads can call into here.
|
|
CWUsCompleted *pWUsCompleted = m_WUsCompletedCS.Lock();
|
|
pWUsCompleted->m_Pending.AddToTail( iWorkUnit );
|
|
m_WUsCompletedCS.Unlock();
|
|
return true;
|
|
}
|
|
}
|
|
|
|
virtual bool HandlePacket( MessageBuffer *pBuf, int iSource, bool bIgnoreContents )
|
|
{
|
|
if ( pBuf->data[1] == DW_SUBPACKETID_REQUEST_SHUFFLE )
|
|
{
|
|
if ( bIgnoreContents )
|
|
return true;
|
|
|
|
m_bShuffleRequested = true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
virtual void DisconnectHandler( int workerID )
|
|
{
|
|
CWorkersReady *pWorkersReady = m_WorkersReadyCS.Lock();
|
|
|
|
if ( pWorkersReady->m_WorkersReady.Find( workerID ) != -1 )
|
|
m_bForceShuffle = true;
|
|
|
|
m_WorkersReadyCS.Unlock();
|
|
}
|
|
|
|
public:
|
|
CDSInfo *m_pInfo;
|
|
|
|
class CWorkersReady
|
|
{
|
|
public:
|
|
CUtlVector<int> m_WorkersReady; // The list of workers who have said they're ready to participate.
|
|
};
|
|
CCriticalSectionData<CWorkersReady> m_WorkersReadyCS;
|
|
|
|
class CWUsCompleted
|
|
{
|
|
public:
|
|
CUtlVector<WUIndexType> m_Completed; // WUs completed that we have sent to workers.
|
|
CUtlVector<WUIndexType> m_Pending; // WUs completed that we haven't sent to workers.
|
|
};
|
|
CCriticalSectionData<CWUsCompleted> m_WUsCompletedCS;
|
|
MessageBuffer m_WUSCompletedMessageBuffer; // Used to send lists of completed WUs.
|
|
int m_bUsingMasterLocalThreads;
|
|
|
|
bool m_bForceShuffle;
|
|
bool m_bShuffleRequested;
|
|
double m_flLastShuffleRequestServiceTime;
|
|
|
|
CShuffledWorkUnitWalker m_WorkUnitWalker;
|
|
};
|
|
|
|
|
|
class CDistributor_SDKWorker : public IWorkUnitDistributorWorker, public IShuffleRequester
|
|
{
|
|
public:
|
|
virtual void Init( CDSInfo *pInfo )
|
|
{
|
|
m_iMyWorkUnitWalkerID = -1;
|
|
m_pInfo = pInfo;
|
|
m_WorkUnitWalker.Init( pInfo->m_nWorkUnits, this );
|
|
}
|
|
|
|
virtual void Release()
|
|
{
|
|
delete this;
|
|
}
|
|
|
|
virtual bool GetNextWorkUnit( WUIndexType *pWUIndex )
|
|
{
|
|
// If we don't have an ID yet, we haven't received a Shuffle() command, so we're waiting for that before working.
|
|
// TODO: we could do some random WUs here while we're waiting, although that could suck if the WUs take forever to do
|
|
// and they're duplicates.
|
|
if ( m_iMyWorkUnitWalkerID == -1 )
|
|
return false;
|
|
|
|
// Look in our current shuffled list of work units for the next one.
|
|
return m_WorkUnitWalker.Thread_GetNextWorkUnit( m_iMyWorkUnitWalkerID, pWUIndex );
|
|
}
|
|
|
|
virtual void NoteLocalWorkUnitCompleted( WUIndexType iWU )
|
|
{
|
|
m_WorkUnitWalker.Thread_NoteLocalWorkUnitCompleted( iWU );
|
|
}
|
|
|
|
virtual bool HandlePacket( MessageBuffer *pBuf, int iSource, bool bIgnoreContents )
|
|
{
|
|
// If it's a SHUFFLE message, then shuffle..
|
|
if ( pBuf->data[1] == DW_SUBPACKETID_SHUFFLE )
|
|
{
|
|
if ( bIgnoreContents )
|
|
return true;
|
|
|
|
unsigned short nWorkers, myID;
|
|
CRC32_t shuffleCRC;
|
|
pBuf->read( &nWorkers, sizeof( nWorkers ) );
|
|
pBuf->read( &shuffleCRC, sizeof( shuffleCRC ) );
|
|
pBuf->read( &myID, sizeof( myID ) );
|
|
m_iMyWorkUnitWalkerID = myID;
|
|
|
|
m_WorkUnitWalker.Shuffle( nWorkers );
|
|
if ( m_WorkUnitWalker.GetShuffleCRC() != shuffleCRC )
|
|
{
|
|
static int nWarnings = 1;
|
|
if ( ++nWarnings <= 2 )
|
|
Warning( "\nShuffle CRC mismatch\n" );
|
|
}
|
|
return true;
|
|
}
|
|
else if ( pBuf->data[1] == DW_SUBPACKETID_WUS_COMPLETED_LIST )
|
|
{
|
|
if ( bIgnoreContents )
|
|
return true;
|
|
|
|
WUIndexType nCompleted;
|
|
m_pInfo->ReadWUIndex( &nCompleted, pBuf );
|
|
for ( WUIndexType i=0; i < nCompleted; i++ )
|
|
{
|
|
WUIndexType iWU;
|
|
m_pInfo->ReadWUIndex( &iWU, pBuf );
|
|
m_WorkUnitWalker.Thread_NoteWorkUnitCompleted( iWU );
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
virtual void RequestShuffle()
|
|
{
|
|
// Ok.. request a reshuffle.
|
|
MessageBuffer mb;
|
|
PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_REQUEST_SHUFFLE );
|
|
VMPI_SendData( mb.data, mb.getLen(), VMPI_MASTER_ID );
|
|
}
|
|
|
|
private:
|
|
CDSInfo *m_pInfo;
|
|
CShuffledWorkUnitWalker m_WorkUnitWalker;
|
|
int m_iMyWorkUnitWalkerID;
|
|
};
|
|
|
|
|
|
|
|
IWorkUnitDistributorMaster* CreateWUDistributor_SDKMaster()
|
|
{
|
|
return new CDistributor_SDKMaster;
|
|
}
|
|
|
|
IWorkUnitDistributorWorker* CreateWUDistributor_SDKWorker()
|
|
{
|
|
return new CDistributor_SDKWorker;
|
|
}
|
|
|