From 28bcab7146821165656aa0a8a12ffe4115029d5c Mon Sep 17 00:00:00 2001 From: Turanszki Janos Date: Mon, 5 Nov 2018 22:33:55 +0000 Subject: [PATCH] updated job system dispatch() --- WickedEngine/wiInitializer.cpp | 2 +- WickedEngine/wiJobSystem.cpp | 63 +++++---- WickedEngine/wiJobSystem.h | 5 +- WickedEngine/wiPhysicsEngine_Bullet.cpp | 4 +- WickedEngine/wiSceneSystem.cpp | 172 ++++++------------------ WickedEngine/wiVersion.cpp | 2 +- 6 files changed, 88 insertions(+), 160 deletions(-) diff --git a/WickedEngine/wiInitializer.cpp b/WickedEngine/wiInitializer.cpp index 401c3954b..c7ce5019e 100644 --- a/WickedEngine/wiInitializer.cpp +++ b/WickedEngine/wiInitializer.cpp @@ -23,10 +23,10 @@ namespace wiInitializer wiJobSystem::Initialize(); - wiJobSystem::Execute([] { wiSoundEffect::Initialize(); wiMusic::Initialize(); }); // this loads slowest... wiJobSystem::Execute([] { wiRenderer::Initialize(); wiWidget::LoadShaders(); }); wiJobSystem::Execute([] { wiFont::Initialize(); }); wiJobSystem::Execute([] { wiImage::Initialize(); }); + wiJobSystem::Execute([] { wiSoundEffect::Initialize(); wiMusic::Initialize(); }); wiJobSystem::Execute([] { wiTextureHelper::Initialize(); }); wiJobSystem::Execute([] { wiSceneSystem::wiHairParticle::Initialize(); }); wiJobSystem::Execute([] { wiSceneSystem::wiEmittedParticle::Initialize(); }); diff --git a/WickedEngine/wiJobSystem.cpp b/WickedEngine/wiJobSystem.cpp index abbe6afcc..6d018de01 100644 --- a/WickedEngine/wiJobSystem.cpp +++ b/WickedEngine/wiJobSystem.cpp @@ -10,30 +10,20 @@ namespace wiJobSystem { - struct Job - { - uint32_t jobIndex = 0; - std::function func; - - Job() {} - Job(const std::function& func) - { - this->func = std::move([func](uint32_t jobIndex) { - func(); - }); - } - Job(uint32_t jobIndex, const std::function& func) : jobIndex(jobIndex), func(func) {} - }; + typedef std::function Job; std::deque jobPool; wiSpinLock jobLock; std::condition_variable wakeCondition; std::mutex wakeMutex; - std::atomic remainingJobs = 0; + std::atomic remainingJobs; + std::atomic_bool waitBarrier; uint32_t numThreads = 0; - std::atomic_bool waitBarrier = false; void Initialize() { + waitBarrier.store(false); + remainingJobs.store(0); + // Retrieve the number of hardware threads in this system: auto numCores = std::thread::hardware_concurrency(); @@ -42,7 +32,7 @@ namespace wiJobSystem for (unsigned int threadID = 0; threadID < numThreads; ++threadID) { - std::thread([threadID] { + std::thread([] { while (true) { @@ -62,7 +52,7 @@ namespace wiJobSystem if (working) { - job.func(job.jobIndex); // execute job + job(); // execute job remainingJobs.fetch_sub(1); } else @@ -86,33 +76,56 @@ namespace wiJobSystem return numThreads; } - void Execute(const std::function& func) + void Execute(const std::function& job) { while (waitBarrier.load() == true) { std::this_thread::yield(); } // can't add jobs while Wait() is in progress + // This is important, and acts as a barrier for Wait(): remainingJobs.fetch_add(1); jobLock.lock(); - jobPool.push_back(Job(std::move(func))); + jobPool.push_back(job); jobLock.unlock(); wakeCondition.notify_one(); // only wake a single thread } - void Dispatch(uint32_t jobCount, const std::function& func) + void Dispatch(uint32_t jobCount, uint32_t groupSize, const std::function& job) { - if (jobCount == 0) + if (jobCount == 0 || groupSize == 0) { return; } while (waitBarrier.load() == true) { std::this_thread::yield(); } // can't add jobs while Wait() is in progress - remainingJobs.fetch_add(jobCount); + // Calculate the amount of job groups to dispatch: + const uint32_t jobGroupCount = (uint32_t)ceilf((float)jobCount / (float)groupSize); + + // This is important, and acts as a barrier for Wait(): + remainingJobs.fetch_add(jobGroupCount); jobLock.lock(); - for (uint32_t i = 0; i < jobCount; ++i) + for (uint32_t i = 0; i < jobGroupCount; ++i) { - jobPool.push_back(Job(i, std::move(func))); + // Calculate the current group's offset into the jobs: + const uint32_t jobOffset = i * groupSize; + + // For each group, generate a real job: + jobPool.push_back([jobCount, groupSize, job, jobOffset]() { + + // Inside the group, loop through all sub-jobs and propagate sub-job index: + for (uint32_t j = 0; j < groupSize; ++j) + { + const uint32_t jobIndex = jobOffset + j; + if (jobIndex >= jobCount) + { + // The amount of sub-jobs can be larger than the jobCount, so if that happens, don't issue the sub-job: + break; + } + job(jobIndex); + } + }); + } jobLock.unlock(); diff --git a/WickedEngine/wiJobSystem.h b/WickedEngine/wiJobSystem.h index 016afc86f..4e474f0e1 100644 --- a/WickedEngine/wiJobSystem.h +++ b/WickedEngine/wiJobSystem.h @@ -9,12 +9,13 @@ namespace wiJobSystem uint32_t GetThreadCount(); // Add a job to execute asynchronously. Any idle thread will execute this job. - void Execute(const std::function& func); + void Execute(const std::function& job); // Divide a job onto multiple jobs and execute in parallel. // jobCount : how many jobs to generate for this task. + // groupSize : how many jobs to execute per thread. Jobs inside a group execute serially. It might be worth to increase for small jobs // func : receives the job invocation index (0, 1, ... jobCount) as lambda argument - void Dispatch(uint32_t jobCount, const std::function& func); + void Dispatch(uint32_t jobCount, uint32_t groupSize, const std::function& job); // Check if any threads are working currently or not bool IsBusy(); diff --git a/WickedEngine/wiPhysicsEngine_Bullet.cpp b/WickedEngine/wiPhysicsEngine_Bullet.cpp index 4f899bc3e..ced057ab4 100644 --- a/WickedEngine/wiPhysicsEngine_Bullet.cpp +++ b/WickedEngine/wiPhysicsEngine_Bullet.cpp @@ -316,7 +316,7 @@ namespace wiPhysicsEngine btVector3 wind = btVector3(weather.windDirection.x, weather.windDirection.y, weather.windDirection.z); // System will register rigidbodies to objects, and update physics engine state for kinematics: - wiJobSystem::Dispatch((uint32_t)rigidbodies.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)rigidbodies.GetCount(), 8, [&](uint32_t jobIndex) { RigidBodyPhysicsComponent& physicscomponent = rigidbodies[jobIndex]; Entity entity = rigidbodies.GetEntity(jobIndex); @@ -366,7 +366,7 @@ namespace wiPhysicsEngine }); // System will register softbodies to meshes and update physics engine state: - wiJobSystem::Dispatch((uint32_t)softbodies.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)softbodies.GetCount(), 1, [&](uint32_t jobIndex) { SoftBodyPhysicsComponent& physicscomponent = softbodies[jobIndex]; Entity entity = softbodies.GetEntity(jobIndex); diff --git a/WickedEngine/wiSceneSystem.cpp b/WickedEngine/wiSceneSystem.cpp index dc4b10f8a..0723ca61e 100644 --- a/WickedEngine/wiSceneSystem.cpp +++ b/WickedEngine/wiSceneSystem.cpp @@ -6,6 +6,7 @@ #include "wiArchive.h" #include "wiRenderer.h" #include "wiJobSystem.h" +#include "wiSpinlock.h" #include #include @@ -1333,39 +1334,20 @@ namespace wiSceneSystem } - + const uint32_t small_subtask_groupsize = 32; void RunPreviousFrameTransformUpdateSystem( const ComponentManager& transforms, ComponentManager& prev_transforms ) { - //for (size_t i = 0; i < prev_transforms.GetCount(); ++i) - //{ - // PreviousFrameTransformComponent& prev_transform = prev_transforms[i]; - // Entity entity = prev_transforms.GetEntity(i); - // const TransformComponent& transform = *transforms.GetComponent(entity); + wiJobSystem::Dispatch((uint32_t)prev_transforms.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { - // prev_transform.world_prev = transform.world; - //} + PreviousFrameTransformComponent& prev_transform = prev_transforms[jobIndex]; + Entity entity = prev_transforms.GetEntity(jobIndex); + const TransformComponent& transform = *transforms.GetComponent(entity); - const uint32_t itemsPerThread = 8; - const uint32_t jobCount = (uint32_t)ceilf((float)prev_transforms.GetCount() / (float)itemsPerThread); - - wiJobSystem::Dispatch(jobCount, [&](uint32_t jobIndex) { - for (uint32_t i = 0; i < itemsPerThread; ++i) - { - const uint32_t itemIndex = jobIndex * itemsPerThread + i; - if (itemIndex >= prev_transforms.GetCount()) - { - break; - } - PreviousFrameTransformComponent& prev_transform = prev_transforms[itemIndex]; - Entity entity = prev_transforms.GetEntity(itemIndex); - const TransformComponent& transform = *transforms.GetComponent(entity); - - prev_transform.world_prev = transform.world; - } + prev_transform.world_prev = transform.world; }); } void RunAnimationUpdateSystem( @@ -1493,26 +1475,10 @@ namespace wiSceneSystem } void RunTransformUpdateSystem(ComponentManager& transforms) { - //for (size_t i = 0; i < transforms.GetCount(); ++i) - //{ - // TransformComponent& transform = transforms[i]; - // transform.UpdateTransform(); - //} + wiJobSystem::Dispatch((uint32_t)transforms.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { - const uint32_t itemsPerThread = 8; - const uint32_t jobCount = (uint32_t)ceilf((float)transforms.GetCount() / (float)itemsPerThread); - - wiJobSystem::Dispatch(jobCount, [&](uint32_t jobIndex) { - for (uint32_t i = 0; i < itemsPerThread; ++i) - { - const uint32_t itemIndex = jobIndex * itemsPerThread + i; - if (itemIndex >= transforms.GetCount()) - { - break; - } - TransformComponent& transform = transforms[itemIndex]; - transform.UpdateTransform(); - } + TransformComponent& transform = transforms[jobIndex]; + transform.UpdateTransform(); }); } void RunHierarchyUpdateSystem( @@ -1550,32 +1516,7 @@ namespace wiSceneSystem ComponentManager& armatures ) { - //for (size_t i = 0; i < armatures.GetCount(); ++i) - //{ - // ArmatureComponent& armature = armatures[i]; - - // if (armature.boneData.size() != armature.boneCollection.size()) - // { - // armature.boneData.resize(armature.boneCollection.size()); - // } - - // XMMATRIX R = XMLoadFloat4x4(&armature.remapMatrix); - - // int boneIndex = 0; - // for (Entity boneEntity : armature.boneCollection) - // { - // const TransformComponent& bone = *transforms.GetComponent(boneEntity); - - // XMMATRIX B = XMLoadFloat4x4(&armature.inverseBindMatrices[boneIndex]); - // XMMATRIX W = XMLoadFloat4x4(&bone.world); - // XMMATRIX M = B * W * R; - - // armature.boneData[boneIndex++].Store(M); - // } - - //} - - wiJobSystem::Dispatch((uint32_t)armatures.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)armatures.GetCount(), 1, [&](uint32_t jobIndex) { ArmatureComponent& armature = armatures[jobIndex]; @@ -1602,62 +1543,30 @@ namespace wiSceneSystem } void RunMaterialUpdateSystem(ComponentManager& materials, float dt) { - //for (size_t i = 0; i < materials.GetCount(); ++i) - //{ - // MaterialComponent& material = materials[i]; + wiJobSystem::Dispatch((uint32_t)materials.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { - // material.texAnimSleep -= dt * material.texAnimFrameRate; - // if (material.texAnimSleep <= 0) - // { - // material.texMulAdd.z = fmodf(material.texMulAdd.z + material.texAnimDirection.x, 1); - // material.texMulAdd.w = fmodf(material.texMulAdd.w + material.texAnimDirection.y, 1); - // material.texAnimSleep = 1.0f; + MaterialComponent& material = materials[jobIndex]; - // material.SetDirty(); // will trigger constant buffer update! - // } - - // material.engineStencilRef = STENCILREF_DEFAULT; - // if (material.subsurfaceScattering > 0) - // { - // material.engineStencilRef = STENCILREF_SKIN; - // } - - //} - - const uint32_t itemsPerThread = 64; - const uint32_t jobCount = (uint32_t)ceilf((float)materials.GetCount() / (float)itemsPerThread); - - wiJobSystem::Dispatch(jobCount, [&](uint32_t jobIndex) { - for (uint32_t i = 0; i < itemsPerThread; ++i) + material.texAnimSleep -= dt * material.texAnimFrameRate; + if (material.texAnimSleep <= 0) { - const uint32_t itemIndex = jobIndex * itemsPerThread + i; - if (itemIndex >= materials.GetCount()) - { - break; - } - MaterialComponent& material = materials[itemIndex]; + material.texMulAdd.z = fmodf(material.texMulAdd.z + material.texAnimDirection.x, 1); + material.texMulAdd.w = fmodf(material.texMulAdd.w + material.texAnimDirection.y, 1); + material.texAnimSleep = 1.0f; - material.texAnimSleep -= dt * material.texAnimFrameRate; - if (material.texAnimSleep <= 0) - { - material.texMulAdd.z = fmodf(material.texMulAdd.z + material.texAnimDirection.x, 1); - material.texMulAdd.w = fmodf(material.texMulAdd.w + material.texAnimDirection.y, 1); - material.texAnimSleep = 1.0f; + material.SetDirty(); // will trigger constant buffer update! + } - material.SetDirty(); // will trigger constant buffer update! - } - - material.engineStencilRef = STENCILREF_DEFAULT; - if (material.subsurfaceScattering > 0) - { - material.engineStencilRef = STENCILREF_SKIN; - } + material.engineStencilRef = STENCILREF_DEFAULT; + if (material.subsurfaceScattering > 0) + { + material.engineStencilRef = STENCILREF_SKIN; } }); } void RunImpostorUpdateSystem(ComponentManager& impostors) { - wiJobSystem::Dispatch((uint32_t)impostors.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)impostors.GetCount(), 1, [&](uint32_t jobIndex) { ImpostorComponent& impostor = impostors[jobIndex]; impostor.aabb = AABB(); @@ -1680,11 +1589,12 @@ namespace wiSceneSystem assert(objects.GetCount() == aabb_objects.GetCount()); sceneBounds = AABB(); + static wiSpinLock lock; // contention for sceneBounds and waterPlane! - for (size_t i = 0; i < objects.GetCount(); ++i) - { - ObjectComponent& object = objects[i]; - AABB& aabb = aabb_objects[i]; + wiJobSystem::Dispatch((uint32_t)objects.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { + + ObjectComponent& object = objects[jobIndex]; + AABB& aabb = aabb_objects[jobIndex]; aabb = AABB(); object.rendertypeMask = 0; @@ -1695,7 +1605,7 @@ namespace wiSceneSystem if (object.meshID != INVALID_ENTITY) { - Entity entity = objects.GetEntity(i); + Entity entity = objects.GetEntity(jobIndex); const MeshComponent* mesh = meshes.GetComponent(object.meshID); // These will only be valid for a single frame: @@ -1747,7 +1657,9 @@ namespace wiSceneSystem XMVECTOR N = XMVectorSet(0, 1, 0, 0); N = XMVector3TransformNormal(N, XMLoadFloat4x4(&transform.world)); XMVECTOR _refPlane = XMPlaneFromPointNormal(P, N); + lock.lock(); XMStoreFloat4(&waterPlane, _refPlane); + lock.unlock(); } object.SetCastShadow(material->IsCastingShadow()); @@ -1784,17 +1696,19 @@ namespace wiSceneSystem } + lock.lock(); sceneBounds = AABB::Merge(sceneBounds, aabb); + lock.unlock(); } } - } + }); } void RunCameraUpdateSystem( const ComponentManager& transforms, ComponentManager& cameras ) { - wiJobSystem::Dispatch((uint32_t)cameras.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)cameras.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { CameraComponent& camera = cameras[jobIndex]; Entity entity = cameras.GetEntity(jobIndex); @@ -1815,7 +1729,7 @@ namespace wiSceneSystem { assert(decals.GetCount() == aabb_decals.GetCount()); - wiJobSystem::Dispatch((uint32_t)decals.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)decals.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { DecalComponent& decal = decals[jobIndex]; Entity entity = decals.GetEntity(jobIndex); @@ -1853,7 +1767,7 @@ namespace wiSceneSystem { assert(probes.GetCount() == aabb_probes.GetCount()); - wiJobSystem::Dispatch((uint32_t)probes.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)probes.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { EnvironmentProbeComponent& probe = probes[jobIndex]; Entity entity = probes.GetEntity(jobIndex); @@ -1880,7 +1794,7 @@ namespace wiSceneSystem ComponentManager& forces ) { - wiJobSystem::Dispatch((uint32_t)forces.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)forces.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { ForceFieldComponent& force = forces[jobIndex]; Entity entity = forces.GetEntity(jobIndex); @@ -1897,7 +1811,7 @@ namespace wiSceneSystem { assert(lights.GetCount() == aabb_lights.GetCount()); - wiJobSystem::Dispatch((uint32_t)lights.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)lights.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { LightComponent& light = lights[jobIndex]; Entity entity = lights.GetEntity(jobIndex); @@ -1944,13 +1858,13 @@ namespace wiSceneSystem float dt ) { - wiJobSystem::Dispatch((uint32_t)emitters.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)emitters.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { wiEmittedParticle& emitter = emitters[jobIndex]; emitter.Update(dt); }); - wiJobSystem::Dispatch((uint32_t)hairs.GetCount(), [&](uint32_t jobIndex) { + wiJobSystem::Dispatch((uint32_t)hairs.GetCount(), small_subtask_groupsize, [&](uint32_t jobIndex) { wiHairParticle& hair = hairs[jobIndex]; Entity entity = hairs.GetEntity(jobIndex); diff --git a/WickedEngine/wiVersion.cpp b/WickedEngine/wiVersion.cpp index b560b2755..b97e5b3cf 100644 --- a/WickedEngine/wiVersion.cpp +++ b/WickedEngine/wiVersion.cpp @@ -9,7 +9,7 @@ namespace wiVersion // minor features, major updates const int minor = 22; // minor bug fixes, alterations, refactors, updates - const int revision = 4; + const int revision = 5; long GetVersion()