birkeh/o3de
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

JobGraph -> TaskGraph (and associated classes/files)

Browse Source 
This commit also addresses all PR feedback

Signed-off-by: Jeremy Ong <jcong@amazon.com>
This commit is contained in:
Jeremy Ong
2021-08-05 14:37:59 -06:00
parent d2f2a186cb
commit eaa6e087cf
19 changed files with 1109 additions and 910 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.cpp
-76
View File
@@ -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 <AzCore/Jobs/Internal/JobTypeEraser.h>
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
Code/Framework/AzCore/AzCore/Jobs/Internal/JobTypeEraser.h
-232
View File
@@ -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 <AzCore/Jobs/JobDescriptor.h>
#include <AzCore/std/containers/fixed_vector.h>
#include <AzCore/std/typetraits/is_assignable.h>
#include <AzCore/std/typetraits/is_destructible.h>
#include <AzCore/std/parallel/atomic.h>
#include <AzCore/Memory/SystemAllocator.h>
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<typename Lambda>
class JobTypeEraser final
{
public:
constexpr JobInvoke_t ErasedInvoker()
{
return reinterpret_cast<JobInvoke_t>(Invoker);
}
constexpr JobRelocate_t ErasedRelocator()
{
if constexpr (AZStd::is_trivially_move_constructible_v<Lambda>)
{
return nullptr;
}
else if constexpr (AZStd::is_move_constructible_v<Lambda>)
{
return reinterpret_cast<JobRelocate_t>(Mover);
}
else if constexpr (AZStd::is_copy_constructible_v<Lambda>)
{
return reinterpret_cast<JobRelocate_t>(Copyer);
}
else
{
static_assert(
AZStd::is_move_constructible_v<Lambda> || AZStd::is_copy_constructible_v<Lambda>,
"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<Lambda>)
{
return nullptr;
}
else
{
return reinterpret_cast<JobDestroy_t>(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<uint32_t>);
TypeErasedJob() = default;
template<typename Lambda>
TypeErasedJob(JobDescriptor const& desc, Lambda&& lambda) noexcept
: m_descriptor{ desc }
{
JobTypeEraser<Lambda> 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>(lambda), m_buffer);
m_lambda = m_buffer;
}
else
{
// Lambda has spilled to the heap (or requires extended alignment)
m_lambda = reinterpret_cast<char*>(azmalloc(sizeof(Lambda), alignof(Lambda)));
TypedRelocate(AZStd::forward<Lambda>(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<uint8_t>(m_descriptor.priority);
}
private:
friend class CompiledJobGraph;
friend class JobWorker;
// This relocation avoids branches needed if the lambda type is unknown
template<typename Lambda>
void TypedRelocate(Lambda&& lambda, char* destination)
{
if constexpr (AZStd::is_trivially_move_constructible_v<Lambda>)
{
memcpy(destination, reinterpret_cast<char*>(&lambda), sizeof(Lambda));
}
else if constexpr (AZStd::is_move_constructible_v<Lambda>)
{
new (destination) Lambda{ AZStd::move(lambda) };
}
else if constexpr (AZStd::is_copy_constructible_v<Lambda>)
{
new (destination) Lambda{ lambda };
}
else
{
static_assert(
AZStd::is_move_constructible_v<Lambda> || AZStd::is_copy_constructible_v<Lambda>,
"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<uint32_t> 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
Code/Framework/AzCore/AzCore/Jobs/JobDescriptor.h
-49
View File
@@ -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 <AzCore/base.h>
#include <AzCore/std/limits.h>
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;
};
}
Code/Framework/AzCore/AzCore/Jobs/JobExecutor.h
-85
View File
@@ -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 <AzCore/Jobs/Internal/JobTypeEraser.h>
#include <AzCore/Jobs/JobDescriptor.h>
#include <AzCore/std/containers/unordered_map.h>
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/parallel/atomic.h>
#include <AzCore/Memory/PoolAllocator.h>
namespace AZ
{
class JobGraphEvent;
class JobGraph;
namespace Internal
{
class CompiledJobGraph final
{
public:
AZ_CLASS_ALLOCATOR(CompiledJobGraph, SystemAllocator, 0)
CompiledJobGraph(
AZStd::vector<TypeErasedJob>&& jobs,
AZStd::unordered_map<uint32_t, AZStd::vector<uint32_t>>& links,
size_t linkCount,
JobGraph* parent);
AZStd::vector<TypeErasedJob>& 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<TypeErasedJob> m_jobs;
AZStd::vector<TypeErasedJob*> 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<uint32_t> 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<uint32_t> m_lastSubmission;
AZStd::atomic<uint64_t> m_remaining;
};
} // namespace AZ
Code/Framework/AzCore/AzCore/Jobs/JobGraph.cpp
-85
View File
@@ -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 <AzCore/Jobs/JobGraph.h>
#include <AzCore/Jobs/JobExecutor.h>
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();
}
}
}
Code/Framework/AzCore/AzCore/Jobs/JobGraph.h
-150
View File
@@ -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 <AzCore/Jobs/Internal/JobTypeEraser.h>
#include <AzCore/Jobs/JobDescriptor.h>
#include <AzCore/std/containers/array.h>
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/containers/unordered_map.h>
#include <AzCore/std/parallel/binary_semaphore.h>
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 <typename... JT>
void Precedes(JT&... tokens);
// Indicate that this job must finish after the job token(s) passed as the argument
template <typename... JT>
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<typename Lambda>
JobToken AddJob(JobDescriptor const& descriptor, Lambda&& lambda);
template <typename... Lambdas>
AZStd::array<JobToken, sizeof...(Lambdas)> 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<T> 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<Internal::TypeErasedJob> m_jobs;
// Job index |-> Dependent job indices
AZStd::unordered_map<uint32_t, AZStd::vector<uint32_t>> m_links;
uint32_t m_linkCount = 0;
bool m_retained = true;
AZStd::atomic<bool> m_submitted = false;
};
} // namespace AZ
#include <AzCore/Jobs/JobGraph.inl>
Code/Framework/AzCore/AzCore/Task/Internal/Task.cpp
+58
View File
@@ -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 <AzCore/Task/Internal/Task.h>
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
Code/Framework/AzCore/AzCore/Task/Internal/Task.h
+180
View File
@@ -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 <AzCore/Task/Internal/TaskConfig.h>
#include <AzCore/Task/TaskDescriptor.h>
#include <AzCore/std/containers/fixed_vector.h>
#include <AzCore/std/typetraits/is_assignable.h>
#include <AzCore/std/typetraits/is_destructible.h>
#include <AzCore/std/parallel/atomic.h>
#include <AzCore/Memory/PoolAllocator.h>
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<typename Lambda>
class TaskTypeEraser final
{
public:
constexpr TaskInvoke_t ErasedInvoker()
{
return reinterpret_cast<TaskInvoke_t>(Invoker);
}
constexpr TaskRelocate_t ErasedRelocator()
{
if constexpr (AZStd::is_trivially_move_constructible_v<Lambda>)
{
return nullptr;
}
else if constexpr (AZStd::is_move_constructible_v<Lambda>)
{
return reinterpret_cast<TaskRelocate_t>(Mover);
}
else if constexpr (AZStd::is_copy_constructible_v<Lambda>)
{
return reinterpret_cast<TaskRelocate_t>(Copier);
}
else
{
static_assert(
AZStd::is_move_constructible_v<Lambda> || AZStd::is_copy_constructible_v<Lambda>,
"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<Lambda>)
{
return nullptr;
}
else
{
return reinterpret_cast<TaskDestroy_t>(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<uint32_t>);
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<typename Lambda>
Task(TaskDescriptor const& desc, Lambda& lambda) = delete;
template<typename Lambda>
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<typename Lambda>
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<uint32_t> 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 <AzCore/Task/Internal/Task.inl>
Code/Framework/AzCore/AzCore/Task/Internal/Task.inl
+84
View File
@@ -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<typename Lambda>
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<Lambda> 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>(lambda), m_lambda);
}
template<typename Lambda>
void Task::TypedRelocate(Lambda&& lambda, char* destination)
{
if constexpr (AZStd::is_trivially_move_constructible_v<Lambda>)
{
memcpy(destination, reinterpret_cast<char*>(&lambda), sizeof(Lambda));
}
else if constexpr (AZStd::is_move_constructible_v<Lambda>)
{
new (destination) Lambda{ AZStd::move(lambda) };
}
else if constexpr (AZStd::is_copy_constructible_v<Lambda>)
{
new (destination) Lambda{ lambda };
}
else
{
static_assert(
AZStd::is_move_constructible_v<Lambda> || AZStd::is_copy_constructible_v<Lambda>,
"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<uint8_t>(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
Code/Framework/AzCore/AzCore/Task/Internal/TaskConfig.h
+14
View File
@@ -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 <AzCore/AzCore_Traits_Platform.h>
#if !defined(AZ_TRAIT_TASK_BYTE_SIZE)
#define AZ_TRAIT_TASK_BYTE_SIZE 128
#endif
Code/Framework/AzCore/AzCore/Task/TaskDescriptor.h
+49
View File
@@ -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 <AzCore/base.h>
#include <AzCore/std/limits.h>
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;
};
}
Code/Framework/AzCore/AzCore/Jobs/JobExecutor.cpp → Code/Framework/AzCore/AzCore/Task/TaskExecutor.cpp
+102 -74
View File
@@ -6,8 +6,8 @@
*
*/
#include <AzCore/Jobs/JobExecutor.h>
#include <AzCore/Jobs/JobGraph.h>
#include <AzCore/Task/TaskExecutor.h>
#include <AzCore/Task/TaskGraph.h>
#include <AzCore/std/containers/queue.h>
#include <AzCore/std/parallel/binary_semaphore.h>
@@ -17,46 +17,45 @@
#include <AzCore/std/parallel/semaphore.h>
#include <AzCore/std/parallel/thread.h>
#include <AzCore/std/string/string.h>
#include <AzCore/Module/Environment.h>
#include <random>
namespace AZ
{
constexpr static size_t PRIORITY_COUNT = static_cast<size_t>(JobPriority::PRIORITY_COUNT);
namespace Internal
{
CompiledJobGraph::CompiledJobGraph(
AZStd::vector<TypeErasedJob>&& jobs,
CompiledTaskGraph::CompiledTaskGraph(
AZStd::vector<Task>&& tasks,
AZStd::unordered_map<uint32_t, AZStd::vector<uint32_t>>& 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<size_t>(job.m_successorOffset) + j] = &m_jobs[links[i][j]];
m_successors[static_cast<size_t>(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<uint16_t> 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<uint8_t>(JobPriority::PRIORITY_COUNT);
constexpr static uint8_t PriorityLevelCount = static_cast<uint8_t>(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<bool> 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<TaskExecutor*> 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<TaskExecutor*>(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<TaskExecutor*>("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<Internal::JobWorker*>(azmalloc(m_threadCount * sizeof(Internal::JobWorker)));
m_workers = reinterpret_cast<Internal::TaskWorker*>(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
Code/Framework/AzCore/AzCore/Task/TaskExecutor.h
+94
View File
@@ -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 <AzCore/Task/Internal/Task.h>
#include <AzCore/Task/TaskDescriptor.h>
#include <AzCore/std/containers/unordered_map.h>
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/parallel/atomic.h>
#include <AzCore/std/parallel/binary_semaphore.h>
#include <AzCore/Memory/PoolAllocator.h>
namespace AZ
{
class TaskGraphEvent;
class TaskGraph;
namespace Internal
{
class CompiledTaskGraph final
{
public:
AZ_CLASS_ALLOCATOR(CompiledTaskGraph, SystemAllocator, 0)
CompiledTaskGraph(
AZStd::vector<Task>&& tasks,
AZStd::unordered_map<uint32_t, AZStd::vector<uint32_t>>& links,
size_t linkCount,
TaskGraph* parent);
AZStd::vector<Task>& 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<Task> m_tasks;
AZStd::vector<Task*> 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<uint32_t> 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<uint32_t> m_lastSubmission;
AZStd::atomic<uint64_t> m_graphsRemaining;
AZStd::atomic<bool> m_isDraining;
AZStd::binary_semaphore m_drainSemaphore;
};
} // namespace AZ
Code/Framework/AzCore/AzCore/Task/TaskGraph.cpp
+85
View File
@@ -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 <AzCore/Task/TaskGraph.h>
#include <AzCore/Task/TaskExecutor.h>
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();
}
}
}
Code/Framework/AzCore/AzCore/Task/TaskGraph.h
+151
View File
@@ -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 <AzCore/Task/Internal/Task.h>
#include <AzCore/Task/TaskDescriptor.h>
#include <AzCore/std/containers/array.h>
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/containers/unordered_map.h>
#include <AzCore/std/parallel/binary_semaphore.h>
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 <typename... JT>
void Precedes(JT&... tokens);
// Indicate that this task must finish after the task token(s) passed as the argument
template <typename... JT>
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<typename Lambda>
TaskToken AddTask(TaskDescriptor const& descriptor, Lambda&& lambda);
template <typename... Lambdas>
AZStd::array<TaskToken, sizeof...(Lambdas)> 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<T> 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<Internal::Task> m_tasks;
// Task index |-> Dependent task indices
AZStd::unordered_map<uint32_t, AZStd::vector<uint32_t>> m_links;
uint32_t m_linkCount = 0;
bool m_retained = true;
AZStd::atomic<bool> m_submitted = false;
};
} // namespace AZ
#include <AzCore/Task/TaskGraph.inl>
Code/Framework/AzCore/AzCore/Jobs/JobGraph.inl → Code/Framework/AzCore/AzCore/Task/TaskGraph.inl
+13 -13
View File
@@ -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<typename... JT>
inline void JobToken::Precedes(JT&... tokens)
void TaskToken::Precedes(JT&... tokens)
{
(PrecedesInternal(tokens), ...);
}
template <typename... JT>
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<typename Lambda>
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>(lambda));
m_tasks.emplace_back(desc, AZStd::forward<Lambda>(lambda));
return { *this, m_jobs.size() - 1 };
return { *this, m_tasks.size() - 1 };
}
template <typename... Lambdas>
inline AZStd::array<JobToken, sizeof...(Lambdas)> JobGraph::AddJobs(JobDescriptor const& descriptor, Lambdas&&... lambdas)
AZStd::array<TaskToken, sizeof...(Lambdas)> TaskGraph::AddTasks(TaskDescriptor const& descriptor, Lambdas&&... lambdas)
{
return { AddJob(descriptor, AZStd::forward<Lambdas>(lambdas))... };
return { AddTask(descriptor, AZStd::forward<Lambdas>(lambdas))... };
}
inline void JobGraph::Detach()
inline void TaskGraph::Detach()
{
m_retained = false;
}
Code/Framework/AzCore/AzCore/azcore_files.cmake
+10 -8
View File
@@ -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
Code/Framework/AzCore/Tests/JobGraphTests.cpp → Code/Framework/AzCore/Tests/TaskTests.cpp
+268 -137
View File
@@ -6,26 +6,26 @@
*
*/
#include <AzCore/Jobs/JobGraph.h>
#include <AzCore/Jobs/JobExecutor.h>
#include <AzCore/Task/TaskGraph.h>
#include <AzCore/Task/TaskExecutor.h>
#include <AzCore/Memory/PoolAllocator.h>
#include <AzCore/UnitTest/TestTypes.h>
#include <random>
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<AZ::PoolAllocator>::Create();
AZ::AllocatorInstance<AZ::ThreadPoolAllocator>::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, &copyCount]
{
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, &copyCount]
{
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<int> 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<int> 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<int> 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();
}
Code/Framework/AzCore/Tests/azcoretests_files.cmake
+1 -1
View File
@@ -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