Pass VDisk burst threshold in config, fix TBucketQuoter bug (#2165)

library/cpp/bucket_quoter/bucket_quoter.h

@@ -172,8 +172,8 @@ public:
 
     i64 UseAndFill(ui64 tokens) {
         TGuard<Lock> g(BucketMutex);
-        UseNoLock(tokens);
         FillBucket();
+        UseNoLock(tokens);
         return Bucket;
     }
 

ydb/core/blobstorage/ut_blobstorage/lib/env.h

@@ -39,6 +39,8 @@ struct TEnvironmentSetup {
         const bool SuppressCompatibilityCheck = false;
         const TFeatureFlags FeatureFlags;
         const NPDisk::EDeviceType DiskType = NPDisk::EDeviceType::DEVICE_TYPE_NVME;
+        const ui32 BurstThresholdNs = 0;
+        const TString VDiskKind = "";
     };
 
     const TSettings Settings;
@@ -323,6 +325,16 @@ struct TEnvironmentSetup {
                     config->CacheAccessor = std::make_unique<TAccessor>(Cache[nodeId]);
                 }
                 config->FeatureFlags = Settings.FeatureFlags;
+                if (Settings.VDiskKind) {
+                    NKikimrBlobStorage::TAllVDiskKinds vdiskConfig;
+                    auto* kind = vdiskConfig.AddVDiskKinds();
+                    kind->SetKind(NKikimrBlobStorage::TVDiskKind::Test1);
+                    if (Settings.BurstThresholdNs) {
+                        kind->MutableConfig()->SetBurstThresholdNs(Settings.BurstThresholdNs);
+                    }
+
+                    config->AllVDiskKinds = MakeIntrusive<TAllVDiskKinds>(vdiskConfig);
+                }
                 warden.reset(CreateBSNodeWarden(config));
             }
 
@@ -408,7 +420,11 @@ struct TEnvironmentSetup {
         cmd2->SetName(StoragePoolName);
         cmd2->SetKind(StoragePoolName);
         cmd2->SetErasureSpecies(TBlobStorageGroupType::ErasureSpeciesName(Settings.Erasure.GetErasure()));
-        cmd2->SetVDiskKind("Default");
+        if (Settings.VDiskKind) {
+            cmd2->SetVDiskKind(Settings.VDiskKind);
+        } else {
+            cmd2->SetVDiskKind("Default");
+        }
         cmd2->SetNumGroups(numGroups ? numGroups : NumGroups);
         cmd2->AddPDiskFilter()->AddProperty()->SetType(pdiskType);
         if (Settings.Encryption) {
@@ -428,7 +444,11 @@ struct TEnvironmentSetup {
         cmd->SetName(poolName);
         cmd->SetKind(poolName);
         cmd->SetErasureSpecies(TBlobStorageGroupType::ErasureSpeciesName(Settings.Erasure.GetErasure()));
-        cmd->SetVDiskKind("Default");
+        if (Settings.VDiskKind) {
+            cmd->SetVDiskKind(Settings.VDiskKind);
+        } else {
+            cmd->SetVDiskKind("Default");
+        }
         cmd->SetNumGroups(1);
         cmd->AddPDiskFilter()->AddProperty()->SetType(NKikimrBlobStorage::EPDiskType::ROT);
         if (Settings.Encryption) {

ydb/core/blobstorage/ut_blobstorage/monitoring.cpp

@@ -33,13 +33,16 @@ ui64 AggregateVDiskCounters(std::unique_ptr<TEnvironmentSetup>& env, TString sto
 };
 
 void SetupEnv(const TBlobStorageGroupInfo::TTopology& topology, std::unique_ptr<TEnvironmentSetup>& env,
-        ui32& groupSize, TBlobStorageGroupType& groupType, ui32& groupId, std::vector<ui32>& pdiskLayout) {
+        ui32& groupSize, TBlobStorageGroupType& groupType, ui32& groupId, std::vector<ui32>& pdiskLayout,
+        ui32 burstThresholdNs = 0, TString vdiskKind = "") {
     groupSize = topology.TotalVDisks;
     groupType = topology.GType;
     env.reset(new TEnvironmentSetup({
         .NodeCount = groupSize,
         .Erasure = groupType,
         .DiskType = NPDisk::EDeviceType::DEVICE_TYPE_ROT,
+        .BurstThresholdNs = burstThresholdNs,
+        .VDiskKind = vdiskKind,
     }));
 
     env->CreateBoxAndPool(1, 1);
@@ -244,13 +247,16 @@ enum class ELoadDistribution : ui8 {
 };
 
 template <typename TInflightActor>
-void TestBurst(const TBlobStorageGroupInfo::TTopology& topology, TInflightActor* actor, ELoadDistribution loadDistribution) {
+void TestBurst(ui32 requests, ui32 inflight, TDuration delay, ELoadDistribution loadDistribution,
+        ui32 burstThresholdNs = 0) {
+    TBlobStorageGroupInfo::TTopology topology(TBlobStorageGroupType::ErasureNone, 1, 1, 1, true);
+    auto* actor = new TInflightActor({requests, inflight, delay}, 8_MB);
     std::unique_ptr<TEnvironmentSetup> env;
     ui32 groupSize;
     TBlobStorageGroupType groupType;
     ui32 groupId;
     std::vector<ui32> pdiskLayout;
-    SetupEnv(topology, env, groupSize, groupType, groupId, pdiskLayout);
+    SetupEnv(topology, env, groupSize, groupType, groupId, pdiskLayout, burstThresholdNs, "Test1");
 
     actor->SetGroupId(groupId);
     env->Runtime->Register(actor, 1);
@@ -266,16 +272,18 @@ void TestBurst(const TBlobStorageGroupInfo::TTopology& topology, TInflightActor*
     }
 }
 
-#define MAKE_BURST_TEST(requestType, requests, inflight, delay, distribution)                       \
-Y_UNIT_TEST(Test##requestType##distribution) {                                                      \
-    TBlobStorageGroupInfo::TTopology topology(TBlobStorageGroupType::ErasureNone, 1, 1, 1, true);   \
-    auto* actor = new TInflightActor##requestType({requests, inflight, delay}, 8_MB);               \
-    TestBurst(topology, actor, ELoadDistribution::Distribution##distribution);                      \
-}
-
 Y_UNIT_TEST_SUITE(BurstDetection) {
-    MAKE_BURST_TEST(Put, 10, 1, TDuration::Seconds(1), Evenly);
-    MAKE_BURST_TEST(Put, 10, 100, TDuration::MilliSeconds(1), Burst);
+    Y_UNIT_TEST(TestPutEvenly) {
+        TestBurst<TInflightActorPut>(10, 1, TDuration::Seconds(1), ELoadDistribution::DistributionEvenly);
+    }
+
+    Y_UNIT_TEST(TestPutBurst) {
+        TestBurst<TInflightActorPut>(10, 10, TDuration::MilliSeconds(1), ELoadDistribution::DistributionBurst);
+    }
+
+    Y_UNIT_TEST(TestOverlySensitive) {
+        TestBurst<TInflightActorPut>(10, 1, TDuration::Seconds(1), ELoadDistribution::DistributionBurst, 1);
+    }
 }
 
 #undef MAKE_BURST_TEST

ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.cpp

@@ -42,7 +42,7 @@ public:
 };
 
 TBsCostTracker::TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::EDeviceType diskType,
-        const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters)
+        const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters, ui64 burstThresholdNs)
     : GroupType(groupType)
     , CostCounters(counters->GetSubgroup("subsystem", "advancedCost"))
     , UserDiskCost(CostCounters->GetCounter("UserDiskCost", true))
@@ -50,6 +50,7 @@ TBsCostTracker::TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::E
     , ScrubDiskCost(CostCounters->GetCounter("ScrubDiskCost", true))
     , DefragDiskCost(CostCounters->GetCounter("DefragDiskCost", true))
     , InternalDiskCost(CostCounters->GetCounter("InternalDiskCost", true))
+    , BucketCapacity(burstThresholdNs / GroupType.BlobSubgroupSize())
     , Bucket(&DiskTimeAvailableNs, &BucketCapacity, nullptr, nullptr, nullptr, nullptr, true)
 {
     BurstDetector.Initialize(CostCounters, "BurstDetector");
@@ -67,7 +68,6 @@ TBsCostTracker::TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::E
         CostModel = std::make_unique<TBsCostModelErasureNone>(diskType);
         break;
     }
-    UpdateBucketCapacity();
 }
 
 } // NKikimr

ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.h

@@ -318,7 +318,7 @@ private:
     ::NMonitoring::TDynamicCounters::TCounterPtr DefragDiskCost;
     ::NMonitoring::TDynamicCounters::TCounterPtr InternalDiskCost;
 
-    TAtomic BucketCapacity = 1'000'000'000;  // 10^9 nsec
+    TAtomic BucketCapacity;  // 10^9 nsec
     TAtomic DiskTimeAvailableNs = 1'000'000'000;
     TBucketQuoter<i64, TSpinLock, TAppDataTimerMs<TInstantTimerMs>> Bucket;
     TLight BurstDetector;
@@ -327,7 +327,7 @@ private:
 
 public:
     TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::EDeviceType diskType,
-            const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters);
+            const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters, ui64 burstThresholdNs);
 
     template<class TEv>
     ui64 GetCost(const TEv& ev) const {
@@ -340,35 +340,15 @@ public:
         return cost;
     }
 
-private:
-    void UpdateBucketCapacity() {
-        if (!CostModel) {
-            return;
-        }
-        ui64 maxPartSize = GroupType.MaxPartSize(TBlobStorageGroupType::ECrcMode::CrcModeWholePart, MaxVDiskBlobSize);
-        ui64 maxHugePartSize = GroupType.MaxPartSize(TBlobStorageGroupType::ECrcMode::CrcModeWholePart,
-                CostModel->HugeBlobSize);
-        ui64 capacity = std::max({
-            CostModel->ReadCost(maxHugePartSize),
-            CostModel->WriteCost(maxPartSize),
-            CostModel->HugeWriteCost(maxHugePartSize)
-        }) * ConcurrentHugeRequestsAllowed;
-
-        if (capacity != (ui64)AtomicGet(BucketCapacity)) {
-            AtomicSet(BucketCapacity, capacity);
-        }
-    }
-
 public:
     void UpdateCostModel(const TCostModel& costModel) {
         if (CostModel) {
             CostModel->Update(costModel);
         }
-        UpdateBucketCapacity();
     }
 
     void CountRequest(ui64 cost) {
-        Bucket.Use(cost);
+        Bucket.UseAndFill(cost);
         BurstDetector.Set(!Bucket.IsAvail(), SeqnoBurstDetector.fetch_add(1));
     }
 

ydb/core/blobstorage/vdisk/common/vdisk_config.cpp

@@ -119,6 +119,8 @@ namespace NKikimr {
 #else
         BarrierValidation = true; // switch by default on debug builds
 #endif
+
+        BurstThresholdNs = NPDisk::DevicePerformance.at(baseInfo.DeviceType).BurstThresholdNs;
     }
 
     void TVDiskConfig::SetupHugeBytes() {
@@ -160,6 +162,8 @@ namespace NKikimr {
         UPDATE_MACRO(ReplInterconnectChannel);
 
         UPDATE_MACRO(BarrierValidation);
+
+        UPDATE_MACRO(BurstThresholdNs);
 #undef UPDATE_MACRO
     }
 
@@ -180,6 +184,15 @@ namespace NKikimr {
         ParseConfig();
     }
 
+    TAllVDiskKinds::TAllVDiskKinds(const NKikimrBlobStorage::TAllVDiskKinds &proto)
+        : AllKindsConfig()
+        , VDiskMegaBaseConfig(TVDiskConfig::TBaseInfo())
+        , KindsMap()
+    {
+        AllKindsConfig.CopyFrom(proto);
+        ParseConfig();
+    }
+
     TIntrusivePtr<TVDiskConfig> TAllVDiskKinds::MakeVDiskConfig(const TVDiskConfig::TBaseInfo &baseInfo) {
         EKind k = baseInfo.Kind;
         TVector<const TKind *> merge;

ydb/core/blobstorage/vdisk/common/vdisk_config.h

@@ -209,6 +209,7 @@ namespace NKikimr {
         TDuration WhiteboardUpdateInterval;
         bool EnableVDiskCooldownTimeout;
         TControlWrapper EnableVPatch = true;
+        ui64 BurstThresholdNs = 1'000'000'000;
 
         ///////////// FEATURE FLAGS ////////////////////////
         NKikimrConfig::TFeatureFlags FeatureFlags;
@@ -228,6 +229,7 @@ namespace NKikimr {
     class TAllVDiskKinds : public TThrRefBase {
     public:
         TAllVDiskKinds(const TString &prototext = TString());
+        TAllVDiskKinds(const NKikimrBlobStorage::TAllVDiskKinds &proto);
         TIntrusivePtr<TVDiskConfig> MakeVDiskConfig(const TVDiskConfig::TBaseInfo &baseInfo);
         void Merge(const NKikimrBlobStorage::TAllVDiskKinds &allVDiskKinds);
 

ydb/core/blobstorage/vdisk/common/vdisk_context.cpp

@@ -30,7 +30,8 @@ namespace NKikimr {
                 TReplQuoter::TPtr replPDiskReadQuoter,
                 TReplQuoter::TPtr replPDiskWriteQuoter,
                 TReplQuoter::TPtr replNodeRequestQuoter,
-                TReplQuoter::TPtr replNodeResponseQuoter)
+                TReplQuoter::TPtr replNodeResponseQuoter,
+                ui64 burstThresholdNs)
         : TBSProxyContext(vdiskCounters->GetSubgroup("subsystem", "memhull"))
         , VDiskActorId(vdiskActorId)
         , Top(std::move(top))
@@ -57,7 +58,7 @@ namespace NKikimr {
         , ReplPDiskWriteQuoter(std::move(replPDiskWriteQuoter))
         , ReplNodeRequestQuoter(std::move(replNodeRequestQuoter))
         , ReplNodeResponseQuoter(std::move(replNodeResponseQuoter))
-        , CostTracker(std::make_shared<TBsCostTracker>(Top->GType, type, vdiskCounters))
+        , CostTracker(std::make_shared<TBsCostTracker>(Top->GType, type, vdiskCounters, burstThresholdNs))
         , OutOfSpaceState(Top->GetTotalVDisksNum(), Top->GetOrderNumber(ShortSelfVDisk))
         , CostMonGroup(vdiskCounters, "subsystem", "cost")
         , Logger(as ? ActorSystemLogger(as) : DevNullLogger())

ydb/core/blobstorage/vdisk/common/vdisk_context.h

@@ -100,7 +100,8 @@ namespace NKikimr {
                 TReplQuoter::TPtr replPDiskReadQuoter = nullptr,
                 TReplQuoter::TPtr replPDiskWriteQuoter = nullptr,
                 TReplQuoter::TPtr replNodeRequestQuoter = nullptr,
-                TReplQuoter::TPtr replNodeResponseQuoter = nullptr);
+                TReplQuoter::TPtr replNodeResponseQuoter = nullptr,
+                ui64 burstThresholdNs = 1'000'000'000);
 
         // The function checks response from PDisk. Normally, it's OK.
         // Other alternatives are: 1) shutdown; 2) FAIL

ydb/core/blobstorage/vdisk/skeleton/blobstorage_skeleton.cpp

@@ -2153,6 +2153,10 @@ namespace NKikimr {
                                     TABLED() {str << "VDiskIncarnationGuid";}
                                     TABLED() {str << Db->GetVDiskIncarnationGuid(true);}
                                 }
+                                TABLER() {
+                                    TABLED() {str << "BurstThresholdNs";}
+                                    TABLED() {str << Config->BurstThresholdNs;}
+                                }
                             }
                         }