Skip to content

Commit dddaa2a

Browse files
nshestakovnshestakov
authored
fixed heap-use-after-free - core/persqueue/pqtablet/partition/read_qu… (#25817)
2 parents 5532329 + 3549d16 commit dddaa2a

File tree

11 files changed

+32
-31
lines changed

11 files changed

+32
-31
lines changed

‎ydb/core/persqueue/pqtablet/partition/partition.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -312,7 +312,7 @@ void AddCheckDiskRequest(TEvKeyValue::TEvRequest *request, ui32 numChannels) {
312312

313313
TPartition::TPartition(ui64 tabletId, const TPartitionId& partition, const TActorId& tablet, ui32 tabletGeneration, const TActorId& blobCache,
314314
const NPersQueue::TTopicConverterPtr& topicConverter, TString dcId, bool isServerless,
315-
const NKikimrPQ::TPQTabletConfig& tabletConfig, const TTabletCountersBase& counters, bool subDomainOutOfSpace, ui32 numChannels,
315+
const NKikimrPQ::TPQTabletConfig& tabletConfig, const std::shared_ptr<TTabletCountersBase>& counters, bool subDomainOutOfSpace, ui32 numChannels,
316316
const TActorId& writeQuoterActorId,
317317
TIntrusivePtr<NJaegerTracing::TSamplingThrottlingControl> samplingControl,
318318
bool newPartition)
@@ -355,7 +355,7 @@ TPartition::TPartition(ui64 tabletId, const TPartitionId& partition, const TActo
355355
, LastEmittedHeartbeat(TRowVersion::Min())
356356
, SamplingControl(samplingControl)
357357
{
358-
TabletCounters.Populate(Counters);
358+
TabletCounters.Populate(*Counters);
359359
}
360360

361361
void TPartition::EmplaceResponse(TMessage&& message, const TActorContext& ctx) {
@@ -4119,7 +4119,7 @@ size_t TPartition::GetQuotaRequestSize(const TEvKeyValue::TEvRequest& request) {
41194119

41204120
void TPartition::CreateMirrorerActor() {
41214121
Mirrorer = MakeHolder<TMirrorerInfo>(
4122-
Register(CreateMirrorer(TabletId, TabletActorId, SelfId(), TopicConverter, Partition.InternalPartitionId, IsLocalDC, GetEndOffset(), Config.GetPartitionConfig().GetMirrorFrom(), TabletCounters)),
4122+
RegisterWithSameMailbox(CreateMirrorer(TabletId, TabletActorId, SelfId(), TopicConverter, Partition.InternalPartitionId, IsLocalDC, GetEndOffset(), Config.GetPartitionConfig().GetMirrorFrom(), TabletCounters)),
41234123
TabletCounters
41244124
);
41254125
}
@@ -4343,7 +4343,7 @@ void TPartition::Handle(TEvPQ::TEvExclusiveLockAcquired::TPtr&) {
43434343

43444344
IActor* CreatePartitionActor(ui64 tabletId, const TPartitionId& partition, const TActorId& tablet, ui32 tabletGeneration,
43454345
const TActorId& blobCache, const NPersQueue::TTopicConverterPtr& topicConverter, TString dcId, bool isServerless,
4346-
const NKikimrPQ::TPQTabletConfig& config, const TTabletCountersBase& counters, bool SubDomainOutOfSpace,
4346+
const NKikimrPQ::TPQTabletConfig& config, const std::shared_ptr<TTabletCountersBase>& counters, bool SubDomainOutOfSpace,
43474347
ui32 numChannels, const TActorId& writeQuoterActorId,
43484348
TIntrusivePtr<NJaegerTracing::TSamplingThrottlingControl> samplingControl, bool newPartition) {
43494349

‎ydb/core/persqueue/pqtablet/partition/partition.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -511,7 +511,7 @@ class TPartition : public TBaseActor<TPartition> {
511511

512512
TPartition(ui64 tabletId, const TPartitionId& partition, const TActorId& tablet, ui32 tabletGeneration, const TActorId& blobCache,
513513
const NPersQueue::TTopicConverterPtr& topicConverter, TString dcId, bool isServerless,
514-
const NKikimrPQ::TPQTabletConfig& config, const TTabletCountersBase& counters, bool SubDomainOutOfSpace, ui32 numChannels,
514+
const NKikimrPQ::TPQTabletConfig& config, const std::shared_ptr<TTabletCountersBase>& counters, bool SubDomainOutOfSpace, ui32 numChannels,
515515
const TActorId& writeQuoterActorId,
516516
TIntrusivePtr<NJaegerTracing::TSamplingThrottlingControl> samplingControl,
517517
bool newPartition = false);
@@ -706,7 +706,7 @@ class TPartition : public TBaseActor<TPartition> {
706706
const NKikimrPQ::TPQTabletConfig::TPartition* PartitionConfig = nullptr;
707707
const NKikimrPQ::TPQTabletConfig::TPartition* PendingPartitionConfig = nullptr;
708708

709-
const TTabletCountersBase& Counters;
709+
std::shared_ptr<TTabletCountersBase> Counters;
710710
NPersQueue::TTopicConverterPtr TopicConverter;
711711
bool IsLocalDC;
712712
TString DCId;

‎ydb/core/persqueue/pqtablet/partition/partition_factory.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ NActors::IActor* CreatePartitionActor(ui64 tabletId,
3434
TString dcId,
3535
bool isServerless,
3636
const NKikimrPQ::TPQTabletConfig& config,
37-
const TTabletCountersBase& counters,
37+
const std::shared_ptr<TTabletCountersBase>& counters,
3838
bool SubDomainOutOfSpace,
3939
ui32 numChannels,
4040
const NActors::TActorId& writeQuoterActorId,

‎ydb/core/persqueue/pqtablet/partition/partition_init.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -978,7 +978,7 @@ void TPartition::Initialize(const TActorContext& ctx) {
978978
CreationTime = ctx.Now();
979979
WriteCycleStartTime = ctx.Now();
980980

981-
ReadQuotaTrackerActor = Register(new TReadQuoter(
981+
ReadQuotaTrackerActor = RegisterWithSameMailbox(new TReadQuoter(
982982
AppData(ctx)->PQConfig,
983983
TopicConverter,
984984
Config,

‎ydb/core/persqueue/pqtablet/partition/read_quoter.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -342,7 +342,7 @@ THolder<TAccountQuoterHolder> TReadQuoter::CreateAccountQuotaTracker(const TStri
342342
if (GetTabletActor() && quotingConfig.GetEnableQuoting()) {
343343
Y_ENSURE(TopicConverter);
344344
if (quotingConfig.GetEnableReadQuoting()) {
345-
actorId = TActivationContext::Register(
345+
actorId = TActivationContext::RegisterWithSameMailbox(
346346
new TAccountReadQuoter(
347347
GetTabletActor(),
348348
ctx.SelfID,

‎ydb/core/persqueue/pqtablet/partition/read_quoter.h‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,7 @@ const TDuration WAKE_UP_TIMEOUT = TDuration::Seconds(1);
114114
TActorId tabletActor,
115115
bool totalPartitionQuotaEnabled,
116116
ui64 tabletId,
117-
const TTabletCountersBase& counters,
117+
const std::shared_ptr<TTabletCountersBase>& counters,
118118
ui64 maxRequestsInflight
119119
)
120120
: TBaseActor(tabletId, tabletActor, NKikimrServices::PERSQUEUE)
@@ -126,7 +126,7 @@ const TDuration WAKE_UP_TIMEOUT = TDuration::Seconds(1);
126126
, MaxInflightRequests(maxRequestsInflight)
127127
, TotalPartitionQuotaEnabled(totalPartitionQuotaEnabled)
128128
{
129-
Counters.Populate(counters);
129+
Counters.Populate(*counters);
130130
}
131131

132132
public:
@@ -239,7 +239,7 @@ const static ui64 DEFAULT_READ_SPEED_AND_BURST = 1'000'000'000;
239239
TActorId tabletActor,
240240
const TActorId& parent,
241241
ui64 tabletId,
242-
const TTabletCountersBase& counters
242+
const std::shared_ptr<TTabletCountersBase>& counters
243243
)
244244
: TPartitionQuoterBase(
245245
topicConverter, config, partition, tabletActor, true, tabletId, counters,
@@ -309,7 +309,7 @@ class TWriteQuoter : public TPartitionQuoterBase {
309309
const TPartitionId& partition,
310310
TActorId tabletActor,
311311
ui64 tabletId,
312-
const TTabletCountersBase& counters
312+
const std::shared_ptr<TTabletCountersBase>& counters
313313
);
314314

315315
public:

‎ydb/core/persqueue/pqtablet/partition/write_quoter.cpp‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ TWriteQuoter::TWriteQuoter(
1010
const TPartitionId& partition,
1111
TActorId tabletActor,
1212
ui64 tabletId,
13-
const TTabletCountersBase& counters
13+
const std::shared_ptr<TTabletCountersBase>& counters
1414
)
1515
: TPartitionQuoterBase(
1616
topicConverter, config, partition, tabletActor, pqConfig.GetQuotingConfig().GetEnableQuoting(),
@@ -60,7 +60,7 @@ void TWriteQuoter::UpdateQuotaConfigImpl(bool, const TActorContext&) {
6060
THolder<TAccountQuoterHolder> TWriteQuoter::CreateAccountQuotaTracker() const {
6161
TActorId actorId;
6262
if (GetTabletActor() && GetAccountQuotingEnabled(AppData()->PQConfig)) {
63-
actorId = TActivationContext::Register(
63+
actorId = TActivationContext::RegisterWithSameMailbox(
6464
new TAccountWriteQuoter(
6565
GetTabletActor(),
6666
SelfId(),
@@ -69,7 +69,8 @@ THolder<TAccountQuoterHolder> TWriteQuoter::CreateAccountQuotaTracker() const {
6969
GetPartition(),
7070
Counters,
7171
ActorContext()
72-
)
72+
),
73+
TabletActorId
7374
);
7475
}
7576
if (actorId) {

‎ydb/core/persqueue/pqtablet/pq_impl.cpp‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -625,7 +625,7 @@ void TPersQueue::CreateOriginalPartition(const NKikimrPQ::TPQTabletConfig& confi
625625
bool newPartition,
626626
const TActorContext& ctx)
627627
{
628-
TActorId actorId = ctx.Register(CreatePartitionActor(partitionId,
628+
TActorId actorId = ctx.RegisterWithSameMailbox(CreatePartitionActor(partitionId,
629629
topicConverter,
630630
config,
631631
newPartition,
@@ -689,7 +689,7 @@ void TPersQueue::CreateSupportivePartitionActor(const TPartitionId& partitionId,
689689
PQ_ENSURE(Partitions.contains(partitionId));
690690

691691
TPartitionInfo& partition = Partitions.at(partitionId);
692-
partition.Actor = ctx.Register(CreatePartitionActor(partitionId,
692+
partition.Actor = ctx.RegisterWithSameMailbox(CreatePartitionActor(partitionId,
693693
TopicConverter,
694694
MakeSupportivePartitionConfig(),
695695
false,
@@ -2595,7 +2595,7 @@ void TPersQueue::Handle(TEvPersQueue::TEvRequest::TPtr& ev, const TActorContext&
25952595
directKey.PartitionSessionId = pipeIter->second.PartitionSessionId;
25962596
}
25972597
TStringBuilder log; log << "PQ - create read proxy" << Endl;
2598-
TActorId rr = ctx.Register(CreateReadProxy(ev->Sender, TabletID(), ctx.SelfID, GetGeneration(), directKey, request));
2598+
TActorId rr = ctx.RegisterWithSameMailbox(CreateReadProxy(ev->Sender, TabletID(), ctx.SelfID, GetGeneration(), directKey, request));
25992599
ans = CreateResponseProxy(rr, ctx.SelfID, TopicName, p, m, s, c, ResourceMetrics, ctx);
26002600
} else {
26012601
ans = CreateResponseProxy(ev->Sender, ctx.SelfID, TopicName, p, m, s, c, ResourceMetrics, ctx);
@@ -2871,7 +2871,6 @@ TPersQueue::TPersQueue(const TActorId& tablet, TTabletStorageInfo *info)
28712871
, OriginalPartitionsCount(0)
28722872
, NewConfigShouldBeApplied(false)
28732873
, TabletState(NKikimrPQ::ENormal)
2874-
, Counters(nullptr)
28752874
, NextResponseCookie(0)
28762875
, ResourceMetrics(nullptr)
28772876
{
@@ -2890,8 +2889,9 @@ TPersQueue::TPersQueue(const TActorId& tablet, TTabletStorageInfo *info)
28902889
ECumulativeCounters_descriptor,
28912890
EPercentileCounters_descriptor> TPersQueueCounters;
28922891
typedef TProtobufTabletCountersPair<TKeyValueCounters, TPersQueueCounters> TCounters;
2892+
28932893
TAutoPtr<TCounters> counters(new TCounters());
2894-
Counters = (counters->GetSecondTabletCounters()).Release();
2894+
Counters.reset(counters->GetSecondTabletCounters().Release());
28952895

28962896
State.SetupTabletCounters(counters->GetFirstTabletCounters().Release()); //FirstTabletCounters is of good type and contains all counters
28972897
State.Clear();
@@ -2900,7 +2900,7 @@ TPersQueue::TPersQueue(const TActorId& tablet, TTabletStorageInfo *info)
29002900
void TPersQueue::CreatedHook(const TActorContext& ctx)
29012901
{
29022902
IsServerless = AppData(ctx)->FeatureFlags.GetEnableDbCounters(); //TODO: find out it via describe
2903-
CacheActor = ctx.Register(new TPQCacheProxy(ctx.SelfID, TabletID()));
2903+
CacheActor = ctx.RegisterWithSameMailbox(new TPQCacheProxy(ctx.SelfID, TabletID()));
29042904

29052905
SamplingControl = AppData(ctx)->TracingConfigurator->GetControl();
29062906

@@ -4751,14 +4751,14 @@ TActorId TPersQueue::GetPartitionQuoter(const TPartitionId& partition) {
47514751

47524752
auto& quoterId = PartitionWriteQuoters[partition.OriginalPartitionId];
47534753
if (!quoterId) {
4754-
quoterId = Register(new TWriteQuoter(
4754+
quoterId = RegisterWithSameMailbox(new TWriteQuoter(
47554755
TopicConverter,
47564756
Config,
47574757
AppData()->PQConfig,
47584758
partition,
47594759
SelfId(),
47604760
TabletID(),
4761-
*Counters
4761+
Counters
47624762
));
47634763
}
47644764
return quoterId;
@@ -4782,7 +4782,7 @@ IActor* TPersQueue::CreatePartitionActor(const TPartitionId& partitionId,
47824782
DCId,
47834783
IsServerless,
47844784
config,
4785-
*Counters,
4785+
Counters,
47864786
SubDomainOutOfSpace,
47874787
(ui32)channels,
47884788
GetPartitionQuoter(partitionId),

‎ydb/core/persqueue/pqtablet/pq_impl.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -238,7 +238,7 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
238238
NKikimrPQ::ETabletState TabletState;
239239
TSet<TChangeNotification> TabletStateRequests;
240240

241-
TAutoPtr<TTabletCountersBase> Counters;
241+
std::shared_ptr<TTabletCountersBase> Counters;
242242
TEvPQ::TEvTabletCacheCounters::TCacheCounters CacheCounters;
243243
TMap<TString, NKikimr::NPQ::TMultiCounter> BytesWrittenFromDC;
244244

‎ydb/core/persqueue/ut/partition_ut.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -331,7 +331,7 @@ class TPartitionFixture : public NUnitTest::TBaseFixture {
331331
NPersQueue::TTopicConverterPtr TopicConverter;
332332
NKikimrPQ::TPQTabletConfig Config;
333333

334-
TAutoPtr<TTabletCountersBase> TabletCounters;
334+
std::shared_ptr<TTabletCountersBase> TabletCounters;
335335

336336
};
337337

@@ -373,7 +373,7 @@ TPartition* TPartitionFixture::CreatePartitionActor(const TPartitionId& id,
373373
>;
374374

375375
TAutoPtr<TCounters> counters(new TCounters());
376-
TabletCounters = counters->GetSecondTabletCounters().Release();
376+
TabletCounters.reset(counters->GetSecondTabletCounters().Release());
377377

378378
Config = MakeConfig(config.Version,
379379
config.Consumers,
@@ -392,7 +392,7 @@ TPartition* TPartitionFixture::CreatePartitionActor(const TPartitionId& id,
392392
id,
393393
Ctx->Edge,
394394
Ctx->TabletId,
395-
*TabletCounters
395+
TabletCounters
396396
));
397397
}
398398
auto samplingControl = Ctx->Runtime->GetAppData(0).TracingConfigurator->GetControl();
@@ -405,7 +405,7 @@ TPartition* TPartitionFixture::CreatePartitionActor(const TPartitionId& id,
405405
"dcId",
406406
false,
407407
Config,
408-
*TabletCounters,
408+
TabletCounters,
409409
false,
410410
1,
411411
quoterId,

0 commit comments

Comments
 (0)