diff --git a/src/external/threadpool/ThreadPool.h b/src/external/threadpool/ThreadPool.h new file mode 100644 index 00000000..2e030687 --- /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; + volatile 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..6c3a08df 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 { std::thread::hardware_concurrency() > 2 ? std::thread::hardware_concurrency() - 2 : 1 }; 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); + lastUsedFiles.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 @@ -364,112 +355,60 @@ void sfz::FilePool::setPreloadSize(uint32_t preloadSize) noexcept } } -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 }; + while (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 +424,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 +455,69 @@ 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 +{ + 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 +542,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 +572,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..84cf06ed 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,59 @@ 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 + { + 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; + 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/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 }; }; 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);