Merge pull request #4424 from aws-lumberyard-dev/rgba16f/AZJobToIdle

Move Job system towards being used for Idle tasks

Code/Framework/AzCore/AzCore/Jobs/JobManagerComponent.cpp

@@ -18,6 +18,14 @@
 #include <AzCore/std/parallel/thread.h>
 #include <AzCore/Math/MathUtils.h>
 
+#include <AzCore/Console/IConsole.h>
+
+#include <AzCore/Threading/ThreadUtils.h>
+
+AZ_CVAR(float, cl_jobThreadsConcurrencyRatio, 0.6f, nullptr, AZ::ConsoleFunctorFlags::Null, "Legacy Job system multiplier on the number of hw threads the machine creates at initialization");
+AZ_CVAR(uint32_t, cl_jobThreadsNumReserved, 2, nullptr, AZ::ConsoleFunctorFlags::Null, "Legacy Job system number of hardware threads that are reserved for O3DE system threads");
+AZ_CVAR(uint32_t, cl_jobThreadsMinNumber, 2, nullptr, AZ::ConsoleFunctorFlags::Null, "Legacy Job system minimum number of worker threads to create after scaling the number of hw threads");
+
 namespace AZ
 {
     //=========================================================================
@@ -46,9 +54,10 @@ namespace AZ
         JobManagerThreadDesc threadDesc;
 
         int numberOfWorkerThreads = m_numberOfWorkerThreads;
-        if (numberOfWorkerThreads <= 0)
+        if (numberOfWorkerThreads <= 0) // spawn default number of threads
         {
-            numberOfWorkerThreads = AZ::GetMin(static_cast<unsigned int>(desc.m_workerThreads.capacity()), AZStd::thread::hardware_concurrency());
+            uint32_t scaledHardwareThreads = Threading::CalcNumWorkerThreads(cl_jobThreadsConcurrencyRatio, cl_jobThreadsMinNumber, cl_jobThreadsNumReserved);
+            numberOfWorkerThreads = AZ::GetMin(static_cast<unsigned int>(desc.m_workerThreads.capacity()), scaledHardwareThreads);
         #if (AZ_TRAIT_MAX_JOB_MANAGER_WORKER_THREADS)
             numberOfWorkerThreads = AZ::GetMin(numberOfWorkerThreads, AZ_TRAIT_MAX_JOB_MANAGER_WORKER_THREADS);
         #endif // (AZ_TRAIT_MAX_JOB_MANAGER_WORKER_THREADS)

Code/Framework/AzCore/AzCore/Jobs/JobManagerDesc.h

@@ -36,7 +36,7 @@ namespace AZ
         */
         int     m_stackSize;
 
-        JobManagerThreadDesc(int cpuId = -1, int priority = -100000, int stackSize = -1)
+        JobManagerThreadDesc(int cpuId = -1, int priority = 0, int stackSize = -1)
             : m_cpuId(cpuId)
             , m_priority(priority)
             , m_stackSize(stackSize)

Code/Framework/AzCore/AzCore/Task/TaskExecutor.cpp

@@ -308,11 +308,11 @@ namespace AZ
 
     void TaskExecutor::SetInstance(TaskExecutor* executor)
     {
-        if (!executor)
+        if (!executor) // allow unsetting the executor
         {
             s_executor.Reset();
         }
-        else if (!s_executor) // ignore any calls to set after the first (this happens in unit tests that create new system entities)
+        else if (!s_executor) // ignore any extra executors after the first (this happens during unit tests)
         {
             s_executor = AZ::Environment::CreateVariable<TaskExecutor*>(s_executorName, executor);
         }

Code/Framework/AzCore/AzCore/Task/TaskGraphSystemComponent.cpp

@@ -11,9 +11,15 @@
 #include <AzCore/Task/TaskGraphSystemComponent.h>
 #include <AzCore/Serialization/SerializeContext.h>
 #include <AzCore/Serialization/EditContext.h>
+#include <AzCore/Math/MathUtils.h>
+#include <AzCore/Threading/ThreadUtils.h>
 
 // Create a cvar as a central location for experimentation with switching from the Job system to TaskGraph system.
 AZ_CVAR(bool, cl_activateTaskGraph, false, nullptr, AZ::ConsoleFunctorFlags::Null, "Flag clients of TaskGraph to switch between jobs/taskgraph (Note does not disable task graph system)");
+AZ_CVAR(float, cl_taskGraphThreadsConcurrencyRatio, 1.0f, nullptr, AZ::ConsoleFunctorFlags::Null, "TaskGraph calculate the number of worker threads to spawn by scaling the number of hw threads, value is clamped between 0.0f and 1.0f");
+AZ_CVAR(uint32_t, cl_taskGraphThreadsNumReserved, 2, nullptr, AZ::ConsoleFunctorFlags::Null, "TaskGraph number of hardware threads that are reserved for O3DE system threads. Value is clamped between 0 and the number of logical cores in the system");
+AZ_CVAR(uint32_t, cl_taskGraphThreadsMinNumber, 2, nullptr, AZ::ConsoleFunctorFlags::Null, "TaskGraph minimum number of worker threads to create after scaling the number of hw threads");
+
 static constexpr uint32_t TaskExecutorServiceCrc = AZ_CRC_CE("TaskExecutorService");
 
 namespace AZ
@@ -24,8 +30,8 @@ namespace AZ
 
         if (Interface<TaskGraphActiveInterface>::Get() == nullptr)
         {
-            Interface<TaskGraphActiveInterface>::Register(this);
-            m_taskExecutor = aznew TaskExecutor();
+            Interface<TaskGraphActiveInterface>::Register(this); // small window that another thread can try to use taskgraph between this line and the set instance.
+            m_taskExecutor = aznew TaskExecutor(Threading::CalcNumWorkerThreads(cl_taskGraphThreadsConcurrencyRatio, cl_taskGraphThreadsMinNumber, cl_taskGraphThreadsNumReserved));
             TaskExecutor::SetInstance(m_taskExecutor);
         }
     }

Code/Framework/AzCore/AzCore/Threading/ThreadUtils.cpp

@@ -0,0 +1,25 @@
+/*
+ * 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/Threading/ThreadUtils.h>
+#include <AzCore/std/parallel/thread.h>
+#include <AzCore/Math/MathUtils.h>
+
+namespace AZ::Threading
+{
+    uint32_t CalcNumWorkerThreads(float workerThreadsRatio, uint32_t minNumWorkerThreads, uint32_t reservedNumThreads)
+    {
+        const uint32_t maxHardwareThreads = AZStd::thread::hardware_concurrency();
+        const uint32_t numReservedThreads = AZ::GetMin<uint32_t>(reservedNumThreads, maxHardwareThreads); // protect against num reserved being bigger than the number of hw threads
+        const uint32_t maxWorkerThreads = maxHardwareThreads - numReservedThreads;
+        const float requestedWorkerThreads = AZ::GetClamp<float>(workerThreadsRatio, 0.0f, 1.0f) * static_cast<float>(maxWorkerThreads);
+        const uint32_t requestedWorkerThreadsRounded = AZStd::lround(requestedWorkerThreads);
+        const uint32_t numWorkerThreads = AZ::GetMax<uint32_t>(minNumWorkerThreads, requestedWorkerThreadsRounded);
+        return numWorkerThreads;
+    }
+};

Code/Framework/AzCore/AzCore/Threading/ThreadUtils.h

@@ -0,0 +1,22 @@
+/*
+ * 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>
+
+namespace AZ::Threading
+{
+    //! Calculates the number of worker threads a system should use based on the number of hardware threads a device has. 
+    //!         result = max (minNumWorkerThreads, workerThreadsRatio * (num_hardware_threads - reservedNumThreads))
+    //! @param workerThreadsRatio scale applied to the calculated maximum number of threads available after reserved threads have been accounted for. Clamped between 0 and 1.
+    //! @param minNumWorkerThreads minimum value that will be returned. Value is unclamped and can be more than num_hardware_threads.
+    //! @param reservedNumThreads number of hardware threads to reserve for O3DE system threads. Value clamped to num_hardware_threads.
+    //! @return number of worker threads for the calling system to allocate
+    uint32_t CalcNumWorkerThreads(float workerThreadsRatio, uint32_t minNumWorkerThreads, uint32_t reservedNumThreads);
+};

Code/Framework/AzCore/AzCore/azcore_files.cmake

@@ -639,6 +639,8 @@ set(FILES
     Threading/ThreadSafeDeque.inl
     Threading/ThreadSafeObject.h
     Threading/ThreadSafeObject.inl
+    Threading/ThreadUtils.h
+    Threading/ThreadUtils.cpp
     Time/ITime.h
     Time/TimeSystemComponent.cpp
     Time/TimeSystemComponent.h

Code/Framework/AzCore/Platform/Common/UnixLike/AzCore/std/parallel/internal/thread_UnixLike.cpp

@@ -59,6 +59,10 @@ namespace AZStd
                 {
                     priority = desc->m_priority;
                 }
+                else
+                {
+                    priority = SCHED_OTHER;
+                }
                 if (desc->m_name)
                 {
                     name = desc->m_name;