caching_component_base.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/cache/caching_component_base.hpp
/// @brief @copybrief components::CachingComponentBase
 
#include <memory>
#include <string>
#include <utility>
 
#include <fmt/format.h>
 
#include <userver/cache/cache_update_trait.hpp>
#include <userver/cache/data_provider.hpp>
#include <userver/cache/exceptions.hpp>
#include <userver/compiler/demangle.hpp>
#include <userver/components/component_base.hpp>
#include <userver/components/component_context.hpp>
#include <userver/components/component_fwd.hpp>
#include <userver/concurrent/async_event_channel.hpp>
#include <userver/dump/helpers.hpp>
#include <userver/dump/meta.hpp>
#include <userver/dump/operations.hpp>
#include <userver/engine/async.hpp>
#include <userver/rcu/rcu.hpp>
#include <userver/utils/assert.hpp>
#include <userver/utils/impl/wait_token_storage.hpp>
#include <userver/utils/meta.hpp>
#include <userver/utils/resource_scopes.hpp>
#include <userver/utils/shared_readable_ptr.hpp>
#include <userver/yaml_config/schema.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace components {
 
/// @ingroup userver_components userver_base_classes
///
/// @brief Base class for caching components
///
/// Provides facilities for creating periodically updated caches.
/// You need to override cache::CacheUpdateTrait::Update.
/// You can also override cache::CachingComponentBase::PreAssignCheck and set
/// has-pre-assign-check: true in the static config to enable check.
///
/// Caching components must be configured in service config (see options below)
/// and may be reconfigured dynamically via components::DynamicConfig.
///
/// @ref scripts/docs/en/userver/caches.md provide a more detailed introduction.
///
/// ## CachingComponentBase Dynamic config
/// * @ref USERVER_CACHES
/// * @ref USERVER_DUMPS
///
/// ## Static options of components::CachingComponentBase :
/// @include{doc} scripts/docs/en/components_schema/core/src/cache/caching_component_base.md
///
/// Options inherited from @ref components::ComponentBase :
/// @include{doc} scripts/docs/en/components_schema/core/src/components/impl/component_base.md
///
/// ### Update types
///  * `full-and-incremental`: both `update-interval` and `full-update-interval`
///    must be specified. Updates with UpdateType::kIncremental will be triggered
///    each `update-interval` (adjusted by jitter) unless `full-update-interval`
///    has passed and UpdateType::kFull is triggered.
///  * `only-full`: only `update-interval` must be specified. UpdateType::kFull
///    will be triggered each `update-interval` (adjusted by jitter).
///  * `only-incremental`: only `update-interval` must be specified. UpdateType::kFull is triggered
///    on the first update, afterwards UpdateType::kIncremental will be triggered
///    each `update-interval` (adjusted by jitter). Warning: use carefully.
///    If the cache loses any data, it is lost until service restart (in the worst case). If possible,
///    use `full-and-incremental` with rare full updates, and completely avoid `only-incremental`.
///    Also you have to explicitly remove outdated items from the cache container, otherwise
///    the cache might grow indefinitely and eventually will lead to OOM.
///    If not sure, just use `full-and-incremental`.
///
/// ### Avoiding memory leaks
///
/// If you don't implement the deletion of objects that are deleted from the data source and don't use full updates,
/// you may get an effective memory leak, because garbage objects will pile up in the cached data.
///
/// Calculation example:
/// * size of database: 1000 objects
/// * removal rate: 30 objects per minute (0.5 objects per second)
///
/// Let's say we allow 20% extra garbage objects in cache in addition to the actual objects from the database. In this
/// case we need:
///
/// full-update-interval = (size-of-database * 20% / removal-rate) = 400s
///
/// ### Dealing with nullptr data in CachingComponentBase
///
/// The cache can become `nullptr` through multiple ways:
///
/// * If the first cache update fails, and `first-update-fail-ok` config
///   option is set to `true` (otherwise the service shutdown at start)
/// * Through manually calling @ref Set with `nullptr` in @ref Update
/// * If `failed-updates-before-expiration` is set, and that many periodic
///   updates fail in a row
///
/// By default, the cache's user can expect that the pointer returned
/// from @ref Get will never be `nullptr`. If the cache for some reason is
/// in `nullptr` state, then @ref Get will throw. This is the safe default
/// behavior for most cases.
///
/// If all systems of a service are expected to work with a cache in `nullptr`
/// state, then such a cache should override `MayReturnNull` to return `true`.
/// It will also serve self-documentation purposes: if a cache defines
/// @ref MayReturnNull, then pointers returned from @ref Get should be checked
/// for `nullptr` before usage.
///
/// ### `first-update-mode` modes
///
/// Further customizes the behavior of @ref dump::Dumper "cache dumps".
///
/// Mode        | Description
/// ----------- | -----------
/// skip        | after successful load from dump, do nothing
/// required    | make a synchronous update of type `first-update-type`, stop the service on failure
/// best-effort | make a synchronous update of type `first-update-type`, keep working and use data from dump on failure
///
/// ### testsuite-force-periodic-update
///  use it to enable periodic cache update for a component in testsuite environment
///  where testsuite-periodic-update-enabled from TestsuiteSupport config is false
///
/// By default, update types are guessed based on update intervals presence.
/// If both `update-interval` and `full-update-interval` are present,
/// `full-and-incremental` types is assumed. Otherwise `only-full` is used.
///
/// @see `dump::Dumper` for more info on persistent cache dumps and
/// corresponding config options.
///
/// @see @ref scripts/docs/en/userver/caches.md. pytest_userver.client.Client.invalidate_caches()
/// for a function to force cache update from testsuite.
template <typename T>
// NOLINTNEXTLINE(fuchsia-multiple-inheritance)
class CachingComponentBase : public ComponentBase, public cache::DataProvider<T>, protected cache::CacheUpdateTrait {
public:
    CachingComponentBase(const ComponentConfig& config, const ComponentContext&);
    ~CachingComponentBase() override;
 
    using cache::CacheUpdateTrait::Name;
 
    using cache::CacheUpdateTrait::InvalidateAsync;
 
    using DataType = T;
 
    /// @return cache contents. May be `nullptr` if and only if `MayReturnNull`
    /// returns `true`.
    /// @throws cache::EmptyCacheError if the contents are `nullptr`, and
    /// `MayReturnNull` returns `false` (which is the default behavior).
    utils::SharedReadablePtr<T> Get() const final;
 
    /// @return cache contents. May be nullptr regardless of `MayReturnNull`.
    utils::SharedReadablePtr<T> GetUnsafe() const;
 
    /// Subscribes to cache updates using a member function. Also immediately
    /// invokes the function with the current cache contents.
    ///
    /// Further updates are delivered after @ref utils::ResourceScopeStorage::AfterConstruction, including those updates
    /// that arrived before it. Unsubscribe runs in @ref utils::ResourceScopeStorage::BeforeDestruction.
    ///
    /// @param scopes storage that owns the subscription lifetime. In a component constructor pass `context.Scopes()`
    /// or @ref components::GetResourceScopes.
    template <class Class>
    void UpdateAndListen(
        utils::ResourceScopeStorage& scopes,
        Class* obj,
        std::string name,
        void (Class::*func)(const std::shared_ptr<const T>&)
    );
 
    /// @overload
    /// @deprecated Use the overload that takes @ref utils::ResourceScopeStorage.
    ///
    /// Store the returned scope as a member and call `Unsubscribe` explicitly.
    template <class Class>
    concurrent::AsyncEventSubscriberScope UpdateAndListen(
        Class* obj,
        std::string name,
        void (Class::*func)(const std::shared_ptr<const T>&)
    );
 
    concurrent::AsyncEventChannel<const std::shared_ptr<const T>&>& GetEventChannel();
 
    static yaml_config::Schema GetStaticConfigSchema();
 
protected:
    /// @brief Sets the new value of cache. As a result, @ref Get member function starts returning the new value.
    ///
    /// Notifies subscribers after setting the new value, see @ref UpdateAndListen.
    /// Should only be called from @ref Update normally.
    ///
    /// @warning Do not forget to update @ref cache::UpdateStatisticsScope, otherwise the behavior is undefined.
    void Set(std::unique_ptr<const T> value_ptr);
 
    /// @overload
    void Set(T&& value);
 
    /// Attach the value of cache. As a result the `Get()` member function starts returning the value passed into
    /// this function after the `Update()` finishes. Does not take over into sole ownership. Do not use unless
    /// absolutely necessary. The object must be strictly thread-safe.
    ///
    /// @warning Do not forget to update @ref cache::UpdateStatisticsScope, otherwise
    /// the behavior is undefined.
    void Attach(const std::shared_ptr<const T>& value_ptr);
 
    /// @overload Set()
    template <typename... Args>
    void Emplace(Args&&... args);
 
    /// Clears the content of the cache by string a default constructed T.
    void Clear();
 
    /// Whether @ref Get is expected to return `nullptr`.
    virtual bool MayReturnNull() const;
 
    /// @{
    /// Override to use custom serialization for cache dumps
    virtual void WriteContents(dump::Writer& writer, const T& contents) const;
 
    virtual std::unique_ptr<const T> ReadContents(dump::Reader& reader) const;
    /// @}
 
    /// @brief If the option has-pre-assign-check is set true in static config,
    /// this function is called before assigning the new value to the cache
    /// @note old_value_ptr and new_value_ptr can be nullptr.
    virtual void PreAssignCheck(const T* old_value_ptr, const T* new_value_ptr) const;
 
private:
    void Cleanup() final;
 
    void MarkAsExpired() final;
 
    void GetAndWrite(dump::Writer& writer) const final;
    void ReadAndSet(dump::Reader& reader) final;
 
    std::shared_ptr<const T> TransformNewValue(std::unique_ptr<const T> new_value);
 
    rcu::Variable<std::shared_ptr<const T>> cache_;
    concurrent::AsyncEventChannel<const std::shared_ptr<const T>&> event_channel_;
    utils::impl::WaitTokenStorage wait_token_storage_;
};
 
template <typename T>
CachingComponentBase<T>::CachingComponentBase(const ComponentConfig& config, const ComponentContext& context)
    : ComponentBase(config, context),
      cache::CacheUpdateTrait(config, context),
      event_channel_(
          components::GetCurrentComponentName(context),
          [this](const auto& function) {
              const auto ptr = cache_.ReadCopy();
              if (ptr) {
                  function(ptr);
              }
          }
      )
{}
 
template <typename T>
CachingComponentBase<T>::~CachingComponentBase() {
    // Avoid a deadlock in WaitForAllTokens
    cache_.Assign(nullptr);
    // We must wait for destruction of all instances of T to finish, otherwise
    // it's UB if T's destructor accesses dependent components
    wait_token_storage_.WaitForAllTokens();
}
 
template <typename T>
utils::SharedReadablePtr<T> CachingComponentBase<T>::Get() const {
    auto ptr = GetUnsafe();
    if (!ptr && !MayReturnNull()) {
        throw cache::EmptyCacheError(Name());
    }
    return ptr;
}
 
template <typename T>
template <typename Class>
void CachingComponentBase<T>::UpdateAndListen(
    utils::ResourceScopeStorage& scopes,
    Class* obj,
    std::string name,
    void (Class::*func)(const std::shared_ptr<const T>&)
) {
    event_channel_.DoUpdateAndListenScoped(scopes, obj, std::move(name), func, [this, obj, func] {
        auto ptr = Get();  // TODO: extra ref
        (obj->*func)(ptr);
    });
}
 
template <typename T>
template <typename Class>
concurrent::AsyncEventSubscriberScope CachingComponentBase<
    T>::UpdateAndListen(Class* obj, std::string name, void (Class::*func)(const std::shared_ptr<const T>&)) {
    return event_channel_.DoUpdateAndListen(obj, std::move(name), func, [&] {
        auto ptr = Get();  // TODO: extra ref
        (obj->*func)(ptr);
    });
}
 
template <typename T>
concurrent::AsyncEventChannel<const std::shared_ptr<const T>&>& CachingComponentBase<T>::GetEventChannel() {
    return event_channel_;
}
 
template <typename T>
utils::SharedReadablePtr<T> CachingComponentBase<T>::GetUnsafe() const {
    return utils::SharedReadablePtr<T>(cache_.ReadCopy());
}
 
template <typename T>
void CachingComponentBase<T>::Set(std::unique_ptr<const T> value_ptr) {
    Attach(TransformNewValue(std::move(value_ptr)));
}
 
template <typename T>
void CachingComponentBase<T>::Set(T&& value) {
    Emplace(std::move(value));
}
 
template <typename T>
void CachingComponentBase<T>::Attach(const std::shared_ptr<const T>& value_ptr) {
    if (HasPreAssignCheck()) {
        auto old_value = cache_.Read();
        PreAssignCheck(old_value->get(), value_ptr.get());
    }
 
    cache_.Assign(value_ptr);
    event_channel_.SendEvent(value_ptr);
    OnCacheModified();
}
 
template <typename T>
template <typename... Args>
void CachingComponentBase<T>::Emplace(Args&&... args) {
    Set(std::make_unique<T>(std::forward<Args>(args)...));
}
 
template <typename T>
void CachingComponentBase<T>::Clear() {
    cache_.Assign(std::make_unique<const T>());
}
 
template <typename T>
bool CachingComponentBase<T>::MayReturnNull() const {
    return false;
}
 
template <typename T>
void CachingComponentBase<T>::GetAndWrite(dump::Writer& writer) const {
    const auto contents = GetUnsafe();
    if (!contents) {
        throw cache::EmptyCacheError(Name());
    }
    WriteContents(writer, *contents);
}
 
template <typename T>
void CachingComponentBase<T>::ReadAndSet(dump::Reader& reader) {
    auto data = ReadContents(reader);
    if constexpr (meta::kIsSizable<T>) {
        if (data) {
            SetDataSizeStatistic(std::size(*data));
        }
    }
    Set(std::move(data));
}
 
template <typename T>
void CachingComponentBase<T>::WriteContents(dump::Writer& writer, const T& contents) const {
    if constexpr (dump::kIsDumpable<T>) {
        writer.Write(contents);
    } else {
        dump::ThrowDumpUnimplemented(Name());
    }
}
 
template <typename T>
std::unique_ptr<const T> CachingComponentBase<T>::ReadContents(dump::Reader& reader) const {
    if constexpr (dump::kIsDumpable<T>) {
        // To avoid an extra move and avoid including common_containers.hpp
        return std::unique_ptr<const T>{new T(reader.Read<T>())};
    } else {
        dump::ThrowDumpUnimplemented(Name());
    }
}
 
template <typename T>
void CachingComponentBase<T>::Cleanup() {
    cache_.Cleanup();
}
 
template <typename T>
void CachingComponentBase<T>::MarkAsExpired() {
    Set(std::unique_ptr<const T>{});
}
 
namespace impl {
 
yaml_config::Schema GetCachingComponentBaseSchema();
 
template <typename T, typename Deleter>
auto MakeAsyncDeleter(engine::TaskProcessor& task_processor, Deleter deleter) {
    return [&task_processor, deleter = std::move(deleter)](const T* raw_ptr) mutable {
        std::unique_ptr<const T, Deleter> ptr(raw_ptr, std::move(deleter));
 
        engine::DetachUnscopedUnsafe(engine::CriticalAsyncNoTracing(task_processor, [ptr = std::move(ptr)]() mutable {})
        );
    };
}
 
}  // namespace impl
 
template <typename T>
yaml_config::Schema CachingComponentBase<T>::GetStaticConfigSchema() {
    return impl::GetCachingComponentBaseSchema();
}
 
template <typename T>
void CachingComponentBase<T>::PreAssignCheck(const T*, [[maybe_unused]] const T* new_value_ptr) const {
    UINVARIANT(
        meta::kIsSizable<T>,
        fmt::format(
            "{} type does not support std::size(), add implementation of "
            "the method size() for this type or "
            "override cache::CachingComponentBase::PreAssignCheck.",
            compiler::GetTypeName<T>()
        )
    );
 
    if constexpr (meta::kIsSizable<T>) {
        if (!new_value_ptr || std::size(*new_value_ptr) == 0) {
            throw cache::EmptyDataError(Name());
        }
    }
}
 
template <typename T>
std::shared_ptr<const T> CachingComponentBase<T>::TransformNewValue(std::unique_ptr<const T> new_value) {
    // Kill garbage asynchronously as T::~T() might be very slow
    if (IsSafeDataLifetime()) {
        // Use token only if `safe-data-lifetime` is true
        auto deleter_with_token = [token = wait_token_storage_.GetToken()](const T* raw_ptr) {
            // Make sure *raw_ptr is deleted before token is destroyed
            std::default_delete<const T>{}(raw_ptr);
        };
        return std::shared_ptr<const T>(
            new_value.release(),
            impl::MakeAsyncDeleter<T>(GetCacheTaskProcessor(), std::move(deleter_with_token))
        );
    } else {
        return std::shared_ptr<const T>(
            new_value.release(),
            impl::MakeAsyncDeleter<T>(GetCacheTaskProcessor(), std::default_delete<const T>{})
        );
    }
}
 
}  // namespace components
 
USERVER_NAMESPACE_END