TaskGraph Library

TaskGraph用于实现将将多个Taskes放入多个线程执行,并且可以设定这些Task之间的依赖。引擎很多模块用到了它。现在我们就来解读一下它的实现，文章后部分再进行使用案例分析。
源码路径:

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

关键类

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.

经过分析源码，得出如下设计思路:

1. 一个TaskGraph对象会依赖多个GraphEvent对象, 也就是说该TaskGraph对象在收到所有先决事件触发后，才能执行任务;
2. 一个TaskGraph对象执行任务后，会触发它的GraphEvent, GraphEvent会尝试唤醒依赖于它的Task。
3. FTaskGraphInterface的实现负责执行TaskGraph的任务。
   综上所述, 整个系统分两个部分:

• TaskGraph和GraphEvent组成了Task依赖网络(当然不能出现循环依赖);
• FTaskGraphInterface的实现负责执行TaskGraph的任务,安排这些任务在某些线程上运行。

GraphTask依赖网络

重点代码摘录:

• 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 依赖于该事件的任务列表 **/
       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. 附加的等待事件数组 **/
       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;  //该任务执行完，需要激发该事件对象
  };
  

GraphTask与Event组成任务执行依赖网络, 注意Event对象是可以动态绑定到一个GraphTask上的，但是根据设计需求同一时刻只能只能绑定到一个GraphTask上,
这个妙用请参阅

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

的实现。
依赖网络图解:

TaskGraph_Library.jpg

任务的调度执行

class FTaskGraphInterface负责对GraphTask的调度执行，它根据GraphTask的要求安排到希望的线程上去执行。细节请阅读它的实现代码。调试如下使用案例

1. FRenderCommandFence类

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