diff --git a/scwx-qt/scwx-qt.cmake b/scwx-qt/scwx-qt.cmake index 1df379b88..614b188fa 100644 --- a/scwx-qt/scwx-qt.cmake +++ b/scwx-qt/scwx-qt.cmake @@ -118,6 +118,7 @@ set(HDR_MANAGER source/scwx/qt/manager/alert_manager.hpp source/scwx/qt/manager/media_manager.hpp source/scwx/qt/manager/placefile_manager.hpp source/scwx/qt/manager/position_manager.hpp + source/scwx/qt/manager/product_datastore.hpp source/scwx/qt/manager/provider_manager.hpp source/scwx/qt/manager/radar_coordinate_table.hpp source/scwx/qt/manager/radar_product_manager.hpp @@ -139,6 +140,7 @@ set(SRC_MANAGER source/scwx/qt/manager/alert_manager.cpp source/scwx/qt/manager/media_manager.cpp source/scwx/qt/manager/placefile_manager.cpp source/scwx/qt/manager/position_manager.cpp + source/scwx/qt/manager/product_datastore.cpp source/scwx/qt/manager/provider_manager.cpp source/scwx/qt/manager/radar_coordinate_table.cpp source/scwx/qt/manager/radar_product_manager.cpp diff --git a/scwx-qt/source/scwx/qt/manager/product_datastore.cpp b/scwx-qt/source/scwx/qt/manager/product_datastore.cpp new file mode 100644 index 000000000..fd1bd9094 --- /dev/null +++ b/scwx-qt/source/scwx/qt/manager/product_datastore.cpp @@ -0,0 +1,626 @@ +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace scwx::qt::manager +{ + +namespace +{ + +static const std::string logPrefix_ = "scwx::qt::manager::product_datastore"; +static const auto logger_ = scwx::util::Logger::Create(logPrefix_); + +using RadarProductRecordList = + std::list>; + +static constexpr std::size_t kMinimumCacheLimit_ {6u}; + +} // namespace + +class ProductDatastore::Impl +{ +public: + void SetCacheLimit(std::size_t cacheLimit) + { + cacheLimit_.store(std::max(cacheLimit, kMinimumCacheLimit_)); + } + + std::size_t cache_limit() const { return cacheLimit_.load(); } + + std::shared_ptr + Store(const std::shared_ptr& record) + { + logger_->trace("Store()"); + + if (record == nullptr) + { + return nullptr; + } + + std::shared_ptr storedRecord = nullptr; + + const auto timeInSeconds = + std::chrono::time_point_cast( + record->time()); + + if (record->radar_product_group() == common::RadarProductGroup::Level2) + { + std::unique_lock const lock {level2ProductRecordMutex_}; + + auto it = level2ProductRecords_.find(timeInSeconds); + if (it != level2ProductRecords_.cend()) + { + storedRecord = it->second.lock(); + + if (storedRecord != nullptr) + { + logger_->debug( + "Level 2 product previously loaded, loading from cache"); + } + } + + if (storedRecord == nullptr) + { + storedRecord = record; + level2ProductRecords_[timeInSeconds] = record; + } + + UpdateRecentRecords(level2ProductRecentRecords_, storedRecord); + } + else if (record->radar_product_group() == + common::RadarProductGroup::Level3) + { + std::unique_lock const lock {level3ProductRecordMutex_}; + + auto& productMap = level3ProductRecordsMap_[record->radar_product()]; + + auto it = productMap.find(timeInSeconds); + if (it != productMap.cend()) + { + storedRecord = it->second.lock(); + + if (storedRecord != nullptr) + { + logger_->debug( + "Level 3 product previously loaded, loading from cache"); + } + } + + if (storedRecord == nullptr) + { + storedRecord = record; + productMap[timeInSeconds] = record; + } + + UpdateRecentRecords( + level3ProductRecentRecordsMap_[record->radar_product()], + storedRecord); + } + else + { + logger_->debug( + "Unsupported radar product group: {}", + common::GetRadarProductGroupName(record->radar_product_group())); + } + + return storedRecord; + } + + void PopulateLevel2ProductTimes( + const std::shared_ptr& level2ProviderManager, + const std::shared_ptr& level2ChunksProviderManager, + std::chrono::system_clock::time_point time, + bool update) + { + PopulateProductTimes(level2ProviderManager, + level2ProductRecords_, + level2ProductRecordMutex_, + time, + update); + PopulateProductTimes(level2ChunksProviderManager, + level2ProductRecords_, + level2ProductRecordMutex_, + time, + update); + } + + void PopulateLevel3ProductTimes( + const std::shared_ptr& level3ProviderManager, + const std::string& product, + std::chrono::system_clock::time_point time, + bool update) + { + std::unique_lock level3ProductRecordLock {level3ProductRecordMutex_}; + auto& level3ProductRecords = level3ProductRecordsMap_[product]; + level3ProductRecordLock.unlock(); + + PopulateProductTimes(level3ProviderManager, + level3ProductRecords, + level3ProductRecordMutex_, + time, + update); + } + + std::shared_ptr + GetCachedNexradFile(common::RadarProductGroup group, + const std::string& level3Product, + std::chrono::system_clock::time_point time) + { + std::shared_ptr existingRecord = nullptr; + + const auto timeInSeconds = + std::chrono::time_point_cast(time); + + if (group == common::RadarProductGroup::Level2) + { + std::shared_lock const sharedLock {level2ProductRecordMutex_}; + + auto it = level2ProductRecords_.find(timeInSeconds); + if (it != level2ProductRecords_.cend()) + { + existingRecord = it->second.lock(); + + if (existingRecord != nullptr) + { + logger_->trace( + "Data previously loaded, loading from data cache"); + } + } + } + else if (group == common::RadarProductGroup::Level3) + { + std::shared_lock const sharedLock {level3ProductRecordMutex_}; + + auto productIt = level3ProductRecordsMap_.find(level3Product); + if (productIt != level3ProductRecordsMap_.cend()) + { + auto it = productIt->second.find(timeInSeconds); + if (it != productIt->second.cend()) + { + existingRecord = it->second.lock(); + + if (existingRecord != nullptr) + { + logger_->trace( + "Data previously loaded, loading from data cache"); + } + } + } + } + + if (existingRecord != nullptr) + { + return existingRecord->nexrad_file(); + } + + return nullptr; + } + + std::vector + FindLevel2RecordEntries(std::chrono::system_clock::time_point time) const + { + std::vector entries {}; + + std::shared_lock const lock {level2ProductRecordMutex_}; + + if (!level2ProductRecords_.empty() && + time == std::chrono::system_clock::time_point {}) + { + const auto& recordEntry = *level2ProductRecords_.rbegin(); + entries.push_back({recordEntry.first, recordEntry.second}); + } + else + { + auto recordIt = + scwx::util::GetBoundedElementIterator(level2ProductRecords_, time); + + if (recordIt != level2ProductRecords_.cend()) + { + entries.push_back({recordIt->first, recordIt->second}); + + if (recordIt != level2ProductRecords_.cbegin()) + { + const auto previousIt = std::prev(recordIt); + entries.push_back({previousIt->first, previousIt->second}); + } + } + } + + return entries; + } + + std::optional + FindLevel3RecordEntry(const std::string& product, + std::chrono::system_clock::time_point time) const + { + std::shared_lock const lock {level3ProductRecordMutex_}; + + auto it = level3ProductRecordsMap_.find(product); + if (it == level3ProductRecordsMap_.cend() || it->second.empty()) + { + return std::nullopt; + } + + if (time == std::chrono::system_clock::time_point {}) + { + const auto& recordEntry = *it->second.rbegin(); + return RadarProductRecordEntry {recordEntry.first, recordEntry.second}; + } + + auto recordPtr = scwx::util::GetBoundedElementPointer(it->second, time); + if (recordPtr == nullptr) + { + return std::nullopt; + } + + return RadarProductRecordEntry {recordPtr->first, recordPtr->second}; + } + + void ForEachLevel2Record( + const std::function& callback) const + { + std::shared_lock const lock {level2ProductRecordMutex_}; + + for (auto& record : level2ProductRecords_) + { + callback(record.first, record.second.expired()); + } + } + + void ForEachLevel3Product( + const std::function& + callback) const + { + std::shared_lock const lock {level3ProductRecordMutex_}; + + for (auto& recordMap : level3ProductRecordsMap_) + { + callback(recordMap.first, recordMap.second); + } + } + +private: + void + UpdateRecentRecords(RadarProductRecordList& recentList, + const std::shared_ptr& record) + { + const std::size_t recentListMaxSize {cacheLimit_.load()}; + bool iteratorErased = false; + + auto it = std::find(recentList.cbegin(), recentList.cend(), record); + if (it != recentList.cbegin() && it != recentList.cend()) + { + recentList.erase(it); + iteratorErased = true; + } + + if (iteratorErased || recentList.size() == 0 || it != recentList.cbegin()) + { + recentList.push_front(record); + } + + while (recentList.size() > recentListMaxSize) + { + recentList.pop_back(); + } + } + + void + PopulateProductTimes(const std::shared_ptr& providerManager, + RadarProductRecordMap& productRecordMap, + std::shared_mutex& productRecordMutex, + std::chrono::system_clock::time_point time, + bool update) + { + if (providerManager == nullptr) + { + return; + } + + const auto providers = providerManager->providers(); + if (providers.empty()) + { + return; + } + + if (update) + { + logger_->debug( + "Populating product times: {}, {}, {}", + common::GetRadarProductGroupName(providerManager->group()), + providerManager->product(), + scwx::util::time::TimeString(time)); + } + else + { + logger_->trace( + "Populating cached product times: {}, {}, {}", + common::GetRadarProductGroupName(providerManager->group()), + providerManager->product(), + scwx::util::time::TimeString(time)); + } + + auto today = std::chrono::floor(time); + + if (today == std::chrono::system_clock::time_point {}) + { + today = std::chrono::floor(scwx::util::time::now()); + } + + const auto yesterday = today - std::chrono::days {1}; + const auto tomorrow = today + std::chrono::days {1}; + const auto dates = std::array {yesterday, today, tomorrow}; + + std::set volumeTimes {}; + std::mutex volumeTimesMutex {}; + + const auto processDate = + [&](const std::shared_ptr& provider, + const auto& date) + { + if (date > scwx::util::time::now()) + { + return; + } + + const auto timePoints = provider->GetTimePointsByDate(date, update); + if (timePoints.empty()) + { + return; + } + + providerManager->NoteVolumeTimes(provider->radar_site(), timePoints); + + const std::unique_lock volumeTimesLock {volumeTimesMutex}; + + std::copy(timePoints.begin(), + timePoints.end(), + std::inserter(volumeTimes, volumeTimes.end())); + }; + + std::for_each( + std::execution::par, + providers.begin(), + providers.end(), + [&](const auto& provider) + { + if (provider->IsDateArchiveAvailable()) + { + std::for_each(std::execution::par, + dates.begin(), + dates.end(), + [&](const auto& date) + { + const auto candidates = + common::GetRadarIdCandidates( + provider->radar_site(), date); + + if (std::ranges::find(candidates, + provider->radar_site()) != + candidates.cend()) + { + processDate(provider, date); + } + }); + } + else + { + const auto candidates = + common::GetRadarIdCandidates(provider->radar_site(), today); + + if (std::ranges::find(candidates, provider->radar_site()) != + candidates.cend()) + { + processDate(provider, today); + } + } + }); + + std::unique_lock const lock {productRecordMutex}; + + std::transform( + volumeTimes.cbegin(), + volumeTimes.cend(), + std::inserter(productRecordMap, productRecordMap.begin()), + [](const std::chrono::system_clock::time_point& volumeTime) + { + return std::pair>( + volumeTime, std::weak_ptr {}); + }); + } + + std::atomic cacheLimit_ {kMinimumCacheLimit_}; + + RadarProductRecordMap level2ProductRecords_ {}; + RadarProductRecordList level2ProductRecentRecords_ {}; + std::unordered_map + level3ProductRecordsMap_ {}; + std::unordered_map + level3ProductRecentRecordsMap_ {}; + mutable std::shared_mutex level2ProductRecordMutex_ {}; + mutable std::shared_mutex level3ProductRecordMutex_ {}; +}; + +ProductDatastore::ProductDatastore() : p(std::make_unique()) {} + +ProductDatastore::~ProductDatastore() = default; + +void ProductDatastore::SetCacheLimit(std::size_t cacheLimit) +{ + p->SetCacheLimit(cacheLimit); +} + +std::size_t ProductDatastore::cache_limit() const +{ + return p->cache_limit(); +} + +std::shared_ptr ProductDatastore::Store( + const std::shared_ptr& record) +{ + return p->Store(record); +} + +bool ProductDatastore::AreProductTimesPopulated( + const std::shared_ptr& providerManager, + std::chrono::system_clock::time_point time) +{ + if (providerManager == nullptr) + { + return false; + } + + const auto providers = providerManager->providers(); + if (providers.empty()) + { + // If providers are not available, assume product times are populated + return true; + } + + auto today = std::chrono::floor(time); + + bool productTimesPopulated = false; + + if (today == std::chrono::system_clock::time_point {}) + { + today = std::chrono::floor(scwx::util::time::now()); + } + + const auto yesterday = today - std::chrono::days {1}; + const auto tomorrow = today + std::chrono::days {1}; + + for (const auto& provider : providers) + { + bool providerTimesPopulated = true; + bool providerValidForDates = false; + + if (provider->IsDateArchiveAvailable()) + { + const auto dates = std::array {yesterday, today, tomorrow}; + + for (const auto& date : dates) + { + if (date > scwx::util::time::now()) + { + continue; + } + + const auto candidates = + common::GetRadarIdCandidates(provider->radar_site(), date); + + if (std::ranges::find(candidates, provider->radar_site()) == + candidates.cend()) + { + continue; + } + + providerValidForDates = true; + + if (!provider->IsDateCached(date)) + { + providerTimesPopulated = false; + } + } + } + else + { + const auto candidates = + common::GetRadarIdCandidates(provider->radar_site(), today); + + if (std::ranges::find(candidates, provider->radar_site()) != + candidates.cend()) + { + providerValidForDates = true; + } + + if (providerValidForDates && !provider->IsDateCached(today)) + { + providerTimesPopulated = false; + } + } + + if (providerValidForDates && providerTimesPopulated) + { + productTimesPopulated = true; + break; + } + } + + return productTimesPopulated; +} + +void ProductDatastore::PopulateLevel2ProductTimes( + const std::shared_ptr& level2ProviderManager, + const std::shared_ptr& level2ChunksProviderManager, + std::chrono::system_clock::time_point time, + bool update) +{ + p->PopulateLevel2ProductTimes( + level2ProviderManager, level2ChunksProviderManager, time, update); +} + +void ProductDatastore::PopulateLevel3ProductTimes( + const std::shared_ptr& level3ProviderManager, + const std::string& product, + std::chrono::system_clock::time_point time, + bool update) +{ + p->PopulateLevel3ProductTimes(level3ProviderManager, product, time, update); +} + +std::shared_ptr ProductDatastore::GetCachedNexradFile( + common::RadarProductGroup group, + const std::string& level3Product, + std::chrono::system_clock::time_point time) +{ + return p->GetCachedNexradFile(group, level3Product, time); +} + +std::vector ProductDatastore::FindLevel2RecordEntries( + std::chrono::system_clock::time_point time) const +{ + return p->FindLevel2RecordEntries(time); +} + +std::optional ProductDatastore::FindLevel3RecordEntry( + const std::string& product, std::chrono::system_clock::time_point time) const +{ + return p->FindLevel3RecordEntry(product, time); +} + +void ProductDatastore::ForEachLevel2Record( + const std::function& callback) const +{ + p->ForEachLevel2Record(callback); +} + +void ProductDatastore::ForEachLevel3Product( + const std::function& callback) + const +{ + p->ForEachLevel3Product(callback); +} + +} // namespace scwx::qt::manager diff --git a/scwx-qt/source/scwx/qt/manager/product_datastore.hpp b/scwx-qt/source/scwx/qt/manager/product_datastore.hpp new file mode 100644 index 000000000..b1d7efbf1 --- /dev/null +++ b/scwx-qt/source/scwx/qt/manager/product_datastore.hpp @@ -0,0 +1,90 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace scwx::qt::manager +{ + +class ProviderManager; + +using RadarProductRecordMap = + std::map>; + +struct RadarProductRecordEntry +{ + std::chrono::system_clock::time_point time {}; + std::weak_ptr record {}; +}; + +class ProductDatastore +{ +public: + ProductDatastore(); + ~ProductDatastore(); + + ProductDatastore(const ProductDatastore&) = delete; + ProductDatastore& operator=(const ProductDatastore&) = delete; + ProductDatastore(ProductDatastore&&) = delete; + ProductDatastore& operator=(ProductDatastore&&) = delete; + + void SetCacheLimit(std::size_t cacheLimit); + [[nodiscard]] std::size_t cache_limit() const; + + std::shared_ptr + Store(const std::shared_ptr& record); + + static bool AreProductTimesPopulated( + const std::shared_ptr& providerManager, + std::chrono::system_clock::time_point time); + + void PopulateLevel2ProductTimes( + const std::shared_ptr& level2ProviderManager, + const std::shared_ptr& level2ChunksProviderManager, + std::chrono::system_clock::time_point time, + bool update = true); + + void PopulateLevel3ProductTimes( + const std::shared_ptr& level3ProviderManager, + const std::string& product, + std::chrono::system_clock::time_point time, + bool update = true); + + [[nodiscard]] std::shared_ptr + GetCachedNexradFile(common::RadarProductGroup group, + const std::string& level3Product, + std::chrono::system_clock::time_point time); + + [[nodiscard]] std::vector + FindLevel2RecordEntries(std::chrono::system_clock::time_point time) const; + + [[nodiscard]] std::optional + FindLevel3RecordEntry(const std::string& product, + std::chrono::system_clock::time_point time) const; + + void ForEachLevel2Record( + const std::function& callback) const; + + void ForEachLevel3Product( + const std::function& + callback) const; + +private: + class Impl; + std::unique_ptr p; +}; + +} // namespace scwx::qt::manager diff --git a/scwx-qt/source/scwx/qt/manager/radar_product_manager.cpp b/scwx-qt/source/scwx/qt/manager/radar_product_manager.cpp index 10cecc228..e4aaee81b 100644 --- a/scwx-qt/source/scwx/qt/manager/radar_product_manager.cpp +++ b/scwx-qt/source/scwx/qt/manager/radar_product_manager.cpp @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -9,7 +10,6 @@ #include #include #include -#include #include #include #include @@ -48,11 +48,6 @@ static const auto logger_ = scwx::util::Logger::Create(logPrefix_); typedef std::function()> CreateNexradFileFunction; -typedef std::map> - RadarProductRecordMap; -typedef std::list> - RadarProductRecordList; static const std::string kDefaultLevel3Product_ {"N0B"}; @@ -160,10 +155,6 @@ class RadarProductManagerImpl types::RadarProductLoadStatus> GetLevel3ProductRecord(const std::string& product, std::chrono::system_clock::time_point time); - std::shared_ptr - StoreRadarProductRecord(std::shared_ptr record); - void UpdateRecentRecords(RadarProductRecordList& recentList, - std::shared_ptr record); void LoadNexradFileAsync( CreateNexradFileFunction load, @@ -173,8 +164,7 @@ class RadarProductManagerImpl void LoadProviderData(std::chrono::system_clock::time_point time, const std::shared_ptr& providerManager, - RadarProductRecordMap& recordMap, - std::shared_mutex& recordMutex, + const std::string& level3Product, std::mutex& loadDataMutex, const std::shared_ptr& request); @@ -194,17 +184,6 @@ class RadarProductManagerImpl static float gate_size(types::RadarType radarType); - static bool AreProductTimesPopulated( - const std::shared_ptr& providerManager, - std::chrono::system_clock::time_point time); - - static void - PopulateProductTimes(std::shared_ptr providerManager, - RadarProductRecordMap& productRecordMap, - std::shared_mutex& productRecordMutex, - std::chrono::system_clock::time_point time, - bool update); - static void LoadNexradFile(CreateNexradFileFunction load, const std::shared_ptr& request, @@ -217,18 +196,10 @@ class RadarProductManagerImpl bool level3AvailabilityReady_ {false}; std::shared_ptr radarSite_; - std::size_t cacheLimit_ {6u}; - std::unique_ptr coordinateTable_ {}; + ProductDatastore productDatastore_ {}; - RadarProductRecordMap level2ProductRecords_ {}; - RadarProductRecordList level2ProductRecentRecords_ {}; - std::unordered_map - level3ProductRecordsMap_ {}; - std::unordered_map - level3ProductRecentRecordsMap_ {}; - std::shared_mutex level2ProductRecordMutex_ {}; - std::shared_mutex level3ProductRecordMutex_ {}; + std::unique_ptr coordinateTable_ {}; std::shared_ptr level2ProviderManager_; std::shared_ptr level2ChunksProviderManager_; @@ -289,38 +260,34 @@ void RadarProductManager::DumpRecords() logger_->info(" Level 2"); { - std::shared_lock level2ProductLock { - radarProductManager->p->level2ProductRecordMutex_}; - - for (auto& record : - radarProductManager->p->level2ProductRecords_) - { - logger_->info(" {}{}", - scwx::util::TimeString(record.first), - record.second.expired() ? " (expired)" : ""); - } + radarProductManager->p->productDatastore_.ForEachLevel2Record( + [&](std::chrono::system_clock::time_point recordTime, + bool expired) + { + logger_->info(" {}{}", + scwx::util::TimeString(recordTime), + expired ? " (expired)" : ""); + }); } logger_->info(" Level 3"); { - std::shared_lock level3ProductLock { - radarProductManager->p->level3ProductRecordMutex_}; - - for (auto& recordMap : - radarProductManager->p->level3ProductRecordsMap_) - { - // Product Name - logger_->info(" {}", recordMap.first); - - for (auto& record : recordMap.second) - { - logger_->info(" {}{}", - scwx::util::TimeString(record.first), - record.second.expired() ? " (expired)" : - ""); - } - } + radarProductManager->p->productDatastore_ + .ForEachLevel3Product( + [&](const std::string& product, + const RadarProductRecordMap& recordMap) + { + logger_->info(" {}", product); + + for (auto& record : recordMap) + { + logger_->info( + " {}{}", + scwx::util::TimeString(record.first), + record.second.expired() ? " (expired)" : ""); + } + }); } } } @@ -708,8 +675,7 @@ RadarProductManager::GetActiveVolumeTimes( void RadarProductManagerImpl::LoadProviderData( std::chrono::system_clock::time_point time, const std::shared_ptr& providerManager, - RadarProductRecordMap& recordMap, - std::shared_mutex& recordMutex, + const std::string& level3Product, std::mutex& loadDataMutex, const std::shared_ptr& request) { @@ -718,29 +684,14 @@ void RadarProductManagerImpl::LoadProviderData( scwx::util::TimeString(time)); LoadNexradFileAsync( - [providerManager, time, &recordMap, &recordMutex]() + [providerManager, level3Product, time, this]() -> std::shared_ptr { - std::shared_ptr existingRecord = nullptr; - std::shared_ptr nexradFile = nullptr; - - { - std::shared_lock sharedLock {recordMutex}; - - auto it = recordMap.find(time); - if (it != recordMap.cend()) - { - existingRecord = it->second.lock(); - - if (existingRecord != nullptr) - { - logger_->trace( - "Data previously loaded, loading from data cache"); - } - } - } + std::shared_ptr nexradFile = + productDatastore_.GetCachedNexradFile( + providerManager->group(), level3Product, time); - if (existingRecord == nullptr) + if (nexradFile == nullptr) { nexradFile = providerManager->LoadObjectByTime(time); @@ -750,10 +701,6 @@ void RadarProductManagerImpl::LoadProviderData( scwx::util::TimeString(time)); } } - else - { - nexradFile = existingRecord->nexrad_file(); - } return nexradFile; }, @@ -768,12 +715,8 @@ void RadarProductManager::LoadLevel2Data( { logger_->trace("LoadLevel2Data: {}", scwx::util::TimeString(time)); - p->LoadProviderData(time, - p->level2ProviderManager_, - p->level2ProductRecords_, - p->level2ProductRecordMutex_, - p->loadLevel2DataMutex_, - request); + p->LoadProviderData( + time, p->level2ProviderManager_, "", p->loadLevel2DataMutex_, request); } void RadarProductManager::LoadLevel3Data( @@ -793,17 +736,10 @@ void RadarProductManager::LoadLevel3Data( } providerManagerLock.unlock(); - // Look up product record - std::unique_lock productRecordLock(p->level3ProductRecordMutex_); - RadarProductRecordMap& level3ProductRecords = - p->level3ProductRecordsMap_[product]; - productRecordLock.unlock(); - // Load provider data p->LoadProviderData(time, level3ProviderManager->second, - level3ProductRecords, - p->level3ProductRecordMutex_, + product, p->loadLevel3DataMutex_, request); } @@ -931,7 +867,7 @@ void RadarProductManagerImpl::LoadNexradFile( manager = RadarProductManager::Instance(recordRadarId); manager->Initialize(); - record = manager->p->StoreRadarProductRecord(record); + record = manager->p->productDatastore_.Store(record); } lock.unlock(); @@ -946,117 +882,26 @@ void RadarProductManagerImpl::LoadNexradFile( bool RadarProductManagerImpl::AreLevel2ProductTimesPopulated( std::chrono::system_clock::time_point time) const { - return AreProductTimesPopulated(level2ProviderManager_, time); + return ProductDatastore::AreProductTimesPopulated(level2ProviderManager_, + time) && + ProductDatastore::AreProductTimesPopulated( + level2ChunksProviderManager_, time); } bool RadarProductManagerImpl::AreLevel3ProductTimesPopulated( const std::string& product, std::chrono::system_clock::time_point time) { - // Get provider manager const auto level3ProviderManager = GetLevel3ProviderManager(product); - return AreProductTimesPopulated(level3ProviderManager, time); -} - -bool RadarProductManagerImpl::AreProductTimesPopulated( - const std::shared_ptr& providerManager, - std::chrono::system_clock::time_point time) -{ - const auto providers = providerManager->providers(); - if (providers.empty()) - { - // If providers are not available, assume product times are populated - return true; - } - - auto today = std::chrono::floor(time); - - bool productTimesPopulated = false; - - // Assume a query for the epoch is a query for now - if (today == std::chrono::system_clock::time_point {}) - { - today = std::chrono::floor(scwx::util::time::now()); - } - - const auto yesterday = today - std::chrono::days {1}; - const auto tomorrow = today + std::chrono::days {1}; - - for (const auto& provider : providers) - { - bool providerTimesPopulated = true; - bool providerValidForDates = false; - - if (provider->IsDateArchiveAvailable()) - { - const auto dates = std::array {yesterday, today, tomorrow}; - - for (const auto& date : dates) - { - // Don't query for a time point in the future - if (date > scwx::util::time::now()) - { - continue; - } - - const auto candidates = - common::GetRadarIdCandidates(provider->radar_site(), date); - - // Skip dates outside this provider's candidate window - if (std::ranges::find(candidates, provider->radar_site()) == - candidates.cend()) - { - continue; - } - - providerValidForDates = true; - - if (!provider->IsDateCached(date)) - { - providerTimesPopulated = false; - } - } - } - else - { - const auto candidates = - common::GetRadarIdCandidates(provider->radar_site(), today); - - if (std::ranges::find(candidates, provider->radar_site()) != - candidates.cend()) - { - providerValidForDates = true; - } - - if (providerValidForDates && !provider->IsDateCached(today)) - { - providerTimesPopulated = false; - } - } - - if (providerValidForDates && providerTimesPopulated) - { - productTimesPopulated = true; - break; - } - } - - return productTimesPopulated; + return ProductDatastore::AreProductTimesPopulated(level3ProviderManager, + time); } void RadarProductManagerImpl::PopulateLevel2ProductTimes( std::chrono::system_clock::time_point time, bool update) { - PopulateProductTimes(level2ProviderManager_, - level2ProductRecords_, - level2ProductRecordMutex_, - time, - update); - PopulateProductTimes(level2ChunksProviderManager_, - level2ProductRecords_, - level2ProductRecordMutex_, - time, - update); + productDatastore_.PopulateLevel2ProductTimes( + level2ProviderManager_, level2ChunksProviderManager_, time, update); } void RadarProductManagerImpl::PopulateLevel3ProductTimes( @@ -1064,144 +909,10 @@ void RadarProductManagerImpl::PopulateLevel3ProductTimes( std::chrono::system_clock::time_point time, bool update) { - // Get provider manager auto level3ProviderManager = GetLevel3ProviderManager(product); - // Get product records - std::unique_lock level3ProductRecordLock {level3ProductRecordMutex_}; - auto& level3ProductRecords = level3ProductRecordsMap_[product]; - level3ProductRecordLock.unlock(); - - PopulateProductTimes(level3ProviderManager, - level3ProductRecords, - level3ProductRecordMutex_, - time, - update); -} - -void RadarProductManagerImpl::PopulateProductTimes( - std::shared_ptr providerManager, - RadarProductRecordMap& productRecordMap, - std::shared_mutex& productRecordMutex, - std::chrono::system_clock::time_point time, - bool update) -{ - const auto providers = providerManager->providers(); - if (providers.empty()) - { - return; - } - - if (update) - { - logger_->debug("Populating product times: {}, {}, {}", - common::GetRadarProductGroupName(providerManager->group()), - providerManager->product(), - scwx::util::time::TimeString(time)); - } - else - { - logger_->trace("Populating cached product times: {}, {}, {}", - common::GetRadarProductGroupName(providerManager->group()), - providerManager->product(), - scwx::util::time::TimeString(time)); - } - - auto today = std::chrono::floor(time); - - // Assume a query for the epoch is a query for now - if (today == std::chrono::system_clock::time_point {}) - { - today = std::chrono::floor(scwx::util::time::now()); - } - - const auto yesterday = today - std::chrono::days {1}; - const auto tomorrow = today + std::chrono::days {1}; - const auto dates = std::array {yesterday, today, tomorrow}; - - std::set volumeTimes {}; - std::mutex volumeTimesMutex {}; - - const auto processDate = - [&](const std::shared_ptr& provider, - const auto& date) - { - // Don't query for a time point in the future - if (date > scwx::util::time::now()) - { - return; - } - - // Query the provider for volume time points - const auto timePoints = provider->GetTimePointsByDate(date, update); - if (timePoints.empty()) - { - return; - } - - providerManager->NoteVolumeTimes(provider->radar_site(), timePoints); - - // Lock the merged volume time list - const std::unique_lock volumeTimesLock {volumeTimesMutex}; - - // Copy time points to the merged list - std::copy(timePoints.begin(), - timePoints.end(), - std::inserter(volumeTimes, volumeTimes.end())); - }; - - // For each provider (in parallel) - std::for_each( - std::execution::par, - providers.begin(), - providers.end(), - [&](const auto& provider) - { - if (provider->IsDateArchiveAvailable()) - { - // For yesterday, today and tomorrow (in parallel) - std::for_each( - std::execution::par, - dates.begin(), - dates.end(), - [&](const auto& date) - { - const auto candidates = - common::GetRadarIdCandidates(provider->radar_site(), date); - - if (std::ranges::find(candidates, provider->radar_site()) != - candidates.cend()) - { - processDate(provider, date); - } - }); - } - else - { - const auto candidates = - common::GetRadarIdCandidates(provider->radar_site(), today); - - if (std::ranges::find(candidates, provider->radar_site()) != - candidates.cend()) - { - processDate(provider, today); - } - } - }); - - // Lock the product record map - std::unique_lock lock {productRecordMutex}; - - // Merge volume times into map - std::transform(volumeTimes.cbegin(), - volumeTimes.cend(), - std::inserter(productRecordMap, productRecordMap.begin()), - [](const std::chrono::system_clock::time_point& time) - { - return std::pair>( - time, std::weak_ptr {}); - }); + productDatastore_.PopulateLevel3ProductTimes( + level3ProviderManager, product, time, update); } std::tuple> - records {}; - std::vector recordPtrs {}; - types::RadarProductLoadStatus status { + records {}; + types::RadarProductLoadStatus status { types::RadarProductLoadStatus::ListingProducts}; - std::size_t recordPtrCount = 0u; - std::size_t recordCount = 0u; + std::size_t recordCount = 0u; // Ensure Level 2 product records are updated if (!AreLevel2ProductTimesPopulated(time)) @@ -1249,69 +958,40 @@ RadarProductManagerImpl::GetLevel2ProductRecords( // Advance to loading product status = types::RadarProductLoadStatus::LoadingProduct; - { - std::shared_lock lock {level2ProductRecordMutex_}; - - if (!level2ProductRecords_.empty() && - time == std::chrono::system_clock::time_point {}) - { - // If a default-initialized time point is given, return the latest - // record - recordPtrs.push_back(&(*level2ProductRecords_.rbegin())); - } - else - { - // Get the requested record - auto recordIt = - scwx::util::GetBoundedElementIterator(level2ProductRecords_, time); + const auto recordEntries = productDatastore_.FindLevel2RecordEntries(time); - if (recordIt != level2ProductRecords_.cend()) - { - recordPtrs.push_back(&(*(recordIt))); + std::size_t validEntryCount = 0u; - // The requested time may be in the previous record, so get that too - if (recordIt != level2ProductRecords_.cbegin()) - { - recordPtrs.push_back(&(*(--recordIt))); - } - } - } - } - - // For each record pointer - for (auto& recordPtr : recordPtrs) + // For each record entry + for (const auto& recordEntry : recordEntries) { std::shared_ptr record {nullptr}; std::chrono::system_clock::time_point recordTime {time}; + bool entryValid = true; + + using namespace std::chrono_literals; + + // Don't check for an exact time match for level 2 products + recordTime = recordEntry.time; - if (recordPtr != nullptr) + if ( + // For latest data, ensure it is from the last 24 hours + (time == std::chrono::system_clock::time_point {} && + (recordTime > scwx::util::time::now() - 24h || recordTime == time)) || + // For time queries, ensure data is within 24 hours of the request + (time != std::chrono::system_clock::time_point {} && + std::chrono::abs(recordTime - time) < 24h)) { - using namespace std::chrono_literals; - - // Don't check for an exact time match for level 2 products - recordTime = recordPtr->first; - - if ( - // For latest data, ensure it is from the last 24 hours - (time == std::chrono::system_clock::time_point {} && - (recordTime > scwx::util::time::now() - 24h || - recordTime == time)) || - // For time queries, ensure data is within 24 hours of the request - (time != std::chrono::system_clock::time_point {} && - std::chrono::abs(recordTime - time) < 24h)) - { - record = recordPtr->second.lock(); - ++recordPtrCount; - } - else - { - // Reset the record - recordPtr = nullptr; - recordTime = time; - } + record = recordEntry.record.lock(); + ++validEntryCount; + } + else + { + entryValid = false; + recordTime = time; } - if (recordPtr != nullptr && record == nullptr && + if (entryValid && record == nullptr && recordTime != std::chrono::system_clock::time_point {}) { // Product is expired, reload it @@ -1343,12 +1023,12 @@ RadarProductManagerImpl::GetLevel2ProductRecords( } } - if (recordPtrCount == 0) + if (validEntryCount == 0) { // If all records are empty, the product is not available status = types::RadarProductLoadStatus::ProductNotAvailable; } - else if (recordCount == recordPtrCount) + else if (recordCount == validEntryCount) { // If all records were populated, the product has been loaded status = types::RadarProductLoadStatus::ProductLoaded; @@ -1364,10 +1044,10 @@ RadarProductManagerImpl::GetLevel3ProductRecord( const std::string& product, std::chrono::system_clock::time_point time) { std::shared_ptr record {nullptr}; - RadarProductRecordMap::const_pointer recordPtr {nullptr}; std::chrono::system_clock::time_point recordTime {time}; types::RadarProductLoadStatus status { types::RadarProductLoadStatus::ListingProducts}; + bool recordEntryValid = false; // Ensure Level 3 product records are updated if (!AreLevel3ProductTimesPopulated(product, time)) @@ -1399,33 +1079,17 @@ RadarProductManagerImpl::GetLevel3ProductRecord( // Advance to loading product status = types::RadarProductLoadStatus::LoadingProduct; - std::unique_lock lock {level3ProductRecordMutex_}; - - auto it = level3ProductRecordsMap_.find(product); - - if (it != level3ProductRecordsMap_.cend() && !it->second.empty()) - { - if (time == std::chrono::system_clock::time_point {}) - { - // If a default-initialized time point is given, return the latest - // record - recordPtr = &(*it->second.rbegin()); - } - else - { - recordPtr = scwx::util::GetBoundedElementPointer(it->second, time); - } - } - - // Lock is no longer needed - lock.unlock(); + const auto recordEntry = + productDatastore_.FindLevel3RecordEntry(product, time); - if (recordPtr != nullptr) + if (recordEntry.has_value()) { using namespace std::chrono_literals; + recordEntryValid = true; + // Don't check for an exact time match for level 3 products - recordTime = recordPtr->first; + recordTime = recordEntry->time; if ( // For latest data, ensure it is from the last 24 hours @@ -1435,17 +1099,16 @@ RadarProductManagerImpl::GetLevel3ProductRecord( (time != std::chrono::system_clock::time_point {} && std::chrono::abs(recordTime - time) < 24h)) { - record = recordPtr->second.lock(); + record = recordEntry->record.lock(); } else { - // Reset the record - recordPtr = nullptr; - recordTime = time; + recordEntryValid = false; + recordTime = time; } } - if (recordPtr != nullptr && record == nullptr && + if (recordEntryValid && record == nullptr && recordTime != std::chrono::system_clock::time_point {}) { // Product is expired, reload it @@ -1469,7 +1132,7 @@ RadarProductManagerImpl::GetLevel3ProductRecord( // Status is already set to LoadingProduct } - if (recordPtr == nullptr) + if (!recordEntryValid) { // If the record is empty, the product is not available status = types::RadarProductLoadStatus::ProductNotAvailable; @@ -1483,101 +1146,6 @@ RadarProductManagerImpl::GetLevel3ProductRecord( return {record, recordTime, status}; } -std::shared_ptr -RadarProductManagerImpl::StoreRadarProductRecord( - std::shared_ptr record) -{ - logger_->trace("StoreRadarProductRecord()"); - - std::shared_ptr storedRecord = nullptr; - - auto timeInSeconds = - std::chrono::time_point_cast(record->time()); - - if (record->radar_product_group() == common::RadarProductGroup::Level2) - { - std::unique_lock lock {level2ProductRecordMutex_}; - - auto it = level2ProductRecords_.find(timeInSeconds); - if (it != level2ProductRecords_.cend()) - { - storedRecord = it->second.lock(); - - if (storedRecord != nullptr) - { - logger_->debug( - "Level 2 product previously loaded, loading from cache"); - } - } - - if (storedRecord == nullptr) - { - storedRecord = record; - level2ProductRecords_[timeInSeconds] = record; - } - - UpdateRecentRecords(level2ProductRecentRecords_, storedRecord); - } - else if (record->radar_product_group() == common::RadarProductGroup::Level3) - { - std::unique_lock lock {level3ProductRecordMutex_}; - - auto& productMap = level3ProductRecordsMap_[record->radar_product()]; - - auto it = productMap.find(timeInSeconds); - if (it != productMap.cend()) - { - storedRecord = it->second.lock(); - - if (storedRecord != nullptr) - { - logger_->debug( - "Level 3 product previously loaded, loading from cache"); - } - } - - if (storedRecord == nullptr) - { - storedRecord = record; - productMap[timeInSeconds] = record; - } - - UpdateRecentRecords( - level3ProductRecentRecordsMap_[record->radar_product()], storedRecord); - } - - return storedRecord; -} - -void RadarProductManagerImpl::UpdateRecentRecords( - RadarProductRecordList& recentList, - std::shared_ptr record) -{ - const std::size_t recentListMaxSize {cacheLimit_}; - bool iteratorErased = false; - - auto it = std::find(recentList.cbegin(), recentList.cend(), record); - if (it != recentList.cbegin() && it != recentList.cend()) - { - // If the record exists beyond the front of the list, remove it - recentList.erase(it); - iteratorErased = true; - } - - if (iteratorErased || recentList.size() == 0 || it != recentList.cbegin()) - { - // Add the record to the front of the list, unless it's already there - recentList.push_front(record); - } - - while (recentList.size() > recentListMaxSize) - { - // Remove from the end of the list while it's too big - recentList.pop_back(); - } -} - std::tuple, float, std::vector, @@ -1766,7 +1334,7 @@ std::vector RadarProductManager::GetLevel3Products() void RadarProductManager::SetCacheLimit(size_t cacheLimit) { - p->cacheLimit_ = std::max(cacheLimit, 6u); + p->productDatastore_.SetCacheLimit(cacheLimit); } void RadarProductManager::UpdateAvailableProducts() diff --git a/test/source/scwx/qt/manager/product_datastore.test.cpp b/test/source/scwx/qt/manager/product_datastore.test.cpp new file mode 100644 index 000000000..bae4c8894 --- /dev/null +++ b/test/source/scwx/qt/manager/product_datastore.test.cpp @@ -0,0 +1,180 @@ +#include +#include +#include + +#include + +#include + +namespace scwx::qt::manager +{ + +namespace +{ + +std::shared_ptr +CreateLevel2Record(std::chrono::system_clock::time_point time) +{ + const std::string filename = std::string(SCWX_TEST_DATA_DIR) + + "/nexrad/level2/Level2_KLSX_20210527_1757.ar2v"; + + auto nexradFile = wsr88d::NexradFileFactory::Create(filename); + auto record = types::RadarProductRecord::Create(nexradFile); + record->set_time(time); + return record; +} + +std::shared_ptr +CreateLevel3Record(std::chrono::system_clock::time_point time) +{ + const std::string filename = + std::string(SCWX_TEST_DATA_DIR) + + "/nexrad/level3/KLSX_SDUS23_N2QLSX_202112110250"; + + auto nexradFile = wsr88d::NexradFileFactory::Create(filename); + auto record = types::RadarProductRecord::Create(nexradFile); + record->set_time(time); + return record; +} + +} // namespace + +TEST(ProductDatastore, SetCacheLimitMinimum) +{ + ProductDatastore datastore {}; + + datastore.SetCacheLimit(1u); + EXPECT_EQ(datastore.cache_limit(), 6u); + + datastore.SetCacheLimit(12u); + EXPECT_EQ(datastore.cache_limit(), 12u); +} + +TEST(ProductDatastore, StoreLevel2Dedup) +{ + ProductDatastore datastore {}; + + using namespace std::chrono_literals; + + const auto time = std::chrono::floor( + std::chrono::system_clock::now()); + + auto record = CreateLevel2Record(time); + ASSERT_NE(record, nullptr); + + const auto storedRecord = datastore.Store(record); + const auto cachedRecord = datastore.Store(record); + + EXPECT_EQ(storedRecord, cachedRecord); + EXPECT_EQ(storedRecord, record); +} + +TEST(ProductDatastore, StoreLevel3Dedup) +{ + ProductDatastore datastore {}; + + using namespace std::chrono_literals; + + const auto time = std::chrono::floor( + std::chrono::system_clock::now()); + + auto record = CreateLevel3Record(time); + ASSERT_NE(record, nullptr); + + const auto storedRecord = datastore.Store(record); + const auto cachedRecord = datastore.Store(record); + + EXPECT_EQ(storedRecord, cachedRecord); + EXPECT_EQ(storedRecord, record); +} + +TEST(ProductDatastore, FindLevel2RecordEntries) +{ + ProductDatastore datastore {}; + + using namespace std::chrono_literals; + + const auto earlierTime = std::chrono::floor( + std::chrono::system_clock::now() - 10min); + const auto laterTime = earlierTime + 5min; + + datastore.Store(CreateLevel2Record(earlierTime)); + datastore.Store(CreateLevel2Record(laterTime)); + + const auto latestEntries = datastore.FindLevel2RecordEntries( + std::chrono::system_clock::time_point {}); + ASSERT_EQ(latestEntries.size(), 1u); + EXPECT_EQ(latestEntries.front().time, laterTime); + + const auto boundedEntries = + datastore.FindLevel2RecordEntries(laterTime + 1min); + ASSERT_EQ(boundedEntries.size(), 2u); + EXPECT_EQ(boundedEntries.back().time, earlierTime); + EXPECT_EQ(boundedEntries.front().time, laterTime); +} + +TEST(ProductDatastore, FindLevel3RecordEntry) +{ + ProductDatastore datastore {}; + + using namespace std::chrono_literals; + + const auto time = std::chrono::floor( + std::chrono::system_clock::now()); + + auto record = CreateLevel3Record(time); + ASSERT_NE(record, nullptr); + + datastore.Store(record); + + const auto product = record->radar_product(); + + const auto latestEntry = datastore.FindLevel3RecordEntry( + product, std::chrono::system_clock::time_point {}); + ASSERT_TRUE(latestEntry.has_value()); + EXPECT_EQ(latestEntry->time, time); + + const auto exactEntry = datastore.FindLevel3RecordEntry(product, time); + ASSERT_TRUE(exactEntry.has_value()); + EXPECT_EQ(exactEntry->time, time); +} + +TEST(ProductDatastore, GetCachedNexradFile) +{ + ProductDatastore datastore {}; + + using namespace std::chrono_literals; + + const auto time = std::chrono::floor( + std::chrono::system_clock::now()); + + auto record = CreateLevel2Record(time); + ASSERT_NE(record, nullptr); + + datastore.Store(record); + + const auto cachedFile = datastore.GetCachedNexradFile( + common::RadarProductGroup::Level2, {}, time); + EXPECT_EQ(cachedFile, record->nexrad_file()); +} + +TEST(ProductDatastore, GetCachedNexradFileSubsecondLookup) +{ + ProductDatastore datastore {}; + + using namespace std::chrono_literals; + + const auto time = std::chrono::floor( + std::chrono::system_clock::now()); + + auto record = CreateLevel2Record(time); + ASSERT_NE(record, nullptr); + + datastore.Store(record); + + const auto cachedFile = datastore.GetCachedNexradFile( + common::RadarProductGroup::Level2, {}, time + 500ms); + EXPECT_EQ(cachedFile, record->nexrad_file()); +} + +} // namespace scwx::qt::manager diff --git a/test/test.cmake b/test/test.cmake index a04c84d24..7e7ad4ed6 100644 --- a/test/test.cmake +++ b/test/test.cmake @@ -36,7 +36,8 @@ set(SRC_QT_CONFIG_TESTS source/scwx/qt/config/county_database.test.cpp set(SRC_QT_MAIN_TESTS source/scwx/qt/main/application_paths.test.cpp source/scwx/qt/main/program_options.test.cpp source/scwx/qt/main/theme.test.cpp) -set(SRC_QT_MANAGER_TESTS source/scwx/qt/manager/radar_product_manager.test.cpp +set(SRC_QT_MANAGER_TESTS source/scwx/qt/manager/product_datastore.test.cpp + source/scwx/qt/manager/radar_product_manager.test.cpp source/scwx/qt/manager/settings_manager.test.cpp source/scwx/qt/manager/update_manager.test.cpp) set(SRC_QT_MAP_TESTS source/scwx/qt/map/map_annotation_layer.test.cpp