From eaa6e087cf7602e4fc57a087d13e80b24ba7941b Mon Sep 17 00:00:00 2001 From: Jeremy Ong Date: Thu, 5 Aug 2021 14:37:59 -0600 Subject: [PATCH] JobGraph -> TaskGraph (and associated classes/files) This commit also addresses all PR feedback Signed-off-by: Jeremy Ong --- .../AzCore/Jobs/Internal/JobTypeEraser.cpp | 76 ---- .../AzCore/Jobs/Internal/JobTypeEraser.h | 232 ---------- .../AzCore/AzCore/Jobs/JobDescriptor.h | 49 --- .../AzCore/AzCore/Jobs/JobExecutor.h | 85 ---- .../Framework/AzCore/AzCore/Jobs/JobGraph.cpp | 85 ---- Code/Framework/AzCore/AzCore/Jobs/JobGraph.h | 150 ------- .../AzCore/AzCore/Task/Internal/Task.cpp | 58 +++ .../AzCore/AzCore/Task/Internal/Task.h | 180 ++++++++ .../AzCore/AzCore/Task/Internal/Task.inl | 84 ++++ .../AzCore/AzCore/Task/Internal/TaskConfig.h | 14 + .../AzCore/AzCore/Task/TaskDescriptor.h | 49 +++ .../JobExecutor.cpp => Task/TaskExecutor.cpp} | 176 ++++---- .../AzCore/AzCore/Task/TaskExecutor.h | 94 ++++ .../AzCore/AzCore/Task/TaskGraph.cpp | 85 ++++ Code/Framework/AzCore/AzCore/Task/TaskGraph.h | 151 +++++++ .../{Jobs/JobGraph.inl => Task/TaskGraph.inl} | 26 +- .../AzCore/AzCore/azcore_files.cmake | 18 +- .../{JobGraphTests.cpp => TaskTests.cpp} | 405 ++++++++++++------ .../AzCore/Tests/azcoretests_files.cmake | 2 +- 19 files changed, 1109 insertions(+), 910 deletions(-) delete mode 100644 Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.cpp delete mode 100644 Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.h delete mode 100644 Code/Framework/AzCore/AzCore/Jobs/JobDescriptor.h delete mode 100644 Code/Framework/AzCore/AzCore/Jobs/JobExecutor.h delete mode 100644 Code/Framework/AzCore/AzCore/Jobs/JobGraph.cpp delete mode 100644 Code/Framework/AzCore/AzCore/Jobs/JobGraph.h create mode 100644 Code/Framework/AzCore/AzCore/Task/Internal/Task.cpp create mode 100644 Code/Framework/AzCore/AzCore/Task/Internal/Task.h create mode 100644 Code/Framework/AzCore/AzCore/Task/Internal/Task.inl create mode 100644 Code/Framework/AzCore/AzCore/Task/Internal/TaskConfig.h create mode 100644 Code/Framework/AzCore/AzCore/Task/TaskDescriptor.h rename Code/Framework/AzCore/AzCore/{Jobs/JobExecutor.cpp => Task/TaskExecutor.cpp} (62%) create mode 100644 Code/Framework/AzCore/AzCore/Task/TaskExecutor.h create mode 100644 Code/Framework/AzCore/AzCore/Task/TaskGraph.cpp create mode 100644 Code/Framework/AzCore/AzCore/Task/TaskGraph.h rename Code/Framework/AzCore/AzCore/{Jobs/JobGraph.inl => Task/TaskGraph.inl} (50%) rename Code/Framework/AzCore/Tests/{JobGraphTests.cpp => TaskTests.cpp} (53%) diff --git a/Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.cpp b/Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.cpp deleted file mode 100644 index 043f232e32..0000000000 --- a/Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.cpp +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Copyright (c) Contributors to the Open 3D Engine Project. - * For complete copyright and license terms please see the LICENSE at the root of this distribution. - * - * SPDX-License-Identifier: Apache-2.0 OR MIT - * - */ - -#include - -namespace AZ::Internal -{ - TypeErasedJob::TypeErasedJob(TypeErasedJob&& other) noexcept - { - if (!other.m_relocator || other.m_lambda != other.m_buffer) - { - // The type-erased lambda is trivially relocatable OR, the lambda is heap allocated - memcpy(this, &other, sizeof(TypeErasedJob)); - - if (other.m_lambda == other.m_buffer) - { - m_lambda = m_buffer; - } - - // Prevent deletion in the event the lambda had spilled to the heap - other.m_lambda = nullptr; - return; - } - - // At this point, we know the lambda was inlined - m_lambda = m_buffer; - - m_invoker = other.m_invoker; - m_relocator = other.m_relocator; - m_destroyer = other.m_destroyer; - - // We now own the lambda, so clear the moved-from job's destroyer - other.m_destroyer = nullptr; - other.m_invoker = nullptr; - - m_relocator(m_buffer, other.m_buffer); - } - - TypeErasedJob& TypeErasedJob::operator=(TypeErasedJob&& other) noexcept - { - if (this == &other) - { - return *this; - } - - this->~TypeErasedJob(); - - new (this) TypeErasedJob{ AZStd::move(other) }; - - return *this; - } - - TypeErasedJob::~TypeErasedJob() - { - if (m_lambda) - { - if (m_destroyer) - { - // The presence of m_destroyer indicates that the lambda is not trivially destructible - m_destroyer(m_lambda); - } - - if (m_lambda != m_buffer) - { - // We've spilled the lambda into the heap, free its memory - azfree(m_lambda); - } - } - } - -} // namespace AZ::Internal diff --git a/Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.h b/Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.h deleted file mode 100644 index e57ba85176..0000000000 --- a/Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.h +++ /dev/null @@ -1,232 +0,0 @@ -/* - * Copyright (c) Contributors to the Open 3D Engine Project. - * For complete copyright and license terms please see the LICENSE at the root of this distribution. - * - * SPDX-License-Identifier: Apache-2.0 OR MIT - * - */ -#pragma once - -#include -#include -#include -#include -#include -#include - -namespace AZ::Internal -{ - using JobInvoke_t = void (*)(void* lambda); - using JobRelocate_t = void (*)(void* dst, void* src); - using JobDestroy_t = void (*)(void* obj); - - class CompiledJobGraph; - - // Lambdas are opaque types and we cannot extract any member function pointers. In order to store lambdas in a - // type erased fashion, we instead use a single function call indirection, invoking the lambda function in a - // static class function which has a stable address in memory. The Erased* methods return addresses to the - // indirect callers of the lambda copy/move assignment operators, call operator, and destructor. - // - // For lambdas that are trivially relocatable, both the returned move and copy assignment function pointers - // will be nullptr. - // - // Lambdas that are trivially destructible will result in a nullptr returned JobDestroy_t pointer. - // - // The class will check that the lambda is copy assignable or movable. - template - class JobTypeEraser final - { - public: - constexpr JobInvoke_t ErasedInvoker() - { - return reinterpret_cast(Invoker); - } - - constexpr JobRelocate_t ErasedRelocator() - { - if constexpr (AZStd::is_trivially_move_constructible_v) - { - return nullptr; - } - else if constexpr (AZStd::is_move_constructible_v) - { - return reinterpret_cast(Mover); - } - else if constexpr (AZStd::is_copy_constructible_v) - { - return reinterpret_cast(Copyer); - } - else - { - static_assert( - AZStd::is_move_constructible_v || AZStd::is_copy_constructible_v, - "Job lambdas must be either move or copy constructible. Please verify that all captured data is move or copy " - "constructible."); - } - } - - constexpr JobDestroy_t ErasedDestroyer() - { - if constexpr (AZStd::is_trivially_destructible_v) - { - return nullptr; - } - else - { - return reinterpret_cast(Destroyer); - } - } - - private: - constexpr static void Invoker(Lambda* lambda) - { - lambda->operator()(); - } - - constexpr static void Mover(Lambda* dst, Lambda* src) - { - new (dst) Lambda{ AZStd::move(*src) }; - } - - constexpr static void Copyer(Lambda* dst, Lambda* src) - { - new (dst) Lambda{ *src }; - } - - constexpr static void Destroyer(Lambda* lambda) - { - lambda->~Lambda(); - } - }; - - // The TypeErasedJob encapsulates member function pointers to store in a homogeneously-typed container - // The function signature of all lambdas encoded in a TypeErasedJob is void(*)(). The lambdas can capture - // data, in which case the data is inlined in this structure if the payload is less than or equal to the - // buffer size. Otherwise, the data is heap allocated. - class alignas(alignof(max_align_t)) TypeErasedJob final - { - public: - // The inline buffer allows the TypeErasedJob to span two cache lines. Lambdas can capture 48 - // bytes of data (6 pointers/references on a 64-bit machine) before spilling to the heap. - constexpr static size_t BufferSize = - 128 - sizeof(size_t) * 6 - sizeof(uint32_t) - sizeof(JobDescriptor) - sizeof(AZStd::atomic); - - TypeErasedJob() = default; - - template - TypeErasedJob(JobDescriptor const& desc, Lambda&& lambda) noexcept - : m_descriptor{ desc } - { - JobTypeEraser eraser; - m_invoker = eraser.ErasedInvoker(); - m_relocator = eraser.ErasedRelocator(); - m_destroyer = eraser.ErasedDestroyer(); - - // NOTE: This code is conservative in that extended alignment requirements result in a heap - // spill, even if the lambda could have occupied a portion of the inline buffer with a base - // pointer adjustment. - if constexpr (sizeof(Lambda) <= BufferSize && alignof(Lambda) <= alignof(max_align_t)) - { - TypedRelocate(AZStd::forward(lambda), m_buffer); - m_lambda = m_buffer; - } - else - { - // Lambda has spilled to the heap (or requires extended alignment) - m_lambda = reinterpret_cast(azmalloc(sizeof(Lambda), alignof(Lambda))); - TypedRelocate(AZStd::forward(lambda), m_lambda); - } - } - - TypeErasedJob(TypeErasedJob&& other) noexcept; - - TypeErasedJob& operator=(TypeErasedJob&& other) noexcept; - - ~TypeErasedJob(); - - void Link(TypeErasedJob& other); - - // Indicates if this job is a root of the graph (with no dependencies) - bool IsRoot(); - - void Init() noexcept - { - m_dependencyCount = m_inboundLinkCount; - } - - void Invoke() - { - m_invoker(m_lambda); - } - - uint8_t GetPriorityNumber() const - { - return static_cast(m_descriptor.priority); - } - - private: - friend class CompiledJobGraph; - friend class JobWorker; - - // This relocation avoids branches needed if the lambda type is unknown - template - void TypedRelocate(Lambda&& lambda, char* destination) - { - if constexpr (AZStd::is_trivially_move_constructible_v) - { - memcpy(destination, reinterpret_cast(&lambda), sizeof(Lambda)); - } - else if constexpr (AZStd::is_move_constructible_v) - { - new (destination) Lambda{ AZStd::move(lambda) }; - } - else if constexpr (AZStd::is_copy_constructible_v) - { - new (destination) Lambda{ lambda }; - } - else - { - static_assert( - AZStd::is_move_constructible_v || AZStd::is_copy_constructible_v, - "Job lambdas must be either move or copy constructible. Please verify that all captured data is move or copy " - "constructible."); - } - } - - // Small buffer optimization for lambdas. We cover our bases here by enforcing alignment on the - // class to equal the alignment of the largest scalar type available on the system (generally - // 16 bytes). - char m_buffer[BufferSize]; - AZStd::atomic m_dependencyCount; - - // This value is an offset in a buffer that stores dependency tracking information. - uint32_t m_successorOffset = 0; - uint32_t m_inboundLinkCount = 0; - uint32_t m_outboundLinkCount = 0; - - // May point to the inlined payload buffer, or heap - char* m_lambda = nullptr; - - CompiledJobGraph* m_graph = nullptr; - - JobInvoke_t m_invoker; - - // If nullptr, the lambda is trivially relocatable (via memcpy). Otherwise, it must be invoked - // when instances of this class are moved. - JobRelocate_t m_relocator; - JobDestroy_t m_destroyer; - - JobDescriptor m_descriptor; - }; - - inline void TypeErasedJob::Link(TypeErasedJob& other) - { - ++m_outboundLinkCount; - ++other.m_inboundLinkCount; - } - - inline bool TypeErasedJob::IsRoot() - { - return m_inboundLinkCount == 0; - } -} // namespace AZ::Internal diff --git a/Code/Framework/AzCore/AzCore/Jobs/JobDescriptor.h b/Code/Framework/AzCore/AzCore/Jobs/JobDescriptor.h deleted file mode 100644 index 82a9603bd5..0000000000 --- a/Code/Framework/AzCore/AzCore/Jobs/JobDescriptor.h +++ /dev/null @@ -1,49 +0,0 @@ -/* - * Copyright (c) Contributors to the Open 3D Engine Project. - * For complete copyright and license terms please see the LICENSE at the root of this distribution. - * - * SPDX-License-Identifier: Apache-2.0 OR MIT - * - */ - -#pragma once - -#include -#include - -namespace AZ -{ - // Job priorities MAY be used judiciously to fine tune runtime execution, with the understanding - // that profiling is needed to understand what the critical path per frame is. Modifying - // job priorities is an EXPERT setting that should succeed a healthy dose of measurement. - enum class JobPriority : uint8_t - { - CRITICAL = 0, - HIGH = 1, - MEDIUM = 2, // Default - LOW = 3, - PRIORITY_COUNT = 4, - }; - - // All submitted jobs are associated with a JobDescriptor which defines the priority, affinitization, - // and tracking of the job resource utilization. - // - // TODO: Define various job kinds and provide a mechanism for cpuMask computation on different systems. - struct JobDescriptor - { - // Unique job kind label (e.g. "frustum culling") - // Job names *must* be provided - const char* jobName = nullptr; - - // Associates a set of job kinds together for budget tracking (e.g. "graphics") - const char* jobGroup = nullptr; - - // EXPERTS ONLY. Jobs of higher priority are executed ahead of any lower priority jobs - // that were queued before it provided they had not yet started - JobPriority priority = JobPriority::MEDIUM; - - // EXPERTS ONLY. A bitmask that restricts jobs of this kind to run only on cores - // corresponding to a set bit. 0 is synonymous with all bits set - uint32_t cpuMask = 0; - }; -} diff --git a/Code/Framework/AzCore/AzCore/Jobs/JobExecutor.h b/Code/Framework/AzCore/AzCore/Jobs/JobExecutor.h deleted file mode 100644 index 746edf00a1..0000000000 --- a/Code/Framework/AzCore/AzCore/Jobs/JobExecutor.h +++ /dev/null @@ -1,85 +0,0 @@ -/* - * Copyright (c) Contributors to the Open 3D Engine Project. - * For complete copyright and license terms please see the LICENSE at the root of this distribution. - * - * SPDX-License-Identifier: Apache-2.0 OR MIT - * - */ - -#pragma once - -#include -#include -#include -#include -#include -#include - -namespace AZ -{ - class JobGraphEvent; - class JobGraph; - - namespace Internal - { - class CompiledJobGraph final - { - public: - AZ_CLASS_ALLOCATOR(CompiledJobGraph, SystemAllocator, 0) - - CompiledJobGraph( - AZStd::vector&& jobs, - AZStd::unordered_map>& links, - size_t linkCount, - JobGraph* parent); - - AZStd::vector& Jobs() noexcept - { - return m_jobs; - } - - // Indicate that a constituent job has finished and decrement a counter to determine if the - // graph should be freed (returns the value after atomic decrement) - uint32_t Release(); - - private: - friend class ::AZ::JobGraph; - friend class JobWorker; - - AZStd::vector m_jobs; - AZStd::vector m_successors; - JobGraphEvent* m_waitEvent = nullptr; - // The pointer to the parent graph is set only if it is retained - JobGraph* m_parent = nullptr; - AZStd::atomic m_remaining; - }; - - class JobWorker; - } // namespace Internal - - class JobExecutor - { - public: - AZ_CLASS_ALLOCATOR(JobExecutor, SystemAllocator, 0); - - static JobExecutor& Instance(); - - // Passing 0 for the threadCount requests for the thread count to match the hardware concurrency - JobExecutor(uint32_t threadCount = 0); - ~JobExecutor(); - - void Submit(Internal::CompiledJobGraph& graph); - - void Submit(Internal::TypeErasedJob& job); - - // Busy wait until jobs are cleared from the executor (note, does not prevent future jobs from being submitted) - void Drain(); - private: - friend class Internal::JobWorker; - - Internal::JobWorker* m_workers; - uint32_t m_threadCount = 0; - AZStd::atomic m_lastSubmission; - AZStd::atomic m_remaining; - }; -} // namespace AZ diff --git a/Code/Framework/AzCore/AzCore/Jobs/JobGraph.cpp b/Code/Framework/AzCore/AzCore/Jobs/JobGraph.cpp deleted file mode 100644 index 6b34f1273a..0000000000 --- a/Code/Framework/AzCore/AzCore/Jobs/JobGraph.cpp +++ /dev/null @@ -1,85 +0,0 @@ -/* - * Copyright (c) Contributors to the Open 3D Engine Project. - * For complete copyright and license terms please see the LICENSE at the root of this distribution. - * - * SPDX-License-Identifier: Apache-2.0 OR MIT - * - */ - -#include - -#include - -namespace AZ -{ - using Internal::CompiledJobGraph; - - void JobToken::PrecedesInternal(JobToken& comesAfter) - { - AZ_Assert(!m_parent.m_submitted, "Cannot mutate a JobGraph that was previously submitted."); - - // Increment inbound/outbound edge counts - m_parent.m_jobs[m_index].Link(m_parent.m_jobs[comesAfter.m_index]); - - m_parent.m_links[m_index].emplace_back(comesAfter.m_index); - - ++m_parent.m_linkCount; - } - - JobGraph::~JobGraph() - { - if (m_retained && m_compiledJobGraph) - { - // This job graph has already finished and we are potentially responsible for its destruction - if (m_compiledJobGraph->Release() == 0) - { - azdestroy(m_compiledJobGraph); - } - } - } - - void JobGraph::Reset() - { - AZ_Assert(!m_submitted, "Cannot reset a job graph while it is in flight"); - if (m_compiledJobGraph) - { - azdestroy(m_compiledJobGraph); - m_compiledJobGraph = nullptr; - } - m_jobs.clear(); - m_links.clear(); - m_linkCount = 0; - } - - void JobGraph::Submit(JobGraphEvent* waitEvent) - { - SubmitOnExecutor(JobExecutor::Instance(), waitEvent); - } - - void JobGraph::SubmitOnExecutor(JobExecutor& executor, JobGraphEvent* waitEvent) - { - if (!m_compiledJobGraph) - { - m_compiledJobGraph = aznew CompiledJobGraph(AZStd::move(m_jobs), m_links, m_linkCount, m_retained ? this : nullptr); - } - - m_compiledJobGraph->m_waitEvent = waitEvent; - m_compiledJobGraph->m_remaining = m_compiledJobGraph->m_jobs.size() + (m_retained ? 1 : 0); - for (size_t i = 0; i != m_compiledJobGraph->m_jobs.size(); ++i) - { - m_compiledJobGraph->m_jobs[i].Init(); - } - - executor.Submit(*m_compiledJobGraph); - - if (m_retained) - { - m_submitted = true; - } - else - { - m_compiledJobGraph = nullptr; - Reset(); - } - } -} diff --git a/Code/Framework/AzCore/AzCore/Jobs/JobGraph.h b/Code/Framework/AzCore/AzCore/Jobs/JobGraph.h deleted file mode 100644 index 872a9a4e5c..0000000000 --- a/Code/Framework/AzCore/AzCore/Jobs/JobGraph.h +++ /dev/null @@ -1,150 +0,0 @@ -/* - * Copyright (c) Contributors to the Open 3D Engine Project. - * For complete copyright and license terms please see the LICENSE at the root of this distribution. - * - * SPDX-License-Identifier: Apache-2.0 OR MIT - * - */ - -#pragma once - -// NOTE: If adding additional header/symbol dependencies, consider if such additions are better -// suited in the private CompiledJobGraph implementation instead to keep this header lean. -#include -#include -#include -#include -#include -#include - -namespace AZ -{ - namespace Internal - { - class CompiledJobGraph; - } - class JobExecutor; - class JobGraph; - - // A JobToken is returned each time a Job is added to the JobGraph. JobTokens are used to - // express dependencies between jobs within the graph. - class JobToken final - { - public: - // Indicate that this job must finish before the job token(s) passed as the argument - template - void Precedes(JT&... tokens); - - // Indicate that this job must finish after the job token(s) passed as the argument - template - void Succeeds(JT&... tokens); - - private: - friend class JobGraph; - - void PrecedesInternal(JobToken& comesAfter); - - // Only the JobGraph should be creating JobToken - JobToken(JobGraph& parent, size_t index); - - JobGraph& m_parent; - size_t m_index; - }; - - // A JobGraphEvent may be used to block until a job graph has finished executing. Usage - // is NOT recommended for the majority of tasks (prefer to simply containing expanding/contracting - // the graph without synchronization over the course of the frame). However, the event - // is useful for the edges of the computation graph. - // - // You are responsible for ensuring the event object lifetime exceeds the job graph lifetime. - // - // After the JobGraphEvent is signaled, you are allowed to reuse the same JobGraphEvent - // for a future submission. - class JobGraphEvent - { - public: - bool IsSignaled(); - void Wait(); - - private: - friend class ::AZ::Internal::CompiledJobGraph; - friend class JobGraph; - void Signal(); - - AZStd::binary_semaphore m_semaphore; - }; - - // The JobGraph encapsulates a set of jobs and their interdependencies. After adding - // jobs, and marking dependencies as necessary, the entire graph is submitted via - // the JobGraph::Submit method. - // - // The JobGraph MAY be retained across multiple frames and resubmitted, provided the - // user provides some guarantees (see comments associated with JobGraph::Retain). - class JobGraph final - { - public: - ~JobGraph(); - - // Reset the state of the job graph to begin recording jobs and edges again - // NOTE: Graph must be in a "settled" state (cannot be in-flight) - void Reset(); - - // Add a job to the graph, retrieiving a token that can be used to express dependencies - // between jobs. The first argument specifies the JobKind, used for tracking the job. - // NOTE: This operation is invalid if the graph is in-flight - template - JobToken AddJob(JobDescriptor const& descriptor, Lambda&& lambda); - - template - AZStd::array AddJobs(JobDescriptor const& descriptor, Lambdas&&... lambdas); - - // By default, you are responsible for retaining the JobGraph, indicating you promise that - // this JobGraph will live as long as it takes for all constituent jobs to complete. - // Once retained, this job graph can be resubmitted after completion without any - // modifications. JobTokens that were created as a result of adding jobs used to - // mark dependencies DO NOT need to outlive the job graph. - // - // Invoking Detach PRIOR to submission indicates you wish the jobs associated with this - // JobGraph to deallocate upon completion. After invoking Detach, you may let this JobGraph - // go out of scope or deallocate after submission. - // - // NOTE: The JobGraph has no concept of resources used by design. Resubmission - // of the job graph is expected to rely on either indirection, or safe overwriting - // of previously used memory to supply new data (this can even be done as the first - // job in the graph). - // NOTE: This operation is invalid if the graph is in-flight - void Detach(); - - // Invoke the job graph, asserting if there are dependency violations. Note that - // submitting the same graph multiple times to process simultaneously is VALID - // behavior. This is, for example, a mechanism that allows a job graph to loop - // in perpetuity (in fact, the entire frame could be modeled as a single job graph, - // where the final job resubmits the job graph again). - // - // This API is not designed to protect against memory safety violations (nothing - // can prevent a user from incorrectly aliasing memory unsafely even without repeated - // submission). To catch memory safety violations, it is ENCOURAGED that you access - // data through JobResource handles. - void Submit(JobGraphEvent* waitEvent = nullptr); - - // Same as submit but run on a different executor than the default system executor - void SubmitOnExecutor(JobExecutor& executor, JobGraphEvent* waitEvent = nullptr); - - private: - friend class JobToken; - friend class Internal::CompiledJobGraph; - - Internal::CompiledJobGraph* m_compiledJobGraph = nullptr; - - AZStd::vector m_jobs; - - // Job index |-> Dependent job indices - AZStd::unordered_map> m_links; - - uint32_t m_linkCount = 0; - bool m_retained = true; - AZStd::atomic m_submitted = false; - }; -} // namespace AZ - -#include diff --git a/Code/Framework/AzCore/AzCore/Task/Internal/Task.cpp b/Code/Framework/AzCore/AzCore/Task/Internal/Task.cpp new file mode 100644 index 0000000000..b0e5da8fc0 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/Internal/Task.cpp @@ -0,0 +1,58 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ + +#include + +namespace AZ::Internal +{ + Task::Task(Task&& other) noexcept + { + if (!other.m_relocator) + { + // The type-erased lambda is trivially relocatable OR, the lambda is heap allocated + memcpy(this, &other, sizeof(Task)); + + // Prevent deletion in the event the lambda had spilled to the heap + other.m_destroyer = nullptr; + return; + } + + m_invoker = other.m_invoker; + m_relocator = other.m_relocator; + m_destroyer = other.m_destroyer; + + // We now own the lambda, so clear the moved-from task's destroyer + other.m_destroyer = nullptr; + + m_relocator(m_lambda, other.m_lambda); + } + + Task& Task::operator=(Task&& other) noexcept + { + if (this == &other) + { + return *this; + } + + this->~Task(); + + new (this) Task{ AZStd::move(other) }; + + return *this; + } + + Task::~Task() + { + if (m_destroyer) + { + // The presence of m_destroyer indicates that the lambda is not trivially destructible + m_destroyer(m_lambda); + } + } + +} // namespace AZ::Internal diff --git a/Code/Framework/AzCore/AzCore/Task/Internal/Task.h b/Code/Framework/AzCore/AzCore/Task/Internal/Task.h new file mode 100644 index 0000000000..e5d3736f2a --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/Internal/Task.h @@ -0,0 +1,180 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +namespace AZ::Internal +{ + using TaskInvoke_t = void (*)(void* lambda); + using TaskRelocate_t = void (*)(void* dst, void* src); + using TaskDestroy_t = void (*)(void* obj); + + class CompiledTaskGraph; + + // Lambdas are opaque types and we cannot extract any member function pointers. In order to store lambdas in a + // type erased fashion, we instead use a single function call indirection, invoking the lambda function in a + // static class function which has a stable address in memory. The Erased* methods return addresses to the + // indirect callers of the lambda copy/move assignment operators, call operator, and destructor. + // + // For lambdas that are trivially relocatable, both the returned move and copy assignment function pointers + // will be nullptr. + // + // Lambdas that are trivially destructible will result in a nullptr returned TaskDestroy_t pointer. + // + // The class will check that the lambda is copy assignable or movable. + template + class TaskTypeEraser final + { + public: + constexpr TaskInvoke_t ErasedInvoker() + { + return reinterpret_cast(Invoker); + } + + constexpr TaskRelocate_t ErasedRelocator() + { + if constexpr (AZStd::is_trivially_move_constructible_v) + { + return nullptr; + } + else if constexpr (AZStd::is_move_constructible_v) + { + return reinterpret_cast(Mover); + } + else if constexpr (AZStd::is_copy_constructible_v) + { + return reinterpret_cast(Copier); + } + else + { + static_assert( + AZStd::is_move_constructible_v || AZStd::is_copy_constructible_v, + "Task lambdas must be either move or copy constructible. Please verify that all captured data is move or copy " + "constructible."); + } + } + + constexpr TaskDestroy_t ErasedDestroyer() + { + if constexpr (AZStd::is_trivially_destructible_v) + { + return nullptr; + } + else + { + return reinterpret_cast(Destroyer); + } + } + + private: + constexpr static void Invoker(Lambda* lambda) + { + lambda->operator()(); + } + + constexpr static void Mover(Lambda* dst, Lambda* src) + { + new (dst) Lambda{ AZStd::move(*src) }; + } + + constexpr static void Copier(Lambda* dst, Lambda* src) + { + new (dst) Lambda{ *src }; + } + + constexpr static void Destroyer(Lambda* lambda) + { + lambda->~Lambda(); + } + }; + + // The Task encapsulates member function pointers to store in a homogeneously-typed container + // The function signature of all lambdas encoded in a Task is void(*)(). The lambdas can capture + // data, in which case the data is inlined in this structure. Attempting to capture more data + // will result in a compile failure, so use indirection and capture a pointer/reference to your + // data if you run into this. + class alignas(alignof(max_align_t)) Task final + { + public: + AZ_CLASS_ALLOCATOR(Task, ThreadPoolAllocator, 0); + + // The inline buffer allows the Task to span two cache lines. Lambdas can capture 56 + // bytes of data (7 pointers/references on a 64-bit machine). + constexpr static size_t BufferSize = + AZ_TRAIT_TASK_BYTE_SIZE - sizeof(size_t) * 5 - sizeof(uint32_t) - sizeof(TaskDescriptor) - sizeof(AZStd::atomic); + + Task() = default; + + // Prevent binding lvalue references to lambdas + // If you are encountering a compiler error here, please either move the lambda into the AddJob function with AZStd::move + // or simply define the lambda directly as a parameter of AddJob + template + Task(TaskDescriptor const& desc, Lambda& lambda) = delete; + + template + Task(TaskDescriptor const& desc, Lambda&& lambda) noexcept; + + Task(Task&& other) noexcept; + + Task& operator=(Task&& other) noexcept; + + ~Task(); + + void Link(Task& other); + + // Indicates if this task is a root of the graph (with no dependencies) + bool IsRoot() const noexcept; + + // Prepare for dispatch (reset the dependency counter to the number of inbound edges) + void Init() noexcept; + + // Invoke the embedded lambda function + void Invoke(); + + uint8_t GetPriorityNumber() const noexcept; + + private: + friend class CompiledTaskGraph; + friend class TaskWorker; + + // This relocation avoids branches needed if the lambda type is unknown + template + void TypedRelocate(Lambda&& lambda, char* destination); + + // Small buffer optimization for lambdas. We cover our bases here by enforcing alignment on the + // class to equal the alignment of the largest scalar type available on the system (generally + // 16 bytes). + char m_lambda[BufferSize]; + AZStd::atomic m_dependencyCount; + + // This value is an offset in a buffer that stores dependency tracking information. + uint32_t m_successorOffset = 0; + uint32_t m_inboundLinkCount = 0; + uint32_t m_outboundLinkCount = 0; + + CompiledTaskGraph* m_graph = nullptr; + + TaskInvoke_t m_invoker; + + // If nullptr, the lambda is trivially relocatable (via memcpy). Otherwise, it must be invoked + // when instances of this class are moved. + TaskRelocate_t m_relocator; + TaskDestroy_t m_destroyer; + + TaskDescriptor m_descriptor; + }; +} // namespace AZ::Internal + +#include diff --git a/Code/Framework/AzCore/AzCore/Task/Internal/Task.inl b/Code/Framework/AzCore/AzCore/Task/Internal/Task.inl new file mode 100644 index 0000000000..83cd311f56 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/Internal/Task.inl @@ -0,0 +1,84 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ + +namespace AZ::Internal +{ + template + Task::Task(TaskDescriptor const& desc, Lambda&& lambda) noexcept + : m_descriptor{ desc } + { + static_assert( + sizeof(Lambda) <= BufferSize, + "Task lambda has too much captured data, please capture no" + "more than 56 bytes of data (likely by capturing a single reference/pointer to a container of data)"); + static_assert( + alignof(Lambda) <= alignof(max_align_t), + "Task lambda has extended alignment which isn't supported." + "Please capture a reference/pointer to the data requiring an extended alignment instead"); + + TaskTypeEraser eraser; + m_invoker = eraser.ErasedInvoker(); + m_relocator = eraser.ErasedRelocator(); + m_destroyer = eraser.ErasedDestroyer(); + + // NOTE: This code is conservative in that extended alignment requirements result in a heap + // spill, even if the lambda could have occupied a portion of the inline buffer with a base + // pointer adjustment. + TypedRelocate(AZStd::forward(lambda), m_lambda); + } + + template + void Task::TypedRelocate(Lambda&& lambda, char* destination) + { + if constexpr (AZStd::is_trivially_move_constructible_v) + { + memcpy(destination, reinterpret_cast(&lambda), sizeof(Lambda)); + } + else if constexpr (AZStd::is_move_constructible_v) + { + new (destination) Lambda{ AZStd::move(lambda) }; + } + else if constexpr (AZStd::is_copy_constructible_v) + { + new (destination) Lambda{ lambda }; + } + else + { + static_assert( + AZStd::is_move_constructible_v || AZStd::is_copy_constructible_v, + "Task lambdas must be either move or copy constructible. Please verify that all captured data is move or copy " + "constructible."); + } + } + + inline void Task::Init() noexcept + { + m_dependencyCount = m_inboundLinkCount; + } + + inline void Task::Invoke() + { + m_invoker(m_lambda); + } + + inline uint8_t Task::GetPriorityNumber() const noexcept + { + return static_cast(m_descriptor.priority); + } + + inline void Task::Link(Task& other) + { + ++m_outboundLinkCount; + ++other.m_inboundLinkCount; + } + + inline bool Task::IsRoot() const noexcept + { + return m_inboundLinkCount == 0; + } +} // namespace AZ::Internal diff --git a/Code/Framework/AzCore/AzCore/Task/Internal/TaskConfig.h b/Code/Framework/AzCore/AzCore/Task/Internal/TaskConfig.h new file mode 100644 index 0000000000..d5550b40d4 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/Internal/TaskConfig.h @@ -0,0 +1,14 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ +#pragma once + +#include + +#if !defined(AZ_TRAIT_TASK_BYTE_SIZE) +#define AZ_TRAIT_TASK_BYTE_SIZE 128 +#endif diff --git a/Code/Framework/AzCore/AzCore/Task/TaskDescriptor.h b/Code/Framework/AzCore/AzCore/Task/TaskDescriptor.h new file mode 100644 index 0000000000..8342fed925 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/TaskDescriptor.h @@ -0,0 +1,49 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ + +#pragma once + +#include +#include + +namespace AZ +{ + // Task priorities MAY be used judiciously to fine tune runtime execution, with the understanding + // that profiling is needed to understand what the critical path per frame is. Modifying + // task priorities is an EXPERT setting that should succeed a healthy dose of measurement. + enum class TaskPriority : uint8_t + { + CRITICAL = 0, + HIGH = 1, + MEDIUM = 2, // Default + LOW = 3, + PRIORITY_COUNT = 4, + }; + + // All submitted tasks are associated with a TaskDescriptor which defines the priority, affinitization, + // and tracking of the task resource utilization. + // + // TODO: Define various task kinds and provide a mechanism for cpuMask computation on different systems. + struct TaskDescriptor + { + // Unique task kind label (e.g. "frustum culling") + // Task names *must* be provided + const char* taskName = nullptr; + + // Associates a set of task kinds together for budget tracking (e.g. "graphics") + const char* taskGroup = nullptr; + + // EXPERTS ONLY. Tasks of higher priority are executed ahead of any lower priority tasks + // that were queued before it provided they had not yet started + TaskPriority priority = TaskPriority::MEDIUM; + + // EXPERTS ONLY. A bitmask that restricts tasks of this kind to run only on cores + // corresponding to a set bit. 0 is synonymous with all bits set + uint32_t cpuMask = 0; + }; +} diff --git a/Code/Framework/AzCore/AzCore/Jobs/JobExecutor.cpp b/Code/Framework/AzCore/AzCore/Task/TaskExecutor.cpp similarity index 62% rename from Code/Framework/AzCore/AzCore/Jobs/JobExecutor.cpp rename to Code/Framework/AzCore/AzCore/Task/TaskExecutor.cpp index 83823515c1..c69763f289 100644 --- a/Code/Framework/AzCore/AzCore/Jobs/JobExecutor.cpp +++ b/Code/Framework/AzCore/AzCore/Task/TaskExecutor.cpp @@ -6,8 +6,8 @@ * */ -#include -#include +#include +#include #include #include @@ -17,46 +17,45 @@ #include #include #include +#include #include namespace AZ { - constexpr static size_t PRIORITY_COUNT = static_cast(JobPriority::PRIORITY_COUNT); - namespace Internal { - CompiledJobGraph::CompiledJobGraph( - AZStd::vector&& jobs, + CompiledTaskGraph::CompiledTaskGraph( + AZStd::vector&& tasks, AZStd::unordered_map>& links, size_t linkCount, - JobGraph* parent) + TaskGraph* parent) : m_parent{ parent } { - m_jobs = AZStd::move(jobs); + m_tasks = AZStd::move(tasks); m_successors.resize(linkCount); - TypeErasedJob** cursor = m_successors.data(); + Task** cursor = m_successors.data(); - for (size_t i = 0; i != m_jobs.size(); ++i) + for (size_t i = 0; i != m_tasks.size(); ++i) { - TypeErasedJob& job = m_jobs[i]; - job.m_graph = this; - job.m_successorOffset = cursor - m_successors.data(); - cursor += job.m_outboundLinkCount; + Task& task = m_tasks[i]; + task.m_graph = this; + task.m_successorOffset = cursor - m_successors.data(); + cursor += task.m_outboundLinkCount; - AZ_Assert(job.m_outboundLinkCount == links[i].size(), "Job outbound link information mismatch"); + AZ_Assert(task.m_outboundLinkCount == links[i].size(), "Task outbound link information mismatch"); - for (uint32_t j = 0; j != job.m_outboundLinkCount; ++j) + for (uint32_t j = 0; j != task.m_outboundLinkCount; ++j) { - m_successors[static_cast(job.m_successorOffset) + j] = &m_jobs[links[i][j]]; + m_successors[static_cast(task.m_successorOffset) + j] = &m_tasks[links[i][j]]; } } // TODO: Check for dependency cycles } - uint32_t CompiledJobGraph::Release() + uint32_t CompiledTaskGraph::Release() { uint32_t remaining = --m_remaining; @@ -94,35 +93,35 @@ namespace AZ AZStd::atomic reserve; }; - // The Job Queue is a lock free 4-priority queue. Its basic operation is as follows: + // The Task Queue is a lock free 4-priority queue. Its basic operation is as follows: // Each priority level is associated with a different queue, corresponding to the maximum size of a uint16_t. // Each queue is implemented as a ring buffer, and a 64 bit atomic maintains the following state per queue: // - offset to the "head" of the ring, from where we acquire elements // - offset to the "tail" of the ring, which tracks where new elements should be enqueued // - offset to a tail reservation index, which is used to reserve a slot to enqueue elements - class JobQueue final + class TaskQueue final { public: - // Preallocating upfront allows us to reserve slots to insert jobs without locks. - // Each thread allocated by the job manager consumes ~2 MB. + // Preallocating upfront allows us to reserve slots to insert tasks without locks. + // Each thread allocated by the task manager consumes ~2 MB. constexpr static uint16_t MaxQueueSize = 0xffff; - constexpr static uint8_t PriorityLevelCount = static_cast(JobPriority::PRIORITY_COUNT); + constexpr static uint8_t PriorityLevelCount = static_cast(TaskPriority::PRIORITY_COUNT); - JobQueue() = default; - JobQueue(const JobQueue&) = delete; - JobQueue& operator=(const JobQueue&) = delete; + TaskQueue() = default; + TaskQueue(const TaskQueue&) = delete; + TaskQueue& operator=(const TaskQueue&) = delete; - void Enqueue(TypeErasedJob* job); - TypeErasedJob* TryDequeue(); + void Enqueue(Task* task); + Task* TryDequeue(); private: QueueStatus m_status[PriorityLevelCount] = {}; - TypeErasedJob* m_queues[PriorityLevelCount][MaxQueueSize] = {}; + Task* m_queues[PriorityLevelCount][MaxQueueSize] = {}; }; - void JobQueue::Enqueue(TypeErasedJob* job) + void TaskQueue::Enqueue(Task* task) { - uint8_t priority = job->GetPriorityNumber(); + uint8_t priority = task->GetPriorityNumber(); QueueStatus& status = m_status[priority]; AZStd::exponential_backoff backoff; @@ -131,18 +130,18 @@ namespace AZ uint16_t reserve = status.reserve.load(); uint16_t head = status.head.load(); - // Enqueuing is done in two phases because we cannot atomically write the job to the slot we reserve + // Enqueuing is done in two phases because we cannot atomically write the task to the slot we reserve // and simulataneously publish the fact that the slot is now available. if (reserve != head - 1) { // Try to reserve a slot if (status.reserve.compare_exchange_weak(reserve, reserve + 1)) { - m_queues[priority][reserve] = job; + m_queues[priority][reserve] = task; uint16_t expectedReserve = reserve; - // Increment the tail to advertise the new job + // Increment the tail to advertise the new task while (!status.tail.compare_exchange_weak(expectedReserve, reserve + 1)) { expectedReserve = reserve; @@ -160,7 +159,7 @@ namespace AZ } } - TypeErasedJob* JobQueue::TryDequeue() + Task* TaskQueue::TryDequeue() { for (size_t priority = 0; priority != PriorityLevelCount; ++priority) { @@ -176,10 +175,10 @@ namespace AZ } else { - TypeErasedJob* job = m_queues[priority][status.head]; + Task* task = m_queues[priority][status.head]; if (status.head.compare_exchange_weak(head, head + 1)) { - return job; + return task; } } } @@ -188,14 +187,14 @@ namespace AZ return nullptr; } - class JobWorker + class TaskWorker { public: - void Spawn(::AZ::JobExecutor& executor, size_t id, AZStd::semaphore& initSemaphore, bool affinitize) + void Spawn(::AZ::TaskExecutor& executor, size_t id, AZStd::semaphore& initSemaphore, bool affinitize) { m_executor = &executor; - AZStd::string threadName = AZStd::string::format("JobWorker %zu", id); + AZStd::string threadName = AZStd::string::format("TaskWorker %zu", id); AZStd::thread_desc desc = {}; desc.m_name = threadName.c_str(); if (affinitize) @@ -219,13 +218,13 @@ namespace AZ m_thread.join(); } - void Enqueue(TypeErasedJob* job) + void Enqueue(Task* task) { - m_queue.Enqueue(job); + m_queue.Enqueue(task); if (!m_busy.exchange(true)) { - // The worker was idle prior to enqueueing the job, release the semaphore + // The worker was idle prior to enqueueing the task, release the semaphore m_semaphore.release(); } } @@ -245,24 +244,26 @@ namespace AZ m_busy = true; - TypeErasedJob* job = m_queue.TryDequeue(); - while (job) + Task* task = m_queue.TryDequeue(); + while (task) { - job->Invoke(); - // Decrement counts for all job successors - for (size_t j = 0; j != job->m_outboundLinkCount; ++j) + task->Invoke(); + // Decrement counts for all task successors + for (size_t j = 0; j != task->m_outboundLinkCount; ++j) { - TypeErasedJob* successor = job->m_graph->m_successors[job->m_successorOffset + j]; + Task* successor = task->m_graph->m_successors[task->m_successorOffset + j]; if (--successor->m_dependencyCount == 0) { m_executor->Submit(*successor); } } - job->m_graph->Release(); - --m_executor->m_remaining; + if (task->m_graph->Release() == (task->m_graph->m_parent ? 1 : 0)) + { + m_executor->ReleaseGraph(); + } - job = m_queue.TryDequeue(); + task = m_queue.TryDequeue(); } } } @@ -272,24 +273,38 @@ namespace AZ AZStd::atomic m_busy; AZStd::binary_semaphore m_semaphore; - ::AZ::JobExecutor* m_executor; - JobQueue m_queue; + ::AZ::TaskExecutor* m_executor; + TaskQueue m_queue; }; } // namespace Internal - JobExecutor& JobExecutor::Instance() + static EnvironmentVariable s_executor; + constexpr static const char* s_executorName = "GlobalTaskExecutor"; + TaskExecutor& TaskExecutor::Instance() { - // TODO: Create the default executor as part of a component (as in JobManagerComponent) - static JobExecutor executor; - return executor; + if (!s_executor) + { + s_executor = AZ::Environment::FindVariable(s_executorName); + } + + return **s_executor; } - JobExecutor::JobExecutor(uint32_t threadCount) + // TODO: Create the default executor as part of a component (as in TaskManagerComponent) + void TaskExecutor::SetInstance(TaskExecutor* executor) + { + AZ_Assert(!s_executor, "Attempting to set the global task executor more than once"); + + s_executor = AZ::Environment::CreateVariable("GlobalTaskExecutor"); + s_executor.Set(executor); + } + + TaskExecutor::TaskExecutor(uint32_t threadCount) { // TODO: Configure thread count + affinity based on configuration m_threadCount = threadCount == 0 ? AZStd::thread::hardware_concurrency() : threadCount; - m_workers = reinterpret_cast(azmalloc(m_threadCount * sizeof(Internal::JobWorker))); + m_workers = reinterpret_cast(azmalloc(m_threadCount * sizeof(Internal::TaskWorker))); bool affinitize = m_threadCount == AZStd::thread::hardware_concurrency(); @@ -297,7 +312,7 @@ namespace AZ for (size_t i = 0; i != m_threadCount; ++i) { - new (m_workers + i) Internal::JobWorker{}; + new (m_workers + i) Internal::TaskWorker{}; m_workers[i].Spawn(*this, i, initSemaphore, affinitize); } @@ -307,43 +322,56 @@ namespace AZ } } - JobExecutor::~JobExecutor() + TaskExecutor::~TaskExecutor() { for (size_t i = 0; i != m_threadCount; ++i) { m_workers[i].Join(); - m_workers[i].~JobWorker(); + m_workers[i].~TaskWorker(); } azfree(m_workers); } - void JobExecutor::Submit(Internal::CompiledJobGraph& graph) + void TaskExecutor::Submit(Internal::CompiledTaskGraph& graph) { - // Submit all jobs that have no inbound edges - for (Internal::TypeErasedJob& job : graph.Jobs()) + ++m_graphsRemaining; + // Submit all tasks that have no inbound edges + for (Internal::Task& task : graph.Tasks()) { - if (job.IsRoot()) + if (task.IsRoot()) { - Submit(job); + Submit(task); } } } - void JobExecutor::Submit(Internal::TypeErasedJob& job) + void TaskExecutor::Submit(Internal::Task& task) { // TODO: Something more sophisticated is likely needed here. // First, we are completely ignoring affinity. // Second, some heuristics on core availability will help distribute work more effectively - ++m_remaining; - m_workers[++m_lastSubmission % m_threadCount].Enqueue(&job); + m_workers[++m_lastSubmission % m_threadCount].Enqueue(&task); } - void JobExecutor::Drain() + void TaskExecutor::Drain() { - while (m_remaining > 0) + m_isDraining = true; + if (m_graphsRemaining == 0) { - AZStd::this_thread::sleep_for(AZStd::chrono::milliseconds{ 100 }); + return; + } + m_drainSemaphore.acquire(); + } + + void TaskExecutor::ReleaseGraph() + { + uint64_t graphsRemaining = --m_graphsRemaining; + + if (graphsRemaining == 0 && m_isDraining) + { + m_drainSemaphore.release(); + m_isDraining = false; } } } // namespace AZ diff --git a/Code/Framework/AzCore/AzCore/Task/TaskExecutor.h b/Code/Framework/AzCore/AzCore/Task/TaskExecutor.h new file mode 100644 index 0000000000..ad4c4b81c2 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/TaskExecutor.h @@ -0,0 +1,94 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +namespace AZ +{ + class TaskGraphEvent; + class TaskGraph; + + namespace Internal + { + class CompiledTaskGraph final + { + public: + AZ_CLASS_ALLOCATOR(CompiledTaskGraph, SystemAllocator, 0) + + CompiledTaskGraph( + AZStd::vector&& tasks, + AZStd::unordered_map>& links, + size_t linkCount, + TaskGraph* parent); + + AZStd::vector& Tasks() noexcept + { + return m_tasks; + } + + // Indicate that a constituent task has finished and decrement a counter to determine if the + // graph should be freed (returns the value after atomic decrement) + uint32_t Release(); + + private: + friend class ::AZ::TaskGraph; + friend class TaskWorker; + + AZStd::vector m_tasks; + AZStd::vector m_successors; + TaskGraphEvent* m_waitEvent = nullptr; + // The pointer to the parent graph is set only if it is retained + TaskGraph* m_parent = nullptr; + AZStd::atomic m_remaining; + }; + + class TaskWorker; + } // namespace Internal + + class TaskExecutor final + { + public: + AZ_CLASS_ALLOCATOR(TaskExecutor, SystemAllocator, 0); + + static TaskExecutor& Instance(); + + // Invoked by a system component on program launch + static void SetInstance(TaskExecutor* executor); + + // Passing 0 for the threadCount requests for the thread count to match the hardware concurrency + explicit TaskExecutor(uint32_t threadCount = 0); + ~TaskExecutor(); + + void Submit(Internal::CompiledTaskGraph& graph); + + void Submit(Internal::Task& task); + + // Wait until tasks are cleared from the executor (note, does not prevent future tasks from being submitted) + // If this is used, it's expected to be used between frames to shutdown the engine + void Drain(); + private: + friend class Internal::TaskWorker; + + void ReleaseGraph(); + + Internal::TaskWorker* m_workers; + uint32_t m_threadCount = 0; + AZStd::atomic m_lastSubmission; + AZStd::atomic m_graphsRemaining; + AZStd::atomic m_isDraining; + AZStd::binary_semaphore m_drainSemaphore; + }; +} // namespace AZ diff --git a/Code/Framework/AzCore/AzCore/Task/TaskGraph.cpp b/Code/Framework/AzCore/AzCore/Task/TaskGraph.cpp new file mode 100644 index 0000000000..86e4f846d5 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/TaskGraph.cpp @@ -0,0 +1,85 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ + +#include + +#include + +namespace AZ +{ + using Internal::CompiledTaskGraph; + + void TaskToken::PrecedesInternal(TaskToken& comesAfter) + { + AZ_Assert(!m_parent.m_submitted, "Cannot mutate a TaskGraph that was previously submitted."); + + // Increment inbound/outbound edge counts + m_parent.m_tasks[m_index].Link(m_parent.m_tasks[comesAfter.m_index]); + + m_parent.m_links[m_index].emplace_back(comesAfter.m_index); + + ++m_parent.m_linkCount; + } + + TaskGraph::~TaskGraph() + { + if (m_retained && m_compiledTaskGraph) + { + // This job graph has already finished and we are potentially responsible for its destruction + if (m_compiledTaskGraph->Release() == 0) + { + azdestroy(m_compiledTaskGraph); + } + } + } + + void TaskGraph::Reset() + { + AZ_Assert(!m_submitted, "Cannot reset a job graph while it is in flight"); + if (m_compiledTaskGraph) + { + azdestroy(m_compiledTaskGraph); + m_compiledTaskGraph = nullptr; + } + m_tasks.clear(); + m_links.clear(); + m_linkCount = 0; + } + + void TaskGraph::Submit(TaskGraphEvent* waitEvent) + { + SubmitOnExecutor(TaskExecutor::Instance(), waitEvent); + } + + void TaskGraph::SubmitOnExecutor(TaskExecutor& executor, TaskGraphEvent* waitEvent) + { + if (!m_compiledTaskGraph) + { + m_compiledTaskGraph = aznew CompiledTaskGraph(AZStd::move(m_tasks), m_links, m_linkCount, m_retained ? this : nullptr); + } + + m_compiledTaskGraph->m_waitEvent = waitEvent; + m_compiledTaskGraph->m_remaining = m_compiledTaskGraph->m_tasks.size() + (m_retained ? 1 : 0); + for (size_t i = 0; i != m_compiledTaskGraph->m_tasks.size(); ++i) + { + m_compiledTaskGraph->m_tasks[i].Init(); + } + + executor.Submit(*m_compiledTaskGraph); + + if (m_retained) + { + m_submitted = true; + } + else + { + m_compiledTaskGraph = nullptr; + Reset(); + } + } +} diff --git a/Code/Framework/AzCore/AzCore/Task/TaskGraph.h b/Code/Framework/AzCore/AzCore/Task/TaskGraph.h new file mode 100644 index 0000000000..d133593508 --- /dev/null +++ b/Code/Framework/AzCore/AzCore/Task/TaskGraph.h @@ -0,0 +1,151 @@ +/* + * Copyright (c) Contributors to the Open 3D Engine Project. + * For complete copyright and license terms please see the LICENSE at the root of this distribution. + * + * SPDX-License-Identifier: Apache-2.0 OR MIT + * + */ + +#pragma once + +// NOTE: If adding additional header/symbol dependencies, consider if such additions are better +// suited in the private CompiledTaskGraph implementation instead to keep this header lean. +#include +#include +#include +#include +#include +#include + +namespace AZ +{ + namespace Internal + { + class CompiledTaskGraph; + } + class TaskExecutor; + class TaskGraph; + + // A TaskToken is returned each time a Task is added to the TaskGraph. TaskTokens are used to + // express dependencies between tasks within the graph, and have no purpose after the graph + // is submitted (simply let them go out of scope) + class TaskToken final + { + public: + // Indicate that this task must finish before the task token(s) passed as the argument + template + void Precedes(JT&... tokens); + + // Indicate that this task must finish after the task token(s) passed as the argument + template + void Follows(JT&... tokens); + + private: + friend class TaskGraph; + + void PrecedesInternal(TaskToken& comesAfter); + + // Only the TaskGraph should be creating TaskToken + TaskToken(TaskGraph& parent, size_t index); + + TaskGraph& m_parent; + size_t m_index; + }; + + // A TaskGraphEvent may be used to block until a task graph has finished executing. Usage + // is NOT recommended for the majority of tasks (prefer to simply containing expanding/contracting + // the graph without synchronization over the course of the frame). However, the event + // is useful for the edges of the computation graph. + // + // You are responsible for ensuring the event object lifetime exceeds the task graph lifetime. + // + // After the TaskGraphEvent is signaled, you are allowed to reuse the same TaskGraphEvent + // for a future submission. + class TaskGraphEvent + { + public: + bool IsSignaled(); + void Wait(); + + private: + friend class ::AZ::Internal::CompiledTaskGraph; + friend class TaskGraph; + void Signal(); + + AZStd::binary_semaphore m_semaphore; + }; + + // The TaskGraph encapsulates a set of tasks and their interdependencies. After adding + // tasks, and marking dependencies as necessary, the entire graph is submitted via + // the TaskGraph::Submit method. + // + // The TaskGraph MAY be retained across multiple frames and resubmitted, provided the + // user provides some guarantees (see comments associated with TaskGraph::Retain). + class TaskGraph final + { + public: + ~TaskGraph(); + + // Reset the state of the task graph to begin recording tasks and edges again + // NOTE: Graph must be in a "settled" state (cannot be in-flight) + void Reset(); + + // Add a task to the graph, retrieiving a token that can be used to express dependencies + // between tasks. The first argument specifies the TaskKind, used for tracking the task. + // NOTE: This operation is invalid if the graph is in-flight + template + TaskToken AddTask(TaskDescriptor const& descriptor, Lambda&& lambda); + + template + AZStd::array AddTasks(TaskDescriptor const& descriptor, Lambdas&&... lambdas); + + // By default, you are responsible for retaining the TaskGraph, indicating you promise that + // this TaskGraph will live as long as it takes for all constituent tasks to complete. + // Once retained, this task graph can be resubmitted after completion without any + // modifications. TaskTokens that were created as a result of adding tasks used to + // mark dependencies DO NOT need to outlive the task graph. + // + // Invoking Detach PRIOR to submission indicates you wish the tasks associated with this + // TaskGraph to deallocate upon completion. After invoking Detach, you may let this TaskGraph + // go out of scope or deallocate after submission. + // + // NOTE: The TaskGraph has no concept of resources used by design. Resubmission + // of the task graph is expected to rely on either indirection, or safe overwriting + // of previously used memory to supply new data (this can even be done as the first + // task in the graph). + // NOTE: This operation is invalid if the graph is in-flight + void Detach(); + + // Invoke the task graph, asserting if there are dependency violations. Note that + // submitting the same graph multiple times to process simultaneously is VALID + // behavior. This is, for example, a mechanism that allows a task graph to loop + // in perpetuity (in fact, the entire frame could be modeled as a single task graph, + // where the final task resubmits the task graph again). + // + // This API is not designed to protect against memory safety violations (nothing + // can prevent a user from incorrectly aliasing memory unsafely even without repeated + // submission). To catch memory safety violations, it is ENCOURAGED that you access + // data through TaskResource handles. + void Submit(TaskGraphEvent* waitEvent = nullptr); + + // Same as submit but run on a different executor than the default system executor + void SubmitOnExecutor(TaskExecutor& executor, TaskGraphEvent* waitEvent = nullptr); + + private: + friend class TaskToken; + friend class Internal::CompiledTaskGraph; + + Internal::CompiledTaskGraph* m_compiledTaskGraph = nullptr; + + AZStd::vector m_tasks; + + // Task index |-> Dependent task indices + AZStd::unordered_map> m_links; + + uint32_t m_linkCount = 0; + bool m_retained = true; + AZStd::atomic m_submitted = false; + }; +} // namespace AZ + +#include diff --git a/Code/Framework/AzCore/AzCore/Jobs/JobGraph.inl b/Code/Framework/AzCore/AzCore/Task/TaskGraph.inl similarity index 50% rename from Code/Framework/AzCore/AzCore/Jobs/JobGraph.inl rename to Code/Framework/AzCore/AzCore/Task/TaskGraph.inl index ad1fb3505c..1971ddbbca 100644 --- a/Code/Framework/AzCore/AzCore/Jobs/JobGraph.inl +++ b/Code/Framework/AzCore/AzCore/Task/TaskGraph.inl @@ -10,56 +10,56 @@ namespace AZ { - inline JobToken::JobToken(JobGraph& parent, size_t index) + inline TaskToken::TaskToken(TaskGraph& parent, size_t index) : m_parent{ parent } , m_index{ index } { } template - inline void JobToken::Precedes(JT&... tokens) + void TaskToken::Precedes(JT&... tokens) { (PrecedesInternal(tokens), ...); } template - inline void JobToken::Succeeds(JT&... tokens) + void TaskToken::Follows(JT&... tokens) { (tokens.PrecedesInternal(*this), ...); } - inline bool JobGraphEvent::IsSignaled() + inline bool TaskGraphEvent::IsSignaled() { return m_semaphore.try_acquire_for(AZStd::chrono::milliseconds{ 0 }); } - inline void JobGraphEvent::Wait() + inline void TaskGraphEvent::Wait() { m_semaphore.acquire(); } - inline void JobGraphEvent::Signal() + inline void TaskGraphEvent::Signal() { m_semaphore.release(); } template - inline JobToken JobGraph::AddJob(JobDescriptor const& desc, Lambda&& lambda) + TaskToken TaskGraph::AddTask(TaskDescriptor const& desc, Lambda&& lambda) { - AZ_Assert(!m_submitted, "Cannot mutate a JobGraph that was previously submitted or in flight."); + AZ_Assert(!m_submitted, "Cannot mutate a TaskGraph that was previously submitted or in flight."); - m_jobs.emplace_back(desc, AZStd::forward(lambda)); + m_tasks.emplace_back(desc, AZStd::forward(lambda)); - return { *this, m_jobs.size() - 1 }; + return { *this, m_tasks.size() - 1 }; } template - inline AZStd::array JobGraph::AddJobs(JobDescriptor const& descriptor, Lambdas&&... lambdas) + AZStd::array TaskGraph::AddTasks(TaskDescriptor const& descriptor, Lambdas&&... lambdas) { - return { AddJob(descriptor, AZStd::forward(lambdas))... }; + return { AddTask(descriptor, AZStd::forward(lambdas))... }; } - inline void JobGraph::Detach() + inline void TaskGraph::Detach() { m_retained = false; } diff --git a/Code/Framework/AzCore/AzCore/azcore_files.cmake b/Code/Framework/AzCore/AzCore/azcore_files.cmake index a23ec9ab82..1e2a0b98a0 100644 --- a/Code/Framework/AzCore/AzCore/azcore_files.cmake +++ b/Code/Framework/AzCore/AzCore/azcore_files.cmake @@ -221,8 +221,6 @@ set(FILES Jobs/Internal/JobManagerWorkStealing.cpp Jobs/Internal/JobManagerWorkStealing.h Jobs/Internal/JobNotify.h - Jobs/Internal/JobTypeEraser.cpp - Jobs/Internal/JobTypeEraser.h Jobs/Job.cpp Jobs/Job.h Jobs/JobCancelGroup.h @@ -230,14 +228,8 @@ set(FILES Jobs/JobCompletionSpin.h Jobs/JobContext.cpp Jobs/JobContext.h - Jobs/JobDescriptor.h Jobs/JobEmpty.h - Jobs/JobExecutor.cpp - Jobs/JobExecutor.h Jobs/JobFunction.h - Jobs/JobGraph.cpp - Jobs/JobGraph.h - Jobs/JobGraph.inl Jobs/JobManager.cpp Jobs/JobManager.h Jobs/JobManagerBus.h @@ -624,6 +616,16 @@ set(FILES Socket/AzSocket_fwd.h Socket/AzSocket.cpp Socket/AzSocket.h + Task/Internal/Task.cpp + Task/Internal/Task.inl + Task/Internal/Task.h + Task/Internal/TaskConfig.h + Task/TaskDescriptor.h + Task/TaskExecutor.cpp + Task/TaskExecutor.h + Task/TaskGraph.cpp + Task/TaskGraph.h + Task/TaskGraph.inl Threading/ThreadSafeDeque.h Threading/ThreadSafeDeque.inl Threading/ThreadSafeObject.h diff --git a/Code/Framework/AzCore/Tests/JobGraphTests.cpp b/Code/Framework/AzCore/Tests/TaskTests.cpp similarity index 53% rename from Code/Framework/AzCore/Tests/JobGraphTests.cpp rename to Code/Framework/AzCore/Tests/TaskTests.cpp index 469773f471..eeb523ae26 100644 --- a/Code/Framework/AzCore/Tests/JobGraphTests.cpp +++ b/Code/Framework/AzCore/Tests/TaskTests.cpp @@ -6,26 +6,26 @@ * */ -#include -#include +#include +#include #include #include #include -using AZ::JobDescriptor; -using AZ::JobGraph; -using AZ::JobGraphEvent; -using AZ::JobExecutor; -using AZ::Internal::TypeErasedJob; -using AZ::JobPriority; +using AZ::TaskDescriptor; +using AZ::TaskGraph; +using AZ::TaskGraphEvent; +using AZ::TaskExecutor; +using AZ::Internal::Task; +using AZ::TaskPriority; -static JobDescriptor defaultJD{ "JobGraphTestJob", "JobGraphTests" }; +static TaskDescriptor defaultTD{ "TaskGraphTestTask", "TaskGraphTests" }; namespace UnitTest { - class JobGraphTestFixture : public AllocatorsTestFixture + class TaskGraphTestFixture : public AllocatorsTestFixture { public: void SetUp() override @@ -34,7 +34,7 @@ namespace UnitTest AZ::AllocatorInstance::Create(); AZ::AllocatorInstance::Create(); - m_executor = aznew JobExecutor(4); + m_executor = aznew TaskExecutor(4); } void TearDown() override @@ -46,38 +46,38 @@ namespace UnitTest } protected: - JobExecutor* m_executor; + TaskExecutor* m_executor; }; - TEST(JobGraphTests, TrivialJobLambda) + TEST(TaskGraphTests, TrivialTaskLambda) { int x = 0; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [&x]() { ++x; }); - job.Invoke(); + task.Invoke(); EXPECT_EQ(1, x); } - TEST(JobGraphTests, TrivialJobLambdaMove) + TEST(TaskGraphTests, TrivialTaskLambdaMove) { int x = 0; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [&x]() { ++x; }); - TypeErasedJob job2 = AZStd::move(job); + Task task2 = AZStd::move(task); - job2.Invoke(); + task2.Invoke(); EXPECT_EQ(1, x); } @@ -110,78 +110,90 @@ namespace UnitTest int copyCount = 0; }; - TEST(JobGraphTests, MoveOnlyJobLambda) + /* + TEST(TaskGraphTests, ThisShouldNotCompile) + { + auto lambda = [] + { + }; + + Task task(defaultTD, lambda); + task.Invoke(); + } + */ + + TEST(TaskGraphTests, MoveOnlyTaskLambda) { TrackMoves tm; int moveCount = 0; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [tm = AZStd::move(tm), &moveCount] { moveCount = tm.moveCount; }); - job.Invoke(); + task.Invoke(); // Two moves are expected. Once into the capture body of the lambda, once to construct - // the type erased job + // the type erased task EXPECT_EQ(2, moveCount); } - TEST(JobGraphTests, MoveOnlyJobLambdaMove) + TEST(TaskGraphTests, MoveOnlyTaskLambdaMove) { TrackMoves tm; int moveCount = 0; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [tm = AZStd::move(tm), &moveCount] { moveCount = tm.moveCount; }); - TypeErasedJob job2 = AZStd::move(job); - job2.Invoke(); + Task task2 = AZStd::move(task); + task2.Invoke(); EXPECT_EQ(3, moveCount); } - TEST(JobGraphTests, CopyOnlyJobLambda) + TEST(TaskGraphTests, CopyOnlyTaskLambda) { TrackCopies tc; int copyCount = 0; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [tc, ©Count] { copyCount = tc.copyCount; }); - job.Invoke(); + task.Invoke(); // Two copies are expected. Once into the capture body of the lambda, once to construct - // the type erased job + // the type erased task EXPECT_EQ(2, copyCount); } - TEST(JobGraphTests, CopyOnlyJobLambdaMove) + TEST(TaskGraphTests, CopyOnlyTaskLambdaMove) { TrackCopies tc; int copyCount = 0; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [tc, ©Count] { copyCount = tc.copyCount; }); - TypeErasedJob job2 = AZStd::move(job); - job2.Invoke(); + Task task2 = AZStd::move(task); + task2.Invoke(); EXPECT_EQ(3, copyCount); } - TEST(JobGraphTests, DestroyLambda) + TEST(TaskGraphTests, DestroyLambda) { // This test ensures that for a lambda with a destructor, the destructor is invoked // exactly once on a non-moved-from object. @@ -209,12 +221,12 @@ namespace UnitTest { TrackDestroy td{ &x }; - TypeErasedJob job( - defaultJD, + Task task( + defaultTD, [td = AZStd::move(td)] { }); - job.Invoke(); + task.Invoke(); // Destructor should not have run yet (except on moved-from instances) EXPECT_EQ(x, 0); } @@ -223,25 +235,21 @@ namespace UnitTest EXPECT_EQ(x, 1); } - TEST_F(JobGraphTestFixture, SerialGraph) + TEST_F(TaskGraphTestFixture, VariadicInterface) { int x = 0; - JobGraph graph; - auto a = graph.AddJob( - defaultJD, + TaskGraph graph; + auto [a, b, c] = graph.AddTasks( + defaultTD, [&] { x += 3; - }); - auto b = graph.AddJob( - defaultJD, + }, [&] { x = 4 * x; - }); - auto c = graph.AddJob( - defaultJD, + }, [&] { x -= 1; @@ -250,35 +258,69 @@ namespace UnitTest a.Precedes(b); b.Precedes(c); - JobGraphEvent ev; + TaskGraphEvent ev; graph.SubmitOnExecutor(*m_executor, &ev); ev.Wait(); EXPECT_EQ(11, x); } - TEST_F(JobGraphTestFixture, DetachedGraph) + TEST_F(TaskGraphTestFixture, SerialGraph) { int x = 0; - JobGraphEvent ev; + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, + [&] + { + x += 3; + }); + auto b = graph.AddTask( + defaultTD, + [&] + { + x = 4 * x; + }); + auto c = graph.AddTask( + defaultTD, + [&] + { + x -= 1; + }); + + a.Precedes(b); + b.Precedes(c); + + TaskGraphEvent ev; + graph.SubmitOnExecutor(*m_executor, &ev); + ev.Wait(); + + EXPECT_EQ(11, x); + } + + TEST_F(TaskGraphTestFixture, DetachedGraph) + { + int x = 0; + + TaskGraphEvent ev; { - JobGraph graph; - auto a = graph.AddJob( - defaultJD, + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, [&] { x += 3; }); - auto b = graph.AddJob( - defaultJD, + auto b = graph.AddTask( + defaultTD, [&] { x = 4 * x; }); - auto c = graph.AddJob( - defaultJD, + auto c = graph.AddTask( + defaultTD, [&] { x -= 1; @@ -295,35 +337,35 @@ namespace UnitTest EXPECT_EQ(11, x); } - TEST_F(JobGraphTestFixture, ForkJoin) + TEST_F(TaskGraphTestFixture, ForkJoin) { AZStd::atomic x = 0; - // Job a initializes x to 3 - // Job b and c toggles the lowest two bits atomically - // Job d decrements x + // Task a initializes x to 3 + // Task b and c toggles the lowest two bits atomically + // Task d decrements x - JobGraph graph; - auto a = graph.AddJob( - defaultJD, + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, [&] { x = 0b111; }); - auto b = graph.AddJob( - defaultJD, + auto b = graph.AddTask( + defaultTD, [&] { x ^= 1; }); - auto c = graph.AddJob( - defaultJD, + auto c = graph.AddTask( + defaultTD, [&] { x ^= 2; }); - auto d = graph.AddJob( - defaultJD, + auto d = graph.AddTask( + defaultTD, [&] { x -= 1; @@ -336,65 +378,65 @@ namespace UnitTest // d a.Precedes(b, c); - d.Succeeds(b, c); + d.Follows(b, c); - JobGraphEvent ev; + TaskGraphEvent ev; graph.SubmitOnExecutor(*m_executor, &ev); ev.Wait(); EXPECT_EQ(3, x); } - TEST_F(JobGraphTestFixture, SpawnSubgraph) + TEST_F(TaskGraphTestFixture, SpawnSubgraph) { AZStd::atomic x = 0; - JobGraph graph; - auto a = graph.AddJob( - defaultJD, + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, [&] { x = 0b111; }); - auto b = graph.AddJob( - defaultJD, + auto b = graph.AddTask( + defaultTD, [&] { x ^= 1; }); - auto c = graph.AddJob( - defaultJD, + auto c = graph.AddTask( + defaultTD, [&] { x ^= 2; - JobGraph subgraph; - auto e = subgraph.AddJob( - defaultJD, + TaskGraph subgraph; + auto e = subgraph.AddTask( + defaultTD, [&] { x ^= 0b1000; }); - auto f = subgraph.AddJob( - defaultJD, + auto f = subgraph.AddTask( + defaultTD, [&] { x ^= 0b10000; }); - auto g = subgraph.AddJob( - defaultJD, + auto g = subgraph.AddTask( + defaultTD, [&] { x += 0b1000; }); e.Precedes(g); f.Precedes(g); - JobGraphEvent ev; + TaskGraphEvent ev; subgraph.SubmitOnExecutor(*m_executor, &ev); ev.Wait(); }); - auto d = graph.AddJob( - defaultJD, + auto d = graph.AddTask( + defaultTD, [&] { x -= 1; @@ -418,56 +460,56 @@ namespace UnitTest b.Precedes(d); c.Precedes(d); - JobGraphEvent ev; + TaskGraphEvent ev; graph.SubmitOnExecutor(*m_executor, &ev); ev.Wait(); EXPECT_EQ(3 | 0b100000, x); } - TEST_F(JobGraphTestFixture, RetainedGraph) + TEST_F(TaskGraphTestFixture, RetainedGraph) { AZStd::atomic x = 0; - JobGraph graph; - auto a = graph.AddJob( - defaultJD, + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, [&] { x = 0b111; }); - auto b = graph.AddJob( - defaultJD, + auto b = graph.AddTask( + defaultTD, [&] { x ^= 1; }); - auto c = graph.AddJob( - defaultJD, + auto c = graph.AddTask( + defaultTD, [&] { x ^= 2; }); - auto d = graph.AddJob( - defaultJD, + auto d = graph.AddTask( + defaultTD, [&] { x -= 1; }); - auto e = graph.AddJob( - defaultJD, + auto e = graph.AddTask( + defaultTD, [&] { x ^= 0b1000; }); - auto f = graph.AddJob( - defaultJD, + auto f = graph.AddTask( + defaultTD, [&] { x ^= 0b10000; }); - auto g = graph.AddJob( - defaultJD, + auto g = graph.AddTask( + defaultTD, [&] { x += 0b1000; @@ -486,10 +528,10 @@ namespace UnitTest a.Precedes(b, c); b.Precedes(d); c.Precedes(e, f); - g.Succeeds(e, f); + g.Follows(e, f); g.Precedes(d); - JobGraphEvent ev; + TaskGraphEvent ev; graph.SubmitOnExecutor(*m_executor, &ev); ev.Wait(); @@ -501,18 +543,107 @@ namespace UnitTest EXPECT_EQ(3 | 0b100000, x); } + + TEST_F(TaskGraphTestFixture, ExecutorDrainRetained) + { + bool drainDone = false; + AZStd::binary_semaphore taskStart; + AZStd::binary_semaphore threadLaunched; + AZStd::binary_semaphore threadFinished; + + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, + [&] + { + taskStart.acquire(); + }); + + graph.SubmitOnExecutor(*m_executor); + + AZStd::thread drainThread{ [this, &drainDone, &threadLaunched, &threadFinished] + { + threadLaunched.release(); + m_executor->Drain(); + drainDone = true; + threadFinished.release(); + } }; + + + // Wait until our drain thread has launched + threadLaunched.acquire(); + + // The task itself hasn't started, so the drain should still be blocking + EXPECT_EQ(false, drainDone); + + // Allow the task to finish + taskStart.release(); + + // Wait for the drain thread to wrap up + threadFinished.acquire(); + + // We successfully drained the executor + EXPECT_EQ(true, drainDone); + + drainThread.join(); + } + + TEST_F(TaskGraphTestFixture, ExecutorDrainDetached) + { + bool drainDone = false; + AZStd::binary_semaphore taskStart; + AZStd::binary_semaphore threadLaunched; + AZStd::binary_semaphore threadFinished; + + TaskGraph graph; + auto a = graph.AddTask( + defaultTD, + [&] + { + taskStart.acquire(); + }); + graph.Detach(); + + graph.SubmitOnExecutor(*m_executor); + + AZStd::thread drainThread{ [this, &drainDone, &threadLaunched, &threadFinished] + { + threadLaunched.release(); + m_executor->Drain(); + drainDone = true; + threadFinished.release(); + } }; + + + // Wait until our drain thread has launched + threadLaunched.acquire(); + + // The task itself hasn't started, so the drain should still be blocking + EXPECT_EQ(false, drainDone); + + // Allow the task to finish + taskStart.release(); + + // Wait for the drain thread to wrap up + threadFinished.acquire(); + + // We successfully drained the executor + EXPECT_EQ(true, drainDone); + + drainThread.join(); + } } // namespace UnitTest #if defined(HAVE_BENCHMARK) namespace Benchmark { - class JobGraphBenchmarkFixture : public ::benchmark::Fixture + class TaskGraphBenchmarkFixture : public ::benchmark::Fixture { public: void SetUp(benchmark::State&) override { - executor = new JobExecutor; - graph = new JobGraph; + executor = new TaskExecutor; + graph = new TaskGraph; } void TearDown(benchmark::State&) override @@ -521,38 +652,38 @@ namespace Benchmark delete executor; } - JobDescriptor descriptors[4] = { { "critical", "benchmark", JobPriority::CRITICAL }, - { "high", "benchmark", JobPriority::HIGH }, - { "medium", "benchmark", JobPriority::MEDIUM }, - { "low", "benchmark", JobPriority::LOW } }; + TaskDescriptor descriptors[4] = { { "critical", "benchmark", TaskPriority::CRITICAL }, + { "high", "benchmark", TaskPriority::HIGH }, + { "medium", "benchmark", TaskPriority::MEDIUM }, + { "low", "benchmark", TaskPriority::LOW } }; - JobGraph* graph; - JobExecutor* executor; + TaskGraph* graph; + TaskExecutor* executor; }; - BENCHMARK_F(JobGraphBenchmarkFixture, QueueToDequeue)(benchmark::State& state) + BENCHMARK_F(TaskGraphBenchmarkFixture, QueueToDequeue)(benchmark::State& state) { - graph->AddJob( + graph->AddTask( descriptors[2], [] { }); for (auto _ : state) { - JobGraphEvent ev; + TaskGraphEvent ev; graph->SubmitOnExecutor(*executor, &ev); ev.Wait(); } } - BENCHMARK_F(JobGraphBenchmarkFixture, OneAfterAnother)(benchmark::State& state) + BENCHMARK_F(TaskGraphBenchmarkFixture, OneAfterAnother)(benchmark::State& state) { - auto a = graph->AddJob( + auto a = graph->AddTask( descriptors[2], [] { }); - auto b = graph->AddJob( + auto b = graph->AddTask( descriptors[2], [] { @@ -561,15 +692,15 @@ namespace Benchmark for (auto _ : state) { - JobGraphEvent ev; + TaskGraphEvent ev; graph->SubmitOnExecutor(*executor, &ev); ev.Wait(); } } - BENCHMARK_F(JobGraphBenchmarkFixture, FourToOneJoin)(benchmark::State& state) + BENCHMARK_F(TaskGraphBenchmarkFixture, FourToOneJoin)(benchmark::State& state) { - auto [a, b, c, d, e] = graph->AddJobs( + auto [a, b, c, d, e] = graph->AddTasks( descriptors[2], [] { @@ -587,11 +718,11 @@ namespace Benchmark { }); - e.Succeeds(a, b, c, d); + e.Follows(a, b, c, d); for (auto _ : state) { - JobGraphEvent ev; + TaskGraphEvent ev; graph->SubmitOnExecutor(*executor, &ev); ev.Wait(); } diff --git a/Code/Framework/AzCore/Tests/azcoretests_files.cmake b/Code/Framework/AzCore/Tests/azcoretests_files.cmake index 480baabe7a..ca0e2862fc 100644 --- a/Code/Framework/AzCore/Tests/azcoretests_files.cmake +++ b/Code/Framework/AzCore/Tests/azcoretests_files.cmake @@ -40,7 +40,6 @@ set(FILES Interface.cpp IO/Path/PathTests.cpp IPC.cpp - JobGraphTests.cpp Jobs.cpp JSON.cpp FixedWidthIntegers.cpp @@ -66,6 +65,7 @@ set(FILES StreamerTests.cpp StringFunc.cpp SystemFile.cpp + TaskTests.cpp TickBusTest.cpp TimeDataStatistics.cpp UUIDTests.cpp