Modified source engine (2017) developed by valve and leaked in 2020. Not for commercial purporses
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

430 lines
13 KiB

//========= Copyright Valve Corporation, All rights reserved. ============//
//
// Purpose:
//
// $NoKeywords: $
//=============================================================================
#ifndef GC_JOBMGR_H
#define GC_JOBMGR_H
#ifdef _WIN32
#pragma once
#endif
#include "tier0/fasttimer.h"
#include "tier1/utlpriorityqueue.h"
#include "job.h"
#include "workthreadpool.h"
class GCConVar;
#include "tier0/memdbgon.h"
namespace GCSDK
{
#if defined(_DEBUG)
// this is restricted to debug builds due to the performance cost
// that could be changed by removing the expensive sm_listAllJobs.Find() command
#define DEBUG_JOB_LIST
#endif // defined(_DEBUG)
struct JobStats_t
{
uint m_cJobsCurrent;
uint m_cJobsTotal;
uint m_cJobsFailed;
uint64 m_cJobsTimedOut; // # of jobs timed out ever
double m_flSumJobTimeMicrosec;
double m_flSumSqJobTimeMicrosec;
uint64 m_unMaxJobTimeMicrosec;
uint m_cTimeslices;
JobStats_t()
{
memset( this, 0, sizeof(JobStats_t) );
}
};
struct JobStatsBucket_t
{
JobStatsBucket_t()
{
memset( this, 0, sizeof(JobStatsBucket_t) );
}
char m_rgchName[64];
uint64 m_cCompletes;
uint64 m_u64RunTimeMax;
uint64 m_cTimeoutNetMsg;
uint64 m_cLongInterYieldTime;
uint64 m_cLocksAttempted;
uint64 m_cLocksWaitedFor;
uint64 m_cLocksFailed;
uint64 m_cLocksLongHeld;
uint64 m_cLocksLongWait;
uint64 m_cWaitTimeout;
uint64 m_u64JobDuration;
uint64 m_cJobsPaused;
uint64 m_cJobsFailed;
uint64 m_u64RunTime;
// use by ListJobs
uint64 m_cPauseReasonNetworkMsg;
uint64 m_cPauseReasonSleepForTime;
uint64 m_cPauseReasonWaitingForLock;
uint64 m_cPauseReasonYield;
uint64 m_cPauseReasonSQL;
uint64 m_cPauseReasonWorkItem;
#ifdef DBGFLAG_VALIDATE
void Validate( CValidator &validator, const char *pchName )
{
VALIDATE_SCOPE();
}
#endif
};
enum EJobProfileAction
{
k_EJobProfileAction_ErrorReport = 0,
k_EJobProfileAction_Start = 1,
k_EJobProfileAction_Stop = 2,
k_EJobProfileAction_Dump = 3,
k_EJobProfileAction_Clear = 4,
};
enum EJobProfileSortOrder
{
k_EJobProfileSortOrder_Alpha = 0,
k_EJobProfileSortOrder_Count = 1,
k_EJobProfileSortOrder_TotalRuntime = 2,
};
struct JobProfileStats_t
{
int m_iJobProfileSort;
CUtlMap< uint32, JobStatsBucket_t, int > *pmapStatsBucket;
};
//-----------------------------------------------------------------------------
// Purpose: This keeps track of all jobs that belong to a given hub.
// It's primarily used for routing incoming messages to jobs.
//-----------------------------------------------------------------------------
class CJobMgr
{
public:
// Constructors & destructors
CJobMgr();
~CJobMgr();
// gets the next available job ID
JobID_t GetNewJobID();
// Set the thread count for the internal thread pool(s)
void SetThreadPoolSize( uint cThreads );
// Run any sleeping jobs who's wakeup time has arrived and check for timeouts
bool BFrameFuncRunSleepingJobs( CLimitTimer &limitTimer );
// Run any yielding jobs, even low priority ones
bool BFrameFuncRunYieldingJobs( CLimitTimer &limitTimer );
// Route this message to an existing Job, or create a new one if that JobID does not exist
bool BRouteMsgToJob( void *pParent, IMsgNetPacket *pNetPacket, const JobMsgInfo_t &jobMsgInfo );
// Adds a new Job to the mgr and generates a JobID for it.
void InsertJob( CJob &job );
// Removes a Job from the mgr (the caller is still responsible for freeing it)
void RemoveJob( CJob &job );
//called by a job that has just been started to place itself on the yield queue instead of running
void AddDelayedJobToYieldList( CJob &job );
#ifdef GC
// resumes the specified job if it is, in fact, waiting for a SQL query to return
bool BResumeSQLJob( JobID_t jobID );
// yields waiting for a query response
bool BYieldingRunQuery( CJob &job, CGCSQLQueryGroup *pQueryGroup, ESchemaCatalog eSchemaCatalog );
// SQL profiling
enum ESQLProfileSort
{
k_ESQLProfileSortTotalTime,
k_ESQLProfileSortTotalCount,
k_ESQLProfileSortAvgTime,
k_ESQLProfileSortName
};
void StartSQLProfiling();
void StopSQLProfiling();
void DumpSQLProfile( ESQLProfileSort eSort );
#endif
// returns true if we're running any jobs of the specified name
// slow to call if lots of jobs are running, should only be used by tests
bool BIsJobRunning( const char *pchJobName );
// passes a network msg directly to the specified job
void PassMsgToJob( CJob &job, IMsgNetPacket *pNetPacket, const JobMsgInfo_t &jobMsgInfo );
// yields until a network message is received
bool BYieldingWaitForMsg( CJob &job );
// yields for a set amount of time
bool BYieldingWaitTime( CJob &job, uint32 cMicrosecondsToSleep );
// simple yield until Run() called again
bool BYield( CJob &job );
// Yield only if job manager decides we need to
bool BYieldIfNeeded( CJob &job, bool *pbYielded );
// Thread pool work item
bool BYieldingWaitForWorkItem( CJob &job, const char *pszWorkItemName = NULL );
bool BRouteWorkItemCompleted( JobID_t jobID, bool bWorkItemCanceled ) { return BRouteWorkItemCompletedInternal( jobID, bWorkItemCanceled, /* bShouldExist */ true, /* bResumeImmediately */ true ); }
bool BRouteWorkItemCompletedIfExists( JobID_t jobID, bool bWorkItemCanceled ) { return BRouteWorkItemCompletedInternal( jobID, bWorkItemCanceled, /* bShouldExist */ false, /* bResumeImmediately */ true ); }
bool BRouteWorkItemCompletedDelayed( JobID_t jobID, bool bWorkItemCanceled ) { return BRouteWorkItemCompletedInternal( jobID, bWorkItemCanceled, /* bShouldExist */ true, /* bResumeImmediately */ false ); }
bool BRouteWorkItemCompletedIfExistsDelayed( JobID_t jobID, bool bWorkItemCanceled ) { return BRouteWorkItemCompletedInternal( jobID, bWorkItemCanceled, /* bShouldExist */ false, /* bResumeImmediately */ false ); }
void AddThreadedJobWorkItem( CWorkItem *pWorkItem );
void StopWorkThreads() { m_WorkThreadPool.StopWorkThreads(); }
static int ProfileSortFunc( void *pCtx, const int *lhs, const int *rhs );
void ProfileJobs( EJobProfileAction ejobProfileAction, EJobProfileSortOrder iSortOrder = k_EJobProfileSortOrder_Alpha );
int DumpJobSummary();
void DumpJob( JobID_t jobID, int nPrintLocksMax = 20 ) const;
int CountJobs() const; // counts currently active jobs
void CheckThreadID(); // make sure we are still in the correct thread
int CountYieldingJobs() const { return m_ListJobsYieldingRegPri.Count(); } // counts jobs currently in a yielding state
bool HasOutstandingThreadPoolWorkItems();
void SetIsShuttingDown();
bool GetIsShuttingDown() const { return m_bIsShuttingDown; }
void *GetMainMemoryDebugInfo() { return g_memMainDebugInfo.Base(); }
#ifdef DBGFLAG_VALIDATE
void Validate( CValidator &validator, const char *pchName ); // Validate our internal structures
static void ValidateStatics( CValidator &validator, const char *pchName );
#endif /* DBGFLAG_VALIDATE */
// wakes up a job that was waiting on a lock
void WakeupLockedJob( CJob &job );
// returns true if there is a job active with the specified ID
bool BJobExists( JobID_t jobID ) const;
// returns a job
CJob *GetPJob( JobID_t jobID );
const CJob *GetPJob( JobID_t jobID ) const;
JobStats_t& GetJobStats() { return m_JobStats; }
// Access work thread pool directly
CWorkThreadPool *AccessWorkThreadPool() { return &m_WorkThreadPool; }
// Debug helpers
// dumps a list of all running jobs across ALL job managers
void DumpJobs( const char *pszJobName, int nMax, int nPrintLocksMax = 1 ) const;
// cause a debug break in the given job
static void DebugJob( int iJob );
// disable/enable yielding for debugging
void SetPauseAllowed( bool bNewPauseAllowed ) { m_bDebugDisallowPause = !bNewPauseAllowed; }
private:
bool BRouteWorkItemCompletedInternal( JobID_t jobID, bool bWorkItemCanceled, bool bShouldExist, bool bResumeImmediately );
// Create a new job for this message
bool BLaunchJobFromNetworkMsg( void *pParent, const JobMsgInfo_t &jobMsgInfo, IMsgNetPacket *pNetPacket );
// Internal add to yield list (looks at priority)
void AddToYieldList( CJob &job );
// Get an IJob given a job ID and pause reason
bool BGetIJob( JobID_t jobID, EJobPauseReason eJobPauseReason, bool bShouldExist, int *pIJob );
// Map containing all of our jobs
CUtlMap<JobID_t, CJob *, int> m_MapJob;
// jobs simply waiting until the next Run()
struct JobYielding_t
{
JobID_t m_JobID;
uint m_nIteration;
};
CUtlLinkedList<JobYielding_t, int> m_ListJobsYieldingRegPri;
bool BResumeYieldingJobs( CLimitTimer &limitTimer );
bool BResumeYieldingJobsFromList( CUtlLinkedList<JobYielding_t, int> &listJobsYielding, uint nCurrentIteration, CLimitTimer &limitTimer );
uint m_nCurrentYieldIterationRegPri;
// jobs waiting on a timer
struct JobSleeping_t
{
JobID_t m_JobID;
CJobTime m_SWakeupTime;
CJobTime m_STimeTouched;
};
CUtlPriorityQueue<JobSleeping_t> m_QueueJobSleeping;
bool BResumeSleepingJobs( CLimitTimer &limitTimer );
static bool JobSleepingLessFunc( JobSleeping_t const &lhs, JobSleeping_t const &rhs );
// timeout list of jobs, ordered from oldest to newest
struct JobTimeout_t
{
JobID_t m_JobID;
CJobTime m_STimePaused;
CJobTime m_STimeTouched;
uint32 m_cHeartbeatsBeforeTimeout;
};
CUtlLinkedList<JobTimeout_t, int> m_ListJobTimeouts;
CUtlMap<JobID_t, int, int> m_MapJobTimeoutsIndexByJobID;
void PauseJob( CJob &job, EJobPauseReason eJobPauseReason );
void CheckForJobTimeouts( CLimitTimer &limitTimer );
void TimeoutJob( CJob &job );
bool m_bJobTimedOut;
// thread pool usage, for running job functions in other threads
CWorkThreadPool m_WorkThreadPool;
void AccumulateStatsofJob( CJob &job );
void RecordOrphanedMessage( MsgType_t eMsg, JobID_t jobIDTarget );
// stats info
JobStats_t m_JobStats;
// static job registration
static void RegisterJobType( const JobType_t *pJobType );
friend void Job_RegisterJobType( const JobType_t *pJobType );
JobID_t m_unNextJobID;
uint m_unFrameFuncThreadID; // the thread is JobMgr is working in
bool m_bProfiling;
bool m_bIsShuttingDown;
int m_cErrorsToReport;
CUtlMap< uint32, JobStatsBucket_t, int > m_mapStatsBucket;
CUtlMap<MsgType_t, int, int> m_mapOrphanMessages;
CUtlMemory<unsigned char> g_memMainDebugInfo;
#ifdef GC
// sql profiling
bool m_bSQLProfiling;
CFastTimer m_sqlTimer;
struct PendingSQLJob_t
{
int64 m_nStartMicrosec;
int32 m_iBucket;
};
struct SQLProfileBucket_t
{
int64 m_nTotalMicrosec;
uint32 m_unCount;
};
CUtlHashMapLarge<GID_t, PendingSQLJob_t> m_mapSQLQueriesInFlight;
CUtlDict<SQLProfileBucket_t> m_dictSQLBuckets;
struct SQLProfileCtx_t
{
ESQLProfileSort m_eSort;
CUtlDict<SQLProfileBucket_t> *pdictBuckets;
};
static int SQLProfileSortFunc( void *pCtx, const int *lhs, const int *rhs );
#endif
#ifdef DEBUG_JOB_LIST
// static job debug list
static CUtlLinkedList<CJob *, int> sm_listAllJobs;
#endif
bool m_bDebugDisallowPause;
};
//-----------------------------------------------------------------------------
// Purpose: passthrough function just so the CJob internal data can be kept private
//-----------------------------------------------------------------------------
inline void Job_RegisterJobType( const JobType_t *pJobType )
{
CJobMgr::RegisterJobType( pJobType );
}
//-----------------------------------------------------------------------------
// Purpose: passthrough function just so the CJob internal data can be kept private
//-----------------------------------------------------------------------------
inline void Job_SetJobType( CJob &job, const JobType_t *pJobType )
{
job.m_pJobType = pJobType;
}
//-----------------------------------------------------------------------------
// Purpose: job registration macro
//-----------------------------------------------------------------------------
#define GC_REG_JOB( parentclass, jobclass, jobname, msg, servertype ) \
GCSDK::CJob *CreateJob_##jobclass( parentclass *pParent, void * pvStartParam ); \
static const GCSDK::JobType_t g_JobType_##jobclass = { jobname, (GCSDK::MsgType_t)msg, servertype, (GCSDK::JobCreationFunc_t)CreateJob_##jobclass }; \
GCSDK::CJob *CreateJob_##jobclass( parentclass *pParent, void * pvStartParam ) \
{ \
GCSDK::CJob *job = GCSDK::CJob::AllocateJob<jobclass>( pParent ); \
if ( job ) \
{ \
Job_SetJobType( *job, &g_JobType_##jobclass ); \
if ( pvStartParam ) job->SetStartParam( pvStartParam ); \
} \
else \
{ \
AssertMsg( job, "CJob::AllocateJob<" #jobclass "> returned NULL!" ); \
} \
return job; \
} \
static class CRegJob_##jobclass \
{ \
public: CRegJob_##jobclass() \
{ \
Job_RegisterJobType( &g_JobType_##jobclass ); \
} \
} g_RegJob_##jobclass;
//-----------------------------------------------------------------------------
// Purpose: job registration macro for job triggered by web api request
//-----------------------------------------------------------------------------
#define REG_WEBAPI_JOB( parentclass, jobclass, jobname, servertype ) \
CJob *CreateJob_##jobclass( parentclass *pParent, void * pvStartParam ); \
static const JobType_t g_JobType_##jobclass = { jobname, k_EGCMsgInvalid, servertype, (JobCreationFunc_t)CreateJob_##jobclass }; \
CJob *CreateJob_##jobclass( parentclass *pParent, void * pvStartParam ) \
{ \
CJob *job = CJob::AllocateJob<jobclass>( pParent ); \
if ( job ) \
{ \
Job_SetJobType( *job, &g_JobType_##jobclass ); \
if ( pvStartParam ) job->SetStartParam( pvStartParam ); \
} \
else \
{ \
AssertMsg( job, "CJob::AllocateJob<" #jobclass "> returned NULL!" ); \
} \
return job; \
} \
static class CRegJob_##jobclass \
{ \
public: CRegJob_##jobclass() \
{ \
Job_RegisterJobType( &g_JobType_##jobclass ); \
} \
} g_RegJob_##jobclass;
} // namespace GCSDK
#include "tier0/memdbgoff.h"
#endif // GC_JOBMGR_H