Intermediate changes

commit_hash:36c8231ac7d17ca0d196613bbe94471e663a2ac9

4 files changed, 20 insertions, 12 deletions

diff --git a/yt/yt/core/concurrency/unittests/quantized_executor_ut.cpp b/yt/yt/core/concurrency/unittests/quantized_executor_ut.cpp
index 7db0cf83c7..4ddc546cdb 100644
--- a/yt/yt/core/concurrency/unittests/quantized_executor_ut.cpp
+++ b/yt/yt/core/concurrency/unittests/quantized_executor_ut.cpp
@@ -152,15 +152,18 @@ TEST_F(TQuantizedExecutorTest, Simple)
 
 TEST_F(TQuantizedExecutorTest, Timeout)
 {
-    InitSimple(/*workerCount*/ 4, /*iterationCount*/ std::numeric_limits<i64>::max());
+    InitSimple(/*workerCount*/ 2, /*iterationCount*/ std::numeric_limits<i64>::max());
 
-    for (int index = 1; index <= 10; ++index) {
-        WaitFor(Executor_->Run(TDuration::MilliSeconds(100)))
-            .ThrowOnError();
+    WaitFor(Executor_->Run(TDuration::MilliSeconds(100)))
+        .ThrowOnError();
 
-        auto counter = SimpleCallbackProvider_->GetCounter();
-        EXPECT_LE(counter, /*workerCount*/ 4 * /*milliseconds*/ 100.0 / /*period*/ 5 * index * 1.25);
-    }
+    auto oldCounter = SimpleCallbackProvider_->GetCounter();
+
+    TDelayedExecutor::WaitForDuration(TDuration::MilliSeconds(100));
+
+    auto newCounter = SimpleCallbackProvider_->GetCounter();
+
+    EXPECT_EQ(oldCounter, newCounter);
 }
 
 TEST_F(TQuantizedExecutorTest, LongCallback1)
diff --git a/yt/yt/core/rpc/unittests/lib/test_service.cpp b/yt/yt/core/rpc/unittests/lib/test_service.cpp
index 51492ad839..8ac8d4720d 100644
--- a/yt/yt/core/rpc/unittests/lib/test_service.cpp
+++ b/yt/yt/core/rpc/unittests/lib/test_service.cpp
@@ -21,6 +21,7 @@ using namespace NConcurrency;
 ////////////////////////////////////////////////////////////////////////////////
 
 YT_DEFINE_GLOBAL(std::unique_ptr<NThreading::TEvent>, Latch_);
+YT_DEFINE_GLOBAL(std::atomic<int>, ConcurrentCalls_, 0);
 
 ////////////////////////////////////////////////////////////////////////////////
 
@@ -202,6 +203,10 @@ public:
 
     DECLARE_RPC_SERVICE_METHOD(NTestRpc, RequestBytesThrottledCall)
     {
+        THROW_ERROR_EXCEPTION_UNLESS(ConcurrentCalls_().fetch_add(1) == 0, "Too many concurrent calls on entry!");
+        Sleep(TDuration::MilliSeconds(100));
+        THROW_ERROR_EXCEPTION_UNLESS(ConcurrentCalls_().fetch_sub(1) == 1, "Too many concurrent calls on exit!");
+
         context->Reply();
     }
 
diff --git a/yt/yt/core/rpc/unittests/rpc_ut.cpp b/yt/yt/core/rpc/unittests/rpc_ut.cpp
index 0221693c5a..157ee8faaa 100644
--- a/yt/yt/core/rpc/unittests/rpc_ut.cpp
+++ b/yt/yt/core/rpc/unittests/rpc_ut.cpp
@@ -661,7 +661,7 @@ TYPED_TEST(TNotGrpcTest, RequestBytesThrottling)
                 methods = {
                     RequestBytesThrottledCall = {
                         request_bytes_throttler = {
-                            limit = 1000000;
+                            limit = 100000;
                         }
                     }
                 }
@@ -675,18 +675,16 @@ TYPED_TEST(TNotGrpcTest, RequestBytesThrottling)
 
     auto makeCall = [&] {
         auto req = proxy.RequestBytesThrottledCall();
-        req->Attachments().push_back(TSharedMutableRef::Allocate(100'000));
+        req->Attachments().push_back(TSharedMutableRef::Allocate(60'000));
         return req->Invoke().AsVoid();
     };
 
     std::vector<TFuture<void>> futures;
-    for (int i = 0; i < 30; ++i) {
+    for (int i = 0; i < 5; ++i) {
         futures.push_back(makeCall());
     }
 
-    NProfiling::TWallTimer timer;
     EXPECT_TRUE(AllSucceeded(std::move(futures)).Get().IsOK());
-    EXPECT_LE(std::abs(static_cast<i64>(timer.GetElapsedTime().MilliSeconds()) - 3000), 200);
 }
 
 // Now test different types of errors.
diff --git a/yt/yt/library/program/config.cpp b/yt/yt/library/program/config.cpp
index 52aa9e5d8f..74b9d61fdd 100644
--- a/yt/yt/library/program/config.cpp
+++ b/yt/yt/library/program/config.cpp
@@ -187,6 +187,8 @@ void TSingletonsDynamicConfig::Register(TRegistrar registrar)
         .Optional();
     registrar.Parameter("tcmalloc", &TThis::TCMalloc)
         .Optional();
+    registrar.Parameter("stockpile", &TThis::Stockpile)
+        .Optional();
     registrar.Parameter("protobuf_interop", &TThis::ProtobufInterop)
         .DefaultNew();
 }