Move Inflight actors to ut_helpers (#1759)

* Move Inflight actors to ut_helpers

* Mute patch tests

* Remove unused assert

Author: serbel324

.github/config/muted_ya.txt

@@ -1,8 +1,7 @@
 ydb/core/blobstorage/dsproxy/ut TBlobStorageProxySequenceTest.TestBlock42PutWithChangingSlowDisk
 ydb/core/blobstorage/dsproxy/ut_fat TBlobStorageProxyTest.TestBatchedPutRequestDoesNotContainAHugeBlob
-ydb/core/blobstorage/ut_blobstorage CostMetricsGetBlock4Plus2.TestGet4Plus2BlockRequests10000Inflight1BlobSize1000
-ydb/core/blobstorage/ut_blobstorage CostMetricsPatchMirror3dc.*
 ydb/core/blobstorage/ut_blobstorage BurstDetection.*
+ydb/core/blobstorage/ut_blobstorage CostMetrics*
 ydb/core/client/ut TClientTest.PromoteFollower
 ydb/core/client/ut TClientTest.ReadFromFollower
 ydb/core/client/ut TFlatTest.AutoSplitMergeQueue

ydb/core/blobstorage/ut_blobstorage/monitoring.cpp

@@ -1,5 +1,6 @@
 #include <ydb/core/blobstorage/ut_blobstorage/lib/env.h>
 #include <ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.h>
+#include "ut_helpers.h"
 
 constexpr bool VERBOSE = false;
 
@@ -11,72 +12,6 @@ TString MakeData(ui32 dataSize) {
     return data;
 }
 
-template <typename TDerived>
-class TInflightActor : public TActorBootstrapped<TDerived> {
-public:
-    struct TSettings {
-        ui32 Requests;
-        ui32 MaxInFlight;
-        TDuration Delay = TDuration::Zero();
-    };
-
-public:
-    TInflightActor(TSettings settings)
-        : RequestsToSend(settings.Requests)
-        , RequestInFlight(settings.MaxInFlight)
-        , Settings(settings)
-    {}
-
-    virtual ~TInflightActor() = default;
-
-    void SetGroupId(ui32 groupId) {
-        GroupId = groupId;
-    }
-    void Bootstrap(const TActorContext &ctx) {
-        BootstrapImpl(ctx);
-    }
-
-protected:
-    void ScheduleRequests() {
-        while (RequestInFlight > 0 && RequestsToSend > 0) {
-            TMonotonic now = TMonotonic::Now();
-            TDuration timePassed = now - LastTs;
-            if (timePassed >= Settings.Delay) {
-                LastTs = now;
-                RequestInFlight--;
-                RequestsToSend--;
-                SendRequest();
-            } else {
-                TActorBootstrapped<TDerived>::Schedule(Settings.Delay - timePassed, new TEvents::TEvWakeup);
-            }
-        }
-    }
-
-    void HandleReply(NKikimrProto::EReplyStatus status) {
-        if (status == NKikimrProto::OK) {
-            OKs++;
-        } else {
-            Fails++;
-        }
-        ++RequestInFlight;
-        ScheduleRequests();
-    }
-
-    virtual void BootstrapImpl(const TActorContext &ctx) = 0;
-    virtual void SendRequest() = 0;
-
-protected:
-    ui32 RequestsToSend;
-    ui32 RequestInFlight;
-    ui32 GroupId;
-    TMonotonic LastTs;
-    TSettings Settings;
-
-public:
-    ui32 OKs = 0;
-    ui32 Fails = 0;
-};
-
 ui64 AggregateVDiskCounters(std::unique_ptr<TEnvironmentSetup>& env, const NKikimrBlobStorage::TBaseConfig& baseConfig,
         TString storagePool, ui32 groupSize, ui32 groupId, const std::vector<ui32>& pdiskLayout, TString subsystem,
         TString counter, bool derivative = false) {
@@ -168,8 +103,8 @@ void TestDSProxyAndVDiskEqualCost(const TBlobStorageGroupInfo::TTopology& topolo
     TStringStream str;
     double proportion = 1. * dsproxyCost / vdiskCost;
     i64 diff = (i64)dsproxyCost - vdiskCost;
-    str << "OKs# " << actor->OKs << ", Fails# " << actor->Fails << ", Cost on dsproxy# "
-            << dsproxyCost << ", Cost on vdisks# " << vdiskCost << ", proportion# " << proportion
+    str << "OKs# " << actor->ResponsesByStatus[NKikimrProto::OK] << ", Errors# " << actor->ResponsesByStatus[NKikimrProto::ERROR]
+            << ", Cost on dsproxy# " << dsproxyCost << ", Cost on vdisks# " << vdiskCost << ", proportion# " << proportion
             << " diff# " << diff;
 
     if constexpr(VERBOSE) {
@@ -179,43 +114,6 @@ void TestDSProxyAndVDiskEqualCost(const TBlobStorageGroupInfo::TTopology& topolo
     UNIT_ASSERT_VALUES_EQUAL_C(dsproxyCost, vdiskCost, str.Str());
 }
 
-class TInflightActorPut : public TInflightActor<TInflightActorPut> {
-public:
-    TInflightActorPut(TSettings settings, ui32 dataSize = 1024)
-        : TInflightActor(settings)
-        , DataSize(dataSize)
-    {}
-
-    STRICT_STFUNC(StateWork,
-        cFunc(TEvBlobStorage::TEvStatusResult::EventType, ScheduleRequests);
-        cFunc(TEvents::TEvWakeup::EventType, ScheduleRequests);
-        hFunc(TEvBlobStorage::TEvPutResult, Handle);
-    )
-
-    virtual void BootstrapImpl(const TActorContext&/* ctx*/) override {
-        // dummy request to establish the session
-        auto ev = new TEvBlobStorage::TEvStatus(TInstant::Max());
-        SendToBSProxy(SelfId(), GroupId, ev, 0);
-        Become(&TInflightActorPut::StateWork);
-    }
-
-protected:
-    virtual void SendRequest() override {
-        TString data = MakeData(DataSize);
-        auto ev = new TEvBlobStorage::TEvPut(TLogoBlobID(1, 1, 1, 10, DataSize, RequestsToSend + 1),
-                data, TInstant::Max(), NKikimrBlobStorage::UserData);
-        SendToBSProxy(SelfId(), GroupId, ev, 0);
-    }
-
-    void Handle(TEvBlobStorage::TEvPutResult::TPtr res) {
-        HandleReply(res->Get()->Status);
-    }
-
-private:
-    std::string Data;
-    ui32 DataSize;
-};
-
 #define MAKE_TEST(erasure, requestType, requests, inflight)                         \
 Y_UNIT_TEST(Test##requestType##erasure##Requests##requests##Inflight##inflight) {   \
     auto groupType = TBlobStorageGroupType::Erasure##erasure;                       \
@@ -236,99 +134,6 @@ Y_UNIT_TEST(Test##requestType##erasure##Requests##requests##Inflight##inflight##
     TestDSProxyAndVDiskEqualCost(topology, actor);                                                      \
 }
 
-class TInflightActorGet : public TInflightActor<TInflightActorGet> {
-public:
-    TInflightActorGet(TSettings settings, ui32 dataSize = 1024)
-        : TInflightActor(settings)
-        , DataSize(dataSize)
-    {}
-
-    STRICT_STFUNC(StateWork,
-        cFunc(TEvBlobStorage::TEvPutResult::EventType, ScheduleRequests);
-        cFunc(TEvents::TEvWakeup::EventType, ScheduleRequests);
-        hFunc(TEvBlobStorage::TEvGetResult, Handle);
-    )
-
-    virtual void BootstrapImpl(const TActorContext&/* ctx*/) override {
-        TString data = MakeData(DataSize);
-        BlobId = TLogoBlobID(1, 1, 1, 10, DataSize, 1);
-        auto ev = new TEvBlobStorage::TEvPut(BlobId, data, TInstant::Max());
-        SendToBSProxy(SelfId(), GroupId, ev, 0);
-        Become(&TInflightActorGet::StateWork);
-    }
-
-protected:
-    virtual void SendRequest() override {
-        auto ev = new TEvBlobStorage::TEvGet(BlobId, 0, 10, TInstant::Max(), NKikimrBlobStorage::EGetHandleClass::FastRead);
-        SendToBSProxy(SelfId(), GroupId, ev, 0);
-    }
-
-    void Handle(TEvBlobStorage::TEvGetResult::TPtr res) {
-        HandleReply(res->Get()->Status);
-    }
-
-private:
-    TLogoBlobID BlobId;
-    std::string Data;
-    ui32 DataSize;
-};
-
-class TInflightActorPatch : public TInflightActor<TInflightActorPatch> {
-public:
-    TInflightActorPatch(TSettings settings, ui32 dataSize = 1024)
-        : TInflightActor(settings)
-        , DataSize(dataSize)
-    {}
-
-    STRICT_STFUNC(StateWork,
-        hFunc(TEvBlobStorage::TEvPatchResult, Handle);
-        hFunc(TEvBlobStorage::TEvPutResult, Handle);
-    )
-
-    virtual void BootstrapImpl(const TActorContext&/* ctx*/) override {
-        TString data = MakeData(DataSize);
-        for (ui32 i = 0; i < RequestInFlight; ++i) {
-            TLogoBlobID blobId(1, 1, 1, 10, DataSize, 1 + i);
-            auto ev = new TEvBlobStorage::TEvPut(blobId, data, TInstant::Max());
-            SendToBSProxy(SelfId(), GroupId, ev, 0);
-        }
-        Become(&TInflightActorPatch::StateWork);
-    }
-
-protected:
-    virtual void SendRequest() override {
-        TLogoBlobID oldId = Blobs.front();
-        Blobs.pop_front();
-        TLogoBlobID newId(1, 1, oldId.Step() + 1, 10, DataSize, oldId.Cookie());
-        Y_ABORT_UNLESS(TEvBlobStorage::TEvPatch::GetBlobIdWithSamePlacement(oldId, &newId, BlobIdMask, GroupId, GroupId));
-        TArrayHolder<TEvBlobStorage::TEvPatch::TDiff> diffs(new TEvBlobStorage::TEvPatch::TDiff[1]);
-        char c = 'a' + RequestsToSend % 26;
-        diffs[0].Set(TString(DataSize, c), 0);
-        auto ev = new TEvBlobStorage::TEvPatch(GroupId, oldId, newId, BlobIdMask, std::move(diffs), 1, TInstant::Max());
-        SendToBSProxy(SelfId(), GroupId, ev, 0);
-    }
-
-
-    void Handle(TEvBlobStorage::TEvPatchResult::TPtr res) {
-        Blobs.push_back(res->Get()->Id);
-        HandleReply(res->Get()->Status);
-    }
-
-    void Handle(TEvBlobStorage::TEvPutResult::TPtr res) {
-        Blobs.push_back(res->Get()->Id);
-        if (++BlobsWritten == RequestInFlight) {
-            ScheduleRequests();
-        }
-    }
-
-protected:
-    std::deque<TLogoBlobID> Blobs;
-    ui32 BlobIdMask = TLogoBlobID::MaxCookie & 0xfffff000;
-    ui32 BlobsWritten = 0;
-    std::string Data;
-    ui32 DataSize;
-};
-
 Y_UNIT_TEST_SUITE(CostMetricsPutMirror3dc) {
     MAKE_TEST_W_DATASIZE(Mirror3dc, Put, 1, 1, 1000);
     MAKE_TEST_W_DATASIZE(Mirror3dc, Put, 10, 1, 1000);

ydb/core/blobstorage/ut_blobstorage/ut_helpers.cpp

@@ -0,0 +1,15 @@
+#include "ut_helpers.h"
+
+namespace NKikimr {
+
+TString MakeData(ui32 dataSize) {
+    TString data(dataSize, '\0');
+    for (ui32 i = 0; i < dataSize; ++i) {
+        data[i] = 'A' + (i % 26);
+    }
+    return data;
+}
+
+ui64 TInflightActor::Cookie = 1;
+
+} // namespace NKikimr