1

utils/vmpi/vmpi_distribute_work_default.cpp

@@ -0,0 +1,602 @@
+//========= 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_WU_ASSIGNMENT	(VMPI_DISTRIBUTE_WORK_EXTRA_SUBPACKET_BASE+0)
+
+
+
+static int s_numWusToDeal = -1;
+
+void VMPI_SetWorkUnitsPartitionSize( int numWusToDeal )
+{
+	s_numWusToDeal = numWusToDeal;
+}
+
+
+class CWorkUnitInfo
+{
+public:
+	WUIndexType m_iWorkUnit;
+};
+
+
+class CWULookupInfo
+{
+public:
+	CWULookupInfo() : m_iWUInfo( -1 ), m_iPartition( -222222 ), m_iPartitionListIndex( -1 ) {}
+
+public:
+	int m_iWUInfo;				// Index into m_WUInfo.
+	int m_iPartition;			// Which partition it's in.
+	int m_iPartitionListIndex;	// Index into its partition's m_WUs.
+};
+
+
+class CPartitionInfo
+{
+public:
+	typedef CUtlLinkedList< WUIndexType, int > PartitionWUs;
+
+public:
+	int m_iPartition;	// Index into m_Partitions.
+	int m_iWorker; // Who owns this partition?
+	PartitionWUs m_WUs;	// Which WUs are in this partition?
+};
+
+
+// Work units tracker to track consecutive finished blocks
+class CWorkUnitsTracker
+{
+public:
+	CWorkUnitsTracker() {}
+
+public:
+	// Initializes the unit tracker to receive numUnits in future
+	void PrepareForWorkUnits( uint64 numUnits );
+	// Signals that a work unit has been finished
+	// returns a zero-based index of the next pending work unit
+	//			up to which the task list has been processed fully now
+	//			because the received work unit filled the gap or was the next pending work unit.
+	// returns 0 to indicate that this work unit is a "faster processed future work unit".
+	uint64 WorkUnitFinished( uint64 iWorkUnit );
+
+public:
+	enum WUInfo { kNone, kTrigger, kDone };
+	CVisibleWindowVector< uint8 > m_arrInfo;
+};
+
+void CWorkUnitsTracker::PrepareForWorkUnits( uint64 numUnits )
+{
+	m_arrInfo.Reset( numUnits + 1 );
+
+	if ( numUnits )
+	{
+		m_arrInfo.ExpandWindow( 2ull, kNone );
+		m_arrInfo.Get( 0ull ) = kTrigger;
+	}
+}
+
+uint64 CWorkUnitsTracker::WorkUnitFinished( uint64 iWorkUnit )
+{
+	uint64 uiResult = uint64( 0 );
+
+	if ( iWorkUnit >= m_arrInfo.FirstPossibleIndex() && iWorkUnit < m_arrInfo.PastPossibleIndex() )
+	{
+		// Need to access the element
+		m_arrInfo.ExpandWindow( iWorkUnit + 1, kNone );
+
+		// Set it done
+		uint8 &rchThere = m_arrInfo.Get( iWorkUnit ), chThere = rchThere;
+		rchThere = kDone;
+
+		// Should we trigger?
+		if ( kTrigger == chThere )
+		{
+			// Go along all "done" work units and trigger the last found one
+			while ( ( ( ++ iWorkUnit ) < m_arrInfo.PastVisibleIndex() ) &&
+				( kDone == m_arrInfo.Get( iWorkUnit ) ) )
+				continue;
+
+			m_arrInfo.Get( iWorkUnit ) = kTrigger;
+			m_arrInfo.ShrinkWindow( iWorkUnit - 1 );
+			uiResult = iWorkUnit;
+		}
+		else if( iWorkUnit == m_arrInfo.FirstPossibleIndex() )
+		{
+			// Go along all "done" work units and shrink including the last found one
+			while ( ( ( ++ iWorkUnit ) < m_arrInfo.PastVisibleIndex() ) &&
+				( kDone == m_arrInfo.Get( iWorkUnit ) ) )
+				continue;
+
+			m_arrInfo.ShrinkWindow( iWorkUnit - 1 );
+		}
+	}
+
+	return uiResult;
+}
+
+CWorkUnitsTracker g_MasterWorkUnitsTracker;
+
+
+
+static bool CompareSoonestWorkUnitSets( CPartitionInfo::PartitionWUs * const &x, CPartitionInfo::PartitionWUs * const &y )
+{
+	// Compare by fourth/second/first job in the partitions
+	WUIndexType missing = ~WUIndexType(0);
+	WUIndexType jobsX[4] = { missing, missing, missing, missing };
+	WUIndexType jobsY[4] = { missing, missing, missing, missing };
+	int counter = 0;
+
+	counter = 0;
+	FOR_EACH_LL( (*x), i )
+	{
+		jobsX[ counter ++ ] = (*x)[i];
+		if ( counter >= 4 )
+			break;
+	}
+
+	counter = 0;
+	FOR_EACH_LL( (*y), i )
+	{
+		jobsY[ counter ++ ] = (*y)[i];
+		if ( counter >= 4 )
+			break;
+	}
+
+	// Compare
+	if ( jobsX[3] != jobsY[3] )
+		return ( jobsX[3] < jobsY[3] );
+
+	if ( jobsX[1] != jobsY[1] )
+		return ( jobsX[1] < jobsY[1] );
+
+	return jobsX[0] < jobsY[0];
+}
+
+
+
+class CDistributor_DefaultMaster : public IWorkUnitDistributorMaster
+{
+public:
+	virtual void Release()
+	{
+		delete this;
+	}
+	
+	virtual void DistributeWork_Master( CDSInfo *pInfo )
+	{
+		m_pInfo = pInfo;
+		g_MasterWorkUnitsTracker.PrepareForWorkUnits( m_pInfo->m_nWorkUnits );
+		
+		m_WULookup.Reset( pInfo->m_nWorkUnits );
+		while ( m_WULookup.FirstPossibleIndex() < m_WULookup.PastPossibleIndex() )
+		{
+			VMPI_DispatchNextMessage( 200 );
+
+			VMPITracker_HandleDebugKeypresses();
+		
+			if ( g_pDistributeWorkCallbacks && g_pDistributeWorkCallbacks->Update() )
+				break;
+		}
+	}
+
+	virtual void OnWorkerReady( int iSource )
+	{
+		AssignWUsToWorker( iSource );
+	}
+
+	virtual bool HandleWorkUnitResults( WUIndexType iWorkUnit )
+	{
+		CWULookupInfo *pLookup = NULL;
+		if ( iWorkUnit >= m_WULookup.FirstPossibleIndex() && iWorkUnit < m_WULookup.PastVisibleIndex() )
+			pLookup = &m_WULookup.Get( iWorkUnit );
+
+		if ( !pLookup || pLookup->m_iWUInfo == -1 )
+			return false;
+		
+		// Mark this WU finished and remove it from the list of pending WUs.
+		m_WUInfo.Remove( pLookup->m_iWUInfo );
+		pLookup->m_iWUInfo = -1;	
+
+
+		// Get rid of the WU from its partition.
+		int iPartition = pLookup->m_iPartition;
+		CPartitionInfo *pPartition = m_Partitions[iPartition];
+		pPartition->m_WUs.Remove( pLookup->m_iPartitionListIndex );
+
+		// Shrink the window of the lookup work units
+		if ( iWorkUnit == m_WULookup.FirstPossibleIndex() )
+		{
+			WUIndexType kwu = iWorkUnit;
+			for ( WUIndexType kwuEnd = m_WULookup.PastVisibleIndex(); kwu < kwuEnd; ++ kwu )
+			{
+				if ( -1 != m_WULookup.Get( kwu ).m_iWUInfo && kwu > iWorkUnit )
+					break;
+			}
+			m_WULookup.ShrinkWindow( kwu - 1 );
+		}
+
+		// Give the worker some new work if need be.
+		if ( pPartition->m_WUs.Count() == 0 )
+		{
+			int iPartitionWorker = pPartition->m_iWorker;
+			delete pPartition;
+			m_Partitions.Remove( iPartition );
+	
+			// If there are any more WUs remaining, give the worker from this partition some more of them.
+			if ( m_WULookup.FirstPossibleIndex() < m_WULookup.PastPossibleIndex() )
+			{
+				AssignWUsToWorker( iPartitionWorker );
+			}
+		}
+
+		uint64 iDoneWorkUnits = g_MasterWorkUnitsTracker.WorkUnitFinished( iWorkUnit );
+		if ( iDoneWorkUnits && g_pDistributeWorkCallbacks )
+		{
+			g_pDistributeWorkCallbacks->OnWorkUnitsCompleted( iDoneWorkUnits );
+		}
+
+		return true;
+	}
+
+	virtual void DisconnectHandler( int workerID )
+	{
+		int iPartitionLookup = FindPartitionByWorker( workerID );
+		if ( iPartitionLookup != -1 )
+		{
+			// Mark this guy's partition as unowned so another worker can get it.
+			CPartitionInfo *pPartition = m_Partitions[iPartitionLookup];
+			pPartition->m_iWorker = -1;
+		}
+	}
+
+	CPartitionInfo* AddPartition( int iWorker )
+	{
+		CPartitionInfo *pNew = new CPartitionInfo;
+		pNew->m_iPartition = m_Partitions.AddToTail( pNew );
+		pNew->m_iWorker = iWorker;
+		return pNew;
+	}
+
+	bool SplitWUsPartition( CPartitionInfo *pPartitionLarge,
+						CPartitionInfo **ppFirstHalf, CPartitionInfo **ppSecondHalf,
+						int iFirstHalfWorker, int iSecondHalfWorker )
+	{
+		int nCount = pPartitionLarge->m_WUs.Count();
+		
+		if ( nCount > 1 )	// Allocate the partitions for the two workers
+		{
+			*ppFirstHalf = AddPartition( iFirstHalfWorker );
+			*ppSecondHalf = AddPartition( iSecondHalfWorker );
+		}
+		else				// Specially transfer a partition with too few work units
+		{
+			*ppFirstHalf = NULL;
+			*ppSecondHalf = AddPartition( iSecondHalfWorker );
+		}
+
+		// Prepare for transfer
+		CPartitionInfo *arrNewParts[2] = { *ppFirstHalf ? *ppFirstHalf : *ppSecondHalf, *ppSecondHalf };
+		
+		// Transfer the work units:
+		// alternate first/second halves
+		// don't put more than "half deal units" tasks into the second half
+		// e.g.               { 1, 2, 3, 4 }
+		// becomes: 1st half { 1, 2 }, 2nd half { 3, 4 }
+		for ( int k = 0; k < nCount; ++ k )
+		{
+			int iHead = pPartitionLarge->m_WUs.Head();
+			WUIndexType iWU = pPartitionLarge->m_WUs[ iHead ];
+			pPartitionLarge->m_WUs.Remove( iHead );
+
+			/*
+			int nHalf = !!( ( k % 2 ) || ( k >= nCount - 1 ) );
+			if ( k == 5 ) // no more than 2 jobs to branch off
+				arrNewParts[ 1 ] = arrNewParts[ 0 ];
+				*/
+			int nHalf = !( k < nCount/2 );
+			CPartitionInfo *pTo = arrNewParts[ nHalf ];
+
+			CWULookupInfo &li = m_WULookup.Get( iWU );
+			li.m_iPartition = pTo->m_iPartition;
+			li.m_iPartitionListIndex = pTo->m_WUs.AddToTail( iWU );
+		}
+
+		// LogPartitionsWorkUnits( pInfo );
+		return true;
+	}
+
+	void AssignWUsToWorker( int iWorker )
+	{
+		// Get rid of this worker's old partition.
+		int iPrevious = FindPartitionByWorker( iWorker );
+		if ( iPrevious != -1 )
+		{
+			delete m_Partitions[iPrevious];
+			m_Partitions.Remove( iPrevious );
+		}
+
+		if ( g_iVMPIVerboseLevel >= 1 )
+			Msg( "A" );
+
+
+		CVisibleWindowVector< CWULookupInfo > &vlkup = m_WULookup;
+		if ( CommandLine()->FindParm( "-mpi_NoScheduler" ) )
+		{
+			Warning( "\n\n-mpi_NoScheduler found: Warning - this should only be used for testing and with 1 worker!\n\n" );
+			vlkup.ExpandWindow( m_pInfo->m_nWorkUnits );
+			CPartitionInfo *pPartition = AddPartition( iWorker );
+			for ( int i=0; i < m_pInfo->m_nWorkUnits; i++ )
+			{
+				CWorkUnitInfo info;
+				info.m_iWorkUnit = i;
+
+				CWULookupInfo &li = vlkup.Get( i );
+				li.m_iPartition = pPartition->m_iPartition;
+				li.m_iPartitionListIndex = pPartition->m_WUs.AddToTail( i );
+				li.m_iWUInfo = m_WUInfo.AddToTail( info );
+			}
+
+			SendPartitionToWorker( pPartition, iWorker );
+			return;
+		}
+		
+
+		// Any partitions abandoned by workers?
+		int iAbandonedPartition = FindPartitionByWorker( -1 );
+		if ( -1 != iAbandonedPartition )
+		{
+			CPartitionInfo *pPartition = m_Partitions[ iAbandonedPartition ];
+			pPartition->m_iWorker = iWorker;
+			SendPartitionToWorker( pPartition, iWorker );
+		}
+		
+		// Any absolutely untouched partitions yet?
+		else if ( vlkup.PastVisibleIndex() < vlkup.PastPossibleIndex() )
+		{
+			// Figure out how many WUs to include in a batch
+			int numWusToDeal = s_numWusToDeal;
+			if ( numWusToDeal <= 0 )
+			{
+				uint64 uiFraction = vlkup.PastPossibleIndex() / g_nMaxWorkerCount;
+				Assert( uiFraction < INT_MAX/2 );
+				
+				numWusToDeal = int( uiFraction );
+				if ( numWusToDeal <= 0 )
+					numWusToDeal = 8;
+			}
+
+			// Allocate room for upcoming work units lookup
+			WUIndexType iBegin = vlkup.PastVisibleIndex();
+			WUIndexType iEnd = min( iBegin + g_nMaxWorkerCount * numWusToDeal, vlkup.PastPossibleIndex() );
+			vlkup.ExpandWindow( iEnd - 1 );
+
+			// Allocate a partition
+			size_t numPartitions = min( ( size_t )(iEnd - iBegin), ( size_t )g_nMaxWorkerCount );
+			CArrayAutoPtr< CPartitionInfo * > spArrPartitions( new CPartitionInfo* [ numPartitions ] );
+			CPartitionInfo **arrPartitions = spArrPartitions.Get();
+
+			arrPartitions[0] = AddPartition( iWorker );
+			for ( size_t k = 1; k < numPartitions; ++ k )
+				arrPartitions[k] = AddPartition( -1 );
+
+			// Assign upcoming work units to the partitions.
+			for ( WUIndexType i = iBegin ; i < iEnd; ++ i )
+			{
+				CWorkUnitInfo info;
+				info.m_iWorkUnit = i;
+
+				CPartitionInfo *pPartition = arrPartitions[ size_t( (i - iBegin) % numPartitions ) ];
+
+				CWULookupInfo &li = vlkup.Get( i );
+				li.m_iPartition = pPartition->m_iPartition;
+				li.m_iPartitionListIndex = pPartition->m_WUs.AddToTail( i );
+				li.m_iWUInfo = m_WUInfo.AddToTail( info );
+			}
+
+			// Now send this guy the WU list in his partition.
+			SendPartitionToWorker( arrPartitions[0], iWorker );
+		}
+
+		// Split one of the last partitions to finish sooner
+		else
+		{
+			// Find a partition to split.
+			int iPartToSplit = FindSoonestPartition();
+			if ( iPartToSplit >= 0 )
+			{
+				CPartitionInfo *pPartition = m_Partitions[ iPartToSplit ];
+
+				CPartitionInfo *pOldHalf = NULL, *pNewHalf = NULL;
+				int iOldWorker = pPartition->m_iWorker, iNewWorker = iWorker;
+				if ( SplitWUsPartition( pPartition, &pOldHalf, &pNewHalf, iOldWorker, iNewWorker ) )
+				{
+					if ( pOldHalf )
+						SendPartitionToWorker( pOldHalf, iOldWorker );
+					if ( pNewHalf )
+						SendPartitionToWorker( pNewHalf, iNewWorker );
+
+					// Delete the partition that got split
+					Assert( pPartition->m_WUs.Count() == 0 );
+					delete pPartition;
+					m_Partitions.Remove( iPartToSplit );
+				}
+			}
+		}
+	}
+
+	int FindSoonestPartition()
+	{
+		CUtlLinkedList < CPartitionInfo *, int > &lst = m_Partitions;
+
+		// Sorted partitions
+		CUtlMap< CPartitionInfo::PartitionWUs *, int > sortedPartitions ( CompareSoonestWorkUnitSets );
+		sortedPartitions.EnsureCapacity( lst.Count() );
+		FOR_EACH_LL( lst, i )
+		{
+			sortedPartitions.Insert( &lst[i]->m_WUs, i );
+		}
+
+		if ( sortedPartitions.Count() )
+		{
+			return sortedPartitions.Element( sortedPartitions.FirstInorder() );
+		}
+
+		return lst.Head();
+	}
+
+	int FindPartitionByWorker( int iWorker )
+	{
+		FOR_EACH_LL( m_Partitions, i )
+		{
+			if ( m_Partitions[i]->m_iWorker == iWorker )
+				return i;
+		}
+		return -1;
+	}
+
+	void SendPartitionToWorker( CPartitionInfo *pPartition, int iWorker )
+	{
+		// Stuff the next nWUs work units into the buffer.
+		MessageBuffer mb;
+		PrepareDistributeWorkHeader( &mb, DW_SUBPACKETID_WU_ASSIGNMENT );
+		
+		FOR_EACH_LL( pPartition->m_WUs, i )
+		{
+			WUIndexType iWU = pPartition->m_WUs[i];
+			mb.write( &iWU, sizeof( iWU ) );
+			VMPITracker_WorkUnitSentToWorker( ( int ) iWU, iWorker );
+		}
+
+		VMPI_SendData( mb.data, mb.getLen(), iWorker );
+	}
+
+	virtual bool HandlePacket( MessageBuffer *pBuf, int iSource, bool bIgnoreContents )
+	{
+		return false;
+	}
+	
+private:
+	
+	CDSInfo *m_pInfo;
+
+	CUtlLinkedList<CPartitionInfo*,int> m_Partitions;	
+	CVisibleWindowVector<CWULookupInfo> m_WULookup;		// Map work unit index to CWorkUnitInfo.
+	CUtlLinkedList<CWorkUnitInfo,int> m_WUInfo;			// Sorted with most elegible WU at the head.
+};
+
+
+
+class CDistributor_DefaultWorker : public IWorkUnitDistributorWorker
+{
+public:
+	virtual void Release()
+	{
+		delete this;
+	}
+
+	virtual void Init( CDSInfo *pInfo )
+	{
+	}
+	
+	virtual bool GetNextWorkUnit( WUIndexType *pWUIndex )
+	{
+		CCriticalSectionLock csLock( &m_CS );
+		csLock.Lock();
+
+		// NOTE: this is called from INSIDE worker threads.
+		if ( m_WorkUnits.Count() == 0 )
+		{
+			return false;
+		}
+		else
+		{
+			*pWUIndex = m_WorkUnits[ m_WorkUnits.Head() ];
+			m_WorkUnits.Remove( m_WorkUnits.Head() );
+			return true;
+		}
+	}
+
+	virtual void NoteLocalWorkUnitCompleted( WUIndexType iWU )
+	{
+	}
+
+	virtual bool HandlePacket( MessageBuffer *pBuf, int iSource, bool bIgnoreContents )
+	{
+		if ( pBuf->data[1] == DW_SUBPACKETID_WU_ASSIGNMENT )
+		{
+			// If the message wasn't even related to the current DistributeWork() call we're on, ignore it.
+			if ( bIgnoreContents )
+				return true;
+
+			if ( ((pBuf->getLen() - pBuf->getOffset()) % sizeof( WUIndexType )) != 0 )
+			{
+				Error( "DistributeWork: invalid work units packet from master" );
+			}
+
+			// Parse out the work unit indices.
+			CCriticalSectionLock csLock( &m_CS );
+			csLock.Lock();
+
+				m_WorkUnits.Purge();
+
+				int nIndices = (pBuf->getLen() - pBuf->getOffset()) / sizeof( WUIndexType );
+				for ( int i=0; i < nIndices; i++ )
+				{
+					WUIndexType iWU;
+					pBuf->read( &iWU, sizeof( iWU ) );
+
+					// Add the index to the list.
+					m_WorkUnits.AddToTail( iWU );
+				}
+			
+			csLock.Unlock();
+
+			return true;
+		}
+		else
+		{
+			return false;
+		}
+	}
+
+	// Threads eat up the list of WUs in here.
+	CCriticalSection m_CS;
+	CUtlLinkedList<WUIndexType, int> m_WorkUnits;			// A list of work units assigned to this worker
+};
+
+
+
+
+IWorkUnitDistributorMaster* CreateWUDistributor_DefaultMaster()
+{
+	return new CDistributor_DefaultMaster;
+}
+IWorkUnitDistributorWorker* CreateWUDistributor_DefaultWorker()
+{
+	return new CDistributor_DefaultWorker;
+}