25-1: introduce VDisk syncronization broker (ydb-platform#15710) (ydb-platform#16861)

alexd65536 · web-flow · commit 83d7a3c79bf0 · 2025-04-08T13:51:06.000+03:00
diff --git a/ydb/core/base/blobstorage.h b/ydb/core/base/blobstorage.h
@@ -757,6 +757,9 @@ struct TEvBlobStorage {
         EvHugeQueryForbiddenChunks,
         EvHugeForbiddenChunks,
         EvContinueShred,
+        EvQuerySyncToken,
+        EvSyncToken,
+        EvReleaseSyncToken,
 
         EvYardInitResult = EvPut + 9 * 512,                     /// 268 636 672
         EvLogResult,
diff --git a/ydb/core/base/services/blobstorage_service_id.h b/ydb/core/base/services/blobstorage_service_id.h
@@ -76,6 +76,11 @@ inline TActorId MakeBlobStorageReplBrokerID() {
     return TActorId(0, TStringBuf(x, 12));
 }
 
+inline TActorId MakeBlobStorageSyncBrokerID() {
+    char x[12] = {'b', 's', 's', 'y', 'n', 'c', 'b', 'r', 'o', 'k', 'e', 'r'};
+    return TActorId(0, TStringBuf(x, 12));
+}
+
 inline TActorId MakeBlobStorageNodeWardenID(ui32 node) {
     char x[12] = {'b','s','n','o','d','e','c','n','t','r','l','r'};
     return TActorId(node, TStringBuf(x, 12));
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_impl.cpp b/ydb/core/blobstorage/nodewarden/node_warden_impl.cpp
@@ -10,6 +10,7 @@
 #include <ydb/core/blobstorage/dsproxy/dsproxy_nodemonactor.h>
 #include <ydb/core/blobstorage/pdisk/drivedata_serializer.h>
 #include <ydb/core/blobstorage/vdisk/repl/blobstorage_replbroker.h>
+#include <ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_broker.h>
 #include <ydb/library/pdisk_io/file_params.h>
 #include <ydb/core/mind/bscontroller/yaml_config_helpers.h>
 #include <ydb/core/base/nameservice.h>
@@ -44,6 +45,7 @@ TNodeWarden::TNodeWarden(const TIntrusivePtr<TNodeWardenConfig> &cfg)
     , ThrottlingMaxOccupancyPerMille(950, 1, 1000)
     , ThrottlingMinLogChunkCount(100, 1, 1000)
     , ThrottlingMaxLogChunkCount(130, 1, 1000)
+    , MaxInProgressSyncCount(0, 0, 1000)
     , MaxCommonLogChunksHDD(200, 1, 1'000'000)
     , MaxCommonLogChunksSSD(200, 1, 1'000'000)
     , CostMetricsParametersByMedia({
@@ -370,6 +372,8 @@ void TNodeWarden::Bootstrap() {
         icb->RegisterSharedControl(ThrottlingMinLogChunkCount, "VDiskControls.ThrottlingMinLogChunkCount");
         icb->RegisterSharedControl(ThrottlingMaxLogChunkCount, "VDiskControls.ThrottlingMaxLogChunkCount");
 
+        icb->RegisterSharedControl(MaxInProgressSyncCount, "VDiskControls.MaxInProgressSyncCount");
+
         icb->RegisterSharedControl(MaxCommonLogChunksHDD, "PDiskControls.MaxCommonLogChunksHDD");
         icb->RegisterSharedControl(MaxCommonLogChunksSSD, "PDiskControls.MaxCommonLogChunksSSD");
 
@@ -424,6 +428,9 @@ void TNodeWarden::Bootstrap() {
     const ui64 maxBytes = replBrokerConfig.GetMaxInFlightReadBytes();
     actorSystem->RegisterLocalService(MakeBlobStorageReplBrokerID(), Register(CreateReplBrokerActor(maxBytes)));
 
+    actorSystem->RegisterLocalService(MakeBlobStorageSyncBrokerID(), Register(
+        CreateSyncBrokerActor(MaxInProgressSyncCount)));
+
     // determine if we are running in 'mock' mode
     EnableProxyMock = Cfg->BlobStorageConfig.GetServiceSet().GetEnableProxyMock();
 
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_impl.h b/ydb/core/blobstorage/nodewarden/node_warden_impl.h
@@ -216,6 +216,8 @@ namespace NKikimr::NStorage {
         TControlWrapper ThrottlingMinLogChunkCount;
         TControlWrapper ThrottlingMaxLogChunkCount;
 
+        TControlWrapper MaxInProgressSyncCount;
+
         TControlWrapper MaxCommonLogChunksHDD;
         TControlWrapper MaxCommonLogChunksSSD;
 
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp b/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
@@ -212,6 +212,8 @@ namespace NKikimr::NStorage {
         vdiskConfig->ThrottlingMinLogChunkCount = ThrottlingMinLogChunkCount;
         vdiskConfig->ThrottlingMaxLogChunkCount = ThrottlingMaxLogChunkCount;
 
+        vdiskConfig->MaxInProgressSyncCount = MaxInProgressSyncCount;
+
         vdiskConfig->CostMetricsParametersByMedia = CostMetricsParametersByMedia;
 
         vdiskConfig->FeatureFlags = Cfg->FeatureFlags;
diff --git a/ydb/core/blobstorage/vdisk/common/vdisk_config.h b/ydb/core/blobstorage/vdisk/common/vdisk_config.h
@@ -258,6 +258,9 @@ namespace NKikimr {
         TControlWrapper ThrottlingMinLogChunkCount;
         TControlWrapper ThrottlingMaxLogChunkCount;
 
+        ///////////// SYNC SETTINGS //////////////////
+        TControlWrapper MaxInProgressSyncCount;
+
         ///////////// FEATURE FLAGS ////////////////////////
         NKikimrConfig::TFeatureFlags FeatureFlags;
 
diff --git a/ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_broker.cpp b/ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_broker.cpp
@@ -0,0 +1,153 @@
+#include "blobstorage_syncer_broker.h"
+
+#include <ydb/core/blobstorage/vdisk/common/vdisk_log.h>
+#include <ydb/core/control/immediate_control_board_wrapper.h>
+
+namespace NKikimr {
+
+    class TSyncBroker : public TActorBootstrapped<TSyncBroker> {
+        TControlWrapper MaxInProgressSyncCount;
+
+        std::unordered_map<TActorId, std::unordered_set<TActorId>> Active;
+
+        struct TWaitSync {
+            TActorId VDiskActorId;
+            std::unordered_set<TActorId> ActorIds;
+        };
+        std::list<TWaitSync> WaitQueue; // TODO: better search
+
+    public:
+        static constexpr auto ActorActivityType() {
+            return NKikimrServices::TActivity::BS_SYNC_BROKER;
+        }
+
+        STRICT_STFUNC(StateFunc,
+            hFunc(TEvQuerySyncToken, Handle)
+            hFunc(TEvReleaseSyncToken, Handle)
+            hFunc(TEvents::TEvWakeup, Handle)
+        )
+
+        explicit TSyncBroker(const TControlWrapper& maxInProgressSyncCount)
+            : MaxInProgressSyncCount(maxInProgressSyncCount)
+        {}
+
+        void Bootstrap() {
+            Become(&TThis::StateFunc, TDuration::MilliSeconds(100), new TEvents::TEvWakeup);
+        }
+
+        void Handle(TEvQuerySyncToken::TPtr& ev) {
+            const auto vDiskActorId = ev->Get()->VDiskActorId;
+            const auto actorId = ev->Sender;
+
+            if (const auto it = Active.find(vDiskActorId); it != Active.end()) {
+                it->second.insert(actorId);
+                Send(actorId, new TEvSyncToken);
+
+                LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_SYNCER,
+                    "TEvQuerySyncToken, VDisk actor id: " << vDiskActorId <<
+                    ", actor id: " << actorId <<
+                    ", token sent, active: " << Active.size() <<
+                    ", waiting: " << WaitQueue.size());
+                return;
+            }
+
+            const auto limit = (ui64)MaxInProgressSyncCount;
+
+            if (!limit || Active.size() < limit) {
+                Active[vDiskActorId].insert(actorId);
+                Send(actorId, new TEvSyncToken);
+
+                LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_SYNCER,
+                    "TEvQuerySyncToken, VDisk actor id: " << vDiskActorId <<
+                    ", actor id: " << actorId <<
+                    ", token sent, active: " << Active.size() <<
+                    ", waiting: " << WaitQueue.size());
+                return;
+            }
+
+            auto pred = [&vDiskActorId](const auto& item) {
+                return item.VDiskActorId == vDiskActorId;
+            };
+
+            if (const auto it = std::find_if(WaitQueue.begin(), WaitQueue.end(), pred); it != WaitQueue.end()) {
+                it->ActorIds.insert(actorId);
+
+                LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_SYNCER,
+                    "TEvQuerySyncToken, VDisk actor id: " << vDiskActorId <<
+                    ", actor id: " << actorId <<
+                    ", enqueued, active: " << Active.size() <<
+                    ", waiting: " << WaitQueue.size());
+                return;
+            }
+            
+            TWaitSync sync{vDiskActorId, {actorId}};
+            WaitQueue.emplace_back(std::move(sync));
+
+            LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_SYNCER,
+                "TEvQuerySyncToken, VDisk actor id: " << vDiskActorId <<
+                ", actor id: " << actorId <<
+                ", enqueued, active: " << Active.size() <<
+                ", waiting: " << WaitQueue.size());
+        }
+
+        void ProcessQueue() {
+            const auto limit = (ui64)MaxInProgressSyncCount;
+
+            while (!WaitQueue.empty() && (!limit || Active.size() < limit)) {
+                const auto& waitSync = WaitQueue.front();
+                for (const auto& actorId : waitSync.ActorIds) {
+                    Send(actorId, new TEvSyncToken);
+                }
+                Active[waitSync.VDiskActorId] = std::move(waitSync.ActorIds);
+                WaitQueue.pop_front();
+            }
+        }
+
+        void Handle(TEvReleaseSyncToken::TPtr& ev) {
+            const auto vDiskActorId = ev->Get()->VDiskActorId;
+            const auto actorId = ev->Sender;
+
+            if (const auto it = Active.find(vDiskActorId); it != Active.end()) {
+                it->second.erase(actorId);
+                if (it->second.empty()) {
+                    Active.erase(vDiskActorId);
+                    ProcessQueue();
+                }
+
+                LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_SYNCER,
+                    "TEvReleaseSyncToken, VDisk actor id: " << vDiskActorId <<
+                    ", actor id: " << actorId <<
+                    ", token released, active: " << Active.size() <<
+                    ", waiting: " << WaitQueue.size());
+                return;
+            }
+
+            auto pred = [&vDiskActorId](const auto& item) {
+                return item.VDiskActorId == vDiskActorId;
+            };
+
+            if (const auto it = std::find_if(WaitQueue.begin(), WaitQueue.end(), pred); it != WaitQueue.end()) {
+                it->ActorIds.erase(actorId);
+                if (it->ActorIds.empty()) {
+                    WaitQueue.erase(it);
+                }
+
+                LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::BS_SYNCER,
+                    "TEvReleaseSyncToken, VDisk actor id: " << vDiskActorId <<
+                    ", actor id: " << actorId <<
+                    ", removed from queue, active: " << Active.size() <<
+                    ", waiting: " << WaitQueue.size());
+            }
+        }
+
+        void Handle(TEvents::TEvWakeup::TPtr&) {
+            ProcessQueue();
+            Schedule(TDuration::MilliSeconds(100), new TEvents::TEvWakeup);
+        }
+    };
+
+    IActor *CreateSyncBrokerActor(const TControlWrapper& maxInProgressSyncCount) {
+        return new TSyncBroker(maxInProgressSyncCount);
+    }
+
+} // NKikimr
diff --git a/ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_broker.h b/ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_broker.h
@@ -0,0 +1,32 @@
+#pragma once
+
+#include "defs.h"
+#include <ydb/core/base/blobstorage.h>
+#include <ydb/core/blobstorage/base/blobstorage_vdiskid.h>
+
+namespace NKikimr {
+
+    struct TEvQuerySyncToken : public TEventLocal<TEvQuerySyncToken, TEvBlobStorage::EvQuerySyncToken> {
+        TActorId VDiskActorId;
+
+        explicit TEvQuerySyncToken(const TActorId& id)
+            : VDiskActorId(id)
+        {}
+    };
+
+    struct TEvSyncToken : public TEventLocal<TEvSyncToken, TEvBlobStorage::EvSyncToken>
+    {};
+
+    struct TEvReleaseSyncToken : public TEventLocal<TEvReleaseSyncToken, TEvBlobStorage::EvReleaseSyncToken> {
+        TActorId VDiskActorId;
+
+        explicit TEvReleaseSyncToken(const TActorId& id)
+            : VDiskActorId(id)
+        {}
+    };
+
+    class TControlWrapper;
+
+    extern IActor *CreateSyncBrokerActor(const TControlWrapper& maxInProgressSyncCount);
+
+} // NKikimr
diff --git a/ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_recoverlostdata_proxy.cpp b/ydb/core/blobstorage/vdisk/syncer/blobstorage_syncer_recoverlostdata_proxy.cpp
@@ -1,4 +1,5 @@
 #include "blobstorage_syncer_recoverlostdata_proxy.h"
+#include "blobstorage_syncer_broker.h"
 #include "blobstorage_syncer_committer.h"
 #include "blobstorage_syncer_data.h"
 #include "blobstorage_syncquorum.h"
@@ -45,6 +46,39 @@ namespace NKikimr {
         TVDiskID TargetVDiskId;
         TActorId TargetActorId;
 
+        void Bootstrap(const TActorContext &ctx) {
+            LOG_DEBUG(ctx, BS_SYNCER,
+                      VDISKP(SyncerCtx->VCtx->VDiskLogPrefix,
+                        "TSyncerRLDFullSyncProxyActor(%s): START",
+                            TargetVDiskId.ToString().data()));
+            ctx.Send(MakeBlobStorageSyncBrokerID(), new TEvQuerySyncToken(SyncerCtx->VCtx->VDiskActorId),
+                IEventHandle::FlagTrackDelivery);
+            Become(&TThis::WaitForBrokerStateFunc);
+        }
+
+        void Handle(TEvSyncToken::TPtr &ev, const TActorContext &ctx) {
+            Y_UNUSED(ev);
+            LOG_DEBUG(ctx, BS_SYNCER,
+                      VDISKP(SyncerCtx->VCtx->VDiskLogPrefix,
+                        "TSyncerRLDFullSyncProxyActor(%s): TEvSyncToken received",
+                            TargetVDiskId.ToString().data()));
+            CreateAndRunTask(ctx);
+        }
+
+        void Handle(TEvents::TEvUndelivered::TPtr &ev, const TActorContext &ctx) {
+            // no sync broker service
+            if (ev->Get()->SourceType == TEvQuerySyncToken::EventType) {
+                CreateAndRunTask(ctx);
+            }
+        }
+
+        STRICT_STFUNC(WaitForBrokerStateFunc,
+            HFunc(TEvents::TEvPoisonPill, HandlePoison)
+            HFunc(TEvSyncToken, Handle)
+            HFunc(TEvents::TEvUndelivered, Handle)
+            HFunc(TEvVGenerationChange, Handle)
+        )
+
         void CreateAndRunTask(const TActorContext &ctx) {
             // create task
             auto task = std::make_unique<TSyncerJobTask>(TSyncerJobTask::EFullRecover, TargetVDiskId, TargetActorId,
@@ -56,17 +90,8 @@ namespace NKikimr {
             Become(&TThis::WaitForSyncStateFunc);
         }
 
-        void Bootstrap(const TActorContext &ctx) {
-            LOG_DEBUG(ctx, BS_SYNCER,
-                      VDISKP(SyncerCtx->VCtx->VDiskLogPrefix,
-                        "TSyncerRLDFullSyncProxyActor(%s): START",
-                            TargetVDiskId.ToString().data()));
-
-            // run job
-            CreateAndRunTask(ctx);
-        }
-
         void Handle(TEvSyncerJobDone::TPtr &ev, const TActorContext &ctx) {
+            ctx.Send(MakeBlobStorageSyncBrokerID(), new TEvReleaseSyncToken(SyncerCtx->VCtx->VDiskActorId));
             LOG_DEBUG(ctx, BS_SYNCER,
                       VDISKP(SyncerCtx->VCtx->VDiskLogPrefix,
                         "TSyncerRLDFullSyncProxyActor(%s): TEvSyncerJobDone; Task# %s",
@@ -103,8 +128,9 @@ namespace NKikimr {
 
         void Handle(TEvSyncerRLDWakeup::TPtr &ev, const TActorContext &ctx) {
             Y_UNUSED(ev);
-            // run job again
-            CreateAndRunTask(ctx);
+            ctx.Send(MakeBlobStorageSyncBrokerID(), new TEvQuerySyncToken(SyncerCtx->VCtx->VDiskActorId),
+                IEventHandle::FlagTrackDelivery);
+            Become(&TThis::WaitForBrokerStateFunc);
         }
 
         STRICT_STFUNC(WaitForTimeoutStateFunc,
@@ -147,6 +173,7 @@ namespace NKikimr {
         // HandlePoison
         ////////////////////////////////////////////////////////////////////////
         void HandlePoison(TEvents::TEvPoisonPill::TPtr &ev, const TActorContext &ctx) {
+            ctx.Send(MakeBlobStorageSyncBrokerID(), new TEvReleaseSyncToken(SyncerCtx->VCtx->VDiskActorId));
             LOG_DEBUG(ctx, BS_SYNCER,
                      VDISKP(SyncerCtx->VCtx->VDiskLogPrefix,
                         "TSyncerRLDFullSyncProxyActor(%s): PoisonPill",
diff --git a/ydb/core/blobstorage/vdisk/syncer/ya.make b/ydb/core/blobstorage/vdisk/syncer/ya.make
@@ -10,11 +10,14 @@ PEERDIR(
     ydb/core/blobstorage/vdisk/anubis_osiris
     ydb/core/blobstorage/vdisk/common
     ydb/core/blobstorage/vdisk/hulldb/base
+    ydb/core/control
     ydb/core/driver_lib/version
 )
 
 SRCS(
     defs.h
+    blobstorage_syncer_broker.cpp
+    blobstorage_syncer_broker.h
     blobstorage_syncer_committer.cpp
     blobstorage_syncer_committer.h
     blobstorage_syncer.cpp
diff --git a/ydb/core/blobstorage/vdisk/vdisk_services.h b/ydb/core/blobstorage/vdisk/vdisk_services.h
@@ -7,6 +7,6 @@ namespace NKikimr {
     // Some services to run on a node for correct functionalily of VDisks
 
     extern IActor *CreateReplBrokerActor(ui64 maxMemBytes);
+    extern IActor *CreateSyncBrokerActor(const TControlWrapper& maxInProgressSyncCount);
 
 } // NKikimr
-
diff --git a/ydb/core/persqueue/ut/partition_ut.cpp b/ydb/core/persqueue/ut/partition_ut.cpp
@@ -1824,6 +1824,8 @@ Y_UNIT_TEST_F(UserActCount, TPartitionFixture)
 
     CreatePartition();
 
+    Ctx->Runtime->SetScheduledLimit(6000);
+
     SendCreateSession(1, "client", "session-id", 2, 3);
     WaitCmdWrite({.Count=2, .UserInfos={{0, {.Session="session-id", .Offset=0, .Generation=2, .Step=3}}}});
     SendCmdWriteResponse(NMsgBusProxy::MSTATUS_OK);
diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto
diff --git a/ydb/library/services/services.proto b/ydb/library/services/services.proto
