From 8ee4105b399e343f5c39233681f5465773534559 Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Fri, 18 Sep 2020 17:12:12 +0200 Subject: [PATCH 1/8] Deep change of the background loadingThe goal was to reduce the number of background threads in big sessions (multiple instances) and avoid loading the same file alot on repeated notes (e.g. drum tracks).- All instances use a common thread pool- Each instance has a dispatching job that starts background loader, and a garbage job that clears data that has been unused for some time- The "FilePromise" object has disappeared. Upon request, Voices get a very thin reference-counting handler to the file data. When the handler is released the reference count is decreased and a "last used" timestamp is added to the file.- The background file loading job checks i) if the file is already loaded, or ii) if some other thread is already loading it. If any case is validated, it exits assuming the background loading is already happening.- The garbage job has to be triggered regularly by e.g. the synth. In the RT thread the triggerGarbageCollection() method will check the list of previously loaded file Ids to see if any has not been used for a while. If so, the memory is set to be discarded by a background thread, otherwise it'll wait for the next ping.Only tested on Linux for now. --- src/external/threadpool/ThreadPool.h | 100 ++++++++ src/sfizz/Config.h | 1 + src/sfizz/FilePool.cpp | 371 +++++++++++++++------------ src/sfizz/FilePool.h | 192 +++++++++----- src/sfizz/Synth.cpp | 10 +- src/sfizz/Synth.h | 2 + src/sfizz/Voice.cpp | 27 +- src/sfizz/Voice.h | 2 +- src/sfizz/Wavetables.cpp | 4 +- 9 files changed, 450 insertions(+), 259 deletions(-) create mode 100644 src/external/threadpool/ThreadPool.h diff --git a/src/external/threadpool/ThreadPool.h b/src/external/threadpool/ThreadPool.h new file mode 100644 index 00000000..36169849 --- /dev/null +++ b/src/external/threadpool/ThreadPool.h @@ -0,0 +1,100 @@ +// SPDX-License-Identifier: zlib + +#ifndef THREAD_POOL_H +#define THREAD_POOL_H + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +class ThreadPool { +public: + ThreadPool(size_t); + template + auto enqueue(F&& f, Args&&... args) + -> std::future::type>; + ~ThreadPool(); +private: + // need to keep track of threads so we can join them + std::vector< std::thread > workers; + // the task queue + std::queue< std::function > tasks; + + // synchronization + std::mutex queue_mutex; + std::condition_variable condition; + bool stop; +}; + +// the constructor just launches some amount of workers +inline ThreadPool::ThreadPool(size_t threads) + : stop(false) +{ + for(size_t i = 0;i task; + + { + std::unique_lock lock(this->queue_mutex); + this->condition.wait(lock, + [this]{ return this->stop || !this->tasks.empty(); }); + if(this->stop && this->tasks.empty()) + return; + task = std::move(this->tasks.front()); + this->tasks.pop(); + } + + task(); + } + } + ); +} + +// add new work item to the pool +template +auto ThreadPool::enqueue(F&& f, Args&&... args) + -> std::future::type> +{ + using return_type = typename std::result_of::type; + + auto task = std::make_shared< std::packaged_task >( + std::bind(std::forward(f), std::forward(args)...) + ); + + std::future res = task->get_future(); + { + std::unique_lock lock(queue_mutex); + + // don't allow enqueueing after stopping the pool + if(stop) + throw std::runtime_error("enqueue on stopped ThreadPool"); + + tasks.emplace([task](){ (*task)(); }); + } + condition.notify_one(); + return res; +} + +// the destructor joins all threads +inline ThreadPool::~ThreadPool() +{ + { + std::unique_lock lock(queue_mutex); + stop = true; + } + condition.notify_all(); + for(std::thread &worker: workers) + worker.join(); +} + +#endif diff --git a/src/sfizz/Config.h b/src/sfizz/Config.h index 477c3ce3..1c8b97a7 100644 --- a/src/sfizz/Config.h +++ b/src/sfizz/Config.h @@ -39,6 +39,7 @@ namespace config { constexpr bool loggingEnabled { false }; constexpr size_t numChannels { 2 }; constexpr int numBackgroundThreads { 4 }; + constexpr unsigned fileClearingPeriod { 5 }; // in seconds constexpr int numVoices { 64 }; constexpr unsigned maxVoices { 256 }; constexpr unsigned smoothingSteps { 512 }; diff --git a/src/sfizz/FilePool.cpp b/src/sfizz/FilePool.cpp index c92ebcbd..ec9dee51 100644 --- a/src/sfizz/FilePool.cpp +++ b/src/sfizz/FilePool.cpp @@ -45,6 +45,9 @@ #else #include #endif +#include "threadpool/ThreadPool.h" +using namespace std::placeholders; +static ThreadPool threadPool { sfz::config::numBackgroundThreads }; void readBaseFile(sfz::AudioReader& reader, sfz::FileAudioBuffer& output, uint32_t numFrames) { @@ -67,18 +70,18 @@ void readBaseFile(sfz::AudioReader& reader, sfz::FileAudioBuffer& output, uint32 } } -std::unique_ptr readFromFile(sfz::AudioReader& reader, uint32_t numFrames, sfz::Oversampling factor) +sfz::FileAudioBuffer readFromFile(sfz::AudioReader& reader, uint32_t numFrames, sfz::Oversampling factor) { - auto baseBuffer = absl::make_unique(); - readBaseFile(reader, *baseBuffer, numFrames); + sfz::FileAudioBuffer baseBuffer; + readBaseFile(reader, baseBuffer, numFrames); if (factor == sfz::Oversampling::x1) return baseBuffer; - auto outputBuffer = absl::make_unique(reader.channels(), numFrames * static_cast(factor)); - outputBuffer->clear(); + sfz::FileAudioBuffer outputBuffer { reader.channels(), numFrames * static_cast(factor) }; + outputBuffer.clear(); sfz::Oversampler oversampler { factor }; - oversampler.stream(*baseBuffer, *outputBuffer); + oversampler.stream(baseBuffer, outputBuffer); return outputBuffer; } @@ -93,37 +96,27 @@ void streamFromFile(sfz::AudioReader& reader, uint32_t numFrames, sfz::Oversampl } sfz::FilePool::FilePool(sfz::Logger& logger) -: logger(logger) + : logger(logger) { - FilePromise promise; - if (!promise.dataStatus.is_lock_free()) - DBG("atomic is not lock-free; could cause issues with locking"); - - for (int i = 0; i < config::numBackgroundThreads; ++i) - threadPool.emplace_back( &FilePool::loadingThread, this ); - - threadPool.emplace_back( &FilePool::clearingThread, this ); - - for (int i = 0; i < config::maxFilePromises; ++i) - emptyPromises.push_back(std::make_shared()); + loadingJobs.reserve(config::maxVoices); + loadingJobs.reserve(config::maxVoices); + garbageToCollect.reserve(config::maxVoices); } sfz::FilePool::~FilePool() { - quitThread = true; - std::error_code ec; - for (unsigned i = 0; i < threadPool.size(); ++i) { - ec = std::error_code(); - workerBarrier.post(ec); - } + garbageFlag = false; + semGarbageBarrier.post(ec); + garbageThread.join(); - ec = std::error_code(); - semClearingRequest.post(ec); + dispatchFlag = false; + dispatchBarrier.post(ec); + dispatchThread.join(); - for (auto& thread: threadPool) - thread.join(); + for (auto& job : loadingJobs) + job.wait(); } bool sfz::FilePool::checkSample(std::string& filename) const noexcept @@ -142,7 +135,7 @@ bool sfz::FilePool::checkSample(std::string& filename) const noexcept static const fs::path dot { "." }; static const fs::path dotdot { ".." }; - for (const fs::path &part : oldPath.relative_path()) { + for (const fs::path& part : oldPath.relative_path()) { if (part == dot || part == dotdot) { path /= part; continue; @@ -153,7 +146,7 @@ bool sfz::FilePool::checkSample(std::string& filename) const noexcept continue; } - auto it = path.empty() ? fs::directory_iterator{ dot, ec } : fs::directory_iterator{ path, ec }; + auto it = path.empty() ? fs::directory_iterator { dot, ec } : fs::directory_iterator { path, ec }; if (ec) { DBG("Error creating a directory iterator for " << filename << " (Error code: " << ec.message() << ")"); return false; @@ -169,10 +162,10 @@ bool sfz::FilePool::checkSample(std::string& filename) const noexcept #endif }; - while (it != fs::directory_iterator{} && !searchPredicate(*it)) + while (it != fs::directory_iterator {} && !searchPredicate(*it)) it.increment(ec); - if (it == fs::directory_iterator{}) { + if (it == fs::directory_iterator {}) { DBG("File not found, could not resolve " << filename); return false; } @@ -275,22 +268,30 @@ bool sfz::FilePool::preloadFile(const FileId& fileId, uint32_t maxOffset) noexce const auto existingFile = preloadedFiles.find(fileId); if (existingFile != preloadedFiles.end()) { - if (framesToLoad > existingFile->second.preloadedData->getNumFrames()) { + if (framesToLoad > existingFile->second.preloadedData.getNumFrames()) { preloadedFiles[fileId].information.maxOffset = maxOffset; preloadedFiles[fileId].preloadedData = readFromFile(*reader, framesToLoad, oversamplingFactor); } } else { - fileInformation->sampleRate = static_cast(oversamplingFactor) * static_cast(reader->sampleRate()); - FileDataHandle handle { + const auto factor = static_cast(oversamplingFactor); + fileInformation->sampleRate = factor * static_cast(reader->sampleRate()); + fileInformation->end = static_cast(factor * fileInformation->end); + fileInformation->loopBegin = static_cast(factor * fileInformation->loopBegin); + fileInformation->loopEnd = static_cast(factor * fileInformation->loopEnd); + auto insertedPair = preloadedFiles.insert_or_assign(fileId, { readFromFile(*reader, framesToLoad, oversamplingFactor), *fileInformation - }; - preloadedFiles.insert_or_assign(fileId, handle); + }); + + if (!insertedPair.second) + return false; + + insertedPair.first->second.status = FileData::Status::Preloaded; } return true; } -absl::optional sfz::FilePool::loadFile(const FileId& fileId) noexcept +sfz::FileDataHolder sfz::FilePool::loadFile(const FileId& fileId) noexcept { auto fileInformation = getFileInformation(fileId); if (!fileInformation) @@ -299,54 +300,44 @@ absl::optional sfz::FilePool::loadFile(const FileId& fileId const fs::path file { rootDirectory / fileId.filename() }; AudioReaderPtr reader = createAudioReader(file, fileId.isReverse()); - // FIXME: Large offsets will require large preloading; is this OK in practice? Apparently sforzando does the same const auto frames = static_cast(reader->frames()); const auto existingFile = loadedFiles.find(fileId); if (existingFile != loadedFiles.end()) { - return existingFile->second; + return { &existingFile->second }; } else { - fileInformation->sampleRate = static_cast(oversamplingFactor) * static_cast(reader->sampleRate()); - FileDataHandle handle { + const auto factor = static_cast(oversamplingFactor); + fileInformation->sampleRate = factor * static_cast(reader->sampleRate()); + fileInformation->end = static_cast(factor * fileInformation->end); + fileInformation->loopBegin = static_cast(factor * fileInformation->loopBegin); + fileInformation->loopEnd = static_cast(factor * fileInformation->loopEnd); + auto insertedPair = preloadedFiles.insert_or_assign(fileId, { readFromFile(*reader, frames, oversamplingFactor), *fileInformation - }; - loadedFiles.insert_or_assign(fileId, handle); - return handle; + }); + insertedPair.first->second.status = FileData::Status::Preloaded; + ASSERT(insertedPair.second); + return { &insertedPair.first->second }; } } -sfz::FilePromisePtr sfz::FilePool::getFilePromise(const FileId& fileId) noexcept +sfz::FileDataHolder sfz::FilePool::getFilePromise(const FileId& fileId) noexcept { - if (emptyPromises.empty()) { - DBG("[sfizz] No empty promises left to honor the one for " << fileId); - return {}; - } - const auto preloaded = preloadedFiles.find(fileId); if (preloaded == preloadedFiles.end()) { DBG("[sfizz] File not found in the preloaded files: " << fileId); return {}; } - - auto promise = emptyPromises.back(); - promise->fileId = preloaded->first; - promise->preloadedData = preloaded->second.preloadedData; - promise->sampleRate = static_cast(preloaded->second.information.sampleRate); - promise->oversamplingFactor = oversamplingFactor; - promise->creationTime = std::chrono::high_resolution_clock::now(); - - if (!promiseQueue.try_push(promise)) { - DBG("[sfizz] Could not enqueue the promise for " << fileId << " (queue capacity " << promiseQueue.capacity() << ")"); + QueuedFileData queuedData { fileId, &preloaded->second, std::chrono::high_resolution_clock::now() }; + if (!filesToLoad.try_push(queuedData)) { + DBG("[sfizz] Could not enqueue the file to load for " << fileId << " (queue capacity " << filesToLoad.capacity() << ")"); return {}; } std::error_code ec; - workerBarrier.post(ec); + dispatchBarrier.post(ec); ASSERT(!ec); - emptyPromises.pop_back(); - - return promise; + return { &preloaded->second }; } void sfz::FilePool::setPreloadSize(uint32_t preloadSize) noexcept @@ -358,118 +349,67 @@ void sfz::FilePool::setPreloadSize(uint32_t preloadSize) noexcept // Update all the preloaded sizes for (auto& preloadedFile : preloadedFiles) { const auto maxOffset = preloadedFile.second.information.maxOffset; + const auto numFrames = preloadedFile.second.preloadedData.getNumFrames() / static_cast(oversamplingFactor); fs::path file { rootDirectory / preloadedFile.first.filename() }; AudioReaderPtr reader = createAudioReader(file, preloadedFile.first.isReverse()); preloadedFile.second.preloadedData = readFromFile(*reader, preloadSize + maxOffset, oversamplingFactor); } } -void sfz::FilePool::tryToClearPromises() +void sfz::FilePool::loadingJob(QueuedFileData data) noexcept { - const std::lock_guard promiseLock { promiseGuard }; + raiseCurrentThreadPriority(); - for (auto& promise: promisesToClear) { - if (promise->dataStatus != FilePromise::DataStatus::Wait) - promise->reset(); + const auto loadStartTime = std::chrono::high_resolution_clock::now(); + const auto waitDuration = loadStartTime - data.queuedTime; + const fs::path file { rootDirectory / data.id.filename() }; + std::error_code readError; + AudioReaderPtr reader = createAudioReader(file, data.id.isReverse(), &readError); + + if (readError) { + DBG("[sfizz] libsndfile errored for " << data.id << " with message " << readError.message()); + return; } -} -void sfz::FilePool::clearingThread() -{ - raiseCurrentThreadPriority(); + FileData::Status currentStatus = data.data->status.load(); - RTSemaphore& request = semClearingRequest; - do { - request.wait(); - if (quitThread) + unsigned spinCounter { 0 }; + if (currentStatus == FileData::Status::Invalid) { + // Spin until the state changes + if (spinCounter > 1024) { + DBG("[sfizz] " << data.id << " is stuck on Invalid? Leaving the load"); return; - tryToClearPromises(); - } while (1); -} - -void sfz::FilePool::loadingThread() noexcept -{ - raiseCurrentThreadPriority(); - - FilePromisePtr promise; - do { - workerBarrier.wait(); - - if (emptyQueue) { - while (promiseQueue.try_pop(promise)) { - // We're just dequeuing - } - emptyQueue = false; - semEmptyQueueFinished.post(); - continue; } - if (quitThread) - return; + std::this_thread::sleep_for(std::chrono::microseconds(100)); + currentStatus = data.data->status.load(); + spinCounter += 1; + } - if (!promiseQueue.try_pop(promise)) { - continue; - } + // Already loading or loaded + if (currentStatus != FileData::Status::Preloaded) + return; - threadsLoading++; - const auto loadStartTime = std::chrono::high_resolution_clock::now(); - const auto waitDuration = loadStartTime - promise->creationTime; + // Someone else got the token + if (!data.data->status.compare_exchange_strong(currentStatus, FileData::Status::Streaming)) + return; - const fs::path file { rootDirectory / promise->fileId.filename() }; - std::error_code readError; - AudioReaderPtr reader = createAudioReader(file, promise->fileId.isReverse(), &readError); - if (readError) { - DBG("[sfizz] libsndfile errored for " << promise->fileId << " with message " << readError.message()); - promise->dataStatus = FilePromise::DataStatus::Error; - continue; - } - const auto frames = static_cast(reader->frames()); - streamFromFile(*reader, frames, oversamplingFactor, promise->fileData, &promise->availableFrames); - promise->dataStatus = FilePromise::DataStatus::Ready; - const auto loadDuration = std::chrono::high_resolution_clock::now() - loadStartTime; - logger.logFileTime(waitDuration, loadDuration, frames, promise->fileId.filename()); + const auto frames = static_cast(reader->frames()); + streamFromFile(*reader, frames, oversamplingFactor, data.data->fileData, &data.data->availableFrames); + const auto loadDuration = std::chrono::high_resolution_clock::now() - loadStartTime; + logger.logFileTime(waitDuration, loadDuration, frames, data.id.filename()); - threadsLoading--; + data.data->status = FileData::Status::Done; - semFilledPromiseQueueAvailable.wait(); - filledPromiseQueue.push(promise); - - promise.reset(); - } while (1); + std::lock_guard guard { lastUsedMutex }; + if (absl::c_find(lastUsedFiles, data.id) == lastUsedFiles.end()) + lastUsedFiles.push_back(data.id); } void sfz::FilePool::clear() { emptyFileLoadingQueues(); preloadedFiles.clear(); - temporaryFilePromises.clear(); - promisesToClear.clear(); -} - -void sfz::FilePool::cleanupPromises() noexcept -{ - const std::unique_lock lock { promiseGuard, std::try_to_lock }; - if (!lock.owns_lock()) - return; - - // The garbage collection cleared the data from these so we can move them - // back to the empty queue - auto promiseWaiting = [](FilePromisePtr& p) { return p->waiting(); }; - auto moveToEmpty = [&](FilePromisePtr& p) { return emptyPromises.push_back(p); }; - swapAndPopAll(promisesToClear, promiseWaiting, moveToEmpty); - - // Remove the promises from the filled queue and put them in a linear - // storage - FilePromisePtr promise; - while (filledPromiseQueue.try_pop(promise)) { - semFilledPromiseQueueAvailable.post(); - temporaryFilePromises.push_back(promise); - } - - auto promiseUsedOnce = [](FilePromisePtr& p) { return p.use_count() == 1; }; - auto moveToClear = [&](FilePromisePtr& p) { return promisesToClear.push_back(p); }; - if (swapAndPopAll(temporaryFilePromises, promiseUsedOnce, moveToClear) > 0) - semClearingRequest.post(); } void sfz::FilePool::setOversamplingFactor(sfz::Oversampling factor) noexcept @@ -485,10 +425,22 @@ void sfz::FilePool::setOversamplingFactor(sfz::Oversampling factor) noexcept preloadedFile.second.information.maxOffset + preloadSize ); }(); + fs::path file { rootDirectory / preloadedFile.first.filename() }; AudioReaderPtr reader = createAudioReader(file, preloadedFile.first.isReverse()); preloadedFile.second.preloadedData = readFromFile(*reader, framesToLoad, factor); - preloadedFile.second.information.sampleRate *= samplerateChange; + FileInformation& information = preloadedFile.second.information; + information.sampleRate *= samplerateChange; + information.end = static_cast(samplerateChange * information.end); + information.loopBegin = static_cast(samplerateChange * information.loopBegin); + information.loopEnd = static_cast(samplerateChange * information.loopEnd); + + if (preloadedFile.second.status == FileData::Status::Done) { + const auto realFrames = + preloadedFile.second.availableFrames.load() / static_cast(this->oversamplingFactor); + preloadedFile.second.fileData = readFromFile(*reader, realFrames, factor); + preloadedFile.second.availableFrames = realFrames * static_cast(factor); + } } this->oversamplingFactor = factor; @@ -504,26 +456,71 @@ uint32_t sfz::FilePool::getPreloadSize() const noexcept return preloadSize; } +template +bool is_ready(std::future const& f) +{ + return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready; +} + +void sfz::FilePool::dispatchingJob() noexcept +{ + QueuedFileData queuedData; + while (dispatchFlag) { + dispatchBarrier.wait(); + + if (emptyQueueFlag) { + while (filesToLoad.try_pop(queuedData)) { + // pass + } + semEmptyQueueFinished.post(); + emptyQueueFlag = false; + continue; + } + + if (filesToLoad.try_pop(queuedData)) { + loadingJobs.push_back( + threadPool.enqueue([this](const QueuedFileData& data) { loadingJob(data); }, queuedData)); + } + + // Clear finished jobs + swapAndPopAll(loadingJobs, [](std::future& future) { + return future.wait_for(std::chrono::seconds(0)) == std::future_status::ready; + }); + } +} + +void sfz::FilePool::garbageJob() noexcept +{ + std::chrono::seconds counter { 0 }; // This avoids waiting to long on the thread + constexpr std::chrono::milliseconds timeAtom { 100 }; + while (garbageFlag) { + semGarbageBarrier.wait(); + { + std::lock_guard guard { garbageMutex }; + for (auto& g: garbageToCollect) + g.reset(); + + garbageToCollect.clear(); + } + } +} + void sfz::FilePool::emptyFileLoadingQueues() noexcept { - emptyQueue = true; - workerBarrier.post(); - semEmptyQueueFinished.wait(); + ASSERT(dispatchFlag); + emptyQueueFlag = true; + std::error_code ec; + dispatchBarrier.post(ec); + + if (!ec) + semEmptyQueueFinished.wait(); } void sfz::FilePool::waitForBackgroundLoading() noexcept { - // TODO: validate that this is enough, otherwise we will need an atomic count - // of the files we need to load still. - // Spinlocking on the size of the background queue - while (!promiseQueue.was_empty()){ - std::this_thread::sleep_for(std::chrono::microseconds(100)); - } - - // Spinlocking on the threads possibly logging in the background - while (threadsLoading > 0) { - std::this_thread::sleep_for(std::chrono::microseconds(100)); - } + for (auto& job : loadingJobs) + job.wait(); + loadingJobs.clear(); } void sfz::FilePool::raiseCurrentThreadPriority() noexcept @@ -548,8 +545,7 @@ void sfz::FilePool::raiseCurrentThreadPriority() noexcept policy = SCHED_RR; const int minprio = sched_get_priority_min(policy); const int maxprio = sched_get_priority_max(policy); - param.sched_priority = minprio + - config::backgroundLoaderPthreadPriority * (maxprio - minprio) / 100; + param.sched_priority = minprio + config::backgroundLoaderPthreadPriority * (maxprio - minprio) / 100; if (pthread_setschedparam(thread, policy, ¶m) != 0) { DBG("[sfizz] Cannot set current thread scheduling parameters"); @@ -579,3 +575,40 @@ void sfz::FilePool::setRamLoading(bool loadInRam) noexcept setPreloadSize(preloadSize); } } + +void sfz::FilePool::triggerGarbageCollection() noexcept +{ + const std::unique_lock lastUsedLock { lastUsedMutex, std::try_to_lock }; + const std::unique_lock garbageLock { garbageMutex, std::try_to_lock }; + if (!lastUsedLock.owns_lock() || !garbageLock.owns_lock()) + return; + + const auto now = std::chrono::high_resolution_clock::now(); + swapAndPopAll(lastUsedFiles, [&](const FileId& id) { + if (garbageToCollect.size() == garbageToCollect.capacity()) + return false; + + auto& data = preloadedFiles[id]; + if (data.status == FileData::Status::Preloaded) + return true; + + if (data.status != FileData::Status::Done) + return false; + + if (data.readerCount != 0) + return false; + + const auto secondsIdle = std::chrono::duration_cast(now - data.lastViewerLeftAt).count(); + if (secondsIdle < config::fileClearingPeriod) + return false; + + data.availableFrames = 0; + data.status = FileData::Status::Preloaded; + garbageToCollect.push_back(std::move(data.fileData)); + return true; + }); + + std::error_code ec; + semGarbageBarrier.post(ec); + ASSERT(!ec); +} diff --git a/src/sfizz/FilePool.h b/src/sfizz/FilePool.h index ad357ff6..53846ad1 100644 --- a/src/sfizz/FilePool.h +++ b/src/sfizz/FilePool.h @@ -42,7 +42,8 @@ #include "Logger.h" #include #include -#include +#include +#include "utility/SpinMutex.h" namespace sfz { using FileAudioBuffer = AudioBuffer getData() { - if (dataStatus == DataStatus::Ready) - return AudioSpan(fileData); - else if (availableFrames > preloadedData->getNumFrames()) + if (availableFrames > preloadedData.getNumFrames()) return AudioSpan(fileData).first(availableFrames); else - return AudioSpan(*preloadedData); + return AudioSpan(preloadedData); } - void reset() + FileData(const FileData& other) = default; + FileData& operator=(const FileData& other) = default; + FileData(FileData&& other) { - fileData.reset(); - preloadedData.reset(); - fileId = FileId {}; - availableFrames = 0; - dataStatus = DataStatus::Wait; - oversamplingFactor = config::defaultOversamplingFactor; - sampleRate = config::defaultSampleRate; + ASSERT(other.readerCount == 0); // Probably should not be moving this... + information = std::move(other.information); + preloadedData = std::move(other.preloadedData); + fileData = std::move(other.fileData); + availableFrames = other.availableFrames.load(); + lastViewerLeftAt = other.lastViewerLeftAt; + status = other.status.load(); } - - enum class DataStatus { - Wait = 0, - Ready, - Error, - }; - - bool waiting() const { return dataStatus == DataStatus::Wait; } - - void sleepUntilComplete() + FileData& operator=(FileData&& other) { - while (waiting()) - std::this_thread::sleep_for(std::chrono::milliseconds(50)); + ASSERT(other.readerCount == 0); // Probably should not be moving this... + information = std::move(other.information); + preloadedData = std::move(other.preloadedData); + fileData = std::move(other.fileData); + availableFrames = other.availableFrames.load(); + lastViewerLeftAt = other.lastViewerLeftAt; + status = other.status.load(); + return *this; } - FileId fileId {}; - FileAudioBufferPtr preloadedData {}; + FileAudioBuffer preloadedData; + FileInformation information; FileAudioBuffer fileData {}; - float sampleRate { config::defaultSampleRate }; - Oversampling oversamplingFactor { config::defaultOversamplingFactor }; + std::atomic status { Status::Invalid }; std::atomic availableFrames { 0 }; - std::atomic dataStatus { DataStatus::Wait }; - std::chrono::time_point creationTime; + std::atomic readerCount { 0 }; + std::chrono::time_point lastViewerLeftAt; - LEAK_DETECTOR(FilePromise); + LEAK_DETECTOR(FileData); +}; + + +class FileDataHolder { +public: + FileDataHolder() = default; + FileDataHolder(const FileDataHolder&) = delete; + FileDataHolder& operator=(const FileDataHolder&) = delete; + FileDataHolder(FileDataHolder&& other) + { + this->data = other.data; + other.data = nullptr; + } + FileDataHolder& operator=(FileDataHolder&& other) + { + this->data = other.data; + other.data = nullptr; + return *this; + } + FileDataHolder(FileData* data) : data(data) + { + if (!data) + return; + + data->readerCount += 1; + } + void reset() + { + if (!data) + return; + + data->readerCount -= 1; + data->lastViewerLeftAt = std::chrono::high_resolution_clock::now(); + data = nullptr; + } + ~FileDataHolder() + { + ASSERT(!data || data->readerCount > 0); + reset(); + } + FileData& operator*() { return *data; } + FileData* operator->() { return data; } + explicit operator bool() const { return data != nullptr; } +private: + FileData* data { nullptr }; + LEAK_DETECTOR(FileDataHolder); }; -using FilePromisePtr = std::shared_ptr; /** * @brief This is a singleton-designed class that holds all the preloaded data * as well as functions to request new file data and collect the file handles to @@ -188,7 +231,7 @@ public: * @param fileId * @return A handle on the file data */ - absl::optional loadFile(const FileId& fileId) noexcept; + FileDataHolder loadFile(const FileId& fileId) noexcept; /** * @brief Check that the sample exists. If not, try to find it in a case insensitive way. @@ -214,19 +257,12 @@ public: */ void clear(); /** - * @brief Moves the filled promises to a linear storage, and checks - * said linear storage for promises that are not used anymore. - * - * This function has to be called on the audio thread. - */ - void cleanupPromises() noexcept; - /** - * @brief Get a file promise + * @brief Get a handle on a file, which triggers background loading * * @param fileId the file to preload - * @return FilePromisePtr a file promise + * @return FileDataHolder a file data handle */ - FilePromisePtr getFilePromise(const FileId& fileId) noexcept; + FileDataHolder getFilePromise(const FileId& fileId) noexcept; /** * @brief Change the preloading size. This will trigger a full * reload of all samples, so don't call it on the audio thread. @@ -277,37 +313,51 @@ public: * @param loadInRam */ void setRamLoading(bool loadInRam) noexcept; + /** + * @brief Prepares unused data to be freed on a background thread. + * This should be called regularly by the Synth, otherwise memory + * risk building up. + */ + void triggerGarbageCollection() noexcept; private: Logger& logger; fs::path rootDirectory; - void loadingThread() noexcept; - void clearingThread(); - void tryToClearPromises(); - atomic_queue::AtomicQueue2 promiseQueue; - atomic_queue::AtomicQueue2 filledPromiseQueue; - RTSemaphore semFilledPromiseQueueAvailable { config::maxVoices }; bool loadInRam { config::loadInRam }; uint32_t preloadSize { config::preloadSize }; Oversampling oversamplingFactor { config::defaultOversamplingFactor }; - // Signals - volatile bool quitThread { false }; - volatile bool emptyQueue { false }; - RTSemaphore semEmptyQueueFinished; - std::atomic threadsLoading { 0 }; - RTSemaphore workerBarrier; - RTSemaphore semClearingRequest; - // File promises data structures along with their guards. - std::vector emptyPromises; - std::vector temporaryFilePromises; - std::vector promisesToClear; - SpinMutex promiseGuard; + // Signals + volatile bool dispatchFlag { true }; + volatile bool garbageFlag { true }; + volatile bool emptyQueueFlag { false }; + RTSemaphore dispatchBarrier; + RTSemaphore semEmptyQueueFinished; + RTSemaphore semGarbageBarrier; + + // Structures for the background loaders + struct QueuedFileData + { + FileId id; + FileData* data; + std::chrono::time_point queuedTime; + }; + atomic_queue::AtomicQueue2 filesToLoad; + void dispatchingJob() noexcept; + void garbageJob() noexcept; + void loadingJob(QueuedFileData data) noexcept; + std::vector> loadingJobs; + std::thread dispatchThread { &FilePool::dispatchingJob, this }; + std::thread garbageThread { &FilePool::garbageJob, this }; + + SpinMutex lastUsedMutex; + std::vector lastUsedFiles; + SpinMutex garbageMutex; + std::vector garbageToCollect; // Preloaded data - absl::flat_hash_map preloadedFiles; - absl::flat_hash_map loadedFiles; - std::vector threadPool { }; + absl::flat_hash_map preloadedFiles; + absl::flat_hash_map loadedFiles; LEAK_DETECTOR(FilePool); }; } diff --git a/src/sfizz/Synth.cpp b/src/sfizz/Synth.cpp index 498a9f62..1a2f7183 100644 --- a/src/sfizz/Synth.cpp +++ b/src/sfizz/Synth.cpp @@ -840,6 +840,15 @@ void sfz::Synth::renderBlock(AudioSpan buffer) noexcept if (resources.synthConfig.freeWheeling) resources.filePool.waitForBackgroundLoading(); + const auto now = std::chrono::high_resolution_clock::now(); + const auto timeSinceLastCollection = + std::chrono::duration_cast(now - lastGarbageCollection); + + if (timeSinceLastCollection.count() > config::fileClearingPeriod) { + lastGarbageCollection = now; + resources.filePool.triggerGarbageCollection(); + } + const std::unique_lock lock { callbackGuard, std::try_to_lock }; if (!lock.owns_lock()) return; @@ -860,7 +869,6 @@ void sfz::Synth::renderBlock(AudioSpan buffer) noexcept { // Main render block ScopedTiming logger { callbackBreakdown.renderMethod, ScopedTiming::Operation::addToDuration }; tempMixSpan->fill(0.0f); - resources.filePool.cleanupPromises(); // Ramp out whatever is in the buffer at this point; should only be killed voice data linearRamp(*rampSpan, 1.0f, -1.0f / static_cast(numFrames)); diff --git a/src/sfizz/Synth.h b/src/sfizz/Synth.h index c0d32c2e..bea54281 100644 --- a/src/sfizz/Synth.h +++ b/src/sfizz/Synth.h @@ -941,6 +941,8 @@ private: Duration dispatchDuration { 0 }; + std::chrono::time_point lastGarbageCollection; + Parser parser; fs::file_time_type modificationTime { }; diff --git a/src/sfizz/Voice.cpp b/src/sfizz/Voice.cpp index 7dd2f948..92092404 100644 --- a/src/sfizz/Voice.cpp +++ b/src/sfizz/Voice.cpp @@ -95,13 +95,13 @@ void sfz::Voice::startVoice(Region* region, int delay, const TriggerEvent& event setupOscillatorUnison(); } else { currentPromise = resources.filePool.getFilePromise(region->sampleId); - if (currentPromise == nullptr) { + if (!currentPromise) { switchState(State::cleanMeUp); return; } updateLoopInformation(); - speedRatio = static_cast(currentPromise->sampleRate / this->sampleRate); - sourcePosition = region->getOffset(currentPromise->oversamplingFactor); + speedRatio = static_cast(currentPromise->information.sampleRate / this->sampleRate); + sourcePosition = region->getOffset(resources.filePool.getOversamplingFactor()); } // do Scala retuning and reconvert the frequency into a 12TET key number @@ -545,7 +545,7 @@ void sfz::Voice::fillWithData(AudioSpan buffer) noexcept if (numSamples == 0) return; - if (currentPromise == nullptr) { + if (!currentPromise) { DBG("[Voice] Missing promise during fillWithData"); return; } @@ -647,17 +647,17 @@ void sfz::Voice::fillWithData(AudioSpan buffer) noexcept else { // cut short the voice at the instant of reaching end of sample const auto sampleEnd = min( - static_cast(region->trueSampleEnd(currentPromise->oversamplingFactor)), + static_cast(currentPromise->information.end), static_cast(source.getNumFrames()) ) - 1; for (unsigned i = 0; i < numSamples; ++i) { if ((*indices)[i] >= sampleEnd) { #ifndef NDEBUG // Check for underflow - if (source.getNumFrames() - 1 < region->trueSampleEnd(currentPromise->oversamplingFactor)) { + if (source.getNumFrames() - 1 < currentPromise->information.end) { DBG("[sfizz] Underflow: source available samples " << source.getNumFrames() << "/" - << region->trueSampleEnd(currentPromise->oversamplingFactor) + << currentPromise->information.end << " for sample " << region->sampleId); } #endif @@ -1091,16 +1091,13 @@ void sfz::Voice::updateLoopInformation() noexcept if (!region->shouldLoop()) return; + const auto& info = currentPromise->information; + const auto rate = info.sampleRate; - const auto factor = currentPromise->oversamplingFactor; - const auto rate = currentPromise->sampleRate; - - loop.end = static_cast(region->loopEnd(factor)); - loop.start = static_cast(region->loopStart(factor)); + loop.end = static_cast(info.loopEnd); + loop.start = static_cast(info.loopBegin); loop.size = loop.end + 1 - loop.start; - loop.xfSize = static_cast( - lroundPositive(region->loopCrossfade * static_cast(factor) * rate) - ); + loop.xfSize = static_cast(lroundPositive(region->loopCrossfade * rate)); loop.xfOutStart = loop.end + 1 - loop.xfSize; loop.xfInStart = loop.start - loop.xfSize; } diff --git a/src/sfizz/Voice.h b/src/sfizz/Voice.h index dbcae7d7..bc0b66ff 100644 --- a/src/sfizz/Voice.h +++ b/src/sfizz/Voice.h @@ -527,7 +527,7 @@ private: */ void updateLoopInformation() noexcept; - FilePromisePtr currentPromise { nullptr }; + FileDataHolder currentPromise; int samplesPerBlock { config::defaultSamplesPerBlock }; float sampleRate { config::defaultSampleRate }; diff --git a/src/sfizz/Wavetables.cpp b/src/sfizz/Wavetables.cpp index c8cba91c..a5655ce6 100644 --- a/src/sfizz/Wavetables.cpp +++ b/src/sfizz/Wavetables.cpp @@ -525,10 +525,10 @@ bool WavetablePool::createFileWave(FilePool& filePool, const std::string& filena if (fileHandle->information.numChannels > 1) DBG("[sfizz] Only the first channel of " << filename << " will be used to create the wavetable"); - auto audioData = fileHandle->preloadedData->getConstSpan(0); + auto audioData = fileHandle->preloadedData.getConstSpan(0); // an even size is required for FFT - static_assert(absl::remove_reference_tpreloadedData)>::PaddingRight > 0, + static_assert(absl::remove_reference_tpreloadedData)>::PaddingRight > 0, "Right padding is required on the audio file buffer"); if (audioData.size() & 1) audioData = absl::MakeConstSpan(audioData.data(), audioData.size() + 1); From e693e82a396212ac1dd4812def6f6a5e25cb5750 Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Sat, 19 Sep 2020 01:17:07 +0200 Subject: [PATCH 2/8] The dispatcher was not spinning --- src/sfizz/FilePool.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sfizz/FilePool.cpp b/src/sfizz/FilePool.cpp index ec9dee51..05bd0770 100644 --- a/src/sfizz/FilePool.cpp +++ b/src/sfizz/FilePool.cpp @@ -374,7 +374,7 @@ void sfz::FilePool::loadingJob(QueuedFileData data) noexcept FileData::Status currentStatus = data.data->status.load(); unsigned spinCounter { 0 }; - if (currentStatus == FileData::Status::Invalid) { + while (currentStatus == FileData::Status::Invalid) { // Spin until the state changes if (spinCounter > 1024) { DBG("[sfizz] " << data.id << " is stuck on Invalid? Leaving the load"); From 9e0a1dff72c41fbe52b1c918aeeb3688fcb3e7cd Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Sat, 19 Sep 2020 02:14:26 +0200 Subject: [PATCH 3/8] Add constructors to the QueuedFileData Honestly a stupid queue would be fine at this point --- src/sfizz/FilePool.h | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/sfizz/FilePool.h b/src/sfizz/FilePool.h index 53846ad1..84cf06ed 100644 --- a/src/sfizz/FilePool.h +++ b/src/sfizz/FilePool.h @@ -338,9 +338,17 @@ private: // Structures for the background loaders struct QueuedFileData { - FileId id; - FileData* data; - std::chrono::time_point queuedTime; + using TimePoint = std::chrono::time_point; + QueuedFileData() = default; + QueuedFileData(FileId id, FileData* data, TimePoint queuedTime) + : id(id), data(data), queuedTime(queuedTime) {} + QueuedFileData(const QueuedFileData&) = default; + QueuedFileData& operator=(const QueuedFileData&) = default; + QueuedFileData(QueuedFileData&&) = default; + QueuedFileData& operator=(QueuedFileData&&) = default; + FileId id {}; + FileData* data { nullptr }; + TimePoint queuedTime {}; }; atomic_queue::AtomicQueue2 filesToLoad; void dispatchingJob() noexcept; From a505b9f037a1827c11e71ae690815977c61c8a52 Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Sat, 19 Sep 2020 02:46:42 +0200 Subject: [PATCH 4/8] gcc 4.9 idiosyncrasies --- src/sfizz/MathHelpers.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sfizz/MathHelpers.h b/src/sfizz/MathHelpers.h index d5b0cd3a..9fb035f1 100644 --- a/src/sfizz/MathHelpers.h +++ b/src/sfizz/MathHelpers.h @@ -674,7 +674,7 @@ public: } private: - std::array seeds_ {}; + std::array seeds_ {{}}; float mean_ { 0 }; float gain_ { 0 }; }; From d35ff4eb422ccc2c0fb2875e3b866df93cd4aac3 Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Mon, 21 Sep 2020 09:21:06 +0200 Subject: [PATCH 5/8] Add a volatile on the stop flag in the thread pool --- src/external/threadpool/ThreadPool.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/external/threadpool/ThreadPool.h b/src/external/threadpool/ThreadPool.h index 36169849..2e030687 100644 --- a/src/external/threadpool/ThreadPool.h +++ b/src/external/threadpool/ThreadPool.h @@ -29,7 +29,7 @@ private: // synchronization std::mutex queue_mutex; std::condition_variable condition; - bool stop; + volatile bool stop; }; // the constructor just launches some amount of workers From 1e2ae6bce79ee99400928fb6407b979c1689c332 Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Mon, 21 Sep 2020 09:37:19 +0200 Subject: [PATCH 6/8] Unused variable and a wrong name --- src/sfizz/FilePool.cpp | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/sfizz/FilePool.cpp b/src/sfizz/FilePool.cpp index 05bd0770..543c9bf6 100644 --- a/src/sfizz/FilePool.cpp +++ b/src/sfizz/FilePool.cpp @@ -99,7 +99,7 @@ sfz::FilePool::FilePool(sfz::Logger& logger) : logger(logger) { loadingJobs.reserve(config::maxVoices); - loadingJobs.reserve(config::maxVoices); + lastUsedFiles.reserve(config::maxVoices); garbageToCollect.reserve(config::maxVoices); } @@ -491,8 +491,6 @@ void sfz::FilePool::dispatchingJob() noexcept void sfz::FilePool::garbageJob() noexcept { - std::chrono::seconds counter { 0 }; // This avoids waiting to long on the thread - constexpr std::chrono::milliseconds timeAtom { 100 }; while (garbageFlag) { semGarbageBarrier.wait(); { From a0ad3b992123494e99a9d455a52b84fc0536a2a9 Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Wed, 23 Sep 2020 12:00:11 +0200 Subject: [PATCH 7/8] Adapt the thread pool to the number of concurrent threads --- src/sfizz/FilePool.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sfizz/FilePool.cpp b/src/sfizz/FilePool.cpp index 543c9bf6..172b57c5 100644 --- a/src/sfizz/FilePool.cpp +++ b/src/sfizz/FilePool.cpp @@ -47,7 +47,7 @@ #endif #include "threadpool/ThreadPool.h" using namespace std::placeholders; -static ThreadPool threadPool { sfz::config::numBackgroundThreads }; +static ThreadPool threadPool { std::thread::hardware_concurrency() > 2 ? std::thread::hardware_concurrency() - 2 : 1 }; void readBaseFile(sfz::AudioReader& reader, sfz::FileAudioBuffer& output, uint32_t numFrames) { From b78fa39d9e801a20607da8ee4ebdd82b5fdce92b Mon Sep 17 00:00:00 2001 From: Paul Ferrand Date: Wed, 7 Oct 2020 00:51:18 +0200 Subject: [PATCH 8/8] Remove unused variable --- src/sfizz/FilePool.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/src/sfizz/FilePool.cpp b/src/sfizz/FilePool.cpp index 172b57c5..6c3a08df 100644 --- a/src/sfizz/FilePool.cpp +++ b/src/sfizz/FilePool.cpp @@ -349,7 +349,6 @@ void sfz::FilePool::setPreloadSize(uint32_t preloadSize) noexcept // Update all the preloaded sizes for (auto& preloadedFile : preloadedFiles) { const auto maxOffset = preloadedFile.second.information.maxOffset; - const auto numFrames = preloadedFile.second.preloadedData.getNumFrames() / static_cast(oversamplingFactor); fs::path file { rootDirectory / preloadedFile.first.filename() }; AudioReaderPtr reader = createAudioReader(file, preloadedFile.first.isReverse()); preloadedFile.second.preloadedData = readFromFile(*reader, preloadSize + maxOffset, oversamplingFactor);