diff options
Diffstat (limited to 'Source/Core/DiscIO/MultithreadedCompressor.h')
| -rw-r--r-- | Source/Core/DiscIO/MultithreadedCompressor.h | 225 |
1 files changed, 225 insertions, 0 deletions
diff --git a/Source/Core/DiscIO/MultithreadedCompressor.h b/Source/Core/DiscIO/MultithreadedCompressor.h new file mode 100644 index 0000000000..36b0163f66 --- /dev/null +++ b/Source/Core/DiscIO/MultithreadedCompressor.h @@ -0,0 +1,225 @@ +// Copyright 2020 Dolphin Emulator Project +// Licensed under GPLv2+ +// Refer to the license.txt file included. + +#pragma once + +#include <atomic> +#include <functional> +#include <memory> +#include <thread> +#include <utility> +#include <variant> +#include <vector> + +#include "Common/Assert.h" +#include "Common/Event.h" +#include "Common/Result.h" + +namespace DiscIO +{ +enum class ConversionResultCode +{ + Success, + Canceled, + ReadFailed, + WriteFailed, + InternalError, +}; + +template <typename T> +using ConversionResult = Common::Result<ConversionResultCode, T>; + +// This class starts a number of compression threads and one output thread. +// The set_up_compress_thread_state function is called at the start of each compression thread. +// When CompressAndWrite is called, the compress function will be called on one of the +// compression threads, and then the output function will be called on the output thread. +// The output thread handles data in the order that data was submitted using CompressAndWrite, +// but the compression threads are not guaranteed to handle data in a predictable order. +// Remember to check GetStatus regularly and cancel if it doesn't return Success, +// and call Shutdown when you want to ensure that everything finishes. +template <typename CompressThreadState, typename CompressParameters, typename OutputParameters> +class MultithreadedCompressor +{ +public: + MultithreadedCompressor( + std::function<ConversionResultCode(CompressThreadState*)> set_up_compress_thread_state, + std::function<ConversionResult<OutputParameters>(CompressThreadState*, CompressParameters)> + compress, + std::function<ConversionResultCode(OutputParameters)> output) + : m_set_up_compress_thread_state(std::move(set_up_compress_thread_state)), + m_compress(std::move(compress)), m_output(std::move(output)), + m_threads(std::max<unsigned int>(1, std::thread::hardware_concurrency())) + { + m_compress_threads = std::make_unique<CompressThread[]>(m_threads); + + for (size_t i = 0; i < m_threads; ++i) + { + m_compress_threads[i].thread = + std::thread(std::mem_fn(&MultithreadedCompressor::CompressThreadFunction), this, + &m_compress_threads[i]); + } + + m_output_thread = + std::thread(std::mem_fn(&MultithreadedCompressor::OutputThreadFunction), this); + } + + ~MultithreadedCompressor() + { + if (!m_shutting_down.load()) + Shutdown(); + } + + void CompressAndWrite(CompressParameters parameters) + { + if (GetStatus() != ConversionResultCode::Success) + return; + + CompressThread& compress_thread = m_compress_threads[m_current_index]; + + compress_thread.compress_ready_event.Wait(); + compress_thread.compress_parameters = std::move(parameters); + compress_thread.compress_event.Set(); + + ++m_current_index; + if (m_current_index >= m_threads) + m_current_index -= m_threads; + } + + void SetError(ConversionResultCode result) + { + ASSERT(result != ConversionResultCode::Success); + + // If we already have an error, don't overwrite it + ConversionResultCode expected = ConversionResultCode::Success; + m_result.compare_exchange_strong(expected, result); + } + + ConversionResultCode GetStatus() const { return m_result.load(); } + + void Shutdown() + { + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].compress_ready_event.Wait(); + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].compress_done_event.Wait(); + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].output_ready_event.Wait(); + + m_shutting_down.store(true); + + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].compress_event.Set(); + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].output_event.Set(); + + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].thread.join(); + + m_output_thread.join(); + } + +private: + struct CompressThread + { + std::thread thread; + + Common::Event compress_ready_event; + Common::Event compress_event; + Common::Event compress_done_event; + Common::Event output_ready_event; + Common::Event output_event; + + CompressParameters compress_parameters; + OutputParameters output_parameters; + }; + + void CompressThreadFunction(CompressThread* state) + { + CompressThreadState compress_thread_state; + + ConversionResultCode setup_result = m_set_up_compress_thread_state(&compress_thread_state); + if (setup_result != ConversionResultCode::Success) + SetError(setup_result); + + state->compress_ready_event.Set(); + state->compress_done_event.Set(); + + while (true) + { + state->compress_event.Wait(); + + if (m_shutting_down.load()) + return; + + CompressParameters parameters = std::move(state->compress_parameters); + + state->compress_done_event.Reset(); + state->compress_ready_event.Set(); + + ConversionResult<OutputParameters> result = + m_compress(&compress_thread_state, std::move(parameters)); + + if (result) + { + state->output_ready_event.Wait(); + state->output_parameters = std::move(*result); + state->output_event.Set(); + } + else + { + SetError(result.Error()); + } + + state->compress_done_event.Set(); + } + } + + void OutputThreadFunction() + { + for (size_t i = 0; i < m_threads; ++i) + m_compress_threads[i].output_ready_event.Set(); + + size_t index = 0; + + while (true) + { + CompressThread& compress_thread = m_compress_threads[index]; + + compress_thread.output_event.Wait(); + + if (m_shutting_down.load()) + return; + + OutputParameters parameters = std::move(compress_thread.output_parameters); + + compress_thread.output_ready_event.Set(); + + const ConversionResultCode result = m_output(std::move(parameters)); + + if (result != ConversionResultCode::Success) + SetError(result); + + ++index; + if (index >= m_threads) + index -= m_threads; + } + } + + std::function<ConversionResultCode(CompressThreadState*)> m_set_up_compress_thread_state; + std::function<ConversionResult<OutputParameters>(CompressThreadState*, CompressParameters)> + m_compress; + std::function<ConversionResultCode(OutputParameters)> m_output; + + // We can't use std::vector for this, because Common::Event is not movable + std::unique_ptr<CompressThread[]> m_compress_threads; + std::thread m_output_thread; + + const size_t m_threads; + size_t m_current_index = 0; + + std::atomic<ConversionResultCode> m_result = ConversionResultCode::Success; + std::atomic<bool> m_shutting_down = false; +}; + +} // namespace DiscIO |