AsyncLoadingThread.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
/*=============================================================================
    AsyncLoadingThread.h: Unreal async loading code.
=============================================================================*/
#pragma once
 
#include "CoreMinimal.h"
#include "Algo/AnyOf.h"
#include "HAL/ThreadSafeCounter.h"
#include "UObject/ObjectMacros.h"
#include "Serialization/AsyncLoading.h"
#include "Serialization/LoadTimeTracePrivate.h"
#include "Misc/ScopeLock.h"
#include "Misc/CommandLine.h"
#include "Misc/App.h"
#include "HAL/Runnable.h"
#include "HAL/ThreadSafeBool.h"
#include "Misc/ConfigCacheIni.h"
#include "HAL/LowLevelMemTracker.h"
#include "Serialization/AsyncPackageLoader.h"
#include "AutoRTFM.h"
 
#include <atomic>
 
struct FPrecacheCallbackHandler;
 
struct FAsyncLoadEvent
{
    enum
    {
        EventSystemPriority_MAX = MAX_int32
    };
 
    int32 UserPriority;
    int32 PackageSerialNumber;
    int32 EventSystemPriority;
    int32 SerialNumber;
    TFunction<void(FAsyncLoadEventArgs& Args)> Payload;
 
    FAsyncLoadEvent()
        : UserPriority(0)
        , PackageSerialNumber(0)
        , EventSystemPriority(0)
        , SerialNumber(0)
    {
    }
    FAsyncLoadEvent(int32 InUserPriority, int32 InPackageSerialNumber, int32 InEventSystemPriority, int32 InSerialNumber, TFunction<void(FAsyncLoadEventArgs& Args)>&& InPayload)
        : UserPriority(InUserPriority)
        , PackageSerialNumber(InPackageSerialNumber)
        , EventSystemPriority(InEventSystemPriority)
        , SerialNumber(InSerialNumber)
        , Payload(Forward<TFunction<void(FAsyncLoadEventArgs& Args)>>(InPayload))
    {
    }
 
    FORCEINLINE bool operator<(const FAsyncLoadEvent& Other) const
    {
        if (UserPriority != Other.UserPriority)
        {
            return UserPriority > Other.UserPriority;
        }
        if (EventSystemPriority != Other.EventSystemPriority)
        {
            return EventSystemPriority > Other.EventSystemPriority;
        }
        if (PackageSerialNumber != Other.PackageSerialNumber)
        {
            return PackageSerialNumber > Other.PackageSerialNumber; // roughly DFS
        }
        return SerialNumber < Other.SerialNumber;
    }
};
 
struct FAsyncLoadEventQueue
{
    //FCriticalSection AsyncLoadEventQueueCritical; //@todoio maybe this doesn't need to be protected by a critical section. 
    int32 RunningSerialNumber;
    TArray<FAsyncLoadEvent> EventQueue;
 
    FAsyncLoadEventQueue()
        : RunningSerialNumber(0)
    {
    }
 
    FORCEINLINE void AddAsyncEvent(int32 UserPriority, int32 PackageSerialNumber, int32 EventSystemPriority, TFunction<void(FAsyncLoadEventArgs& Args)>&& Payload)
    {
        //UE::TScopeLock Lock(AsyncLoadEventQueueCritical);
        //@todoio check(FAsyncLoadingThread::IsInAsyncLoadThread());
        EventQueue.HeapPush(FAsyncLoadEvent(UserPriority, PackageSerialNumber, EventSystemPriority, ++RunningSerialNumber, Forward<TFunction<void(FAsyncLoadEventArgs& Args)>>(Payload)));
    }
    bool PopAndExecute(FAsyncLoadEventArgs& Args)
    {
        FAsyncLoadEvent Event;
        bool bResult = false;
        {
            //UE::TScopeLock Lock(AsyncLoadEventQueueCritical);
            //@todoio check(FAsyncLoadingThread::IsInAsyncLoadThread());
            if (EventQueue.Num())
            {
                EventQueue.HeapPop(Event, EAllowShrinking::No);
                bResult = true;
            }
        }
        if (bResult)
        {
            Event.Payload(Args);
        }
        return bResult;
    }
};
 
// Shared data for FFlushRequest
struct FFlushRequestData
{
    FFlushRequestData(TConstArrayView<int32> InRequestIDs)
        : RequestIDs(MakeUnique<std::atomic<int32>[]>(InRequestIDs.Num()))
        , NumRequests(InRequestIDs.Num())
    {
        for (int32 i=0; i < InRequestIDs.Num(); ++i)
        {
            RequestIDs.Get()[i] = InRequestIDs[i];
        }
    }
    
    TConstArrayView<std::atomic<int32>> GetRequestIDs() const
    {
        return MakeArrayView(RequestIDs.Get(), NumRequests);
    }
 
    TArrayView<std::atomic<int32>> GetRequestIDs() 
    {
        return MakeArrayView(RequestIDs.Get(), NumRequests);
    }
    
    void OnRequestComplete(int32 InCompletedRequest)
    {
        for (std::atomic<int32>& RequestID : GetRequestIDs())
        {
            if (RequestID.load(std::memory_order_relaxed) == InCompletedRequest)
            {
                ++NumCompletedRequests;
            }
        }
    }
    
    bool IsComplete()
    {
        return NumRequests == NumCompletedRequests;
    }
 
private:
    TUniquePtr<std::atomic<int32>[]> RequestIDs;    
    int32 NumRequests;
    int32 NumCompletedRequests = 0;
};
 
struct FFlushRequest
{
    FFlushRequest() = default;
 
    FFlushRequest(TConstArrayView<int32> InRequestIDs)
        : Data(MakeShared<FFlushRequestData>(InRequestIDs))
    {
    }
    
    TConstArrayView<std::atomic<int32>> GetRequestIDs() const 
    {
        return Data->GetRequestIDs();
    }
 
    bool IsValid() const
    {
        return Data.IsValid();
    }
    
    bool IsComplete() const
    {
        return Data->IsComplete();
    }
 
    operator bool() const
    {
        return IsValid();
    }
 
private:
    void OnRequestComplete(int32 RequestID)
    {
        Data->OnRequestComplete(RequestID); 
    }
 
    void AdjustRequestIDs(TMap<int32, int32>& DuplicateRequestMap)
    {
        for (std::atomic<int32>& RequestID : Data->GetRequestIDs())
        {   
            if (int32* MainRequest = DuplicateRequestMap.Find(RequestID.load(std::memory_order_relaxed)))
            {
                RequestID.store(*MainRequest, std::memory_order_relaxed);
            }
        }
    }
 
    friend class FAsyncLoadingThread;
 
    TSharedPtr<FFlushRequestData, ESPMode::ThreadSafe> Data;
};
 
struct FMaxPackageSummarySize
{
    static int32 Value;
    static void Init();
};
 
class FAsyncLoadingThread final : public FRunnable, public IAsyncPackageLoader
{
    friend struct FAsyncPackage;
 
    struct FEDLBootNotificationManager& EDLBootNotificationManager;
 
    FRunnableThread* Thread;
    FThreadSafeCounter StopTaskCounter;
 
    static bool bThreadStarted;
 
    FEvent* QueuedRequestsEvent;
    FEvent* CancelLoadingEvent;
    FEvent* ThreadSuspendedEvent;
    FEvent* ThreadResumedEvent;
    TArray<FAsyncPackageDesc*> QueuedPackages;
#if THREADSAFE_UOBJECTS
    FCriticalSection QueueCritical;
#endif
    FThreadSafeBool bShouldCancelLoading;
    FThreadSafeCounter IsLoadingSuspended;
    TArray<FAsyncPackage*> LoadedPackages;  
    TMap<FName, FAsyncPackage*> LoadedPackagesNameLookup;
#if THREADSAFE_UOBJECTS
    FCriticalSection LoadedPackagesCritical;
#endif
    TArray<FAsyncPackage*> LoadedPackagesToProcess;
    TArray<FAsyncPackage*> PackagesToDelete;
    TMap<FName, FAsyncPackage*> LoadedPackagesToProcessNameLookup;
#if WITH_EDITOR
    TSet<FWeakObjectPtr> LoadedAssets;
#endif
#if THREADSAFE_UOBJECTS
    FCriticalSection LoadedPackagesToProcessCritical;
#endif
 
    TArray<FFlushRequest, TInlineAllocator<4>> FlushRequests;
#if THREADSAFE_UOBJECTS
    FCriticalSection FlushRequestCritical;
#endif
    TMap<int32, int32> DuplicateRequestMap;
 
    TArray<FAsyncPackage*> AsyncPackages;
    TMap<FName, FAsyncPackage*> AsyncPackageNameLookup;
public:
    TArray<FAsyncPackage*> AsyncPackagesReadyForTick;
private:
 
#if THREADSAFE_UOBJECTS
    FCriticalSection AsyncPackagesCritical;
#endif
 
    TSet<int32> PendingRequests;
#if THREADSAFE_UOBJECTS
    FCriticalSection PendingRequestsCritical;
#endif
 
    FThreadSafeCounter QueuedPackagesCounter;
    FThreadSafeCounter ExistingAsyncPackagesCounter;
 
    FThreadSafeCounter AsyncThreadReady;
 
    TArray<FAsyncPackageDesc*> QueuedPackagesToCancel;
    TSet<FAsyncPackage*> PackagesToCancel;
 
    static uint32 AsyncLoadingThreadID;
 
    enum class EAsyncPackageInsertMode
    {
        InsertBeforeMatchingPriorities, // Insert this package before all other packages of the same priority
        InsertAfterMatchingPriorities   // Insert this package after all other packages of the same priority
    };
 
#if LOOKING_FOR_PERF_ISSUES
    static FThreadSafeCounter BlockingCycles;
#endif
 
    IAsyncPackageLoader* IoStorePackageLoader = nullptr;
 
public:
 
    FAsyncLoadingThread(int32 InThreadIndex);
    virtual ~FAsyncLoadingThread();
 
    virtual ELoaderType GetLoaderType() const override
    {
        return ELoaderType::LegacyLoader;
    }
 
    IAsyncPackageLoader* GetIoStorePackageLoader() const
    {
        return IoStorePackageLoader;
    }
 
    void SetIoStorePackageLoader(IAsyncPackageLoader* InIoStorePackageLoader)
    {
        IoStorePackageLoader = InIoStorePackageLoader;
    }
 
    //~ Begin FRunnable Interface.
    virtual bool Init();
    virtual uint32 Run();
    virtual void Stop();
    //~ End FRunnable Interface
 
    static FAsyncLoadingThread& Get()
    {
        check(Instance);
        return *Instance;
    }
 
    void StartThread() override;
 
    bool ShouldAlwaysLoadPackageAsync(const FPackagePath& InPackagePath) override;
 
    int32 LoadPackage(const FPackagePath& PackagePath, FLoadPackageAsyncOptionalParams OptionalParams) override;
 
    int32 LoadPackage(
            const FPackagePath& InPackagePath,
            FName InCustomName,
            FLoadPackageAsyncDelegate InCompletionDelegate,
            EPackageFlags InPackageFlags,
            int32 InPIEInstanceID,
            int32 InPackagePriority,
            const FLinkerInstancingContext* InInstancingContext,
            uint32 InLoadFlags) override;
 
    EAsyncPackageState::Type ProcessLoading(bool bUseTimeLimit, bool bUseFullTimeLimit, double TimeLimit) override;
 
    EAsyncPackageState::Type ProcessLoadingUntilComplete(TFunctionRef<bool()> CompletionPredicate, double TimeLimit) override;
 
    void FlushLoading(TConstArrayView<int32> RequestIDs) override;
 
    void NotifyConstructedDuringAsyncLoading(UObject* Object, bool bSubObject) override;
 
    void NotifyUnreachableObjects(const TArrayView<FUObjectItem*>& UnreachableObjects) override {};
 
    void FireCompletedCompiledInImport(void* AsyncPacakge, FPackageIndex Import);
 
    void NotifyRegistrationEvent(FName PackageName, FName Name, ENotifyRegistrationType NotifyRegistrationType, ENotifyRegistrationPhase NotifyRegistrationPhase, UObject* (*InRegister)(), bool InbDynamic, UObject* FinishedObject) override;
 
    void NotifyRegistrationComplete() override;
 
    FAsyncLoadEventQueue EventQueue;
 
    FORCEINLINE FAsyncPackage* GetPackage(FWeakAsyncPackagePtr Ptr)
    {
        if (Ptr.PackageName != NAME_None && Ptr.SerialNumber)
        {
            FAsyncPackage* Package = FindAsyncPackage(Ptr.PackageName);
            if (Package && Package->SerialNumber == Ptr.SerialNumber)
            {
                return Package;
            }
        }
        return nullptr;
    }
 
    void QueueEvent_CreateLinker(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_FinishLinker(FWeakAsyncPackagePtr WeakPtr, int32 EventSystemPriority = 0);
    void QueueEvent_StartImportPackages(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_SetupImports(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_SetupExports(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_ProcessImportsAndExports(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_ExportsDone(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_ProcessPostloadWait(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
    void QueueEvent_StartPostLoad(FAsyncPackage* Pkg, int32 EventSystemPriority = 0);
 
    FORCEINLINE bool IsMultithreaded() override
    {
        return bThreadStarted;
    }
 
    static void EnterAsyncLoadingTick(int32 ThreadIndex)
    {
        AsyncLoadingTickCounter.Increment();
        check(CurrentAsyncLoadingTickThreadIndex == INDEX_NONE || CurrentAsyncLoadingTickThreadIndex == ThreadIndex);
        CurrentAsyncLoadingTickThreadIndex = ThreadIndex;
    }
 
    static void LeaveAsyncLoadingTick(int32 ThreadIndex)
    {
        int32 AsyncLoadingTickCounterValue = AsyncLoadingTickCounter.Decrement();
        check(AsyncLoadingTickCounterValue >= 0);
        check(CurrentAsyncLoadingTickThreadIndex == ThreadIndex);
        if (AsyncLoadingTickCounterValue == 0)
        {
            CurrentAsyncLoadingTickThreadIndex = INDEX_NONE;
        }
    }
 
    UE_AUTORTFM_ALWAYS_OPEN static bool GetIsInAsyncLoadingTick()
    {
        return !!AsyncLoadingTickCounter.GetValue();
    }
 
    UE_AUTORTFM_ALWAYS_OPEN bool IsAsyncLoadingPackages() override
    {
        return QueuedPackagesCounter.GetValue() != 0 || ExistingAsyncPackagesCounter.GetValue() != 0;
    }
 
    FORCEINLINE bool IsInAsyncLoadThread() override
    {
        bool bResult = false;
        if (IsMultithreaded())
        {
            // We still need to report we're in async loading thread even if 
            // we're on game thread but inside of async loading code (PostLoad mostly)
            // to make it behave exactly like the non-threaded version
            bResult = FPlatformTLS::GetCurrentThreadId() == AsyncLoadingThreadID || 
                (IsInGameThread() && GetIsInAsyncLoadingTick());
        }
        else
        {
            bResult = IsInGameThread() && GetIsInAsyncLoadingTick();
        }
        return bResult;
    }
 
    FORCEINLINE bool IsAsyncLoadingSuspendedInternal() const
    {
        return !!IsLoadingSuspended.GetValue();
    }
 
    virtual bool IsAsyncLoadingSuspended() override
    {
        return IsAsyncLoadingSuspendedInternal();
    }
 
    FORCEINLINE int32 GetAsyncLoadingSuspendedCount()
    {
        FPlatformMisc::MemoryBarrier();
        return IsLoadingSuspended.GetValue();
    }
 
    FORCEINLINE int32 GetNumQueuedPackages() override
    {
        return QueuedPackagesCounter.GetValue();
    }
 
    FORCEINLINE int32 GetNumAsyncPackages() override
    {
        return ExistingAsyncPackagesCounter.GetValue();
    }
 
    FORCEINLINE FAsyncPackage* FindAsyncPackage(const FName& PackageName)
    {
        checkSlow(IsInAsyncLoadThread());
        return AsyncPackageNameLookup.FindRef(PackageName);
    }
 
    void InsertPackage(FAsyncPackage* Package, bool bReinsert = false, EAsyncPackageInsertMode InsertMode = EAsyncPackageInsertMode::InsertBeforeMatchingPriorities);
 
    FORCEINLINE FAsyncPackage* FindLoadedPackage(const FName& PackageName)
    {
        checkSlow(IsInAsyncLoadThread());
        return LoadedPackagesNameLookup.FindRef(PackageName);
    }
 
    void QueuePackage(FAsyncPackageDesc& Package);
 
    void CancelLoading() override;
 
    void ShutdownLoading() override;
 
    void SuspendLoading() override;
 
    void ResumeLoading() override;
 
    FORCEINLINE FAsyncPackage* GetPackage(int32 PackageIndex)
    {
        checkSlow(IsInAsyncLoadThread());
        return AsyncPackages[PackageIndex];
    }
 
    EAsyncPackageState::Type ProcessAsyncLoading(int32& OutPackagesProcessed, bool bUseTimeLimit, bool bUseFullTimeLimit, float TimeLimit, FFlushRequest& FlushRequest);
 
    void CheckForCycles();
 
    EAsyncPackageState::Type TickAsyncLoading(bool bUseTimeLimit, bool bUseFullTimeLimit, double TimeLimit, FFlushRequest FlushRequest = FFlushRequest());
 
    EAsyncPackageState::Type TickAsyncThread(bool bUseTimeLimit, bool bUseFullTimeLimit, double TimeLimit, bool& bDidSomething, FFlushRequest& FlushRequest);
 
    void InitializeLoading() override;
 
    float GetAsyncLoadPercentage(const FName& PackageName) override;
 
 
private:
    void AddPendingRequest(int32 RequestID)
    {
#if THREADSAFE_UOBJECTS
        UE::TScopeLock Lock(PendingRequestsCritical);
#endif
        if (!PendingRequests.Contains(RequestID))
        {
            PendingRequests.Add(RequestID);
        }
    }
 
    void RemovePendingRequests(TArray<int32>& RequestIDs)
    {
#if THREADSAFE_UOBJECTS
        UE::TScopeLock Lock(PendingRequestsCritical);
#endif
        for (int32 ID : RequestIDs)
        {
            PendingRequests.Remove(ID);
            TRACE_LOADTIME_END_REQUEST(ID);
        }       
    }
 
    void CompleteFlushRequests(TArray<int32>& RequestIDs);
 
    void MapDuplicateRequestID(int32 DuplicateRequestId, int32 MainRequestId);
 
 
    int32 GetQueuedPackagesCount() const
    {
        return QueuedPackagesCounter.GetValue();
    }
    int32 GetExistingAsyncPackagesCount() const
    {
        return ExistingAsyncPackagesCounter.GetValue();
    }
 
    bool ContainsRequestInternal(int32 RequestID)
    {
#if THREADSAFE_UOBJECTS
        UE::TScopeLock Lock(PendingRequestsCritical);
#endif
        return PendingRequests.Contains(RequestID);
    }
 
    bool ContainsAnyRequestInternal(TConstArrayView<int32> RequestIDs)
    {
#if THREADSAFE_UOBJECTS
        UE::TScopeLock Lock(PendingRequestsCritical);
#endif
        return Algo::AnyOf(RequestIDs, [this](int32 RequestID){ return PendingRequests.Contains(RequestID); });
    }
 
    bool ContainsAnyRequestInternal(TConstArrayView<std::atomic<int32>> RequestIDs)
    {
#if THREADSAFE_UOBJECTS
        UE::TScopeLock Lock(PendingRequestsCritical);
#endif
        return Algo::AnyOf(RequestIDs, [this](const std::atomic<int32>& RequestID){ return PendingRequests.Contains(RequestID.load(std::memory_order_relaxed)); });
    }
 
    FFlushRequest AddFlushRequest(TConstArrayView<int32> RequestIDs);
 
    FFlushRequest PeekFlushRequest();
 
    void AdjustFlushRequest(FFlushRequest& FlushRequest);
 
    bool ShouldProcessPackage(FAsyncPackage* InAsyncPackage, const FFlushRequest& FlushRequest);
    
    EAsyncPackageState::Type ProcessLoadedPackages(bool bUseTimeLimit, bool bUseFullTimeLimit, double TimeLimit, bool& bDidSomething, const FFlushRequest& FlushRequest);
 
    int32 CreateAsyncPackagesFromQueue(bool bUseTimeLimit, bool bUseFullTimeLimit, double TimeLimit);
 
    void ProcessAsyncPackageRequest(FAsyncPackageDesc* InRequest, FAsyncPackage* InRootPackage);
 
    void UpdateExistingPackagePriorities(FAsyncPackage* InPackage, TAsyncLoadPriority InNewPriority);
 
    FAsyncPackage* FindExistingPackageAndAddCompletionCallback(FAsyncPackageDesc* PackageRequest, TMap<FName, FAsyncPackage*>& PackageList);
 
    void AddToLoadedPackages(FAsyncPackage* Package);
 
    void CancelAsyncLoadingInternal();
    void FinalizeCancelAsyncLoadingInternal();
 
    FEventLoadGraph EventGraph; 
    FPrecacheCallbackHandler* PrecacheHandler;
    int32 AsyncLoadingThreadIndex;
 
    static FThreadSafeCounter AsyncLoadingTickCounter;
    static int32 CurrentAsyncLoadingTickThreadIndex;
 
public:
 
    FEventLoadGraph& GetEventGraph()
    {
        return EventGraph;
    }
 
    FPrecacheCallbackHandler& GetPrecacheHandler()
    {
        return *PrecacheHandler;
    }
 
    int32 GetThreadIndex() const
    {
        return AsyncLoadingThreadIndex;
    }
 
#if !UE_BUILD_SHIPPING && !UE_BUILD_TEST
    FThreadSafeCounter RecursionNotAllowed;
#endif
 
    static FAsyncLoadingThread* Instance;
};