1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
#pragma once
#include <library/cpp/actors/core/actorsystem.h>
#include <library/cpp/actors/core/executor_pool_basic.h>
#include <library/cpp/actors/core/scheduler_basic.h>
#include <library/cpp/actors/core/mailbox.h>
#include <library/cpp/actors/dnsresolver/dnsresolver.h>
#include <library/cpp/actors/interconnect/handshake_broker.h>
#include <library/cpp/actors/interconnect/interconnect_tcp_server.h>
#include <library/cpp/actors/interconnect/interconnect_tcp_proxy.h>
#include <library/cpp/actors/interconnect/interconnect_proxy_wrapper.h>
using namespace NActors;
class TNode {
THolder<TActorSystem> ActorSystem;
public:
TNode(ui32 nodeId, ui32 numNodes, const THashMap<ui32, ui16>& nodeToPort, const TString& address,
NMonitoring::TDynamicCounterPtr counters, TDuration deadPeerTimeout,
TChannelsConfig channelsSettings = TChannelsConfig(),
ui32 numDynamicNodes = 0, ui32 numThreads = 1,
TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr) {
TActorSystemSetup setup;
setup.NodeId = nodeId;
setup.ExecutorsCount = 1;
setup.Executors.Reset(new TAutoPtr<IExecutorPool>[setup.ExecutorsCount]);
for (ui32 i = 0; i < setup.ExecutorsCount; ++i) {
setup.Executors[i].Reset(new TBasicExecutorPool(i, numThreads, 20 /* magic number */));
}
setup.Scheduler.Reset(new TBasicSchedulerThread());
const ui32 interconnectPoolId = 0;
auto common = MakeIntrusive<TInterconnectProxyCommon>();
common->NameserviceId = GetNameserviceActorId();
common->MonCounters = counters->GetSubgroup("nodeId", ToString(nodeId));
common->ChannelsConfig = channelsSettings;
common->ClusterUUID = "cluster";
common->AcceptUUID = {common->ClusterUUID};
common->TechnicalSelfHostName = address;
common->Settings.Handshake = TDuration::Seconds(1);
common->Settings.DeadPeer = deadPeerTimeout;
common->Settings.CloseOnIdle = TDuration::Minutes(1);
common->Settings.SendBufferDieLimitInMB = 512;
common->Settings.TotalInflightAmountOfData = 512 * 1024;
common->Settings.TCPSocketBufferSize = 2048 * 1024;
common->OutgoingHandshakeInflightLimit = 3;
setup.Interconnect.ProxyActors.resize(numNodes + 1 - numDynamicNodes);
setup.Interconnect.ProxyWrapperFactory = CreateProxyWrapperFactory(common, interconnectPoolId);
for (ui32 i = 1; i <= numNodes; ++i) {
if (i == nodeId) {
// create listener actor for local node "nodeId"
setup.LocalServices.emplace_back(TActorId(), TActorSetupCmd(new TInterconnectListenerTCP(address,
nodeToPort.at(nodeId), common), TMailboxType::ReadAsFilled, interconnectPoolId));
} else if (i <= numNodes - numDynamicNodes) {
// create proxy actor to reach node "i"
setup.Interconnect.ProxyActors[i] = {new TInterconnectProxyTCP(i, common),
TMailboxType::ReadAsFilled, interconnectPoolId};
}
}
setup.LocalServices.emplace_back(MakePollerActorId(), TActorSetupCmd(CreatePollerActor(),
TMailboxType::ReadAsFilled, 0));
const TActorId loggerActorId = loggerSettings ? loggerSettings->LoggerActorId : TActorId(0, "logger");
if (!loggerSettings) {
constexpr ui32 LoggerComponentId = 410; // NKikimrServices::LOGGER
loggerSettings = MakeIntrusive<NLog::TSettings>(
loggerActorId,
(NLog::EComponent)LoggerComponentId,
NLog::PRI_INFO,
NLog::PRI_DEBUG,
0U);
loggerSettings->Append(
NActorsServices::EServiceCommon_MIN,
NActorsServices::EServiceCommon_MAX,
NActorsServices::EServiceCommon_Name
);
constexpr ui32 WilsonComponentId = 430; // NKikimrServices::WILSON
static const TString WilsonComponentName = "WILSON";
loggerSettings->Append(
(NLog::EComponent)WilsonComponentId,
(NLog::EComponent)WilsonComponentId + 1,
[](NLog::EComponent) -> const TString & { return WilsonComponentName; });
}
// register nameserver table
auto names = MakeIntrusive<TTableNameserverSetup>();
for (ui32 i = 1; i <= numNodes; ++i) {
names->StaticNodeTable[i] = TTableNameserverSetup::TNodeInfo(address, address, nodeToPort.at(i));
}
setup.LocalServices.emplace_back(
NDnsResolver::MakeDnsResolverActorId(),
TActorSetupCmd(
NDnsResolver::CreateOnDemandDnsResolver(),
TMailboxType::ReadAsFilled, interconnectPoolId));
setup.LocalServices.emplace_back(GetNameserviceActorId(), TActorSetupCmd(
CreateNameserverTable(names, interconnectPoolId), TMailboxType::ReadAsFilled,
interconnectPoolId));
// register logger
setup.LocalServices.emplace_back(loggerActorId, TActorSetupCmd(new TLoggerActor(loggerSettings,
CreateStderrBackend(), counters->GetSubgroup("subsystem", "logger")),
TMailboxType::ReadAsFilled, interconnectPoolId));
if (common->OutgoingHandshakeInflightLimit) {
// create handshake broker actor
setup.LocalServices.emplace_back(MakeHandshakeBrokerOutId(), TActorSetupCmd(
CreateHandshakeBroker(*common->OutgoingHandshakeInflightLimit),
TMailboxType::ReadAsFilled, interconnectPoolId));
}
auto sp = MakeHolder<TActorSystemSetup>(std::move(setup));
ActorSystem.Reset(new TActorSystem(sp, nullptr, loggerSettings));
ActorSystem->Start();
}
~TNode() {
ActorSystem->Stop();
}
bool Send(const TActorId& recipient, IEventBase* ev) {
return ActorSystem->Send(recipient, ev);
}
TActorId RegisterActor(IActor* actor) {
return ActorSystem->Register(actor);
}
TActorId InterconnectProxy(ui32 peerNodeId) {
return ActorSystem->InterconnectProxy(peerNodeId);
}
void RegisterServiceActor(const TActorId& serviceId, IActor* actor) {
const TActorId actorId = ActorSystem->Register(actor);
ActorSystem->RegisterLocalService(serviceId, actorId);
}
TActorSystem *GetActorSystem() const {
return ActorSystem.Get();
}
};
|
