6#include <userver/cache/base_postgres_cache_fwd.hpp>
12#include <unordered_map>
14#include <fmt/format.h>
16#include <userver/cache/cache_statistics.hpp>
17#include <userver/cache/caching_component_base.hpp>
18#include <userver/components/component_config.hpp>
19#include <userver/components/component_context.hpp>
21#include <userver/storages/postgres/cluster.hpp>
22#include <userver/storages/postgres/component.hpp>
23#include <userver/storages/postgres/io/chrono.hpp>
25#include <userver/compiler/demangle.hpp>
26#include <userver/logging/log.hpp>
27#include <userver/tracing/span.hpp>
28#include <userver/utils/assert.hpp>
29#include <userver/utils/cpu_relax.hpp>
30#include <userver/utils/meta.hpp>
31#include <userver/utils/void_t.hpp>
32#include <userver/yaml_config/merge_schemas.hpp>
34USERVER_NAMESPACE_BEGIN
128namespace pg_cache::detail {
131using ValueType =
typename T::ValueType;
133inline constexpr bool kHasValueType = meta::IsDetected<ValueType, T>;
136using RawValueTypeImpl =
typename T::RawValueType;
138inline constexpr bool kHasRawValueType = meta::IsDetected<RawValueTypeImpl, T>;
140using RawValueType = meta::DetectedOr<ValueType<T>, RawValueTypeImpl, T>;
142template <
typename PostgreCachePolicy>
143auto ExtractValue(RawValueType<PostgreCachePolicy>&& raw) {
144 if constexpr (kHasRawValueType<PostgreCachePolicy>) {
145 return Convert(std::move(raw),
formats::
parse::
To<ValueType<PostgreCachePolicy>>());
147 return std::move(raw);
153using HasNameImpl = std::enable_if_t<!std::string_view{T::kName}.empty()>;
155inline constexpr bool kHasName = meta::IsDetected<HasNameImpl, T>;
159using HasQueryImpl =
decltype(T::kQuery);
161inline constexpr bool kHasQuery = meta::IsDetected<HasQueryImpl, T>;
165using HasGetQueryImpl =
decltype(T::GetQuery());
167inline constexpr bool kHasGetQuery = meta::IsDetected<HasGetQueryImpl, T>;
171using HasWhere =
decltype(T::kWhere);
173inline constexpr bool kHasWhere = meta::IsDetected<HasWhere, T>;
177using HasOrderBy =
decltype(T::kOrderBy);
179inline constexpr bool kHasOrderBy = meta::IsDetected<HasOrderBy, T>;
183using HasUpdatedField =
decltype(T::kUpdatedField);
185inline constexpr bool kHasUpdatedField = meta::IsDetected<HasUpdatedField, T>;
188using WantIncrementalUpdates = std::enable_if_t<!std::string_view{T::kUpdatedField}.empty()>;
190inline constexpr bool kWantIncrementalUpdates = meta::IsDetected<WantIncrementalUpdates, T>;
194using KeyMemberTypeImpl = std::decay_t<std::invoke_result_t<
decltype(T::kKeyMember), ValueType<T>>>;
196inline constexpr bool kHasKeyMember = meta::IsDetected<KeyMemberTypeImpl, T>;
198using KeyMemberType = meta::DetectedType<KeyMemberTypeImpl, T>;
202using SizeMethodInvokeResultImpl =
decltype(std::declval<T>().size());
204inline constexpr bool kHasSizeMethod = meta::IsDetected<SizeMethodInvokeResultImpl, T> &&
205 std::is_convertible_v<SizeMethodInvokeResultImpl<T>, std::size_t>;
209using InsertOrAssignMethodInvokeResultImpl =
decltype(std::declval<
typename T::CacheContainer>().insert_or_assign(
210 std::declval<KeyMemberTypeImpl<T>>(),
211 std::declval<ValueType<T>>()
214inline constexpr bool kHasInsertOrAssignMethod = meta::IsDetected<InsertOrAssignMethodInvokeResultImpl, T>;
218using CacheInsertOrAssignFunctionInvokeResultImpl =
decltype(CacheInsertOrAssign(
219 std::declval<
typename T::CacheContainer&>(),
220 std::declval<ValueType<T>>(),
221 std::declval<KeyMemberTypeImpl<T>>()
224inline constexpr bool kHasCacheInsertOrAssignFunction =
225 meta::IsDetected<CacheInsertOrAssignFunctionInvokeResultImpl, T>;
228template <
typename T,
typename = USERVER_NAMESPACE::
utils::void_t<>>
229struct DataCacheContainer {
231 meta::kIsStdHashable<KeyMemberType<T>>,
232 "With default CacheContainer, key type must be std::hash-able"
235 using type = std::unordered_map<KeyMemberType<T>, ValueType<T>>;
239struct DataCacheContainer<T, USERVER_NAMESPACE::
utils::void_t<
typename T::CacheContainer>> {
240 static_assert(kHasSizeMethod<
typename T::CacheContainer>,
"Custom CacheContainer must provide `size` method");
242 kHasInsertOrAssignMethod<T> || kHasCacheInsertOrAssignFunction<T>,
243 "Custom CacheContainer must provide `insert_or_assign` method similar to std::unordered_map's "
244 "one or CacheInsertOrAssign function"
247 using type =
typename T::CacheContainer;
251using DataCacheContainerType =
typename DataCacheContainer<T>::type;
256inline constexpr bool kIsContainerCopiedByElement =
257 meta::kIsInstantiationOf<std::unordered_map, T> || meta::kIsInstantiationOf<std::map, T>;
261CopyContainer(
const T& container, [[maybe_unused]] std::size_t cpu_relax_iterations,
tracing::ScopeTime& scope) {
262 if constexpr (kIsContainerCopiedByElement<T>) {
263 auto copy = std::make_unique<T>();
264 if constexpr (meta::kIsReservable<T>) {
265 copy->reserve(container.size());
269 for (
const auto& kv : container) {
275 return std::make_unique<T>(container);
279template <
typename Container,
typename Value,
typename KeyMember,
typename... Args>
280void CacheInsertOrAssign(Container& container, Value&& value,
const KeyMember& key_member, Args&&... ) {
282 static_assert(
sizeof...(Args) == 0);
284 auto key = std::invoke(key_member, value);
285 container.insert_or_assign(std::move(key), std::forward<Value>(value));
289using HasOnWritesDoneImpl =
decltype(std::declval<T&>().OnWritesDone());
292void OnWritesDone(T& container) {
293 if constexpr (meta::IsDetected<HasOnWritesDoneImpl, T>) {
294 container.OnWritesDone();
299using HasCustomUpdatedImpl =
decltype(T::GetLastKnownUpdated(std::declval<DataCacheContainerType<T>>()));
302inline constexpr bool kHasCustomUpdated = meta::IsDetected<HasCustomUpdatedImpl, T>;
305using UpdatedFieldTypeImpl =
typename T::UpdatedFieldType;
307inline constexpr bool kHasUpdatedFieldType = meta::IsDetected<UpdatedFieldTypeImpl, T>;
309using UpdatedFieldType = meta::DetectedOr<storages::
postgres::TimePointTz, UpdatedFieldTypeImpl, T>;
312constexpr bool CheckUpdatedFieldType() {
313 if constexpr (kHasUpdatedFieldType<T>) {
314#if USERVER_POSTGRES_ENABLE_LEGACY_TIMESTAMP
316 std::is_same_v<
typename T::UpdatedFieldType, storages::postgres::TimePointTz> ||
317 std::is_same_v<
typename T::UpdatedFieldType, storages::postgres::TimePointWithoutTz> ||
318 std::is_same_v<
typename T::UpdatedFieldType, storages::postgres::TimePoint> || kHasCustomUpdated<T>,
319 "Invalid UpdatedFieldType, must be either TimePointTz or "
321 "or (legacy) system_clock::time_point"
325 std::is_same_v<
typename T::UpdatedFieldType, storages::
postgres::TimePointTz> ||
326 std::is_same_v<
typename T::UpdatedFieldType, storages::
postgres::TimePointWithoutTz> ||
327 kHasCustomUpdated<T>,
328 "Invalid UpdatedFieldType, must be either TimePointTz or "
334 !kWantIncrementalUpdates<T>,
335 "UpdatedFieldType must be explicitly specified when using "
336 "incremental updates"
344using HasClusterHostTypeImpl =
decltype(T::kClusterHostType);
347constexpr storages::
postgres::ClusterHostTypeFlags ClusterHostType() {
348 if constexpr (meta::IsDetected<HasClusterHostTypeImpl, T>) {
349 return T::kClusterHostType;
357using HasMayReturnNull =
decltype(T::kMayReturnNull);
360constexpr bool MayReturnNull() {
361 if constexpr (meta::IsDetected<HasMayReturnNull, T>) {
362 return T::kMayReturnNull;
368template <
typename PostgreCachePolicy>
369struct PolicyChecker {
371 static_assert(kHasName<PostgreCachePolicy>,
"The PosgreSQL cache policy must contain a static member `kName`");
372 static_assert(kHasValueType<PostgreCachePolicy>,
"The PosgreSQL cache policy must define a type alias `ValueType`");
374 kHasKeyMember<PostgreCachePolicy>,
375 "The PostgreSQL cache policy must contain a static member `kKeyMember` "
376 "with a pointer to a data or a function member with the object's key"
379 kHasQuery<PostgreCachePolicy> || kHasGetQuery<PostgreCachePolicy>,
380 "The PosgreSQL cache policy must contain a static data member "
381 "`kQuery` with a select statement or a static member function "
382 "`GetQuery` returning the query"
385 !(kHasQuery<PostgreCachePolicy> && kHasGetQuery<PostgreCachePolicy>),
386 "The PosgreSQL cache policy must define `kQuery` or "
387 "`GetQuery`, not both"
390 kHasUpdatedField<PostgreCachePolicy>,
391 "The PosgreSQL cache policy must contain a static member "
392 "`kUpdatedField`. If you don't want to use incremental updates, "
393 "please set its value to `nullptr`"
395 static_assert(CheckUpdatedFieldType<PostgreCachePolicy>());
398 ClusterHostType<PostgreCachePolicy>() & storages::
postgres::kClusterHostRolesMask,
399 "Cluster host role must be specified for caching component, "
400 "please be more specific"
403 static storages::
postgres::Query GetQuery() {
404 if constexpr (kHasGetQuery<PostgreCachePolicy>) {
405 return PostgreCachePolicy::GetQuery();
407 return PostgreCachePolicy::kQuery;
414inline constexpr std::chrono::minutes kDefaultFullUpdateTimeout{1};
415inline constexpr std::chrono::seconds kDefaultIncrementalUpdateTimeout{1};
416inline constexpr std::chrono::milliseconds kStatementTimeoutOff{0};
417inline constexpr std::chrono::milliseconds kCpuRelaxThreshold{10};
418inline constexpr std::chrono::milliseconds kCpuRelaxInterval{2};
420inline constexpr std::string_view kCopyStage =
"copy_data";
421inline constexpr std::string_view kFetchStage =
"fetch";
422inline constexpr std::string_view kParseStage =
"parse";
424inline constexpr std::size_t kDefaultChunkSize = 1000;
432template <
typename PostgreCachePolicy>
433class PostgreCache
final :
public pg_cache::detail::PolicyChecker<PostgreCachePolicy>::BaseType {
436 using PolicyType = PostgreCachePolicy;
437 using ValueType = pg_cache::detail::ValueType<PolicyType>;
438 using RawValueType = pg_cache::detail::RawValueType<PolicyType>;
439 using DataType = pg_cache::detail::DataCacheContainerType<PolicyType>;
440 using PolicyCheckerType = pg_cache::detail::PolicyChecker<PostgreCachePolicy>;
441 using UpdatedFieldType = pg_cache::detail::UpdatedFieldType<PostgreCachePolicy>;
442 using BaseType =
typename PolicyCheckerType::BaseType;
445 constexpr static bool kIncrementalUpdates = pg_cache::detail::kWantIncrementalUpdates<PolicyType>;
446 constexpr static auto kClusterHostTypeFlags = pg_cache::detail::ClusterHostType<PolicyType>();
447 constexpr static auto kName = PolicyType::kName;
449 PostgreCache(
const ComponentConfig&,
const ComponentContext&);
450 ~PostgreCache()
override;
452 static yaml_config::Schema GetStaticConfigSchema();
455 using CachedData = std::unique_ptr<DataType>;
457 UpdatedFieldType GetLastUpdated(std::chrono::system_clock::time_point last_update,
const DataType& cache)
const;
461 const std::chrono::system_clock::time_point& last_update,
462 const std::chrono::system_clock::time_point& now,
463 cache::UpdateStatisticsScope& stats_scope
466 bool MayReturnNull()
const override;
471 CachedData& data_cache,
472 cache::UpdateStatisticsScope& stats_scope,
476 static storages::
postgres::Query GetAllQuery();
477 static storages::
postgres::Query GetDeltaQuery();
478 static std::string GetWhereClause();
479 static std::string GetDeltaWhereClause();
480 static std::string GetOrderByClause();
482 std::chrono::milliseconds ParseCorrection(
const ComponentConfig& config);
484 std::vector<storages::
postgres::ClusterPtr> clusters_;
486 const std::chrono::system_clock::duration correction_;
487 const std::chrono::milliseconds full_update_timeout_;
488 const std::chrono::milliseconds incremental_update_timeout_;
489 const std::size_t chunk_size_;
490 std::size_t cpu_relax_iterations_parse_{0};
491 std::size_t cpu_relax_iterations_copy_{0};
433class PostgreCache
final :
public pg_cache::detail::PolicyChecker<PostgreCachePolicy>::BaseType {
…};
494template <
typename PostgreCachePolicy>
495inline constexpr bool kHasValidate<PostgreCache<PostgreCachePolicy>> =
true;
497template <
typename PostgreCachePolicy>
498PostgreCache<PostgreCachePolicy>::PostgreCache(
const ComponentConfig& config,
const ComponentContext& context)
499 : BaseType{config, context},
500 correction_{ParseCorrection(config)},
501 full_update_timeout_{
502 config
["full-update-op-timeout"].As<std
::chrono
::milliseconds
>(pg_cache::detail::kDefaultFullUpdateTimeout
)},
503 incremental_update_timeout_{config
["incremental-update-op-timeout"].As<std
::chrono
::milliseconds
>(
504 pg_cache::detail::kDefaultIncrementalUpdateTimeout
506 chunk_size_{config
["chunk-size"].As<size_t
>(pg_cache::detail::kDefaultChunkSize
)} {
509 "Either set 'chunk-size' to 0, or enable PostgreSQL portals by building "
510 "the framework with CMake option USERVER_FEATURE_PATCH_LIBPQ set to ON."
514 throw std::logic_error(
515 "Incremental update support is requested in config but no update field "
516 "name is specified in traits of '" +
520 if (correction_.count() < 0) {
521 throw std::logic_error(
522 "Refusing to set forward (negative) update correction requested in "
528 const auto pg_alias = config
["pgcomponent"].As<std
::string
>("");
529 if (pg_alias.empty()) {
534 clusters_.resize(shard_count);
535 for (size_t i = 0; i < shard_count; ++i) {
542 this->StartPeriodicUpdates();
545template <
typename PostgreCachePolicy>
546PostgreCache<PostgreCachePolicy>::~PostgreCache() {
547 this->StopPeriodicUpdates();
550template <
typename PostgreCachePolicy>
551std::string PostgreCache<PostgreCachePolicy>::GetWhereClause() {
552 if constexpr (pg_cache::detail::kHasWhere<PostgreCachePolicy>) {
553 return fmt::format(FMT_COMPILE(
"where {}"), PostgreCachePolicy::kWhere);
559template <
typename PostgreCachePolicy>
560std::string PostgreCache<PostgreCachePolicy>::GetDeltaWhereClause() {
561 if constexpr (pg_cache::detail::kHasWhere<PostgreCachePolicy>) {
563 FMT_COMPILE(
"where ({}) and {} >= $1"), PostgreCachePolicy::kWhere, PostgreCachePolicy::kUpdatedField
566 return fmt::format(FMT_COMPILE(
"where {} >= $1"), PostgreCachePolicy::kUpdatedField);
570template <
typename PostgreCachePolicy>
571std::string PostgreCache<PostgreCachePolicy>::GetOrderByClause() {
572 if constexpr (pg_cache::detail::kHasOrderBy<PostgreCachePolicy>) {
573 return fmt::format(FMT_COMPILE(
"order by {}"), PostgreCachePolicy::kOrderBy);
579template <
typename PostgreCachePolicy>
580storages::
postgres::Query PostgreCache<PostgreCachePolicy>::GetAllQuery() {
581 const storages::
postgres::Query query = PolicyCheckerType::GetQuery();
582 return fmt::format(
"{} {} {}", query
.GetStatementView(), GetWhereClause(), GetOrderByClause());
585template <
typename PostgreCachePolicy>
586storages::
postgres::Query PostgreCache<PostgreCachePolicy>::GetDeltaQuery() {
587 if constexpr (kIncrementalUpdates) {
588 const storages::
postgres::Query query = PolicyCheckerType::GetQuery();
590 fmt::format(
"{} {} {}", query
.GetStatementView(), GetDeltaWhereClause(), GetOrderByClause()),
594 return GetAllQuery();
598template <
typename PostgreCachePolicy>
599std::chrono::milliseconds PostgreCache<PostgreCachePolicy>::ParseCorrection(
const ComponentConfig& config) {
600 static constexpr std::string_view kUpdateCorrection =
"update-correction";
601 if (pg_cache::detail::kHasCustomUpdated<PostgreCachePolicy> ||
603 return config
[kUpdateCorrection
].As<std
::chrono
::milliseconds
>(0
);
605 return config
[kUpdateCorrection
].As<std
::chrono
::milliseconds
>();
609template <
typename PostgreCachePolicy>
610typename PostgreCache<PostgreCachePolicy>::UpdatedFieldType PostgreCache<PostgreCachePolicy>::GetLastUpdated(
611 [[maybe_unused]] std::chrono::system_clock::time_point last_update,
612 const DataType& cache
614 if constexpr (pg_cache::detail::kHasCustomUpdated<PostgreCachePolicy>) {
615 return PostgreCachePolicy::GetLastKnownUpdated(cache);
617 return UpdatedFieldType{last_update - correction_};
621template <
typename PostgreCachePolicy>
622void PostgreCache<PostgreCachePolicy>::Update(
624 const std::chrono::system_clock::time_point& last_update,
625 const std::chrono::system_clock::time_point& ,
626 cache::UpdateStatisticsScope& stats_scope
629 if constexpr (!kIncrementalUpdates) {
633 const std::chrono::milliseconds timeout =
638 auto data_cache = GetDataSnapshot(type, scope);
639 [[maybe_unused]]
const auto old_size = data_cache->size();
641 scope.Reset(std::string{pg_cache::detail::kFetchStage});
645 for (
auto& cluster : clusters_) {
646 if (chunk_size_ > 0) {
647 auto trx = cluster->Begin(
648 kClusterHostTypeFlags,
650 pg::
CommandControl{timeout, pg_cache::detail::kStatementTimeoutOff}
652 auto portal = trx.MakePortal(query, GetLastUpdated(last_update, *data_cache));
654 scope.Reset(std::string{pg_cache::detail::kFetchStage});
655 auto res = portal.Fetch(chunk_size_);
658 scope.Reset(std::string{pg_cache::detail::kParseStage});
659 CacheResults(res, data_cache, stats_scope, scope);
660 changes += res.Size();
665 auto res = has_parameter ? cluster->Execute(
666 kClusterHostTypeFlags,
667 pg::
CommandControl{timeout, pg_cache::detail::kStatementTimeoutOff},
669 GetLastUpdated(last_update, *data_cache)
672 kClusterHostTypeFlags,
673 pg::
CommandControl{timeout, pg_cache::detail::kStatementTimeoutOff},
678 scope.Reset(std::string{pg_cache::detail::kParseStage});
679 CacheResults(res, data_cache, stats_scope, scope);
680 changes += res.Size();
686 if constexpr (pg_cache::detail::kIsContainerCopiedByElement<DataType>) {
688 const auto elapsed_copy = scope.ElapsedTotal(std::string{pg_cache::detail::kCopyStage});
689 if (elapsed_copy > pg_cache::detail::kCpuRelaxThreshold) {
690 cpu_relax_iterations_copy_ =
static_cast<std::size_t>(
691 static_cast<
double>(old_size) / (elapsed_copy / pg_cache::detail::kCpuRelaxInterval)
693 LOG_TRACE() <<
"Elapsed time for copying " << kName <<
" " << elapsed_copy.count() <<
" for " << changes
694 <<
" data items is over threshold. Will relax CPU every " << cpu_relax_iterations_parse_
701 const auto elapsed_parse = scope.ElapsedTotal(std::string{pg_cache::detail::kParseStage});
702 if (elapsed_parse > pg_cache::detail::kCpuRelaxThreshold) {
703 cpu_relax_iterations_parse_ =
static_cast<std::size_t>(
704 static_cast<
double>(changes) / (elapsed_parse / pg_cache::detail::kCpuRelaxInterval)
706 LOG_TRACE() <<
"Elapsed time for parsing " << kName <<
" " << elapsed_parse.count() <<
" for " << changes
707 <<
" data items is over threshold. Will relax CPU every " << cpu_relax_iterations_parse_
713 pg_cache::detail::OnWritesDone(*data_cache);
715 this->Set(std::move(data_cache));
721template <
typename PostgreCachePolicy>
722bool PostgreCache<PostgreCachePolicy>::MayReturnNull()
const {
723 return pg_cache::detail::MayReturnNull<PolicyType>();
726template <
typename PostgreCachePolicy>
727void PostgreCache<PostgreCachePolicy>::CacheResults(
729 CachedData& data_cache,
730 cache::UpdateStatisticsScope& stats_scope,
733 auto values = res.AsSetOf<RawValueType>(storages::
postgres::kRowTag);
735 for (
auto p = values.begin(); p != values.end(); ++p) {
738 using pg_cache::detail::CacheInsertOrAssign;
740 *data_cache, pg_cache::detail::ExtractValue<PostgreCachePolicy>(*p), PostgreCachePolicy::kKeyMember
742 }
catch (
const std::exception& e) {
744 LOG_ERROR() <<
"Error parsing data row in cache '" << kName <<
"' to '"
745 <<
compiler::GetTypeName<ValueType>() <<
"': " << e.what();
750template <
typename PostgreCachePolicy>
751typename PostgreCache<PostgreCachePolicy>::CachedData
754 auto data =
this->Get();
756 return pg_cache::detail::CopyContainer(*data, cpu_relax_iterations_copy_, scope);
759 return std::make_unique<DataType>();
764std::string GetPostgreCacheSchema();
768template <
typename PostgreCachePolicy>
769yaml_config::Schema PostgreCache<PostgreCachePolicy>::GetStaticConfigSchema() {
770 using ParentType =
typename pg_cache::detail::PolicyChecker<PostgreCachePolicy>::BaseType;
771 return yaml_config::MergeSchemas<ParentType>(impl::GetPostgreCacheSchema());
776namespace utils::impl::projected_set {
778template <
typename Set,
typename Value,
typename KeyMember>
779void CacheInsertOrAssign(Set& set, Value&& value,
const KeyMember& ) {
780 DoInsert(set, std::forward<Value>(value));