wow thanks multi-threading!

Author: mastercoms

gameui/OptionsSubVideo.cpp

@@ -842,7 +842,7 @@ class COptionsSubVideoAdvancedDlg : public vgui::Frame
 
 		if (mat_reducefillrate.GetBool())
 		{
-			if (mat_phong.GetBool() && r_force_fastpath.GetBool() && r_skybox_lowend.GetBool())
+			if (!mat_phong.GetBool() && r_force_fastpath.GetBool() && r_skybox_lowend.GetBool())
 			{
 			    m_pShaderDetail->ActivateItem( 0 );
 			}

materialsystem/cmaterialsystem.cpp

@@ -3702,9 +3702,9 @@ void CMaterialSystem::EndFrame( void )
 
 			if ( m_pActiveAsyncJob )
 			{
-				if ( !m_pActiveAsyncJob->IsFinished() )
+				while ( !m_pActiveAsyncJob->IsFinished() )
 				{
-					m_pActiveAsyncJob->WaitForFinish();
+					m_pActiveAsyncJob->WaitForFinish(0);
 				}
 				// Sync with GPU if we had a job for it, even if it finished early on CPU!
 				if (!IsPC() && g_config.ForceHWSync())
@@ -3786,7 +3786,10 @@ void CMaterialSystem::EndFrame( void )
 			{
 				if ( m_pActiveAsyncJob )
 				{
-					m_pActiveAsyncJob->WaitForFinish();
+					while (!m_pActiveAsyncJob->IsFinished())
+					{
+						m_pActiveAsyncJob->WaitForFinish(0);
+					}
 					SafeRelease( m_pActiveAsyncJob );
 				}
 				// probably have a queued context set here, need hardware to flush the queue if the job isn't active
@@ -4871,7 +4874,10 @@ void CMaterialSystem::ThreadRelease( )
 	CJob		*pActiveAsyncJob = new CThreadRelease();
 	IThreadPool *pThreadPool = CreateMatQueueThreadPool();
 	pThreadPool->AddJob( pActiveAsyncJob );
-	pActiveAsyncJob->WaitForFinish();
+	while (!pActiveAsyncJob->IsFinished())
+	{
+		pActiveAsyncJob->WaitForFinish(0);
+	}
 
 	SafeRelease( pActiveAsyncJob );
 
@@ -4958,7 +4964,10 @@ MaterialLock_t CMaterialSystem::Lock()
 #if 1 // Rick's optimization: not sure this is needed anymore
 	if ( pCurContext != &m_HardwareRenderContext && m_pActiveAsyncJob )
 	{
-		m_pActiveAsyncJob->WaitForFinish();
+		while (!m_pActiveAsyncJob->IsFinished())
+		{
+			m_pActiveAsyncJob->WaitForFinish(0);
+		}
 		// threadsafety note: not releasing or nulling pointer.
 	}
 

public/vstdlib/jobthread.h

@@ -86,6 +86,8 @@ enum JobFlags_t
 	JF_BOOST_THREAD		= ( 1 << 1 ),	// Up the thread priority to max allowed while processing task
 	JF_SERIAL			= ( 1 << 2 ),	// Job cannot be executed out of order relative to other "strict" jobs
 	JF_QUEUE			= ( 1 << 3 ),	// Queue it, even if not an IO job
+	JF_NO_NOTIFY        = ( 1 << 4 ),   // Do not notify the queue, because we are gonna queue up more calls!
+	JF_NO_ADD           = ( 1 << 5 ),   // Do not add to the queue, because we are gonna modify the job!
 };
 
 enum JobPriority_t
@@ -346,6 +348,17 @@ abstract_class IThreadPool : public IRefCounted
 			return pJob; \
 		}
 
+	#define DEFINE_MEMBER_QUEUE_CALL_WITH_FLAGS(N) \
+		template <typename OBJECT_TYPE, typename FUNCTION_CLASS, typename FUNCTION_RETTYPE FUNC_TEMPLATE_FUNC_PARAMS_##N FUNC_TEMPLATE_ARG_PARAMS_##N> \
+		CJob *QueueCall(OBJECT_TYPE *pObject, int flags, FUNCTION_RETTYPE ( FUNCTION_CLASS::*pfnProxied )( FUNC_BASE_TEMPLATE_FUNC_PARAMS_##N ) FUNC_ARG_FORMAL_PARAMS_##N ) \
+		{ \
+			CJob *pJob; \
+			int jobFlags = JF_QUEUE; \
+	        jobFlags |= flags; \
+			AddFunctorInternal( CreateFunctor( pObject, pfnProxied FUNC_FUNCTOR_CALL_ARGS_##N ), &pJob, NULL, jobFlags ); \
+			return pJob; \
+		}
+
 	//-------------------------------------
 
 	#define DEFINE_CONST_MEMBER_QUEUE_CALL(N) \
@@ -387,6 +400,7 @@ abstract_class IThreadPool : public IRefCounted
 	FUNC_GENERATE_ALL( DEFINE_REF_COUNTING_CONST_MEMBER_ADD_CALL );
 	FUNC_GENERATE_ALL( DEFINE_NONMEMBER_QUEUE_CALL );
 	FUNC_GENERATE_ALL( DEFINE_MEMBER_QUEUE_CALL );
+	FUNC_GENERATE_ALL( DEFINE_MEMBER_QUEUE_CALL_WITH_FLAGS );
 	FUNC_GENERATE_ALL( DEFINE_CONST_MEMBER_QUEUE_CALL );
 	FUNC_GENERATE_ALL( DEFINE_REF_COUNTING_MEMBER_QUEUE_CALL );
 	FUNC_GENERATE_ALL( DEFINE_REF_COUNTING_CONST_MEMBER_QUEUE_CALL );
@@ -398,6 +412,7 @@ abstract_class IThreadPool : public IRefCounted
 	#undef DEFINE_REF_COUNTING_CONST_MEMBER_ADD_CALL
 	#undef DEFINE_NONMEMBER_QUEUE_CALL
 	#undef DEFINE_MEMBER_QUEUE_CALL
+    #undef DEFINE_MEMBER_QUEUE_CALL_WITH_FLAGS
 	#undef DEFINE_CONST_MEMBER_QUEUE_CALL
 	#undef DEFINE_REF_COUNTING_MEMBER_QUEUE_CALL
 	#undef DEFINE_REF_COUNTING_CONST_MEMBER_QUEUE_CALL
@@ -896,12 +911,18 @@ class CParallelProcessor
 			CJob **jobs = (CJob **)stackalloc( nJobs * sizeof(CJob **) );
 			int i = nJobs;
 
-			while( i-- )
+			// For the first jobs, don't notify the thread pool yet so every thread has a fair chance at getting a stab at it.
+			while( --i )
 			{
-				jobs[i] = pThreadPool->QueueCall( this, &CParallelProcessor<ITEM_TYPE, ITEM_PROCESSOR_TYPE>::DoExecute );
+				jobs[i] = pThreadPool->QueueCall( this, JF_NO_NOTIFY, &CParallelProcessor<ITEM_TYPE, ITEM_PROCESSOR_TYPE>::DoExecute );
 				jobs[i]->SetDescription( m_szDescription );
 			}
 
+			// Last job notifies for all of them!
+			jobs[0] = pThreadPool->QueueCall( this, &CParallelProcessor<ITEM_TYPE, ITEM_PROCESSOR_TYPE>::DoExecute );
+			jobs[0]->SetDescription( m_szDescription );
+
+			// Do jobs alongside the threads
 			DoExecute();
 
 			for ( i = 0; i < nJobs; i++ )

tier0/threadtools.cpp

@@ -1621,7 +1621,9 @@ int ThreadWaitForEvents(int nEvents, CThreadEvent* const* pEvents, bool bWaitAll
 			return false;
 		};
 
+		// UNDONE(mastercoms): TOO OPTIMISTIC: what if its signaled on this thread, but actually not signaled?
 		// Most optimistic case: we have signal state synced already.
+#if 0
 		for (int i = 0; i < 20; i++)
 		{
 			if (lPredSignaledAny())
@@ -1635,6 +1637,7 @@ int ThreadWaitForEvents(int nEvents, CThreadEvent* const* pEvents, bool bWaitAll
 			}
 			ThreadPause();
 		}
+#endif
 		if (!bRet)
 		{
 			// Second optimistic case: we can do an initial check to minimize contention

vstdlib/jobthread.cpp

@@ -92,7 +92,7 @@ class ALIGN16 CJobQueue
 		return NULL;
 	}
 
-	int Push( CJob *pJob, int iThread = -1 )
+	int Push( CJob *pJob, bool bNotify = true )
 	{
 		pJob->AddRef();
 
@@ -106,12 +106,15 @@ class ALIGN16 CJobQueue
 
 		m_pQueues[pJob->GetPriority()]->PushItem( pJob );
 
-		m_mutex.Lock();
-		if ( ++m_nItems == 1 )
+		if (bNotify)
 		{
-			m_JobAvailableEvent.Set();
+		    m_mutex.Lock();
+		    if ( ++m_nItems == 1 )
+		    {
+			    m_JobAvailableEvent.Set();
+		    }
+		    m_mutex.Unlock();
 		}
-		m_mutex.Unlock();
 
 		return nOverflow;
 	}
@@ -598,35 +601,64 @@ int CThreadPool::YieldWait( CThreadEvent **pEvents, int nEvents, bool bWaitAll,
 {
 	tmZone( TELEMETRY_LEVEL0, TMZF_IDLE, "%s(%d) SPINNING %t", __FUNCTION__, timeout, tmSendCallStack( TELEMETRY_LEVEL0, 0 ) );
 
-	Assert( timeout == TT_INFINITE ); // unimplemented
-
 	int result;
 	CJob *pJob;
-	// Always wait for zero milliseconds initially, to let us process jobs on this thread.
-	timeout = 0;
-	while ( ( result = ThreadWaitForEvents( nEvents, pEvents, bWaitAll, timeout ) ) == WAIT_TIMEOUT )
+
+	if (timeout != TT_INFINITE)
 	{
-		if ( !m_bExecOnThreadPoolThreadsOnly && m_SharedQueue.Pop( &pJob ) )
+	    DevWarning("Yielding for non-infinite timeout %d!\n", timeout);
+	}
+
+	// If we aren't exiting this wait immediately, take the opportunity to process jobs on this thread as much as possible.
+	if (timeout != 0)
+	{
+		// If we're still waiting on this thread pool, then process its jobs as much as possible.
+		while ( (result = ThreadWaitForEvents( nEvents, pEvents, bWaitAll, 0 )) == WAIT_TIMEOUT )
 		{
-			ServiceJobAndRelease( pJob );
-			m_nJobs--;
+		    if ( !m_bExecOnThreadPoolThreadsOnly && m_SharedQueue.Pop( &pJob ) )
+		    {
+			    ServiceJobAndRelease( pJob );
+			    m_nJobs--;
+		    }
+#if 1
+			else
+			{
+				// If we can't process jobs or there are no jobs to process, stop trying to respond to them.
+			    break;
+			}
+#else
+		    else
+		    {
+			    // UNDONE(mastercoms): we can just use TT_INFINITE as needed?
+
+			    // Since there are no jobs for the main thread set the timeout to infinite.
+			    // The only disadvantage to this is that if a job thread creates a new job
+			    // then the main thread will not be available to pick it up, but if that
+			    // is a problem you can just create more worker threads. Debugging test runs
+			    // of TF2 suggests that jobs are only ever added from the main thread which
+			    // means that there is no disadvantage.
+			    // Waiting on the events instead of busy spinning has multiple advantages.
+			    // It avoids wasting CPU time/electricity, it makes it more obvious in profiles
+			    // when the main thread is idle versus busy, and it allows ready thread analysis
+			    // in xperf to find out what woke up a waiting thread.
+			    // It also avoids unnecessary CPU starvation -- seen on customer traces of TF2.
+			    timeout = TT_INFINITE;
+		    }
+#endif
 		}
-		else
+
+		// Now that we have responded to jobs with near zero latency, and there's no more jobs to process, enter our extended wait with timeout if we do need to wait more.
+		if (result == WAIT_TIMEOUT)
 		{
-			// Since there are no jobs for the main thread set the timeout to infinite.
-			// The only disadvantage to this is that if a job thread creates a new job
-			// then the main thread will not be available to pick it up, but if that
-			// is a problem you can just create more worker threads. Debugging test runs
-			// of TF2 suggests that jobs are only ever added from the main thread which
-			// means that there is no disadvantage.
-			// Waiting on the events instead of busy spinning has multiple advantages.
-			// It avoids wasting CPU time/electricity, it makes it more obvious in profiles
-			// when the main thread is idle versus busy, and it allows ready thread analysis
-			// in xperf to find out what woke up a waiting thread.
-			// It also avoids unnecessary CPU starvation -- seen on customer traces of TF2.
-			timeout = TT_INFINITE;
+			result = ThreadWaitForEvents( nEvents, pEvents, bWaitAll, timeout );
 		}
 	}
+	else
+	{
+		// We explicitly asked for it, so trust that we have a good reason.
+		result = ThreadWaitForEvents( nEvents, pEvents, bWaitAll, 0 );
+	}
+	
 	return result;
 }
 
@@ -679,6 +711,12 @@ void CThreadPool::AddJob( CJob *pJob )
 		return;
 	}
 
+	int flags = pJob->GetFlags();
+	if (flags & JF_NO_ADD)
+	{
+		return;
+	}
+
 	if ( m_Threads.Count() == 0 )
 	{
 		// So only threadpool jobs are supposed to execute the jobs, but there are no threadpool threads?
@@ -688,8 +726,6 @@ void CThreadPool::AddJob( CJob *pJob )
 		return;
 	}
 
-	int flags = pJob->GetFlags();
-
 	if ( !m_bExecOnThreadPoolThreadsOnly && ( ( flags & ( JF_IO | JF_QUEUE ) ) == 0 ) /* @TBD && !m_queue.Count() */ )
 	{
 		if ( !NumIdleThreads() )
@@ -722,7 +758,9 @@ void CThreadPool::InsertJobInQueue( CJob *pJob )
 {
 	CJobQueue *pQueue;
 
-	if ( !( pJob->GetFlags() & JF_SERIAL ) )
+	int flags = pJob->GetFlags();
+
+	if ( !( flags & JF_SERIAL ) )
 	{
 		int iThread = pJob->GetServiceThread();
 		if ( iThread == -1 || !m_Threads.IsValidIndex( iThread ) )
@@ -739,7 +777,7 @@ void CThreadPool::InsertJobInQueue( CJob *pJob )
 		pQueue = &(m_Threads[0]->AccessDirectQueue());
 	}
 
-	m_nJobs -= pQueue->Push( pJob );
+	m_nJobs -= pQueue->Push( pJob, (flags & JF_NO_NOTIFY) == 0 );
 }
 
 //---------------------------------------------------------