Skip to content

Commit ab4c3b2

Browse files
pixccpixcc
authored
Merge 5a65d95 into af01dce 
2 parents af01dce + 5a65d95 commit ab4c3b2

10 files changed

+701
-9
lines changed

‎ydb/core/mind/node_broker.cpp‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -586,6 +586,36 @@ void TNodeBroker::PrepareEpochCache()
586586
TabletCounters->Simple()[COUNTER_EPOCH_DELTAS_SIZE_BYTES].Set(EpochDeltasCache.size());
587587
}
588588

589+
void TNodeBroker::PrepareUpdateNodesLog()
590+
{
591+
LOG_DEBUG_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
592+
"Preparing update nodes log for epoch #" << Committed.Epoch.ToString()
593+
<< " nodes=" << Committed.Nodes.size()
594+
<< " expired=" << Committed.ExpiredNodes.size()
595+
<< " removed=" << Committed.RemovedNodes.size());
596+
597+
UpdateNodesLog.clear();
598+
UpdateNodesLogVersions.clear();
599+
600+
TVector<TVersionedNodeID> nodeIdsSortedByVersion;
601+
for (auto &entry : Committed.Nodes) {
602+
nodeIdsSortedByVersion.emplace_back(entry.second.NodeId, entry.second.Version);
603+
}
604+
for (auto &entry : Committed.ExpiredNodes) {
605+
nodeIdsSortedByVersion.emplace_back(entry.second.NodeId, entry.second.Version);
606+
}
607+
for (auto &entry : Committed.RemovedNodes) {
608+
nodeIdsSortedByVersion.emplace_back(entry.second.NodeId, entry.second.Version);
609+
}
610+
std::sort(nodeIdsSortedByVersion.begin(), nodeIdsSortedByVersion.end(), TVersionedNodeID::TCmpByVersion());
611+
612+
for (const auto &id : nodeIdsSortedByVersion) {
613+
const auto& node = *Committed.FindNode(id.NodeId);
614+
AddNodeToUpdateNodesLog(node);
615+
}
616+
TabletCounters->Simple()[COUNTER_UPDATE_NODES_LOG_SIZE_BYTES].Set(UpdateNodesLog.size());
617+
}
618+
589619
void TNodeBroker::AddNodeToEpochCache(const TNodeInfo &node)
590620
{
591621
LOG_DEBUG_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
@@ -615,13 +645,83 @@ void TNodeBroker::AddDeltaToEpochDeltasCache(const TString &delta, ui64 version)
615645
TabletCounters->Simple()[COUNTER_EPOCH_DELTAS_SIZE_BYTES].Set(EpochDeltasCache.size());
616646
}
617647

648+
void TNodeBroker::AddNodeToUpdateNodesLog(const TNodeInfo &node)
649+
{
650+
LOG_DEBUG_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
651+
"Add node " << node.IdShortString() << " to update nodes log");
652+
653+
NKikimrNodeBroker::TUpdateNodes updateNodes;
654+
655+
switch (node.State) {
656+
case ENodeState::Active:
657+
FillNodeInfo(node, *updateNodes.AddUpdates()->MutableNode());
658+
break;
659+
case ENodeState::Expired:
660+
updateNodes.AddUpdates()->SetExpiredNode(node.NodeId);
661+
break;
662+
case ENodeState::Removed:
663+
updateNodes.AddUpdates()->SetRemovedNode(node.NodeId);
664+
break;
665+
}
666+
667+
TString delta;
668+
Y_PROTOBUF_SUPPRESS_NODISCARD updateNodes.SerializeToString(&delta);
669+
670+
Y_ENSURE(UpdateNodesLogVersions.empty() || UpdateNodesLogVersions.back().Version <= node.Version);
671+
if (!UpdateNodesLogVersions.empty() && UpdateNodesLogVersions.back().Version == node.Version) {
672+
UpdateNodesLog += delta;
673+
UpdateNodesLogVersions.back().CacheEndOffset = UpdateNodesLog.size();
674+
} else {
675+
UpdateNodesLog += delta;
676+
UpdateNodesLogVersions.emplace_back(node.Version, UpdateNodesLog.size());
677+
}
678+
TabletCounters->Simple()[COUNTER_UPDATE_NODES_LOG_SIZE_BYTES].Set(UpdateNodesLog.size());
679+
}
680+
618681
void TNodeBroker::SubscribeForConfigUpdates(const TActorContext &ctx)
619682
{
620683
ui32 nodeBrokerItem = (ui32)NKikimrConsole::TConfigItem::NodeBrokerConfigItem;
621684
ui32 featureFlagsItem = (ui32)NKikimrConsole::TConfigItem::FeatureFlagsItem;
622685
NConsole::SubscribeViaConfigDispatcher(ctx, {nodeBrokerItem, featureFlagsItem}, ctx.SelfID);
623686
}
624687

688+
void TNodeBroker::SendUpdateNodes(const TActorContext &ctx)
689+
{
690+
if (SentVersion >= Committed.Epoch.Version) {
691+
return;
692+
}
693+
694+
for (const auto& [_, subscriber] : ServerPipeToSubscriber) {
695+
SendUpdateNodes(subscriber, SentVersion, ctx);
696+
}
697+
SentVersion = Committed.Epoch.Version;
698+
}
699+
700+
void TNodeBroker::SendUpdateNodes(TActorId subscriber, ui64 version, const TActorContext &ctx)
701+
{
702+
if (version >= Committed.Epoch.Version) {
703+
return;
704+
}
705+
706+
NKikimrNodeBroker::TUpdateNodes record;
707+
record.SetFromVersion(version);
708+
Committed.Epoch.Serialize(*record.MutableEpoch());
709+
auto response = MakeHolder<TEvNodeBroker::TEvUpdateNodes>(record);
710+
711+
auto it = std::lower_bound(UpdateNodesLogVersions.begin(), UpdateNodesLogVersions.end(), version + 1);
712+
if (it != UpdateNodesLogVersions.begin()) {
713+
response->PreSerializedData = UpdateNodesLog.substr(std::prev(it)->CacheEndOffset);
714+
} else {
715+
response->PreSerializedData = UpdateNodesLog;
716+
}
717+
718+
TabletCounters->Percentile()[COUNTER_UPDATE_NODES_BYTES].IncrementFor(response->GetCachedByteSize());
719+
LOG_TRACE_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
720+
"Send TEvUpdateNodes v" << version << " -> v" << Committed.Epoch.Version
721+
<< " to " << subscriber);
722+
ctx.Send(subscriber, response.Release());
723+
}
724+
625725
void TNodeBroker::TState::LoadConfigFromProto(const NKikimrNodeBroker::TConfig &config)
626726
{
627727
Config = config;
@@ -1452,6 +1552,38 @@ void TNodeBroker::Handle(TEvNodeBroker::TEvSetConfigRequest::TPtr &ev,
14521552
Execute(CreateTxUpdateConfig(ev), ctx);
14531553
}
14541554

1555+
void TNodeBroker::Handle(TEvNodeBroker::TEvSubscribeNodesRequest::TPtr &ev,
1556+
const TActorContext &ctx)
1557+
{
1558+
TabletCounters->Cumulative()[COUNTER_SUBSCRIBE_NODES_REQUESTS].Increment(1);
1559+
1560+
if (auto [_, inserted] = ServerPipeToSubscriber.emplace(ev->Recipient, ev->Sender); inserted) {
1561+
LOG_DEBUG_S(ctx, NKikimrServices::NODE_BROKER,
1562+
"New subscriber " << ev->Sender << ", server pipe id: " << ev->Recipient
1563+
<< ", cached version: " << ev->Get()->Record.GetCachedVersion());
1564+
SendUpdateNodes(ev->Sender, ev->Get()->Record.GetCachedVersion(), ctx);
1565+
}
1566+
}
1567+
1568+
void TNodeBroker::Handle(TEvNodeBroker::TEvSyncNodesRequest::TPtr &ev,
1569+
const TActorContext &ctx)
1570+
{
1571+
TabletCounters->Cumulative()[COUNTER_SYNC_NODES_REQUESTS].Increment(1);
1572+
SendUpdateNodes(ev->Sender, ev->Get()->Record.GetCachedVersion(), ctx);
1573+
auto response = MakeHolder<TEvNodeBroker::TEvSyncNodesResponse>();
1574+
ctx.Send(ev->Sender, response.Release(), 0, ev->Cookie);
1575+
}
1576+
1577+
void TNodeBroker::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr &ev,
1578+
const TActorContext &ctx)
1579+
{
1580+
if (auto it = ServerPipeToSubscriber.find(ev->Get()->ServerId); it != ServerPipeToSubscriber.end()) {
1581+
LOG_DEBUG_S(ctx, NKikimrServices::NODE_BROKER,
1582+
"Unsubscribed " << it->second << ", server pipe id: " << it->first);
1583+
ServerPipeToSubscriber.erase(it);
1584+
}
1585+
}
1586+
14551587
void TNodeBroker::Handle(TEvPrivate::TEvUpdateEpoch::TPtr &ev,
14561588
const TActorContext &ctx)
14571589
{

‎ydb/core/mind/node_broker.h‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,12 @@ struct TEvNodeBroker {
109109
EvGracefulShutdownRequest,
110110
EvGracefulShutdownResponse,
111111

112+
// delta protocol
113+
EvSubscribeNodesRequest,
114+
EvUpdateNodes,
115+
EvSyncNodesRequest,
116+
EvSyncNodesResponse,
117+
112118
// TODO: remove
113119
// internal
114120
//EvNodeExpire = EvListNodes + 512,
@@ -204,6 +210,31 @@ struct TEvNodeBroker {
204210
NKikimrNodeBroker::TSetConfigResponse,
205211
EvSetConfigResponse> {
206212
};
213+
214+
struct TEvSubscribeNodesRequest : public TEventPB<TEvSubscribeNodesRequest,
215+
NKikimrNodeBroker::TSubscribeNodesRequest,
216+
EvSubscribeNodesRequest> {
217+
};
218+
219+
struct TEvUpdateNodes : public TEventPreSerializedPB<TEvUpdateNodes,
220+
NKikimrNodeBroker::TUpdateNodes,
221+
EvUpdateNodes> {
222+
TEvUpdateNodes() = default;
223+
TEvUpdateNodes(const NKikimrNodeBroker::TUpdateNodes &record)
224+
: TEventPreSerializedPB(record)
225+
{
226+
}
227+
};
228+
229+
struct TEvSyncNodesRequest : public TEventPB<TEvSyncNodesRequest,
230+
NKikimrNodeBroker::TSyncNodesRequest,
231+
EvSyncNodesRequest> {
232+
};
233+
234+
struct TEvSyncNodesResponse : public TEventPB<TEvSyncNodesResponse,
235+
NKikimrNodeBroker::TSyncNodesResponse,
236+
EvSyncNodesResponse> {
237+
};
207238
};
208239

209240
constexpr ui32 DOMAIN_BITS = TDomainsInfo::DomainBits;

‎ydb/core/mind/node_broker__extend_lease.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,9 @@ class TNodeBroker::TTxExtendLease : public TTransactionBase<TNodeBroker> {
8787
Self->Committed.ExtendLease(node);
8888
Self->Committed.UpdateEpochVersion();
8989
Self->AddNodeToEpochCache(node);
90+
Self->AddNodeToUpdateNodesLog(node);
9091
}
92+
Self->SendUpdateNodes(ctx);
9193
}
9294

9395
private:

‎ydb/core/mind/node_broker__migrate_state.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,11 +96,13 @@ class TNodeBroker::TTxMigrateState : public TTransactionBase<TNodeBroker> {
9696

9797
if (Finalized) {
9898
Self->Committed = Self->Dirty;
99+
Self->SentVersion = Self->Committed.Epoch.Version;
99100
Self->Become(&TNodeBroker::StateWork);
100101
Self->SubscribeForConfigUpdates(ctx);
101102
Self->ScheduleEpochUpdate(ctx);
102103
Self->PrepareEpochCache();
103-
Self->SignalTabletActive(ctx);
104+
Self->PrepareUpdateNodesLog();
105+
Self->SignalTabletActive(ctx, "1.0");
104106
} else {
105107
Self->Execute(Self->CreateTxMigrateState(std::move(DbChanges)));
106108
}

‎ydb/core/mind/node_broker__register_node.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -221,9 +221,11 @@ class TNodeBroker::TTxRegisterNode : public TTransactionBase<TNodeBroker> {
221221
if (ShouldUpdateVersion()) {
222222
Self->Committed.UpdateEpochVersion();
223223
Self->AddNodeToEpochCache(node);
224+
Self->AddNodeToUpdateNodesLog(node);
224225
}
225226

226227
Reply(ctx);
228+
Self->SendUpdateNodes(ctx);
227229
}
228230

229231
private:

‎ydb/core/mind/node_broker__update_epoch.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,9 @@ class TNodeBroker::TTxUpdateEpoch : public TTransactionBase<TNodeBroker> {
3333
Self->Committed.ApplyStateDiff(Diff);
3434
Self->ScheduleEpochUpdate(ctx);
3535
Self->PrepareEpochCache();
36+
Self->PrepareUpdateNodesLog();
3637
Self->ProcessDelayedListNodesRequests();
38+
Self->SendUpdateNodes(ctx);
3739
}
3840

3941
private:

‎ydb/core/mind/node_broker_impl.h‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -223,10 +223,12 @@ class TNodeBroker : public TActor<TNodeBroker>
223223
HFuncTraced(TEvNodeBroker::TEvCompactTables, Handle);
224224
HFuncTraced(TEvNodeBroker::TEvGetConfigRequest, Handle);
225225
HFuncTraced(TEvNodeBroker::TEvSetConfigRequest, Handle);
226+
HFuncTraced(TEvNodeBroker::TEvSubscribeNodesRequest, Handle);
227+
HFuncTraced(TEvNodeBroker::TEvSyncNodesRequest, Handle);
226228
HFuncTraced(TEvPrivate::TEvUpdateEpoch, Handle);
227229
HFuncTraced(TEvPrivate::TEvResolvedRegistrationRequest, Handle);
230+
HFunc(TEvTabletPipe::TEvServerDisconnected, Handle);
228231
IgnoreFunc(TEvTabletPipe::TEvServerConnected);
229-
IgnoreFunc(TEvTabletPipe::TEvServerDisconnected);
230232
IgnoreFunc(NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionResponse);
231233
IgnoreFunc(NConsole::TEvConfigsDispatcher::TEvRemoveConfigSubscriptionResponse);
232234

@@ -251,11 +253,16 @@ class TNodeBroker : public TActor<TNodeBroker>
251253
NKikimrNodeBroker::TNodeInfo &info) const;
252254

253255
void PrepareEpochCache();
256+
void PrepareUpdateNodesLog();
254257
void AddNodeToEpochCache(const TNodeInfo &node);
255258
void AddDeltaToEpochDeltasCache(const TString& delta, ui64 version);
259+
void AddNodeToUpdateNodesLog(const TNodeInfo &node);
256260

257261
void SubscribeForConfigUpdates(const TActorContext &ctx);
258262

263+
void SendUpdateNodes(const TActorContext &ctx);
264+
void SendUpdateNodes(TActorId subscriber, ui64 version, const TActorContext &ctx);
265+
259266
void Handle(TEvConsole::TEvConfigNotificationRequest::TPtr &ev,
260267
const TActorContext &ctx);
261268
void Handle(TEvConsole::TEvReplaceConfigSubscriptionsResponse::TPtr &ev,
@@ -276,6 +283,12 @@ class TNodeBroker : public TActor<TNodeBroker>
276283
const TActorContext &ctx);
277284
void Handle(TEvNodeBroker::TEvSetConfigRequest::TPtr &ev,
278285
const TActorContext &ctx);
286+
void Handle(TEvNodeBroker::TEvSubscribeNodesRequest::TPtr &ev,
287+
const TActorContext &ctx);
288+
void Handle(TEvNodeBroker::TEvSyncNodesRequest::TPtr &ev,
289+
const TActorContext &ctx);
290+
void Handle(TEvTabletPipe::TEvServerDisconnected::TPtr &ev,
291+
const TActorContext &ctx);
279292
void Handle(TEvPrivate::TEvUpdateEpoch::TPtr &ev,
280293
const TActorContext &ctx);
281294
void Handle(TEvPrivate::TEvResolvedRegistrationRequest::TPtr &ev,
@@ -288,11 +301,18 @@ class TNodeBroker : public TActor<TNodeBroker>
288301
// Events collected during initialization phase.
289302
TMultiMap<ui64, TEvNodeBroker::TEvListNodes::TPtr> DelayedListNodesRequests;
290303
TSchedulerCookieHolder EpochTimerCookieHolder;
291-
TString EpochCache;
292304

305+
// old epoch protocol
306+
TString EpochCache;
293307
TString EpochDeltasCache;
294308
TVector<TCacheVersion> EpochDeltasVersions;
295309

310+
// new delta protocol
311+
TString UpdateNodesLog;
312+
TVector<TCacheVersion> UpdateNodesLogVersions;
313+
ui64 SentVersion;
314+
THashMap<TActorId, TActorId> ServerPipeToSubscriber;
315+
296316
TTabletCountersBase* TabletCounters;
297317
TAutoPtr<TTabletCountersBase> TabletCountersPtr;
298318

0 commit comments

Comments
 (0)