TaskGraph Library

TaskGraph用于實現(xiàn)將將多個Taskes放入多個線程執(zhí)行,并且可以設(shè)定這些Task之間的依賴励稳。引擎很多模塊用到了它∮蚕耍現(xiàn)在我們就來解讀一下它的實現(xiàn)杆逗，文章后部分再進行使用案例分析预茄。
源碼路徑:

• Engine\Source\Runtime\Core\Public\Async\TaskGraphInterfaces.h
• Engine\Source\Runtime\Core\Private\Async\TaskGraph.cpp

關(guān)鍵類

1. FTaskGraphInterface
   Interface tot he task graph system.
2. FBaseGraphTask
   Base class for all tasks.
3. FGraphEvent
   A FGraphEvent is a list of tasks waiting for something. These tasks are call the subsequents. A graph event is a prerequisite for each of its subsequents. Graph events have a lifetime managed by reference counting.
4. template< typename TTask> class TGraphTask
   Embeds a user defined task, as exemplified above, for doing the work and provides the functionality for setting up and handling prerequisites and subsequents.
5. FReturnGraphTask
   a task used to return flow control from a named thread back to the original caller of ProcessThreadUntilRequestReturn.
6. FNullGraphTask
   a task that does nothing. It can be used to "gather" tasks into one prerequisite.
7. FTriggerEventGraphTask
   a task that triggers an event(operating system Event object)
8. FSimpleDelegateGraphTask
   for simple delegate based tasks. This is less efficient than a custom task, doesn't provide the task arguments, doesn't allow specification of the current thread, etc.
9. FDelegateGraphTask
   class for more full featured delegate based tasks. Still less efficient than a custom task, but provides all of the args.
10. FFunctionGraphTask
    Task class for lambda based tasks.
11. FCompletionList
    List of tasks that can be "joined" into one task which can be waited on or used as a prerequisite.

經(jīng)過分析源碼狠怨，得出如下設(shè)計思路:

1. 一個TaskGraph對象會依賴多個GraphEvent對象, 也就是說該TaskGraph對象在收到所有先決事件觸發(fā)后坑资，才能執(zhí)行任務(wù);
2. 一個TaskGraph對象執(zhí)行任務(wù)后运敢，會觸發(fā)它的GraphEvent, GraphEvent會嘗試喚醒依賴于它的Task校仑。
3. FTaskGraphInterface的實現(xiàn)負責(zé)執(zhí)行TaskGraph的任務(wù)。
   綜上所述, 整個系統(tǒng)分兩個部分:

• TaskGraph和GraphEvent組成了Task依賴網(wǎng)絡(luò)(當(dāng)然不能出現(xiàn)循環(huán)依賴);
• FTaskGraphInterface的實現(xiàn)負責(zé)執(zhí)行TaskGraph的任務(wù),安排這些任務(wù)在某些線程上運行传惠。

GraphTask依賴網(wǎng)絡(luò)

重點代碼摘錄:

• FGraphEvent

  class FGraphEvent
  {
   public:
       bool AddSubsequent(class FBaseGraphTask* Task)
       {
           return SubsequentList.PushIfNotClosed(Task);
       }
  
       /**
        *  Delay the firing of this event until the given event fires.
        *  CAUTION: This is only legal while executing the task associated with this event.
        *  @param EventToWaitFor event to wait for until we fire.
       **/
       void DontCompleteUntil(FGraphEventRef EventToWaitFor)
       {
           checkThreadGraph(!IsComplete()); // it is not legal to add a DontCompleteUntil after the event has been completed. Basically, this is only legal within a task function.
           new (EventsToWaitFor) FGraphEventRef(EventToWaitFor);
       }
  
       /**
        *  "Complete" the event. This grabs the list of subsequents and atomically closes it. Then for each subsequent it reduces the number of prerequisites outstanding and if that drops to zero, the task is queued.
        *  @param CurrentThreadIfKnown if the current thread is known, provide it here. Otherwise it will be determined via TLS if any task ends up being queued.
       **/
       CORE_API void DispatchSubsequents(TArray<FBaseGraphTask*>& NewTasks, ENamedThreads::Type CurrentThreadIfKnown = ENamedThreads::AnyThread);
  
       bool IsComplete() const
       {
           return SubsequentList.IsClosed();
       }
  
   private:
  
       /** Threadsafe list of subsequents for the event 依賴于該事件的任務(wù)列表 **/
       TClosableLockFreePointerListUnorderedSingleConsumer<FBaseGraphTask, 0>      SubsequentList;
       /** List of events to wait for until firing. This is not thread safe as it is only legal to fill it in within the context of an executing task. 附加的等待事件數(shù)組 **/
       FGraphEventArray                                                        EventsToWaitFor;
       /** Number of outstanding references to this graph event **/
       FThreadSafeCounter                                                      ReferenceCount;
   };
  

• FBaseGraphTask

   class FBaseGraphTask
   {
   public:
       FBaseGraphTask(
           int32 InNumberOfPrerequistitesOutstanding
           )
           : ThreadToExecuteOn(ENamedThreads::AnyThread)
           , NumberOfPrerequistitesOutstanding(InNumberOfPrerequistitesOutstanding + 1) // + 1 is not a prerequisite, it is a lock to prevent it from executing while it is getting prerequisites, one it is safe to execute, call PrerequisitesComplete
       {
           checkThreadGraph(LifeStage.Increment() == int32(LS_Contructed));
       }
       
       void PrerequisitesComplete(ENamedThreads::Type CurrentThread, int32 NumAlreadyFinishedPrequistes, bool bUnlock = true)
       {
           checkThreadGraph(LifeStage.Increment() == int32(LS_PrequisitesSetup));
           int32 NumToSub = NumAlreadyFinishedPrequistes + (bUnlock ? 1 : 0); // the +1 is for the "lock" we set up in the constructor
           if (NumberOfPrerequistitesOutstanding.Subtract(NumToSub) == NumToSub) 
           {
               QueueTask(CurrentThread);
           }
       }
       void ConditionalQueueTask(ENamedThreads::Type CurrentThread)
       {
           if (NumberOfPrerequistitesOutstanding.Decrement()==0)
           {
               QueueTask(CurrentThread);
           }
       }
  
       // Subclass API
       FORCEINLINE void Execute(TArray<FBaseGraphTask*>& NewTasks, ENamedThreads::Type CurrentThread)
       {
           checkThreadGraph(LifeStage.Increment() == int32(LS_Executing));
           ExecuteTask(NewTasks, CurrentThread);
       }
       // Internal Use
       void QueueTask(ENamedThreads::Type CurrentThreadIfKnown)
       {
           checkThreadGraph(LifeStage.Increment() == int32(LS_Queued));
           FTaskGraphInterface::Get().QueueTask(this, ThreadToExecuteOn, CurrentThreadIfKnown);
       }
  
       /** Thread to execute on, can be ENamedThreads::AnyThread to execute on any unnamed thread **/
       ENamedThreads::Type         ThreadToExecuteOn;
       /** Number of prerequisites outstanding. When this drops to zero, the thread is queued for execution.  **/
       FThreadSafeCounter          NumberOfPrerequistitesOutstanding; 
   };
  

• TGraphTask

    /** 
    *  TGraphTask
    *  Embeds a user defined task, as exemplified above, for doing the work and provides the functionality for setting up and handling prerequisites and subsequents
    **/
   template<typename TTask>
   class TGraphTask : public FBaseGraphTask
   {
   public:
       virtual void ExecuteTask(TArray<FBaseGraphTask*>& NewTasks, ENamedThreads::Type CurrentThread) final override
       {
           checkThreadGraph(TaskConstructed);
  
           // Fire and forget mode must not have subsequents
           // Track subsequents mode must have subsequents
           checkThreadGraph(XOR(TTask::GetSubsequentsMode() == ESubsequentsMode::FireAndForget, IsValidRef(Subsequents)));
  
           if (TTask::GetSubsequentsMode() == ESubsequentsMode::TrackSubsequents)
           {
               Subsequents->CheckDontCompleteUntilIsEmpty(); // we can only add wait for tasks while executing the task
           }
  
           TTask& Task = *(TTask*)&TaskStorage;
           {
               FScopeCycleCounter Scope(Task.GetStatId(), true); 
               Task.DoTask(CurrentThread, Subsequents);
               Task.~TTask();
               checkThreadGraph(ENamedThreads::GetThreadIndex(CurrentThread) <= ENamedThreads::RenderThread || FMemStack::Get().IsEmpty()); // you must mark and pop memstacks if you use them in tasks! Named threads are excepted.
           }
  
           TaskConstructed = false;
  
           if (TTask::GetSubsequentsMode() == ESubsequentsMode::TrackSubsequents)
           {
               FPlatformMisc::MemoryBarrier();
               Subsequents->DispatchSubsequents(NewTasks, CurrentThread);
           }
  
           if (sizeof(TGraphTask) <= FBaseGraphTask::SMALL_TASK_SIZE)
           {
               this->TGraphTask::~TGraphTask();
               FBaseGraphTask::GetSmallTaskAllocator().Free(this);
           }
           else
           {
               delete this;
           }
       }
       
       /** An aligned bit of storage to hold the embedded task **/
       TAlignedBytes<sizeof(TTask),ALIGNOF(TTask)> TaskStorage;
       /** Used to sanity check the state of the object **/
       bool                        TaskConstructed;
       /** A reference counted pointer to the completion event which lists the tasks that have me as a prerequisite. **/
       FGraphEventRef              Subsequents;  //該任務(wù)執(zhí)行完迄沫，需要激發(fā)該事件對象
  };
  

GraphTask與Event組成任務(wù)執(zhí)行依賴網(wǎng)絡(luò), 注意Event對象是可以動態(tài)綁定到一個GraphTask上的，但是根據(jù)設(shè)計需求同一時刻只能只能綁定到一個GraphTask上,
這個妙用請參閱

void FGraphEvent::DispatchSubsequents(TArray<FBaseGraphTask*>& NewTasks, ENamedThreads::Type CurrentThreadIfKnown = ENamedThreads::AnyThread)

的實現(xiàn)涉枫。
依賴網(wǎng)絡(luò)圖解:

TaskGraph_Library.jpg

任務(wù)的調(diào)度執(zhí)行

class FTaskGraphInterface負責(zé)對GraphTask的調(diào)度執(zhí)行邢滑，它根據(jù)GraphTask的要求安排到希望的線程上去執(zhí)行。細節(jié)請閱讀它的實現(xiàn)代碼愿汰。調(diào)試如下使用案例

1. FRenderCommandFence類

    // Issue a fence command to the rendering thread and wait for it to complete.
    FRenderCommandFence Fence;
    Fence.BeginFence();
    Fence.Wait();