Adjust cost bucket capacity dynamically (#1681)

* Adjust cost bucket capacity dynamically

* ui32 for BlobSize

5 files changed, 36 insertions, 8 deletions

diff --git a/ydb/core/base/blobstorage_grouptype.cpp b/ydb/core/base/blobstorage_grouptype.cpp
index 747a91ead9..65effc85ff 100644
--- a/ydb/core/base/blobstorage_grouptype.cpp
+++ b/ydb/core/base/blobstorage_grouptype.cpp
@@ -196,7 +196,11 @@ ui64 TBlobStorageGroupType::PartSize(const TLogoBlobID &id) const {
 
 ui64 TBlobStorageGroupType::MaxPartSize(const TLogoBlobID &id) const {
     Y_ABORT_UNLESS(!id.PartId());
-    return TErasureType::PartSize((TErasureType::ECrcMode)id.CrcMode(), id.BlobSize());
+    return MaxPartSize((TErasureType::ECrcMode)id.CrcMode(), id.BlobSize());
+}
+
+ui64 TBlobStorageGroupType::MaxPartSize(TErasureType::ECrcMode crcMode, ui32 blobSize) const {
+    return TErasureType::PartSize(crcMode, blobSize);
 }
 
 bool TBlobStorageGroupType::PartFits(ui32 partId, ui32 idxInSubgroup) const {
diff --git a/ydb/core/base/blobstorage_grouptype.h b/ydb/core/base/blobstorage_grouptype.h
index aeca719493..5d067087fb 100644
--- a/ydb/core/base/blobstorage_grouptype.h
+++ b/ydb/core/base/blobstorage_grouptype.h
@@ -144,6 +144,7 @@ struct TBlobStorageGroupType : public TErasureType {
 
     ui64 PartSize(const TLogoBlobID &id) const;
     ui64 MaxPartSize(const TLogoBlobID &id) const;
+    ui64 MaxPartSize(TErasureType::ECrcMode crcMode, ui32 blobSize) const;
 
     bool PartFits(ui32 partId, ui32 idxInSubgroup) const;
 };
diff --git a/ydb/core/blobstorage/ut_blobstorage/monitoring.cpp b/ydb/core/blobstorage/ut_blobstorage/monitoring.cpp
index f5293b14de..58c8f833c4 100644
--- a/ydb/core/blobstorage/ut_blobstorage/monitoring.cpp
+++ b/ydb/core/blobstorage/ut_blobstorage/monitoring.cpp
@@ -440,13 +440,13 @@ 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}, 10_KB);               \
+    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, 3000, 1, TDuration::MilliSeconds(5), Evenly);
-    MAKE_BURST_TEST(Put, 3000, 1000000, TDuration::Zero(), Burst);
+    MAKE_BURST_TEST(Put, 50, 1, TDuration::MilliSeconds(100), Evenly);
+    MAKE_BURST_TEST(Put, 50, 100, TDuration::Zero(), Burst);
 }
 
 #undef MAKE_BURST_TEST
diff --git a/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.cpp b/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.cpp
index 9c9e458f42..1a5ec980f8 100644
--- a/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.cpp
+++ b/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.cpp
@@ -50,7 +50,7 @@ TBsCostTracker::TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::E
     , ScrubDiskCost(CostCounters->GetCounter("ScrubDiskCost", true))
     , DefragDiskCost(CostCounters->GetCounter("DefragDiskCost", true))
     , InternalDiskCost(CostCounters->GetCounter("InternalDiskCost", true))
-    , Bucket(1'000'000'000, BucketCapacity)
+    , Bucket(&BucketInflow, &BucketCapacity, nullptr, nullptr, nullptr, nullptr, true)
 {
     BurstDetector.Initialize(CostCounters, "BurstDetector");
     switch (GroupType.GetErasure()) {
@@ -67,6 +67,7 @@ TBsCostTracker::TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::E
         CostModel = std::make_unique<TBsCostModelErasureNone>(diskType);
         break;
     }
+    UpdateBucketCapacity();
 }
 
 } // NKikimr
diff --git a/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.h b/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.h
index 5d486503ab..58e3f425eb 100644
--- a/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.h
+++ b/ydb/core/blobstorage/vdisk/common/blobstorage_cost_tracker.h
@@ -7,6 +7,7 @@
 
 #include <library/cpp/bucket_quoter/bucket_quoter.h>
 #include <util/system/compiler.h>
+#include <ydb/core/base/blobstorage.h>
 #include <ydb/core/blobstorage/base/blobstorage_events.h>
 #include <ydb/core/util/light.h>
 
@@ -50,6 +51,8 @@ public:
 
     virtual ~TBsCostModelBase() = default;
 
+    friend class TBsCostTracker;
+
 protected:
     NPDisk::EDeviceType DeviceType = NPDisk::DEVICE_TYPE_UNKNOWN;
 
@@ -292,10 +295,12 @@ private:
     ::NMonitoring::TDynamicCounters::TCounterPtr DefragDiskCost;
     ::NMonitoring::TDynamicCounters::TCounterPtr InternalDiskCost;
 
+    TAtomic BucketCapacity = 1'000'000'000;  // 10^9 nsec
+    TAtomic BucketInflow = 1'000'000'000;  // 10^9 nsec
     TBucketQuoter<i64, TSpinLock, THPTimerUs> Bucket;
-    static constexpr ui64 BucketCapacity = 1'000'000'000;
     TLight BurstDetector;
     std::atomic<ui64> SeqnoBurstDetector = 0;
+    static constexpr ui32 ConcurrentHugeRequestsAllowed = 3;
 
 public:
     TBsCostTracker(const TBlobStorageGroupType& groupType, NPDisk::EDeviceType diskType,
@@ -312,14 +317,31 @@ public:
         return cost;
     }
 
-    /// SETTINGS
-    void UpdateFromVDiskSettings(NKikimrBlobStorage::TVDiskCostSettings &settings) const;
+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) {