Merge pull request #2815 from aws-lumberyard-dev/AzCore/JobGraph
Add initial TaskGraph prototype
This commit is contained in:
Jeremy Ong
GitHub
@@ -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
|
||||
@@ -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>
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,362 @@
|
||||
/*
|
||||
* 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/TaskExecutor.h>
|
||||
#include <AzCore/Task/TaskGraph.h>
|
||||
|
||||
#include <AzCore/std/containers/queue.h>
|
||||
#include <AzCore/std/parallel/binary_semaphore.h>
|
||||
#include <AzCore/std/parallel/exponential_backoff.h>
|
||||
#include <AzCore/std/parallel/mutex.h>
|
||||
#include <AzCore/std/parallel/scoped_lock.h>
|
||||
#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
|
||||
{
|
||||
namespace Internal
|
||||
{
|
||||
CompiledTaskGraph::CompiledTaskGraph(
|
||||
AZStd::vector<Task>&& tasks,
|
||||
AZStd::unordered_map<uint32_t, AZStd::vector<uint32_t>>& links,
|
||||
size_t linkCount,
|
||||
TaskGraph* parent)
|
||||
: m_parent{ parent }
|
||||
{
|
||||
m_tasks = AZStd::move(tasks);
|
||||
m_successors.resize(linkCount);
|
||||
|
||||
Task** cursor = m_successors.data();
|
||||
|
||||
for (size_t i = 0; i != m_tasks.size(); ++i)
|
||||
{
|
||||
Task& task = m_tasks[i];
|
||||
task.m_graph = this;
|
||||
task.m_successorOffset = cursor - m_successors.data();
|
||||
cursor += task.m_outboundLinkCount;
|
||||
|
||||
AZ_Assert(task.m_outboundLinkCount == links[i].size(), "Task outbound link information mismatch");
|
||||
|
||||
for (uint32_t j = 0; j != task.m_outboundLinkCount; ++j)
|
||||
{
|
||||
m_successors[static_cast<size_t>(task.m_successorOffset) + j] = &m_tasks[links[i][j]];
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Check for dependency cycles
|
||||
}
|
||||
|
||||
uint32_t CompiledTaskGraph::Release()
|
||||
{
|
||||
uint32_t remaining = --m_remaining;
|
||||
|
||||
if (m_parent)
|
||||
{
|
||||
if (remaining == 1)
|
||||
{
|
||||
// Allow the parent graph to be submitted again
|
||||
m_parent->m_submitted = false;
|
||||
}
|
||||
}
|
||||
else if (remaining == 0)
|
||||
{
|
||||
if (m_waitEvent)
|
||||
{
|
||||
m_waitEvent->Signal();
|
||||
}
|
||||
|
||||
azdestroy(this);
|
||||
return remaining;
|
||||
}
|
||||
|
||||
if (m_waitEvent && remaining == (m_parent ? 1 : 0))
|
||||
{
|
||||
m_waitEvent->Signal();
|
||||
}
|
||||
|
||||
return remaining;
|
||||
}
|
||||
|
||||
struct QueueStatus
|
||||
{
|
||||
AZStd::atomic<uint16_t> head;
|
||||
AZStd::atomic<uint16_t> tail;
|
||||
AZStd::atomic<uint16_t> reserve;
|
||||
};
|
||||
|
||||
// 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 TaskQueue final
|
||||
{
|
||||
public:
|
||||
// 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>(TaskPriority::PRIORITY_COUNT);
|
||||
|
||||
TaskQueue() = default;
|
||||
TaskQueue(const TaskQueue&) = delete;
|
||||
TaskQueue& operator=(const TaskQueue&) = delete;
|
||||
|
||||
void Enqueue(Task* task);
|
||||
Task* TryDequeue();
|
||||
|
||||
private:
|
||||
QueueStatus m_status[PriorityLevelCount] = {};
|
||||
Task* m_queues[PriorityLevelCount][MaxQueueSize] = {};
|
||||
};
|
||||
|
||||
void TaskQueue::Enqueue(Task* task)
|
||||
{
|
||||
uint8_t priority = task->GetPriorityNumber();
|
||||
QueueStatus& status = m_status[priority];
|
||||
|
||||
AZStd::exponential_backoff backoff;
|
||||
while (true)
|
||||
{
|
||||
uint16_t reserve = status.reserve.load();
|
||||
uint16_t head = status.head.load();
|
||||
|
||||
// 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] = task;
|
||||
|
||||
uint16_t expectedReserve = reserve;
|
||||
|
||||
// Increment the tail to advertise the new task
|
||||
while (!status.tail.compare_exchange_weak(expectedReserve, reserve + 1))
|
||||
{
|
||||
expectedReserve = reserve;
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// We failed to reserve a slot, try again
|
||||
}
|
||||
else
|
||||
{
|
||||
backoff.wait();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Task* TaskQueue::TryDequeue()
|
||||
{
|
||||
for (size_t priority = 0; priority != PriorityLevelCount; ++priority)
|
||||
{
|
||||
QueueStatus& status = m_status[priority];
|
||||
while (true)
|
||||
{
|
||||
uint16_t head = status.head.load();
|
||||
uint16_t tail = status.tail.load();
|
||||
if (head == tail)
|
||||
{
|
||||
// Queue empty
|
||||
break;
|
||||
}
|
||||
else
|
||||
{
|
||||
Task* task = m_queues[priority][status.head];
|
||||
if (status.head.compare_exchange_weak(head, head + 1))
|
||||
{
|
||||
return task;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
class TaskWorker
|
||||
{
|
||||
public:
|
||||
void Spawn(::AZ::TaskExecutor& executor, size_t id, AZStd::semaphore& initSemaphore, bool affinitize)
|
||||
{
|
||||
m_executor = &executor;
|
||||
|
||||
AZStd::string threadName = AZStd::string::format("TaskWorker %zu", id);
|
||||
AZStd::thread_desc desc = {};
|
||||
desc.m_name = threadName.c_str();
|
||||
if (affinitize)
|
||||
{
|
||||
desc.m_cpuId = 1 << id;
|
||||
}
|
||||
m_active.store(true, AZStd::memory_order_release);
|
||||
|
||||
m_thread = AZStd::thread{ [this, &initSemaphore]
|
||||
{
|
||||
initSemaphore.release();
|
||||
Run();
|
||||
},
|
||||
&desc };
|
||||
}
|
||||
|
||||
void Join()
|
||||
{
|
||||
m_active.store(false, AZStd::memory_order_release);
|
||||
m_semaphore.release();
|
||||
m_thread.join();
|
||||
}
|
||||
|
||||
void Enqueue(Task* task)
|
||||
{
|
||||
m_queue.Enqueue(task);
|
||||
|
||||
if (!m_busy.exchange(true))
|
||||
{
|
||||
// The worker was idle prior to enqueueing the task, release the semaphore
|
||||
m_semaphore.release();
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
void Run()
|
||||
{
|
||||
while (m_active)
|
||||
{
|
||||
m_busy = false;
|
||||
m_semaphore.acquire();
|
||||
|
||||
if (!m_active)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
m_busy = true;
|
||||
|
||||
Task* task = m_queue.TryDequeue();
|
||||
while (task)
|
||||
{
|
||||
task->Invoke();
|
||||
// Decrement counts for all task successors
|
||||
for (size_t j = 0; j != task->m_outboundLinkCount; ++j)
|
||||
{
|
||||
Task* successor = task->m_graph->m_successors[task->m_successorOffset + j];
|
||||
if (--successor->m_dependencyCount == 0)
|
||||
{
|
||||
m_executor->Submit(*successor);
|
||||
}
|
||||
}
|
||||
|
||||
bool isRetained = task->m_graph->m_parent != nullptr;
|
||||
if (task->m_graph->Release() == (isRetained ? 1 : 0))
|
||||
{
|
||||
m_executor->ReleaseGraph();
|
||||
}
|
||||
|
||||
task = m_queue.TryDequeue();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
AZStd::thread m_thread;
|
||||
AZStd::atomic<bool> m_active;
|
||||
AZStd::atomic<bool> m_busy;
|
||||
AZStd::binary_semaphore m_semaphore;
|
||||
|
||||
::AZ::TaskExecutor* m_executor;
|
||||
TaskQueue m_queue;
|
||||
};
|
||||
} // namespace Internal
|
||||
|
||||
static EnvironmentVariable<TaskExecutor*> s_executor;
|
||||
constexpr static const char* s_executorName = "GlobalTaskExecutor";
|
||||
TaskExecutor& TaskExecutor::Instance()
|
||||
{
|
||||
if (!s_executor)
|
||||
{
|
||||
s_executor = AZ::Environment::FindVariable<TaskExecutor*>(s_executorName);
|
||||
}
|
||||
|
||||
return **s_executor;
|
||||
}
|
||||
|
||||
// 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::TaskWorker*>(azmalloc(m_threadCount * sizeof(Internal::TaskWorker)));
|
||||
|
||||
bool affinitize = m_threadCount == AZStd::thread::hardware_concurrency();
|
||||
|
||||
AZStd::semaphore initSemaphore;
|
||||
|
||||
for (size_t i = 0; i != m_threadCount; ++i)
|
||||
{
|
||||
new (m_workers + i) Internal::TaskWorker{};
|
||||
m_workers[i].Spawn(*this, i, initSemaphore, affinitize);
|
||||
}
|
||||
|
||||
for (size_t i = 0; i != m_threadCount; ++i)
|
||||
{
|
||||
initSemaphore.acquire();
|
||||
}
|
||||
}
|
||||
|
||||
TaskExecutor::~TaskExecutor()
|
||||
{
|
||||
for (size_t i = 0; i != m_threadCount; ++i)
|
||||
{
|
||||
m_workers[i].Join();
|
||||
m_workers[i].~TaskWorker();
|
||||
}
|
||||
|
||||
azfree(m_workers);
|
||||
}
|
||||
|
||||
void TaskExecutor::Submit(Internal::CompiledTaskGraph& graph)
|
||||
{
|
||||
++m_graphsRemaining;
|
||||
// Submit all tasks that have no inbound edges
|
||||
for (Internal::Task& task : graph.Tasks())
|
||||
{
|
||||
if (task.IsRoot())
|
||||
{
|
||||
Submit(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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_workers[++m_lastSubmission % m_threadCount].Enqueue(&task);
|
||||
}
|
||||
|
||||
void TaskExecutor::ReleaseGraph()
|
||||
{
|
||||
--m_graphsRemaining;
|
||||
}
|
||||
} // namespace AZ
|
||||
@@ -0,0 +1,89 @@
|
||||
/*
|
||||
* 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);
|
||||
|
||||
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;
|
||||
};
|
||||
} // namespace AZ
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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>
|
||||
@@ -0,0 +1,66 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
namespace AZ
|
||||
{
|
||||
inline TaskToken::TaskToken(TaskGraph& parent, size_t index)
|
||||
: m_parent{ parent }
|
||||
, m_index{ index }
|
||||
{
|
||||
}
|
||||
|
||||
template<typename... JT>
|
||||
void TaskToken::Precedes(JT&... tokens)
|
||||
{
|
||||
(PrecedesInternal(tokens), ...);
|
||||
}
|
||||
|
||||
template <typename... JT>
|
||||
void TaskToken::Follows(JT&... tokens)
|
||||
{
|
||||
(tokens.PrecedesInternal(*this), ...);
|
||||
}
|
||||
|
||||
inline bool TaskGraphEvent::IsSignaled()
|
||||
{
|
||||
return m_semaphore.try_acquire_for(AZStd::chrono::milliseconds{ 0 });
|
||||
}
|
||||
|
||||
inline void TaskGraphEvent::Wait()
|
||||
{
|
||||
m_semaphore.acquire();
|
||||
}
|
||||
|
||||
inline void TaskGraphEvent::Signal()
|
||||
{
|
||||
m_semaphore.release();
|
||||
}
|
||||
|
||||
template<typename Lambda>
|
||||
TaskToken TaskGraph::AddTask(TaskDescriptor const& desc, Lambda&& lambda)
|
||||
{
|
||||
AZ_Assert(!m_submitted, "Cannot mutate a TaskGraph that was previously submitted or in flight.");
|
||||
|
||||
m_tasks.emplace_back(desc, AZStd::forward<Lambda>(lambda));
|
||||
|
||||
return { *this, m_tasks.size() - 1 };
|
||||
}
|
||||
|
||||
template <typename... Lambdas>
|
||||
AZStd::array<TaskToken, sizeof...(Lambdas)> TaskGraph::AddTasks(TaskDescriptor const& descriptor, Lambdas&&... lambdas)
|
||||
{
|
||||
return { AddTask(descriptor, AZStd::forward<Lambdas>(lambdas))... };
|
||||
}
|
||||
|
||||
inline void TaskGraph::Detach()
|
||||
{
|
||||
m_retained = false;
|
||||
}
|
||||
} // namespace AZ
|
||||
@@ -616,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
|
||||
|
||||
@@ -69,8 +69,7 @@ namespace AZStd
|
||||
int m_priority{ -100000 };
|
||||
|
||||
//! The CPU ids (as a bitfield) that this thread will be running on, see \ref AZStd::thread_desc::m_cpuId.
|
||||
//! Windows: This parameter is ignored.
|
||||
//! On other platforms, each bit maps directly to the core numbers [0-n], default is 0
|
||||
//! Each bit maps directly to the core numbers [0-n], default is 0
|
||||
int m_cpuId{ AFFINITY_MASK_ALL };
|
||||
|
||||
//! If we can join the thread.
|
||||
|
||||
@@ -0,0 +1,642 @@
|
||||
/*
|
||||
* 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>
|
||||
#include <AzCore/Memory/PoolAllocator.h>
|
||||
|
||||
#include <AzCore/UnitTest/TestTypes.h>
|
||||
|
||||
#include <random>
|
||||
|
||||
using AZ::TaskDescriptor;
|
||||
using AZ::TaskGraph;
|
||||
using AZ::TaskGraphEvent;
|
||||
using AZ::TaskExecutor;
|
||||
using AZ::Internal::Task;
|
||||
using AZ::TaskPriority;
|
||||
|
||||
static TaskDescriptor defaultTD{ "TaskGraphTestTask", "TaskGraphTests" };
|
||||
|
||||
namespace UnitTest
|
||||
{
|
||||
class TaskGraphTestFixture : public AllocatorsTestFixture
|
||||
{
|
||||
public:
|
||||
void SetUp() override
|
||||
{
|
||||
AllocatorsTestFixture::SetUp();
|
||||
AZ::AllocatorInstance<AZ::PoolAllocator>::Create();
|
||||
AZ::AllocatorInstance<AZ::ThreadPoolAllocator>::Create();
|
||||
|
||||
m_executor = aznew TaskExecutor(4);
|
||||
}
|
||||
|
||||
void TearDown() override
|
||||
{
|
||||
azdestroy(m_executor);
|
||||
AZ::AllocatorInstance<AZ::ThreadPoolAllocator>::Destroy();
|
||||
AZ::AllocatorInstance<AZ::PoolAllocator>::Destroy();
|
||||
AllocatorsTestFixture::TearDown();
|
||||
}
|
||||
|
||||
protected:
|
||||
TaskExecutor* m_executor;
|
||||
};
|
||||
|
||||
TEST(TaskGraphTests, TrivialTaskLambda)
|
||||
{
|
||||
int x = 0;
|
||||
|
||||
Task task(
|
||||
defaultTD,
|
||||
[&x]()
|
||||
{
|
||||
++x;
|
||||
});
|
||||
task.Invoke();
|
||||
|
||||
EXPECT_EQ(1, x);
|
||||
}
|
||||
|
||||
TEST(TaskGraphTests, TrivialTaskLambdaMove)
|
||||
{
|
||||
int x = 0;
|
||||
|
||||
Task task(
|
||||
defaultTD,
|
||||
[&x]()
|
||||
{
|
||||
++x;
|
||||
});
|
||||
|
||||
Task task2 = AZStd::move(task);
|
||||
|
||||
task2.Invoke();
|
||||
|
||||
EXPECT_EQ(1, x);
|
||||
}
|
||||
|
||||
struct TrackMoves
|
||||
{
|
||||
TrackMoves() = default;
|
||||
|
||||
TrackMoves(const TrackMoves&) = delete;
|
||||
|
||||
TrackMoves(TrackMoves&& other)
|
||||
: moveCount{other.moveCount + 1}
|
||||
{
|
||||
}
|
||||
|
||||
int moveCount = 0;
|
||||
};
|
||||
|
||||
struct TrackCopies
|
||||
{
|
||||
TrackCopies() = default;
|
||||
|
||||
TrackCopies(TrackCopies&&) = delete;
|
||||
|
||||
TrackCopies(const TrackCopies& other)
|
||||
: copyCount{other.copyCount + 1}
|
||||
{
|
||||
}
|
||||
|
||||
int copyCount = 0;
|
||||
};
|
||||
|
||||
/*
|
||||
TEST(TaskGraphTests, ThisShouldNotCompile)
|
||||
{
|
||||
auto lambda = []
|
||||
{
|
||||
};
|
||||
|
||||
Task task(defaultTD, lambda);
|
||||
task.Invoke();
|
||||
}
|
||||
*/
|
||||
|
||||
TEST(TaskGraphTests, MoveOnlyTaskLambda)
|
||||
{
|
||||
TrackMoves tm;
|
||||
int moveCount = 0;
|
||||
|
||||
Task task(
|
||||
defaultTD,
|
||||
[tm = AZStd::move(tm), &moveCount]
|
||||
{
|
||||
moveCount = tm.moveCount;
|
||||
});
|
||||
task.Invoke();
|
||||
|
||||
// Two moves are expected. Once into the capture body of the lambda, once to construct
|
||||
// the type erased task
|
||||
EXPECT_EQ(2, moveCount);
|
||||
}
|
||||
|
||||
TEST(TaskGraphTests, MoveOnlyTaskLambdaMove)
|
||||
{
|
||||
TrackMoves tm;
|
||||
int moveCount = 0;
|
||||
|
||||
Task task(
|
||||
defaultTD,
|
||||
[tm = AZStd::move(tm), &moveCount]
|
||||
{
|
||||
moveCount = tm.moveCount;
|
||||
});
|
||||
|
||||
Task task2 = AZStd::move(task);
|
||||
task2.Invoke();
|
||||
|
||||
EXPECT_EQ(3, moveCount);
|
||||
}
|
||||
|
||||
TEST(TaskGraphTests, CopyOnlyTaskLambda)
|
||||
{
|
||||
TrackCopies tc;
|
||||
int copyCount = 0;
|
||||
|
||||
Task task(
|
||||
defaultTD,
|
||||
[tc, ©Count]
|
||||
{
|
||||
copyCount = tc.copyCount;
|
||||
});
|
||||
task.Invoke();
|
||||
|
||||
// Two copies are expected. Once into the capture body of the lambda, once to construct
|
||||
// the type erased task
|
||||
EXPECT_EQ(2, copyCount);
|
||||
}
|
||||
|
||||
TEST(TaskGraphTests, CopyOnlyTaskLambdaMove)
|
||||
{
|
||||
TrackCopies tc;
|
||||
int copyCount = 0;
|
||||
|
||||
Task task(
|
||||
defaultTD,
|
||||
[tc, ©Count]
|
||||
{
|
||||
copyCount = tc.copyCount;
|
||||
});
|
||||
Task task2 = AZStd::move(task);
|
||||
task2.Invoke();
|
||||
|
||||
EXPECT_EQ(3, copyCount);
|
||||
}
|
||||
|
||||
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.
|
||||
int x = 0;
|
||||
struct TrackDestroy
|
||||
{
|
||||
TrackDestroy(int* px)
|
||||
: count{ px }
|
||||
{
|
||||
}
|
||||
TrackDestroy(TrackDestroy&& other)
|
||||
: count{ other.count }
|
||||
{
|
||||
other.count = nullptr;
|
||||
}
|
||||
~TrackDestroy()
|
||||
{
|
||||
if (count)
|
||||
{
|
||||
++*count;
|
||||
}
|
||||
}
|
||||
int* count = nullptr;
|
||||
};
|
||||
|
||||
{
|
||||
TrackDestroy td{ &x };
|
||||
Task task(
|
||||
defaultTD,
|
||||
[td = AZStd::move(td)]
|
||||
{
|
||||
});
|
||||
task.Invoke();
|
||||
// Destructor should not have run yet (except on moved-from instances)
|
||||
EXPECT_EQ(x, 0);
|
||||
}
|
||||
|
||||
// Destructor should have run now
|
||||
EXPECT_EQ(x, 1);
|
||||
}
|
||||
|
||||
TEST_F(TaskGraphTestFixture, VariadicInterface)
|
||||
{
|
||||
int x = 0;
|
||||
|
||||
TaskGraph graph;
|
||||
auto [a, b, c] = graph.AddTasks(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x += 3;
|
||||
},
|
||||
[&]
|
||||
{
|
||||
x = 4 * x;
|
||||
},
|
||||
[&]
|
||||
{
|
||||
x -= 1;
|
||||
});
|
||||
|
||||
a.Precedes(b);
|
||||
b.Precedes(c);
|
||||
|
||||
TaskGraphEvent ev;
|
||||
graph.SubmitOnExecutor(*m_executor, &ev);
|
||||
ev.Wait();
|
||||
|
||||
EXPECT_EQ(11, x);
|
||||
}
|
||||
|
||||
TEST_F(TaskGraphTestFixture, SerialGraph)
|
||||
{
|
||||
int x = 0;
|
||||
|
||||
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;
|
||||
|
||||
{
|
||||
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);
|
||||
graph.Detach();
|
||||
graph.SubmitOnExecutor(*m_executor, &ev);
|
||||
}
|
||||
|
||||
ev.Wait();
|
||||
|
||||
EXPECT_EQ(11, x);
|
||||
}
|
||||
|
||||
TEST_F(TaskGraphTestFixture, ForkJoin)
|
||||
{
|
||||
AZStd::atomic<int> x = 0;
|
||||
|
||||
// Task a initializes x to 3
|
||||
// Task b and c toggles the lowest two bits atomically
|
||||
// Task d decrements x
|
||||
|
||||
TaskGraph graph;
|
||||
auto a = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x = 0b111;
|
||||
});
|
||||
auto b = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 1;
|
||||
});
|
||||
auto c = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 2;
|
||||
});
|
||||
auto d = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x -= 1;
|
||||
});
|
||||
|
||||
// a <-- Root
|
||||
// / \
|
||||
// b c
|
||||
// \ /
|
||||
// d
|
||||
|
||||
a.Precedes(b, c);
|
||||
d.Follows(b, c);
|
||||
|
||||
TaskGraphEvent ev;
|
||||
graph.SubmitOnExecutor(*m_executor, &ev);
|
||||
ev.Wait();
|
||||
|
||||
EXPECT_EQ(3, x);
|
||||
}
|
||||
|
||||
TEST_F(TaskGraphTestFixture, SpawnSubgraph)
|
||||
{
|
||||
AZStd::atomic<int> x = 0;
|
||||
|
||||
TaskGraph graph;
|
||||
auto a = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x = 0b111;
|
||||
});
|
||||
auto b = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 1;
|
||||
});
|
||||
auto c = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 2;
|
||||
|
||||
TaskGraph subgraph;
|
||||
auto e = subgraph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 0b1000;
|
||||
});
|
||||
auto f = subgraph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 0b10000;
|
||||
});
|
||||
auto g = subgraph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x += 0b1000;
|
||||
});
|
||||
e.Precedes(g);
|
||||
f.Precedes(g);
|
||||
TaskGraphEvent ev;
|
||||
subgraph.SubmitOnExecutor(*m_executor, &ev);
|
||||
ev.Wait();
|
||||
});
|
||||
auto d = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x -= 1;
|
||||
});
|
||||
|
||||
// NOTE: The ideal way to express this topology is without the wait on the subgraph
|
||||
// at task g, but this is more an illustrative test. Better is to express the entire
|
||||
// graph in a single larger graph.
|
||||
// a <-- Root
|
||||
// / \
|
||||
// b c - f
|
||||
// \ \ \
|
||||
// \ e - g
|
||||
// \ /
|
||||
// \ /
|
||||
// \ /
|
||||
// d
|
||||
|
||||
a.Precedes(b);
|
||||
a.Precedes(c);
|
||||
b.Precedes(d);
|
||||
c.Precedes(d);
|
||||
|
||||
TaskGraphEvent ev;
|
||||
graph.SubmitOnExecutor(*m_executor, &ev);
|
||||
ev.Wait();
|
||||
|
||||
EXPECT_EQ(3 | 0b100000, x);
|
||||
}
|
||||
|
||||
TEST_F(TaskGraphTestFixture, RetainedGraph)
|
||||
{
|
||||
AZStd::atomic<int> x = 0;
|
||||
|
||||
TaskGraph graph;
|
||||
auto a = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x = 0b111;
|
||||
});
|
||||
auto b = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 1;
|
||||
});
|
||||
auto c = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 2;
|
||||
});
|
||||
auto d = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x -= 1;
|
||||
});
|
||||
auto e = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 0b1000;
|
||||
});
|
||||
auto f = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x ^= 0b10000;
|
||||
});
|
||||
auto g = graph.AddTask(
|
||||
defaultTD,
|
||||
[&]
|
||||
{
|
||||
x += 0b1000;
|
||||
});
|
||||
|
||||
// a <-- Root
|
||||
// / \
|
||||
// b c - f
|
||||
// \ \ \
|
||||
// \ e - g
|
||||
// \ /
|
||||
// \ /
|
||||
// \ /
|
||||
// d
|
||||
|
||||
a.Precedes(b, c);
|
||||
b.Precedes(d);
|
||||
c.Precedes(e, f);
|
||||
g.Follows(e, f);
|
||||
g.Precedes(d);
|
||||
|
||||
TaskGraphEvent ev;
|
||||
graph.SubmitOnExecutor(*m_executor, &ev);
|
||||
ev.Wait();
|
||||
|
||||
EXPECT_EQ(3 | 0b100000, x);
|
||||
x = 0;
|
||||
|
||||
graph.SubmitOnExecutor(*m_executor, &ev);
|
||||
ev.Wait();
|
||||
|
||||
EXPECT_EQ(3 | 0b100000, x);
|
||||
}
|
||||
} // namespace UnitTest
|
||||
|
||||
#if defined(HAVE_BENCHMARK)
|
||||
namespace Benchmark
|
||||
{
|
||||
class TaskGraphBenchmarkFixture : public ::benchmark::Fixture
|
||||
{
|
||||
public:
|
||||
void SetUp(benchmark::State&) override
|
||||
{
|
||||
executor = new TaskExecutor;
|
||||
graph = new TaskGraph;
|
||||
}
|
||||
|
||||
void TearDown(benchmark::State&) override
|
||||
{
|
||||
delete graph;
|
||||
delete executor;
|
||||
}
|
||||
|
||||
TaskDescriptor descriptors[4] = { { "critical", "benchmark", TaskPriority::CRITICAL },
|
||||
{ "high", "benchmark", TaskPriority::HIGH },
|
||||
{ "medium", "benchmark", TaskPriority::MEDIUM },
|
||||
{ "low", "benchmark", TaskPriority::LOW } };
|
||||
|
||||
TaskGraph* graph;
|
||||
TaskExecutor* executor;
|
||||
};
|
||||
|
||||
BENCHMARK_F(TaskGraphBenchmarkFixture, QueueToDequeue)(benchmark::State& state)
|
||||
{
|
||||
graph->AddTask(
|
||||
descriptors[2],
|
||||
[]
|
||||
{
|
||||
});
|
||||
for (auto _ : state)
|
||||
{
|
||||
TaskGraphEvent ev;
|
||||
graph->SubmitOnExecutor(*executor, &ev);
|
||||
ev.Wait();
|
||||
}
|
||||
}
|
||||
|
||||
BENCHMARK_F(TaskGraphBenchmarkFixture, OneAfterAnother)(benchmark::State& state)
|
||||
{
|
||||
auto a = graph->AddTask(
|
||||
descriptors[2],
|
||||
[]
|
||||
{
|
||||
});
|
||||
auto b = graph->AddTask(
|
||||
descriptors[2],
|
||||
[]
|
||||
{
|
||||
});
|
||||
a.Precedes(b);
|
||||
|
||||
for (auto _ : state)
|
||||
{
|
||||
TaskGraphEvent ev;
|
||||
graph->SubmitOnExecutor(*executor, &ev);
|
||||
ev.Wait();
|
||||
}
|
||||
}
|
||||
|
||||
BENCHMARK_F(TaskGraphBenchmarkFixture, FourToOneJoin)(benchmark::State& state)
|
||||
{
|
||||
auto [a, b, c, d, e] = graph->AddTasks(
|
||||
descriptors[2],
|
||||
[]
|
||||
{
|
||||
},
|
||||
[]
|
||||
{
|
||||
},
|
||||
[]
|
||||
{
|
||||
},
|
||||
[]
|
||||
{
|
||||
},
|
||||
[]
|
||||
{
|
||||
});
|
||||
|
||||
e.Follows(a, b, c, d);
|
||||
|
||||
for (auto _ : state)
|
||||
{
|
||||
TaskGraphEvent ev;
|
||||
graph->SubmitOnExecutor(*executor, &ev);
|
||||
ev.Wait();
|
||||
}
|
||||
}
|
||||
} // namespace Benchmark
|
||||
#endif
|
||||
@@ -65,6 +65,7 @@ set(FILES
|
||||
StreamerTests.cpp
|
||||
StringFunc.cpp
|
||||
SystemFile.cpp
|
||||
TaskTests.cpp
|
||||
TickBusTest.cpp
|
||||
TimeDataStatistics.cpp
|
||||
UUIDTests.cpp
|
||||
|
||||
Reference in New Issue
Block a user