Add UT for accelerate parameters, fix bugs

Author: serbel324

ydb/core/base/blobstorage.h

@@ -488,6 +488,8 @@ struct TEvBlobStorage {
         EvInplacePatch,
         EvAssimilate,
 
+        EvGetQueuesInfo,     // for debugging purposes
+
         //
         EvPutResult = EvPut + 512,                              /// 268 632 576
         EvGetResult,
@@ -502,6 +504,8 @@ struct TEvBlobStorage {
         EvInplacePatchResult,
         EvAssimilateResult,
 
+        EvQueuesInfo,  // for debugging purposes
+
         // proxy <-> vdisk interface
         EvVPut = EvPut + 2 * 512,                               /// 268 633 088
         EvVGet,

ydb/core/blobstorage/backpressure/common.h

@@ -15,3 +15,39 @@
 #define QLOG_DEBUG_S(marker, arg)  QLOG_LOG_S(marker, NActors::NLog::PRI_DEBUG , arg)
 
 LWTRACE_USING(BLOBSTORAGE_PROVIDER);
+
+namespace NKikimr::NBsQueue {
+
+// Special timer for debug purposes, which works with virtual time of TTestActorSystem
+struct TActivationContextTimer {
+    TActivationContextTimer()
+        : CreationTimestamp(NActors::TActivationContext::Monotonic())
+    {}
+
+    double Passed() const {
+        return (NActors::TActivationContext::Monotonic() - CreationTimestamp).SecondsFloat();
+    }
+
+    TMonotonic CreationTimestamp;
+};
+
+struct TBSQueueTimer {
+    TBSQueueTimer(bool useActorSystemTime)
+    {
+        if (useActorSystemTime) {
+            Timer.emplace<TActivationContextTimer>();
+        } else {
+            Timer.emplace<THPTimer>();
+        }
+    }
+
+    std::variant<THPTimer, TActivationContextTimer> Timer;
+
+    double Passed() const {
+        return std::visit([](const auto& timer) -> double {
+            return timer.Passed();
+        }, Timer);
+    }
+};
+
+} // namespace NKikimr::NBsQueue

ydb/core/blobstorage/backpressure/event.cpp

@@ -27,7 +27,7 @@ IEventBase *TEventHolder::MakeErrorReply(NKikimrProto::EReplyStatus status, cons
 
 void TEventHolder::SendToVDisk(const TActorContext& ctx, const TActorId& remoteVDisk, ui64 queueCookie, ui64 msgId,
         ui64 sequenceId, bool sendMeCostSettings, NWilson::TTraceId traceId, const NBackpressure::TQueueClientId& clientId,
-        const THPTimer& processingTimer) {
+        const TBSQueueTimer& processingTimer) {
     // check that we are not discarded yet
     Y_ABORT_UNLESS(Type != 0);
 

ydb/core/blobstorage/backpressure/event.h

@@ -142,7 +142,7 @@ class TEventHolder {
 
     void SendToVDisk(const TActorContext& ctx, const TActorId& remoteVDisk, ui64 queueCookie, ui64 msgId, ui64 sequenceId,
             bool sendMeCostSettings, NWilson::TTraceId traceId, const NBackpressure::TQueueClientId& clientId,
-            const THPTimer& processingTimer);
+            const TBSQueueTimer& processingTimer);
 
     void Discard();
 };

ydb/core/blobstorage/backpressure/queue.cpp

@@ -4,7 +4,7 @@ namespace NKikimr::NBsQueue {
 
 TBlobStorageQueue::TBlobStorageQueue(const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters, TString& logPrefix,
         const TBSProxyContextPtr& bspctx, const NBackpressure::TQueueClientId& clientId, ui32 interconnectChannel,
-        const TBlobStorageGroupType& gType, NMonitoring::TCountableBase::EVisibility visibility)
+        const TBlobStorageGroupType& gType, NMonitoring::TCountableBase::EVisibility visibility, bool useActorSystemTime)
     : Queues(bspctx)
     , WindowSize(0)
     , InFlightCost(0)
@@ -16,6 +16,7 @@ TBlobStorageQueue::TBlobStorageQueue(const TIntrusivePtr<::NMonitoring::TDynamic
     , ClientId(clientId)
     , BytesWaiting(0)
     , InterconnectChannel(interconnectChannel)
+    , UseActorSystemTime(useActorSystemTime)
     // use parent group visibility
     , QueueWaitingItems(counters->GetCounter("QueueWaitingItems", false, visibility))
     , QueueWaitingBytes(counters->GetCounter("QueueWaitingBytes", false, visibility))

ydb/core/blobstorage/backpressure/queue.h

@@ -51,15 +51,16 @@ class TBlobStorageQueue {
         const ui64 QueueCookie;
         ui64 Cost;
         bool DirtyCost;
-        THPTimer ProcessingTimer;
+        TBSQueueTimer ProcessingTimer;
+
         TTrackableList<TItem>::iterator Iterator;
 
         template<typename TEvent>
         TItem(TAutoPtr<TEventHandle<TEvent>>& event, TInstant deadline,
                 const ::NMonitoring::TDynamicCounters::TCounterPtr& serItems,
                 const ::NMonitoring::TDynamicCounters::TCounterPtr& serBytes,
                 const TBSProxyContextPtr& bspctx, ui32 interconnectChannel,
-                bool local)
+                bool local, bool useActorSystemTime)
             : Queue(EItemQueue::NotSet)
             , CostEssence(*event->Get())
             , Span(TWilson::VDiskTopLevel, std::move(event->TraceId), "Backpressure.InFlight")
@@ -70,6 +71,7 @@ class TBlobStorageQueue {
             , QueueCookie(RandomNumber<ui64>())
             , Cost(0)
             , DirtyCost(true)
+            , ProcessingTimer(useActorSystemTime)
         {
             if (Span) {
                 Span
@@ -129,6 +131,8 @@ class TBlobStorageQueue {
 
     const ui32 InterconnectChannel;
 
+    const bool UseActorSystemTime;
+
 public:
     ::NMonitoring::TDynamicCounters::TCounterPtr QueueWaitingItems;
     ::NMonitoring::TDynamicCounters::TCounterPtr QueueWaitingBytes;
@@ -156,7 +160,8 @@ class TBlobStorageQueue {
     TBlobStorageQueue(const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters, TString& logPrefix,
             const TBSProxyContextPtr& bspctx, const NBackpressure::TQueueClientId& clientId, ui32 interconnectChannel,
             const TBlobStorageGroupType &gType,
-            NMonitoring::TCountableBase::EVisibility visibility = NMonitoring::TCountableBase::EVisibility::Public);
+            NMonitoring::TCountableBase::EVisibility visibility = NMonitoring::TCountableBase::EVisibility::Public,
+            bool useActorSystemTime = false);
 
     ~TBlobStorageQueue();
 
@@ -213,7 +218,8 @@ class TBlobStorageQueue {
         TItemList::iterator newIt;
         if (Queues.Unused.empty()) {
             newIt = Queues.Waiting.emplace(Queues.Waiting.end(), event, deadline,
-                QueueSerializedItems, QueueSerializedBytes, BSProxyCtx, InterconnectChannel, local);
+                QueueSerializedItems, QueueSerializedBytes, BSProxyCtx, InterconnectChannel, local,
+                UseActorSystemTime);
             ++*QueueSize;
         } else {
             newIt = Queues.Unused.begin();
@@ -222,7 +228,7 @@ class TBlobStorageQueue {
             TItem& item = *newIt;
             item.~TItem();
             new(&item) TItem(event, deadline, QueueSerializedItems, QueueSerializedBytes, BSProxyCtx,
-                InterconnectChannel, local);
+                InterconnectChannel, local, UseActorSystemTime);
         }
 
         newIt->Iterator = newIt;

ydb/core/blobstorage/backpressure/queue_backpressure_client.cpp

@@ -77,12 +77,13 @@ class TVDiskBackpressureClientActor : public TActorBootstrapped<TVDiskBackpressu
             NKikimrBlobStorage::EVDiskQueueId queueId,const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters,
             const TBSProxyContextPtr& bspctx, const NBackpressure::TQueueClientId& clientId, const TString& queueName,
             ui32 interconnectChannel, bool /*local*/, TDuration watchdogTimeout,
-            TIntrusivePtr<NBackpressure::TFlowRecord> &flowRecord, NMonitoring::TCountableBase::EVisibility visibility)
+            TIntrusivePtr<NBackpressure::TFlowRecord> &flowRecord, NMonitoring::TCountableBase::EVisibility visibility,
+            bool useActorSystemTime)
         : BSProxyCtx(bspctx)
         , QueueName(queueName)
         , Counters(counters->GetSubgroup("queue", queueName))
         , Queue(Counters, LogPrefix, bspctx, clientId, interconnectChannel,
-                (info ? info->Type : TErasureType::ErasureNone), visibility)
+                (info ? info->Type : TErasureType::ErasureNone), visibility, useActorSystemTime)
         , VDiskIdShort(vdiskId)
         , QueueId(queueId)
         , QueueWatchdogTimeout(watchdogTimeout)
@@ -975,9 +976,10 @@ IActor* CreateVDiskBackpressureClient(const TIntrusivePtr<TBlobStorageGroupInfo>
         NKikimrBlobStorage::EVDiskQueueId queueId,const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters,
         const TBSProxyContextPtr& bspctx, const NBackpressure::TQueueClientId& clientId, const TString& queueName,
         ui32 interconnectChannel, bool local, TDuration watchdogTimeout,
-        TIntrusivePtr<NBackpressure::TFlowRecord> &flowRecord, NMonitoring::TCountableBase::EVisibility visibility) {
+        TIntrusivePtr<NBackpressure::TFlowRecord> &flowRecord, NMonitoring::TCountableBase::EVisibility visibility,
+        bool useActorSystemTime) {
     return new NBsQueue::TVDiskBackpressureClientActor(info, vdiskId, queueId, counters, bspctx, clientId, queueName,
-        interconnectChannel, local, watchdogTimeout, flowRecord, visibility);
+        interconnectChannel, local, watchdogTimeout, flowRecord, visibility, useActorSystemTime);
 }
 
 } // NKikimr

ydb/core/blobstorage/backpressure/queue_backpressure_client.h

@@ -50,6 +50,7 @@ namespace NKikimr {
         NKikimrBlobStorage::EVDiskQueueId queueId,const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters,
         const TBSProxyContextPtr& bspctx, const NBackpressure::TQueueClientId& clientId, const TString& queueName,
         ui32 interconnectChannel, bool local, TDuration watchdogTimeout,
-        TIntrusivePtr<NBackpressure::TFlowRecord> &flowRecord, NMonitoring::TCountableBase::EVisibility visibility);
+        TIntrusivePtr<NBackpressure::TFlowRecord> &flowRecord, NMonitoring::TCountableBase::EVisibility visibility,
+        bool useActorSystemTime = false);
 
 } // NKikimr

ydb/core/blobstorage/dsproxy/dsproxy.h

@@ -54,7 +54,9 @@ const ui32 MaskSizeBits = 32;
 
 constexpr bool DefaultEnablePutBatching = true;
 constexpr bool DefaultEnableVPatch = false;
-constexpr float DefaultSlowDiskThreshold = 2000;
+
+constexpr float DefaultSlowDiskThreshold = 2;
+constexpr float DefaultPredictedDelayMultiplier = 1;
 
 constexpr bool WithMovingPatchRequestToStaticNode = true;
 
@@ -169,6 +171,11 @@ inline void SetExecutionRelay(IEventBase& ev, std::shared_ptr<TEvBlobStorage::TE
     }
 }
 
+struct TAccelerationParams {
+    double SlowDiskThreshold = 2;
+    double PredictedDelayMultiplier = 1;
+};
+
 template<typename TDerived>
 class TBlobStorageGroupRequestActor : public TActor<TDerived> {
 public:
@@ -649,7 +656,7 @@ IActor* CreateBlobStorageGroupPutRequest(const TIntrusivePtr<TBlobStorageGroupIn
     ui64 cookie, NWilson::TTraceId traceId, bool timeStatsEnabled,
     TDiskResponsivenessTracker::TPerDiskStatsPtr stats,
     TMaybe<TGroupStat::EKind> latencyQueueKind, TInstant now, TIntrusivePtr<TStoragePoolCounters> &storagePoolCounters,
-    bool enableRequestMod3x3ForMinLatecy, float slowDiskThreshold);
+    bool enableRequestMod3x3ForMinLatecy, const TAccelerationParams& accelerationParams);
 
 IActor* CreateBlobStorageGroupPutRequest(const TIntrusivePtr<TBlobStorageGroupInfo> &info,
     const TIntrusivePtr<TGroupQueues> &state,
@@ -659,14 +666,14 @@ IActor* CreateBlobStorageGroupPutRequest(const TIntrusivePtr<TBlobStorageGroupIn
     TDiskResponsivenessTracker::TPerDiskStatsPtr stats,
     TMaybe<TGroupStat::EKind> latencyQueueKind, TInstant now, TIntrusivePtr<TStoragePoolCounters> &storagePoolCounters,
     NKikimrBlobStorage::EPutHandleClass handleClass, TEvBlobStorage::TEvPut::ETactic tactic,
-    bool enableRequestMod3x3ForMinLatecy, float slowDiskThreshold);
+    bool enableRequestMod3x3ForMinLatecy, const TAccelerationParams& accelerationParams);
 
 IActor* CreateBlobStorageGroupGetRequest(const TIntrusivePtr<TBlobStorageGroupInfo> &info,
     const TIntrusivePtr<TGroupQueues> &state, const TActorId &source,
     const TIntrusivePtr<TBlobStorageGroupProxyMon> &mon, TEvBlobStorage::TEvGet *ev,
     ui64 cookie, NWilson::TTraceId traceId, TNodeLayoutInfoPtr&& nodeLayout,
     TMaybe<TGroupStat::EKind> latencyQueueKind, TInstant now, TIntrusivePtr<TStoragePoolCounters> &storagePoolCounters,
-    float slowDiskThreshold);
+    const TAccelerationParams& accelerationParams);
 
 IActor* CreateBlobStorageGroupPatchRequest(const TIntrusivePtr<TBlobStorageGroupInfo> &info,
     const TIntrusivePtr<TGroupQueues> &state, const TActorId &source,
@@ -729,12 +736,14 @@ IActor* CreateBlobStorageGroupAssimilateRequest(const TIntrusivePtr<TBlobStorage
 IActor* CreateBlobStorageGroupEjectedProxy(ui32 groupId, TIntrusivePtr<TDsProxyNodeMon> &nodeMon);
 
 IActor* CreateBlobStorageGroupProxyConfigured(TIntrusivePtr<TBlobStorageGroupInfo>&& info,
-    bool forceWaitAllDrives, TIntrusivePtr<TDsProxyNodeMon> &nodeMon,
+    bool forceWaitAllDrives, bool useActorSystemTimeInBSQueue, TIntrusivePtr<TDsProxyNodeMon> &nodeMon,
     TIntrusivePtr<TStoragePoolCounters>&& storagePoolCounters, const TControlWrapper &enablePutBatching,
-    const TControlWrapper &enableVPatch, const TControlWrapper &slowDiskThreshold);
+    const TControlWrapper &enableVPatch, const TControlWrapper &slowDiskThreshold,
+    const TControlWrapper &predictedDelayMultiplier);
 
-IActor* CreateBlobStorageGroupProxyUnconfigured(ui32 groupId, TIntrusivePtr<TDsProxyNodeMon> &nodeMon,
-    const TControlWrapper &enablePutBatching, const TControlWrapper &enableVPatch,
-    const TControlWrapper &slowDiskThreshold);
+IActor* CreateBlobStorageGroupProxyUnconfigured(ui32 groupId, bool useActorSystemInBSQueue,
+    TIntrusivePtr<TDsProxyNodeMon> &nodeMon, const TControlWrapper &enablePutBatching,
+    const TControlWrapper &enableVPatch, const TControlWrapper &slowDiskThreshold,
+    const TControlWrapper &predictedDelayMultiplier);
 
 }//NKikimr

ydb/core/blobstorage/dsproxy/dsproxy_blackboard.cpp

@@ -174,16 +174,22 @@ void TBlobState::AddNotYetResponse(const TBlobStorageGroupInfo &info, const TLog
 ui64 TBlobState::GetPredictedDelayNs(const TBlobStorageGroupInfo &info, TGroupQueues &groupQueues,
         ui32 diskIdxInSubring, NKikimrBlobStorage::EVDiskQueueId queueId) const {
     Y_UNUSED(info);
-    return groupQueues.GetPredictedDelayNsByOrderNumber(Disks[diskIdxInSubring].OrderNumber, queueId);
+    auto res = groupQueues.GetPredictedDelayNsByOrderNumber(Disks[diskIdxInSubring].OrderNumber, queueId);
+    return res;
 }
 
 void TBlobState::GetWorstPredictedDelaysNs(const TBlobStorageGroupInfo &info, TGroupQueues &groupQueues,
-        NKikimrBlobStorage::EVDiskQueueId queueId, ui32 nWorst, TDiskDelayPredictions *outNWorst) const {
+        NKikimrBlobStorage::EVDiskQueueId queueId, TDiskDelayPredictions *outNWorst,
+        double multiplier) const {
     outNWorst->resize(Disks.size());
     for (ui32 diskIdx = 0; diskIdx < Disks.size(); ++diskIdx) {
-        (*outNWorst)[diskIdx] = { GetPredictedDelayNs(info, groupQueues, diskIdx, queueId), diskIdx };
+        (*outNWorst)[diskIdx] = {
+            static_cast<ui64>(GetPredictedDelayNs(info, groupQueues, diskIdx, queueId) * multiplier),
+            diskIdx
+        };
     }
-    std::partial_sort(outNWorst->begin(), outNWorst->begin() + std::min(nWorst, (ui32)Disks.size()), outNWorst->end());
+    ui32 sortedPrefixSize = std::min(3u, (ui32)Disks.size());
+    std::partial_sort(outNWorst->begin(), outNWorst->begin() + sortedPrefixSize, outNWorst->end());
 }
 
 bool TBlobState::HasWrittenQuorum(const TBlobStorageGroupInfo& info, const TBlobStorageGroupInfo::TGroupVDisks& expired) const {
@@ -361,7 +367,7 @@ void TBlackboard::AddErrorResponse(const TLogoBlobID &id, ui32 orderNumber) {
 }
 
 EStrategyOutcome TBlackboard::RunStrategies(TLogContext &logCtx, const TStackVec<IStrategy*, 1>& s,
-        float slowDiskThreshold, TBatchedVec<TFinishedBlob> *finished,
+        const TAccelerationParams& accelerationParams, TBatchedVec<TFinishedBlob> *finished,
         const TBlobStorageGroupInfo::TGroupVDisks *expired) {
     for (auto it = BlobStates.begin(); it != BlobStates.end(); ) {
         auto& blob = it->second;
@@ -374,7 +380,7 @@ EStrategyOutcome TBlackboard::RunStrategies(TLogContext &logCtx, const TStackVec
         NKikimrProto::EReplyStatus status = NKikimrProto::OK;
         TString errorReason;
         for (IStrategy *strategy : s) {
-            switch (auto res = strategy->Process(logCtx, blob, *Info, *this, GroupDiskRequests, slowDiskThreshold)) {
+            switch (auto res = strategy->Process(logCtx, blob, *Info, *this, GroupDiskRequests, accelerationParams)) {
                 case EStrategyOutcome::IN_PROGRESS:
                     status = NKikimrProto::UNKNOWN;
                     break;
@@ -415,9 +421,10 @@ EStrategyOutcome TBlackboard::RunStrategies(TLogContext &logCtx, const TStackVec
     return BlobStates.empty() ? EStrategyOutcome::DONE : EStrategyOutcome::IN_PROGRESS;
 }
 
-EStrategyOutcome TBlackboard::RunStrategy(TLogContext &logCtx, const IStrategy& s, float slowDiskThreshold,
-        TBatchedVec<TFinishedBlob> *finished, const TBlobStorageGroupInfo::TGroupVDisks *expired) {
-    return RunStrategies(logCtx, {const_cast<IStrategy*>(&s)}, slowDiskThreshold, finished, expired);
+EStrategyOutcome TBlackboard::RunStrategy(TLogContext &logCtx, const IStrategy& s,
+        const TAccelerationParams& accelerationParams, TBatchedVec<TFinishedBlob> *finished,
+        const TBlobStorageGroupInfo::TGroupVDisks *expired) {
+    return RunStrategies(logCtx, {const_cast<IStrategy*>(&s)}, accelerationParams, finished, expired);
 }
 
 TBlobState& TBlackboard::GetState(const TLogoBlobID &id) {
@@ -459,13 +466,17 @@ void TBlackboard::ReportPartMapStatus(const TLogoBlobID &id, ssize_t partMapInde
 }
 
 void TBlackboard::GetWorstPredictedDelaysNs(const TBlobStorageGroupInfo &info, TGroupQueues &groupQueues,
-        NKikimrBlobStorage::EVDiskQueueId queueId, ui32 nWorst, TDiskDelayPredictions *outNWorst) const {
+        NKikimrBlobStorage::EVDiskQueueId queueId, TDiskDelayPredictions *outNWorst,
+        double multiplier) const {
     ui32 totalVDisks = info.GetTotalVDisksNum();
     outNWorst->resize(totalVDisks);
     for (ui32 orderNumber = 0; orderNumber < totalVDisks; ++orderNumber) {
-        (*outNWorst)[orderNumber] = { groupQueues.GetPredictedDelayNsByOrderNumber(orderNumber, queueId), orderNumber };
+        (*outNWorst)[orderNumber] = { 
+            static_cast<ui64>(groupQueues.GetPredictedDelayNsByOrderNumber(orderNumber, queueId) * multiplier),
+            orderNumber
+        };
     }
-    std::partial_sort(outNWorst->begin(), outNWorst->begin() + std::min(nWorst, totalVDisks), outNWorst->end());
+    std::partial_sort(outNWorst->begin(), outNWorst->begin() + std::min(3u, totalVDisks), outNWorst->end());
 }
 
 void TBlackboard::RegisterBlobForPut(const TLogoBlobID& id, size_t blobIdx) {