《Exploring in UE4》多线程机制详解[原理分析]

//Runnable.h
class CORE_API FRunnable
{
public:
        /**
         * Initializes the runnable object.
         *
         * This method is called in the context of the thread object that aggregates this, not the
         * thread that passes this runnable to a new thread.
         *
         * @return True if initialization was successful, false otherwise
         * @see Run, Stop, Exit
         */
        virtual bool Init()
        {
                return true;
        }

        /**
         * Runs the runnable object.
         *
         * This is where all per object thread work is done. This is only called if the initialization was successful.
         *
         * @return The exit code of the runnable object
         * @see Init, Stop, Exit
         */
        virtual uint32 Run() = 0;

        /**
         * Stops the runnable object.
         *
         * This is called if a thread is requested to terminate early.
         * @see Init, Run, Exit
         */
        virtual void Stop() { }

        /**
         * Exits the runnable object.
         *
         * Called in the context of the aggregating thread to perform any cleanup.
         * @see Init, Run, Stop
         */
        virtual void Exit() { }

        /**
         * Gets single thread interface pointer used for ticking this runnable when multi-threading is disabled.
         * If the interface is not implemented, this runnable will not be ticked when FPlatformProcess::SupportsMultithreading() is false.
         *
        * @return Pointer to the single thread interface or nullptr if not implemented.
         */
        virtual class FSingleThreadRunnable* GetSingleThreadInterface( )
        {
                return nullptr;
        }

        /** Virtual destructor */
        virtual ~FRunnable() { }
};

//AsyncWork.h
        class ExampleAsyncTask : public FNonAbandonableTask
        {
                friend class FAsyncTask<ExampleAsyncTask>;

                int32 ExampleData;

                ExampleAsyncTask(int32 InExampleData)
                 : ExampleData(InExampleData)
                {
                }

                void DoWork()
                {
                        ... do the work here
                }

                FORCEINLINE TStatId GetStatId() const
                {
                        RETURN_QUICK_DECLARE_CYCLE_STAT(ExampleAsyncTask, STATGROUP_ThreadPoolAsyncTasks);
                }
        };

        void Example()
        {

                //start an example job

                FAsyncTask<ExampleAsyncTask>* MyTask = new FAsyncTask<ExampleAsyncTask>( 5 );
                MyTask->StartBackgroundTask();

                //--or --

                MyTask->StartSynchronousTask();

                //to just do it now on this thread
                //Check if the task is done :

                if (MyTask->IsDone())
                {
                }

                //Spinning on IsDone is not acceptable( see EnsureCompletion ), but it is ok to check once a frame.
                //Ensure the task is done, doing the task on the current thread if it has not been started, waiting until completion in all cases.

                MyTask->EnsureCompletion();
                delete Task;
        }

// FEngineLoop.PreInit   LaunchEngineLoop.cpp
if (FPlatformProcess::SupportsMultithreading())
{
        {
                GThreadPool = FQueuedThreadPool::Allocate();
                int32 NumThreadsInThreadPool = FPlatformMisc::NumberOfWorkerThreadsToSpawn();

                // we are only going to give dedicated servers one pool thread
                if (FPlatformProperties::IsServerOnly())
                {
                    NumThreadsInThreadPool = 1;
                }
                verify(GThreadPool->Create(NumThreadsInThreadPool, 128 * 1024));
        }
#ifUSE_NEW_ASYNC_IO
        {
                GIOThreadPool = FQueuedThreadPool::Allocate();
                int32 NumThreadsInThreadPool = FPlatformMisc::NumberOfIOWorkerThreadsToSpawn();
                if (FPlatformProperties::IsServerOnly())
                {
                    NumThreadsInThreadPool = 2;
                }
                verify(GIOThreadPool->Create(NumThreadsInThreadPool, 16 * 1024, TPri_AboveNormal));
        }
#endif// USE_NEW_ASYNC_IO

#ifWITH_EDITOR
        // when we are in the editor we like to do things like build lighting and such
        // this thread pool can be used for those purposes
        GLargeThreadPool = FQueuedThreadPool::Allocate();
        int32 NumThreadsInLargeThreadPool = FMath::Max(FPlatformMisc::NumberOfCoresIncludingHyperthreads() - 2, 2);
                
        verify(GLargeThreadPool->Create(NumThreadsInLargeThreadPool, 128 * 1024));
#endif
}

//ThreadingBase.cpp
bool bContinueWaiting = true;
while(bContinueWaiting )
{                                
        DECLARE_SCOPE_CYCLE_COUNTER(TEXT( "FQueuedThread::Run.WaitForWork" ), STAT_FQueuedThread_Run_WaitForWork, STATGROUP_ThreadPoolAsyncTasks );
        // Wait for some work to do
        bContinueWaiting = !DoWorkEvent->Wait( 10 );
}
//windows平台下的wait
bool FEventWin::Wait(uint32 WaitTime, const bool bIgnoreThreadIdleStats/*= false*/)
{
        WaitForStats();

        SCOPE_CYCLE_COUNTER(STAT_EventWait );
        check(Event );

        FThreadIdleStats::FScopeIdleScope(bIgnoreThreadIdleStats );
        return (WaitForSingleObject( Event, WaitTime ) == WAIT_OBJECT_0);
}

      // FAutoDeleteAsyncTask
        virtual void Abandon(void)
        {
                if (Task.CanAbandon())
                {
                        Task.Abandon();
                        delete this;
                }
                else
                {
                        DoWork();
                }
        }
        // FAsyncTask
        virtual void Abandon(void)
        {
                if (Task.CanAbandon())
                {
                        Task.Abandon();
                        check(WorkNotFinishedCounter.GetValue() == 1);
                        WorkNotFinishedCounter.Decrement();
                }
                else
                {
                        DoWork();
                }
                FinishThreadedWork();
        }

/** Forwarding constructor. */
template <typename Arg0Type, typename... ArgTypes>
FAsyncTask(Arg0Type&& Arg0, ArgTypes&&... Args)
        : Task(Forward<Arg0Type>(Arg0), Forward<ArgTypes>(Args)...)
{
        Init();
}

class FMyTestTask
{
        public:
         FMyTestTask()//send in property defaults here
        {
        }
        static const TCHAR*GetTaskName()
        {
                return TEXT("FMyTestTask");
        }
        FORCEINLINE static TStatId GetStatId()
        {
                RETURN_QUICK_DECLARE_CYCLE_STAT(FMyTestTask, STATGROUP_TaskGraphTasks);
        }
        /** return the thread for this task **/
        static ENamedThreads::Type GetDesiredThread()
        {
                return ENamedThreads::AnyThread;
        }

        /*
        namespace ESubsequentsMode
       {
                enum Type
                {
                        // 存在后续任务
                        TrackSubsequents,
                        // 没有后续任务
                        FireAndForget
                };
        }
        */
        static ESubsequentsMode::Type GetSubsequentsMode()
        {
                return ESubsequentsMode::TrackSubsequents;
        }

        void DoTask(ENamedThreads::Type CurrentThread, const FGraphEventRef& MyCompletionGraphEvent)
        {
                
        }
};

FTaskGraphImplementation(int32)
{
        bCreatedHiPriorityThreads = !!ENamedThreads::bHasHighPriorityThreads;
        bCreatedBackgroundPriorityThreads = !!ENamedThreads::bHasBackgroundThreads;

        int32 MaxTaskThreads = MAX_THREADS;
        int32 NumTaskThreads = FPlatformMisc::NumberOfWorkerThreadsToSpawn();

        // if we don‘t want any performance-based threads, then force the task graph to not create any worker threads, and run in game thread
        if (!FPlatformProcess::SupportsMultithreading())
        {
                // this is the logic that used to be spread over a couple of places, that will make the rest of this function disable a worker thread
                // @todo: it could probably be made simpler/clearer
                // this - 1 tells the below code there is no rendering thread
                MaxTaskThreads = 1;
                NumTaskThreads = 1;
                LastExternalThread = (ENamedThreads::Type)(ENamedThreads::ActualRenderingThread - 1);
                bCreatedHiPriorityThreads = false;
                bCreatedBackgroundPriorityThreads = false;
                ENamedThreads::bHasBackgroundThreads = 0;
                ENamedThreads::bHasHighPriorityThreads = 0;
        }
        else
        {
                LastExternalThread = ENamedThreads::ActualRenderingThread;
        }
                
        NumNamedThreads = LastExternalThread + 1;

        NumTaskThreadSets = 1 + bCreatedHiPriorityThreads + bCreatedBackgroundPriorityThreads;

        // if we don‘t have enough threads to allow all of the sets asked for, then we can‘t create what was asked for.
        check(NumTaskThreadSets == 1 || FMath::Min<int32>(NumTaskThreads * NumTaskThreadSets + NumNamedThreads, MAX_THREADS) == NumTaskThreads * NumTaskThreadSets + NumNamedThreads);
        NumThreads = FMath::Max<int32>(FMath::Min<int32>(NumTaskThreads * NumTaskThreadSets + NumNamedThreads, MAX_THREADS), NumNamedThreads + 1);
        .......
}
//GenericPlatformMisc.cpp
int32 FGenericPlatformMisc::NumberOfWorkerThreadsToSpawn()
{
        static int32 MaxGameThreads = 4;
        static int32 MaxThreads = 16;

        int32 NumberOfCores = FPlatformMisc::NumberOfCores();//物理核数，4核8线程的机器返回的是4
        int32 MaxWorkerThreadsWanted = (IsRunningGame() || IsRunningDedicatedServer() || IsRunningClientOnly()) ? MaxGameThreads :MaxThreads;
        // need to spawn at least one worker thread (see FTaskGraphImplementation)
        return FMath::Max(FMath::Min(NumberOfCores - 1, MaxWorkerThreadsWanted), 1);
}

/** return the thread for this task **/
FORCEINLINEENamedThreads::TypeGetDesiredThread()
{
        return Context.Thread;
}

FGraphEventRef Join=TGraphTask<FVictoryTestTask>::CreateTask(NULL, ENamedThreads::GameThread).ConstructAndDispatchWhenReady();

FTaskGraphInterface::Get().WaitUntilTaskCompletes(Join, ENamedThreads::GameThread_Local);

int32 Add( int32 Amount )
{
        return FPlatformAtomics::InterlockedAdd(&Counter, Amount);
}

// named thread process tasks while we wait
TGraphTask<FReturnGraphTask>::CreateTask(&Tasks, CurrentThread).ConstructAndDispatchWhenReady(CurrentThread);
ProcessThreadUntilRequestReturn(CurrentThread);

FScopedEvent Event;
TriggerEventWhenTasksComplete(Event.Get(), Tasks, CurrentThreadIfKnown);