YQ-4253 added CPU limit per process in CS (#17804)

Author: GrigoriyPA

ydb/core/kqp/compute_actor/kqp_compute_actor.cpp

@@ -126,6 +126,20 @@ void TShardsScanningPolicy::FillRequestScanFeatures(const NKikimrTxDataShard::TK
  maxInFlight = ProtoConfig.GetScanLimit();
  }
 }
+
+TConclusionStatus TCPULimits::DeserializeFromProto(const NKikimrKqp::TEvStartKqpTasksRequest& config) {
+ const auto share = config.GetPoolMaxCpuShare();
+ if (share <= 0 || 1 < share) {
+ return TConclusionStatus::Fail("cpu share have to be in (0, 1] interval");
+ }
+ NActors::TExecutorPoolStats poolStats;
+ TVector<NActors::TExecutorThreadStats> threadsStats;
+ TActivationContext::ActorSystem()->GetPoolStats(TActivationContext::AsActorContext().SelfID.PoolID(), poolStats, threadsStats);
+ CPUGroupThreadsLimit = Max<ui64>(poolStats.MaxThreadCount, 1) * share;
+ CPUGroupName = config.GetSchedulerGroup();
+ return TConclusionStatus::Success();
+}
+
 }
 } // namespace NKikimr
 
@@ -146,8 +160,10 @@ IActor* CreateKqpScanComputeActor(const TActorId& executerId, ui64 txId,
 
 IActor* CreateKqpScanFetcher(const NKikimrKqp::TKqpSnapshot& snapshot, std::vector<NActors::TActorId>&& computeActors,
  const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta, const NYql::NDq::TComputeRuntimeSettings& settings,
- const ui64 txId, TMaybe<ui64> lockTxId, ui32 lockNodeId, TMaybe<NKikimrDataEvents::ELockMode> lockMode, const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId) {
- return new NScanPrivate::TKqpScanFetcherActor(snapshot, settings, std::move(computeActors), txId, lockTxId, lockNodeId, lockMode, meta, shardsScanningPolicy, counters, std::move(traceId));
+ const ui64 txId, TMaybe<ui64> lockTxId, ui32 lockNodeId, TMaybe<NKikimrDataEvents::ELockMode> lockMode,
+ const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId,
+ const TCPULimits& cpuLimits) {
+ return new NScanPrivate::TKqpScanFetcherActor(snapshot, settings, std::move(computeActors), txId, lockTxId, lockNodeId, lockMode, meta, shardsScanningPolicy, counters, std::move(traceId), cpuLimits);
 }
 
 }

ydb/core/kqp/compute_actor/kqp_compute_actor.h

@@ -44,6 +44,13 @@ class TShardsScanningPolicy {
 
 };
 
+class TCPULimits {
+ YDB_OPT(double, CPUGroupThreadsLimit);
+ YDB_READONLY_DEF(TString, CPUGroupName);
+public:
+ TConclusionStatus DeserializeFromProto(const NKikimrKqp::TEvStartKqpTasksRequest& config);
+};
+
 IActor* CreateKqpComputeActor(const TActorId& executerId, ui64 txId, NYql::NDqProto::TDqTask* task,
  NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory,
  const NYql::NDq::TComputeRuntimeSettings& settings, const NYql::NDq::TComputeMemoryLimits& memoryLimits,
@@ -62,7 +69,8 @@ IActor* CreateKqpScanComputeActor(const TActorId& executerId, ui64 txId, NYql::N
 IActor* CreateKqpScanFetcher(const NKikimrKqp::TKqpSnapshot& snapshot, std::vector<NActors::TActorId>&& computeActors,
  const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta, const NYql::NDq::TComputeRuntimeSettings& settings,
  const ui64 txId, TMaybe<ui64> lockTxId, ui32 lockNodeId, TMaybe<NKikimrDataEvents::ELockMode> lockMode,
- const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId);
+ const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId,
+ const TCPULimits& cpuLimits);
 
 NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
  TIntrusivePtr<TKqpCounters> counters,

ydb/core/kqp/compute_actor/kqp_scan_fetcher_actor.cpp

@@ -26,14 +26,16 @@ static constexpr ui64 MAX_SHARD_RESOLVES = 3;
 TKqpScanFetcherActor::TKqpScanFetcherActor(const NKikimrKqp::TKqpSnapshot& snapshot, const TComputeRuntimeSettings& settings,
  std::vector<NActors::TActorId>&& computeActors, const ui64 txId, const TMaybe<ui64> lockTxId, const ui32 lockNodeId,
  const TMaybe<NKikimrDataEvents::ELockMode> lockMode, const NKikimrTxDataShard::TKqpTransaction_TScanTaskMeta& meta,
- const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId)
+ const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId,
+ const TCPULimits& cpuLimits)
  : Meta(meta)
  , ScanDataMeta(Meta)
  , RuntimeSettings(settings)
  , TxId(txId)
  , LockTxId(lockTxId)
  , LockNodeId(lockNodeId)
  , LockMode(lockMode)
+ , CPULimits(cpuLimits)
  , ComputeActorIds(std::move(computeActors))
  , Snapshot(snapshot)
  , ShardsScanningPolicy(shardsScanningPolicy)
@@ -498,6 +500,11 @@ std::unique_ptr<NKikimr::TEvDataShard::TEvKqpScan> TKqpScanFetcherActor::BuildEv
  ev->Record.SetOlapProgramType(NKikimrSchemeOp::EOlapProgramType::OLAP_PROGRAM_SSA_PROGRAM_WITH_PARAMETERS);
  }
 
+ if (const auto cpuGroupThreadsLimit = CPULimits.GetCPUGroupThreadsLimitOptional()) {
+ ev->Record.SetCpuGroupThreadsLimit(*cpuGroupThreadsLimit);
+ ev->Record.SetCpuGroupName(CPULimits.GetCPUGroupName());
+ }
+
  ev->Record.SetDataFormat(Meta.GetDataFormat());
  return ev;
 }

ydb/core/kqp/compute_actor/kqp_scan_fetcher_actor.h

@@ -53,6 +53,7 @@ class TKqpScanFetcherActor: public NActors::TActorBootstrapped<TKqpScanFetcherAc
  const TMaybe<ui64> LockTxId;
  const ui32 LockNodeId;
  const TMaybe<NKikimrDataEvents::ELockMode> LockMode;
+ const TCPULimits CPULimits;
 
 public:
  static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -63,7 +64,8 @@ class TKqpScanFetcherActor: public NActors::TActorBootstrapped<TKqpScanFetcherAc
  std::vector<NActors::TActorId>&& computeActors,
  const ui64 txId, const TMaybe<ui64> lockTxId, const ui32 lockNodeId, const TMaybe<NKikimrDataEvents::ELockMode> lockMode,
  const NKikimrTxDataShard::TKqpTransaction_TScanTaskMeta& meta,
- const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId);
+ const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId,
+ const TCPULimits& cpuLimits);
 
  static TVector<TSerializedTableRange> BuildSerializedTableRanges(const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta::TReadOpMeta& readData);
 

ydb/core/kqp/node_service/kqp_node_service.cpp

@@ -314,11 +314,17 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
  ActorIdToProto(taskCtx.ComputeActorId, startedTask->MutableActorId());
  }
 
+ TCPULimits cpuLimits;
+ if (msg.GetPoolMaxCpuShare() > 0) {
+ // Share <= 0 means disabled limit
+ cpuLimits.DeserializeFromProto(msg).Validate();
+ }
+
  for (auto&& i : computesByStage) {
  for (auto&& m : i.second.MutableMetaInfo()) {
  Register(CreateKqpScanFetcher(msg.GetSnapshot(), std::move(m.MutableActorIds()),
  m.GetMeta(), runtimeSettingsBase, txId, lockTxId, lockNodeId, lockMode,
- scanPolicy, Counters, NWilson::TTraceId(ev->TraceId)));
+ scanPolicy, Counters, NWilson::TTraceId(ev->TraceId), cpuLimits));
  }
  }
 

ydb/core/protos/tx_datashard.proto

@@ -1762,6 +1762,8 @@ message TEvKqpScan {
  optional string CSScanPolicy = 26;
  optional NKikimrKqp.TEvKqpScanCursor ScanCursor = 27;
  optional NKikimrDataEvents.ELockMode LockMode = 28;
+ optional double CpuGroupThreadsLimit = 29;
+ optional string CpuGroupName = 30;
 }
 
 message TEvCompactTable {

ydb/core/tx/columnshard/engines/reader/abstract/read_context.cpp

@@ -13,7 +13,7 @@ TReadContext::TReadContext(const std::shared_ptr<IStoragesManager>& storagesMana
  const std::shared_ptr<NDataAccessorControl::IDataAccessorsManager>& dataAccessorsManager,
  const NColumnShard::TConcreteScanCounters& counters, const TReadMetadataBase::TConstPtr& readMetadata, const TActorId& scanActorId,
  const TActorId& resourceSubscribeActorId, const TActorId& readCoordinatorActorId, const TComputeShardingPolicy& computeShardingPolicy,
- const ui64 scanId)
+ const ui64 scanId, const NConveyor::TCPULimitsConfig& cpuLimits)
  : StoragesManager(storagesManager)
  , DataAccessorsManager(dataAccessorsManager)
  , Counters(counters)
@@ -24,7 +24,7 @@ TReadContext::TReadContext(const std::shared_ptr<IStoragesManager>& storagesMana
  , ResourceSubscribeActorId(resourceSubscribeActorId)
  , ReadCoordinatorActorId(readCoordinatorActorId)
  , ComputeShardingPolicy(computeShardingPolicy)
- , ConveyorProcessGuard(NConveyor::TScanServiceOperator::StartProcess(ScanId)) {
+ , ConveyorProcessGuard(NConveyor::TScanServiceOperator::StartProcess(ScanId, cpuLimits)) {
  Y_ABORT_UNLESS(ReadMetadata);
  if (ReadMetadata->HasResultSchema()) {
  Resolver = std::make_shared<NCommon::TIndexColumnResolver>(ReadMetadata->GetResultSchema()->GetIndexInfo());

ydb/core/tx/columnshard/engines/reader/abstract/read_context.h

@@ -7,6 +7,7 @@
 #include <ydb/core/tx/columnshard/data_accessor/manager.h>
 #include <ydb/core/tx/columnshard/resource_subscriber/task.h>
 #include <ydb/core/tx/conveyor/usage/abstract.h>
+#include <ydb/core/tx/conveyor/usage/config.h>
 
 #include <ydb/library/accessor/accessor.h>
 
@@ -148,7 +149,7 @@ class TReadContext {
  const std::shared_ptr<NDataAccessorControl::IDataAccessorsManager>& dataAccessorsManager,
  const NColumnShard::TConcreteScanCounters& counters, const TReadMetadataBase::TConstPtr& readMetadata, const TActorId& scanActorId,
  const TActorId& resourceSubscribeActorId, const TActorId& readCoordinatorActorId, const TComputeShardingPolicy& computeShardingPolicy,
- const ui64 scanId);
+ const ui64 scanId, const NConveyor::TCPULimitsConfig& cpuLimits);
 };
 
 class IDataReader {

ydb/core/tx/columnshard/engines/reader/abstract/ya.make

@@ -13,6 +13,7 @@ PEERDIR(
  ydb/core/tx/program
  ydb/core/protos
  ydb/core/tx/columnshard/data_sharing/protos
+ ydb/core/tx/conveyor/usage
 )
 
 GENERATE_ENUM_SERIALIZATION(read_metadata.h)

ydb/core/tx/columnshard/engines/reader/actor/actor.cpp

@@ -21,7 +21,7 @@ TColumnShardScan::TColumnShardScan(const TActorId& columnShardActorId, const TAc
  const std::shared_ptr<NDataAccessorControl::IDataAccessorsManager>& dataAccessorsManager,
  const TComputeShardingPolicy& computeShardingPolicy, ui32 scanId, ui64 txId, ui32 scanGen, ui64 requestCookie, ui64 tabletId,
  TDuration timeout, const TReadMetadataBase::TConstPtr& readMetadataRange, NKikimrDataEvents::EDataFormat dataFormat,
- const NColumnShard::TScanCounters& scanCountersPool)
+ const NColumnShard::TScanCounters& scanCountersPool, const NConveyor::TCPULimitsConfig& cpuLimits)
  : StoragesManager(storagesManager)
  , DataAccessorsManager(dataAccessorsManager)
  , ColumnShardActorId(columnShardActorId)
@@ -33,6 +33,7 @@ TColumnShardScan::TColumnShardScan(const TActorId& columnShardActorId, const TAc
  , RequestCookie(requestCookie)
  , DataFormat(dataFormat)
  , TabletId(tabletId)
+ , CPULimits(cpuLimits)
  , ReadMetadataRange(readMetadataRange)
  , Timeout(timeout ? timeout : COMPUTE_HARD_TIMEOUT)
  , ScanCountersPool(scanCountersPool, TValidator::CheckNotNull(ReadMetadataRange)->GetProgram().GetGraphOptional())
@@ -53,7 +54,7 @@ void TColumnShardScan::Bootstrap(const TActorContext& ctx) {
  ReadCoordinatorActorId = ctx.Register(new NBlobOperations::NRead::TReadCoordinatorActor(TabletId, SelfId()));
 
  std::shared_ptr<TReadContext> context = std::make_shared<TReadContext>(StoragesManager, DataAccessorsManager, ScanCountersPool,
- ReadMetadataRange, SelfId(), ResourceSubscribeActorId, ReadCoordinatorActorId, ComputeShardingPolicy, ScanId);
+ ReadMetadataRange, SelfId(), ResourceSubscribeActorId, ReadCoordinatorActorId, ComputeShardingPolicy, ScanId, CPULimits);
  ScanIterator = ReadMetadataRange->StartScan(context);
  auto startResult = ScanIterator->Start();
  StartInstant = TMonotonic::Now();