Async.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "Async/Future.h"
#include "Async/TaskGraphInterfaces.h"
#include "Containers/UnrealString.h"
#include "CoreTypes.h"
#include "HAL/PlatformAffinity.h"
#include "HAL/PlatformProcess.h"
#include "HAL/Runnable.h"
#include "HAL/RunnableThread.h"
#include "HAL/ThreadSafeCounter.h"
#include "Misc/AssertionMacros.h"
#include "Misc/CoreStats.h"
#include "Misc/Fork.h"
#include "Misc/IQueuedWork.h"
#include "Misc/QueuedThreadPool.h"
#include "Containers/Ticker.h"
#include "Stats/Stats.h"
#include "Templates/Function.h"
#include "Templates/UnrealTemplate.h"
 
enum class EAsyncExecution
{
    TaskGraph,
 
    TaskGraphMainThread,
 
    TaskGraphMainTick,
 
    Thread,
 
    ThreadIfForkSafe,
 
    ThreadPool,
 
#if WITH_EDITOR
    LargeThreadPool
#endif
};
 
 
template<typename ResultType, typename CallableType>
inline void SetPromise(TPromise<ResultType>& Promise, CallableType&& Callable)
{
    Promise.SetValue(Forward<CallableType>(Callable)());
}
 
template<typename CallableType>
inline void SetPromise(TPromise<void>& Promise, CallableType&& Callable)
{
    Forward<CallableType>(Callable)();
    Promise.SetValue();
}
 
class FAsyncGraphTaskBase
{
public:
 
    TStatId GetStatId() const
    {
        return GET_STATID(STAT_TaskGraph_OtherTasks);
    }
 
    static ESubsequentsMode::Type GetSubsequentsMode()
    {
        return ESubsequentsMode::FireAndForget;
    }
};
 
 
template<typename ResultType>
class TAsyncGraphTask
    : public FAsyncGraphTaskBase
{
public:
 
    TAsyncGraphTask(TUniqueFunction<ResultType()>&& InFunction, TPromise<ResultType>&& InPromise, ENamedThreads::Type InDesiredThread = ENamedThreads::AnyThread)
        : Function(MoveTemp(InFunction))
        , Promise(MoveTemp(InPromise))
        , DesiredThread(InDesiredThread)
    { }
 
public:
 
    void DoTask(ENamedThreads::Type CurrentThread, const FGraphEventRef& MyCompletionGraphEvent)
    {
        SetPromise(Promise, Function);
    }
 
    ENamedThreads::Type GetDesiredThread()
    {
        return DesiredThread;
    }
 
    TFuture<ResultType> GetFuture()
    {
        return Promise.GetFuture();
    }
 
private:
 
    TUniqueFunction<ResultType()> Function;
 
    TPromise<ResultType> Promise;
 
    ENamedThreads::Type DesiredThread;
};
 
 
template<typename ResultType>
class TAsyncRunnable
    : public FRunnable
{
public:
 
    TAsyncRunnable(TUniqueFunction<ResultType()>&& InFunction, TPromise<ResultType>&& InPromise, TFuture<FRunnableThread*>&& InThreadFuture)
        : Function(MoveTemp(InFunction))
        , Promise(MoveTemp(InPromise))
        , ThreadFuture(MoveTemp(InThreadFuture))
    { }
 
public:
 
    //~ FRunnable interface
 
    virtual uint32 Run() override;
 
private:
 
    TUniqueFunction<ResultType()> Function;
 
    TPromise<ResultType> Promise;
 
    TFuture<FRunnableThread*> ThreadFuture;
};
 
 
template<typename ResultType>
class TAsyncQueuedWork
    : public IQueuedWork
{
public:
 
    TAsyncQueuedWork(TUniqueFunction<ResultType()>&& InFunction, TPromise<ResultType>&& InPromise)
        : Function(MoveTemp(InFunction))
        , Promise(MoveTemp(InPromise))
    { }
 
public:
 
    //~ IQueuedWork interface
 
    virtual void DoThreadedWork() override
    {
        SetPromise(Promise, Function);
        delete this;
    }
 
    virtual void Abandon() override
    {
        // not supported
    }
 
private:
 
    TUniqueFunction<ResultType()> Function;
 
    TPromise<ResultType> Promise;
};
 
 
struct FAsyncThreadIndex
{
    static CORE_API int32 GetNext();
};
 
 
template<typename CallableType>
auto Async(EAsyncExecution Execution, CallableType&& Callable, TUniqueFunction<void()> CompletionCallback = nullptr) -> TFuture<decltype(Forward<CallableType>(Callable)())>
{
    using ResultType = decltype(Forward<CallableType>(Callable)());
    TUniqueFunction<ResultType()> Function(Forward<CallableType>(Callable));
    TPromise<ResultType> Promise(MoveTemp(CompletionCallback));
    TFuture<ResultType> Future = Promise.GetFuture();
 
    switch (Execution)
    {
    case EAsyncExecution::TaskGraphMainThread:
        // fallthrough
    case EAsyncExecution::TaskGraph:
        {
            TGraphTask<TAsyncGraphTask<ResultType>>::CreateTask().ConstructAndDispatchWhenReady(MoveTemp(Function), MoveTemp(Promise), Execution == EAsyncExecution::TaskGraph ? ENamedThreads::AnyThread : ENamedThreads::GameThread);
        }
        break;
    case EAsyncExecution::TaskGraphMainTick:
        FTSTicker::GetCoreTicker().AddTicker(TEXT("Async_TaskGraphMainTick"), 0.0f,
            [Function = MoveTemp(Function), Promise = MoveTemp(Promise)](float) mutable
            {
                SetPromise(Promise, Function);
                return false;
            }
        );
        break;
    case EAsyncExecution::Thread:
        if (FPlatformProcess::SupportsMultithreading())
        {
            TPromise<FRunnableThread*> ThreadPromise;
            TAsyncRunnable<ResultType>* Runnable = new TAsyncRunnable<ResultType>(MoveTemp(Function), MoveTemp(Promise), ThreadPromise.GetFuture());
            
            const FString TAsyncThreadName = FString::Printf(TEXT("TAsync %d"), FAsyncThreadIndex::GetNext());
            FRunnableThread* RunnableThread = FRunnableThread::Create(Runnable, *TAsyncThreadName);
 
            check(RunnableThread != nullptr);
            check(RunnableThread->GetThreadType() == FRunnableThread::ThreadType::Real);
 
            ThreadPromise.SetValue(RunnableThread);
        }
        else
        {
            SetPromise(Promise, Function);
        }
        break;
 
    case EAsyncExecution::ThreadIfForkSafe:
        if (FPlatformProcess::SupportsMultithreading() || FForkProcessHelper::IsForkedMultithreadInstance())
        {
            TPromise<FRunnableThread*> ThreadPromise;
            TAsyncRunnable<ResultType>* Runnable = new TAsyncRunnable<ResultType>(MoveTemp(Function), MoveTemp(Promise), ThreadPromise.GetFuture());
 
            const FString TAsyncThreadName = FString::Printf(TEXT("TAsync %d"), FAsyncThreadIndex::GetNext());
            FRunnableThread* RunnableThread = FForkProcessHelper::CreateForkableThread(Runnable, *TAsyncThreadName);
 
            check(RunnableThread != nullptr);
            check(RunnableThread->GetThreadType() == FRunnableThread::ThreadType::Real);
 
            ThreadPromise.SetValue(RunnableThread);
        }
        else
        {
            SetPromise(Promise, Function);
        }
        break;
 
    case EAsyncExecution::ThreadPool:
        if (FPlatformProcess::SupportsMultithreading())
        {
            check(GThreadPool != nullptr);
            GThreadPool->AddQueuedWork(new TAsyncQueuedWork<ResultType>(MoveTemp(Function), MoveTemp(Promise)));
        }
        else
        {
            SetPromise(Promise, Function);
        }
        break;
 
#if WITH_EDITOR
    case EAsyncExecution::LargeThreadPool:
        if (FPlatformProcess::SupportsMultithreading())
        {
            check(GLargeThreadPool != nullptr);
            GLargeThreadPool->AddQueuedWork(new TAsyncQueuedWork<ResultType>(MoveTemp(Function), MoveTemp(Promise)));
        }
        else
        {
            SetPromise(Promise, Function);
        }
        break;
#endif
 
    default:
        check(false); // not implemented yet!
    }
 
    return MoveTemp(Future);
}
 
template<typename CallableType>
auto AsyncPool(FQueuedThreadPool& ThreadPool, CallableType&& Callable, TUniqueFunction<void()> CompletionCallback = nullptr, EQueuedWorkPriority InQueuedWorkPriority = EQueuedWorkPriority::Normal) -> TFuture<decltype(Forward<CallableType>(Callable)())>
{
    using ResultType = decltype(Forward<CallableType>(Callable)());
    TUniqueFunction<ResultType()> Function(Forward<CallableType>(Callable));
    TPromise<ResultType> Promise(MoveTemp(CompletionCallback));
    TFuture<ResultType> Future = Promise.GetFuture();
 
    ThreadPool.AddQueuedWork(new TAsyncQueuedWork<ResultType>(MoveTemp(Function), MoveTemp(Promise)), InQueuedWorkPriority);
 
    return MoveTemp(Future);
}
 
template<typename CallableType>
auto AsyncThread(CallableType&& Callable, uint32 StackSize = 0, EThreadPriority ThreadPri = TPri_Normal, TUniqueFunction<void()> CompletionCallback = nullptr) -> TFuture<decltype(Forward<CallableType>(Callable)())>
{
    using ResultType = decltype(Forward<CallableType>(Callable)());
    TUniqueFunction<ResultType()> Function(Forward<CallableType>(Callable));
    TPromise<ResultType> Promise(MoveTemp(CompletionCallback));
    TFuture<ResultType> Future = Promise.GetFuture();
 
    if (FPlatformProcess::SupportsMultithreading())
    {
        TPromise<FRunnableThread*> ThreadPromise;
        TAsyncRunnable<ResultType>* Runnable = new TAsyncRunnable<ResultType>(MoveTemp(Function), MoveTemp(Promise), ThreadPromise.GetFuture());
 
        const FString TAsyncThreadName = FString::Printf(TEXT("TAsyncThread %d"), FAsyncThreadIndex::GetNext());
        FRunnableThread* RunnableThread = FRunnableThread::Create(Runnable, *TAsyncThreadName, StackSize, ThreadPri);
 
        check(RunnableThread != nullptr);
 
        ThreadPromise.SetValue(RunnableThread);
    }
    else
    {
        SetPromise(Promise, Function);
    }
 
    return MoveTemp(Future);
}
 
CORE_API void AsyncTask(ENamedThreads::Type Thread, TUniqueFunction<void()> Function);
 
/* Inline functions
 *****************************************************************************/
 
template<typename ResultType>
uint32 TAsyncRunnable<ResultType>::Run()
{
    SetPromise(Promise, Function);
    FRunnableThread* Thread = ThreadFuture.Get();
 
    // Enqueue deletion of the thread to a different thread.
    Async(EAsyncExecution::TaskGraph, [Thread, this]() {
            delete Thread;
            delete this;
        }
    );
 
    return 0;
}