diff options
Diffstat (limited to 'library/cpp/actors/interconnect/ut')
4 files changed, 46 insertions, 22 deletions
diff --git a/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h b/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h index 2b6d27cd3f3..dd2557e25ee 100644 --- a/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h +++ b/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h @@ -26,13 +26,15 @@ private: TList<TTrafficInterrupter> interrupters; NActors::TChannelsConfig ChannelsConfig; TPortManager PortManager; + TIntrusivePtr<NLog::TSettings> LoggerSettings; public: TTestICCluster(ui32 numNodes = 1, NActors::TChannelsConfig channelsConfig = NActors::TChannelsConfig(), - TTrafficInterrupterSettings* tiSettings = nullptr) + TTrafficInterrupterSettings* tiSettings = nullptr, TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr) : NumNodes(numNodes) , Counters(new NMonitoring::TDynamicCounters) , ChannelsConfig(channelsConfig) + , LoggerSettings(loggerSettings) { THashMap<ui32, ui16> nodeToPortMap; THashMap<ui32, THashMap<ui32, ui16>> specificNodePortMap; @@ -59,7 +61,8 @@ public: for (ui32 i = 1; i <= NumNodes; ++i) { auto& portMap = tiSettings ? specificNodePortMap[i] : nodeToPortMap; - Nodes.emplace(i, MakeHolder<TNode>(i, NumNodes, portMap, Address, Counters, DeadPeerTimeout, ChannelsConfig)); + Nodes.emplace(i, MakeHolder<TNode>(i, NumNodes, portMap, Address, Counters, DeadPeerTimeout, ChannelsConfig, + /*numDynamicNodes=*/0, /*numThreads=*/1, LoggerSettings)); } } diff --git a/library/cpp/actors/interconnect/ut/lib/node.h b/library/cpp/actors/interconnect/ut/lib/node.h index ff30b1445e8..0b538cdb1cb 100644 --- a/library/cpp/actors/interconnect/ut/lib/node.h +++ b/library/cpp/actors/interconnect/ut/lib/node.h @@ -6,6 +6,7 @@ #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> @@ -19,7 +20,8 @@ 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) { + ui32 numDynamicNodes = 0, ui32 numThreads = 1, + TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr) { TActorSystemSetup setup; setup.NodeId = nodeId; setup.ExecutorsCount = 1; @@ -43,6 +45,7 @@ public: 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); @@ -62,29 +65,31 @@ public: setup.LocalServices.emplace_back(MakePollerActorId(), TActorSetupCmd(CreatePollerActor(), TMailboxType::ReadAsFilled, 0)); - const TActorId loggerActorId(0, "logger"); - constexpr ui32 LoggerComponentId = 410; // NKikimrServices::LOGGER + const TActorId loggerActorId = loggerSettings ? loggerSettings->LoggerActorId : TActorId(0, "logger"); - auto loggerSettings = MakeIntrusive<NLog::TSettings>( - loggerActorId, - (NLog::EComponent)LoggerComponentId, - NLog::PRI_INFO, - NLog::PRI_DEBUG, - 0U); + 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 - ); + loggerSettings->Append( + NActorsServices::EServiceCommon_MIN, + NActorsServices::EServiceCommon_MAX, + NActorsServices::EServiceCommon_Name + ); - constexpr ui32 WilsonComponentId = 430; // NKikimrServices::WILSON - static const TString WilsonComponentName = "WILSON"; + 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; }); + loggerSettings->Append( + (NLog::EComponent)WilsonComponentId, + (NLog::EComponent)WilsonComponentId + 1, + [](NLog::EComponent) -> const TString & { return WilsonComponentName; }); + } // register nameserver table auto names = MakeIntrusive<TTableNameserverSetup>(); @@ -105,6 +110,14 @@ public: 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(); diff --git a/library/cpp/actors/interconnect/ut/lib/test_events.h b/library/cpp/actors/interconnect/ut/lib/test_events.h index cd0d9e01520..1bb5eb7d383 100644 --- a/library/cpp/actors/interconnect/ut/lib/test_events.h +++ b/library/cpp/actors/interconnect/ut/lib/test_events.h @@ -9,6 +9,7 @@ namespace NActors { EvTestSmall, EvTestLarge, EvTestResponse, + EvTestStartPolling, }; struct TEvTest : TEventPB<TEvTest, NInterconnectTest::TEvTest, EvTest> { @@ -46,4 +47,8 @@ namespace NActors { } }; + struct TEvTestStartPolling : TEventPB<TEvTestStartPolling, NInterconnectTest::TEvTestStartPolling, EvTestStartPolling> { + TEvTestStartPolling() = default; + }; + } diff --git a/library/cpp/actors/interconnect/ut/protos/interconnect_test.proto b/library/cpp/actors/interconnect/ut/protos/interconnect_test.proto index b9b2bd6a4e3..b74d068a8b8 100644 --- a/library/cpp/actors/interconnect/ut/protos/interconnect_test.proto +++ b/library/cpp/actors/interconnect/ut/protos/interconnect_test.proto @@ -23,3 +23,6 @@ message TEvTestSmall { message TEvTestResponse { optional uint64 ConfirmedSequenceNumber = 1; } + +message TEvTestStartPolling { +} |