Intermediate changes

commit_hash:8b7eb71badc9f2fcd168ee34e8c379b35577eccb

library/cpp/containers/dense_hash/dense_hash.h

@@ -2,6 +2,7 @@
 
 #include "fwd.h"
 
+#include <util/generic/bitops.h>
 #include <util/generic/utility.h>
 #include <util/generic/vector.h>
 #include <util/generic/mapfindptr.h>

library/cpp/yt/stockpile/stockpile.h

@@ -1,5 +1,9 @@
 #pragma once
 
+#include <library/cpp/yt/cpu_clock/clock.h>
+
+#include <library/cpp/yt/misc/enum.h>
+
 #include <util/system/types.h>
 
 #include <util/generic/size_literals.h>
@@ -10,6 +14,14 @@ namespace NYT {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+DEFINE_ENUM(EStockpileStrategy,
+    ((FixedBreaks)          (0))
+    ((FlooredLoad)          (1))
+    ((ProgressiveBackoff)   (2))
+);
+
+////////////////////////////////////////////////////////////////////////////////
+
 struct TStockpileOptions
 {
     static constexpr i64 DefaultBufferSize = 4_GBs;
@@ -18,11 +30,18 @@ struct TStockpileOptions
     static constexpr int DefaultThreadCount = 4;
     int ThreadCount = DefaultThreadCount;
 
+    static constexpr EStockpileStrategy DefaultStrategy = EStockpileStrategy::FixedBreaks;
+    EStockpileStrategy Strategy = DefaultStrategy;
+
     static constexpr TDuration DefaultPeriod = TDuration::MilliSeconds(10);
     TDuration Period = DefaultPeriod;
 };
 
-void ConfigureStockpile(const TStockpileOptions& options);
+////////////////////////////////////////////////////////////////////////////////
+
+void RunStockpileThread(TStockpileOptions options, std::atomic<bool>* shouldProceed);
+
+void RunDetachedStockpileThreads(TStockpileOptions options);
 
 ////////////////////////////////////////////////////////////////////////////////
 

library/cpp/yt/stockpile/stockpile_linux.cpp

@@ -1,38 +1,117 @@
 #include "stockpile.h"
 
+#include "library/cpp/yt/logging/logger.h"
+
 #include <thread>
 #include <mutex>
 
 #include <sys/mman.h>
 
 #include <util/system/thread.h>
+#include <string.h>
 
 namespace NYT {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+static const auto Logger = NLogging::TLogger("Stockpile");
+
+constexpr int MADV_STOCKPILE = 0x59410004;
+
+////////////////////////////////////////////////////////////////////////////////
+
 namespace {
 
-void RunStockpile(const TStockpileOptions& options)
+void RunWithFixedBreaks(i64 bufferSize, TDuration period)
 {
-    TThread::SetCurrentThreadName("Stockpile");
+    auto returnCode = ::madvise(nullptr, bufferSize, MADV_STOCKPILE);
+    YT_LOG_DEBUG_IF(returnCode, "System call \"madvise\" returned %v", strerror(returnCode));
+    Sleep(period);
+}
 
-    constexpr int MADV_STOCKPILE = 0x59410004;
+void RunWithCappedLoad(i64 bufferSize, TDuration period)
+{
+    auto started = GetApproximateCpuInstant();
+    auto returnCode = ::madvise(nullptr, bufferSize, MADV_STOCKPILE);
+    YT_LOG_DEBUG_IF(returnCode, "System call \"madvise\" returned %v", strerror(returnCode));
+    auto duration = CpuDurationToDuration(GetApproximateCpuInstant() - started);
 
-    while (true) {
-        ::madvise(nullptr, options.BufferSize, MADV_STOCKPILE);
-        Sleep(options.Period);
+    if (duration < period) {
+        Sleep(period - duration);
+    }
+}
+
+std::pair<i64, TDuration> RunWithBackoffs(
+    i64 adjustedBufferSize,
+    TDuration adjustedPeriod,
+    const TStockpileOptions& options,
+    i64 pageSize)
+{
+    int returnCode = ::madvise(nullptr, adjustedBufferSize, MADV_STOCKPILE);
+    YT_LOG_DEBUG_IF(returnCode, "System call \"madvise\" returned %v", strerror(returnCode));
+
+    switch(returnCode) {
+        case 0:
+            Sleep(options.Period);
+            return {options.BufferSize, options.Period};
+
+        case ENOMEM:
+            if (adjustedBufferSize / 2 >= pageSize) {
+                // Immediately make an attempt to reclaim half as much.
+                adjustedBufferSize = adjustedBufferSize / 2;
+            } else {
+                // Unless there is not even a single reclaimable page.
+                Sleep(options.Period);
+            }
+            return {adjustedBufferSize, options.Period};
+
+        case EAGAIN:
+        case EINTR:
+            Sleep(adjustedPeriod);
+            return {options.BufferSize, adjustedPeriod + options.Period};
+
+        default:
+            Sleep(options.Period);
+            return {options.BufferSize, options.Period};
     }
 }
 
 } // namespace
 
-void ConfigureStockpile(const TStockpileOptions& options)
+void RunStockpileThread(TStockpileOptions options, std::atomic<bool>* shouldProceed)
+{
+    TThread::SetCurrentThreadName("Stockpile");
+
+    const i64 pageSize = sysconf(_SC_PAGESIZE);
+    auto bufferSize = options.BufferSize;
+    auto period = options.Period;
+
+    while (!shouldProceed || shouldProceed->load()) {
+        switch (options.Strategy) {
+            case EStockpileStrategy::FixedBreaks:
+                RunWithFixedBreaks(options.BufferSize, options.Period);
+                break;
+
+            case EStockpileStrategy::FlooredLoad:
+                RunWithCappedLoad(options.BufferSize, options.Period);
+                break;
+
+            case EStockpileStrategy::ProgressiveBackoff:
+                std::tie(bufferSize, period) = RunWithBackoffs(bufferSize, period, options, pageSize);
+                break;
+
+            default:
+                YT_ABORT();
+        }
+    }
+}
+
+void RunDetachedStockpileThreads(TStockpileOptions options)
 {
     static std::once_flag OnceFlag;
-    std::call_once(OnceFlag, [options] {
-        for (int i = 0; i < options.ThreadCount; i++) {
-            std::thread(RunStockpile, options).detach();
+    std::call_once(OnceFlag, [options = std::move(options)] {
+        for (int i = 0; i < options.ThreadCount; ++i) {
+            std::thread(RunStockpileThread, options, nullptr).detach();
         }
     });
 }

library/cpp/yt/stockpile/stockpile_other.cpp

@@ -4,7 +4,10 @@ namespace NYT {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-void ConfigureStockpile(const TStockpileOptions& /*options*/)
+void RunStockpileThread(TStockpileOptions /*options*/, std::atomic<bool>* /*shouldProceed*/)
+{ }
+
+void RunDetachedStockpileThreads(TStockpileOptions /*options*/)
 { }
 
 ////////////////////////////////////////////////////////////////////////////////

yt/yt/library/program/config.cpp

@@ -54,13 +54,52 @@ void TTCMallocConfig::Register(TRegistrar registrar)
 void TStockpileConfig::Register(TRegistrar registrar)
 {
     registrar.BaseClassParameter("buffer_size", &TThis::BufferSize)
-        .Default(DefaultBufferSize);
+        .Default(DefaultBufferSize)
+        .GreaterThan(0);
     registrar.BaseClassParameter("thread_count", &TThis::ThreadCount)
         .Default(DefaultThreadCount);
+    registrar.BaseClassParameter("strategy", &TThis::Strategy)
+        .Default(DefaultStrategy);
     registrar.BaseClassParameter("period", &TThis::Period)
         .Default(DefaultPeriod);
 }
 
+TStockpileConfigPtr TStockpileConfig::ApplyDynamic(const TStockpileDynamicConfigPtr& dynamicConfig) const
+{
+    auto mergedConfig = CloneYsonStruct(MakeStrong(this));
+
+    if (dynamicConfig->BufferSize) {
+        mergedConfig->BufferSize = *dynamicConfig->BufferSize;
+    }
+    if (dynamicConfig->ThreadCount) {
+        mergedConfig->ThreadCount = *dynamicConfig->ThreadCount;
+    }
+    if (dynamicConfig->Strategy) {
+        mergedConfig->Strategy = *dynamicConfig->Strategy;
+    }
+    if (dynamicConfig->Period) {
+        mergedConfig->Period = *dynamicConfig->Period;
+    }
+
+    return mergedConfig;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+void TStockpileDynamicConfig::Register(TRegistrar registrar)
+{
+    registrar.BaseClassParameter("buffer_size", &TThis::BufferSize)
+        .Optional()
+        .GreaterThan(0);
+    registrar.BaseClassParameter("thread_count", &TThis::ThreadCount)
+        .Optional()
+        .GreaterThan(0);
+    registrar.BaseClassParameter("strategy", &TThis::Strategy)
+        .Optional();
+    registrar.BaseClassParameter("period", &TThis::Period)
+        .Optional();
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 void THeapProfilerConfig::Register(TRegistrar registrar)

yt/yt/library/program/config.h

@@ -101,6 +101,8 @@ class TStockpileConfig
     , public NYTree::TYsonStruct
 {
 public:
+    TStockpileConfigPtr ApplyDynamic(const TStockpileDynamicConfigPtr& dynamicConfig) const;
+
     REGISTER_YSON_STRUCT(TStockpileConfig);
 
     static void Register(TRegistrar registrar);
@@ -110,6 +112,24 @@ DEFINE_REFCOUNTED_TYPE(TStockpileConfig)
 
 ////////////////////////////////////////////////////////////////////////////////
 
+class TStockpileDynamicConfig
+    : public NYTree::TYsonStruct
+{
+public:
+    std::optional<i64> BufferSize;
+    std::optional<int> ThreadCount;
+    std::optional<EStockpileStrategy> Strategy;
+    std::optional<TDuration> Period;
+
+    REGISTER_YSON_STRUCT(TStockpileDynamicConfig);
+
+    static void Register(TRegistrar registrar);
+};
+
+DEFINE_REFCOUNTED_TYPE(TStockpileDynamicConfig)
+
+////////////////////////////////////////////////////////////////////////////////
+
 class THeapProfilerConfig
     : public NYTree::TYsonStruct
 {
@@ -178,6 +198,7 @@ public:
     NTracing::TJaegerTracerDynamicConfigPtr Jaeger;
     NTracing::TTracingTransportConfigPtr TracingTransport;
     TTCMallocConfigPtr TCMalloc;
+    TStockpileDynamicConfigPtr Stockpile;
     NYson::TProtobufInteropDynamicConfigPtr ProtobufInterop;
 
     REGISTER_YSON_STRUCT(TSingletonsDynamicConfig);

yt/yt/library/program/helpers.cpp

@@ -1,6 +1,7 @@
 #include "helpers.h"
 #include "config.h"
 #include "private.h"
+#include "stockpile.h"
 
 #include <yt/yt/core/ytalloc/bindings.h>
 
@@ -243,7 +244,7 @@ void ConfigureSingletons(const TSingletonsConfigPtr& config)
 
     ConfigureTCMalloc(config->TCMalloc);
 
-    ConfigureStockpile(*config->Stockpile);
+    TStockpileManager::Get()->Reconfigure(*config->Stockpile);
 
     if (config->EnableRefCountedTrackerProfiling) {
         EnableRefCountedTrackerProfiling();
@@ -305,6 +306,10 @@ void ReconfigureSingletons(const TSingletonsConfigPtr& config, const TSingletons
         ConfigureTCMalloc(config->TCMalloc);
     }
 
+    if (dynamicConfig->Stockpile) {
+        TStockpileManager::Get()->Reconfigure(*config->Stockpile->ApplyDynamic(dynamicConfig->Stockpile));
+    }
+
     NYson::SetProtobufInteropConfig(config->ProtobufInterop->ApplyDynamic(dynamicConfig->ProtobufInterop));
 }
 

yt/yt/library/program/public.h

@@ -10,6 +10,7 @@ DECLARE_REFCOUNTED_CLASS(TBuildInfo)
 DECLARE_REFCOUNTED_CLASS(TRpcConfig)
 DECLARE_REFCOUNTED_CLASS(TTCMallocConfig)
 DECLARE_REFCOUNTED_CLASS(TStockpileConfig)
+DECLARE_REFCOUNTED_CLASS(TStockpileDynamicConfig)
 DECLARE_REFCOUNTED_CLASS(TSingletonsConfig)
 DECLARE_REFCOUNTED_CLASS(TSingletonsDynamicConfig)
 DECLARE_REFCOUNTED_CLASS(TDiagnosticDumpConfig)

yt/yt/library/program/stockpile.cpp

@@ -0,0 +1,44 @@
+#include "stockpile.h"
+
+namespace NYT {
+
+////////////////////////////////////////////////////////////////////////////////
+
+TStockpileManager* TStockpileManager::Get()
+{
+    return Singleton<TStockpileManager>();
+}
+
+void TStockpileManager::Reconfigure(TStockpileOptions options)
+{
+    auto guard = Guard(SpinLock_);
+
+    ThreadState_->ShouldProceed.store(false);
+
+    for (auto& thread : Threads_) {
+        thread->join();
+    }
+
+    Threads_.clear();
+    ThreadState_ = New<TStockpileThreadState>();
+    ThreadState_->ShouldProceed.store(true, std::memory_order_release);
+
+    for (int threadIndex = 0; threadIndex < options.ThreadCount; ++threadIndex) {
+        Threads_.push_back(std::make_unique<std::thread>([options, state = ThreadState_] {
+            RunStockpileThread(options, &state->ShouldProceed);
+        }));
+    }
+}
+
+TStockpileManager::~TStockpileManager()
+{
+    ThreadState_->ShouldProceed.store(false);
+
+    for (auto& thread : Threads_) {
+        thread->detach();
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NYT

yt/yt/library/program/stockpile.h

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <library/cpp/yt/memory/intrusive_ptr.h>
+#include "library/cpp/yt/memory/new.h"
+
+#include <library/cpp/yt/stockpile/stockpile.h>
+
+#include <library/cpp/yt/threading/spin_lock.h>
+
+#include <thread>
+
+namespace NYT {
+
+////////////////////////////////////////////////////////////////////////////////
+
+DECLARE_REFCOUNTED_STRUCT(TStockpileThreadState)
+
+struct TStockpileThreadState final
+{
+    std::atomic<bool> ShouldProceed;
+};
+
+DEFINE_REFCOUNTED_TYPE(TStockpileThreadState)
+
+////////////////////////////////////////////////////////////////////////////////
+
+class TStockpileManager
+{
+public:
+    static TStockpileManager* Get();
+
+    void Reconfigure(TStockpileOptions options);
+
+    ~TStockpileManager();
+
+private:
+    YT_DECLARE_SPIN_LOCK(NThreading::TSpinLock, SpinLock_);
+    std::vector<std::unique_ptr<std::thread>> Threads_;
+
+    TStockpileThreadStatePtr ThreadState_ = New<TStockpileThreadState>();
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NYT