YT-25865: Fix for TAsyncExpiringCache: separate expiration & refresh

* Changelog entry
  Type: fix
  Component: core

Separated expiration and refresh in TAsyncExpiringCache
commit_hash:759a12d941f29591611b67641c1b758010c03d72

Author: cherepashka

yt/yt/client/driver/config.cpp

@@ -70,6 +70,7 @@ void TDriverConfig::Register(TRegistrar registrar)
     registrar.Preprocessor([] (TThis* config) {
         config->ClientCache->Capacity = 1024_KB;
         config->ProxyDiscoveryCache->RefreshTime = TDuration::Seconds(15);
+        config->ProxyDiscoveryCache->ExpirationPeriod = TDuration::Seconds(15);
         config->ProxyDiscoveryCache->ExpireAfterSuccessfulUpdateTime = TDuration::Seconds(15);
         config->ProxyDiscoveryCache->ExpireAfterFailedUpdateTime = TDuration::Seconds(15);
     });

yt/yt/core/misc/async_expiring_cache-inl.h

@@ -3,11 +3,13 @@
 #include "public.h"
 #include "cache_config.h"
 
-#include <type_traits>
 #include <yt/yt/core/actions/future.h>
 
 #include <yt/yt/core/logging/log.h>
 
+// TODO(cherepashka): remove dependency.
+#include <yt/yt/core/rpc/dispatcher.h>
+
 #include <yt/yt/library/profiling/sensor.h>
 
 #include <library/cpp/yt/threading/spin_lock.h>
@@ -20,7 +22,7 @@ namespace NYT {
 
 /*!
  *  \note
- *  Thread affinity: delayed executor's thread
+ *  Thread affinity: user defined invoker
  */
 template <class TKey, class TValue>
 class TAsyncExpiringCache
@@ -39,7 +41,9 @@ class TAsyncExpiringCache
     explicit TAsyncExpiringCache(
         TAsyncExpiringCacheConfigPtr config,
         NLogging::TLogger logger = {},
-        NProfiling::TProfiler profiler = {});
+        NProfiling::TProfiler profiler = {},
+        // TODO(cherepashka): remove default value and move upper.
+        const IInvokerPtr& invoker = NYT::NRpc::TDispatcher::Get()->GetHeavyInvoker());
 
     TFuture<TValue> Get(const TKey& key);
     TExtendedGetResult GetExtended(const TKey& key);
@@ -101,6 +105,8 @@ class TAsyncExpiringCache
 
 private:
     const NLogging::TLogger Logger_;
+    const NConcurrency::TPeriodicExecutorPtr ExpirationExecutor_;
+    const NConcurrency::TPeriodicExecutorPtr RefreshExecutor_;
 
     struct TEntry
         : public TRefCounted
@@ -131,7 +137,8 @@ class TAsyncExpiringCache
 
     YT_DECLARE_SPIN_LOCK(NThreading::TReaderWriterSpinLock, SpinLock_);
     THashMap<TKey, TEntryPtr> Map_;
-    TAsyncExpiringCacheConfigPtr Config_;
+
+    TAtomicIntrusivePtr<TAsyncExpiringCacheConfig> Config_;
 
     NProfiling::TCounter HitCounter_;
     NProfiling::TCounter MissedCounter_;
@@ -152,22 +159,29 @@ class TAsyncExpiringCache
         TWeakPtr<TEntry> weakEntry,
         const TKey& key);
 
+    enum EEraseReason
+    {
+        Refresh,
+        Expiration,
+    };
+
     bool TryEraseExpired(
         const TEntryPtr& Entry,
-        const TKey& key);
+        const TKey& key,
+        EEraseReason reason);
 
     void Erase(THashMap<TKey, TEntryPtr>::iterator it);
 
-    void UpdateAll();
+    void DeleteExpiredItems();
+    void RefreshAllItems();
 
-    void ScheduleEntryRefresh(
+    // Schedules entry expiration and refresh.
+    void ScheduleEntryUpdate(
         const TEntryPtr& entry,
         const TKey& key,
-        std::optional<TDuration> refreshTime);
+        const TAsyncExpiringCacheConfigPtr& config);
 
     TPromise<TValue> GetPromise(const TEntryPtr& entry) noexcept;
-
-    const TAsyncExpiringCacheConfigPtr& Config() const;
 };
 
 ////////////////////////////////////////////////////////////////////////////////

yt/yt/core/misc/async_expiring_cache.h

@@ -73,6 +73,8 @@ void TAsyncExpiringCacheConfig::Register(TRegistrar registrar)
     registrar.Parameter("refresh_time", &TThis::RefreshTime)
         .Alias("success_probation_time")
         .Default(TDuration::Seconds(10));
+    registrar.Parameter("expiration_period", &TThis::ExpirationPeriod)
+        .Default(TDuration::Seconds(10));
     registrar.Parameter("batch_update", &TThis::BatchUpdate)
         .Default(false);
 
@@ -88,13 +90,24 @@ void TAsyncExpiringCacheConfig::Register(TRegistrar registrar)
 void TAsyncExpiringCacheConfig::ApplyDynamicInplace(
     const TAsyncExpiringCacheDynamicConfigPtr& dynamicConfig)
 {
-    ExpireAfterAccessTime = dynamicConfig->ExpireAfterAccessTime.value_or(ExpireAfterAccessTime);
-    ExpireAfterSuccessfulUpdateTime = dynamicConfig->ExpireAfterSuccessfulUpdateTime.value_or(ExpireAfterSuccessfulUpdateTime);
-    ExpireAfterFailedUpdateTime = dynamicConfig->ExpireAfterFailedUpdateTime.value_or(ExpireAfterFailedUpdateTime);
-    RefreshTime = dynamicConfig->RefreshTime.has_value()
-        ? dynamicConfig->RefreshTime
-        : RefreshTime;
-    BatchUpdate = dynamicConfig->BatchUpdate.value_or(BatchUpdate);
+    if (dynamicConfig->ExpireAfterAccessTime) {
+        ExpireAfterAccessTime = *dynamicConfig->ExpireAfterAccessTime;
+    }
+    if (dynamicConfig->ExpireAfterSuccessfulUpdateTime) {
+        ExpireAfterSuccessfulUpdateTime = *dynamicConfig->ExpireAfterSuccessfulUpdateTime;
+    }
+    if (dynamicConfig->ExpireAfterFailedUpdateTime) {
+        ExpireAfterFailedUpdateTime = *dynamicConfig->ExpireAfterFailedUpdateTime;
+    }
+    if (dynamicConfig->RefreshTime) {
+        RefreshTime = *dynamicConfig->RefreshTime;
+    }
+    if (dynamicConfig->ExpirationPeriod) {
+        ExpirationPeriod = *dynamicConfig->ExpirationPeriod;
+    }
+    if (dynamicConfig->BatchUpdate) {
+        BatchUpdate = *dynamicConfig->BatchUpdate;
+    }
 }
 
 TAsyncExpiringCacheConfigPtr TAsyncExpiringCacheConfig::ApplyDynamic(

yt/yt/core/misc/cache_config.cpp

@@ -98,7 +98,10 @@ struct TAsyncExpiringCacheConfig
     TDuration ExpireAfterFailedUpdateTime;
 
     //! Time before next (background) update. If nullopt - background updates are disabled.
+    // TODO(cherepashka): rename into RefreshPeriod.
     std::optional<TDuration> RefreshTime;
+    //! Time before next (background) expiration. If nullopt - background expirations are disabled.
+    std::optional<TDuration> ExpirationPeriod;
 
     //! If set to true, cache will invoke DoGetMany once instead of DoGet on every entry during an update.
     bool BatchUpdate;
@@ -124,6 +127,7 @@ struct TAsyncExpiringCacheDynamicConfig
     std::optional<TDuration> ExpireAfterSuccessfulUpdateTime;
     std::optional<TDuration> ExpireAfterFailedUpdateTime;
     std::optional<TDuration> RefreshTime;
+    std::optional<TDuration> ExpirationPeriod;
 
     std::optional<bool> BatchUpdate;
 

yt/yt/core/misc/cache_config.h

@@ -43,6 +43,7 @@ void TAddressResolverConfig::Register(TRegistrar registrar)
 
     registrar.Preprocessor([] (TThis* config) {
         config->RefreshTime = TDuration::Seconds(60);
+        config->ExpirationPeriod = TDuration::Seconds(60);
         config->ExpireAfterSuccessfulUpdateTime = TDuration::Seconds(120);
         config->ExpireAfterFailedUpdateTime = TDuration::Seconds(30);
     });

yt/yt/core/net/config.cpp

@@ -32,6 +32,7 @@ void TServiceDiscoveryConfig::Register(TRegistrar registrar)
         config->ExpireAfterSuccessfulUpdateTime = TDuration::Days(1);
         config->ExpireAfterFailedUpdateTime = TDuration::Seconds(5);
         config->RefreshTime = TDuration::Seconds(5);
+        config->ExpirationPeriod = TDuration::Seconds(5);
     });
 }