Skip to content

Commit 1cb70e3

Browse files
kruallkruall
authored
[stable-25-1] Improve tiny ydb perfomance (#19909) (#19926)
2 parents a38de79 + a2e8e48 commit 1cb70e3

File tree

10 files changed

+240
-48
lines changed

10 files changed

+240
-48
lines changed

‎ydb/core/driver_lib/run/auto_config_initializer.cpp

Lines changed: 54 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -244,15 +244,66 @@ namespace NKikimr::NAutoConfigInitializer {
244244
scheduler->SetProgressThreshold(10'000);
245245
}
246246

247+
auto *serviceExecutor = config->AddServiceExecutor();
248+
serviceExecutor->SetServiceName("Interconnect");
249+
250+
if (useSharedThreads && cpuCount >= 1 && cpuCount <= 3) {
251+
config->SetUserExecutor(0);
252+
config->SetSysExecutor(1);
253+
config->SetBatchExecutor(2);
254+
config->SetIoExecutor(3);
255+
serviceExecutor->SetExecutorId(4);
256+
257+
auto *systemExecutor = config->AddExecutor();
258+
auto *userExecutor = config->AddExecutor();
259+
auto *batchExecutor = config->AddExecutor();
260+
auto *ioExecutor = config->AddExecutor();
261+
auto *icExecutor = config->AddExecutor();
262+
263+
ioExecutor->SetType(NKikimrConfig::TActorSystemConfig::TExecutor::IO);
264+
ioExecutor->SetThreads(config->HasForceIOPoolThreads() ? config->GetForceIOPoolThreads() : 1);
265+
ioExecutor->SetName("IO");
266+
267+
auto assignPool = [&](auto *executor, TString name, i16 priority, bool hasSharedThread) {
268+
executor->SetType(NKikimrConfig::TActorSystemConfig::TExecutor::BASIC);
269+
executor->SetThreads(hasSharedThread);
270+
executor->SetMaxThreads(hasSharedThread);
271+
executor->SetName(name);
272+
executor->SetPriority(priority);
273+
executor->SetSpinThreshold(0);
274+
executor->SetHasSharedThread(hasSharedThread);
275+
};
276+
277+
assignPool(systemExecutor, "System", 30, cpuCount >= 3);
278+
assignPool(userExecutor, "User", 20, cpuCount >= 2);
279+
assignPool(batchExecutor, "Batch", 10, false);
280+
assignPool(icExecutor, "IC", 40, true);
281+
282+
batchExecutor->SetForcedForeignSlots(1);
283+
userExecutor->SetForcedForeignSlots(2);
284+
icExecutor->SetForcedForeignSlots(2);
285+
systemExecutor->SetForcedForeignSlots(2);
286+
287+
if (cpuCount >= 2) {
288+
userExecutor->AddAdjacentPools(2);
289+
}
290+
if (cpuCount <= 2) {
291+
icExecutor->AddAdjacentPools(0);
292+
}
293+
if (cpuCount == 1) {
294+
icExecutor->AddAdjacentPools(1);
295+
icExecutor->AddAdjacentPools(2);
296+
}
297+
298+
return;
299+
}
300+
247301
TASPools pools = GetASPools(cpuCount);
248302
ui8 poolCount = pools.GetRealPoolCount();
249303
std::vector<TString> names = pools.GetRealPoolNames();
250304
std::vector<ui8> executorIds = pools.GetIndeces();
251305
std::vector<ui8> priorities = pools.GetPriorities();
252306

253-
auto *serviceExecutor = config->AddServiceExecutor();
254-
serviceExecutor->SetServiceName("Interconnect");
255-
256307
config->SetUserExecutor(pools.SystemPoolId);
257308
config->SetSysExecutor(pools.UserPoolId);
258309
config->SetBatchExecutor(pools.BatchPoolId);

‎ydb/core/driver_lib/run/config_helpers.cpp

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,8 +92,14 @@ void AddExecutorPool(NActors::TCpuManagerConfig& cpuManager, const NKikimrConfig
9292
if (poolConfig.HasMaxLocalQueueSize()) {
9393
basic.MaxLocalQueueSize = poolConfig.GetMaxLocalQueueSize();
9494
}
95+
for (const auto& pool : poolConfig.GetAdjacentPools()) {
96+
basic.AdjacentPools.push_back(pool);
97+
}
98+
if (poolConfig.HasForcedForeignSlots()) {
99+
basic.ForcedForeignSlotCount = poolConfig.GetForcedForeignSlots();
100+
}
95101
cpuManager.Basic.emplace_back(std::move(basic));
96-
102+
97103
break;
98104
}
99105

‎ydb/core/driver_lib/run/run.cpp

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1115,9 +1115,20 @@ void TKikimrRunner::InitializeAppData(const TKikimrRunConfig& runConfig)
11151115
const auto& cfg = runConfig.AppConfig;
11161116

11171117
bool useAutoConfig = !cfg.HasActorSystemConfig() || (cfg.GetActorSystemConfig().HasUseAutoConfig() && cfg.GetActorSystemConfig().GetUseAutoConfig());
1118+
bool useSharedThreads = cfg.HasActorSystemConfig() && cfg.GetActorSystemConfig().HasUseSharedThreads() && cfg.GetActorSystemConfig().GetUseSharedThreads();
11181119
NAutoConfigInitializer::TASPools pools = NAutoConfigInitializer::GetASPools(cfg.GetActorSystemConfig(), useAutoConfig);
11191120
TMap<TString, ui32> servicePools = NAutoConfigInitializer::GetServicePools(cfg.GetActorSystemConfig(), useAutoConfig);
11201121

1122+
if (useSharedThreads) {
1123+
pools.SystemPoolId = 0;
1124+
pools.UserPoolId = 1;
1125+
pools.BatchPoolId = 2;
1126+
pools.IOPoolId = 3;
1127+
pools.ICPoolId = 4;
1128+
servicePools.clear();
1129+
servicePools["Interconnect"] = 4;
1130+
}
1131+
11211132
AppData.Reset(new TAppData(pools.SystemPoolId, pools.UserPoolId, pools.IOPoolId, pools.BatchPoolId,
11221133
servicePools,
11231134
TypeRegistry.Get(),

‎ydb/core/protos/config.proto

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,10 @@ message TActorSystemConfig {
8888
optional bool HasSharedThread = 18;
8989
optional uint32 MaxLocalQueueSize = 20;
9090
optional uint32 MinLocalQueueSize = 21;
91+
92+
// Tiny YDB
93+
repeated uint32 AdjacentPools = 22;
94+
optional uint32 ForcedForeignSlots = 23;
9195
}
9296

9397
message TScheduler {

‎ydb/library/actors/core/config.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,10 @@ namespace NActors {
3434
bool UseRingQueue = false;
3535
ui16 MinLocalQueueSize = 0;
3636
ui16 MaxLocalQueueSize = 0;
37+
38+
// tiny-ydb configs
39+
std::vector<i16> AdjacentPools;
40+
i16 ForcedForeignSlotCount = 0;
3741
};
3842

3943
struct TSharedExecutorPoolConfig {

‎ydb/library/actors/core/cpu_manager.cpp

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,10 +56,26 @@ namespace NActors {
5656
if (sharedThreadCount) {
5757
sht = 1;
5858
}
59-
poolInfos.push_back(TPoolShortInfo{static_cast<i16>(Config.Basic[poolIds[i]].PoolId), sharedThreadCount, true, Config.Basic[poolIds[i]].PoolName});
59+
poolInfos.push_back(TPoolShortInfo{
60+
.PoolId = static_cast<i16>(Config.Basic[poolIds[i]].PoolId),
61+
.SharedThreadCount = sharedThreadCount,
62+
.ForeignSlots = Config.Basic[poolIds[i]].ForcedForeignSlotCount,
63+
.InPriorityOrder = true,
64+
.PoolName = Config.Basic[poolIds[i]].PoolName,
65+
.ForcedForeignSlots = Config.Basic[poolIds[i]].ForcedForeignSlotCount > 0,
66+
.AdjacentPools = Config.Basic[poolIds[i]].AdjacentPools,
67+
});
6068
}
6169
for (ui32 i = 0; i < Config.IO.size(); ++i) {
62-
poolInfos.push_back(TPoolShortInfo{static_cast<i16>(Config.IO[i].PoolId), 0, false, Config.IO[i].PoolName});
70+
poolInfos.push_back(TPoolShortInfo{
71+
.PoolId = static_cast<i16>(Config.IO[poolIds[i]].PoolId),
72+
.SharedThreadCount = 0,
73+
.ForeignSlots = 0,
74+
.InPriorityOrder = false,
75+
.PoolName = Config.IO[i].PoolName,
76+
.ForcedForeignSlots = false,
77+
.AdjacentPools = {},
78+
});
6379
}
6480
Shared = std::make_unique<TSharedExecutorPool>(Config.Shared, poolInfos);
6581

‎ydb/library/actors/core/executor_pool_shared.cpp

Lines changed: 57 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,48 @@ namespace NActors {
4949
});
5050
}
5151

52+
namespace {
53+
bool CheckPoolAdjacency(const TPoolManager& poolManager, i16 poolId, i16 adjacentPoolId) {
54+
if (poolId == adjacentPoolId) {
55+
return true;
56+
}
57+
Y_ABORT_UNLESS((ui32)poolId < poolManager.PoolInfos.size());
58+
const auto& poolInfo = poolManager.PoolInfos[poolId];
59+
return std::find(poolInfo.AdjacentPools.begin(), poolInfo.AdjacentPools.end(), adjacentPoolId) != poolInfo.AdjacentPools.end();
60+
}
61+
62+
bool HasAdjacentPools(const TPoolManager& poolManager, i16 poolId) {
63+
Y_ABORT_UNLESS((ui32)poolId < poolManager.PoolInfos.size());
64+
const auto& poolInfo = poolManager.PoolInfos[poolId];
65+
return !poolInfo.AdjacentPools.empty();
66+
}
67+
68+
i16 NextAdjacentPool(const TPoolManager& poolManager, i16 poolId, i16 currentPoolId) {
69+
if (poolId == currentPoolId) {
70+
if (poolManager.PoolInfos[poolId].AdjacentPools.empty()) {
71+
return poolId;
72+
}
73+
return poolManager.PoolInfos[poolId].AdjacentPools[0];
74+
}
75+
Y_ABORT_UNLESS((ui32)poolId < poolManager.PoolInfos.size());
76+
const auto& poolInfo = poolManager.PoolInfos[poolId];
77+
auto it = std::find(poolInfo.AdjacentPools.begin(), poolInfo.AdjacentPools.end(), currentPoolId);
78+
if (it == poolInfo.AdjacentPools.end() || it + 1 == poolInfo.AdjacentPools.end()) {
79+
return poolId;
80+
}
81+
return *(it + 1);
82+
}
83+
84+
std::optional<i16> GetForcedForeignSlots(const TPoolManager& poolManager, i16 poolId) {
85+
const auto& poolInfo = poolManager.PoolInfos[poolId];
86+
if (poolInfo.ForcedForeignSlots) {
87+
return poolInfo.ForcedForeignSlots;
88+
}
89+
return std::nullopt;
90+
}
91+
92+
}
93+
5294
LWTRACE_USING(ACTORLIB_PROVIDER);
5395

5496
TSharedExecutorPool::TSharedExecutorPool(
@@ -80,8 +122,8 @@ namespace NActors {
80122
}
81123
}
82124
for (ui64 i = 0; i < PoolManager.PoolInfos.size(); ++i) {
83-
ForeignThreadsAllowedByPool[i].store(0, std::memory_order_release);
84-
ForeignThreadSlots[i].store(0, std::memory_order_release);
125+
ForeignThreadsAllowedByPool[i].store(PoolManager.PoolInfos[i].ForeignSlots, std::memory_order_release);
126+
ForeignThreadSlots[i].store(PoolManager.PoolInfos[i].ForeignSlots, std::memory_order_release);
85127
LocalThreads[i].store(PoolManager.PoolInfos[i].SharedThreadCount, std::memory_order_release);
86128
LocalNotifications[i].store(0, std::memory_order_release);
87129
}
@@ -118,7 +160,7 @@ namespace NActors {
118160
continue;
119161
}
120162

121-
if (thread.OwnerPoolId == i) {
163+
if (CheckPoolAdjacency(PoolManager, thread.OwnerPoolId, i)) {
122164
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::Executor, "ownerPoolId == poolId; ownerPoolId == ", thread.OwnerPoolId, " poolId == ", i);
123165
return i;
124166
}
@@ -162,8 +204,10 @@ namespace NActors {
162204
auto &thread = Threads[workerId];
163205
thread.UnsetWork();
164206
TMailbox *mailbox = nullptr;
207+
bool hasAdjacentPools = HasAdjacentPools(PoolManager, thread.OwnerPoolId);
165208
while (!StopFlag.load(std::memory_order_acquire)) {
166-
if (hpnow < thread.SoftDeadlineForPool || thread.CurrentPoolId == thread.OwnerPoolId) {
209+
bool adjacentPool = CheckPoolAdjacency(PoolManager, thread.OwnerPoolId, thread.CurrentPoolId);
210+
if (hpnow < thread.SoftDeadlineForPool || !hasAdjacentPools && adjacentPool) {
167211
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::Activation, "continue same pool; ownerPoolId == ", thread.OwnerPoolId, " currentPoolId == ", thread.CurrentPoolId);
168212
if (thread.SoftDeadlineForPool == Max<NHPTimer::STime>()) {
169213
thread.SoftDeadlineForPool = GetCycleCountFast() + thread.SoftProcessingDurationTs;
@@ -181,6 +225,7 @@ namespace NActors {
181225
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::Executor, "no mailbox and need to find new pool; ownerPoolId == ", thread.OwnerPoolId, " currentPoolId == ", thread.CurrentPoolId, " processedActivationsByCurrentPool == ", TlsThreadContext->ProcessedActivationsByCurrentPool);
182226
TlsThreadContext->ProcessedActivationsByCurrentPool = 0;
183227
if (thread.CurrentPoolId != thread.OwnerPoolId) {
228+
thread.AdjacentPoolId = NextAdjacentPool(PoolManager, thread.OwnerPoolId, thread.AdjacentPoolId);
184229
SwitchToPool(thread.OwnerPoolId, hpnow);
185230
continue;
186231
}
@@ -190,10 +235,11 @@ namespace NActors {
190235
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::Activation, "no mailbox and no need to wait; ownerPoolId == ", thread.OwnerPoolId, " currentPoolId == ", thread.CurrentPoolId);
191236
return nullptr;
192237
} else {
193-
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::Executor, "comeback to owner pool; ownerPoolId == ", thread.OwnerPoolId, " currentPoolId == ", thread.CurrentPoolId, " processedActivationsByCurrentPool == ", TlsThreadContext->ProcessedActivationsByCurrentPool, " hpnow == ", hpnow, " softDeadlineForPool == ", thread.SoftDeadlineForPool);
238+
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::Executor, "change adjacent pool; ownerPoolId == ", thread.OwnerPoolId, " currentPoolId == ", thread.CurrentPoolId, " processedActivationsByCurrentPool == ", TlsThreadContext->ProcessedActivationsByCurrentPool, " hpnow == ", hpnow, " softDeadlineForPool == ", thread.SoftDeadlineForPool);
194239
TlsThreadContext->ProcessedActivationsByCurrentPool = 0;
195-
SwitchToPool(thread.OwnerPoolId, hpnow);
196-
// after soft deadline we check owner pool again
240+
thread.AdjacentPoolId = NextAdjacentPool(PoolManager, thread.OwnerPoolId, thread.AdjacentPoolId);
241+
SwitchToPool(thread.AdjacentPoolId, hpnow);
242+
// after soft deadline we check adjacent pool
197243
continue;
198244
}
199245
bool goToSleep = true;
@@ -308,7 +354,7 @@ namespace NActors {
308354
Y_ABORT_UNLESS(Threads[i].OwnerPoolId < static_cast<i16>(Pools.size()), "OwnerPoolId is out of range i %" PRIu16 " OwnerPoolId == %" PRIu16, i, Threads[i].OwnerPoolId);
309355
Y_ABORT_UNLESS(Threads[i].OwnerPoolId >= 0, "OwnerPoolId is out of range i %" PRIu16 " OwnerPoolId == %" PRIu16, i, Threads[i].OwnerPoolId);
310356
EXECUTOR_POOL_SHARED_DEBUG(EDebugLevel::ExecutorPool, "create thread ", i, " OwnerPoolId == ", Threads[i].OwnerPoolId);
311-
Threads[i].Thread.reset(
357+
Threads[i].Thread.reset(
312358
new TExecutorThread(
313359
i,
314360
actorSystem,
@@ -482,8 +528,6 @@ namespace NActors {
482528
return false;
483529
}
484530

485-
486-
487531
void TSharedExecutorPool::FillForeignThreadsAllowed(std::vector<i16>& foreignThreadsAllowed) const {
488532
foreignThreadsAllowed.resize(PoolManager.PoolInfos.size());
489533
for (ui64 i = 0; i < foreignThreadsAllowed.size(); ++i) {
@@ -513,6 +557,9 @@ namespace NActors {
513557
}
514558

515559
void TSharedExecutorPool::SetForeignThreadSlots(i16 poolId, i16 slots) {
560+
if (auto forcedSlots = GetForcedForeignSlots(PoolManager, poolId)) {
561+
return;
562+
}
516563
i16 current = ForeignThreadsAllowedByPool[poolId].load(std::memory_order_acquire);
517564
if (current == slots) {
518565
return;

‎ydb/library/actors/core/executor_pool_shared.h

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,11 @@ namespace NActors {
2929
struct TPoolShortInfo {
3030
i16 PoolId = 0;
3131
i16 SharedThreadCount = 0;
32+
i16 ForeignSlots = 0;
3233
bool InPriorityOrder = false;
3334
TString PoolName;
35+
bool ForcedForeignSlots = false;
36+
std::vector<i16> AdjacentPools;
3437
};
3538

3639
struct TPoolThreadRange {
@@ -42,7 +45,7 @@ namespace NActors {
4245
TStackVec<TPoolShortInfo, 8> PoolInfos;
4346
TStackVec<TPoolThreadRange, 8> PoolThreadRanges;
4447
TStackVec<i16, 8> PriorityOrder;
45-
48+
4649
TPoolManager(const TVector<TPoolShortInfo> &poolInfos);
4750
};
4851

‎ydb/library/actors/core/executor_thread_ctx.h

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ namespace NActors {
6464
template <typename TDerived, typename TWaitState>
6565
bool Sleep(std::atomic<bool> *stopFlag);
6666
};
67-
67+
6868
struct TExecutorThreadCtx : public TGenericExecutorThreadCtx {
6969
using TBase = TGenericExecutorThreadCtx;
7070

@@ -104,6 +104,7 @@ namespace NActors {
104104
i16 PoolLeaseIndex = -1;
105105
i16 OwnerPoolId = -1;
106106
i16 CurrentPoolId = -1;
107+
i16 AdjacentPoolId = -1;
107108
NHPTimer::STime SoftDeadlineForPool = 0;
108109
NHPTimer::STime SoftProcessingDurationTs = 0;
109110

0 commit comments

Comments
 (0)