Skip to content

Commit 49b9770

Browse files
niksavelievniksaveliev
authored
Merge 1fed88f into c2619f2 
2 parents c2619f2 + 1fed88f commit 49b9770

17 files changed

Lines changed: 598 additions & 26 deletions

‎ydb/core/persqueue/events/internal.h‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,8 @@ struct TEvPQ {
186186
EvReadingPartitionStatusRequest,
187187
EvProcessChangeOwnerRequests,
188188
EvWakeupReleasePartition,
189+
EvPartitionScaleStatusChanged,
190+
EvPartitionScaleRequestDone,
189191
EvEnd
190192
};
191193

@@ -1112,6 +1114,15 @@ struct TEvPQ {
11121114
ui32 PartitionId;
11131115
ui64 Cookie;
11141116
};
1117+
1118+
struct TEvPartitionScaleStatusChanged : public TEventPB<TEvPartitionScaleStatusChanged, NKikimrPQ::TEvPartitionScaleStatusChanged, EvPartitionScaleStatusChanged> {
1119+
TEvPartitionScaleStatusChanged() = default;
1120+
1121+
TEvPartitionScaleStatusChanged(ui32 partitionId, NKikimrPQ::EScaleStatus scaleStatus) {
1122+
Record.SetPartitionId(partitionId);
1123+
Record.SetScaleStatus(scaleStatus);
1124+
}
1125+
};
11151126
};
11161127

11171128
} //NKikimr

‎ydb/core/persqueue/partition.cpp‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -789,6 +789,9 @@ void TPartition::Handle(TEvPQ::TEvPartitionStatus::TPtr& ev, const TActorContext
789789
}
790790
}
791791
}
792+
793+
result.SetScaleStatus(SplitMergeEnabled(TabletConfig) ? ScaleStatus :NKikimrPQ::EScaleStatus::NORMAL);
794+
792795
ctx.Send(ev->Get()->Sender, new TEvPQ::TEvPartitionStatusResponse(result, Partition));
793796
}
794797

@@ -2073,6 +2076,10 @@ void TPartition::EndChangePartitionConfig(NKikimrPQ::TPQTabletConfig&& config,
20732076

20742077
Y_ABORT_UNLESS(Config.GetPartitionConfig().GetTotalPartitions() > 0);
20752078

2079+
if (Config.GetPartitionStrategy().GetScaleThresholdSeconds() != SplitMergeAvgWriteBytes->GetDuration().Seconds()) {
2080+
InitSplitMergeSlidingWindow();
2081+
}
2082+
20762083
Send(ReadQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
20772084
Send(WriteQuotaTrackerActor, new TEvPQ::TEvChangePartitionConfig(TopicConverter, Config));
20782085
TotalPartitionWriteSpeed = config.GetPartitionConfig().GetWriteSpeedInBytesPerSecond();

‎ydb/core/persqueue/partition.h‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
8989

9090
private:
9191
static const ui32 MAX_ERRORS_COUNT_TO_STORE = 10;
92+
static const ui32 SCALE_REQUEST_REPEAT_MIN_SECONDS = 60;
9293

9394
private:
9495
struct THasDataReq;
@@ -333,7 +334,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
333334
const TActorContext& ctx);
334335

335336
void Initialize(const TActorContext& ctx);
336-
337+
void InitSplitMergeSlidingWindow();
337338
template <typename T>
338339
void EmplacePendingRequest(T&& body, const TActorContext& ctx) {
339340
const auto now = ctx.Now();
@@ -358,6 +359,9 @@ class TPartition : public TActorBootstrapped<TPartition> {
358359

359360
void Handle(TEvPQ::TEvCheckPartitionStatusRequest::TPtr& ev, const TActorContext& ctx);
360361

362+
NKikimrPQ::EScaleStatus CheckScaleStatus(const TActorContext& ctx);
363+
void ChangeScaleStatusIfNeeded(NKikimrPQ::EScaleStatus scaleStatus);
364+
361365
TString LogPrefix() const;
362366

363367
void Handle(TEvPQ::TEvProcessChangeOwnerRequests::TPtr& ev, const TActorContext& ctx);
@@ -720,6 +724,10 @@ class TPartition : public TActorBootstrapped<TPartition> {
720724
NSlidingWindow::TSlidingWindow<NSlidingWindow::TSumOperation<ui64>> AvgReadBytes;
721725
TVector<NSlidingWindow::TSlidingWindow<NSlidingWindow::TSumOperation<ui64>>> AvgQuotaBytes;
722726

727+
std::unique_ptr<NSlidingWindow::TSlidingWindow<NSlidingWindow::TSumOperation<ui64>>> SplitMergeAvgWriteBytes;
728+
TInstant LastScaleRequestTime = TInstant::Zero();
729+
NKikimrPQ::EScaleStatus ScaleStatus = NKikimrPQ::EScaleStatus::NORMAL;
730+
723731
ui64 ReservedSize;
724732
std::deque<THolder<TEvPQ::TEvReserveBytes>> ReserveRequests;
725733

‎ydb/core/persqueue/partition_init.cpp‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -699,6 +699,8 @@ void TPartition::Initialize(const TActorContext& ctx) {
699699
LastUsedStorageMeterTimestamp = ctx.Now();
700700
WriteTimestampEstimate = ManageWriteTimestampEstimate ? ctx.Now() : TInstant::Zero();
701701

702+
InitSplitMergeSlidingWindow();
703+
702704
CloudId = Config.GetYcCloudId();
703705
DbId = Config.GetYdbDatabaseId();
704706
DbPath = Config.GetYdbDatabasePath();
@@ -956,6 +958,10 @@ void TPartition::SetupStreamCounters(const TActorContext& ctx) {
956958
);
957959
}
958960

961+
void TPartition::InitSplitMergeSlidingWindow() {
962+
using Tui64SumSlidingWindow = NSlidingWindow::TSlidingWindow<NSlidingWindow::TSumOperation<ui64>>;
963+
SplitMergeAvgWriteBytes = std::make_unique<Tui64SumSlidingWindow>(TDuration::Seconds(Config.GetPartitionStrategy().GetScaleThresholdSeconds()), 1000);
964+
}
959965

960966
//
961967
// Functions

‎ydb/core/persqueue/partition_scale_manager.cpp‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
#include "ydb/core/persqueue/partition_scale_manager.h"
2+
3+
namespace NKikimr {
4+
namespace NPQ {
5+
6+
7+
TPartitionScaleManager::TPartitionScaleManager(
8+
const TString& topicName,
9+
const TString& databasePath,
10+
NKikimrPQ::TUpdateBalancerConfig& balancerConfig
11+
)
12+
: TopicName(topicName)
13+
, DatabasePath(databasePath)
14+
, BalancerConfig(balancerConfig) {
15+
16+
}
17+
18+
void TPartitionScaleManager::HandleScaleStatusChange(const TPartitionInfo& partition, NKikimrPQ::EScaleStatus scaleStatus, const TActorContext& ctx) {
19+
if (scaleStatus == NKikimrPQ::EScaleStatus::NEED_SPLIT) {
20+
PartitionsToSplit.emplace(partition.Id, partition);
21+
TrySendScaleRequest(ctx);
22+
} else {
23+
PartitionsToSplit.erase(partition.Id);
24+
}
25+
}
26+
27+
void TPartitionScaleManager::TrySendScaleRequest(const TActorContext& ctx) {
28+
TInstant delayDeadline = LastResponseTime + RequestTimeout;
29+
if (!DatabasePath || RequestInflight || delayDeadline > ctx.Now()) {
30+
return;
31+
}
32+
33+
auto splitMergePair = BuildScaleRequest();
34+
if (splitMergePair.first.empty() && splitMergePair.second.empty()) {
35+
return;
36+
}
37+
38+
RequestInflight = true;
39+
CurrentScaleRequest = ctx.Register(new TPartitionScaleRequest(
40+
TopicName,
41+
DatabasePath,
42+
BalancerConfig.PathId,
43+
BalancerConfig.PathVersion,
44+
splitMergePair.first,
45+
splitMergePair.second,
46+
ctx.SelfID
47+
));
48+
}
49+
50+
51+
using TPartitionSplit = NKikimrSchemeOp::TPersQueueGroupDescription_TPartitionSplit;
52+
using TPartitionMerge = NKikimrSchemeOp::TPersQueueGroupDescription_TPartitionMerge;
53+
54+
std::pair<std::vector<TPartitionSplit>, std::vector<TPartitionMerge>> TPartitionScaleManager::BuildScaleRequest() {
55+
std::vector<TPartitionSplit> splitsToApply;
56+
std::vector<TPartitionMerge> mergesToApply;
57+
58+
size_t allowedSplitsCount = BalancerConfig.PartitionCountLimit > BalancerConfig.CurPartitions ? BalancerConfig.PartitionCountLimit - BalancerConfig.CurPartitions : 0;
59+
auto itSplit = PartitionsToSplit.begin();
60+
while (allowedSplitsCount > 0 && itSplit != PartitionsToSplit.end()) {
61+
const auto partitionId = itSplit->first;
62+
const auto& partition = itSplit->second;
63+
64+
if (BalancerConfig.PartitionGraph.GetPartition(partitionId)->Children.empty()) {
65+
auto mid = GetRangeMid(partition.KeyRange.FromBound ? *partition.KeyRange.FromBound : "", partition.KeyRange.ToBound ?*partition.KeyRange.ToBound : "");
66+
if (mid.empty()) {
67+
itSplit = PartitionsToSplit.erase(itSplit);
68+
continue;
69+
}
70+
71+
TPartitionSplit split;
72+
split.set_partition(partition.Id);
73+
split.set_splitboundary(mid);
74+
splitsToApply.push_back(split);
75+
76+
allowedSplitsCount--;
77+
itSplit++;
78+
} else {
79+
itSplit = PartitionsToSplit.erase(itSplit);
80+
}
81+
}
82+
83+
return {splitsToApply, mergesToApply};
84+
}
85+
86+
void TPartitionScaleManager::HandleScaleRequestResult(TPartitionScaleRequest::TEvPartitionScaleRequestDone::TPtr& ev, const TActorContext& ctx) {
87+
RequestInflight = false;
88+
LastResponseTime = ctx.Now();
89+
auto result = ev->Get();
90+
if (result->Status == TEvTxUserProxy::TResultStatus::ExecComplete) {
91+
TrySendScaleRequest(ctx);
92+
} else {
93+
ui64 newTimeout = RequestTimeout.MilliSeconds() == 0 ? MIN_SCALE_REQUEST_REPEAT_SECONDS_TIMEOUT + RandomNumber<ui64>(50) : RequestTimeout.MilliSeconds() * 2;
94+
RequestTimeout = TDuration::MilliSeconds(std::min(newTimeout, static_cast<ui64>(MAX_SCALE_REQUEST_REPEAT_SECONDS_TIMEOUT)));
95+
ctx.Schedule(RequestTimeout, new TEvents::TEvWakeup(TRY_SCALE_REQUEST_WAKE_UP_TAG));
96+
}
97+
}
98+
99+
void TPartitionScaleManager::Die(const TActorContext& ctx) {
100+
if (CurrentScaleRequest) {
101+
ctx.Send(CurrentScaleRequest, new TEvents::TEvPoisonPill());
102+
}
103+
}
104+
105+
void TPartitionScaleManager::UpdateBalancerConfig(NKikimrPQ::TUpdateBalancerConfig& config) {
106+
BalancerConfig = TBalancerConfig(config);
107+
}
108+
109+
void TPartitionScaleManager::UpdateDatabasePath(const TString& dbPath) {
110+
DatabasePath = dbPath;
111+
}
112+
113+
TString TPartitionScaleManager::GetRangeMid(const TString& from, const TString& to) {
114+
if (from > to) {
115+
return "";
116+
}
117+
118+
TStringBuilder result;
119+
120+
unsigned char fromPadding = 0;
121+
unsigned char toPadding = 255;
122+
123+
size_t maxSize = std::max(from.size(), to.size());
124+
for (size_t i = 0; i < maxSize; ++i) {
125+
ui16 fromChar = i < from.size() ? static_cast<ui16>(from[i]) : fromPadding;
126+
unsigned char toChar = i < to.size() ? static_cast<unsigned char>(to[i]) : toPadding;
127+
128+
ui16 sum = fromChar + toChar;
129+
130+
result += static_cast<unsigned char>(sum / 2);
131+
}
132+
133+
if (result == from) {
134+
result += static_cast<unsigned char>(127);
135+
}
136+
137+
return result;
138+
}
139+
140+
} // namespace NPQ
141+
} // namespace NKikimr

‎ydb/core/persqueue/partition_scale_manager.h‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
#pragma once
2+
3+
#include <ydb/core/base/path.h>
4+
#include "ydb/core/persqueue/utils.h"
5+
#include <ydb/core/persqueue/partition_scale_request.h>
6+
#include <ydb/core/protos/pqconfig.pb.h>
7+
#include <ydb/core/tx/tx_proxy/proxy.h>
8+
#include <ydb/core/tx/schemeshard/schemeshard.h>
9+
#include <ydb/core/tx/schemeshard/schemeshard_info_types.h>
10+
#include <ydb/library/actors/core/actor_bootstrapped.h>
11+
#include <ydb/library/actors/core/actorsystem.h>
12+
13+
#include <util/system/types.h>
14+
#include <util/generic/fwd.h>
15+
#include <util/generic/string.h>
16+
17+
#include <unordered_map>
18+
#include <utility>
19+
20+
namespace NKikimr {
21+
namespace NPQ {
22+
23+
class TPartitionScaleManager {
24+
25+
public:
26+
struct TPartitionInfo {
27+
ui32 Id;
28+
NSchemeShard::TTopicTabletInfo::TKeyRange KeyRange;
29+
};
30+
31+
private:
32+
struct TBalancerConfig {
33+
TBalancerConfig(
34+
NKikimrPQ::TUpdateBalancerConfig& config
35+
)
36+
: PathId(config.GetPathId())
37+
, PathVersion(config.GetVersion())
38+
, PartitionGraph(MakePartitionGraph(config))
39+
, PartitionCountLimit(config.GetTabletConfig().GetPartitionStrategy().GetMaxPartitionCount())
40+
, MinActivePartitions(config.GetTabletConfig().GetPartitionStrategy().GetMinPartitionCount())
41+
, CurPartitions(config.PartitionsSize()) {
42+
}
43+
44+
ui64 PathId;
45+
int PathVersion;
46+
TPartitionGraph PartitionGraph;
47+
ui64 PartitionCountLimit;
48+
ui64 MinActivePartitions;
49+
ui64 CurPartitions;
50+
};
51+
52+
public:
53+
TPartitionScaleManager(const TString& topicPath, const TString& databasePath, NKikimrPQ::TUpdateBalancerConfig& balancerConfig);
54+
55+
public:
56+
void HandleScaleStatusChange(const TPartitionInfo& partition, NKikimrPQ::EScaleStatus scaleStatus, const TActorContext& ctx);
57+
void HandleScaleRequestResult(TPartitionScaleRequest::TEvPartitionScaleRequestDone::TPtr& ev, const TActorContext& ctx);
58+
void TrySendScaleRequest(const TActorContext& ctx);
59+
void UpdateBalancerConfig(NKikimrPQ::TUpdateBalancerConfig& config);
60+
void UpdateDatabasePath(const TString& dbPath);
61+
void Die(const TActorContext& ctx);
62+
63+
static TString GetRangeMid(const TString& from, const TString& to);
64+
65+
private:
66+
using TPartitionSplit = NKikimrSchemeOp::TPersQueueGroupDescription_TPartitionSplit;
67+
using TPartitionMerge = NKikimrSchemeOp::TPersQueueGroupDescription_TPartitionMerge;
68+
69+
std::pair<std::vector<TPartitionSplit>, std::vector<TPartitionMerge>> BuildScaleRequest();
70+
71+
public:
72+
static const ui64 TRY_SCALE_REQUEST_WAKE_UP_TAG = 10;
73+
74+
private:
75+
static const ui32 MIN_SCALE_REQUEST_REPEAT_SECONDS_TIMEOUT = 10;
76+
static const ui32 MAX_SCALE_REQUEST_REPEAT_SECONDS_TIMEOUT = 1000;
77+
78+
const TString TopicName;
79+
TString DatabasePath;
80+
TActorId CurrentScaleRequest;
81+
TDuration RequestTimeout = TDuration::MilliSeconds(0);
82+
TInstant LastResponseTime = TInstant::Zero();
83+
84+
std::unordered_map<ui64, TPartitionInfo> PartitionsToSplit;
85+
86+
TBalancerConfig BalancerConfig;
87+
bool RequestInflight = false;
88+
};
89+
90+
} // namespace NPQ
91+
} // namespace NKikimr

0 commit comments

Comments
 (0)