summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--ydb/core/mind/hive/hive.h40
-rw-r--r--ydb/core/mind/hive/hive_events.h4
-rw-r--r--ydb/core/mind/hive/hive_impl.cpp37
-rw-r--r--ydb/core/mind/hive/hive_impl.h7
-rw-r--r--ydb/core/mind/hive/tx__load_everything.cpp9
-rw-r--r--ydb/core/mind/hive/tx__update_dc_followers.cpp117
6 files changed, 162 insertions, 52 deletions
diff --git a/ydb/core/mind/hive/hive.h b/ydb/core/mind/hive/hive.h
index cd2528e7e7e..bf555106276 100644
--- a/ydb/core/mind/hive/hive.h
+++ b/ydb/core/mind/hive/hive.h
@@ -332,6 +332,46 @@ struct TNodeFilter {
bool IsAllowedDataCenter(TDataCenterId dc) const;
};
+struct TFollowerUpdates {
+ enum class EAction {
+ Create,
+ Update,
+ Delete,
+ };
+
+ struct TUpdate {
+ EAction Action;
+ TFullTabletId TabletId;
+ TFollowerGroupId GroupId;
+ TDataCenterId DataCenter;
+ };
+
+ std::deque<TUpdate> Updates;
+
+ bool Empty() const {
+ return Updates.empty();
+ }
+
+ void Create(TFullTabletId leaderTablet, TFollowerGroupId group, TDataCenterId dc) {
+ Updates.emplace_back(EAction::Create, leaderTablet, group, dc);
+ }
+
+ void Update(TFullTabletId tablet, TDataCenterId dc) {
+ Updates.emplace_back(EAction::Update, tablet, 0, dc);
+ }
+
+ void Delete(TFullTabletId tablet, TFollowerGroupId group, TDataCenterId dc) {
+ Updates.emplace_back(EAction::Delete, tablet, group, dc);
+ }
+
+ TUpdate Pop() {
+ TUpdate update = Updates.front();
+ Updates.pop_front();
+ return update;
+ }
+};
+
+
} // NHive
} // NKikimr
diff --git a/ydb/core/mind/hive/hive_events.h b/ydb/core/mind/hive/hive_events.h
index 5f5ca3ffa13..23e0ed6edc9 100644
--- a/ydb/core/mind/hive/hive_events.h
+++ b/ydb/core/mind/hive/hive_events.h
@@ -36,6 +36,7 @@ struct TEvPrivate {
EvUpdateDataCenterFollowers,
EvGenerateTestData,
EvRefreshScaleRecommendation,
+ EvUpdateFollowers,
EvEnd
};
@@ -133,6 +134,9 @@ struct TEvPrivate {
struct TEvGenerateTestData : TEventLocal<TEvGenerateTestData, EvGenerateTestData> {};
struct TEvRefreshScaleRecommendation : TEventLocal<TEvRefreshScaleRecommendation, EvRefreshScaleRecommendation> {};
+
+ struct TEvUpdateFollowers : TEventLocal<TEvUpdateFollowers, EvUpdateFollowers> {
+ };
};
} // NHive
diff --git a/ydb/core/mind/hive/hive_impl.cpp b/ydb/core/mind/hive/hive_impl.cpp
index a8dacbf0bc5..6522f7603c7 100644
--- a/ydb/core/mind/hive/hive_impl.cpp
+++ b/ydb/core/mind/hive/hive_impl.cpp
@@ -3072,6 +3072,7 @@ void THive::ProcessEvent(std::unique_ptr<IEventHandle> event) {
hFunc(TEvPrivate::TEvGenerateTestData, Handle);
hFunc(TEvPrivate::TEvRefreshScaleRecommendation, Handle);
hFunc(TEvHive::TEvConfigureScaleRecommender, Handle);
+ hFunc(TEvPrivate::TEvUpdateFollowers, Handle);
}
}
@@ -3179,6 +3180,7 @@ STFUNC(THive::StateWork) {
fFunc(TEvPrivate::TEvGenerateTestData::EventType, EnqueueIncomingEvent);
fFunc(TEvPrivate::TEvRefreshScaleRecommendation::EventType, EnqueueIncomingEvent);
fFunc(TEvHive::TEvConfigureScaleRecommender::EventType, EnqueueIncomingEvent);
+ fFunc(TEvPrivate::TEvUpdateFollowers::EventType, EnqueueIncomingEvent);
hFunc(TEvPrivate::TEvProcessIncomingEvent, Handle);
default:
if (!HandleDefaultEvents(ev, SelfId())) {
@@ -3480,7 +3482,40 @@ void THive::Handle(TEvHive::TEvRequestTabletDistribution::TPtr& ev) {
}
void THive::Handle(TEvPrivate::TEvUpdateDataCenterFollowers::TPtr& ev) {
- Execute(CreateUpdateDcFollowers(ev->Get()->DataCenter));
+ auto dataCenterId = ev->Get()->DataCenter;
+ auto& dataCenter = DataCenters[dataCenterId];
+ if (!dataCenter.UpdateScheduled) {
+ return;
+ }
+ dataCenter.UpdateScheduled = false;
+ if (dataCenter.IsRegistered()) {
+ for (auto& [tabletId, tablet] : Tablets) {
+ for (auto& group : tablet.FollowerGroups) {
+ auto& followers = dataCenter.Followers[{tabletId, group.Id}];
+ i64 neededCount = group.GetFollowerCountForDataCenter(dataCenterId);
+ i64 delta = neededCount - std::ssize(followers);
+ for (i64 i = 0; i < delta; ++i) {
+ BLOG_TRACE("UpdateDataCenterFollowers: Pending create follower for " << tabletId);
+ PendingFollowerUpdates.Create(tablet.GetFullTabletId(), group.Id, dataCenterId);
+ }
+ }
+ }
+ } else {
+ for (auto& [group, followers] : dataCenter.Followers) {
+ for (auto follower : followers) {
+ BLOG_TRACE("UpdateDataCenterFollowers: Pending delete follower for " << follower->GetFullTabletId());
+ PendingFollowerUpdates.Delete(follower->GetFullTabletId(), group.second, dataCenterId);
+ }
+ }
+ }
+ if (!PendingFollowerUpdates.Empty() && !ProcessFollowerUpdatesScheduled) {
+ Send(SelfId(), new TEvPrivate::TEvUpdateFollowers);
+ ProcessFollowerUpdatesScheduled = true;
+ }
+}
+
+void THive::Handle(TEvPrivate::TEvUpdateFollowers::TPtr&) {
+ Execute(CreateProcessUpdateFollowers());
}
void THive::MakeScaleRecommendation() {
diff --git a/ydb/core/mind/hive/hive_impl.h b/ydb/core/mind/hive/hive_impl.h
index d689464f580..b18bf921491 100644
--- a/ydb/core/mind/hive/hive_impl.h
+++ b/ydb/core/mind/hive/hive_impl.h
@@ -240,7 +240,7 @@ protected:
friend class TTxUpdateTabletGroups;
friend class TTxMonEvent_TabletAvailability;
friend class TLoggedMonTransaction;
- friend class TTxUpdateDcFollowers;
+ friend class TTxProcessUpdateFollowers;
friend class TDeleteTabletActor;
@@ -303,7 +303,7 @@ protected:
ITransaction* CreateRequestTabletOwners(TEvHive::TEvRequestTabletOwners::TPtr event);
ITransaction* CreateUpdateTabletsObject(TEvHive::TEvUpdateTabletsObject::TPtr event);
ITransaction* CreateUpdateDomain(TSubDomainKey subdomainKey, TEvHive::TEvUpdateDomain::TPtr event = {});
- ITransaction* CreateUpdateDcFollowers(const TDataCenterId& dc);
+ ITransaction* CreateProcessUpdateFollowers();
ITransaction* CreateGenerateTestData(uint64_t seed);
ITransaction* CreateDeleteNode(TNodeId nodeId);
ITransaction* CreateConfigureScaleRecommender(TEvHive::TEvConfigureScaleRecommender::TPtr event);
@@ -405,6 +405,7 @@ protected:
bool ProcessPendingOperationsScheduled = false;
bool LogTabletMovesScheduled = false;
bool ProcessStorageBalancerScheduled = false;
+ bool ProcessFollowerUpdatesScheduled = false;
TResourceRawValues TotalRawResourceValues = {};
TResourceNormalizedValues TotalNormalizedResourceValues = {};
TInstant LastResourceChangeReaction;
@@ -422,6 +423,7 @@ protected:
std::vector<TActorId> ActorsWaitingToMoveTablets;
std::queue<TActorId> NodePingQueue;
std::unordered_set<TNodeId> NodePingsInProgress;
+ TFollowerUpdates PendingFollowerUpdates;
struct TPendingCreateTablet {
NKikimrHive::TEvCreateTablet CreateTablet;
@@ -586,6 +588,7 @@ protected:
void Handle(TEvPrivate::TEvGenerateTestData::TPtr& ev);
void Handle(TEvPrivate::TEvRefreshScaleRecommendation::TPtr& ev);
void Handle(TEvHive::TEvConfigureScaleRecommender::TPtr& ev);
+ void Handle(TEvPrivate::TEvUpdateFollowers::TPtr& ev);
protected:
void RestartPipeTx(ui64 tabletId);
diff --git a/ydb/core/mind/hive/tx__load_everything.cpp b/ydb/core/mind/hive/tx__load_everything.cpp
index b0237d223f2..6b298ce09b3 100644
--- a/ydb/core/mind/hive/tx__load_everything.cpp
+++ b/ydb/core/mind/hive/tx__load_everything.cpp
@@ -674,7 +674,7 @@ public:
--cnt;
allowedDc = {dc};
Self->DataCenters[dc].Followers[{tabletId, groupId}].push_back(followerIt);
- db.Table<Schema::TabletFollowerTablet>().Key(tabletId, followerIt->Id).Update<Schema::TabletFollowerTablet::DataCenter>(dc);
+ Self->PendingFollowerUpdates.Update({tabletId, followerIt->Id}, dc);
ok = true;
}
}
@@ -692,7 +692,7 @@ public:
}
follower->NodeFilter.AllowedDataCenters = {dcIt->first};
Self->DataCenters[dcIt->first].Followers[{tabletId, groupId}].push_back(follower);
- db.Table<Schema::TabletFollowerTablet>().Key(follower->GetFullTabletId()).Update<Schema::TabletFollowerTablet::DataCenter>(dcIt->first);
+ Self->PendingFollowerUpdates.Update(follower->GetFullTabletId(), dcIt->first);
--dcIt->second;
}
}
@@ -859,6 +859,11 @@ public:
Self->MigrationState = NKikimrHive::EMigrationState::MIGRATION_READY;
ctx.Send(Self->SelfId(), new TEvPrivate::TEvBootTablets());
+ if (!Self->PendingFollowerUpdates.Empty()) {
+ ctx.Send(Self->SelfId(), new TEvPrivate::TEvUpdateFollowers);
+ Self->ProcessFollowerUpdatesScheduled = true;
+ }
+
for (auto it = Self->Nodes.begin(); it != Self->Nodes.end(); ++it) {
Self->ScheduleUnlockTabletExecution(it->second);
}
diff --git a/ydb/core/mind/hive/tx__update_dc_followers.cpp b/ydb/core/mind/hive/tx__update_dc_followers.cpp
index 9d6c26591bc..b196d6d3979 100644
--- a/ydb/core/mind/hive/tx__update_dc_followers.cpp
+++ b/ydb/core/mind/hive/tx__update_dc_followers.cpp
@@ -4,64 +4,87 @@
namespace NKikimr {
namespace NHive {
-class TTxUpdateDcFollowers : public TTransactionBase<THive> {
- TDataCenterId DataCenterId;
+class TTxProcessUpdateFollowers : public TTransactionBase<THive> {
TSideEffects SideEffects;
+
+ static constexpr size_t MAX_UPDATES_PROCESSED = 1000;
public:
- TTxUpdateDcFollowers(const TDataCenterId& dataCenter, THive* hive)
+ TTxProcessUpdateFollowers(THive* hive)
: TBase(hive)
- , DataCenterId(dataCenter)
{}
TTxType GetTxType() const override { return NHive::TXTYPE_UPDATE_DC_FOLLOWERS; }
bool Execute(TTransactionContext& txc, const TActorContext&) override {
- BLOG_D("THive::TTxUpdateDcFollowers::Execute(" << DataCenterId << ")");
- SideEffects.Reset(Self->SelfId());
+ BLOG_D("TTxProcessUpdateFollowers::Execute()");
NIceDb::TNiceDb db(txc.DB);
- auto& dataCenter = Self->DataCenters[DataCenterId];
- if (!dataCenter.UpdateScheduled) {
- return true;
- }
- dataCenter.UpdateScheduled = false;
- if (dataCenter.IsRegistered()) {
- for (auto& [tabletId, tablet] : Self->Tablets) {
- for (auto& group : tablet.FollowerGroups) {
- auto& followers = dataCenter.Followers[{tabletId, group.Id}];
- auto neededCount = group.GetFollowerCountForDataCenter(DataCenterId);
- while (followers.size() < neededCount) {
- TFollowerTabletInfo& follower = tablet.AddFollower(group);
- follower.NodeFilter.AllowedDataCenters = {DataCenterId};
- follower.Statistics.SetLastAliveTimestamp(TlsActivationContext->Now().MilliSeconds());
- db.Table<Schema::TabletFollowerTablet>().Key(tabletId, follower.Id).Update(
- NIceDb::TUpdate<Schema::TabletFollowerTablet::GroupID>(follower.FollowerGroup.Id),
- NIceDb::TUpdate<Schema::TabletFollowerTablet::FollowerNode>(0),
- NIceDb::TUpdate<Schema::TabletFollowerTablet::Statistics>(follower.Statistics),
- NIceDb::TUpdate<Schema::TabletFollowerTablet::DataCenter>(DataCenterId));
- follower.InitTabletMetrics();
- follower.BecomeStopped();
- follower.InitiateBoot();
- followers.push_back(std::prev(tablet.Followers.end()));
- BLOG_D("THive::TTxUpdateDcFollowers::Execute(" << DataCenterId << "): created follower " << follower.GetFullTabletId());
+ SideEffects.Reset(Self->SelfId());
+ for (size_t i = 0; !Self->PendingFollowerUpdates.Empty() && i < MAX_UPDATES_PROCESSED; ++i) {
+ auto op = Self->PendingFollowerUpdates.Pop();
+ TTabletInfo* tablet = Self->FindTablet(op.TabletId);
+ auto& dc = Self->DataCenters[op.DataCenter];
+ if (tablet == nullptr) {
+ continue;
+ }
+ switch (op.Action) {
+ case TFollowerUpdates::EAction::Create:
+ {
+ if (!dc.IsRegistered()) {
+ continue;
+ }
+ TFollowerGroup& group = tablet->AsLeader().GetFollowerGroup(op.GroupId);
+ auto& followers = dc.Followers[{op.TabletId.first, op.GroupId}];
+ if (group.GetFollowerCountForDataCenter(op.DataCenter) <= followers.size()) {
+ continue;
}
+ TFollowerTabletInfo& follower = tablet->AsLeader().AddFollower(group);
+ follower.NodeFilter.AllowedDataCenters = {op.DataCenter};
+ follower.Statistics.SetLastAliveTimestamp(TlsActivationContext->Now().MilliSeconds());
+ db.Table<Schema::TabletFollowerTablet>().Key(op.TabletId.first, follower.Id).Update(
+ NIceDb::TUpdate<Schema::TabletFollowerTablet::GroupID>(follower.FollowerGroup.Id),
+ NIceDb::TUpdate<Schema::TabletFollowerTablet::FollowerNode>(0),
+ NIceDb::TUpdate<Schema::TabletFollowerTablet::Statistics>(follower.Statistics),
+ NIceDb::TUpdate<Schema::TabletFollowerTablet::DataCenter>(op.DataCenter));
+ follower.InitTabletMetrics();
+ follower.BecomeStopped();
+ follower.InitiateBoot();
+ followers.push_back(std::prev(tablet->AsLeader().Followers.end()));
+ BLOG_D("THive::TTxProcessUpdateFollowers::Execute(): created follower " << follower.GetFullTabletId());
+ break;
}
- }
- } else {
- // deleting followers
- i64 deletedFollowers = 0;
- for (auto& [_, followers] : dataCenter.Followers) {
- for (auto follower : followers) {
- db.Table<Schema::TabletFollowerTablet>().Key(follower->GetFullTabletId()).Delete();
- db.Table<Schema::Metrics>().Key(follower->GetFullTabletId()).Delete();
- follower->InitiateStop(SideEffects);
- auto& leader = follower->GetLeader();
- leader.Followers.erase(follower);
- ++deletedFollowers;
+ case TFollowerUpdates::EAction::Update:
+ {
+ // This is updated in memory in LoadEverything
+ bool exists = db.Table<Schema::TabletFollowerTablet>().Key(op.TabletId).Select().IsValid();
+ Y_ABORT_UNLESS(exists, "%s", (TStringBuilder() << "trying to update tablet " << op.TabletId).data());
+ db.Table<Schema::TabletFollowerTablet>().Key(op.TabletId).Update<Schema::TabletFollowerTablet::DataCenter>(op.DataCenter);
+ break;
+ }
+ case TFollowerUpdates::EAction::Delete:
+ {
+ if (dc.IsRegistered()) {
+ continue;
+ }
+ db.Table<Schema::TabletFollowerTablet>().Key(op.TabletId).Delete();
+ db.Table<Schema::Metrics>().Key(op.TabletId).Delete();
+ tablet->InitiateStop(SideEffects);
+ auto& followers = dc.Followers[{op.TabletId.first, op.GroupId}]; // Note: there are at most 3 followers here, see TPartitionConfigMerger
+ auto iter = std::find_if(followers.begin(), followers.end(), [tabletId = op.TabletId](const auto& fw) {
+ return fw->GetFullTabletId() == tabletId;
+ });
+ Y_ABORT_UNLESS(iter != followers.end());
+ auto& leader = tablet->GetLeader();
+ leader.Followers.erase(*iter);
+ followers.erase(iter);
+ Self->UpdateCounterTabletsTotal(-1);
+ break;
}
}
- BLOG_D("THive::TTxUpdateDcFollowers::Execute(" << DataCenterId << "): deleted " << deletedFollowers << " followers");
- Self->UpdateCounterTabletsTotal(-deletedFollowers);
- dataCenter.Followers.clear();
+ }
+ if (Self->PendingFollowerUpdates.Empty()) {
+ Self->ProcessFollowerUpdatesScheduled = false;
+ } else {
+ SideEffects.Send(Self->SelfId(), new TEvPrivate::TEvUpdateFollowers);
}
return true;
}
@@ -71,8 +94,8 @@ public:
}
};
-ITransaction* THive::CreateUpdateDcFollowers(const TDataCenterId& dc) {
- return new TTxUpdateDcFollowers(dc, this);
+ITransaction* THive::CreateProcessUpdateFollowers() {
+ return new TTxProcessUpdateFollowers(this);
}
} // NHive
generated by cgit v1.3 (git 2.53.0) at 2026-07-18 23:31:18 +0000