From 8ccaa2622316dc4f6fc76f9dcf8f9cd920bba9ae Mon Sep 17 00:00:00 2001 From: Dynamitos Date: Wed, 15 Dec 2021 00:05:42 +0100 Subject: [PATCH] Somewhat fixed thread pool --- WRITEUP.md | 123 +++++++++- .../Graphics/RenderPass/LightCullingPass.cpp | 1 + .../Graphics/Vulkan/VulkanCommandBuffer.cpp | 2 + src/Engine/ThreadPool.cpp | 164 +++++++------ src/Engine/ThreadPool.h | 226 ++++++++++-------- src/Engine/Window/SceneView.cpp | 6 +- src/Engine/Window/Window.cpp | 6 +- src/Engine/main.cpp | 10 +- 8 files changed, 353 insertions(+), 185 deletions(-) diff --git a/WRITEUP.md b/WRITEUP.md index de3da7c..c4a01f4 100644 --- a/WRITEUP.md +++ b/WRITEUP.md @@ -1,9 +1,7 @@ # Seele Engine Seele is a Proof-of-Concept 3D render engine using Vulkan as it's primary graphics API, but other APIs could also be implemented. Due to this, a lot of terminology will be Vulkan-related, but it can be translated to other APIs. -The focus of the project is to make the renderer use as many of the system resources as efficiently as possible to maximize the framerate -while minimizing resource usage. -It does this by being designed to support multithreading for more recent APIs, like DX12 or Vulkan, while still providing an interface that could be implemented by an +It is designed to support multithreading for more recent APIs, like DX12 or Vulkan, while still providing an interface that could be implemented by an API that doesn't support it, like OpenGL. These APIs would suffer a significant performance loss, but would still work correctly. ## Custom containers and other utilities @@ -206,3 +204,122 @@ struct Placeholder: IMaterial { ### Renderpass interaction Each render pass has a "base file" containing the vertex and fragment stages. From that base file a variant is generated for each material that will be used with that render pass using `#define` macros to change the material import, and a `type_param` to update the `ParameterBlock` containing the material itself. Other renderpass dependent data like camera parameter or lights are stored in separate descriptor sets, and to avoid code duplication, they parts that are used by more than one render pass are implemented in separate slang files. But since in Vulkan pipeline layouts cannot have empty indices, the descriptor set indices sometimes need to change. This is done using the static `modifyRenderPassMacros` function in each renderpass to `#define` the correct set indices for each shared descriptor layout. + +## Job System + +The Job System is based on C++20 coroutines, running on a thread pool with 1 main thread and as many worker threads as the native CPU has. This is so that when the main thread needs to block for some thread pool work, the CPU can utilize all its resources. Synchronization is done using `Event`s, which are awaitable wrappers for atomic flags. + +### Creating Jobs + +In order to run a function as a Job, its return type must simply be either `Job` or `MainJob`. Also, any `Job` must contain either a `co_await` or `co_return` statement. As the names suggest, `MainJob`s will always run on the main thread, while `Job`s can run on any thread. Some graphics operations, especially regarding windowing, can only occur on the main thread, which is the reason for the separation. Due to the specification for [coroutines](https://en.cppreference.com/w/cpp/language/coroutines) being incredibly complicated and non-intuitive, a detailed description will be neglected for some code snippets. + +```cpp +Job doSomething() +{ + std::cout << "Doing something" << std::endl; + co_return; +} +Job doSomethingElse() +{ + std::cout << "Doing something else" << std::endl; + co_return; +} +Job doSomethingAfterwards() +{ + std::cout << "Doing something afterwards" << std::endl; + co_return; +} +Job startHere() +{ + Job job1 = doSomething(); + Job job2 = doSomethingElse(); + co_await job1; + co_await job2; + co_await doSomethingAfterwards(); +} +``` + +This example will execute the first two functions in an undefined order, while the last one will always be executed last. This is because the jobs are launched as soon as the functions are called, while the co_await operator waits for them to finish. Keeping the return value of every job launched is a bit inconvenient, so there is are some helper functions to help with that. `Job::all()` takes a variable amount of `Job`s which it 'combines' into one. This reduces the previous example into this: + +```cpp +Job doSomething() +{ + std::cout << "Doing something" << std::endl; + co_return; +} +Job doSomethingElse() +{ + std::cout << "Doing something else" << std::endl; + co_return; +} +Job doSomethingAfterwards() +{ + std::cout << "Doing something afterwards" << std::endl; + co_return; +} +Job startHere() +{ + co_await Job::all( + doSomething(), + doSomethingElse()); + co_await doSomethingAfterwards(); +} +``` + +Any Job can also be continued with `then()`, taking in another `Job` or anything that evaluates to a `Job` if called. This further simplifies the example: + +```cpp +Job doSomething() +{ + std::cout << "Doing something" << std::endl; + co_return; +} +Job doSomethingElse() +{ + std::cout << "Doing something else" << std::endl; + co_return; +} +Job doSomethingAfterwards() +{ + std::cout << "Doing something afterwards" << std::endl; + co_return; +} +Job startHere() +{ + co_await Job::all( + doSomething(), + doSomethingElse()) + .then(doSomethingAfterwards()); +} +``` + +### Events + +Aside from `Job`s, `Event`s can also be used to synchronize functions. They are constructed with an optional name for easier debugging, and can be `co_await`ed. For the waiting job to continue, `raise()` must be called, after which one of the waiting jobs should `reset()` the event to make it reusable. + +```cpp +Event updateFinished; +std::mutex dataLock; +uint64 data = 0; + +Job updateSomething() +{ + { + std::scoped_lock lock(dataLock); + data++; + } + updateFinished.raise(); + updateSomething(); // if we call the function again it behaves like a loop +} + +Job processUpdates() +{ + co_await updateFinished; + updateFinished.reset(); + { + std::scoped_lock lock(dataLock); + std::cout << data << std::endl; + } + processUpdates(); +} +``` diff --git a/src/Engine/Graphics/RenderPass/LightCullingPass.cpp b/src/Engine/Graphics/RenderPass/LightCullingPass.cpp index 3df6084..fdd0a48 100644 --- a/src/Engine/Graphics/RenderPass/LightCullingPass.cpp +++ b/src/Engine/Graphics/RenderPass/LightCullingPass.cpp @@ -75,6 +75,7 @@ MainJob LightCullingPass::beginFrame() lightEnvDescriptorSet->updateBuffer(2, pointLightBuffer); lightEnvDescriptorSet->updateBuffer(3, numPointLightBuffer); lightEnvDescriptorSet->writeChanges(); + std::cout << "Finished light culling beginFrame()" << std::endl; co_return; } diff --git a/src/Engine/Graphics/Vulkan/VulkanCommandBuffer.cpp b/src/Engine/Graphics/Vulkan/VulkanCommandBuffer.cpp index d6758bb..0f06be9 100644 --- a/src/Engine/Graphics/Vulkan/VulkanCommandBuffer.cpp +++ b/src/Engine/Graphics/Vulkan/VulkanCommandBuffer.cpp @@ -60,6 +60,7 @@ void CmdBuffer::end() void CmdBuffer::beginRenderPass(PRenderPass newRenderPass, PFramebuffer newFramebuffer) { + assert(state == State::InsideBegin); std::unique_lock lock(handleLock); renderPass = newRenderPass; framebuffer = newFramebuffer; @@ -87,6 +88,7 @@ void CmdBuffer::executeCommands(const Array& commands) assert(state == State::RenderPassActive); if(commands.size() == 0) { + std::cout << "No commands!" << std::endl; return; } std::unique_lock lock(handleLock); diff --git a/src/Engine/ThreadPool.cpp b/src/Engine/ThreadPool.cpp index 05ceb98..90f01f5 100644 --- a/src/Engine/ThreadPool.cpp +++ b/src/Engine/ThreadPool.cpp @@ -3,19 +3,20 @@ using namespace Seele; - Event::Event() - : flag(new std::atomic_bool()) -{} + : flag(new std::atomic_bool()) +{ +} Event::Event(nullptr_t) : flag(nullptr) -{} +{ +} -Event::Event(const std::string& name) - : name(name) - , flag(new std::atomic_bool()) -{} +Event::Event(const std::string &name) + : name(name), flag(new std::atomic_bool()) +{ +} void Event::raise() { @@ -23,7 +24,7 @@ void Event::raise() getGlobalThreadPool().notify(*this); } void Event::reset() -{ +{ flag->store(0); } @@ -36,7 +37,7 @@ ThreadPool::ThreadPool(uint32 threadCount) : workers(threadCount) { running.store(true); - for(uint32 i = 0; i < threadCount; ++i) + for (uint32 i = 0; i < threadCount; ++i) { workers[i] = std::thread(&ThreadPool::threadLoop, this, false); } @@ -45,48 +46,52 @@ ThreadPool::ThreadPool(uint32 threadCount) ThreadPool::~ThreadPool() { running.store(false); - for(auto& thread : workers) + for (auto &thread : workers) { thread.join(); } } -void ThreadPool::addJob(Job&& job) +void ThreadPool::enqueueWaiting(Event &event, Promise* job) { - std::unique_lock lock(jobQueueLock); - //std::cout << "Adding job " << job.id << std::endl; - jobQueue.add(std::move(job)); - jobQueueCV.notify_one(); + assert(!job->done()); + if(event == nullptr) + { + std::unique_lock lock(jobQueueLock); + jobQueue.add(job); + jobQueueCV.notify_one(); + } + else + { + std::unique_lock lock(waitingLock); + waitingJobs[event].add(job); + } } -void ThreadPool::addJob(MainJob&& job) +void ThreadPool::enqueueWaiting(Event &event, MainPromise* job) { - std::unique_lock lock(mainJobLock); - //std::cout << "Adding main job " << job.id << std::endl; - mainJobs.add(std::move(job)); - mainJobCV.notify_one(); + assert(!job->done()); + if(event == nullptr) + { + std::unique_lock lock(mainJobLock); + mainJobs.add(job); + mainJobCV.notify_one(); + } + else + { + std::unique_lock lock(waitingMainLock); + waitingMainJobs[event].add(job); + } } -void ThreadPool::enqueueWaiting(Event& event, Job&& job) -{ - //std::cout << job.id << " waiting for event " << event.name << std::endl; - std::unique_lock lock(waitingLock); - waitingJobs[event].add(std::move(job)); -} -void ThreadPool::enqueueWaiting(Event& event, MainJob&& job) -{ - //std::cout << job.id << " main waiting for event " << event.name << std::endl; - std::unique_lock lock(waitingMainLock); - waitingMainJobs[event].add(std::move(job)); -} -void ThreadPool::notify(Event& event) +void ThreadPool::notify(Event &event) { { std::unique_lock lock(jobQueueLock); std::unique_lock lock2(waitingLock); - List& jobs = waitingJobs[event]; - for(auto& job : jobs) + List &jobs = waitingJobs[event]; + for (auto &job : jobs) { //assert(job.id != -1ull); //std::cout << "Waking up job " << job.id << std::endl; - jobQueue.add(std::move(job)); + jobQueue.add(job); jobQueueCV.notify_one(); } jobs.clear(); @@ -94,67 +99,74 @@ void ThreadPool::notify(Event& event) { std::unique_lock lock(mainJobLock); std::unique_lock lock2(waitingMainLock); - List& jobs = waitingMainJobs[event]; - for(auto& job : jobs) + List &jobs = waitingMainJobs[event]; + for (auto &job : jobs) { //assert(job.id != -1ull); //std::cout << "Waking up main job " << job.id << std::endl; - mainJobs.add(std::move(job)); + mainJobs.add(job); mainJobCV.notify_one(); } jobs.clear(); } } -void ThreadPool::tryMainJob() +void ThreadPool::threadLoop(const bool mainThread) { - MainJob job; + while (running.load()) { - std::unique_lock lock(mainJobLock); - if(!mainJobs.empty()) + if (mainThread) { - job = std::move(mainJobs.retrieve()); + MainPromise* job; + { + std::unique_lock lock(mainJobLock); + if(mainJobs.empty()) + { + mainJobCV.wait(lock); + } + if (!mainJobs.empty()) + { + job = mainJobs.retrieve(); + } + else + { + continue; + } + } + job->resume(); + if (job->done()) + { + job->raise(); + } } else { - return; - } - } - job.resume(); -} -void ThreadPool::threadLoop(const bool mainThread) -{ - while(running.load()) - { - if(mainThread) - { - tryMainJob(); - } - Job job; - { - std::unique_lock lock(jobQueueLock); - if(jobQueue.empty()) + Promise* job; { - jobQueueCV.wait(lock); + std::unique_lock lock(jobQueueLock); + if (jobQueue.empty()) + { + jobQueueCV.wait(lock); + } + if (!jobQueue.empty()) + { + job = jobQueue.retrieve(); + } + else + { + continue; + } } - if(!jobQueue.empty()) + //std::cout << "Starting job " << job.id << std::endl; + job->resume(); + if (job->done()) { - job = std::move(jobQueue.retrieve()); + job->raise(); } - else - { - continue; - } - } - //std::cout << "Starting job " << job.id << std::endl; - job.resume(); - if(job.done()) - { - job.raise(); } } } -ThreadPool& Seele::getGlobalThreadPool() +ThreadPool &Seele::getGlobalThreadPool() { static ThreadPool threadPool; return threadPool; diff --git a/src/Engine/ThreadPool.h b/src/Engine/ThreadPool.h index 1fcaaaa..b9244e3 100644 --- a/src/Engine/ThreadPool.h +++ b/src/Engine/ThreadPool.h @@ -8,29 +8,8 @@ namespace Seele { extern class ThreadPool& getGlobalThreadPool(); - template -struct JobBase; -struct Event; -static std::atomic_uint64_t globalCounter; -template -struct JobPromiseBase -{ - JobBase get_return_object() noexcept; - - inline auto initial_suspend() noexcept; - inline auto final_suspend() const noexcept; - - void return_void() noexcept {} - void unhandled_exception() noexcept { - std::cerr << "Unhandled exception" << std::endl; - exit(1); - }; - std::coroutine_handle<> continuation = std::noop_coroutine(); - uint64 id; - Event finishedEvent; -}; - +struct JobPromiseBase; struct Event { public: @@ -41,6 +20,10 @@ public: { return flag <=> other.flag; } + bool operator==(const Event& other) const + { + return flag == other.flag; + } Event operator co_await() { return *this; @@ -62,65 +45,43 @@ private: friend class ThreadPool; }; -template -struct JobBase +static std::atomic_uint64_t globalCounter; +template +struct JobBase; +template +struct JobPromiseBase { -public: - using promise_type = JobPromiseBase; - - explicit JobBase() - : id(-1) + JobPromiseBase() { + handle = std::coroutine_handle>::from_promise(*this); + id = globalCounter++; + finishedEvent = Event(std::format("Job {}", id)); } - explicit JobBase(std::coroutine_handle handle, uint64 id, Event finishedEvent) - : handle(handle) - , event(std::move(finishedEvent)) - , id(id) - { - } - JobBase(const JobBase & rhs) = delete; - JobBase(JobBase&& rhs) - : handle(std::move(rhs.handle)) - , event(rhs.event) - , id(std::move(rhs.id)) - { - rhs.id = -1; - rhs.handle = nullptr; - } - ~JobBase() - { - if(handle && handle.done()) - { - std::cout << "Destroy job " << handle.address() << std::endl; - handle.destroy(); - } - } - JobBase& operator=(const JobBase& rhs) = delete; - JobBase& operator=(JobBase&& rhs) - { - if(this != &rhs) - { - handle = std::move(rhs.handle); - event = std::move(rhs.event); - id = std::move(rhs.id); - rhs.id = -1; - rhs.handle = nullptr; - } - return *this; - } + JobBase get_return_object() noexcept; + + inline auto initial_suspend() noexcept; + inline auto final_suspend() const noexcept; + + void return_void() noexcept {} + void unhandled_exception() noexcept { + std::cerr << "Unhandled exception" << std::endl; + exit(1); + }; + void resume() { - handle.resume(); + std::unique_lock lock(promiseLock); + if(handle && !handle.done()) + { + handle.resume(); + } } - template - inline JobBase then(Callable callable) + void setContinuation(JobPromiseBase* cont) { - return then(callable()); - } - JobBase then(JobBase continuation) - { - handle.promise().continuation = continuation.handle; - return continuation; + std::unique_lock lock(promiseLock); + std::unique_lock lock2(cont->promiseLock); + continuation = cont->handle; + cont->precondition = finishedEvent; } bool done() { @@ -128,15 +89,87 @@ public: } void raise() { - event.raise(); + std::unique_lock lock(promiseLock); + finishedEvent.raise(); } void reset() { - event.reset(); + std::unique_lock lock(promiseLock); + finishedEvent.reset(); + } + void enqueue(Event& event) + { + getGlobalThreadPool().enqueueWaiting(event, this); + } + void schedule() + { + std::unique_lock lock(promiseLock); + if(!handle || handle.done()) + { + return; + } + getGlobalThreadPool().enqueueWaiting(precondition, this); + } + + std::mutex promiseLock; + std::coroutine_handle handle; + std::coroutine_handle<> continuation = std::noop_coroutine(); + uint64 id; + Event finishedEvent; + Event precondition = nullptr; +}; + +template +struct JobBase +{ +public: + using promise_type = JobPromiseBase; + + explicit JobBase() + : promise(nullptr) + { + } + explicit JobBase(JobPromiseBase* promise) + : promise(promise) + { + } + ~JobBase() + { + if(promise) + { + promise->schedule(); + } + } + void resume() + { + promise->resume(); + } + template + inline JobBase then(Callable callable) + { + return then(callable()); + } + JobBase then(JobBase&& continuation) + { + promise->setContinuation(continuation.promise); + return continuation; + } + bool done() + { + return promise->done(); + } + void raise() + { + promise->raise(); + } + void reset() + { + promise->reset(); } Event operator co_await() { - return event; + promise->resume(); + return promise->finishedEvent; } static JobBase all() = delete; template @@ -166,58 +199,54 @@ public: } } private: - std::coroutine_handle handle; - Event event; - uint64 id; + JobPromiseBase* promise; }; using MainJob = JobBase; using Job = JobBase; +using MainPromise = JobPromiseBase; +using Promise = JobPromiseBase; class ThreadPool { public: ThreadPool(uint32 threadCount = std::thread::hardware_concurrency()); virtual ~ThreadPool(); - void addJob(Job&& job); - void addJob(MainJob&& job); - void enqueueWaiting(Event& event, Job&& job); - void enqueueWaiting(Event& event, MainJob&& job); + // Adds a job to the waiting queue for event, or directly to the job queue if event is nullptr + void enqueueWaiting(Event& event, Promise* job); + // Adds a job to the waiting queue for event, or directly to the job queue if event is nullptr + void enqueueWaiting(Event& event, MainPromise* job); void notify(Event& event); void threadLoop(const bool isMainThread); private: std::atomic_bool running; std::vector workers; - List mainJobs; + List mainJobs; std::mutex mainJobLock; std::condition_variable mainJobCV; - List jobQueue; + List jobQueue; std::mutex jobQueueLock; std::condition_variable jobQueueCV; - Map> waitingJobs; - std::mutex waitingLock; - - Map> waitingMainJobs; + Map> waitingMainJobs; std::mutex waitingMainLock; - - void tryMainJob(); -}; + Map> waitingJobs; + std::mutex waitingLock; +}; template inline JobBase JobPromiseBase::get_return_object() noexcept { - id = globalCounter++; - return JobBase(std::coroutine_handle>::from_promise(*this), id, finishedEvent); + return JobBase(this); } template inline auto JobPromiseBase::initial_suspend() noexcept { - struct JobAwaitable + /*struct JobAwaitable { constexpr bool await_ready() { return false; } constexpr void await_suspend(std::coroutine_handle> h) @@ -228,7 +257,8 @@ inline auto JobPromiseBase::initial_suspend() noexcept uint64 id; Event& event; }; - return JobAwaitable{id, finishedEvent}; + return JobAwaitable{id, finishedEvent};*/ + return std::suspend_always{}; } template inline auto JobPromiseBase::final_suspend() const noexcept @@ -248,7 +278,7 @@ inline auto JobPromiseBase::final_suspend() const noexcept template inline constexpr void Event::await_suspend(std::coroutine_handle> h) { - getGlobalThreadPool().enqueueWaiting(*this, std::move(JobBase(h, h.promise().id, *this))); + h.promise().enqueue(*this); } } // namespace Seele diff --git a/src/Engine/Window/SceneView.cpp b/src/Engine/Window/SceneView.cpp index 145547f..bc096ea 100644 --- a/src/Engine/Window/SceneView.cpp +++ b/src/Engine/Window/SceneView.cpp @@ -93,11 +93,11 @@ MainJob SceneView::render() .then(depthPrepass.endFrame()) .then(lightCullingPass.endFrame()) .then(basePass.endFrame()) - .then([=]() -> MainJob + .then([&](Event& event) -> MainJob { - renderFinishedEvent.raise(); + event.raise(); co_return; - }); + }(renderFinishedEvent)); } void SceneView::keyCallback(KeyCode code, InputAction action, KeyModifier) diff --git a/src/Engine/Window/Window.cpp b/src/Engine/Window/Window.cpp index 815e775..6672715 100644 --- a/src/Engine/Window/Window.cpp +++ b/src/Engine/Window/Window.cpp @@ -43,7 +43,11 @@ MainJob Window::render() windowView->view->resetRender(); } gfxHandle->endFrame(); - render(); + if(owner->isActive()) + { + render(); + } + co_return; } Gfx::PWindow Window::getGfxHandle() diff --git a/src/Engine/main.cpp b/src/Engine/main.cpp index 1865485..ad352da 100644 --- a/src/Engine/main.cpp +++ b/src/Engine/main.cpp @@ -19,7 +19,7 @@ int main() mainWindowInfo.pixelFormat = Gfx::SE_FORMAT_B8G8R8A8_UNORM; auto window = windowManager->addWindow(mainWindowInfo); ViewportCreateInfo sceneViewInfo; - sceneViewInfo.sizeX = 640; + sceneViewInfo.sizeX = 1280; sceneViewInfo.sizeY = 720; sceneViewInfo.offsetX = 0; sceneViewInfo.offsetY = 0; @@ -31,10 +31,12 @@ int main() inspectorViewInfo.sizeY = 720; inspectorViewInfo.offsetX = 640; inspectorViewInfo.offsetY = 0; - PInspectorView inspectorView = new InspectorView(windowManager->getGraphics(), window, inspectorViewInfo); - window->addView(inspectorView); + //PInspectorView inspectorView = new InspectorView(windowManager->getGraphics(), window, inspectorViewInfo); + //window->addView(inspectorView); sceneView->setFocused(); - window->render(); + { + window->render(); + } getGlobalThreadPool().threadLoop(true); return 0; } \ No newline at end of file