Unreal Engine Async and Threading

#include "Tasks/Task.h"
// Tasks/Task.h:299
template<typename TaskBodyType>
UE::Tasks::TTask<TInvokeResult_T<TaskBodyType>> Launch(const TCHAR* DebugName, TaskBodyType&& TaskBody,
    ETaskPriority Priority = ETaskPriority::Normal,
    EExtendedTaskPriority ExtendedPriority = EExtendedTaskPriority::None,
    ETaskFlags Flags = ETaskFlags::None);
// Tasks/Task.h:324 — same, with a prerequisites collection as the third parameter
template<typename TaskBodyType, typename PrerequisitesCollectionType>
UE::Tasks::TTask<TInvokeResult_T<TaskBodyType>> Launch(const TCHAR* DebugName, TaskBodyType&& TaskBody,
    PrerequisitesCollectionType&& Prerequisites,
    ETaskPriority Priority = ETaskPriority::Normal,
    EExtendedTaskPriority ExtendedPriority = EExtendedTaskPriority::None,
    ETaskFlags Flags = ETaskFlags::None);

using namespace UE::Tasks;

TArray<int32> Data;
TTask<int32> Sum = Launch(UE_SOURCE_LOCATION, [Data]() { return ComputeSum(Data); });

// Prerequisites: TaskB runs after TaskA completes. Handles are copyable; GetResult() is non-const.
TTask<FVector> TaskA = Launch(UE_SOURCE_LOCATION, []() { return FVector(1.f, 2.f, 3.f); });
TTask<void> TaskB = Launch(UE_SOURCE_LOCATION,
    [TaskA]() mutable { const FVector Pos = TaskA.GetResult(); ConsumePosition(Pos); },
    Prerequisites(TaskA), ETaskPriority::BackgroundNormal);

// Manual gate: nothing after the event starts until Trigger()
FTaskEvent Gate{ UE_SOURCE_LOCATION };
TTask<void> Gated = Launch(UE_SOURCE_LOCATION, []() { DoWork(); }, Prerequisites(Gate));
Gate.Trigger();

// Wait for a group, with timeout (Tasks/Task.h:393)
TArray<FTask> Group{ TaskB, Gated };
const bool bAllDone = Wait(Group, FTimespan::FromMilliseconds(5.0));

// Cooperative cancellation (Tasks/Task.h: FCancellationToken)
FCancellationToken Token;
Launch(UE_SOURCE_LOCATION, [&Token]() { for (int32 i = 0; i < 1000; ++i) { if (Token.IsCanceled()) { return; } Step(i); } });
Token.Cancel();

// Already-completed task holding a value (useful for uniform interfaces)
TTask<int32> Ready = MakeCompletedTask<int32>(42);

#include "Tasks/Pipe.h"
#include "Tasks/TaskConcurrencyLimiter.h"

UE::Tasks::FPipe SavePipe{ TEXT("SavePipe") };                       // explicit FPipe(const TCHAR* InDebugName)
UE::Tasks::TTask<void> First = SavePipe.Launch(UE_SOURCE_LOCATION, []() { WriteChunk(0); });
UE::Tasks::TTask<bool> Second = SavePipe.Launch(UE_SOURCE_LOCATION, []() { return WriteChunk(1); },
    UE::Tasks::ETaskPriority::BackgroundNormal);
const bool bInsidePipe = SavePipe.IsInContext();                     // true only while a pipe task runs on this thread
SavePipe.WaitUntilEmpty();                                           // before destroying the pipe

UE::Tasks::FTaskConcurrencyLimiter Limiter(4 /*MaxConcurrency*/, UE::Tasks::ETaskPriority::BackgroundHigh);
for (int32 Index = 0; Index < 64; ++Index)
{
    Limiter.Push(UE_SOURCE_LOCATION, [Index](uint32 Slot) { ProcessWithScratch(Index, Slot); }); // Slot in [0, MaxConcurrency)
}
Limiter.Wait();                                                      // Wait(FTimespan Timeout = FTimespan::MaxValue())

#include "Async/Async.h"
// Async/Async.h:299
template<typename CallableType>
auto Async(EAsyncExecution Execution, CallableType&& Callable, TUniqueFunction<void()> CompletionCallback = nullptr) -> TFuture<decltype(Forward<CallableType>(Callable)())>;
// Async/Async.h:407 — takes a reference, so pass *GThreadPool
template<typename CallableType>
auto AsyncPool(FQueuedThreadPool& ThreadPool, CallableType&& Callable, TUniqueFunction<void()> CompletionCallback = nullptr, EQueuedWorkPriority InQueuedWorkPriority = EQueuedWorkPriority::Normal);
// Async/Async.h:430
template<typename CallableType>
auto AsyncThread(CallableType&& Callable, uint32 StackSize = 0, EThreadPriority ThreadPri = TPri_Normal, TUniqueFunction<void()> CompletionCallback = nullptr);
// Async/Async.h:463
CORE_API void AsyncTask(ENamedThreads::Type Thread, TUniqueFunction<void()> Function);

TFuture<FMyResult> Future = Async(EAsyncExecution::ThreadPool, []() { return ComputeResult(); });
if (Future.IsReady()) { UseResult(Future.Get()); }       // non-blocking check
FMyResult Copy = Future.Get();                            // blocks; does NOT invalidate (Async/Future.h:226)
Future.Next([](FMyResult Value) { UseResult(Value); });   // continuation receives the value; runs on the completing thread

TPromise<FMyResult> Promise;
TFuture<FMyResult> FromPromise = Promise.GetFuture();   // call once
Async(EAsyncExecution::Thread, [P = MoveTemp(Promise)]() mutable { P.SetValue(ComputeResult()); });

AMyActor* MyActor = FindMyActor();
AsyncTask(ENamedThreads::GameThread, [WeakActor = TWeakObjectPtr<AMyActor>(MyActor), Copy]()
{
    if (AMyActor* Actor = WeakActor.Get()) { Actor->ApplyResult(Copy); }
});

#include "Async/AsyncWork.h"

class FMyChunkTask : public FNonAbandonableTask
{
public:
    friend class FAsyncTask<FMyChunkTask>;
    friend class FAutoDeleteAsyncTask<FMyChunkTask>;

    explicit FMyChunkTask(TArray<int32> InInput) : Input(MoveTemp(InInput)) {}

    int32 Result = 0;

    void DoWork()
    {
        for (int32 Value : Input) { Result += Value; }
    }

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

private:
    TArray<int32> Input;
};

// Owner-managed lifetime
TArray<int32> Numbers;
FAsyncTask<FMyChunkTask>* Task = new FAsyncTask<FMyChunkTask>(MoveTemp(Numbers));
Task->StartBackgroundTask();                       // (FQueuedThreadPool* = GThreadPool, EQueuedWorkPriority = Normal, EQueuedWorkFlags = None, int64 RequiredMemory = -1, const TCHAR* DebugName = nullptr)
const bool bReady = Task->IsDone();                // poll once per frame, never spin
Task->EnsureCompletion();                          // (bool bDoWorkOnThisThreadIfNotStarted = true, bool bIsLatencySensitive = false)
const int32 Sum = Task->GetTask().Result;
delete Task;

// Fire-and-forget: deletes itself after DoWork
(new FAutoDeleteAsyncTask<FMyChunkTask>(MoveTemp(Numbers)))->StartBackgroundTask();

#include "Async/ParallelFor.h"
// Async/ParallelFor.h:526, :543
inline void ParallelFor(int32 Num, TFunctionRef<void(int32)> Body, EParallelForFlags Flags = EParallelForFlags::None);
inline void ParallelFor(const TCHAR* DebugName, int32 Num, int32 MinBatchSize, TFunctionRef<void(int32)> Body, EParallelForFlags Flags = EParallelForFlags::None);
// Async/ParallelFor.h:792 — Body is called as Body(ContextType&, int32 Index)
template <typename ContextType, typename ContextAllocatorType, typename FunctionType>
inline void ParallelForWithTaskContext(const TCHAR* DebugName, TArray<ContextType, ContextAllocatorType>& OutContexts, int32 Num, int32 MinBatchSize, const FunctionType& Body, EParallelForFlags Flags = EParallelForFlags::None);

TArray<UStaticMesh*> Meshes;
ParallelFor(Meshes.Num(), [&Meshes](int32 Index) { ProcessMesh(Meshes[Index]); });

ParallelFor(TEXT("ProcessMeshes"), Meshes.Num(), 64, [&Meshes](int32 Index) { ProcessMesh(Meshes[Index]); },
    EParallelForFlags::Unbalanced | EParallelForFlags::BackgroundPriority);

struct FMyScratch { TArray<FVector> Buffer; };
TArray<FMyScratch> Contexts;                                          // one per worker task, reused across iterations
ParallelForWithTaskContext(TEXT("Normals"), Contexts, Meshes.Num(), 32,
    [&Meshes](FMyScratch& Scratch, int32 Index) { Scratch.Buffer.Reset(); ComputeNormals(Meshes[Index], Scratch.Buffer); });

// HAL/Runnable.h:32-69 — override verbatim
virtual bool Init();
virtual uint32 Run() = 0;
virtual void Stop();
virtual void Exit();
virtual class FSingleThreadRunnable* GetSingleThreadInterface();   // return a fallback for -nothreading platforms, or nullptr

// HAL/RunnableThread.h:44
static CORE_API FRunnableThread* Create(
    class FRunnable* InRunnable,
    const TCHAR* ThreadName,
    uint32 InStackSize = 0,
    EThreadPriority InThreadPri = TPri_Normal,
    uint64 InThreadAffinityMask = FPlatformAffinity::GetNoAffinityMask(),
    EThreadCreateFlags InCreateFlags = EThreadCreateFlags::None);
virtual bool Kill(bool bShouldWait = true) = 0;    // calls Stop(); with bShouldWait blocks until Run() returns
virtual void WaitForCompletion() = 0;
virtual void Suspend(bool bShouldPause = true) = 0;
virtual void SetThreadPriority(EThreadPriority NewPriority) = 0;

TArray<FVector> Points;
mutable FCriticalSection Mutex;              // mutable so const getters can lock
void Add(const FVector& P) { FScopeLock Lock(&Mutex); Points.Add(P); }

TMap<FName, FVector> Cache;
mutable FRWLock CacheLock;
FVector Read(FName Key) const { FReadScopeLock Lock(CacheLock); return Cache.FindRef(Key); }
void Write(FName Key, FVector V) { FWriteScopeLock Lock(CacheLock); Cache.Add(Key, V); }

int32 Counter = 0;
UE::FMutex SmallMutex;
void Bump() { UE::TUniqueLock Lock(SmallMutex); ++Counter; }