diff options
| author | Alexey Borzenkov <snaury@yandex-team.ru> | 2022-02-10 16:47:43 +0300 |
|---|---|---|
| committer | Daniil Cherednik <dcherednik@yandex-team.ru> | 2022-02-10 16:47:43 +0300 |
| commit | 330c83f8c116bd45316397b179275e9d87007e7d (patch) | |
| tree | c0748b5dcbade83af788c0abfa89c0383d6b779c /library/cpp/actors | |
| parent | 22d92781ba2a10b7fb5b977b7d1a5c40ff53885f (diff) | |
| download | ydb-330c83f8c116bd45316397b179275e9d87007e7d.tar.gz | |
Restoring authorship annotation for Alexey Borzenkov <snaury@yandex-team.ru>. Commit 2 of 2.
Diffstat (limited to 'library/cpp/actors')
55 files changed, 3344 insertions, 3344 deletions
diff --git a/library/cpp/actors/core/actor.cpp b/library/cpp/actors/core/actor.cpp index 638bfb72fa..6f9ba6a42b 100644 --- a/library/cpp/actors/core/actor.cpp +++ b/library/cpp/actors/core/actor.cpp @@ -37,10 +37,10 @@ namespace NActors { TlsActivationContext->ExecutorThread.Schedule(deadline, ev, cookie); } - void TActivationContext::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) { - TlsActivationContext->ExecutorThread.Schedule(deadline, ev, cookie); - } - + void TActivationContext::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) { + TlsActivationContext->ExecutorThread.Schedule(deadline, ev, cookie); + } + void TActivationContext::Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) { TlsActivationContext->ExecutorThread.Schedule(delta, ev, cookie); } @@ -87,14 +87,14 @@ namespace NActors { return TlsActivationContext->ExecutorThread.ActorSystem; } - i64 TActivationContext::GetCurrentEventTicks() { + i64 TActivationContext::GetCurrentEventTicks() { return GetCycleCountFast() - TlsActivationContext->EventStart; - } - - double TActivationContext::GetCurrentEventTicksAsSeconds() { - return NHPTimer::GetSeconds(GetCurrentEventTicks()); - } - + } + + double TActivationContext::GetCurrentEventTicksAsSeconds() { + return NHPTimer::GetSeconds(GetCurrentEventTicks()); + } + TActorId TActorContext::Register(IActor* actor, TMailboxType::EType mailboxType, ui32 poolId) const { return ExecutorThread.RegisterActor(actor, mailboxType, poolId, SelfID); } @@ -107,10 +107,10 @@ namespace NActors { ExecutorThread.Schedule(deadline, new IEventHandle(SelfID, TActorId(), ev), cookie); } - void TActorContext::Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie) const { - ExecutorThread.Schedule(deadline, new IEventHandle(SelfID, TActorId(), ev), cookie); - } - + void TActorContext::Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie) const { + ExecutorThread.Schedule(deadline, new IEventHandle(SelfID, TActorId(), ev), cookie); + } + void TActorContext::Schedule(TDuration delta, IEventBase* ev, ISchedulerCookie* cookie) const { ExecutorThread.Schedule(delta, new IEventHandle(SelfID, TActorId(), ev), cookie); } @@ -119,10 +119,10 @@ namespace NActors { TlsActivationContext->ExecutorThread.Schedule(deadline, new IEventHandle(SelfActorId, TActorId(), ev), cookie); } - void IActor::Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie) const noexcept { - TlsActivationContext->ExecutorThread.Schedule(deadline, new IEventHandle(SelfActorId, TActorId(), ev), cookie); - } - + void IActor::Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie) const noexcept { + TlsActivationContext->ExecutorThread.Schedule(deadline, new IEventHandle(SelfActorId, TActorId(), ev), cookie); + } + void IActor::Schedule(TDuration delta, IEventBase* ev, ISchedulerCookie* cookie) const noexcept { TlsActivationContext->ExecutorThread.Schedule(delta, new IEventHandle(SelfActorId, TActorId(), ev), cookie); } @@ -131,10 +131,10 @@ namespace NActors { return TlsActivationContext->ExecutorThread.ActorSystem->Timestamp(); } - TMonotonic TActivationContext::Monotonic() { - return TlsActivationContext->ExecutorThread.ActorSystem->Monotonic(); - } - + TMonotonic TActivationContext::Monotonic() { + return TlsActivationContext->ExecutorThread.ActorSystem->Monotonic(); + } + TInstant TActorContext::Now() const { return ExecutorThread.ActorSystem->Timestamp(); } diff --git a/library/cpp/actors/core/actor.h b/library/cpp/actors/core/actor.h index 9c167ab595..ed29bd14b9 100644 --- a/library/cpp/actors/core/actor.h +++ b/library/cpp/actors/core/actor.h @@ -1,7 +1,7 @@ #pragma once #include "event.h" -#include "monotonic.h" +#include "monotonic.h" #include <util/system/tls.h> #include <library/cpp/actors/util/local_process_key.h> @@ -24,13 +24,13 @@ namespace NActors { public: TMailboxHeader& Mailbox; TExecutorThread& ExecutorThread; - const NHPTimer::STime EventStart; + const NHPTimer::STime EventStart; protected: - explicit TActivationContext(TMailboxHeader& mailbox, TExecutorThread& executorThread, NHPTimer::STime eventStart) + explicit TActivationContext(TMailboxHeader& mailbox, TExecutorThread& executorThread, NHPTimer::STime eventStart) : Mailbox(mailbox) , ExecutorThread(executorThread) - , EventStart(eventStart) + , EventStart(eventStart) { } @@ -40,22 +40,22 @@ namespace NActors { /** * Schedule one-shot event that will be send at given time point in the future. * - * @param deadline the wallclock time point in future when event must be send + * @param deadline the wallclock time point in future when event must be send * @param ev the event to send * @param cookie cookie that will be piggybacked with event */ static void Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); /** - * Schedule one-shot event that will be send at given time point in the future. - * - * @param deadline the monotonic time point in future when event must be send - * @param ev the event to send - * @param cookie cookie that will be piggybacked with event - */ - static void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); - - /** + * Schedule one-shot event that will be send at given time point in the future. + * + * @param deadline the monotonic time point in future when event must be send + * @param ev the event to send + * @param cookie cookie that will be piggybacked with event + */ + static void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); + + /** * Schedule one-shot event that will be send after given delay. * * @param delta the time from now to delay event sending @@ -65,7 +65,7 @@ namespace NActors { static void Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); static TInstant Now(); - static TMonotonic Monotonic(); + static TMonotonic Monotonic(); NLog::TSettings* LoggerSettings() const; // register new actor in ActorSystem on new fresh mailbox. @@ -83,16 +83,16 @@ namespace NActors { static TActorId InterconnectProxy(ui32 nodeid); static TActorSystem* ActorSystem(); - - static i64 GetCurrentEventTicks(); - static double GetCurrentEventTicksAsSeconds(); + + static i64 GetCurrentEventTicks(); + static double GetCurrentEventTicksAsSeconds(); }; struct TActorContext: public TActivationContext { const TActorId SelfID; explicit TActorContext(TMailboxHeader& mailbox, TExecutorThread& executorThread, NHPTimer::STime eventStart, const TActorId& selfID) - : TActivationContext(mailbox, executorThread, eventStart) + : TActivationContext(mailbox, executorThread, eventStart) , SelfID(selfID) { } @@ -110,22 +110,22 @@ namespace NActors { /** * Schedule one-shot event that will be send at given time point in the future. * - * @param deadline the wallclock time point in future when event must be send + * @param deadline the wallclock time point in future when event must be send * @param ev the event to send * @param cookie cookie that will be piggybacked with event */ void Schedule(TInstant deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const; /** - * Schedule one-shot event that will be send at given time point in the future. - * - * @param deadline the monotonic time point in future when event must be send - * @param ev the event to send - * @param cookie cookie that will be piggybacked with event - */ - void Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const; - - /** + * Schedule one-shot event that will be send at given time point in the future. + * + * @param deadline the monotonic time point in future when event must be send + * @param ev the event to send + * @param cookie cookie that will be piggybacked with event + */ + void Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const; + + /** * Schedule one-shot event that will be send after given delay. * * @param delta the time from now to delay event sending @@ -135,7 +135,7 @@ namespace NActors { void Schedule(TDuration delta, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const; TActorContext MakeFor(const TActorId& otherId) const { - return TActorContext(Mailbox, ExecutorThread, EventStart, otherId); + return TActorContext(Mailbox, ExecutorThread, EventStart, otherId); } // register new actor in ActorSystem on new fresh mailbox. @@ -179,22 +179,22 @@ namespace NActors { /** * Schedule one-shot event that will be send at given time point in the future. * - * @param deadline the wallclock time point in future when event must be send + * @param deadline the wallclock time point in future when event must be send * @param ev the event to send * @param cookie cookie that will be piggybacked with event */ virtual void Schedule(TInstant deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept = 0; /** - * Schedule one-shot event that will be send at given time point in the future. - * - * @param deadline the monotonic time point in future when event must be send - * @param ev the event to send - * @param cookie cookie that will be piggybacked with event - */ - virtual void Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept = 0; - - /** + * Schedule one-shot event that will be send at given time point in the future. + * + * @param deadline the monotonic time point in future when event must be send + * @param ev the event to send + * @param cookie cookie that will be piggybacked with event + */ + virtual void Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept = 0; + + /** * Schedule one-shot event that will be send after given delay. * * @param delta the time from now to delay event sending @@ -240,11 +240,11 @@ namespace NActors { INTERCONNECT_SESSION_KILLER = 286, ACTOR_SYSTEM_SCHEDULER_ACTOR = 312, ACTOR_FUTURE_CALLBACK = 337, - INTERCONNECT_MONACTOR = 362, + INTERCONNECT_MONACTOR = 362, INTERCONNECT_LOAD_ACTOR = 376, INTERCONNECT_LOAD_RESPONDER = 377, NAMESERVICE = 450, - DNS_RESOLVER = 481, + DNS_RESOLVER = 481, INTERCONNECT_PROXY_WRAPPER = 546, }; @@ -362,7 +362,7 @@ namespace NActors { } void Schedule(TInstant deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept final; - void Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept final; + void Schedule(TMonotonic deadline, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept final; void Schedule(TDuration delta, IEventBase* ev, ISchedulerCookie* cookie = nullptr) const noexcept final; // register new actor in ActorSystem on new fresh mailbox. @@ -456,7 +456,7 @@ namespace NActors { inline TActorContext TActivationContext::ActorContextFor(TActorId id) { auto& tls = *TlsActivationContext; - return TActorContext(tls.Mailbox, tls.ExecutorThread, tls.EventStart, id); + return TActorContext(tls.Mailbox, tls.ExecutorThread, tls.EventStart, id); } class TDecorator : public IActor { diff --git a/library/cpp/actors/core/actor_coroutine.cpp b/library/cpp/actors/core/actor_coroutine.cpp index d3bcdafbd3..0ab4d2b24d 100644 --- a/library/cpp/actors/core/actor_coroutine.cpp +++ b/library/cpp/actors/core/actor_coroutine.cpp @@ -1,9 +1,9 @@ #include "actor_coroutine.h" #include "executor_thread.h" -#include <util/system/sanitizers.h> +#include <util/system/sanitizers.h> #include <util/system/type_name.h> - + namespace NActors { static constexpr size_t StackOverflowGap = 4096; static char GoodStack[StackOverflowGap]; @@ -92,8 +92,8 @@ namespace NActors { } // prepare actor context for in-coroutine use - TActivationContext *ac = TlsActivationContext; - TlsActivationContext = nullptr; + TActivationContext *ac = TlsActivationContext; + TlsActivationContext = nullptr; TActorContext ctx(ac->Mailbox, ac->ExecutorThread, ac->EventStart, SelfActorId); ActorContext = &ctx; diff --git a/library/cpp/actors/core/actorid.h b/library/cpp/actors/core/actorid.h index c9e6173173..d972b1a0ff 100644 --- a/library/cpp/actors/core/actorid.h +++ b/library/cpp/actors/core/actorid.h @@ -12,13 +12,13 @@ namespace NActors { // next 20 bits - node id itself struct TActorId { - static constexpr ui32 MaxServiceIDLength = 12; - static constexpr ui32 MaxPoolID = 0x000007FF; - static constexpr ui32 MaxNodeId = 0x000FFFFF; - static constexpr ui32 PoolIndexShift = 20; - static constexpr ui32 PoolIndexMask = MaxPoolID << PoolIndexShift; - static constexpr ui32 ServiceMask = 0x80000000; - static constexpr ui32 NodeIdMask = MaxNodeId; + static constexpr ui32 MaxServiceIDLength = 12; + static constexpr ui32 MaxPoolID = 0x000007FF; + static constexpr ui32 MaxNodeId = 0x000FFFFF; + static constexpr ui32 PoolIndexShift = 20; + static constexpr ui32 PoolIndexMask = MaxPoolID << PoolIndexShift; + static constexpr ui32 ServiceMask = 0x80000000; + static constexpr ui32 NodeIdMask = MaxNodeId; private: union { diff --git a/library/cpp/actors/core/actorsystem.cpp b/library/cpp/actors/core/actorsystem.cpp index 4130b0d4da..c58698a206 100644 --- a/library/cpp/actors/core/actorsystem.cpp +++ b/library/cpp/actors/core/actorsystem.cpp @@ -43,7 +43,7 @@ namespace NActors { , Scheduler(setup->Scheduler) , InterconnectCount((ui32)setup->Interconnect.ProxyActors.size()) , CurrentTimestamp(0) - , CurrentMonotonic(0) + , CurrentMonotonic(0) , CurrentIDCounter(RandomNumber<ui64>()) , SystemSetup(setup.Release()) , DefSelfID(NodeId, "actorsystem") @@ -69,15 +69,15 @@ namespace NActors { #endif TActorId recipient = ev->GetRecipientRewrite(); - const ui32 recpNodeId = recipient.NodeId(); + const ui32 recpNodeId = recipient.NodeId(); if (recpNodeId != NodeId && recpNodeId != 0) { // if recipient is not local one - rewrite with forward instruction Y_VERIFY_DEBUG(!ev->HasEvent() || ev->GetBase()->IsSerializable()); - Y_VERIFY(ev->Recipient == recipient, - "Event rewrite from %s to %s would be lost via interconnect", - ev->Recipient.ToString().c_str(), - recipient.ToString().c_str()); + Y_VERIFY(ev->Recipient == recipient, + "Event rewrite from %s to %s would be lost via interconnect", + ev->Recipient.ToString().c_str(), + recipient.ToString().c_str()); recipient = InterconnectProxy(recpNodeId); ev->Rewrite(TEvInterconnect::EvForward, recipient); } @@ -119,20 +119,20 @@ namespace NActors { } void TActorSystem::Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const { - Schedule(deadline - Timestamp(), ev, cookie); + Schedule(deadline - Timestamp(), ev, cookie); + } + + void TActorSystem::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const { + const auto current = Monotonic(); + if (deadline < current) + deadline = current; + + TTicketLock::TGuard guard(&ScheduleLock); + ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie); } - void TActorSystem::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const { - const auto current = Monotonic(); - if (deadline < current) - deadline = current; - - TTicketLock::TGuard guard(&ScheduleLock); - ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie); - } - void TActorSystem::Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) const { - const auto deadline = Monotonic() + delta; + const auto deadline = Monotonic() + delta; TTicketLock::TGuard guard(&ScheduleLock); ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie); @@ -211,7 +211,7 @@ namespace NActors { TVector<NSchedulerQueue::TReader*> scheduleReaders; scheduleReaders.push_back(&ScheduleQueue->Reader); CpuManager->PrepareStart(scheduleReaders, this); - Scheduler->Prepare(this, &CurrentTimestamp, &CurrentMonotonic); + Scheduler->Prepare(this, &CurrentTimestamp, &CurrentMonotonic); Scheduler->PrepareSchedules(&scheduleReaders.front(), (ui32)scheduleReaders.size()); // setup interconnect proxies @@ -242,9 +242,9 @@ namespace NActors { // ok, setup complete, we could destroy setup config SystemSetup.Destroy(); - Scheduler->PrepareStart(); + Scheduler->PrepareStart(); CpuManager->Start(); - Send(MakeSchedulerActorId(), new TEvSchedulerInitialize(scheduleReaders, &CurrentTimestamp, &CurrentMonotonic)); + Send(MakeSchedulerActorId(), new TEvSchedulerInitialize(scheduleReaders, &CurrentTimestamp, &CurrentMonotonic)); Scheduler->Start(); } diff --git a/library/cpp/actors/core/actorsystem.h b/library/cpp/actors/core/actorsystem.h index 80d33901b0..40499d7586 100644 --- a/library/cpp/actors/core/actorsystem.h +++ b/library/cpp/actors/core/actorsystem.h @@ -27,7 +27,7 @@ namespace NActors { char data[12]; memcpy(data, "ICProxy@", 8); memcpy(data + 8, &destNodeId, sizeof(ui32)); - return TActorId(0, TStringBuf(data, 12)); + return TActorId(0, TStringBuf(data, 12)); } inline bool IsInterconnectProxyId(const TActorId& actorId) { @@ -69,7 +69,7 @@ namespace NActors { /** * Schedule one-shot event that will be send at given time point in the future. * - * @param deadline the wallclock time point in future when event must be send + * @param deadline the wallclock time point in future when event must be send * @param ev the event to send * @param cookie cookie that will be piggybacked with event * @param workerId index of thread which will perform event dispatching @@ -77,16 +77,16 @@ namespace NActors { virtual void Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) = 0; /** - * Schedule one-shot event that will be send at given time point in the future. - * - * @param deadline the monotonic time point in future when event must be send - * @param ev the event to send - * @param cookie cookie that will be piggybacked with event + * Schedule one-shot event that will be send at given time point in the future. + * + * @param deadline the monotonic time point in future when event must be send + * @param ev the event to send + * @param cookie cookie that will be piggybacked with event * @param workerId index of thread which will perform event dispatching - */ + */ virtual void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) = 0; - - /** + + /** * Schedule one-shot event that will be send after given delay. * * @param delta the time from now to delay event sending @@ -136,9 +136,9 @@ namespace NActors { virtual ~ISchedulerThread() { } - virtual void Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) = 0; + virtual void Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) = 0; virtual void PrepareSchedules(NSchedulerQueue::TReader** readers, ui32 scheduleReadersCount) = 0; - virtual void PrepareStart() { /* empty */ } + virtual void PrepareStart() { /* empty */ } virtual void Start() = 0; virtual void PrepareStop() = 0; virtual void Stop() = 0; @@ -226,7 +226,7 @@ namespace NActors { TArrayHolder<TActorId> Interconnect; volatile ui64 CurrentTimestamp; - volatile ui64 CurrentMonotonic; + volatile ui64 CurrentMonotonic; volatile ui64 CurrentIDCounter; THolder<NSchedulerQueue::TQueueType> ScheduleQueue; @@ -264,22 +264,22 @@ namespace NActors { /** * Schedule one-shot event that will be send at given time point in the future. * - * @param deadline the wallclock time point in future when event must be send + * @param deadline the wallclock time point in future when event must be send * @param ev the event to send * @param cookie cookie that will be piggybacked with event */ void Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr) const; /** - * Schedule one-shot event that will be send at given time point in the future. - * - * @param deadline the monotonic time point in future when event must be send - * @param ev the event to send - * @param cookie cookie that will be piggybacked with event - */ - void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr) const; - - /** + * Schedule one-shot event that will be send at given time point in the future. + * + * @param deadline the monotonic time point in future when event must be send + * @param ev the event to send + * @param cookie cookie that will be piggybacked with event + */ + void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr) const; + + /** * Schedule one-shot event that will be send after given delay. * * @param delta the time from now to delay event sending @@ -340,10 +340,10 @@ namespace NActors { return TInstant::MicroSeconds(RelaxedLoad(&CurrentTimestamp)); } - TMonotonic Monotonic() const { - return TMonotonic::MicroSeconds(RelaxedLoad(&CurrentMonotonic)); - } - + TMonotonic Monotonic() const { + return TMonotonic::MicroSeconds(RelaxedLoad(&CurrentMonotonic)); + } + template <typename T> T* AppData() const { return (T*)AppData0; diff --git a/library/cpp/actors/core/events.h b/library/cpp/actors/core/events.h index 381dac03af..702cf50fad 100644 --- a/library/cpp/actors/core/events.h +++ b/library/cpp/actors/core/events.h @@ -4,7 +4,7 @@ #include "event_pb.h" #include <library/cpp/actors/protos/actors.pb.h> -#include <util/system/unaligned_mem.h> +#include <util/system/unaligned_mem.h> namespace NActors { struct TEvents { @@ -161,9 +161,9 @@ namespace NActors { private: static TString MakeData(ui32 sourceType, ui32 reason) { TString s = TString::Uninitialized(sizeof(ui32) + sizeof(ui32)); - char *p = s.Detach(); - WriteUnaligned<ui32>(p + 0, sourceType); - WriteUnaligned<ui32>(p + 4, reason); + char *p = s.Detach(); + WriteUnaligned<ui32>(p + 0, sourceType); + WriteUnaligned<ui32>(p + 4, reason); return s; } }; diff --git a/library/cpp/actors/core/events_undelivered.cpp b/library/cpp/actors/core/events_undelivered.cpp index 3b96625cab..23deaffd10 100644 --- a/library/cpp/actors/core/events_undelivered.cpp +++ b/library/cpp/actors/core/events_undelivered.cpp @@ -32,9 +32,9 @@ namespace NActors { IEventBase* TEvents::TEvUndelivered::Load(TEventSerializedData* bufs) { TString str = bufs->GetString(); Y_VERIFY(str.size() == (sizeof(ui32) + sizeof(ui32))); - const char* p = str.data(); - const ui64 sourceType = ReadUnaligned<ui32>(p + 0); - const ui64 reason = ReadUnaligned<ui32>(p + 4); + const char* p = str.data(); + const ui64 sourceType = ReadUnaligned<ui32>(p + 0); + const ui64 reason = ReadUnaligned<ui32>(p + 4); return new TEvUndelivered(sourceType, reason); } diff --git a/library/cpp/actors/core/executor_pool_basic.cpp b/library/cpp/actors/core/executor_pool_basic.cpp index 87315a6a6a..4dce16939a 100644 --- a/library/cpp/actors/core/executor_pool_basic.cpp +++ b/library/cpp/actors/core/executor_pool_basic.cpp @@ -87,17 +87,17 @@ namespace NActors { if (x < 0) { #if defined ACTORSLIB_COLLECT_EXEC_STATS if (AtomicGetAndIncrement(ThreadUtilization) == 0) { - // Initially counter contains -t0, the pool start timestamp - // When the first thread goes to sleep we add t1, so the counter - // becomes t1-t0 >= 0, or the duration of max utilization so far. - // If the counter was negative and becomes positive, that means - // counter just turned into a duration and we should store that - // duration. Otherwise another thread raced with us and - // subtracted some other timestamp t2. + // Initially counter contains -t0, the pool start timestamp + // When the first thread goes to sleep we add t1, so the counter + // becomes t1-t0 >= 0, or the duration of max utilization so far. + // If the counter was negative and becomes positive, that means + // counter just turned into a duration and we should store that + // duration. Otherwise another thread raced with us and + // subtracted some other timestamp t2. const i64 t = GetCycleCountFast(); - const i64 x = AtomicGetAndAdd(MaxUtilizationCounter, t); - if (x < 0 && x + t > 0) - AtomicStore(&MaxUtilizationAccumulator, x + t); + const i64 x = AtomicGetAndAdd(MaxUtilizationCounter, t); + if (x < 0 && x + t > 0) + AtomicStore(&MaxUtilizationAccumulator, x + t); } #endif @@ -126,7 +126,7 @@ namespace NActors { if (!doSpin) { break; } - if (RelaxedLoad(&StopFlag)) { + if (RelaxedLoad(&StopFlag)) { break; } } @@ -159,20 +159,20 @@ namespace NActors { #if defined ACTORSLIB_COLLECT_EXEC_STATS if (AtomicDecrement(ThreadUtilization) == 0) { - // When we started sleeping counter contained t1-t0, or the - // last duration of max utilization. Now we subtract t2 >= t1, - // which turns counter negative again, and the next sleep cycle - // at timestamp t3 would be adding some new duration t3-t2. - // If the counter was positive and becomes negative that means - // there are no current races with other threads and we should - // store the last positive duration we observed. Multiple - // threads may be adding and subtracting values in potentially - // arbitrary order, which would cause counter to oscillate - // around zero. When it crosses zero is a good indication of a - // correct value. + // When we started sleeping counter contained t1-t0, or the + // last duration of max utilization. Now we subtract t2 >= t1, + // which turns counter negative again, and the next sleep cycle + // at timestamp t3 would be adding some new duration t3-t2. + // If the counter was positive and becomes negative that means + // there are no current races with other threads and we should + // store the last positive duration we observed. Multiple + // threads may be adding and subtracting values in potentially + // arbitrary order, which would cause counter to oscillate + // around zero. When it crosses zero is a good indication of a + // correct value. const i64 t = GetCycleCountFast(); - const i64 x = AtomicGetAndAdd(MaxUtilizationCounter, -t); - if (x > 0 && x - t < 0) + const i64 x = AtomicGetAndAdd(MaxUtilizationCounter, -t); + if (x > 0 && x - t < 0) AtomicStore(&MaxUtilizationAccumulator, x); } #endif @@ -305,18 +305,18 @@ namespace NActors { void TBasicExecutorPool::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) { Y_VERIFY_DEBUG(workerId < PoolThreads); - - const auto current = ActorSystem->Monotonic(); - if (deadline < current) - deadline = current; - + + const auto current = ActorSystem->Monotonic(); + if (deadline < current) + deadline = current; + ScheduleWriters[workerId].Push(deadline.MicroSeconds(), ev.Release(), cookie); - } - + } + void TBasicExecutorPool::Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) { Y_VERIFY_DEBUG(workerId < PoolThreads); - const auto deadline = ActorSystem->Monotonic() + delta; + const auto deadline = ActorSystem->Monotonic() + delta; ScheduleWriters[workerId].Push(deadline.MicroSeconds(), ev.Release(), cookie); } diff --git a/library/cpp/actors/core/executor_pool_basic_ut.cpp b/library/cpp/actors/core/executor_pool_basic_ut.cpp index b3d6d8b329..76dff693af 100644 --- a/library/cpp/actors/core/executor_pool_basic_ut.cpp +++ b/library/cpp/actors/core/executor_pool_basic_ut.cpp @@ -47,8 +47,8 @@ public: if (GetCounter() == 0) { break; } - - Sleep(TDuration::MilliSeconds(1)); + + Sleep(TDuration::MilliSeconds(1)); } } @@ -69,8 +69,8 @@ private: { Y_UNUSED(ev); Action(); - TAtomicBase count = AtomicDecrement(Counter); - Y_VERIFY(count != Max<TAtomicBase>()); + TAtomicBase count = AtomicDecrement(Counter); + Y_VERIFY(count != Max<TAtomicBase>()); if (count) { Send(Receiver, new TEvMsg()); } @@ -206,19 +206,19 @@ Y_UNIT_TEST_SUITE(BasicExecutorPool) { actorSystem.Send(changerActorId, new TEvMsg()); while (true) { - size_t maxCounter = 0; + size_t maxCounter = 0; for (size_t i = 0; i < size; ++i) { - maxCounter = Max(maxCounter, actors[i]->GetCounter()); + maxCounter = Max(maxCounter, actors[i]->GetCounter()); + } + + if (maxCounter == 0) { + break; } - if (maxCounter == 0) { - break; - } - auto now = TInstant::Now(); - UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); + UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); - Sleep(TDuration::MilliSeconds(1)); + Sleep(TDuration::MilliSeconds(1)); } changerActor->Stop(); @@ -242,9 +242,9 @@ Y_UNIT_TEST_SUITE(BasicExecutorPool) { while (actor->GetCounter()) { auto now = TInstant::Now(); - UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Counter is " << actor->GetCounter()); - - Sleep(TDuration::MilliSeconds(1)); + UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Counter is " << actor->GetCounter()); + + Sleep(TDuration::MilliSeconds(1)); } } @@ -275,19 +275,19 @@ Y_UNIT_TEST_SUITE(BasicExecutorPool) { while (true) { - size_t maxCounter = 0; + size_t maxCounter = 0; for (size_t i = 0; i < size; ++i) { - maxCounter = Max(maxCounter, actors[i]->GetCounter()); + maxCounter = Max(maxCounter, actors[i]->GetCounter()); + } + + if (maxCounter == 0) { + break; } - if (maxCounter == 0) { - break; - } - auto now = TInstant::Now(); - UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); + UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); - Sleep(TDuration::MilliSeconds(1)); + Sleep(TDuration::MilliSeconds(1)); } } @@ -319,19 +319,19 @@ Y_UNIT_TEST_SUITE(BasicExecutorPool) { while (true) { - size_t maxCounter = 0; + size_t maxCounter = 0; for (size_t i = 0; i < actorsCount; ++i) { - maxCounter = Max(maxCounter, actors[i]->GetCounter()); + maxCounter = Max(maxCounter, actors[i]->GetCounter()); + } + + if (maxCounter == 0) { + break; } - if (maxCounter == 0) { - break; - } - auto now = TInstant::Now(); - UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); + UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); - Sleep(TDuration::MilliSeconds(1)); + Sleep(TDuration::MilliSeconds(1)); } } @@ -362,19 +362,19 @@ Y_UNIT_TEST_SUITE(BasicExecutorPool) { } while (true) { - size_t maxCounter = 0; + size_t maxCounter = 0; for (size_t i = 0; i < actorsCount; ++i) { - maxCounter = Max(maxCounter, actors[i]->GetCounter()); + maxCounter = Max(maxCounter, actors[i]->GetCounter()); + } + + if (maxCounter == 0) { + break; } - if (maxCounter == 0) { - break; - } - auto now = TInstant::Now(); - UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); + UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Max counter is " << maxCounter); - Sleep(TDuration::MilliSeconds(1)); + Sleep(TDuration::MilliSeconds(1)); } } diff --git a/library/cpp/actors/core/executor_pool_io.cpp b/library/cpp/actors/core/executor_pool_io.cpp index f4f13c9c20..fb557ae6b0 100644 --- a/library/cpp/actors/core/executor_pool_io.cpp +++ b/library/cpp/actors/core/executor_pool_io.cpp @@ -30,33 +30,33 @@ namespace NActors { ui32 workerId = wctx.WorkerId; Y_VERIFY_DEBUG(workerId < PoolThreads); - NHPTimer::STime elapsed = 0; - NHPTimer::STime parked = 0; + NHPTimer::STime elapsed = 0; + NHPTimer::STime parked = 0; NHPTimer::STime hpstart = GetCycleCountFast(); - NHPTimer::STime hpnow; - + NHPTimer::STime hpnow; + const TAtomic x = AtomicDecrement(Semaphore); if (x < 0) { TThreadCtx& threadCtx = Threads[workerId]; ThreadQueue.Push(workerId + 1, revolvingCounter); hpnow = GetCycleCountFast(); - elapsed += hpnow - hpstart; + elapsed += hpnow - hpstart; if (threadCtx.Pad.Park()) return 0; hpstart = GetCycleCountFast(); - parked += hpstart - hpnow; + parked += hpstart - hpnow; } while (!RelaxedLoad(&StopFlag)) { - if (const ui32 activation = Activations.Pop(++revolvingCounter)) { + if (const ui32 activation = Activations.Pop(++revolvingCounter)) { hpnow = GetCycleCountFast(); - elapsed += hpnow - hpstart; + elapsed += hpnow - hpstart; wctx.AddElapsedCycles(IActor::ACTOR_SYSTEM, elapsed); - if (parked > 0) { + if (parked > 0) { wctx.AddParkedCycles(parked); - } + } return activation; - } + } SpinLockPause(); } @@ -69,18 +69,18 @@ namespace NActors { void TIOExecutorPool::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) { Y_UNUSED(workerId); - - const auto current = ActorSystem->Monotonic(); - if (deadline < current) - deadline = current; - - TTicketLock::TGuard guard(&ScheduleLock); - ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie); - } - + + const auto current = ActorSystem->Monotonic(); + if (deadline < current) + deadline = current; + + TTicketLock::TGuard guard(&ScheduleLock); + ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie); + } + void TIOExecutorPool::Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) { Y_UNUSED(workerId); - const auto deadline = ActorSystem->Monotonic() + delta; + const auto deadline = ActorSystem->Monotonic() + delta; TTicketLock::TGuard guard(&ScheduleLock); ScheduleQueue->Writer.Push(deadline.MicroSeconds(), ev.Release(), cookie); diff --git a/library/cpp/actors/core/executor_pool_io.h b/library/cpp/actors/core/executor_pool_io.h index 7de6fd0528..e576d642a1 100644 --- a/library/cpp/actors/core/executor_pool_io.h +++ b/library/cpp/actors/core/executor_pool_io.h @@ -17,7 +17,7 @@ namespace NActors { }; TArrayHolder<TThreadCtx> Threads; - TUnorderedCache<ui32, 512, 4> ThreadQueue; + TUnorderedCache<ui32, 512, 4> ThreadQueue; THolder<NSchedulerQueue::TQueueType> ScheduleQueue; TTicketLock ScheduleLock; diff --git a/library/cpp/actors/core/executor_pool_united.cpp b/library/cpp/actors/core/executor_pool_united.cpp index 4ae9ec4c73..dac6245635 100644 --- a/library/cpp/actors/core/executor_pool_united.cpp +++ b/library/cpp/actors/core/executor_pool_united.cpp @@ -1235,7 +1235,7 @@ namespace NActors { inline void TUnitedWorkers::TryWake(TPoolId pool) { // Avoid using multiple atomic seq_cst loads in cycle, use barrier once AtomicBarrier(); - + // Scan every allowed cpu in pool's wakeup order and try to wake the first idle cpu if (RelaxedLoad(&Pools[pool].Waiters) > 0) { for (TCpu* cpu : Pools[pool].WakeOrderCpus) { @@ -1247,11 +1247,11 @@ namespace NActors { // Cpu has not been woken up } - + inline void TUnitedWorkers::BeginExecution(TPoolId pool, ui32& activation, ui64 revolvingCounter) { Pools[pool].BeginExecution(activation, revolvingCounter); - } - + } + inline bool TUnitedWorkers::NextExecution(TPoolId pool, ui32& activation, ui64 revolvingCounter) { return Pools[pool].NextExecution(activation, revolvingCounter); } diff --git a/library/cpp/actors/core/executor_pool_united_ut.cpp b/library/cpp/actors/core/executor_pool_united_ut.cpp index ddc4c348ee..d4df17f1b8 100644 --- a/library/cpp/actors/core/executor_pool_united_ut.cpp +++ b/library/cpp/actors/core/executor_pool_united_ut.cpp @@ -58,8 +58,8 @@ public: if (GetCounter() == 0) { break; } - - Sleep(TDuration::MilliSeconds(1)); + + Sleep(TDuration::MilliSeconds(1)); } } @@ -78,8 +78,8 @@ private: void Handle(TEvMsg::TPtr &ev) { Y_UNUSED(ev); Action(); - TAtomicBase count = AtomicDecrement(Counter); - Y_VERIFY(count != Max<TAtomicBase>()); + TAtomicBase count = AtomicDecrement(Counter); + Y_VERIFY(count != Max<TAtomicBase>()); if (count) { Send(Receiver, new TEvMsg()); } @@ -149,9 +149,9 @@ Y_UNIT_TEST_SUITE(UnitedExecutorPool) { while (actor->GetCounter()) { auto now = TInstant::Now(); - UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Counter is " << actor->GetCounter()); - - Sleep(TDuration::MilliSeconds(1)); + UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "Counter is " << actor->GetCounter()); + + Sleep(TDuration::MilliSeconds(1)); } TVector<TExecutorThreadStats> stats; @@ -212,11 +212,11 @@ Y_UNIT_TEST_SUITE(UnitedExecutorPool) { left += actor->GetCounter(); } if (left == 0) { - break; - } + break; + } auto now = TInstant::Now(); UNIT_ASSERT_C(now - begin < TDuration::Seconds(5), "left " << left); - Sleep(TDuration::MilliSeconds(1)); + Sleep(TDuration::MilliSeconds(1)); } for (size_t pool = 0; pool < pools; pool++) { @@ -311,11 +311,11 @@ Y_UNIT_TEST_SUITE(UnitedExecutorPool) { left += actor->GetCounter(); } if (left == 0) { - break; - } + break; + } auto now = TInstant::Now(); UNIT_ASSERT_C(now - begin < TDuration::Seconds(15), "left " << left); - Sleep(TDuration::MilliSeconds(1)); + Sleep(TDuration::MilliSeconds(1)); } for (size_t pool = 0; pool < pools; pool++) { diff --git a/library/cpp/actors/core/executor_thread.cpp b/library/cpp/actors/core/executor_thread.cpp index 7e961584e6..446b651efd 100644 --- a/library/cpp/actors/core/executor_thread.cpp +++ b/library/cpp/actors/core/executor_thread.cpp @@ -76,11 +76,11 @@ namespace NActors { Ctx.Executor->Schedule(deadline, ev, cookie, Ctx.WorkerId); } - void TExecutorThread::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) { - ++CurrentActorScheduledEventsCounter; + void TExecutorThread::Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) { + ++CurrentActorScheduledEventsCounter; Ctx.Executor->Schedule(deadline, ev, cookie, Ctx.WorkerId); - } - + } + void TExecutorThread::Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie) { ++CurrentActorScheduledEventsCounter; Ctx.Executor->Schedule(delta, ev, cookie, Ctx.WorkerId); @@ -145,7 +145,7 @@ namespace NActors { NHPTimer::STime hpnow; recipient = ev->GetRecipientRewrite(); if (actor = mailbox->FindActor(recipient.LocalId())) { - TActorContext ctx(*mailbox, *this, hpprev, recipient); + TActorContext ctx(*mailbox, *this, hpprev, recipient); TlsActivationContext = &ctx; #ifdef USE_ACTOR_CALLSTACK diff --git a/library/cpp/actors/core/executor_thread.h b/library/cpp/actors/core/executor_thread.h index 72b8f28c1d..9d3c573f0d 100644 --- a/library/cpp/actors/core/executor_thread.h +++ b/library/cpp/actors/core/executor_thread.h @@ -47,7 +47,7 @@ namespace NActors { const std::vector<THolder<IActor>>& GetUnregistered() const { return DyingActors; } void Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); - void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); + void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); void Schedule(TDuration delta, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie = nullptr); bool Send(TAutoPtr<IEventHandle> ev) { diff --git a/library/cpp/actors/core/log.cpp b/library/cpp/actors/core/log.cpp index 391844dcff..5f63b5af58 100644 --- a/library/cpp/actors/core/log.cpp +++ b/library/cpp/actors/core/log.cpp @@ -223,9 +223,9 @@ namespace NActors { } } - void TLoggerActor::Throttle(const NLog::TSettings& settings) { + void TLoggerActor::Throttle(const NLog::TSettings& settings) { if (AtomicGet(IsOverflow)) - Sleep(settings.ThrottleDelay); + Sleep(settings.ThrottleDelay); } void TLoggerActor::LogIgnoredCount(TInstant now) { diff --git a/library/cpp/actors/core/log.h b/library/cpp/actors/core/log.h index f063642ad2..c11a7cf3c1 100644 --- a/library/cpp/actors/core/log.h +++ b/library/cpp/actors/core/log.h @@ -232,7 +232,7 @@ namespace NActors { // Directly call logger instead of sending a message void Log(TInstant time, NLog::EPriority priority, NLog::EComponent component, const char* c, ...); - static void Throttle(const NLog::TSettings& settings); + static void Throttle(const NLog::TSettings& settings); private: TIntrusivePtr<NLog::TSettings> Settings; @@ -322,7 +322,7 @@ namespace NActors { inline void DeliverLogMessage(TCtx& ctx, NLog::EPriority mPriority, NLog::EComponent mComponent, TString &&str) { const NLog::TSettings *mSettings = ctx.LoggerSettings(); - TLoggerActor::Throttle(*mSettings); + TLoggerActor::Throttle(*mSettings); ctx.Send(new IEventHandle(mSettings->LoggerActorId, TActorId(), new NLog::TEvLog(mPriority, mComponent, std::move(str)))); } diff --git a/library/cpp/actors/core/log_settings.cpp b/library/cpp/actors/core/log_settings.cpp index d77688021d..f52f2fc5d2 100644 --- a/library/cpp/actors/core/log_settings.cpp +++ b/library/cpp/actors/core/log_settings.cpp @@ -12,7 +12,7 @@ namespace NActors { , LoggerComponent(loggerComponent) , TimeThresholdMs(timeThresholdMs) , AllowDrop(true) - , ThrottleDelay(TDuration::MilliSeconds(100)) + , ThrottleDelay(TDuration::MilliSeconds(100)) , MinVal(0) , MaxVal(0) , Mask(0) @@ -34,7 +34,7 @@ namespace NActors { , LoggerComponent(loggerComponent) , TimeThresholdMs(timeThresholdMs) , AllowDrop(true) - , ThrottleDelay(TDuration::MilliSeconds(100)) + , ThrottleDelay(TDuration::MilliSeconds(100)) , MinVal(0) , MaxVal(0) , Mask(0) @@ -205,10 +205,10 @@ namespace NActors { AllowDrop = val; } - void TSettings::SetThrottleDelay(TDuration value) { - ThrottleDelay = value; - } - + void TSettings::SetThrottleDelay(TDuration value) { + ThrottleDelay = value; + } + void TSettings::SetUseLocalTimestamps(bool value) { UseLocalTimestamps = value; } diff --git a/library/cpp/actors/core/log_settings.h b/library/cpp/actors/core/log_settings.h index 5f4898bda1..7fe4504edd 100644 --- a/library/cpp/actors/core/log_settings.h +++ b/library/cpp/actors/core/log_settings.h @@ -73,7 +73,7 @@ namespace NActors { EComponent LoggerComponent; ui64 TimeThresholdMs; bool AllowDrop; - TDuration ThrottleDelay; + TDuration ThrottleDelay; TArrayHolder<TAtomic> ComponentInfo; TVector<TString> ComponentNames; EComponent MinVal; @@ -162,7 +162,7 @@ namespace NActors { static bool IsValidPriority(EPriority priority); bool IsValidComponent(EComponent component); void SetAllowDrop(bool val); - void SetThrottleDelay(TDuration value); + void SetThrottleDelay(TDuration value); void SetUseLocalTimestamps(bool value); private: diff --git a/library/cpp/actors/core/mailbox.cpp b/library/cpp/actors/core/mailbox.cpp index b63577c7d6..d84b4f9e46 100644 --- a/library/cpp/actors/core/mailbox.cpp +++ b/library/cpp/actors/core/mailbox.cpp @@ -529,7 +529,7 @@ namespace NActors { Y_FAIL(); } - AtomicStore(Lines + lineIndex, header); + AtomicStore(Lines + lineIndex, header); ui32 ret = lineIndexMask | 1; diff --git a/library/cpp/actors/core/mailbox.h b/library/cpp/actors/core/mailbox.h index 8f4f901bd9..0bd9c4d314 100644 --- a/library/cpp/actors/core/mailbox.h +++ b/library/cpp/actors/core/mailbox.h @@ -277,7 +277,7 @@ namespace NActors { TAtomic LastAllocatedLine; TAtomic AllocatedMailboxCount; - typedef TUnorderedCache<ui32, 512, 4> TMailboxCache; + typedef TUnorderedCache<ui32, 512, 4> TMailboxCache; TMailboxCache MailboxCacheSimple; TAtomic CachedSimpleMailboxes; TMailboxCache MailboxCacheRevolving; diff --git a/library/cpp/actors/core/monotonic.cpp b/library/cpp/actors/core/monotonic.cpp index eefd8913cc..3465149dbe 100644 --- a/library/cpp/actors/core/monotonic.cpp +++ b/library/cpp/actors/core/monotonic.cpp @@ -1,23 +1,23 @@ -#include "monotonic.h" - -#include <chrono> - -namespace NActors { - - namespace { - // Unfortunately time_since_epoch() is sometimes negative on wine - // Remember initial time point at program start and use offsets from that - std::chrono::steady_clock::time_point MonotonicOffset = std::chrono::steady_clock::now(); - } - - ui64 GetMonotonicMicroSeconds() { - auto microseconds = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - MonotonicOffset).count(); - // Steady clock is supposed to never jump backwards, but it's better to be safe in case of buggy implementations - if (Y_UNLIKELY(microseconds < 0)) { - microseconds = 0; - } - // Add one so we never return zero - return microseconds + 1; - } - -} // namespace NActors +#include "monotonic.h" + +#include <chrono> + +namespace NActors { + + namespace { + // Unfortunately time_since_epoch() is sometimes negative on wine + // Remember initial time point at program start and use offsets from that + std::chrono::steady_clock::time_point MonotonicOffset = std::chrono::steady_clock::now(); + } + + ui64 GetMonotonicMicroSeconds() { + auto microseconds = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - MonotonicOffset).count(); + // Steady clock is supposed to never jump backwards, but it's better to be safe in case of buggy implementations + if (Y_UNLIKELY(microseconds < 0)) { + microseconds = 0; + } + // Add one so we never return zero + return microseconds + 1; + } + +} // namespace NActors diff --git a/library/cpp/actors/core/monotonic.h b/library/cpp/actors/core/monotonic.h index cc0136b558..6fceb91dbe 100644 --- a/library/cpp/actors/core/monotonic.h +++ b/library/cpp/actors/core/monotonic.h @@ -1,111 +1,111 @@ -#pragma once - -#include <util/datetime/base.h> - -namespace NActors { - - /** - * Returns current monotonic time in microseconds - */ - ui64 GetMonotonicMicroSeconds(); - - /** - * Similar to TInstant, but measuring monotonic time - */ - class TMonotonic : public TTimeBase<TMonotonic> { - using TBase = TTimeBase<TMonotonic>; - - private: - constexpr explicit TMonotonic(TValue value) noexcept - : TBase(value) - { } - - public: - constexpr TMonotonic() noexcept { - } - - static constexpr TMonotonic FromValue(TValue value) noexcept { - return TMonotonic(value); - } - - static inline TMonotonic Now() { - return TMonotonic::MicroSeconds(GetMonotonicMicroSeconds()); - } - - using TBase::Days; - using TBase::Hours; - using TBase::MicroSeconds; - using TBase::MilliSeconds; - using TBase::Minutes; - using TBase::Seconds; - - static constexpr TMonotonic Max() noexcept { - return TMonotonic(::Max<ui64>()); - } - - static constexpr TMonotonic Zero() noexcept { - return TMonotonic(); - } - - static constexpr TMonotonic MicroSeconds(ui64 us) noexcept { - return TMonotonic(TInstant::MicroSeconds(us).GetValue()); - } - - static constexpr TMonotonic MilliSeconds(ui64 ms) noexcept { - return TMonotonic(TInstant::MilliSeconds(ms).GetValue()); - } - - static constexpr TMonotonic Seconds(ui64 s) noexcept { - return TMonotonic(TInstant::Seconds(s).GetValue()); - } - - static constexpr TMonotonic Minutes(ui64 m) noexcept { - return TMonotonic(TInstant::Minutes(m).GetValue()); - } - - static constexpr TMonotonic Hours(ui64 h) noexcept { - return TMonotonic(TInstant::Hours(h).GetValue()); - } - - static constexpr TMonotonic Days(ui64 d) noexcept { - return TMonotonic(TInstant::Days(d).GetValue()); - } - - template<class T> - inline TMonotonic& operator+=(const T& t) noexcept { - return (*this = (*this + t)); - } - - template<class T> - inline TMonotonic& operator-=(const T& t) noexcept { - return (*this = (*this - t)); - } - }; -} // namespace NActors - -Y_DECLARE_PODTYPE(NActors::TMonotonic); - -template<> -struct THash<NActors::TMonotonic> { - size_t operator()(const NActors::TMonotonic& key) const { - return THash<NActors::TMonotonic::TValue>()(key.GetValue()); - } -}; - -namespace NActors { - - constexpr TDuration operator-(const TMonotonic& l, const TMonotonic& r) { - return TInstant::FromValue(l.GetValue()) - TInstant::FromValue(r.GetValue()); - } - - constexpr TMonotonic operator+(const TMonotonic& l, const TDuration& r) { - TInstant result = TInstant::FromValue(l.GetValue()) + r; - return TMonotonic::FromValue(result.GetValue()); - } - - constexpr TMonotonic operator-(const TMonotonic& l, const TDuration& r) { - TInstant result = TInstant::FromValue(l.GetValue()) - r; - return TMonotonic::FromValue(result.GetValue()); - } - -} // namespace NActors +#pragma once + +#include <util/datetime/base.h> + +namespace NActors { + + /** + * Returns current monotonic time in microseconds + */ + ui64 GetMonotonicMicroSeconds(); + + /** + * Similar to TInstant, but measuring monotonic time + */ + class TMonotonic : public TTimeBase<TMonotonic> { + using TBase = TTimeBase<TMonotonic>; + + private: + constexpr explicit TMonotonic(TValue value) noexcept + : TBase(value) + { } + + public: + constexpr TMonotonic() noexcept { + } + + static constexpr TMonotonic FromValue(TValue value) noexcept { + return TMonotonic(value); + } + + static inline TMonotonic Now() { + return TMonotonic::MicroSeconds(GetMonotonicMicroSeconds()); + } + + using TBase::Days; + using TBase::Hours; + using TBase::MicroSeconds; + using TBase::MilliSeconds; + using TBase::Minutes; + using TBase::Seconds; + + static constexpr TMonotonic Max() noexcept { + return TMonotonic(::Max<ui64>()); + } + + static constexpr TMonotonic Zero() noexcept { + return TMonotonic(); + } + + static constexpr TMonotonic MicroSeconds(ui64 us) noexcept { + return TMonotonic(TInstant::MicroSeconds(us).GetValue()); + } + + static constexpr TMonotonic MilliSeconds(ui64 ms) noexcept { + return TMonotonic(TInstant::MilliSeconds(ms).GetValue()); + } + + static constexpr TMonotonic Seconds(ui64 s) noexcept { + return TMonotonic(TInstant::Seconds(s).GetValue()); + } + + static constexpr TMonotonic Minutes(ui64 m) noexcept { + return TMonotonic(TInstant::Minutes(m).GetValue()); + } + + static constexpr TMonotonic Hours(ui64 h) noexcept { + return TMonotonic(TInstant::Hours(h).GetValue()); + } + + static constexpr TMonotonic Days(ui64 d) noexcept { + return TMonotonic(TInstant::Days(d).GetValue()); + } + + template<class T> + inline TMonotonic& operator+=(const T& t) noexcept { + return (*this = (*this + t)); + } + + template<class T> + inline TMonotonic& operator-=(const T& t) noexcept { + return (*this = (*this - t)); + } + }; +} // namespace NActors + +Y_DECLARE_PODTYPE(NActors::TMonotonic); + +template<> +struct THash<NActors::TMonotonic> { + size_t operator()(const NActors::TMonotonic& key) const { + return THash<NActors::TMonotonic::TValue>()(key.GetValue()); + } +}; + +namespace NActors { + + constexpr TDuration operator-(const TMonotonic& l, const TMonotonic& r) { + return TInstant::FromValue(l.GetValue()) - TInstant::FromValue(r.GetValue()); + } + + constexpr TMonotonic operator+(const TMonotonic& l, const TDuration& r) { + TInstant result = TInstant::FromValue(l.GetValue()) + r; + return TMonotonic::FromValue(result.GetValue()); + } + + constexpr TMonotonic operator-(const TMonotonic& l, const TDuration& r) { + TInstant result = TInstant::FromValue(l.GetValue()) - r; + return TMonotonic::FromValue(result.GetValue()); + } + +} // namespace NActors diff --git a/library/cpp/actors/core/scheduler_actor.cpp b/library/cpp/actors/core/scheduler_actor.cpp index c189653e99..febc5e40dd 100644 --- a/library/cpp/actors/core/scheduler_actor.cpp +++ b/library/cpp/actors/core/scheduler_actor.cpp @@ -43,7 +43,7 @@ namespace NActors { TPollerToken::TPtr PollerToken; ui64 RealTime; - ui64 MonotonicTime; + ui64 MonotonicTime; ui64 ActiveTick; typedef TMap<ui64, TAutoPtr<NSchedulerQueue::TQueueType>> TMomentMap; // intrasecond queues @@ -56,7 +56,7 @@ namespace NActors { static const ui64 IntrasecondThreshold = 1048576; // ~second TAutoPtr<TMomentMap> ActiveSec; volatile ui64* CurrentTimestamp = nullptr; - volatile ui64* CurrentMonotonic = nullptr; + volatile ui64* CurrentMonotonic = nullptr; TDeque<TAutoPtr<IEventHandle>> EventsToBeSent; public: @@ -84,9 +84,9 @@ namespace NActors { Y_ASSERT(evInitialize.CurrentTimestamp != nullptr); CurrentTimestamp = evInitialize.CurrentTimestamp; - Y_ASSERT(evInitialize.CurrentMonotonic != nullptr); - CurrentMonotonic = evInitialize.CurrentMonotonic; - + Y_ASSERT(evInitialize.CurrentMonotonic != nullptr); + CurrentMonotonic = evInitialize.CurrentMonotonic; + struct itimerspec new_time; memset(&new_time, 0, sizeof(new_time)); new_time.it_value.tv_nsec = Cfg.ResolutionMicroseconds * 1000; @@ -96,10 +96,10 @@ namespace NActors { const bool success = ctx.Send(PollerActor, new TEvPollerRegister(TimerDescriptor, SelfId(), {})); Y_VERIFY(success); - RealTime = RelaxedLoad(CurrentTimestamp); - MonotonicTime = RelaxedLoad(CurrentMonotonic); + RealTime = RelaxedLoad(CurrentTimestamp); + MonotonicTime = RelaxedLoad(CurrentMonotonic); - ActiveTick = AlignUp<ui64>(MonotonicTime, IntrasecondThreshold); + ActiveTick = AlignUp<ui64>(MonotonicTime, IntrasecondThreshold); } void Handle(TEvPollerRegisterResult::TPtr ev, const TActorContext& ctx) { @@ -108,10 +108,10 @@ namespace NActors { } void UpdateTime() { - RealTime = TInstant::Now().MicroSeconds(); - MonotonicTime = Max(MonotonicTime, GetMonotonicMicroSeconds()); - AtomicStore(CurrentTimestamp, RealTime); - AtomicStore(CurrentMonotonic, MonotonicTime); + RealTime = TInstant::Now().MicroSeconds(); + MonotonicTime = Max(MonotonicTime, GetMonotonicMicroSeconds()); + AtomicStore(CurrentTimestamp, RealTime); + AtomicStore(CurrentMonotonic, MonotonicTime); } void TryUpdateTime(NHPTimer::STime* lastTimeUpdate) { diff --git a/library/cpp/actors/core/scheduler_actor.h b/library/cpp/actors/core/scheduler_actor.h index 4209db0ab6..c2c561b43d 100644 --- a/library/cpp/actors/core/scheduler_actor.h +++ b/library/cpp/actors/core/scheduler_actor.h @@ -9,12 +9,12 @@ namespace NActors { struct TEvSchedulerInitialize : TEventLocal<TEvSchedulerInitialize, TEvents::TSystem::Bootstrap> { TVector<NSchedulerQueue::TReader*> ScheduleReaders; volatile ui64* CurrentTimestamp; - volatile ui64* CurrentMonotonic; + volatile ui64* CurrentMonotonic; - TEvSchedulerInitialize(const TVector<NSchedulerQueue::TReader*>& scheduleReaders, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) + TEvSchedulerInitialize(const TVector<NSchedulerQueue::TReader*>& scheduleReaders, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) : ScheduleReaders(scheduleReaders) , CurrentTimestamp(currentTimestamp) - , CurrentMonotonic(currentMonotonic) + , CurrentMonotonic(currentMonotonic) { } }; diff --git a/library/cpp/actors/core/scheduler_basic.cpp b/library/cpp/actors/core/scheduler_basic.cpp index b0c80eb6d2..fba200e16b 100644 --- a/library/cpp/actors/core/scheduler_basic.cpp +++ b/library/cpp/actors/core/scheduler_basic.cpp @@ -9,35 +9,35 @@ #endif namespace NActors { - - struct TBasicSchedulerThread::TMonCounters { - NMonitoring::TDynamicCounters::TCounterPtr TimeDelayMs; - NMonitoring::TDynamicCounters::TCounterPtr QueueSize; - NMonitoring::TDynamicCounters::TCounterPtr EventsSent; - NMonitoring::TDynamicCounters::TCounterPtr EventsDropped; - NMonitoring::TDynamicCounters::TCounterPtr EventsAdded; - NMonitoring::TDynamicCounters::TCounterPtr Iterations; - NMonitoring::TDynamicCounters::TCounterPtr Sleeps; - NMonitoring::TDynamicCounters::TCounterPtr ElapsedMicrosec; - - TMonCounters(const NMonitoring::TDynamicCounterPtr& counters) - : TimeDelayMs(counters->GetCounter("Scheduler/TimeDelayMs", false)) - , QueueSize(counters->GetCounter("Scheduler/QueueSize", false)) - , EventsSent(counters->GetCounter("Scheduler/EventsSent", true)) - , EventsDropped(counters->GetCounter("Scheduler/EventsDropped", true)) - , EventsAdded(counters->GetCounter("Scheduler/EventsAdded", true)) - , Iterations(counters->GetCounter("Scheduler/Iterations", true)) - , Sleeps(counters->GetCounter("Scheduler/Sleeps", true)) - , ElapsedMicrosec(counters->GetCounter("Scheduler/ElapsedMicrosec", true)) - { } - }; - + + struct TBasicSchedulerThread::TMonCounters { + NMonitoring::TDynamicCounters::TCounterPtr TimeDelayMs; + NMonitoring::TDynamicCounters::TCounterPtr QueueSize; + NMonitoring::TDynamicCounters::TCounterPtr EventsSent; + NMonitoring::TDynamicCounters::TCounterPtr EventsDropped; + NMonitoring::TDynamicCounters::TCounterPtr EventsAdded; + NMonitoring::TDynamicCounters::TCounterPtr Iterations; + NMonitoring::TDynamicCounters::TCounterPtr Sleeps; + NMonitoring::TDynamicCounters::TCounterPtr ElapsedMicrosec; + + TMonCounters(const NMonitoring::TDynamicCounterPtr& counters) + : TimeDelayMs(counters->GetCounter("Scheduler/TimeDelayMs", false)) + , QueueSize(counters->GetCounter("Scheduler/QueueSize", false)) + , EventsSent(counters->GetCounter("Scheduler/EventsSent", true)) + , EventsDropped(counters->GetCounter("Scheduler/EventsDropped", true)) + , EventsAdded(counters->GetCounter("Scheduler/EventsAdded", true)) + , Iterations(counters->GetCounter("Scheduler/Iterations", true)) + , Sleeps(counters->GetCounter("Scheduler/Sleeps", true)) + , ElapsedMicrosec(counters->GetCounter("Scheduler/ElapsedMicrosec", true)) + { } + }; + TBasicSchedulerThread::TBasicSchedulerThread(const TSchedulerConfig& config) : Config(config) - , MonCounters(Config.MonCounters ? new TMonCounters(Config.MonCounters) : nullptr) + , MonCounters(Config.MonCounters ? new TMonCounters(Config.MonCounters) : nullptr) , ActorSystem(nullptr) , CurrentTimestamp(nullptr) - , CurrentMonotonic(nullptr) + , CurrentMonotonic(nullptr) , TotalReaders(0) , StopFlag(false) , ScheduleMap(3600) @@ -55,45 +55,45 @@ namespace NActors { #endif ::SetCurrentThreadName("Scheduler"); - ui64 currentMonotonic = RelaxedLoad(CurrentMonotonic); - ui64 throttledMonotonic = currentMonotonic; + ui64 currentMonotonic = RelaxedLoad(CurrentMonotonic); + ui64 throttledMonotonic = currentMonotonic; - ui64 activeTick = AlignUp<ui64>(throttledMonotonic, IntrasecondThreshold); + ui64 activeTick = AlignUp<ui64>(throttledMonotonic, IntrasecondThreshold); TAutoPtr<TMomentMap> activeSec; NHPTimer::STime hpprev = GetCycleCountFast(); - ui64 nextTimestamp = TInstant::Now().MicroSeconds(); - ui64 nextMonotonic = Max(currentMonotonic, GetMonotonicMicroSeconds()); - + ui64 nextTimestamp = TInstant::Now().MicroSeconds(); + ui64 nextMonotonic = Max(currentMonotonic, GetMonotonicMicroSeconds()); + while (!AtomicLoad(&StopFlag)) { { - const ui64 delta = nextMonotonic - throttledMonotonic; - const ui64 elapsedDelta = nextMonotonic - currentMonotonic; - const ui64 threshold = Max(Min(Config.ProgressThreshold, 2 * elapsedDelta), ui64(1)); - - throttledMonotonic = (delta > threshold) ? throttledMonotonic + threshold : nextMonotonic; - - if (MonCounters) { - *MonCounters->TimeDelayMs = (nextMonotonic - throttledMonotonic) / 1000; - } + const ui64 delta = nextMonotonic - throttledMonotonic; + const ui64 elapsedDelta = nextMonotonic - currentMonotonic; + const ui64 threshold = Max(Min(Config.ProgressThreshold, 2 * elapsedDelta), ui64(1)); + + throttledMonotonic = (delta > threshold) ? throttledMonotonic + threshold : nextMonotonic; + + if (MonCounters) { + *MonCounters->TimeDelayMs = (nextMonotonic - throttledMonotonic) / 1000; + } + } + AtomicStore(CurrentTimestamp, nextTimestamp); + AtomicStore(CurrentMonotonic, nextMonotonic); + currentMonotonic = nextMonotonic; + + if (MonCounters) { + ++*MonCounters->Iterations; } - AtomicStore(CurrentTimestamp, nextTimestamp); - AtomicStore(CurrentMonotonic, nextMonotonic); - currentMonotonic = nextMonotonic; - - if (MonCounters) { - ++*MonCounters->Iterations; - } - + bool somethingDone = false; // first step - send everything triggered on schedule - ui64 eventsSent = 0; - ui64 eventsDropped = 0; + ui64 eventsSent = 0; + ui64 eventsDropped = 0; for (;;) { while (!!activeSec && !activeSec->empty()) { TMomentMap::iterator it = activeSec->begin(); - if (it->first <= throttledMonotonic) { + if (it->first <= throttledMonotonic) { if (NSchedulerQueue::TQueueType* q = it->second.Get()) { while (NSchedulerQueue::TEntry* x = q->Reader.Pop()) { somethingDone = true; @@ -102,16 +102,16 @@ namespace NActors { ISchedulerCookie* cookie = x->Cookie; // TODO: lazy send with backoff queue to not hang over contended mailboxes if (cookie) { - if (cookie->Detach()) { + if (cookie->Detach()) { ActorSystem->Send(ev); - ++eventsSent; - } else { + ++eventsSent; + } else { delete ev; - ++eventsDropped; - } + ++eventsDropped; + } } else { ActorSystem->Send(ev); - ++eventsSent; + ++eventsSent; } } } @@ -120,7 +120,7 @@ namespace NActors { break; } - if (activeTick <= throttledMonotonic) { + if (activeTick <= throttledMonotonic) { Y_VERIFY_DEBUG(!activeSec || activeSec->empty()); activeSec.Destroy(); activeTick += IntrasecondThreshold; @@ -138,7 +138,7 @@ namespace NActors { // second step - collect everything from queues - ui64 eventsAdded = 0; + ui64 eventsAdded = 0; for (ui32 i = 0; i != TotalReaders; ++i) { while (NSchedulerQueue::TEntry* x = Readers[i]->Pop()) { somethingDone = true; @@ -165,57 +165,57 @@ namespace NActors { queue.Reset(new NSchedulerQueue::TQueueType()); queue->Writer.Push(instant, ev, cookie); } - - ++eventsAdded; + + ++eventsAdded; } } NHPTimer::STime hpnow = GetCycleCountFast(); - - if (MonCounters) { - *MonCounters->QueueSize -= eventsSent + eventsDropped; - *MonCounters->QueueSize += eventsAdded; - *MonCounters->EventsSent += eventsSent; - *MonCounters->EventsDropped += eventsDropped; - *MonCounters->EventsAdded += eventsAdded; - *MonCounters->ElapsedMicrosec += NHPTimer::GetSeconds(hpnow - hpprev) * 1000000; - } - - hpprev = hpnow; - nextTimestamp = TInstant::Now().MicroSeconds(); - nextMonotonic = Max(currentMonotonic, GetMonotonicMicroSeconds()); - + + if (MonCounters) { + *MonCounters->QueueSize -= eventsSent + eventsDropped; + *MonCounters->QueueSize += eventsAdded; + *MonCounters->EventsSent += eventsSent; + *MonCounters->EventsDropped += eventsDropped; + *MonCounters->EventsAdded += eventsAdded; + *MonCounters->ElapsedMicrosec += NHPTimer::GetSeconds(hpnow - hpprev) * 1000000; + } + + hpprev = hpnow; + nextTimestamp = TInstant::Now().MicroSeconds(); + nextMonotonic = Max(currentMonotonic, GetMonotonicMicroSeconds()); + // ok complete, if nothing left - sleep if (!somethingDone) { - const ui64 nextInstant = AlignDown<ui64>(throttledMonotonic + Config.ResolutionMicroseconds, Config.ResolutionMicroseconds); - if (nextMonotonic >= nextInstant) // already in next time-slice + const ui64 nextInstant = AlignDown<ui64>(throttledMonotonic + Config.ResolutionMicroseconds, Config.ResolutionMicroseconds); + if (nextMonotonic >= nextInstant) // already in next time-slice continue; - const ui64 delta = nextInstant - nextMonotonic; + const ui64 delta = nextInstant - nextMonotonic; if (delta < Config.SpinThreshold) // not so much time left, just spin continue; - if (MonCounters) { - ++*MonCounters->Sleeps; - } - + if (MonCounters) { + ++*MonCounters->Sleeps; + } + NanoSleep(delta * 1000); // ok, looks like we should sleep a bit. - - // Don't count sleep in elapsed microseconds + + // Don't count sleep in elapsed microseconds hpprev = GetCycleCountFast(); - nextTimestamp = TInstant::Now().MicroSeconds(); - nextMonotonic = Max(currentMonotonic, GetMonotonicMicroSeconds()); + nextTimestamp = TInstant::Now().MicroSeconds(); + nextMonotonic = Max(currentMonotonic, GetMonotonicMicroSeconds()); } } // ok, die! } - void TBasicSchedulerThread::Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) { + void TBasicSchedulerThread::Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) { ActorSystem = actorSystem; CurrentTimestamp = currentTimestamp; - CurrentMonotonic = currentMonotonic; - *CurrentTimestamp = TInstant::Now().MicroSeconds(); - *CurrentMonotonic = GetMonotonicMicroSeconds(); + CurrentMonotonic = currentMonotonic; + *CurrentTimestamp = TInstant::Now().MicroSeconds(); + *CurrentMonotonic = GetMonotonicMicroSeconds(); } void TBasicSchedulerThread::PrepareSchedules(NSchedulerQueue::TReader** readers, ui32 scheduleReadersCount) { @@ -225,16 +225,16 @@ namespace NActors { Copy(readers, readers + scheduleReadersCount, Readers.Get()); } - void TBasicSchedulerThread::PrepareStart() { - // Called after actor system is initialized, but before executor threads - // are started, giving us a chance to update current timestamp with a - // more recent value, taking initialization time into account. This is - // safe to do, since scheduler thread is not started yet, so no other - // threads are updating time concurrently. - AtomicStore(CurrentTimestamp, TInstant::Now().MicroSeconds()); - AtomicStore(CurrentMonotonic, Max(RelaxedLoad(CurrentMonotonic), GetMonotonicMicroSeconds())); - } - + void TBasicSchedulerThread::PrepareStart() { + // Called after actor system is initialized, but before executor threads + // are started, giving us a chance to update current timestamp with a + // more recent value, taking initialization time into account. This is + // safe to do, since scheduler thread is not started yet, so no other + // threads are updating time concurrently. + AtomicStore(CurrentTimestamp, TInstant::Now().MicroSeconds()); + AtomicStore(CurrentMonotonic, Max(RelaxedLoad(CurrentMonotonic), GetMonotonicMicroSeconds())); + } + void TBasicSchedulerThread::Start() { MainCycle.Reset(new NThreading::TLegacyFuture<void, false>(std::bind(&TBasicSchedulerThread::CycleFunc, this))); } diff --git a/library/cpp/actors/core/scheduler_basic.h b/library/cpp/actors/core/scheduler_basic.h index 89ef8323f5..2ccde39235 100644 --- a/library/cpp/actors/core/scheduler_basic.h +++ b/library/cpp/actors/core/scheduler_basic.h @@ -1,7 +1,7 @@ #pragma once #include "actorsystem.h" -#include "monotonic.h" +#include "monotonic.h" #include "scheduler_queue.h" #include <library/cpp/actors/util/queue_chunk.h> #include <library/cpp/threading/future/legacy_future.h> @@ -9,17 +9,17 @@ #include <util/generic/map.h> namespace NActors { - + class TBasicSchedulerThread: public ISchedulerThread { // TODO: replace with NUMA-local threads and per-thread schedules const TSchedulerConfig Config; - struct TMonCounters; - const THolder<TMonCounters> MonCounters; - + struct TMonCounters; + const THolder<TMonCounters> MonCounters; + TActorSystem* ActorSystem; volatile ui64* CurrentTimestamp; - volatile ui64* CurrentMonotonic; + volatile ui64* CurrentMonotonic; ui32 TotalReaders; TArrayHolder<NSchedulerQueue::TReader*> Readers; @@ -44,7 +44,7 @@ namespace NActors { void Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) override; void PrepareSchedules(NSchedulerQueue::TReader** readers, ui32 scheduleReadersCount) override; - void PrepareStart() override; + void PrepareStart() override; void Start() override; void PrepareStop() override; void Stop() override; @@ -55,10 +55,10 @@ namespace NActors { virtual ~TMockSchedulerThread() override { } - void Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) override { + void Prepare(TActorSystem* actorSystem, volatile ui64* currentTimestamp, volatile ui64* currentMonotonic) override { Y_UNUSED(actorSystem); - *currentTimestamp = TInstant::Now().MicroSeconds(); - *currentMonotonic = GetMonotonicMicroSeconds(); + *currentTimestamp = TInstant::Now().MicroSeconds(); + *currentMonotonic = GetMonotonicMicroSeconds(); } void PrepareSchedules(NSchedulerQueue::TReader** readers, ui32 scheduleReadersCount) override { diff --git a/library/cpp/actors/core/scheduler_queue.h b/library/cpp/actors/core/scheduler_queue.h index 51b597be1f..3b8fac28f0 100644 --- a/library/cpp/actors/core/scheduler_queue.h +++ b/library/cpp/actors/core/scheduler_queue.h @@ -76,10 +76,10 @@ namespace NActors { } void Push(ui64 instantMicrosends, IEventHandle* ev, ISchedulerCookie* cookie) { - if (Y_UNLIKELY(instantMicrosends == 0)) { - // Protect against Pop() getting stuck forever - instantMicrosends = 1; - } + if (Y_UNLIKELY(instantMicrosends == 0)) { + // Protect against Pop() getting stuck forever + instantMicrosends = 1; + } if (WritePosition != TChunk::EntriesCount) { volatile TEntry& entry = WriteTo->Entries[WritePosition]; entry.Cookie = cookie; diff --git a/library/cpp/actors/core/ya.make b/library/cpp/actors/core/ya.make index 25fc0ce902..880a9d00db 100644 --- a/library/cpp/actors/core/ya.make +++ b/library/cpp/actors/core/ya.make @@ -81,7 +81,7 @@ SRCS( memory_tracker.h mon.h mon_stats.h - monotonic.cpp + monotonic.cpp monotonic.h worker_context.cpp worker_context.h diff --git a/library/cpp/actors/dnsresolver/dnsresolver.cpp b/library/cpp/actors/dnsresolver/dnsresolver.cpp index 791360bd82..6329bb0083 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver.cpp +++ b/library/cpp/actors/dnsresolver/dnsresolver.cpp @@ -1,475 +1,475 @@ -#include "dnsresolver.h" - -#include <library/cpp/actors/core/hfunc.h> -#include <library/cpp/threading/queue/mpsc_htswap.h> -#include <util/network/pair.h> -#include <util/network/socket.h> -#include <util/string/builder.h> -#include <util/system/thread.h> - -#include <ares.h> - -#include <atomic> - -namespace NActors { -namespace NDnsResolver { - - class TAresLibraryInitBase { - protected: - TAresLibraryInitBase() noexcept { - int status = ares_library_init(ARES_LIB_INIT_ALL); - Y_VERIFY(status == ARES_SUCCESS, "Unexpected failure to initialize c-ares library"); - } - - ~TAresLibraryInitBase() noexcept { - ares_library_cleanup(); - } - }; - - class TCallbackQueueBase { - protected: - TCallbackQueueBase() noexcept { - int err = SocketPair(Sockets, false, true); - Y_VERIFY(err == 0, "Unexpected failure to create a socket pair"); - SetNonBlock(Sockets[0]); - SetNonBlock(Sockets[1]); - } - - ~TCallbackQueueBase() noexcept { - closesocket(Sockets[0]); - closesocket(Sockets[1]); - } - - protected: - using TCallback = std::function<void()>; - using TCallbackQueue = NThreading::THTSwapQueue<TCallback>; - - void PushCallback(TCallback callback) { - Y_VERIFY(callback, "Cannot push an empty callback"); - CallbackQueue.Push(std::move(callback)); // this is a lockfree queue - - // Wake up worker thread on the first activation - if (Activations.fetch_add(1, std::memory_order_acq_rel) == 0) { - char ch = 'x'; - ssize_t ret; -#ifdef _win_ - ret = send(SignalSock(), &ch, 1, 0); - if (ret == -1) { - Y_VERIFY(WSAGetLastError() == WSAEWOULDBLOCK, "Unexpected send error"); - return; - } -#else - do { - ret = send(SignalSock(), &ch, 1, 0); - } while (ret == -1 && errno == EINTR); - if (ret == -1) { - Y_VERIFY(errno == EAGAIN || errno == EWOULDBLOCK, "Unexpected send error"); - return; - } -#endif - Y_VERIFY(ret == 1, "Unexpected send result"); - } - } - - void RunCallbacks() noexcept { - char ch[32]; - ssize_t ret; - bool signalled = false; - for (;;) { - ret = recv(WaitSock(), ch, sizeof(ch), 0); - if (ret > 0) { - signalled = true; - } - if (ret == sizeof(ch)) { - continue; - } - if (ret != -1) { - break; - } -#ifdef _win_ - if (WSAGetLastError() == WSAEWOULDBLOCK) { - break; - } - Y_FAIL("Unexpected recv error"); -#else - if (errno == EAGAIN || errno == EWOULDBLOCK) { - break; - } - Y_VERIFY(errno == EINTR, "Unexpected recv error"); -#endif - } - - if (signalled) { - // There's exactly one write to SignalSock while Activations != 0 - // It's impossible to get signalled while Activations == 0 - // We must set Activations = 0 to receive new signals - size_t count = Activations.exchange(0, std::memory_order_acq_rel); - Y_VERIFY(count != 0); - - // N.B. due to the way HTSwap works we may not be able to pop - // all callbacks on this activation, however we expect a new - // delayed activation to happen at a later time. - while (auto callback = CallbackQueue.Pop()) { - callback(); - } - } - } - - SOCKET SignalSock() { - return Sockets[0]; - } - - SOCKET WaitSock() { - return Sockets[1]; - } - - private: - SOCKET Sockets[2]; - TCallbackQueue CallbackQueue; - std::atomic<size_t> Activations{ 0 }; - }; - - class TSimpleDnsResolver - : public TActor<TSimpleDnsResolver> - , private TAresLibraryInitBase - , private TCallbackQueueBase - { - public: - TSimpleDnsResolver(TSimpleDnsResolverOptions options) noexcept - : TActor(&TThis::StateWork) - , Options(std::move(options)) - , WorkerThread(&TThis::WorkerThreadStart, this) - { - InitAres(); - - WorkerThread.Start(); - } - - ~TSimpleDnsResolver() noexcept override { - if (!Stopped) { - PushCallback([this] { - // Mark as stopped first - Stopped = true; - - // Cancel all current ares requests (will not send replies) - ares_cancel(AresChannel); - }); - - WorkerThread.Join(); - } - - StopAres(); - } - - static constexpr EActivityType ActorActivityType() { - return DNS_RESOLVER; - } - - private: - void InitAres() noexcept { - struct ares_options options; - memset(&options, 0, sizeof(options)); - int optmask = 0; - - options.flags = ARES_FLAG_STAYOPEN; - optmask |= ARES_OPT_FLAGS; - - options.sock_state_cb = &TThis::SockStateCallback; - options.sock_state_cb_data = this; - optmask |= ARES_OPT_SOCK_STATE_CB; - - options.timeout = Options.Timeout.MilliSeconds(); - if (options.timeout > 0) { - optmask |= ARES_OPT_TIMEOUTMS; - } - - options.tries = Options.Attempts; - if (options.tries > 0) { - optmask |= ARES_OPT_TRIES; - } - - int err = ares_init_options(&AresChannel, &options, optmask); - Y_VERIFY(err == 0, "Unexpected failure to initialize c-ares channel"); - - if (Options.Servers) { - TStringBuilder csv; - for (const TString& server : Options.Servers) { - if (csv) { - csv << ','; - } - csv << server; - } - err = ares_set_servers_ports_csv(AresChannel, csv.c_str()); - Y_VERIFY(err == 0, "Unexpected failure to set a list of dns servers: %s", ares_strerror(err)); - } - } - - void StopAres() noexcept { - // Destroy the ares channel - ares_destroy(AresChannel); - AresChannel = nullptr; - } - - private: - STRICT_STFUNC(StateWork, { - hFunc(TEvents::TEvPoison, Handle); - hFunc(TEvDns::TEvGetHostByName, Handle); - hFunc(TEvDns::TEvGetAddr, Handle); - }) - - void Handle(TEvents::TEvPoison::TPtr&) { - Y_VERIFY(!Stopped); - - PushCallback([this] { - // Cancel all current ares requests (will send notifications) - ares_cancel(AresChannel); - - // Mark as stopped last - Stopped = true; - }); - - WorkerThread.Join(); - PassAway(); - } - - private: - enum class ERequestType { - GetHostByName, - GetAddr, - }; - - struct TRequestContext : public TThrRefBase { - using TPtr = TIntrusivePtr<TRequestContext>; - - TThis* Self; - TActorSystem* ActorSystem; - TActorId SelfId; - TActorId Sender; - ui64 Cookie; - ERequestType Type; - - TRequestContext(TThis* self, TActorSystem* as, TActorId selfId, TActorId sender, ui64 cookie, ERequestType type) - : Self(self) - , ActorSystem(as) - , SelfId(selfId) - , Sender(sender) - , Cookie(cookie) - , Type(type) - { } - }; - - private: - void Handle(TEvDns::TEvGetHostByName::TPtr& ev) { - auto* msg = ev->Get(); - auto reqCtx = MakeIntrusive<TRequestContext>( - this, TActivationContext::ActorSystem(), SelfId(), ev->Sender, ev->Cookie, ERequestType::GetHostByName); - PushCallback([this, reqCtx = std::move(reqCtx), name = std::move(msg->Name), family = msg->Family] () mutable { - StartGetHostByName(std::move(reqCtx), std::move(name), family); - }); - } - - void Handle(TEvDns::TEvGetAddr::TPtr& ev) { - auto* msg = ev->Get(); - auto reqCtx = MakeIntrusive<TRequestContext>( - this, TActivationContext::ActorSystem(), SelfId(), ev->Sender, ev->Cookie, ERequestType::GetAddr); - PushCallback([this, reqCtx = std::move(reqCtx), name = std::move(msg->Name), family = msg->Family] () mutable { - StartGetHostByName(std::move(reqCtx), std::move(name), family); - }); - } - - void StartGetHostByName(TRequestContext::TPtr reqCtx, TString name, int family) noexcept { - reqCtx->Ref(); - ares_gethostbyname(AresChannel, name.c_str(), family, - &TThis::GetHostByNameAresCallback, reqCtx.Get()); - } - - private: - static void GetHostByNameAresCallback(void* arg, int status, int timeouts, struct hostent* info) { - Y_UNUSED(timeouts); - TRequestContext::TPtr reqCtx(static_cast<TRequestContext*>(arg)); - reqCtx->UnRef(); - - if (reqCtx->Self->Stopped) { - // Don't send any replies after destruction - return; - } - - switch (reqCtx->Type) { - case ERequestType::GetHostByName: { - auto result = MakeHolder<TEvDns::TEvGetHostByNameResult>(); - if (status == 0) { - switch (info->h_addrtype) { - case AF_INET: { - for (int i = 0; info->h_addr_list[i] != nullptr; ++i) { - result->AddrsV4.emplace_back(*(struct in_addr*)(info->h_addr_list[i])); - } - break; - } - case AF_INET6: { - for (int i = 0; info->h_addr_list[i] != nullptr; ++i) { - result->AddrsV6.emplace_back(*(struct in6_addr*)(info->h_addr_list[i])); - } - break; - } - default: - Y_FAIL("unknown address family in ares callback"); - } - } else { - result->ErrorText = ares_strerror(status); - } - result->Status = status; - - reqCtx->ActorSystem->Send(new IEventHandle(reqCtx->Sender, reqCtx->SelfId, result.Release(), 0, reqCtx->Cookie)); - break; - } - - case ERequestType::GetAddr: { - auto result = MakeHolder<TEvDns::TEvGetAddrResult>(); - if (status == 0 && Y_UNLIKELY(info->h_addr_list[0] == nullptr)) { - status = ARES_ENODATA; - } - if (status == 0) { - switch (info->h_addrtype) { - case AF_INET: { - result->Addr = *(struct in_addr*)(info->h_addr_list[0]); - break; - } - case AF_INET6: { - result->Addr = *(struct in6_addr*)(info->h_addr_list[0]); - break; - } - default: - Y_FAIL("unknown address family in ares callback"); - } - } else { - result->ErrorText = ares_strerror(status); - } - result->Status = status; - - reqCtx->ActorSystem->Send(new IEventHandle(reqCtx->Sender, reqCtx->SelfId, result.Release(), 0, reqCtx->Cookie)); - break; - } - } - } - - private: - static void SockStateCallback(void* data, ares_socket_t socket_fd, int readable, int writable) { - static_cast<TThis*>(data)->DoSockStateCallback(socket_fd, readable, writable); - } - - void DoSockStateCallback(ares_socket_t socket_fd, int readable, int writable) noexcept { - int events = (readable ? (POLLRDNORM | POLLIN) : 0) | (writable ? (POLLWRNORM | POLLOUT) : 0); - if (events == 0) { - AresSockStates.erase(socket_fd); - } else { - AresSockStates[socket_fd].NeededEvents = events; - } - } - - private: - static void* WorkerThreadStart(void* arg) noexcept { - static_cast<TSimpleDnsResolver*>(arg)->WorkerThreadLoop(); - return nullptr; - } - - void WorkerThreadLoop() noexcept { - TThread::SetCurrentThreadName("DnsResolver"); - - TVector<struct pollfd> fds; - while (!Stopped) { - fds.clear(); - fds.reserve(1 + AresSockStates.size()); - { - auto& entry = fds.emplace_back(); - entry.fd = WaitSock(); - entry.events = POLLRDNORM | POLLIN; - } - for (auto& kv : AresSockStates) { - auto& entry = fds.emplace_back(); - entry.fd = kv.first; - entry.events = kv.second.NeededEvents; - } - - int timeout = -1; - struct timeval tv; - if (ares_timeout(AresChannel, nullptr, &tv)) { - timeout = tv.tv_sec * 1000 + tv.tv_usec / 1000; - } - - int ret = poll(fds.data(), fds.size(), timeout); - if (ret == -1) { - if (errno == EINTR) { - continue; - } - // we cannot handle failures, run callbacks and pretend everything is ok - RunCallbacks(); - if (Stopped) { - break; - } - ret = 0; - } - - bool ares_called = false; - if (ret > 0) { - for (size_t i = 0; i < fds.size(); ++i) { - auto& entry = fds[i]; - - // Handle WaitSock activation and run callbacks - if (i == 0) { - if (entry.revents & (POLLRDNORM | POLLIN)) { - RunCallbacks(); - if (Stopped) { - break; - } - } - continue; - } - - // All other sockets belong to ares - if (entry.revents == 0) { - continue; - } - // Previous invocation of aress_process_fd might have removed some sockets - if (Y_UNLIKELY(!AresSockStates.contains(entry.fd))) { - continue; - } - ares_process_fd( - AresChannel, - entry.revents & (POLLRDNORM | POLLIN) ? entry.fd : ARES_SOCKET_BAD, - entry.revents & (POLLWRNORM | POLLOUT) ? entry.fd : ARES_SOCKET_BAD); - ares_called = true; - } - - if (Stopped) { - break; - } - } - - if (!ares_called) { - // Let ares handle timeouts - ares_process_fd(AresChannel, ARES_SOCKET_BAD, ARES_SOCKET_BAD); - } - } - } - - private: - struct TSockState { - short NeededEvents = 0; // poll events - }; - - private: - TSimpleDnsResolverOptions Options; - TThread WorkerThread; - - ares_channel AresChannel; - THashMap<SOCKET, TSockState> AresSockStates; - - bool Stopped = false; - }; - - IActor* CreateSimpleDnsResolver(TSimpleDnsResolverOptions options) { - return new TSimpleDnsResolver(std::move(options)); - } - -} // namespace NDnsResolver -} // namespace NActors +#include "dnsresolver.h" + +#include <library/cpp/actors/core/hfunc.h> +#include <library/cpp/threading/queue/mpsc_htswap.h> +#include <util/network/pair.h> +#include <util/network/socket.h> +#include <util/string/builder.h> +#include <util/system/thread.h> + +#include <ares.h> + +#include <atomic> + +namespace NActors { +namespace NDnsResolver { + + class TAresLibraryInitBase { + protected: + TAresLibraryInitBase() noexcept { + int status = ares_library_init(ARES_LIB_INIT_ALL); + Y_VERIFY(status == ARES_SUCCESS, "Unexpected failure to initialize c-ares library"); + } + + ~TAresLibraryInitBase() noexcept { + ares_library_cleanup(); + } + }; + + class TCallbackQueueBase { + protected: + TCallbackQueueBase() noexcept { + int err = SocketPair(Sockets, false, true); + Y_VERIFY(err == 0, "Unexpected failure to create a socket pair"); + SetNonBlock(Sockets[0]); + SetNonBlock(Sockets[1]); + } + + ~TCallbackQueueBase() noexcept { + closesocket(Sockets[0]); + closesocket(Sockets[1]); + } + + protected: + using TCallback = std::function<void()>; + using TCallbackQueue = NThreading::THTSwapQueue<TCallback>; + + void PushCallback(TCallback callback) { + Y_VERIFY(callback, "Cannot push an empty callback"); + CallbackQueue.Push(std::move(callback)); // this is a lockfree queue + + // Wake up worker thread on the first activation + if (Activations.fetch_add(1, std::memory_order_acq_rel) == 0) { + char ch = 'x'; + ssize_t ret; +#ifdef _win_ + ret = send(SignalSock(), &ch, 1, 0); + if (ret == -1) { + Y_VERIFY(WSAGetLastError() == WSAEWOULDBLOCK, "Unexpected send error"); + return; + } +#else + do { + ret = send(SignalSock(), &ch, 1, 0); + } while (ret == -1 && errno == EINTR); + if (ret == -1) { + Y_VERIFY(errno == EAGAIN || errno == EWOULDBLOCK, "Unexpected send error"); + return; + } +#endif + Y_VERIFY(ret == 1, "Unexpected send result"); + } + } + + void RunCallbacks() noexcept { + char ch[32]; + ssize_t ret; + bool signalled = false; + for (;;) { + ret = recv(WaitSock(), ch, sizeof(ch), 0); + if (ret > 0) { + signalled = true; + } + if (ret == sizeof(ch)) { + continue; + } + if (ret != -1) { + break; + } +#ifdef _win_ + if (WSAGetLastError() == WSAEWOULDBLOCK) { + break; + } + Y_FAIL("Unexpected recv error"); +#else + if (errno == EAGAIN || errno == EWOULDBLOCK) { + break; + } + Y_VERIFY(errno == EINTR, "Unexpected recv error"); +#endif + } + + if (signalled) { + // There's exactly one write to SignalSock while Activations != 0 + // It's impossible to get signalled while Activations == 0 + // We must set Activations = 0 to receive new signals + size_t count = Activations.exchange(0, std::memory_order_acq_rel); + Y_VERIFY(count != 0); + + // N.B. due to the way HTSwap works we may not be able to pop + // all callbacks on this activation, however we expect a new + // delayed activation to happen at a later time. + while (auto callback = CallbackQueue.Pop()) { + callback(); + } + } + } + + SOCKET SignalSock() { + return Sockets[0]; + } + + SOCKET WaitSock() { + return Sockets[1]; + } + + private: + SOCKET Sockets[2]; + TCallbackQueue CallbackQueue; + std::atomic<size_t> Activations{ 0 }; + }; + + class TSimpleDnsResolver + : public TActor<TSimpleDnsResolver> + , private TAresLibraryInitBase + , private TCallbackQueueBase + { + public: + TSimpleDnsResolver(TSimpleDnsResolverOptions options) noexcept + : TActor(&TThis::StateWork) + , Options(std::move(options)) + , WorkerThread(&TThis::WorkerThreadStart, this) + { + InitAres(); + + WorkerThread.Start(); + } + + ~TSimpleDnsResolver() noexcept override { + if (!Stopped) { + PushCallback([this] { + // Mark as stopped first + Stopped = true; + + // Cancel all current ares requests (will not send replies) + ares_cancel(AresChannel); + }); + + WorkerThread.Join(); + } + + StopAres(); + } + + static constexpr EActivityType ActorActivityType() { + return DNS_RESOLVER; + } + + private: + void InitAres() noexcept { + struct ares_options options; + memset(&options, 0, sizeof(options)); + int optmask = 0; + + options.flags = ARES_FLAG_STAYOPEN; + optmask |= ARES_OPT_FLAGS; + + options.sock_state_cb = &TThis::SockStateCallback; + options.sock_state_cb_data = this; + optmask |= ARES_OPT_SOCK_STATE_CB; + + options.timeout = Options.Timeout.MilliSeconds(); + if (options.timeout > 0) { + optmask |= ARES_OPT_TIMEOUTMS; + } + + options.tries = Options.Attempts; + if (options.tries > 0) { + optmask |= ARES_OPT_TRIES; + } + + int err = ares_init_options(&AresChannel, &options, optmask); + Y_VERIFY(err == 0, "Unexpected failure to initialize c-ares channel"); + + if (Options.Servers) { + TStringBuilder csv; + for (const TString& server : Options.Servers) { + if (csv) { + csv << ','; + } + csv << server; + } + err = ares_set_servers_ports_csv(AresChannel, csv.c_str()); + Y_VERIFY(err == 0, "Unexpected failure to set a list of dns servers: %s", ares_strerror(err)); + } + } + + void StopAres() noexcept { + // Destroy the ares channel + ares_destroy(AresChannel); + AresChannel = nullptr; + } + + private: + STRICT_STFUNC(StateWork, { + hFunc(TEvents::TEvPoison, Handle); + hFunc(TEvDns::TEvGetHostByName, Handle); + hFunc(TEvDns::TEvGetAddr, Handle); + }) + + void Handle(TEvents::TEvPoison::TPtr&) { + Y_VERIFY(!Stopped); + + PushCallback([this] { + // Cancel all current ares requests (will send notifications) + ares_cancel(AresChannel); + + // Mark as stopped last + Stopped = true; + }); + + WorkerThread.Join(); + PassAway(); + } + + private: + enum class ERequestType { + GetHostByName, + GetAddr, + }; + + struct TRequestContext : public TThrRefBase { + using TPtr = TIntrusivePtr<TRequestContext>; + + TThis* Self; + TActorSystem* ActorSystem; + TActorId SelfId; + TActorId Sender; + ui64 Cookie; + ERequestType Type; + + TRequestContext(TThis* self, TActorSystem* as, TActorId selfId, TActorId sender, ui64 cookie, ERequestType type) + : Self(self) + , ActorSystem(as) + , SelfId(selfId) + , Sender(sender) + , Cookie(cookie) + , Type(type) + { } + }; + + private: + void Handle(TEvDns::TEvGetHostByName::TPtr& ev) { + auto* msg = ev->Get(); + auto reqCtx = MakeIntrusive<TRequestContext>( + this, TActivationContext::ActorSystem(), SelfId(), ev->Sender, ev->Cookie, ERequestType::GetHostByName); + PushCallback([this, reqCtx = std::move(reqCtx), name = std::move(msg->Name), family = msg->Family] () mutable { + StartGetHostByName(std::move(reqCtx), std::move(name), family); + }); + } + + void Handle(TEvDns::TEvGetAddr::TPtr& ev) { + auto* msg = ev->Get(); + auto reqCtx = MakeIntrusive<TRequestContext>( + this, TActivationContext::ActorSystem(), SelfId(), ev->Sender, ev->Cookie, ERequestType::GetAddr); + PushCallback([this, reqCtx = std::move(reqCtx), name = std::move(msg->Name), family = msg->Family] () mutable { + StartGetHostByName(std::move(reqCtx), std::move(name), family); + }); + } + + void StartGetHostByName(TRequestContext::TPtr reqCtx, TString name, int family) noexcept { + reqCtx->Ref(); + ares_gethostbyname(AresChannel, name.c_str(), family, + &TThis::GetHostByNameAresCallback, reqCtx.Get()); + } + + private: + static void GetHostByNameAresCallback(void* arg, int status, int timeouts, struct hostent* info) { + Y_UNUSED(timeouts); + TRequestContext::TPtr reqCtx(static_cast<TRequestContext*>(arg)); + reqCtx->UnRef(); + + if (reqCtx->Self->Stopped) { + // Don't send any replies after destruction + return; + } + + switch (reqCtx->Type) { + case ERequestType::GetHostByName: { + auto result = MakeHolder<TEvDns::TEvGetHostByNameResult>(); + if (status == 0) { + switch (info->h_addrtype) { + case AF_INET: { + for (int i = 0; info->h_addr_list[i] != nullptr; ++i) { + result->AddrsV4.emplace_back(*(struct in_addr*)(info->h_addr_list[i])); + } + break; + } + case AF_INET6: { + for (int i = 0; info->h_addr_list[i] != nullptr; ++i) { + result->AddrsV6.emplace_back(*(struct in6_addr*)(info->h_addr_list[i])); + } + break; + } + default: + Y_FAIL("unknown address family in ares callback"); + } + } else { + result->ErrorText = ares_strerror(status); + } + result->Status = status; + + reqCtx->ActorSystem->Send(new IEventHandle(reqCtx->Sender, reqCtx->SelfId, result.Release(), 0, reqCtx->Cookie)); + break; + } + + case ERequestType::GetAddr: { + auto result = MakeHolder<TEvDns::TEvGetAddrResult>(); + if (status == 0 && Y_UNLIKELY(info->h_addr_list[0] == nullptr)) { + status = ARES_ENODATA; + } + if (status == 0) { + switch (info->h_addrtype) { + case AF_INET: { + result->Addr = *(struct in_addr*)(info->h_addr_list[0]); + break; + } + case AF_INET6: { + result->Addr = *(struct in6_addr*)(info->h_addr_list[0]); + break; + } + default: + Y_FAIL("unknown address family in ares callback"); + } + } else { + result->ErrorText = ares_strerror(status); + } + result->Status = status; + + reqCtx->ActorSystem->Send(new IEventHandle(reqCtx->Sender, reqCtx->SelfId, result.Release(), 0, reqCtx->Cookie)); + break; + } + } + } + + private: + static void SockStateCallback(void* data, ares_socket_t socket_fd, int readable, int writable) { + static_cast<TThis*>(data)->DoSockStateCallback(socket_fd, readable, writable); + } + + void DoSockStateCallback(ares_socket_t socket_fd, int readable, int writable) noexcept { + int events = (readable ? (POLLRDNORM | POLLIN) : 0) | (writable ? (POLLWRNORM | POLLOUT) : 0); + if (events == 0) { + AresSockStates.erase(socket_fd); + } else { + AresSockStates[socket_fd].NeededEvents = events; + } + } + + private: + static void* WorkerThreadStart(void* arg) noexcept { + static_cast<TSimpleDnsResolver*>(arg)->WorkerThreadLoop(); + return nullptr; + } + + void WorkerThreadLoop() noexcept { + TThread::SetCurrentThreadName("DnsResolver"); + + TVector<struct pollfd> fds; + while (!Stopped) { + fds.clear(); + fds.reserve(1 + AresSockStates.size()); + { + auto& entry = fds.emplace_back(); + entry.fd = WaitSock(); + entry.events = POLLRDNORM | POLLIN; + } + for (auto& kv : AresSockStates) { + auto& entry = fds.emplace_back(); + entry.fd = kv.first; + entry.events = kv.second.NeededEvents; + } + + int timeout = -1; + struct timeval tv; + if (ares_timeout(AresChannel, nullptr, &tv)) { + timeout = tv.tv_sec * 1000 + tv.tv_usec / 1000; + } + + int ret = poll(fds.data(), fds.size(), timeout); + if (ret == -1) { + if (errno == EINTR) { + continue; + } + // we cannot handle failures, run callbacks and pretend everything is ok + RunCallbacks(); + if (Stopped) { + break; + } + ret = 0; + } + + bool ares_called = false; + if (ret > 0) { + for (size_t i = 0; i < fds.size(); ++i) { + auto& entry = fds[i]; + + // Handle WaitSock activation and run callbacks + if (i == 0) { + if (entry.revents & (POLLRDNORM | POLLIN)) { + RunCallbacks(); + if (Stopped) { + break; + } + } + continue; + } + + // All other sockets belong to ares + if (entry.revents == 0) { + continue; + } + // Previous invocation of aress_process_fd might have removed some sockets + if (Y_UNLIKELY(!AresSockStates.contains(entry.fd))) { + continue; + } + ares_process_fd( + AresChannel, + entry.revents & (POLLRDNORM | POLLIN) ? entry.fd : ARES_SOCKET_BAD, + entry.revents & (POLLWRNORM | POLLOUT) ? entry.fd : ARES_SOCKET_BAD); + ares_called = true; + } + + if (Stopped) { + break; + } + } + + if (!ares_called) { + // Let ares handle timeouts + ares_process_fd(AresChannel, ARES_SOCKET_BAD, ARES_SOCKET_BAD); + } + } + } + + private: + struct TSockState { + short NeededEvents = 0; // poll events + }; + + private: + TSimpleDnsResolverOptions Options; + TThread WorkerThread; + + ares_channel AresChannel; + THashMap<SOCKET, TSockState> AresSockStates; + + bool Stopped = false; + }; + + IActor* CreateSimpleDnsResolver(TSimpleDnsResolverOptions options) { + return new TSimpleDnsResolver(std::move(options)); + } + +} // namespace NDnsResolver +} // namespace NActors diff --git a/library/cpp/actors/dnsresolver/dnsresolver.h b/library/cpp/actors/dnsresolver/dnsresolver.h index 6aa28641fc..88fc74df7d 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver.h +++ b/library/cpp/actors/dnsresolver/dnsresolver.h @@ -1,128 +1,128 @@ -#pragma once - -#include <library/cpp/actors/core/actor.h> -#include <library/cpp/actors/core/events.h> -#include <library/cpp/actors/core/event_local.h> -#include <library/cpp/monlib/dynamic_counters/counters.h> -#include <util/network/address.h> -#include <variant> - -namespace NActors { -namespace NDnsResolver { - - struct TEvDns { - enum EEv { - EvGetHostByName = EventSpaceBegin(TEvents::ES_DNS), - EvGetHostByNameResult, - EvGetAddr, - EvGetAddrResult, - }; - - /** - * TEvGetHostByName returns the result of ares_gethostbyname - */ - struct TEvGetHostByName : public TEventLocal<TEvGetHostByName, EvGetHostByName> { - TString Name; - int Family; - - explicit TEvGetHostByName(TString name, int family = AF_UNSPEC) - : Name(std::move(name)) - , Family(family) - { } - }; - - struct TEvGetHostByNameResult : public TEventLocal<TEvGetHostByNameResult, EvGetHostByNameResult> { - TVector<struct in_addr> AddrsV4; - TVector<struct in6_addr> AddrsV6; - TString ErrorText; - int Status = 0; - }; - - /** - * TEvGetAddr returns a single address for a given hostname - */ - struct TEvGetAddr : public TEventLocal<TEvGetAddr, EvGetAddr> { - TString Name; - int Family; - - explicit TEvGetAddr(TString name, int family = AF_UNSPEC) - : Name(std::move(name)) - , Family(family) - { } - }; - - struct TEvGetAddrResult : public TEventLocal<TEvGetAddrResult, EvGetAddrResult> { - // N.B. "using" here doesn't work with Visual Studio compiler - typedef struct in6_addr TIPv6Addr; - typedef struct in_addr TIPv4Addr; - - std::variant<std::monostate, TIPv6Addr, TIPv4Addr> Addr; - TString ErrorText; - int Status = 0; - - bool IsV6() const { - return std::holds_alternative<TIPv6Addr>(Addr); - } - - bool IsV4() const { - return std::holds_alternative<TIPv4Addr>(Addr); - } - - const TIPv6Addr& GetAddrV6() const { - const TIPv6Addr* p = std::get_if<TIPv6Addr>(&Addr); - Y_VERIFY(p, "Result is not an ipv6 address"); - return *p; - } - - const TIPv4Addr& GetAddrV4() const { - const TIPv4Addr* p = std::get_if<TIPv4Addr>(&Addr); - Y_VERIFY(p, "Result is not an ipv4 address"); - return *p; - } - }; - }; - - struct TSimpleDnsResolverOptions { - // Initial per-server timeout, grows exponentially with each retry - TDuration Timeout = TDuration::Seconds(1); - // Number of attempts per-server - int Attempts = 2; - // Optional list of custom dns servers (ip.v4[:port], ip::v6 or [ip::v6]:port format) - TVector<TString> Servers; - }; - - IActor* CreateSimpleDnsResolver(TSimpleDnsResolverOptions options = TSimpleDnsResolverOptions()); - - struct TCachingDnsResolverOptions { - // Soft expire time specifies delay before name is refreshed in background - TDuration SoftNegativeExpireTime = TDuration::Seconds(1); - TDuration SoftPositiveExpireTime = TDuration::Seconds(10); - // Hard expire time specifies delay before the last result is forgotten - TDuration HardNegativeExpireTime = TDuration::Seconds(10); - TDuration HardPositiveExpireTime = TDuration::Hours(2); - // Allow these request families - bool AllowIPv6 = true; - bool AllowIPv4 = true; - // Optional counters - NMonitoring::TDynamicCounterPtr MonCounters = nullptr; - }; - - IActor* CreateCachingDnsResolver(TActorId upstream, TCachingDnsResolverOptions options = TCachingDnsResolverOptions()); - - struct TOnDemandDnsResolverOptions - : public TSimpleDnsResolverOptions - , public TCachingDnsResolverOptions - { - }; - - IActor* CreateOnDemandDnsResolver(TOnDemandDnsResolverOptions options = TOnDemandDnsResolverOptions()); - - /** - * Returns actor id of a globally registered dns resolver - */ - inline TActorId MakeDnsResolverActorId() { - return TActorId(0, TStringBuf("dnsresolver")); - } - -} // namespace NDnsResolver -} // namespace NActors +#pragma once + +#include <library/cpp/actors/core/actor.h> +#include <library/cpp/actors/core/events.h> +#include <library/cpp/actors/core/event_local.h> +#include <library/cpp/monlib/dynamic_counters/counters.h> +#include <util/network/address.h> +#include <variant> + +namespace NActors { +namespace NDnsResolver { + + struct TEvDns { + enum EEv { + EvGetHostByName = EventSpaceBegin(TEvents::ES_DNS), + EvGetHostByNameResult, + EvGetAddr, + EvGetAddrResult, + }; + + /** + * TEvGetHostByName returns the result of ares_gethostbyname + */ + struct TEvGetHostByName : public TEventLocal<TEvGetHostByName, EvGetHostByName> { + TString Name; + int Family; + + explicit TEvGetHostByName(TString name, int family = AF_UNSPEC) + : Name(std::move(name)) + , Family(family) + { } + }; + + struct TEvGetHostByNameResult : public TEventLocal<TEvGetHostByNameResult, EvGetHostByNameResult> { + TVector<struct in_addr> AddrsV4; + TVector<struct in6_addr> AddrsV6; + TString ErrorText; + int Status = 0; + }; + + /** + * TEvGetAddr returns a single address for a given hostname + */ + struct TEvGetAddr : public TEventLocal<TEvGetAddr, EvGetAddr> { + TString Name; + int Family; + + explicit TEvGetAddr(TString name, int family = AF_UNSPEC) + : Name(std::move(name)) + , Family(family) + { } + }; + + struct TEvGetAddrResult : public TEventLocal<TEvGetAddrResult, EvGetAddrResult> { + // N.B. "using" here doesn't work with Visual Studio compiler + typedef struct in6_addr TIPv6Addr; + typedef struct in_addr TIPv4Addr; + + std::variant<std::monostate, TIPv6Addr, TIPv4Addr> Addr; + TString ErrorText; + int Status = 0; + + bool IsV6() const { + return std::holds_alternative<TIPv6Addr>(Addr); + } + + bool IsV4() const { + return std::holds_alternative<TIPv4Addr>(Addr); + } + + const TIPv6Addr& GetAddrV6() const { + const TIPv6Addr* p = std::get_if<TIPv6Addr>(&Addr); + Y_VERIFY(p, "Result is not an ipv6 address"); + return *p; + } + + const TIPv4Addr& GetAddrV4() const { + const TIPv4Addr* p = std::get_if<TIPv4Addr>(&Addr); + Y_VERIFY(p, "Result is not an ipv4 address"); + return *p; + } + }; + }; + + struct TSimpleDnsResolverOptions { + // Initial per-server timeout, grows exponentially with each retry + TDuration Timeout = TDuration::Seconds(1); + // Number of attempts per-server + int Attempts = 2; + // Optional list of custom dns servers (ip.v4[:port], ip::v6 or [ip::v6]:port format) + TVector<TString> Servers; + }; + + IActor* CreateSimpleDnsResolver(TSimpleDnsResolverOptions options = TSimpleDnsResolverOptions()); + + struct TCachingDnsResolverOptions { + // Soft expire time specifies delay before name is refreshed in background + TDuration SoftNegativeExpireTime = TDuration::Seconds(1); + TDuration SoftPositiveExpireTime = TDuration::Seconds(10); + // Hard expire time specifies delay before the last result is forgotten + TDuration HardNegativeExpireTime = TDuration::Seconds(10); + TDuration HardPositiveExpireTime = TDuration::Hours(2); + // Allow these request families + bool AllowIPv6 = true; + bool AllowIPv4 = true; + // Optional counters + NMonitoring::TDynamicCounterPtr MonCounters = nullptr; + }; + + IActor* CreateCachingDnsResolver(TActorId upstream, TCachingDnsResolverOptions options = TCachingDnsResolverOptions()); + + struct TOnDemandDnsResolverOptions + : public TSimpleDnsResolverOptions + , public TCachingDnsResolverOptions + { + }; + + IActor* CreateOnDemandDnsResolver(TOnDemandDnsResolverOptions options = TOnDemandDnsResolverOptions()); + + /** + * Returns actor id of a globally registered dns resolver + */ + inline TActorId MakeDnsResolverActorId() { + return TActorId(0, TStringBuf("dnsresolver")); + } + +} // namespace NDnsResolver +} // namespace NActors diff --git a/library/cpp/actors/dnsresolver/dnsresolver_caching.cpp b/library/cpp/actors/dnsresolver/dnsresolver_caching.cpp index a1688854c4..02760f4c27 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver_caching.cpp +++ b/library/cpp/actors/dnsresolver/dnsresolver_caching.cpp @@ -1,730 +1,730 @@ -#include "dnsresolver.h" - -#include <library/cpp/actors/core/hfunc.h> -#include <util/generic/intrlist.h> - -#include <ares.h> - -#include <queue> - -namespace NActors { -namespace NDnsResolver { - - class TCachingDnsResolver : public TActor<TCachingDnsResolver> { - public: - struct TMonCounters { - NMonitoring::TDynamicCounters::TCounterPtr OutgoingInFlightV4; - NMonitoring::TDynamicCounters::TCounterPtr OutgoingInFlightV6; - NMonitoring::TDynamicCounters::TCounterPtr OutgoingErrorsV4; - NMonitoring::TDynamicCounters::TCounterPtr OutgoingErrorsV6; - NMonitoring::TDynamicCounters::TCounterPtr OutgoingTotalV4; - NMonitoring::TDynamicCounters::TCounterPtr OutgoingTotalV6; - - NMonitoring::TDynamicCounters::TCounterPtr IncomingInFlight; - NMonitoring::TDynamicCounters::TCounterPtr IncomingErrors; - NMonitoring::TDynamicCounters::TCounterPtr IncomingTotal; - - NMonitoring::TDynamicCounters::TCounterPtr CacheSize; - NMonitoring::TDynamicCounters::TCounterPtr CacheHits; - NMonitoring::TDynamicCounters::TCounterPtr CacheMisses; - - TMonCounters(const NMonitoring::TDynamicCounterPtr& counters) - : OutgoingInFlightV4(counters->GetCounter("DnsResolver/Outgoing/InFlight/V4", false)) - , OutgoingInFlightV6(counters->GetCounter("DnsResolver/Outgoing/InFlight/V6", false)) - , OutgoingErrorsV4(counters->GetCounter("DnsResolver/Outgoing/Errors/V4", true)) - , OutgoingErrorsV6(counters->GetCounter("DnsResolver/Outgoing/Errors/V6", true)) - , OutgoingTotalV4(counters->GetCounter("DnsResolver/Outgoing/Total/V4", true)) - , OutgoingTotalV6(counters->GetCounter("DnsResolver/Outgoing/Total/V6", true)) - , IncomingInFlight(counters->GetCounter("DnsResolver/Incoming/InFlight", false)) - , IncomingErrors(counters->GetCounter("DnsResolver/Incoming/Errors", true)) - , IncomingTotal(counters->GetCounter("DnsResolver/Incoming/Total", true)) - , CacheSize(counters->GetCounter("DnsResolver/Cache/Size", false)) - , CacheHits(counters->GetCounter("DnsResolver/Cache/Hits", true)) - , CacheMisses(counters->GetCounter("DnsResolver/Cache/Misses", true)) - { } - }; - - public: - TCachingDnsResolver(TActorId upstream, TCachingDnsResolverOptions options) - : TActor(&TThis::StateWork) - , Upstream(upstream) - , Options(std::move(options)) - , MonCounters(Options.MonCounters ? new TMonCounters(Options.MonCounters) : nullptr) - { } - - static constexpr EActivityType ActorActivityType() { - return DNS_RESOLVER; - } - - private: - STRICT_STFUNC(StateWork, { - hFunc(TEvents::TEvPoison, Handle); - hFunc(TEvDns::TEvGetHostByName, Handle); - hFunc(TEvDns::TEvGetAddr, Handle); - hFunc(TEvDns::TEvGetHostByNameResult, Handle); - hFunc(TEvents::TEvUndelivered, Handle); - }); - - void Handle(TEvents::TEvPoison::TPtr&) { - DropPending(ARES_ECANCELLED); - PassAway(); - } - - void Handle(TEvDns::TEvGetHostByName::TPtr& ev) { - auto req = MakeHolder<TIncomingRequest>(); - req->Type = EIncomingRequestType::GetHostByName; - req->Sender = ev->Sender; - req->Cookie = ev->Cookie; - req->Name = std::move(ev->Get()->Name); - req->Family = ev->Get()->Family; - EnqueueRequest(std::move(req)); - } - - void Handle(TEvDns::TEvGetAddr::TPtr& ev) { - auto req = MakeHolder<TIncomingRequest>(); - req->Type = EIncomingRequestType::GetAddr; - req->Sender = ev->Sender; - req->Cookie = ev->Cookie; - req->Name = std::move(ev->Get()->Name); - req->Family = ev->Get()->Family; - EnqueueRequest(std::move(req)); - } - - void Handle(TEvDns::TEvGetHostByNameResult::TPtr& ev) { - auto waitingIt = WaitingRequests.find(ev->Cookie); - Y_VERIFY(waitingIt != WaitingRequests.end(), "Unexpected reply, reqId=%" PRIu64, ev->Cookie); - auto waitingInfo = waitingIt->second; - WaitingRequests.erase(waitingIt); - - switch (waitingInfo.Family) { - case AF_INET6: - if (ev->Get()->Status) { - ProcessErrorV6(waitingInfo.Position, ev->Get()->Status, std::move(ev->Get()->ErrorText)); - } else { - ProcessAddrsV6(waitingInfo.Position, std::move(ev->Get()->AddrsV6)); - } - break; - - case AF_INET: - if (ev->Get()->Status) { - ProcessErrorV4(waitingInfo.Position, ev->Get()->Status, std::move(ev->Get()->ErrorText)); - } else { - ProcessAddrsV4(waitingInfo.Position, std::move(ev->Get()->AddrsV4)); - } - break; - - default: - Y_FAIL("Unexpected request family %d", waitingInfo.Family); - } - } - - void Handle(TEvents::TEvUndelivered::TPtr& ev) { - switch (ev->Get()->SourceType) { - case TEvDns::TEvGetHostByName::EventType: { - auto waitingIt = WaitingRequests.find(ev->Cookie); - Y_VERIFY(waitingIt != WaitingRequests.end(), "Unexpected TEvUndelivered, reqId=%" PRIu64, ev->Cookie); - auto waitingInfo = waitingIt->second; - WaitingRequests.erase(waitingIt); - - switch (waitingInfo.Family) { - case AF_INET6: - ProcessErrorV6(waitingInfo.Position, ARES_ENOTINITIALIZED, "Caching dns resolver cannot deliver to the underlying resolver"); - break; - case AF_INET: - ProcessErrorV4(waitingInfo.Position, ARES_ENOTINITIALIZED, "Caching dns resolver cannot deliver to the underlying resolver"); - break; - default: - Y_FAIL("Unexpected request family %d", waitingInfo.Family); - } - - break; - } - - default: - Y_FAIL("Unexpected TEvUndelievered, type=%" PRIu32, ev->Get()->SourceType); - } - } - - private: - enum EIncomingRequestType { - GetHostByName, - GetAddr, - }; - - struct TIncomingRequest : public TIntrusiveListItem<TIncomingRequest> { - EIncomingRequestType Type; - TActorId Sender; - ui64 Cookie; - TString Name; - int Family; - }; - - using TIncomingRequestList = TIntrusiveListWithAutoDelete<TIncomingRequest, TDelete>; - - void EnqueueRequest(THolder<TIncomingRequest> req) { - if (MonCounters) { - ++*MonCounters->IncomingTotal; - } - - CleanupExpired(TActivationContext::Now()); - - switch (req->Family) { - case AF_UNSPEC: - if (Options.AllowIPv6) { - EnqueueRequestIPv6(std::move(req)); - return; - } - if (Options.AllowIPv4) { - EnqueueRequestIPv4(std::move(req)); - return; - } - break; - - case AF_INET6: - if (Options.AllowIPv6) { - EnqueueRequestIPv6(std::move(req)); - return; - } - break; - - case AF_INET: - if (Options.AllowIPv4) { - EnqueueRequestIPv4(std::move(req)); - return; - } - break; - } - - ReplyWithError(std::move(req), ARES_EBADFAMILY); - } - - void EnqueueRequestIPv6(THolder<TIncomingRequest> req) { - auto now = TActivationContext::Now(); - - auto& fullState = NameToState[req->Name]; - if (MonCounters) { - *MonCounters->CacheSize = NameToState.size(); - } - - auto& state = fullState.StateIPv6; - EnsureRequest(state, req->Name, AF_INET6, now); - - if (state.IsHardExpired(now)) { - Y_VERIFY(state.Waiting); - if (MonCounters) { - ++*MonCounters->CacheMisses; - } - // We need to wait for ipv6 reply, schedule ipv4 request in parallel if needed - if (Options.AllowIPv4) { - EnsureRequest(fullState.StateIPv4, req->Name, AF_INET, now); - } - state.WaitingRequests.PushBack(req.Release()); - return; - } - - // We want to retry AF_UNSPEC with IPv4 in some cases - if (req->Family == AF_UNSPEC && Options.AllowIPv4 && state.RetryUnspec()) { - EnqueueRequestIPv4(std::move(req)); - return; - } - - if (MonCounters) { - ++*MonCounters->CacheHits; - } - - if (state.Status != 0) { - ReplyWithError(std::move(req), state.Status, state.ErrorText); - } else { - ReplyWithAddrs(std::move(req), fullState.AddrsIPv6); - } - } - - void EnqueueRequestIPv4(THolder<TIncomingRequest> req, bool isCacheMiss = false) { - auto now = TActivationContext::Now(); - - auto& fullState = NameToState[req->Name]; - if (MonCounters) { - *MonCounters->CacheSize = NameToState.size(); - } - - auto& state = fullState.StateIPv4; - EnsureRequest(state, req->Name, AF_INET, now); - - if (state.IsHardExpired(now)) { - Y_VERIFY(state.Waiting); - if (MonCounters && !isCacheMiss) { - ++*MonCounters->CacheMisses; - } - state.WaitingRequests.PushBack(req.Release()); - return; - } - - if (MonCounters && !isCacheMiss) { - ++*MonCounters->CacheHits; - } - - if (state.Status != 0) { - ReplyWithError(std::move(req), state.Status, state.ErrorText); - } else { - ReplyWithAddrs(std::move(req), fullState.AddrsIPv4); - } - } - - private: - struct TFamilyState { - TIncomingRequestList WaitingRequests; - TInstant SoftDeadline; - TInstant HardDeadline; - TInstant NextSoftDeadline; - TInstant NextHardDeadline; - TString ErrorText; - int Status = -1; // never requested before - bool InSoftHeap = false; - bool InHardHeap = false; - bool Waiting = false; - - bool Needed() const { - return InSoftHeap || InHardHeap || Waiting; - } - - bool RetryUnspec() const { - return ( - Status == ARES_ENODATA || - Status == ARES_EBADRESP || - Status == ARES_ETIMEOUT); - } - - bool ServerReplied() const { - return ServerReplied(Status); - } - - bool IsSoftExpired(TInstant now) const { - return !InSoftHeap || NextSoftDeadline < now; - } - - bool IsHardExpired(TInstant now) const { - return !InHardHeap || NextHardDeadline < now; - } - - static bool ServerReplied(int status) { - return ( - status == ARES_SUCCESS || - status == ARES_ENODATA || - status == ARES_ENOTFOUND); - } - }; - - struct TState { - TFamilyState StateIPv6; - TFamilyState StateIPv4; - TVector<struct in6_addr> AddrsIPv6; - TVector<struct in_addr> AddrsIPv4; - - bool Needed() const { - return StateIPv6.Needed() || StateIPv4.Needed(); - } - }; - - using TNameToState = THashMap<TString, TState>; - - template<const TFamilyState TState::* StateToFamily, - const TInstant TFamilyState::* FamilyToDeadline> - struct THeapCompare { - // returns true when b < a - bool operator()(TNameToState::iterator a, TNameToState::iterator b) const { - const TState& aState = a->second; - const TState& bState = b->second; - const TFamilyState& aFamily = aState.*StateToFamily; - const TFamilyState& bFamily = bState.*StateToFamily; - const TInstant& aDeadline = aFamily.*FamilyToDeadline; - const TInstant& bDeadline = bFamily.*FamilyToDeadline; - return bDeadline < aDeadline; - } - }; - - template<const TFamilyState TState::* StateToFamily, - const TInstant TFamilyState::* FamilyToDeadline> - using TStateHeap = std::priority_queue< - TNameToState::iterator, - std::vector<TNameToState::iterator>, - THeapCompare<StateToFamily, FamilyToDeadline> - >; - - struct TWaitingInfo { - TNameToState::iterator Position; - int Family; - }; - - private: - void EnsureRequest(TFamilyState& state, const TString& name, int family, TInstant now) { - if (state.Waiting) { - return; // request is already pending - } - - if (!state.IsSoftExpired(now) && !state.IsHardExpired(now)) { - return; // response is not expired yet - } - - if (MonCounters) { - switch (family) { - case AF_INET6: - ++*MonCounters->OutgoingInFlightV6; - ++*MonCounters->OutgoingTotalV6; - break; - case AF_INET: - ++*MonCounters->OutgoingInFlightV4; - ++*MonCounters->OutgoingTotalV4; - break; - } - } - - ui64 reqId = ++LastRequestId; - auto& req = WaitingRequests[reqId]; - req.Position = NameToState.find(name); - req.Family = family; - Y_VERIFY(req.Position != NameToState.end()); - - Send(Upstream, new TEvDns::TEvGetHostByName(name, family), IEventHandle::FlagTrackDelivery, reqId); - state.Waiting = true; - } - - template<TFamilyState TState::* StateToFamily, - TInstant TFamilyState::* FamilyToDeadline, - TInstant TFamilyState::* FamilyToNextDeadline, - bool TFamilyState::* FamilyToFlag, - class THeap> - void PushToHeap(THeap& heap, TNameToState::iterator it, TInstant newDeadline) { - auto& family = it->second.*StateToFamily; - TInstant& deadline = family.*FamilyToDeadline; - TInstant& nextDeadline = family.*FamilyToNextDeadline; - bool& flag = family.*FamilyToFlag; - nextDeadline = newDeadline; - if (!flag) { - deadline = newDeadline; - heap.push(it); - flag = true; - } - } - - void PushSoftV6(TNameToState::iterator it, TInstant newDeadline) { - PushToHeap<&TState::StateIPv6, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv6, it, newDeadline); - } - - void PushHardV6(TNameToState::iterator it, TInstant newDeadline) { - PushToHeap<&TState::StateIPv6, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv6, it, newDeadline); - } - - void PushSoftV4(TNameToState::iterator it, TInstant newDeadline) { - PushToHeap<&TState::StateIPv4, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv4, it, newDeadline); - } - - void PushHardV4(TNameToState::iterator it, TInstant newDeadline) { - PushToHeap<&TState::StateIPv4, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv4, it, newDeadline); - } - - void ProcessErrorV6(TNameToState::iterator it, int status, TString errorText) { - auto now = TActivationContext::Now(); - if (MonCounters) { - --*MonCounters->OutgoingInFlightV6; - ++*MonCounters->OutgoingErrorsV6; - } - - auto& state = it->second.StateIPv6; - Y_VERIFY(state.Waiting, "Got error for a state we are not waiting"); - state.Waiting = false; - - // When we have a cached positive reply, don't overwrite it with spurious errors - const bool serverReplied = TFamilyState::ServerReplied(status); - if (!serverReplied && state.ServerReplied() && !state.IsHardExpired(now)) { - PushSoftV6(it, now + Options.SoftNegativeExpireTime); - if (state.Status == ARES_SUCCESS) { - SendAddrsV6(it); - } else { - SendErrorsV6(it, now); - } - return; - } - - state.Status = status; - state.ErrorText = std::move(errorText); - PushSoftV6(it, now + Options.SoftNegativeExpireTime); - if (serverReplied) { - // Server actually replied, so keep it cached for longer - PushHardV6(it, now + Options.HardPositiveExpireTime); - } else { - PushHardV6(it, now + Options.HardNegativeExpireTime); - } - - SendErrorsV6(it, now); - } - - void SendErrorsV6(TNameToState::iterator it, TInstant now) { - bool cleaned = false; - auto& state = it->second.StateIPv6; - while (state.WaitingRequests) { - THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); - if (req->Family == AF_UNSPEC && Options.AllowIPv4 && state.RetryUnspec()) { - if (!cleaned) { - CleanupExpired(now); - cleaned = true; - } - EnqueueRequestIPv4(std::move(req), /* isCacheMiss */ true); - } else { - ReplyWithError(std::move(req), state.Status, state.ErrorText); - } - } - } - - void ProcessErrorV4(TNameToState::iterator it, int status, TString errorText) { - auto now = TActivationContext::Now(); - if (MonCounters) { - --*MonCounters->OutgoingInFlightV4; - ++*MonCounters->OutgoingErrorsV4; - } - - auto& state = it->second.StateIPv4; - Y_VERIFY(state.Waiting, "Got error for a state we are not waiting"); - state.Waiting = false; - - // When we have a cached positive reply, don't overwrite it with spurious errors - const bool serverReplied = TFamilyState::ServerReplied(status); - if (!serverReplied && state.ServerReplied() && !state.IsHardExpired(now)) { - PushSoftV4(it, now + Options.SoftNegativeExpireTime); - if (state.Status == ARES_SUCCESS) { - SendAddrsV4(it); - } else { - SendErrorsV4(it); - } - return; - } - - state.Status = status; - state.ErrorText = std::move(errorText); - PushSoftV4(it, now + Options.SoftNegativeExpireTime); - if (serverReplied) { - // Server actually replied, so keep it cached for longer - PushHardV4(it, now + Options.HardPositiveExpireTime); - } else { - PushHardV4(it, now + Options.HardNegativeExpireTime); - } - - SendErrorsV4(it); - } - - void SendErrorsV4(TNameToState::iterator it) { - auto& state = it->second.StateIPv4; - while (state.WaitingRequests) { - THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); - ReplyWithError(std::move(req), state.Status, state.ErrorText); - } - } - - void ProcessAddrsV6(TNameToState::iterator it, TVector<struct in6_addr> addrs) { - if (Y_UNLIKELY(addrs.empty())) { - // Probably unnecessary: we don't want to deal with empty address lists - return ProcessErrorV6(it, ARES_ENODATA, ares_strerror(ARES_ENODATA)); - } - - auto now = TActivationContext::Now(); - if (MonCounters) { - --*MonCounters->OutgoingInFlightV6; - } - - auto& state = it->second.StateIPv6; - Y_VERIFY(state.Waiting, "Got reply for a state we are not waiting"); - state.Waiting = false; - - state.Status = ARES_SUCCESS; - it->second.AddrsIPv6 = std::move(addrs); - PushSoftV6(it, now + Options.SoftPositiveExpireTime); - PushHardV6(it, now + Options.HardPositiveExpireTime); - - SendAddrsV6(it); - } - - void SendAddrsV6(TNameToState::iterator it) { - auto& state = it->second.StateIPv6; - while (state.WaitingRequests) { - THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); - ReplyWithAddrs(std::move(req), it->second.AddrsIPv6); - } - } - - void ProcessAddrsV4(TNameToState::iterator it, TVector<struct in_addr> addrs) { - if (Y_UNLIKELY(addrs.empty())) { - // Probably unnecessary: we don't want to deal with empty address lists - return ProcessErrorV4(it, ARES_ENODATA, ares_strerror(ARES_ENODATA)); - } - - auto now = TActivationContext::Now(); - if (MonCounters) { - --*MonCounters->OutgoingInFlightV4; - } - - auto& state = it->second.StateIPv4; - Y_VERIFY(state.Waiting, "Got reply for a state we are not waiting"); - state.Waiting = false; - - state.Status = ARES_SUCCESS; - it->second.AddrsIPv4 = std::move(addrs); - PushSoftV4(it, now + Options.SoftPositiveExpireTime); - PushHardV4(it, now + Options.HardPositiveExpireTime); - - SendAddrsV4(it); - } - - void SendAddrsV4(TNameToState::iterator it) { - auto& state = it->second.StateIPv4; - while (state.WaitingRequests) { - THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); - ReplyWithAddrs(std::move(req), it->second.AddrsIPv4); - } - } - - private: - template<TFamilyState TState::*StateToFamily, - TInstant TFamilyState::* FamilyToDeadline, - TInstant TFamilyState::* FamilyToNextDeadline, - bool TFamilyState::* FamilyToFlag> - void DoCleanupExpired(TStateHeap<StateToFamily, FamilyToDeadline>& heap, TInstant now) { - while (!heap.empty()) { - auto it = heap.top(); - auto& family = it->second.*StateToFamily; - TInstant& deadline = family.*FamilyToDeadline; - if (now <= deadline) { - break; - } - - bool& flag = family.*FamilyToFlag; - Y_VERIFY(flag); - heap.pop(); - flag = false; - - TInstant& nextDeadline = family.*FamilyToNextDeadline; - if (now < nextDeadline) { - deadline = nextDeadline; - heap.push(it); - flag = true; - continue; - } - - // Remove unnecessary items - if (!it->second.Needed()) { - NameToState.erase(it); - if (MonCounters) { - *MonCounters->CacheSize = NameToState.size(); - } - } - } - } - - void CleanupExpired(TInstant now) { - DoCleanupExpired<&TState::StateIPv6, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv6, now); - DoCleanupExpired<&TState::StateIPv6, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv6, now); - DoCleanupExpired<&TState::StateIPv4, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv4, now); - DoCleanupExpired<&TState::StateIPv4, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv4, now); - } - - template<class TEvent> - void SendError(TActorId replyTo, ui64 cookie, int status, const TString& errorText) { - auto reply = MakeHolder<TEvent>(); - reply->Status = status; - reply->ErrorText = errorText; - this->Send(replyTo, reply.Release(), 0, cookie); - } - - void ReplyWithError(THolder<TIncomingRequest> req, int status, const TString& errorText) { - if (MonCounters) { - ++*MonCounters->IncomingErrors; - } - switch (req->Type) { - case EIncomingRequestType::GetHostByName: { - SendError<TEvDns::TEvGetHostByNameResult>(req->Sender, req->Cookie, status, errorText); - break; - } - case EIncomingRequestType::GetAddr: { - SendError<TEvDns::TEvGetAddrResult>(req->Sender, req->Cookie, status, errorText); - break; - } - } - } - - void ReplyWithAddrs(THolder<TIncomingRequest> req, const TVector<struct in6_addr>& addrs) { - switch (req->Type) { - case EIncomingRequestType::GetHostByName: { - auto reply = MakeHolder<TEvDns::TEvGetHostByNameResult>(); - reply->AddrsV6 = addrs; - Send(req->Sender, reply.Release(), 0, req->Cookie); - break; - } - case EIncomingRequestType::GetAddr: { - Y_VERIFY(!addrs.empty()); - auto reply = MakeHolder<TEvDns::TEvGetAddrResult>(); - reply->Addr = addrs.front(); - Send(req->Sender, reply.Release(), 0, req->Cookie); - break; - } - } - } - - void ReplyWithAddrs(THolder<TIncomingRequest> req, const TVector<struct in_addr>& addrs) { - switch (req->Type) { - case EIncomingRequestType::GetHostByName: { - auto reply = MakeHolder<TEvDns::TEvGetHostByNameResult>(); - reply->AddrsV4 = addrs; - Send(req->Sender, reply.Release(), 0, req->Cookie); - break; - } - case EIncomingRequestType::GetAddr: { - Y_VERIFY(!addrs.empty()); - auto reply = MakeHolder<TEvDns::TEvGetAddrResult>(); - reply->Addr = addrs.front(); - Send(req->Sender, reply.Release(), 0, req->Cookie); - break; - } - } - } - - void ReplyWithError(THolder<TIncomingRequest> req, int status) { - ReplyWithError(std::move(req), status, ares_strerror(status)); - } - - void DropPending(TIncomingRequestList& list, int status, const TString& errorText) { - while (list) { - THolder<TIncomingRequest> req(list.PopFront()); - ReplyWithError(std::move(req), status, errorText); - } - } - - void DropPending(int status, const TString& errorText) { - for (auto& [name, state] : NameToState) { - DropPending(state.StateIPv6.WaitingRequests, status, errorText); - DropPending(state.StateIPv4.WaitingRequests, status, errorText); - } - } - - void DropPending(int status) { - DropPending(status, ares_strerror(status)); - } - - private: - const TActorId Upstream; - const TCachingDnsResolverOptions Options; - const THolder<TMonCounters> MonCounters; - - TNameToState NameToState; - TStateHeap<&TState::StateIPv6, &TFamilyState::SoftDeadline> SoftHeapIPv6; - TStateHeap<&TState::StateIPv6, &TFamilyState::HardDeadline> HardHeapIPv6; - TStateHeap<&TState::StateIPv4, &TFamilyState::SoftDeadline> SoftHeapIPv4; - TStateHeap<&TState::StateIPv4, &TFamilyState::HardDeadline> HardHeapIPv4; - - THashMap<ui64, TWaitingInfo> WaitingRequests; - ui64 LastRequestId = 0; - }; - - IActor* CreateCachingDnsResolver(TActorId upstream, TCachingDnsResolverOptions options) { - return new TCachingDnsResolver(upstream, std::move(options)); - } - -} // namespace NDnsResolver -} // namespace NActors +#include "dnsresolver.h" + +#include <library/cpp/actors/core/hfunc.h> +#include <util/generic/intrlist.h> + +#include <ares.h> + +#include <queue> + +namespace NActors { +namespace NDnsResolver { + + class TCachingDnsResolver : public TActor<TCachingDnsResolver> { + public: + struct TMonCounters { + NMonitoring::TDynamicCounters::TCounterPtr OutgoingInFlightV4; + NMonitoring::TDynamicCounters::TCounterPtr OutgoingInFlightV6; + NMonitoring::TDynamicCounters::TCounterPtr OutgoingErrorsV4; + NMonitoring::TDynamicCounters::TCounterPtr OutgoingErrorsV6; + NMonitoring::TDynamicCounters::TCounterPtr OutgoingTotalV4; + NMonitoring::TDynamicCounters::TCounterPtr OutgoingTotalV6; + + NMonitoring::TDynamicCounters::TCounterPtr IncomingInFlight; + NMonitoring::TDynamicCounters::TCounterPtr IncomingErrors; + NMonitoring::TDynamicCounters::TCounterPtr IncomingTotal; + + NMonitoring::TDynamicCounters::TCounterPtr CacheSize; + NMonitoring::TDynamicCounters::TCounterPtr CacheHits; + NMonitoring::TDynamicCounters::TCounterPtr CacheMisses; + + TMonCounters(const NMonitoring::TDynamicCounterPtr& counters) + : OutgoingInFlightV4(counters->GetCounter("DnsResolver/Outgoing/InFlight/V4", false)) + , OutgoingInFlightV6(counters->GetCounter("DnsResolver/Outgoing/InFlight/V6", false)) + , OutgoingErrorsV4(counters->GetCounter("DnsResolver/Outgoing/Errors/V4", true)) + , OutgoingErrorsV6(counters->GetCounter("DnsResolver/Outgoing/Errors/V6", true)) + , OutgoingTotalV4(counters->GetCounter("DnsResolver/Outgoing/Total/V4", true)) + , OutgoingTotalV6(counters->GetCounter("DnsResolver/Outgoing/Total/V6", true)) + , IncomingInFlight(counters->GetCounter("DnsResolver/Incoming/InFlight", false)) + , IncomingErrors(counters->GetCounter("DnsResolver/Incoming/Errors", true)) + , IncomingTotal(counters->GetCounter("DnsResolver/Incoming/Total", true)) + , CacheSize(counters->GetCounter("DnsResolver/Cache/Size", false)) + , CacheHits(counters->GetCounter("DnsResolver/Cache/Hits", true)) + , CacheMisses(counters->GetCounter("DnsResolver/Cache/Misses", true)) + { } + }; + + public: + TCachingDnsResolver(TActorId upstream, TCachingDnsResolverOptions options) + : TActor(&TThis::StateWork) + , Upstream(upstream) + , Options(std::move(options)) + , MonCounters(Options.MonCounters ? new TMonCounters(Options.MonCounters) : nullptr) + { } + + static constexpr EActivityType ActorActivityType() { + return DNS_RESOLVER; + } + + private: + STRICT_STFUNC(StateWork, { + hFunc(TEvents::TEvPoison, Handle); + hFunc(TEvDns::TEvGetHostByName, Handle); + hFunc(TEvDns::TEvGetAddr, Handle); + hFunc(TEvDns::TEvGetHostByNameResult, Handle); + hFunc(TEvents::TEvUndelivered, Handle); + }); + + void Handle(TEvents::TEvPoison::TPtr&) { + DropPending(ARES_ECANCELLED); + PassAway(); + } + + void Handle(TEvDns::TEvGetHostByName::TPtr& ev) { + auto req = MakeHolder<TIncomingRequest>(); + req->Type = EIncomingRequestType::GetHostByName; + req->Sender = ev->Sender; + req->Cookie = ev->Cookie; + req->Name = std::move(ev->Get()->Name); + req->Family = ev->Get()->Family; + EnqueueRequest(std::move(req)); + } + + void Handle(TEvDns::TEvGetAddr::TPtr& ev) { + auto req = MakeHolder<TIncomingRequest>(); + req->Type = EIncomingRequestType::GetAddr; + req->Sender = ev->Sender; + req->Cookie = ev->Cookie; + req->Name = std::move(ev->Get()->Name); + req->Family = ev->Get()->Family; + EnqueueRequest(std::move(req)); + } + + void Handle(TEvDns::TEvGetHostByNameResult::TPtr& ev) { + auto waitingIt = WaitingRequests.find(ev->Cookie); + Y_VERIFY(waitingIt != WaitingRequests.end(), "Unexpected reply, reqId=%" PRIu64, ev->Cookie); + auto waitingInfo = waitingIt->second; + WaitingRequests.erase(waitingIt); + + switch (waitingInfo.Family) { + case AF_INET6: + if (ev->Get()->Status) { + ProcessErrorV6(waitingInfo.Position, ev->Get()->Status, std::move(ev->Get()->ErrorText)); + } else { + ProcessAddrsV6(waitingInfo.Position, std::move(ev->Get()->AddrsV6)); + } + break; + + case AF_INET: + if (ev->Get()->Status) { + ProcessErrorV4(waitingInfo.Position, ev->Get()->Status, std::move(ev->Get()->ErrorText)); + } else { + ProcessAddrsV4(waitingInfo.Position, std::move(ev->Get()->AddrsV4)); + } + break; + + default: + Y_FAIL("Unexpected request family %d", waitingInfo.Family); + } + } + + void Handle(TEvents::TEvUndelivered::TPtr& ev) { + switch (ev->Get()->SourceType) { + case TEvDns::TEvGetHostByName::EventType: { + auto waitingIt = WaitingRequests.find(ev->Cookie); + Y_VERIFY(waitingIt != WaitingRequests.end(), "Unexpected TEvUndelivered, reqId=%" PRIu64, ev->Cookie); + auto waitingInfo = waitingIt->second; + WaitingRequests.erase(waitingIt); + + switch (waitingInfo.Family) { + case AF_INET6: + ProcessErrorV6(waitingInfo.Position, ARES_ENOTINITIALIZED, "Caching dns resolver cannot deliver to the underlying resolver"); + break; + case AF_INET: + ProcessErrorV4(waitingInfo.Position, ARES_ENOTINITIALIZED, "Caching dns resolver cannot deliver to the underlying resolver"); + break; + default: + Y_FAIL("Unexpected request family %d", waitingInfo.Family); + } + + break; + } + + default: + Y_FAIL("Unexpected TEvUndelievered, type=%" PRIu32, ev->Get()->SourceType); + } + } + + private: + enum EIncomingRequestType { + GetHostByName, + GetAddr, + }; + + struct TIncomingRequest : public TIntrusiveListItem<TIncomingRequest> { + EIncomingRequestType Type; + TActorId Sender; + ui64 Cookie; + TString Name; + int Family; + }; + + using TIncomingRequestList = TIntrusiveListWithAutoDelete<TIncomingRequest, TDelete>; + + void EnqueueRequest(THolder<TIncomingRequest> req) { + if (MonCounters) { + ++*MonCounters->IncomingTotal; + } + + CleanupExpired(TActivationContext::Now()); + + switch (req->Family) { + case AF_UNSPEC: + if (Options.AllowIPv6) { + EnqueueRequestIPv6(std::move(req)); + return; + } + if (Options.AllowIPv4) { + EnqueueRequestIPv4(std::move(req)); + return; + } + break; + + case AF_INET6: + if (Options.AllowIPv6) { + EnqueueRequestIPv6(std::move(req)); + return; + } + break; + + case AF_INET: + if (Options.AllowIPv4) { + EnqueueRequestIPv4(std::move(req)); + return; + } + break; + } + + ReplyWithError(std::move(req), ARES_EBADFAMILY); + } + + void EnqueueRequestIPv6(THolder<TIncomingRequest> req) { + auto now = TActivationContext::Now(); + + auto& fullState = NameToState[req->Name]; + if (MonCounters) { + *MonCounters->CacheSize = NameToState.size(); + } + + auto& state = fullState.StateIPv6; + EnsureRequest(state, req->Name, AF_INET6, now); + + if (state.IsHardExpired(now)) { + Y_VERIFY(state.Waiting); + if (MonCounters) { + ++*MonCounters->CacheMisses; + } + // We need to wait for ipv6 reply, schedule ipv4 request in parallel if needed + if (Options.AllowIPv4) { + EnsureRequest(fullState.StateIPv4, req->Name, AF_INET, now); + } + state.WaitingRequests.PushBack(req.Release()); + return; + } + + // We want to retry AF_UNSPEC with IPv4 in some cases + if (req->Family == AF_UNSPEC && Options.AllowIPv4 && state.RetryUnspec()) { + EnqueueRequestIPv4(std::move(req)); + return; + } + + if (MonCounters) { + ++*MonCounters->CacheHits; + } + + if (state.Status != 0) { + ReplyWithError(std::move(req), state.Status, state.ErrorText); + } else { + ReplyWithAddrs(std::move(req), fullState.AddrsIPv6); + } + } + + void EnqueueRequestIPv4(THolder<TIncomingRequest> req, bool isCacheMiss = false) { + auto now = TActivationContext::Now(); + + auto& fullState = NameToState[req->Name]; + if (MonCounters) { + *MonCounters->CacheSize = NameToState.size(); + } + + auto& state = fullState.StateIPv4; + EnsureRequest(state, req->Name, AF_INET, now); + + if (state.IsHardExpired(now)) { + Y_VERIFY(state.Waiting); + if (MonCounters && !isCacheMiss) { + ++*MonCounters->CacheMisses; + } + state.WaitingRequests.PushBack(req.Release()); + return; + } + + if (MonCounters && !isCacheMiss) { + ++*MonCounters->CacheHits; + } + + if (state.Status != 0) { + ReplyWithError(std::move(req), state.Status, state.ErrorText); + } else { + ReplyWithAddrs(std::move(req), fullState.AddrsIPv4); + } + } + + private: + struct TFamilyState { + TIncomingRequestList WaitingRequests; + TInstant SoftDeadline; + TInstant HardDeadline; + TInstant NextSoftDeadline; + TInstant NextHardDeadline; + TString ErrorText; + int Status = -1; // never requested before + bool InSoftHeap = false; + bool InHardHeap = false; + bool Waiting = false; + + bool Needed() const { + return InSoftHeap || InHardHeap || Waiting; + } + + bool RetryUnspec() const { + return ( + Status == ARES_ENODATA || + Status == ARES_EBADRESP || + Status == ARES_ETIMEOUT); + } + + bool ServerReplied() const { + return ServerReplied(Status); + } + + bool IsSoftExpired(TInstant now) const { + return !InSoftHeap || NextSoftDeadline < now; + } + + bool IsHardExpired(TInstant now) const { + return !InHardHeap || NextHardDeadline < now; + } + + static bool ServerReplied(int status) { + return ( + status == ARES_SUCCESS || + status == ARES_ENODATA || + status == ARES_ENOTFOUND); + } + }; + + struct TState { + TFamilyState StateIPv6; + TFamilyState StateIPv4; + TVector<struct in6_addr> AddrsIPv6; + TVector<struct in_addr> AddrsIPv4; + + bool Needed() const { + return StateIPv6.Needed() || StateIPv4.Needed(); + } + }; + + using TNameToState = THashMap<TString, TState>; + + template<const TFamilyState TState::* StateToFamily, + const TInstant TFamilyState::* FamilyToDeadline> + struct THeapCompare { + // returns true when b < a + bool operator()(TNameToState::iterator a, TNameToState::iterator b) const { + const TState& aState = a->second; + const TState& bState = b->second; + const TFamilyState& aFamily = aState.*StateToFamily; + const TFamilyState& bFamily = bState.*StateToFamily; + const TInstant& aDeadline = aFamily.*FamilyToDeadline; + const TInstant& bDeadline = bFamily.*FamilyToDeadline; + return bDeadline < aDeadline; + } + }; + + template<const TFamilyState TState::* StateToFamily, + const TInstant TFamilyState::* FamilyToDeadline> + using TStateHeap = std::priority_queue< + TNameToState::iterator, + std::vector<TNameToState::iterator>, + THeapCompare<StateToFamily, FamilyToDeadline> + >; + + struct TWaitingInfo { + TNameToState::iterator Position; + int Family; + }; + + private: + void EnsureRequest(TFamilyState& state, const TString& name, int family, TInstant now) { + if (state.Waiting) { + return; // request is already pending + } + + if (!state.IsSoftExpired(now) && !state.IsHardExpired(now)) { + return; // response is not expired yet + } + + if (MonCounters) { + switch (family) { + case AF_INET6: + ++*MonCounters->OutgoingInFlightV6; + ++*MonCounters->OutgoingTotalV6; + break; + case AF_INET: + ++*MonCounters->OutgoingInFlightV4; + ++*MonCounters->OutgoingTotalV4; + break; + } + } + + ui64 reqId = ++LastRequestId; + auto& req = WaitingRequests[reqId]; + req.Position = NameToState.find(name); + req.Family = family; + Y_VERIFY(req.Position != NameToState.end()); + + Send(Upstream, new TEvDns::TEvGetHostByName(name, family), IEventHandle::FlagTrackDelivery, reqId); + state.Waiting = true; + } + + template<TFamilyState TState::* StateToFamily, + TInstant TFamilyState::* FamilyToDeadline, + TInstant TFamilyState::* FamilyToNextDeadline, + bool TFamilyState::* FamilyToFlag, + class THeap> + void PushToHeap(THeap& heap, TNameToState::iterator it, TInstant newDeadline) { + auto& family = it->second.*StateToFamily; + TInstant& deadline = family.*FamilyToDeadline; + TInstant& nextDeadline = family.*FamilyToNextDeadline; + bool& flag = family.*FamilyToFlag; + nextDeadline = newDeadline; + if (!flag) { + deadline = newDeadline; + heap.push(it); + flag = true; + } + } + + void PushSoftV6(TNameToState::iterator it, TInstant newDeadline) { + PushToHeap<&TState::StateIPv6, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv6, it, newDeadline); + } + + void PushHardV6(TNameToState::iterator it, TInstant newDeadline) { + PushToHeap<&TState::StateIPv6, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv6, it, newDeadline); + } + + void PushSoftV4(TNameToState::iterator it, TInstant newDeadline) { + PushToHeap<&TState::StateIPv4, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv4, it, newDeadline); + } + + void PushHardV4(TNameToState::iterator it, TInstant newDeadline) { + PushToHeap<&TState::StateIPv4, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv4, it, newDeadline); + } + + void ProcessErrorV6(TNameToState::iterator it, int status, TString errorText) { + auto now = TActivationContext::Now(); + if (MonCounters) { + --*MonCounters->OutgoingInFlightV6; + ++*MonCounters->OutgoingErrorsV6; + } + + auto& state = it->second.StateIPv6; + Y_VERIFY(state.Waiting, "Got error for a state we are not waiting"); + state.Waiting = false; + + // When we have a cached positive reply, don't overwrite it with spurious errors + const bool serverReplied = TFamilyState::ServerReplied(status); + if (!serverReplied && state.ServerReplied() && !state.IsHardExpired(now)) { + PushSoftV6(it, now + Options.SoftNegativeExpireTime); + if (state.Status == ARES_SUCCESS) { + SendAddrsV6(it); + } else { + SendErrorsV6(it, now); + } + return; + } + + state.Status = status; + state.ErrorText = std::move(errorText); + PushSoftV6(it, now + Options.SoftNegativeExpireTime); + if (serverReplied) { + // Server actually replied, so keep it cached for longer + PushHardV6(it, now + Options.HardPositiveExpireTime); + } else { + PushHardV6(it, now + Options.HardNegativeExpireTime); + } + + SendErrorsV6(it, now); + } + + void SendErrorsV6(TNameToState::iterator it, TInstant now) { + bool cleaned = false; + auto& state = it->second.StateIPv6; + while (state.WaitingRequests) { + THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); + if (req->Family == AF_UNSPEC && Options.AllowIPv4 && state.RetryUnspec()) { + if (!cleaned) { + CleanupExpired(now); + cleaned = true; + } + EnqueueRequestIPv4(std::move(req), /* isCacheMiss */ true); + } else { + ReplyWithError(std::move(req), state.Status, state.ErrorText); + } + } + } + + void ProcessErrorV4(TNameToState::iterator it, int status, TString errorText) { + auto now = TActivationContext::Now(); + if (MonCounters) { + --*MonCounters->OutgoingInFlightV4; + ++*MonCounters->OutgoingErrorsV4; + } + + auto& state = it->second.StateIPv4; + Y_VERIFY(state.Waiting, "Got error for a state we are not waiting"); + state.Waiting = false; + + // When we have a cached positive reply, don't overwrite it with spurious errors + const bool serverReplied = TFamilyState::ServerReplied(status); + if (!serverReplied && state.ServerReplied() && !state.IsHardExpired(now)) { + PushSoftV4(it, now + Options.SoftNegativeExpireTime); + if (state.Status == ARES_SUCCESS) { + SendAddrsV4(it); + } else { + SendErrorsV4(it); + } + return; + } + + state.Status = status; + state.ErrorText = std::move(errorText); + PushSoftV4(it, now + Options.SoftNegativeExpireTime); + if (serverReplied) { + // Server actually replied, so keep it cached for longer + PushHardV4(it, now + Options.HardPositiveExpireTime); + } else { + PushHardV4(it, now + Options.HardNegativeExpireTime); + } + + SendErrorsV4(it); + } + + void SendErrorsV4(TNameToState::iterator it) { + auto& state = it->second.StateIPv4; + while (state.WaitingRequests) { + THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); + ReplyWithError(std::move(req), state.Status, state.ErrorText); + } + } + + void ProcessAddrsV6(TNameToState::iterator it, TVector<struct in6_addr> addrs) { + if (Y_UNLIKELY(addrs.empty())) { + // Probably unnecessary: we don't want to deal with empty address lists + return ProcessErrorV6(it, ARES_ENODATA, ares_strerror(ARES_ENODATA)); + } + + auto now = TActivationContext::Now(); + if (MonCounters) { + --*MonCounters->OutgoingInFlightV6; + } + + auto& state = it->second.StateIPv6; + Y_VERIFY(state.Waiting, "Got reply for a state we are not waiting"); + state.Waiting = false; + + state.Status = ARES_SUCCESS; + it->second.AddrsIPv6 = std::move(addrs); + PushSoftV6(it, now + Options.SoftPositiveExpireTime); + PushHardV6(it, now + Options.HardPositiveExpireTime); + + SendAddrsV6(it); + } + + void SendAddrsV6(TNameToState::iterator it) { + auto& state = it->second.StateIPv6; + while (state.WaitingRequests) { + THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); + ReplyWithAddrs(std::move(req), it->second.AddrsIPv6); + } + } + + void ProcessAddrsV4(TNameToState::iterator it, TVector<struct in_addr> addrs) { + if (Y_UNLIKELY(addrs.empty())) { + // Probably unnecessary: we don't want to deal with empty address lists + return ProcessErrorV4(it, ARES_ENODATA, ares_strerror(ARES_ENODATA)); + } + + auto now = TActivationContext::Now(); + if (MonCounters) { + --*MonCounters->OutgoingInFlightV4; + } + + auto& state = it->second.StateIPv4; + Y_VERIFY(state.Waiting, "Got reply for a state we are not waiting"); + state.Waiting = false; + + state.Status = ARES_SUCCESS; + it->second.AddrsIPv4 = std::move(addrs); + PushSoftV4(it, now + Options.SoftPositiveExpireTime); + PushHardV4(it, now + Options.HardPositiveExpireTime); + + SendAddrsV4(it); + } + + void SendAddrsV4(TNameToState::iterator it) { + auto& state = it->second.StateIPv4; + while (state.WaitingRequests) { + THolder<TIncomingRequest> req(state.WaitingRequests.PopFront()); + ReplyWithAddrs(std::move(req), it->second.AddrsIPv4); + } + } + + private: + template<TFamilyState TState::*StateToFamily, + TInstant TFamilyState::* FamilyToDeadline, + TInstant TFamilyState::* FamilyToNextDeadline, + bool TFamilyState::* FamilyToFlag> + void DoCleanupExpired(TStateHeap<StateToFamily, FamilyToDeadline>& heap, TInstant now) { + while (!heap.empty()) { + auto it = heap.top(); + auto& family = it->second.*StateToFamily; + TInstant& deadline = family.*FamilyToDeadline; + if (now <= deadline) { + break; + } + + bool& flag = family.*FamilyToFlag; + Y_VERIFY(flag); + heap.pop(); + flag = false; + + TInstant& nextDeadline = family.*FamilyToNextDeadline; + if (now < nextDeadline) { + deadline = nextDeadline; + heap.push(it); + flag = true; + continue; + } + + // Remove unnecessary items + if (!it->second.Needed()) { + NameToState.erase(it); + if (MonCounters) { + *MonCounters->CacheSize = NameToState.size(); + } + } + } + } + + void CleanupExpired(TInstant now) { + DoCleanupExpired<&TState::StateIPv6, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv6, now); + DoCleanupExpired<&TState::StateIPv6, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv6, now); + DoCleanupExpired<&TState::StateIPv4, &TFamilyState::SoftDeadline, &TFamilyState::NextSoftDeadline, &TFamilyState::InSoftHeap>(SoftHeapIPv4, now); + DoCleanupExpired<&TState::StateIPv4, &TFamilyState::HardDeadline, &TFamilyState::NextHardDeadline, &TFamilyState::InHardHeap>(HardHeapIPv4, now); + } + + template<class TEvent> + void SendError(TActorId replyTo, ui64 cookie, int status, const TString& errorText) { + auto reply = MakeHolder<TEvent>(); + reply->Status = status; + reply->ErrorText = errorText; + this->Send(replyTo, reply.Release(), 0, cookie); + } + + void ReplyWithError(THolder<TIncomingRequest> req, int status, const TString& errorText) { + if (MonCounters) { + ++*MonCounters->IncomingErrors; + } + switch (req->Type) { + case EIncomingRequestType::GetHostByName: { + SendError<TEvDns::TEvGetHostByNameResult>(req->Sender, req->Cookie, status, errorText); + break; + } + case EIncomingRequestType::GetAddr: { + SendError<TEvDns::TEvGetAddrResult>(req->Sender, req->Cookie, status, errorText); + break; + } + } + } + + void ReplyWithAddrs(THolder<TIncomingRequest> req, const TVector<struct in6_addr>& addrs) { + switch (req->Type) { + case EIncomingRequestType::GetHostByName: { + auto reply = MakeHolder<TEvDns::TEvGetHostByNameResult>(); + reply->AddrsV6 = addrs; + Send(req->Sender, reply.Release(), 0, req->Cookie); + break; + } + case EIncomingRequestType::GetAddr: { + Y_VERIFY(!addrs.empty()); + auto reply = MakeHolder<TEvDns::TEvGetAddrResult>(); + reply->Addr = addrs.front(); + Send(req->Sender, reply.Release(), 0, req->Cookie); + break; + } + } + } + + void ReplyWithAddrs(THolder<TIncomingRequest> req, const TVector<struct in_addr>& addrs) { + switch (req->Type) { + case EIncomingRequestType::GetHostByName: { + auto reply = MakeHolder<TEvDns::TEvGetHostByNameResult>(); + reply->AddrsV4 = addrs; + Send(req->Sender, reply.Release(), 0, req->Cookie); + break; + } + case EIncomingRequestType::GetAddr: { + Y_VERIFY(!addrs.empty()); + auto reply = MakeHolder<TEvDns::TEvGetAddrResult>(); + reply->Addr = addrs.front(); + Send(req->Sender, reply.Release(), 0, req->Cookie); + break; + } + } + } + + void ReplyWithError(THolder<TIncomingRequest> req, int status) { + ReplyWithError(std::move(req), status, ares_strerror(status)); + } + + void DropPending(TIncomingRequestList& list, int status, const TString& errorText) { + while (list) { + THolder<TIncomingRequest> req(list.PopFront()); + ReplyWithError(std::move(req), status, errorText); + } + } + + void DropPending(int status, const TString& errorText) { + for (auto& [name, state] : NameToState) { + DropPending(state.StateIPv6.WaitingRequests, status, errorText); + DropPending(state.StateIPv4.WaitingRequests, status, errorText); + } + } + + void DropPending(int status) { + DropPending(status, ares_strerror(status)); + } + + private: + const TActorId Upstream; + const TCachingDnsResolverOptions Options; + const THolder<TMonCounters> MonCounters; + + TNameToState NameToState; + TStateHeap<&TState::StateIPv6, &TFamilyState::SoftDeadline> SoftHeapIPv6; + TStateHeap<&TState::StateIPv6, &TFamilyState::HardDeadline> HardHeapIPv6; + TStateHeap<&TState::StateIPv4, &TFamilyState::SoftDeadline> SoftHeapIPv4; + TStateHeap<&TState::StateIPv4, &TFamilyState::HardDeadline> HardHeapIPv4; + + THashMap<ui64, TWaitingInfo> WaitingRequests; + ui64 LastRequestId = 0; + }; + + IActor* CreateCachingDnsResolver(TActorId upstream, TCachingDnsResolverOptions options) { + return new TCachingDnsResolver(upstream, std::move(options)); + } + +} // namespace NDnsResolver +} // namespace NActors diff --git a/library/cpp/actors/dnsresolver/dnsresolver_caching_ut.cpp b/library/cpp/actors/dnsresolver/dnsresolver_caching_ut.cpp index f63e594d30..c3b7cb3c77 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver_caching_ut.cpp +++ b/library/cpp/actors/dnsresolver/dnsresolver_caching_ut.cpp @@ -1,630 +1,630 @@ -#include "dnsresolver.h" - -#include <library/cpp/actors/core/hfunc.h> -#include <library/cpp/actors/testlib/test_runtime.h> -#include <library/cpp/testing/unittest/registar.h> -#include <util/string/builder.h> - -#include <ares.h> - -using namespace NActors; -using namespace NActors::NDnsResolver; - -// FIXME: use a mock resolver -Y_UNIT_TEST_SUITE(CachingDnsResolver) { - - struct TAddrToString { - TString operator()(const std::monostate&) const { - return "<nothing>"; - } - - TString operator()(const struct in6_addr& addr) const { - char dst[INET6_ADDRSTRLEN]; - auto res = ares_inet_ntop(AF_INET6, &addr, dst, INET6_ADDRSTRLEN); - Y_VERIFY(res, "Cannot convert ipv6 address"); - return dst; - } - - TString operator()(const struct in_addr& addr) const { - char dst[INET_ADDRSTRLEN]; - auto res = ares_inet_ntop(AF_INET, &addr, dst, INET_ADDRSTRLEN); - Y_VERIFY(res, "Cannot convert ipv4 address"); - return dst; - } - }; - - TString AddrToString(const std::variant<std::monostate, struct in6_addr, struct in_addr>& v) { - return std::visit(TAddrToString(), v); - } - - struct TMockReply { - static constexpr TDuration DefaultDelay = TDuration::MilliSeconds(1); - - int Status = 0; - TDuration Delay; - TVector<struct in6_addr> AddrsV6; - TVector<struct in_addr> AddrsV4; - - static TMockReply Error(int status, TDuration delay = DefaultDelay) { - Y_VERIFY(status != 0); - TMockReply reply; - reply.Status = status; - reply.Delay = delay; - return reply; - } - - static TMockReply Empty(TDuration delay = DefaultDelay) { - TMockReply reply; - reply.Delay = delay; - return reply; - } - - static TMockReply ManyV6(const TVector<TString>& addrs, TDuration delay = DefaultDelay) { - TMockReply reply; - reply.Delay = delay; - for (const TString& addr : addrs) { - void* dst = &reply.AddrsV6.emplace_back(); - int status = ares_inet_pton(AF_INET6, addr.c_str(), dst); - Y_VERIFY(status == 1, "Invalid ipv6 address: %s", addr.c_str()); - } - return reply; - } - - static TMockReply ManyV4(const TVector<TString>& addrs, TDuration delay = DefaultDelay) { - TMockReply reply; - reply.Delay = delay; - for (const TString& addr : addrs) { - void* dst = &reply.AddrsV4.emplace_back(); - int status = ares_inet_pton(AF_INET, addr.c_str(), dst); - Y_VERIFY(status == 1, "Invalid ipv4 address: %s", addr.c_str()); - } - return reply; - } - - static TMockReply SingleV6(const TString& addr, TDuration delay = DefaultDelay) { - return ManyV6({ addr }, delay); - } - - static TMockReply SingleV4(const TString& addr, TDuration delay = DefaultDelay) { - return ManyV4({ addr }, delay); - } - }; - - using TMockDnsCallback = std::function<TMockReply (const TString&, int)>; - - class TMockDnsResolver : public TActor<TMockDnsResolver> { - public: - TMockDnsResolver(TMockDnsCallback callback) - : TActor(&TThis::StateWork) - , Callback(std::move(callback)) - { } - - private: - struct TEvPrivate { - enum EEv { - EvScheduled = EventSpaceBegin(TEvents::ES_PRIVATE), - }; - - struct TEvScheduled : public TEventLocal<TEvScheduled, EvScheduled> { - TActorId Sender; - ui64 Cookie; - TMockReply Reply; - - TEvScheduled(TActorId sender, ui64 cookie, TMockReply reply) - : Sender(sender) - , Cookie(cookie) - , Reply(std::move(reply)) - { } - }; - }; - - private: - STRICT_STFUNC(StateWork, { - hFunc(TEvents::TEvPoison, Handle); - hFunc(TEvDns::TEvGetHostByName, Handle); - hFunc(TEvPrivate::TEvScheduled, Handle); - }); - - void Handle(TEvents::TEvPoison::TPtr&) { - PassAway(); - } - - void Handle(TEvDns::TEvGetHostByName::TPtr& ev) { - auto reply = Callback(ev->Get()->Name, ev->Get()->Family); - if (reply.Delay) { - Schedule(reply.Delay, new TEvPrivate::TEvScheduled(ev->Sender, ev->Cookie, std::move(reply))); - } else { - SendReply(ev->Sender, ev->Cookie, std::move(reply)); - } - } - - void Handle(TEvPrivate::TEvScheduled::TPtr& ev) { - SendReply(ev->Get()->Sender, ev->Get()->Cookie, std::move(ev->Get()->Reply)); - } - - private: - void SendReply(const TActorId& sender, ui64 cookie, TMockReply&& reply) { - auto res = MakeHolder<TEvDns::TEvGetHostByNameResult>(); - res->Status = reply.Status; - if (res->Status != 0) { - res->ErrorText = ares_strerror(res->Status); - } else { - res->AddrsV6 = std::move(reply.AddrsV6); - res->AddrsV4 = std::move(reply.AddrsV4); - } - Send(sender, res.Release(), 0, cookie); - } - - private: - TMockDnsCallback Callback; - }; - - struct TCachingDnsRuntime : public TTestActorRuntimeBase { - TCachingDnsResolverOptions ResolverOptions; - TActorId MockResolver; - TActorId Resolver; - TActorId Sleeper; - TString Section_; - - NMonitoring::TDynamicCounters::TCounterPtr InFlight6; - NMonitoring::TDynamicCounters::TCounterPtr InFlight4; - NMonitoring::TDynamicCounters::TCounterPtr Total6; - NMonitoring::TDynamicCounters::TCounterPtr Total4; - NMonitoring::TDynamicCounters::TCounterPtr Misses; - NMonitoring::TDynamicCounters::TCounterPtr Hits; - - THashMap<TString, TMockReply> ReplyV6; - THashMap<TString, TMockReply> ReplyV4; - - TCachingDnsRuntime() { - SetScheduledEventFilter([](auto&&, auto&&, auto&&, auto&&) { return false; }); - ResolverOptions.MonCounters = new NMonitoring::TDynamicCounters; - - ReplyV6["localhost"] = TMockReply::SingleV6("::1"); - ReplyV4["localhost"] = TMockReply::SingleV4("127.0.0.1"); - ReplyV6["yandex.ru"] = TMockReply::SingleV6("2a02:6b8:a::a", TDuration::MilliSeconds(500)); - ReplyV4["yandex.ru"] = TMockReply::SingleV4("77.88.55.77", TDuration::MilliSeconds(250)); - ReplyV6["router.asus.com"] = TMockReply::Error(ARES_ENODATA); - ReplyV4["router.asus.com"] = TMockReply::SingleV4("192.168.0.1"); - } - - void Start(TMockDnsCallback callback) { - MockResolver = Register(new TMockDnsResolver(std::move(callback))); - EnableScheduleForActor(MockResolver); - Resolver = Register(CreateCachingDnsResolver(MockResolver, ResolverOptions)); - Sleeper = AllocateEdgeActor(); - - InFlight6 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/InFlight/V6", false); - InFlight4 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/InFlight/V4", false); - Total6 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/Total/V6", true); - Total4 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/Total/V4", true); - Misses = ResolverOptions.MonCounters->GetCounter("DnsResolver/Cache/Misses", true); - Hits = ResolverOptions.MonCounters->GetCounter("DnsResolver/Cache/Hits", true); - } - - void Start() { - Start([this](const TString& name, int family) { - switch (family) { - case AF_INET6: { - auto it = ReplyV6.find(name); - if (it != ReplyV6.end()) { - return it->second; - } - break; - } - case AF_INET: { - auto it = ReplyV4.find(name); - if (it != ReplyV4.end()) { - return it->second; - } - break; - } - } - return TMockReply::Error(ARES_ENOTFOUND); - }); - } - - void Section(const TString& section) { - Section_ = section; - } - - void Sleep(TDuration duration) { - Schedule(new IEventHandle(Sleeper, Sleeper, new TEvents::TEvWakeup), duration); - GrabEdgeEventRethrow<TEvents::TEvWakeup>(Sleeper); - } - - void WaitNoInFlight() { - if (*InFlight6 || *InFlight4) { - TDispatchOptions options; - options.CustomFinalCondition = [&]() { - return !*InFlight6 && !*InFlight4; - }; - DispatchEvents(options); - UNIT_ASSERT_C(!*InFlight6 && !*InFlight4, "Failed to wait for no inflight in " << Section_); - } - } - - void SendGetHostByName(const TActorId& sender, const TString& name, int family = AF_UNSPEC) { - Send(new IEventHandle(Resolver, sender, new TEvDns::TEvGetHostByName(name, family)), 0, true); - } - - void SendGetAddr(const TActorId& sender, const TString& name, int family = AF_UNSPEC) { - Send(new IEventHandle(Resolver, sender, new TEvDns::TEvGetAddr(name, family)), 0, true); - } - - TEvDns::TEvGetHostByNameResult::TPtr WaitGetHostByName(const TActorId& sender) { - return GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); - } - - TEvDns::TEvGetAddrResult::TPtr WaitGetAddr(const TActorId& sender) { - return GrabEdgeEventRethrow<TEvDns::TEvGetAddrResult>(sender); - } - - void ExpectInFlight6(i64 count) { - UNIT_ASSERT_VALUES_EQUAL_C(InFlight6->Val(), count, Section_); - } - - void ExpectInFlight4(i64 count) { - UNIT_ASSERT_VALUES_EQUAL_C(InFlight4->Val(), count, Section_); - } - - void ExpectTotal6(i64 count) { - UNIT_ASSERT_VALUES_EQUAL_C(Total6->Val(), count, Section_); - } - - void ExpectTotal4(i64 count) { - UNIT_ASSERT_VALUES_EQUAL_C(Total4->Val(), count, Section_); - } - - void Expect6(i64 total, i64 inflight) { - UNIT_ASSERT_C( - Total6->Val() == total && InFlight6->Val() == inflight, - Section_ << ": Expect6(" << total << ", " << inflight << ") " - << " but got (" << Total6->Val() << ", " << InFlight6->Val() << ")"); - } - - void Expect4(i64 total, i64 inflight) { - UNIT_ASSERT_C( - Total4->Val() == total && InFlight4->Val() == inflight, - Section_ << ": Expect4(" << total << ", " << inflight << ") " - << " got (" << Total4->Val() << ", " << InFlight4->Val() << ")"); - } - - void ExpectMisses(i64 count) { - UNIT_ASSERT_VALUES_EQUAL_C(Misses->Val(), count, Section_); - } - - void ExpectHits(i64 count) { - UNIT_ASSERT_VALUES_EQUAL_C(Hits->Val(), count, Section_); - } - - void ExpectGetHostByNameError(const TActorId& sender, int status) { - auto ev = WaitGetHostByName(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, status, Section_ << ": " << ev->Get()->ErrorText); - } - - void ExpectGetAddrError(const TActorId& sender, int status) { - auto ev = WaitGetAddr(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, status, Section_ << ": " << ev->Get()->ErrorText); - } - - void ExpectGetHostByNameSuccess(const TActorId& sender, const TString& expected) { - auto ev = WaitGetHostByName(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, Section_ << ": " << ev->Get()->ErrorText); - TStringBuilder result; - for (const auto& addr : ev->Get()->AddrsV6) { - if (result) { - result << ','; - } - result << TAddrToString()(addr); - } - for (const auto& addr : ev->Get()->AddrsV4) { - if (result) { - result << ','; - } - result << TAddrToString()(addr); - } - UNIT_ASSERT_VALUES_EQUAL_C(TString(result), expected, Section_); - } - - void ExpectGetAddrSuccess(const TActorId& sender, const TString& expected) { - auto ev = WaitGetAddr(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, Section_ << ": " << ev->Get()->ErrorText); - TString result = AddrToString(ev->Get()->Addr); - UNIT_ASSERT_VALUES_EQUAL_C(result, expected, Section_); - } - }; - - Y_UNIT_TEST(UnusableResolver) { - TCachingDnsRuntime runtime; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - - runtime.Send(new IEventHandle(runtime.MockResolver, { }, new TEvents::TEvPoison), 0, true); - runtime.SendGetAddr(sender, "foo.ru", AF_UNSPEC); - runtime.ExpectGetAddrError(sender, ARES_ENOTINITIALIZED); - } - - Y_UNIT_TEST(ResolveCaching) { - TCachingDnsRuntime runtime; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - // First time resolve, ipv4 and ipv6 sent in parallel, we wait for ipv6 result - runtime.Section("First time resolve"); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.Expect6(1, 0); - runtime.Expect4(1, 0); - runtime.ExpectMisses(1); - runtime.ExpectHits(0); - - // Second resolve, ipv6 and ipv4 queries result in a cache hit - runtime.Section("Second resolve, ipv6"); - runtime.SendGetAddr(sender, "yandex.ru", AF_INET6); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.Expect6(1, 0); - runtime.ExpectHits(1); - runtime.Section("Second resolve, ipv4"); - runtime.SendGetAddr(sender, "yandex.ru", AF_INET); - runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); - runtime.Expect4(1, 0); - runtime.ExpectHits(2); - - // Wait until soft expiration and try ipv4 again - // Will cause a cache hit, but will start a new ipv4 request in background - runtime.Section("Retry ipv4 after soft expiration"); - runtime.Sleep(TDuration::Seconds(15)); - runtime.SendGetAddr(sender, "yandex.ru", AF_INET); - runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); - runtime.Expect6(1, 0); - runtime.Expect4(2, 1); - runtime.ExpectMisses(1); - runtime.ExpectHits(3); - runtime.WaitNoInFlight(); - - // Wait until soft expiration and try both again - // Will cause a cache hit, but will start a new ipv6 request in background - runtime.Section("Retry both after soft expiration"); - runtime.Sleep(TDuration::Seconds(15)); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.Expect6(2, 1); - runtime.Expect4(2, 0); - runtime.ExpectMisses(1); - runtime.ExpectHits(4); - runtime.WaitNoInFlight(); - - // Wait until hard expiration and try both again - // Will cause a cache miss and new resolve requests - runtime.Section("Retry both after hard expiration"); - runtime.Sleep(TDuration::Hours(2)); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.Expect6(3, 0); - runtime.Expect4(3, 0); - runtime.ExpectMisses(2); - runtime.ExpectHits(4); - - // Wait half the hard expiration time, must always result in a cache hit - runtime.Section("Retry both after half hard expiration"); - for (ui64 i = 1; i <= 4; ++i) { - runtime.Sleep(TDuration::Hours(1)); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.Expect6(3 + i, 1); - runtime.ExpectHits(4 + i); - runtime.WaitNoInFlight(); - } - - // Change v6 result to a timeout, must keep using cached result until hard expiration - runtime.Section("Dns keeps timing out"); - runtime.ReplyV6["yandex.ru"] = TMockReply::Error(ARES_ETIMEOUT); - for (ui64 i = 1; i <= 4; ++i) { - runtime.Sleep(TDuration::Seconds(15)); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.Expect6(7 + i, 1); - runtime.ExpectHits(8 + i); - runtime.WaitNoInFlight(); - } - - // Change v6 result to nodata, must switch to a v4 result eventually - runtime.Section("Host changes to being ipv4 only"); - runtime.ReplyV6["yandex.ru"] = TMockReply::Error(ARES_ENODATA); - runtime.Sleep(TDuration::Seconds(2)); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - runtime.WaitNoInFlight(); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); - runtime.Expect6(12, 0); - runtime.Expect4(4, 0); - runtime.ExpectMisses(3); - - // Change v6 result to nxdomain, must not fall back to a v4 result - runtime.Section("Host is removed from dns"); - runtime.ReplyV6["yandex.ru"] = TMockReply::Error(ARES_ENOTFOUND); - runtime.Sleep(TDuration::Seconds(15)); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); - runtime.WaitNoInFlight(); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - } - - Y_UNIT_TEST(ResolveCachingV4) { - TCachingDnsRuntime runtime; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.Section("First request"); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); - runtime.ExpectMisses(1); - - runtime.Section("Second request"); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); - runtime.ExpectHits(1); - - runtime.Section("Dns keeps timing out"); - runtime.ReplyV6["router.asus.com"] = TMockReply::Error(ARES_ETIMEOUT); - runtime.ReplyV4["router.asus.com"] = TMockReply::Error(ARES_ETIMEOUT); - for (ui64 i = 1; i <= 4; ++i) { - runtime.Sleep(TDuration::Seconds(15)); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); - runtime.Expect6(1 + i, 1); - runtime.Expect4(1 + i, 1); - runtime.ExpectHits(1 + i); - runtime.WaitNoInFlight(); - } - - runtime.Section("Host is removed from ipv4 dns"); - runtime.ReplyV4["router.asus.com"] = TMockReply::Error(ARES_ENOTFOUND); - runtime.Sleep(TDuration::Seconds(15)); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); - runtime.WaitNoInFlight(); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - } - - Y_UNIT_TEST(EventualTimeout) { - TCachingDnsRuntime runtime; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.ReplyV6["notfound.ru"] = TMockReply::Error(ARES_ENODATA); - runtime.ReplyV4["notfound.ru"] = TMockReply::Error(ARES_ENOTFOUND); - runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - - runtime.ReplyV4["notfound.ru"] = TMockReply::Error(ARES_ETIMEOUT); - runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - runtime.WaitNoInFlight(); - - bool timeout = false; - for (ui64 i = 1; i <= 8; ++i) { - runtime.Sleep(TDuration::Minutes(30)); - runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); - auto ev = runtime.WaitGetAddr(sender); - if (ev->Get()->Status == ARES_ETIMEOUT && i > 2) { - timeout = true; - break; - } - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, ARES_ENOTFOUND, - "Iteration " << i << ": " << ev->Get()->ErrorText); - } - - UNIT_ASSERT_C(timeout, "DnsResolver did not reply with a timeout"); - } - - Y_UNIT_TEST(MultipleRequestsAndHosts) { - TCachingDnsRuntime runtime; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.SendGetHostByName(sender, "router.asus.com", AF_UNSPEC); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetHostByNameSuccess(sender, "192.168.0.1"); - runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); - runtime.ExpectGetHostByNameSuccess(sender, "2a02:6b8:a::a"); - runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); - - runtime.SendGetHostByName(sender, "notfound.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); - runtime.ExpectGetHostByNameError(sender, ARES_ENOTFOUND); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - } - - Y_UNIT_TEST(DisabledIPv6) { - TCachingDnsRuntime runtime; - runtime.ResolverOptions.AllowIPv6 = false; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetHostByNameSuccess(sender, "77.88.55.77"); - runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); - - runtime.SendGetHostByName(sender, "yandex.ru", AF_INET6); - runtime.SendGetAddr(sender, "yandex.ru", AF_INET6); - runtime.ExpectGetHostByNameError(sender, ARES_EBADFAMILY); - runtime.ExpectGetAddrError(sender, ARES_EBADFAMILY); - - runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.ExpectGetHostByNameSuccess(sender, "77.88.55.77"); - runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); - - runtime.SendGetHostByName(sender, "notfound.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); - runtime.ExpectGetHostByNameError(sender, ARES_ENOTFOUND); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - } - - Y_UNIT_TEST(DisabledIPv4) { - TCachingDnsRuntime runtime; - runtime.ResolverOptions.AllowIPv4 = false; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.SendGetHostByName(sender, "router.asus.com", AF_UNSPEC); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetHostByNameError(sender, ARES_ENODATA); - runtime.ExpectGetAddrError(sender, ARES_ENODATA); - - runtime.SendGetHostByName(sender, "router.asus.com", AF_INET); - runtime.SendGetAddr(sender, "router.asus.com", AF_INET); - runtime.ExpectGetHostByNameError(sender, ARES_EBADFAMILY); - runtime.ExpectGetAddrError(sender, ARES_EBADFAMILY); - - runtime.SendGetHostByName(sender, "router.asus.com", AF_UNSPEC); - runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); - runtime.ExpectGetHostByNameError(sender, ARES_ENODATA); - runtime.ExpectGetAddrError(sender, ARES_ENODATA); - - runtime.SendGetHostByName(sender, "notfound.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); - runtime.ExpectGetHostByNameError(sender, ARES_ENOTFOUND); - runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); - } - - Y_UNIT_TEST(PoisonPill) { - TCachingDnsRuntime runtime; - runtime.Initialize(); - runtime.Start(); - - auto sender = runtime.AllocateEdgeActor(); - - runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); - runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); - runtime.Send(new IEventHandle(runtime.Resolver, sender, new TEvents::TEvPoison), 0, true); - runtime.ExpectGetHostByNameError(sender, ARES_ECANCELLED); - runtime.ExpectGetAddrError(sender, ARES_ECANCELLED); - } - -} +#include "dnsresolver.h" + +#include <library/cpp/actors/core/hfunc.h> +#include <library/cpp/actors/testlib/test_runtime.h> +#include <library/cpp/testing/unittest/registar.h> +#include <util/string/builder.h> + +#include <ares.h> + +using namespace NActors; +using namespace NActors::NDnsResolver; + +// FIXME: use a mock resolver +Y_UNIT_TEST_SUITE(CachingDnsResolver) { + + struct TAddrToString { + TString operator()(const std::monostate&) const { + return "<nothing>"; + } + + TString operator()(const struct in6_addr& addr) const { + char dst[INET6_ADDRSTRLEN]; + auto res = ares_inet_ntop(AF_INET6, &addr, dst, INET6_ADDRSTRLEN); + Y_VERIFY(res, "Cannot convert ipv6 address"); + return dst; + } + + TString operator()(const struct in_addr& addr) const { + char dst[INET_ADDRSTRLEN]; + auto res = ares_inet_ntop(AF_INET, &addr, dst, INET_ADDRSTRLEN); + Y_VERIFY(res, "Cannot convert ipv4 address"); + return dst; + } + }; + + TString AddrToString(const std::variant<std::monostate, struct in6_addr, struct in_addr>& v) { + return std::visit(TAddrToString(), v); + } + + struct TMockReply { + static constexpr TDuration DefaultDelay = TDuration::MilliSeconds(1); + + int Status = 0; + TDuration Delay; + TVector<struct in6_addr> AddrsV6; + TVector<struct in_addr> AddrsV4; + + static TMockReply Error(int status, TDuration delay = DefaultDelay) { + Y_VERIFY(status != 0); + TMockReply reply; + reply.Status = status; + reply.Delay = delay; + return reply; + } + + static TMockReply Empty(TDuration delay = DefaultDelay) { + TMockReply reply; + reply.Delay = delay; + return reply; + } + + static TMockReply ManyV6(const TVector<TString>& addrs, TDuration delay = DefaultDelay) { + TMockReply reply; + reply.Delay = delay; + for (const TString& addr : addrs) { + void* dst = &reply.AddrsV6.emplace_back(); + int status = ares_inet_pton(AF_INET6, addr.c_str(), dst); + Y_VERIFY(status == 1, "Invalid ipv6 address: %s", addr.c_str()); + } + return reply; + } + + static TMockReply ManyV4(const TVector<TString>& addrs, TDuration delay = DefaultDelay) { + TMockReply reply; + reply.Delay = delay; + for (const TString& addr : addrs) { + void* dst = &reply.AddrsV4.emplace_back(); + int status = ares_inet_pton(AF_INET, addr.c_str(), dst); + Y_VERIFY(status == 1, "Invalid ipv4 address: %s", addr.c_str()); + } + return reply; + } + + static TMockReply SingleV6(const TString& addr, TDuration delay = DefaultDelay) { + return ManyV6({ addr }, delay); + } + + static TMockReply SingleV4(const TString& addr, TDuration delay = DefaultDelay) { + return ManyV4({ addr }, delay); + } + }; + + using TMockDnsCallback = std::function<TMockReply (const TString&, int)>; + + class TMockDnsResolver : public TActor<TMockDnsResolver> { + public: + TMockDnsResolver(TMockDnsCallback callback) + : TActor(&TThis::StateWork) + , Callback(std::move(callback)) + { } + + private: + struct TEvPrivate { + enum EEv { + EvScheduled = EventSpaceBegin(TEvents::ES_PRIVATE), + }; + + struct TEvScheduled : public TEventLocal<TEvScheduled, EvScheduled> { + TActorId Sender; + ui64 Cookie; + TMockReply Reply; + + TEvScheduled(TActorId sender, ui64 cookie, TMockReply reply) + : Sender(sender) + , Cookie(cookie) + , Reply(std::move(reply)) + { } + }; + }; + + private: + STRICT_STFUNC(StateWork, { + hFunc(TEvents::TEvPoison, Handle); + hFunc(TEvDns::TEvGetHostByName, Handle); + hFunc(TEvPrivate::TEvScheduled, Handle); + }); + + void Handle(TEvents::TEvPoison::TPtr&) { + PassAway(); + } + + void Handle(TEvDns::TEvGetHostByName::TPtr& ev) { + auto reply = Callback(ev->Get()->Name, ev->Get()->Family); + if (reply.Delay) { + Schedule(reply.Delay, new TEvPrivate::TEvScheduled(ev->Sender, ev->Cookie, std::move(reply))); + } else { + SendReply(ev->Sender, ev->Cookie, std::move(reply)); + } + } + + void Handle(TEvPrivate::TEvScheduled::TPtr& ev) { + SendReply(ev->Get()->Sender, ev->Get()->Cookie, std::move(ev->Get()->Reply)); + } + + private: + void SendReply(const TActorId& sender, ui64 cookie, TMockReply&& reply) { + auto res = MakeHolder<TEvDns::TEvGetHostByNameResult>(); + res->Status = reply.Status; + if (res->Status != 0) { + res->ErrorText = ares_strerror(res->Status); + } else { + res->AddrsV6 = std::move(reply.AddrsV6); + res->AddrsV4 = std::move(reply.AddrsV4); + } + Send(sender, res.Release(), 0, cookie); + } + + private: + TMockDnsCallback Callback; + }; + + struct TCachingDnsRuntime : public TTestActorRuntimeBase { + TCachingDnsResolverOptions ResolverOptions; + TActorId MockResolver; + TActorId Resolver; + TActorId Sleeper; + TString Section_; + + NMonitoring::TDynamicCounters::TCounterPtr InFlight6; + NMonitoring::TDynamicCounters::TCounterPtr InFlight4; + NMonitoring::TDynamicCounters::TCounterPtr Total6; + NMonitoring::TDynamicCounters::TCounterPtr Total4; + NMonitoring::TDynamicCounters::TCounterPtr Misses; + NMonitoring::TDynamicCounters::TCounterPtr Hits; + + THashMap<TString, TMockReply> ReplyV6; + THashMap<TString, TMockReply> ReplyV4; + + TCachingDnsRuntime() { + SetScheduledEventFilter([](auto&&, auto&&, auto&&, auto&&) { return false; }); + ResolverOptions.MonCounters = new NMonitoring::TDynamicCounters; + + ReplyV6["localhost"] = TMockReply::SingleV6("::1"); + ReplyV4["localhost"] = TMockReply::SingleV4("127.0.0.1"); + ReplyV6["yandex.ru"] = TMockReply::SingleV6("2a02:6b8:a::a", TDuration::MilliSeconds(500)); + ReplyV4["yandex.ru"] = TMockReply::SingleV4("77.88.55.77", TDuration::MilliSeconds(250)); + ReplyV6["router.asus.com"] = TMockReply::Error(ARES_ENODATA); + ReplyV4["router.asus.com"] = TMockReply::SingleV4("192.168.0.1"); + } + + void Start(TMockDnsCallback callback) { + MockResolver = Register(new TMockDnsResolver(std::move(callback))); + EnableScheduleForActor(MockResolver); + Resolver = Register(CreateCachingDnsResolver(MockResolver, ResolverOptions)); + Sleeper = AllocateEdgeActor(); + + InFlight6 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/InFlight/V6", false); + InFlight4 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/InFlight/V4", false); + Total6 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/Total/V6", true); + Total4 = ResolverOptions.MonCounters->GetCounter("DnsResolver/Outgoing/Total/V4", true); + Misses = ResolverOptions.MonCounters->GetCounter("DnsResolver/Cache/Misses", true); + Hits = ResolverOptions.MonCounters->GetCounter("DnsResolver/Cache/Hits", true); + } + + void Start() { + Start([this](const TString& name, int family) { + switch (family) { + case AF_INET6: { + auto it = ReplyV6.find(name); + if (it != ReplyV6.end()) { + return it->second; + } + break; + } + case AF_INET: { + auto it = ReplyV4.find(name); + if (it != ReplyV4.end()) { + return it->second; + } + break; + } + } + return TMockReply::Error(ARES_ENOTFOUND); + }); + } + + void Section(const TString& section) { + Section_ = section; + } + + void Sleep(TDuration duration) { + Schedule(new IEventHandle(Sleeper, Sleeper, new TEvents::TEvWakeup), duration); + GrabEdgeEventRethrow<TEvents::TEvWakeup>(Sleeper); + } + + void WaitNoInFlight() { + if (*InFlight6 || *InFlight4) { + TDispatchOptions options; + options.CustomFinalCondition = [&]() { + return !*InFlight6 && !*InFlight4; + }; + DispatchEvents(options); + UNIT_ASSERT_C(!*InFlight6 && !*InFlight4, "Failed to wait for no inflight in " << Section_); + } + } + + void SendGetHostByName(const TActorId& sender, const TString& name, int family = AF_UNSPEC) { + Send(new IEventHandle(Resolver, sender, new TEvDns::TEvGetHostByName(name, family)), 0, true); + } + + void SendGetAddr(const TActorId& sender, const TString& name, int family = AF_UNSPEC) { + Send(new IEventHandle(Resolver, sender, new TEvDns::TEvGetAddr(name, family)), 0, true); + } + + TEvDns::TEvGetHostByNameResult::TPtr WaitGetHostByName(const TActorId& sender) { + return GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); + } + + TEvDns::TEvGetAddrResult::TPtr WaitGetAddr(const TActorId& sender) { + return GrabEdgeEventRethrow<TEvDns::TEvGetAddrResult>(sender); + } + + void ExpectInFlight6(i64 count) { + UNIT_ASSERT_VALUES_EQUAL_C(InFlight6->Val(), count, Section_); + } + + void ExpectInFlight4(i64 count) { + UNIT_ASSERT_VALUES_EQUAL_C(InFlight4->Val(), count, Section_); + } + + void ExpectTotal6(i64 count) { + UNIT_ASSERT_VALUES_EQUAL_C(Total6->Val(), count, Section_); + } + + void ExpectTotal4(i64 count) { + UNIT_ASSERT_VALUES_EQUAL_C(Total4->Val(), count, Section_); + } + + void Expect6(i64 total, i64 inflight) { + UNIT_ASSERT_C( + Total6->Val() == total && InFlight6->Val() == inflight, + Section_ << ": Expect6(" << total << ", " << inflight << ") " + << " but got (" << Total6->Val() << ", " << InFlight6->Val() << ")"); + } + + void Expect4(i64 total, i64 inflight) { + UNIT_ASSERT_C( + Total4->Val() == total && InFlight4->Val() == inflight, + Section_ << ": Expect4(" << total << ", " << inflight << ") " + << " got (" << Total4->Val() << ", " << InFlight4->Val() << ")"); + } + + void ExpectMisses(i64 count) { + UNIT_ASSERT_VALUES_EQUAL_C(Misses->Val(), count, Section_); + } + + void ExpectHits(i64 count) { + UNIT_ASSERT_VALUES_EQUAL_C(Hits->Val(), count, Section_); + } + + void ExpectGetHostByNameError(const TActorId& sender, int status) { + auto ev = WaitGetHostByName(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, status, Section_ << ": " << ev->Get()->ErrorText); + } + + void ExpectGetAddrError(const TActorId& sender, int status) { + auto ev = WaitGetAddr(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, status, Section_ << ": " << ev->Get()->ErrorText); + } + + void ExpectGetHostByNameSuccess(const TActorId& sender, const TString& expected) { + auto ev = WaitGetHostByName(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, Section_ << ": " << ev->Get()->ErrorText); + TStringBuilder result; + for (const auto& addr : ev->Get()->AddrsV6) { + if (result) { + result << ','; + } + result << TAddrToString()(addr); + } + for (const auto& addr : ev->Get()->AddrsV4) { + if (result) { + result << ','; + } + result << TAddrToString()(addr); + } + UNIT_ASSERT_VALUES_EQUAL_C(TString(result), expected, Section_); + } + + void ExpectGetAddrSuccess(const TActorId& sender, const TString& expected) { + auto ev = WaitGetAddr(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, Section_ << ": " << ev->Get()->ErrorText); + TString result = AddrToString(ev->Get()->Addr); + UNIT_ASSERT_VALUES_EQUAL_C(result, expected, Section_); + } + }; + + Y_UNIT_TEST(UnusableResolver) { + TCachingDnsRuntime runtime; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + + runtime.Send(new IEventHandle(runtime.MockResolver, { }, new TEvents::TEvPoison), 0, true); + runtime.SendGetAddr(sender, "foo.ru", AF_UNSPEC); + runtime.ExpectGetAddrError(sender, ARES_ENOTINITIALIZED); + } + + Y_UNIT_TEST(ResolveCaching) { + TCachingDnsRuntime runtime; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + // First time resolve, ipv4 and ipv6 sent in parallel, we wait for ipv6 result + runtime.Section("First time resolve"); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.Expect6(1, 0); + runtime.Expect4(1, 0); + runtime.ExpectMisses(1); + runtime.ExpectHits(0); + + // Second resolve, ipv6 and ipv4 queries result in a cache hit + runtime.Section("Second resolve, ipv6"); + runtime.SendGetAddr(sender, "yandex.ru", AF_INET6); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.Expect6(1, 0); + runtime.ExpectHits(1); + runtime.Section("Second resolve, ipv4"); + runtime.SendGetAddr(sender, "yandex.ru", AF_INET); + runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); + runtime.Expect4(1, 0); + runtime.ExpectHits(2); + + // Wait until soft expiration and try ipv4 again + // Will cause a cache hit, but will start a new ipv4 request in background + runtime.Section("Retry ipv4 after soft expiration"); + runtime.Sleep(TDuration::Seconds(15)); + runtime.SendGetAddr(sender, "yandex.ru", AF_INET); + runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); + runtime.Expect6(1, 0); + runtime.Expect4(2, 1); + runtime.ExpectMisses(1); + runtime.ExpectHits(3); + runtime.WaitNoInFlight(); + + // Wait until soft expiration and try both again + // Will cause a cache hit, but will start a new ipv6 request in background + runtime.Section("Retry both after soft expiration"); + runtime.Sleep(TDuration::Seconds(15)); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.Expect6(2, 1); + runtime.Expect4(2, 0); + runtime.ExpectMisses(1); + runtime.ExpectHits(4); + runtime.WaitNoInFlight(); + + // Wait until hard expiration and try both again + // Will cause a cache miss and new resolve requests + runtime.Section("Retry both after hard expiration"); + runtime.Sleep(TDuration::Hours(2)); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.Expect6(3, 0); + runtime.Expect4(3, 0); + runtime.ExpectMisses(2); + runtime.ExpectHits(4); + + // Wait half the hard expiration time, must always result in a cache hit + runtime.Section("Retry both after half hard expiration"); + for (ui64 i = 1; i <= 4; ++i) { + runtime.Sleep(TDuration::Hours(1)); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.Expect6(3 + i, 1); + runtime.ExpectHits(4 + i); + runtime.WaitNoInFlight(); + } + + // Change v6 result to a timeout, must keep using cached result until hard expiration + runtime.Section("Dns keeps timing out"); + runtime.ReplyV6["yandex.ru"] = TMockReply::Error(ARES_ETIMEOUT); + for (ui64 i = 1; i <= 4; ++i) { + runtime.Sleep(TDuration::Seconds(15)); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.Expect6(7 + i, 1); + runtime.ExpectHits(8 + i); + runtime.WaitNoInFlight(); + } + + // Change v6 result to nodata, must switch to a v4 result eventually + runtime.Section("Host changes to being ipv4 only"); + runtime.ReplyV6["yandex.ru"] = TMockReply::Error(ARES_ENODATA); + runtime.Sleep(TDuration::Seconds(2)); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + runtime.WaitNoInFlight(); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); + runtime.Expect6(12, 0); + runtime.Expect4(4, 0); + runtime.ExpectMisses(3); + + // Change v6 result to nxdomain, must not fall back to a v4 result + runtime.Section("Host is removed from dns"); + runtime.ReplyV6["yandex.ru"] = TMockReply::Error(ARES_ENOTFOUND); + runtime.Sleep(TDuration::Seconds(15)); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); + runtime.WaitNoInFlight(); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + } + + Y_UNIT_TEST(ResolveCachingV4) { + TCachingDnsRuntime runtime; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.Section("First request"); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); + runtime.ExpectMisses(1); + + runtime.Section("Second request"); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); + runtime.ExpectHits(1); + + runtime.Section("Dns keeps timing out"); + runtime.ReplyV6["router.asus.com"] = TMockReply::Error(ARES_ETIMEOUT); + runtime.ReplyV4["router.asus.com"] = TMockReply::Error(ARES_ETIMEOUT); + for (ui64 i = 1; i <= 4; ++i) { + runtime.Sleep(TDuration::Seconds(15)); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); + runtime.Expect6(1 + i, 1); + runtime.Expect4(1 + i, 1); + runtime.ExpectHits(1 + i); + runtime.WaitNoInFlight(); + } + + runtime.Section("Host is removed from ipv4 dns"); + runtime.ReplyV4["router.asus.com"] = TMockReply::Error(ARES_ENOTFOUND); + runtime.Sleep(TDuration::Seconds(15)); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); + runtime.WaitNoInFlight(); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + } + + Y_UNIT_TEST(EventualTimeout) { + TCachingDnsRuntime runtime; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.ReplyV6["notfound.ru"] = TMockReply::Error(ARES_ENODATA); + runtime.ReplyV4["notfound.ru"] = TMockReply::Error(ARES_ENOTFOUND); + runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + + runtime.ReplyV4["notfound.ru"] = TMockReply::Error(ARES_ETIMEOUT); + runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + runtime.WaitNoInFlight(); + + bool timeout = false; + for (ui64 i = 1; i <= 8; ++i) { + runtime.Sleep(TDuration::Minutes(30)); + runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); + auto ev = runtime.WaitGetAddr(sender); + if (ev->Get()->Status == ARES_ETIMEOUT && i > 2) { + timeout = true; + break; + } + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, ARES_ENOTFOUND, + "Iteration " << i << ": " << ev->Get()->ErrorText); + } + + UNIT_ASSERT_C(timeout, "DnsResolver did not reply with a timeout"); + } + + Y_UNIT_TEST(MultipleRequestsAndHosts) { + TCachingDnsRuntime runtime; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.SendGetHostByName(sender, "router.asus.com", AF_UNSPEC); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetHostByNameSuccess(sender, "192.168.0.1"); + runtime.ExpectGetAddrSuccess(sender, "192.168.0.1"); + runtime.ExpectGetHostByNameSuccess(sender, "2a02:6b8:a::a"); + runtime.ExpectGetAddrSuccess(sender, "2a02:6b8:a::a"); + + runtime.SendGetHostByName(sender, "notfound.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); + runtime.ExpectGetHostByNameError(sender, ARES_ENOTFOUND); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + } + + Y_UNIT_TEST(DisabledIPv6) { + TCachingDnsRuntime runtime; + runtime.ResolverOptions.AllowIPv6 = false; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetHostByNameSuccess(sender, "77.88.55.77"); + runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); + + runtime.SendGetHostByName(sender, "yandex.ru", AF_INET6); + runtime.SendGetAddr(sender, "yandex.ru", AF_INET6); + runtime.ExpectGetHostByNameError(sender, ARES_EBADFAMILY); + runtime.ExpectGetAddrError(sender, ARES_EBADFAMILY); + + runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.ExpectGetHostByNameSuccess(sender, "77.88.55.77"); + runtime.ExpectGetAddrSuccess(sender, "77.88.55.77"); + + runtime.SendGetHostByName(sender, "notfound.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); + runtime.ExpectGetHostByNameError(sender, ARES_ENOTFOUND); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + } + + Y_UNIT_TEST(DisabledIPv4) { + TCachingDnsRuntime runtime; + runtime.ResolverOptions.AllowIPv4 = false; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.SendGetHostByName(sender, "router.asus.com", AF_UNSPEC); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetHostByNameError(sender, ARES_ENODATA); + runtime.ExpectGetAddrError(sender, ARES_ENODATA); + + runtime.SendGetHostByName(sender, "router.asus.com", AF_INET); + runtime.SendGetAddr(sender, "router.asus.com", AF_INET); + runtime.ExpectGetHostByNameError(sender, ARES_EBADFAMILY); + runtime.ExpectGetAddrError(sender, ARES_EBADFAMILY); + + runtime.SendGetHostByName(sender, "router.asus.com", AF_UNSPEC); + runtime.SendGetAddr(sender, "router.asus.com", AF_UNSPEC); + runtime.ExpectGetHostByNameError(sender, ARES_ENODATA); + runtime.ExpectGetAddrError(sender, ARES_ENODATA); + + runtime.SendGetHostByName(sender, "notfound.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "notfound.ru", AF_UNSPEC); + runtime.ExpectGetHostByNameError(sender, ARES_ENOTFOUND); + runtime.ExpectGetAddrError(sender, ARES_ENOTFOUND); + } + + Y_UNIT_TEST(PoisonPill) { + TCachingDnsRuntime runtime; + runtime.Initialize(); + runtime.Start(); + + auto sender = runtime.AllocateEdgeActor(); + + runtime.SendGetHostByName(sender, "yandex.ru", AF_UNSPEC); + runtime.SendGetAddr(sender, "yandex.ru", AF_UNSPEC); + runtime.Send(new IEventHandle(runtime.Resolver, sender, new TEvents::TEvPoison), 0, true); + runtime.ExpectGetHostByNameError(sender, ARES_ECANCELLED); + runtime.ExpectGetAddrError(sender, ARES_ECANCELLED); + } + +} diff --git a/library/cpp/actors/dnsresolver/dnsresolver_ondemand.cpp b/library/cpp/actors/dnsresolver/dnsresolver_ondemand.cpp index da3f364601..2025162e95 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver_ondemand.cpp +++ b/library/cpp/actors/dnsresolver/dnsresolver_ondemand.cpp @@ -1,64 +1,64 @@ -#include "dnsresolver.h" - -#include <library/cpp/actors/core/hfunc.h> - -namespace NActors { -namespace NDnsResolver { - - class TOnDemandDnsResolver : public TActor<TOnDemandDnsResolver> { - public: - TOnDemandDnsResolver(TOnDemandDnsResolverOptions options) - : TActor(&TThis::StateWork) - , Options(std::move(options)) - { } - - static constexpr EActivityType ActorActivityType() { - return DNS_RESOLVER; - } - - private: - STRICT_STFUNC(StateWork, { - cFunc(TEvents::TEvPoison::EventType, PassAway); - fFunc(TEvDns::TEvGetHostByName::EventType, Forward); - fFunc(TEvDns::TEvGetAddr::EventType, Forward); - }); - - void Forward(STATEFN_SIG) { - ev->Rewrite(ev->GetTypeRewrite(), GetUpstream()); - TActivationContext::Send(std::move(ev)); - } - - private: - TActorId GetUpstream() { - if (Y_UNLIKELY(!CachingResolverId)) { - if (Y_LIKELY(!SimpleResolverId)) { +#include "dnsresolver.h" + +#include <library/cpp/actors/core/hfunc.h> + +namespace NActors { +namespace NDnsResolver { + + class TOnDemandDnsResolver : public TActor<TOnDemandDnsResolver> { + public: + TOnDemandDnsResolver(TOnDemandDnsResolverOptions options) + : TActor(&TThis::StateWork) + , Options(std::move(options)) + { } + + static constexpr EActivityType ActorActivityType() { + return DNS_RESOLVER; + } + + private: + STRICT_STFUNC(StateWork, { + cFunc(TEvents::TEvPoison::EventType, PassAway); + fFunc(TEvDns::TEvGetHostByName::EventType, Forward); + fFunc(TEvDns::TEvGetAddr::EventType, Forward); + }); + + void Forward(STATEFN_SIG) { + ev->Rewrite(ev->GetTypeRewrite(), GetUpstream()); + TActivationContext::Send(std::move(ev)); + } + + private: + TActorId GetUpstream() { + if (Y_UNLIKELY(!CachingResolverId)) { + if (Y_LIKELY(!SimpleResolverId)) { SimpleResolverId = RegisterWithSameMailbox(CreateSimpleDnsResolver(Options)); - } + } CachingResolverId = RegisterWithSameMailbox(CreateCachingDnsResolver(SimpleResolverId, Options)); - } - return CachingResolverId; - } - - void PassAway() override { - if (CachingResolverId) { - Send(CachingResolverId, new TEvents::TEvPoison); - CachingResolverId = { }; - } - if (SimpleResolverId) { - Send(SimpleResolverId, new TEvents::TEvPoison); - SimpleResolverId = { }; - } - } - - private: - TOnDemandDnsResolverOptions Options; - TActorId SimpleResolverId; - TActorId CachingResolverId; - }; - - IActor* CreateOnDemandDnsResolver(TOnDemandDnsResolverOptions options) { - return new TOnDemandDnsResolver(std::move(options)); - } - -} // namespace NDnsResolver -} // namespace NActors + } + return CachingResolverId; + } + + void PassAway() override { + if (CachingResolverId) { + Send(CachingResolverId, new TEvents::TEvPoison); + CachingResolverId = { }; + } + if (SimpleResolverId) { + Send(SimpleResolverId, new TEvents::TEvPoison); + SimpleResolverId = { }; + } + } + + private: + TOnDemandDnsResolverOptions Options; + TActorId SimpleResolverId; + TActorId CachingResolverId; + }; + + IActor* CreateOnDemandDnsResolver(TOnDemandDnsResolverOptions options) { + return new TOnDemandDnsResolver(std::move(options)); + } + +} // namespace NDnsResolver +} // namespace NActors diff --git a/library/cpp/actors/dnsresolver/dnsresolver_ondemand_ut.cpp b/library/cpp/actors/dnsresolver/dnsresolver_ondemand_ut.cpp index ee6b4113f0..2758484552 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver_ondemand_ut.cpp +++ b/library/cpp/actors/dnsresolver/dnsresolver_ondemand_ut.cpp @@ -1,24 +1,24 @@ -#include "dnsresolver.h" - -#include <library/cpp/actors/testlib/test_runtime.h> -#include <library/cpp/testing/unittest/registar.h> - -using namespace NActors; -using namespace NActors::NDnsResolver; - -Y_UNIT_TEST_SUITE(OnDemandDnsResolver) { - - Y_UNIT_TEST(ResolveLocalHost) { - TTestActorRuntimeBase runtime; - runtime.Initialize(); - auto sender = runtime.AllocateEdgeActor(); - auto resolver = runtime.Register(CreateOnDemandDnsResolver()); - runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("localhost", AF_UNSPEC)), - 0, true); - auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); - size_t addrs = ev->Get()->AddrsV4.size() + ev->Get()->AddrsV6.size(); - UNIT_ASSERT_C(addrs > 0, "Got " << addrs << " addresses"); - } - -} +#include "dnsresolver.h" + +#include <library/cpp/actors/testlib/test_runtime.h> +#include <library/cpp/testing/unittest/registar.h> + +using namespace NActors; +using namespace NActors::NDnsResolver; + +Y_UNIT_TEST_SUITE(OnDemandDnsResolver) { + + Y_UNIT_TEST(ResolveLocalHost) { + TTestActorRuntimeBase runtime; + runtime.Initialize(); + auto sender = runtime.AllocateEdgeActor(); + auto resolver = runtime.Register(CreateOnDemandDnsResolver()); + runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("localhost", AF_UNSPEC)), + 0, true); + auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); + size_t addrs = ev->Get()->AddrsV4.size() + ev->Get()->AddrsV6.size(); + UNIT_ASSERT_C(addrs > 0, "Got " << addrs << " addresses"); + } + +} diff --git a/library/cpp/actors/dnsresolver/dnsresolver_ut.cpp b/library/cpp/actors/dnsresolver/dnsresolver_ut.cpp index 86d2ae7a1c..0c343a805c 100644 --- a/library/cpp/actors/dnsresolver/dnsresolver_ut.cpp +++ b/library/cpp/actors/dnsresolver/dnsresolver_ut.cpp @@ -1,98 +1,98 @@ -#include "dnsresolver.h" - -#include <library/cpp/actors/testlib/test_runtime.h> -#include <library/cpp/testing/unittest/registar.h> -#include <util/string/builder.h> - -#include <ares.h> - -using namespace NActors; -using namespace NActors::NDnsResolver; - -Y_UNIT_TEST_SUITE(DnsResolver) { - - struct TSilentUdpServer { - TInetDgramSocket Socket; - ui16 Port; - - TSilentUdpServer() { - TSockAddrInet addr("127.0.0.1", 0); - int err = Socket.Bind(&addr); - Y_VERIFY(err == 0, "Cannot bind a udp socket"); - Port = addr.GetPort(); - } - }; - - Y_UNIT_TEST(ResolveLocalHost) { - TTestActorRuntimeBase runtime; - runtime.Initialize(); - auto sender = runtime.AllocateEdgeActor(); - auto resolver = runtime.Register(CreateSimpleDnsResolver()); - runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("localhost", AF_UNSPEC)), - 0, true); - auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); - size_t addrs = ev->Get()->AddrsV4.size() + ev->Get()->AddrsV6.size(); - UNIT_ASSERT_C(addrs > 0, "Got " << addrs << " addresses"); - } - - Y_UNIT_TEST(ResolveYandexRu) { - TTestActorRuntimeBase runtime; - runtime.Initialize(); - auto sender = runtime.AllocateEdgeActor(); - auto resolver = runtime.Register(CreateSimpleDnsResolver()); - runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("yandex.ru", AF_UNSPEC)), - 0, true); - auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); - size_t addrs = ev->Get()->AddrsV4.size() + ev->Get()->AddrsV6.size(); - UNIT_ASSERT_C(addrs > 0, "Got " << addrs << " addresses"); - } - - Y_UNIT_TEST(GetAddrYandexRu) { - TTestActorRuntimeBase runtime; - runtime.Initialize(); - auto sender = runtime.AllocateEdgeActor(); - auto resolver = runtime.Register(CreateSimpleDnsResolver()); - - runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetAddr("yandex.ru", AF_UNSPEC)), - 0, true); - auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetAddrResult>(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); - UNIT_ASSERT_C(ev->Get()->IsV4() || ev->Get()->IsV6(), "Expect v4 or v6 address"); - } - - Y_UNIT_TEST(ResolveTimeout) { - TSilentUdpServer server; - TTestActorRuntimeBase runtime; - runtime.Initialize(); - auto sender = runtime.AllocateEdgeActor(); - TSimpleDnsResolverOptions options; - options.Timeout = TDuration::MilliSeconds(250); - options.Attempts = 2; - options.Servers.emplace_back(TStringBuilder() << "127.0.0.1:" << server.Port); - auto resolver = runtime.Register(CreateSimpleDnsResolver(options)); - runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("timeout.yandex.ru", AF_INET)), - 0, true); - auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, ARES_ETIMEOUT, ev->Get()->ErrorText); - } - - Y_UNIT_TEST(ResolveGracefulStop) { - TSilentUdpServer server; - TTestActorRuntimeBase runtime; - runtime.Initialize(); - auto sender = runtime.AllocateEdgeActor(); - TSimpleDnsResolverOptions options; - options.Timeout = TDuration::Seconds(5); - options.Attempts = 5; - options.Servers.emplace_back(TStringBuilder() << "127.0.0.1:" << server.Port); - auto resolver = runtime.Register(CreateSimpleDnsResolver(options)); - runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("timeout.yandex.ru", AF_INET)), - 0, true); - runtime.Send(new IEventHandle(resolver, sender, new TEvents::TEvPoison), 0, true); - auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); - UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, ARES_ECANCELLED, ev->Get()->ErrorText); - } - -} +#include "dnsresolver.h" + +#include <library/cpp/actors/testlib/test_runtime.h> +#include <library/cpp/testing/unittest/registar.h> +#include <util/string/builder.h> + +#include <ares.h> + +using namespace NActors; +using namespace NActors::NDnsResolver; + +Y_UNIT_TEST_SUITE(DnsResolver) { + + struct TSilentUdpServer { + TInetDgramSocket Socket; + ui16 Port; + + TSilentUdpServer() { + TSockAddrInet addr("127.0.0.1", 0); + int err = Socket.Bind(&addr); + Y_VERIFY(err == 0, "Cannot bind a udp socket"); + Port = addr.GetPort(); + } + }; + + Y_UNIT_TEST(ResolveLocalHost) { + TTestActorRuntimeBase runtime; + runtime.Initialize(); + auto sender = runtime.AllocateEdgeActor(); + auto resolver = runtime.Register(CreateSimpleDnsResolver()); + runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("localhost", AF_UNSPEC)), + 0, true); + auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); + size_t addrs = ev->Get()->AddrsV4.size() + ev->Get()->AddrsV6.size(); + UNIT_ASSERT_C(addrs > 0, "Got " << addrs << " addresses"); + } + + Y_UNIT_TEST(ResolveYandexRu) { + TTestActorRuntimeBase runtime; + runtime.Initialize(); + auto sender = runtime.AllocateEdgeActor(); + auto resolver = runtime.Register(CreateSimpleDnsResolver()); + runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("yandex.ru", AF_UNSPEC)), + 0, true); + auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); + size_t addrs = ev->Get()->AddrsV4.size() + ev->Get()->AddrsV6.size(); + UNIT_ASSERT_C(addrs > 0, "Got " << addrs << " addresses"); + } + + Y_UNIT_TEST(GetAddrYandexRu) { + TTestActorRuntimeBase runtime; + runtime.Initialize(); + auto sender = runtime.AllocateEdgeActor(); + auto resolver = runtime.Register(CreateSimpleDnsResolver()); + + runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetAddr("yandex.ru", AF_UNSPEC)), + 0, true); + auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetAddrResult>(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, 0, ev->Get()->ErrorText); + UNIT_ASSERT_C(ev->Get()->IsV4() || ev->Get()->IsV6(), "Expect v4 or v6 address"); + } + + Y_UNIT_TEST(ResolveTimeout) { + TSilentUdpServer server; + TTestActorRuntimeBase runtime; + runtime.Initialize(); + auto sender = runtime.AllocateEdgeActor(); + TSimpleDnsResolverOptions options; + options.Timeout = TDuration::MilliSeconds(250); + options.Attempts = 2; + options.Servers.emplace_back(TStringBuilder() << "127.0.0.1:" << server.Port); + auto resolver = runtime.Register(CreateSimpleDnsResolver(options)); + runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("timeout.yandex.ru", AF_INET)), + 0, true); + auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, ARES_ETIMEOUT, ev->Get()->ErrorText); + } + + Y_UNIT_TEST(ResolveGracefulStop) { + TSilentUdpServer server; + TTestActorRuntimeBase runtime; + runtime.Initialize(); + auto sender = runtime.AllocateEdgeActor(); + TSimpleDnsResolverOptions options; + options.Timeout = TDuration::Seconds(5); + options.Attempts = 5; + options.Servers.emplace_back(TStringBuilder() << "127.0.0.1:" << server.Port); + auto resolver = runtime.Register(CreateSimpleDnsResolver(options)); + runtime.Send(new IEventHandle(resolver, sender, new TEvDns::TEvGetHostByName("timeout.yandex.ru", AF_INET)), + 0, true); + runtime.Send(new IEventHandle(resolver, sender, new TEvents::TEvPoison), 0, true); + auto ev = runtime.GrabEdgeEventRethrow<TEvDns::TEvGetHostByNameResult>(sender); + UNIT_ASSERT_VALUES_EQUAL_C(ev->Get()->Status, ARES_ECANCELLED, ev->Get()->ErrorText); + } + +} diff --git a/library/cpp/actors/dnsresolver/ut/ya.make b/library/cpp/actors/dnsresolver/ut/ya.make index 2af5f5f260..ad936bdacd 100644 --- a/library/cpp/actors/dnsresolver/ut/ya.make +++ b/library/cpp/actors/dnsresolver/ut/ya.make @@ -1,20 +1,20 @@ -UNITTEST_FOR(library/cpp/actors/dnsresolver) - -OWNER(g:kikimr) - -PEERDIR( - library/cpp/actors/testlib -) - -SRCS( - dnsresolver_caching_ut.cpp - dnsresolver_ondemand_ut.cpp - dnsresolver_ut.cpp -) - -ADDINCL(contrib/libs/c-ares) - -TAG(ya:external) -REQUIREMENTS(network:full) - -END() +UNITTEST_FOR(library/cpp/actors/dnsresolver) + +OWNER(g:kikimr) + +PEERDIR( + library/cpp/actors/testlib +) + +SRCS( + dnsresolver_caching_ut.cpp + dnsresolver_ondemand_ut.cpp + dnsresolver_ut.cpp +) + +ADDINCL(contrib/libs/c-ares) + +TAG(ya:external) +REQUIREMENTS(network:full) + +END() diff --git a/library/cpp/actors/dnsresolver/ya.make b/library/cpp/actors/dnsresolver/ya.make index 7ee69fe4b5..329c56c5b3 100644 --- a/library/cpp/actors/dnsresolver/ya.make +++ b/library/cpp/actors/dnsresolver/ya.make @@ -1,20 +1,20 @@ -LIBRARY() - -OWNER(g:kikimr) - -SRCS( - dnsresolver.cpp - dnsresolver_caching.cpp - dnsresolver_ondemand.cpp -) - -PEERDIR( - library/cpp/actors/core - contrib/libs/c-ares -) - -ADDINCL(contrib/libs/c-ares) - -END() - -RECURSE_FOR_TESTS(ut) +LIBRARY() + +OWNER(g:kikimr) + +SRCS( + dnsresolver.cpp + dnsresolver_caching.cpp + dnsresolver_ondemand.cpp +) + +PEERDIR( + library/cpp/actors/core + contrib/libs/c-ares +) + +ADDINCL(contrib/libs/c-ares) + +END() + +RECURSE_FOR_TESTS(ut) diff --git a/library/cpp/actors/interconnect/interconnect.h b/library/cpp/actors/interconnect/interconnect.h index f885377d2d..225a5243fd 100644 --- a/library/cpp/actors/interconnect/interconnect.h +++ b/library/cpp/actors/interconnect/interconnect.h @@ -133,7 +133,7 @@ namespace NActors { const TIntrusivePtr<TTableNameserverSetup>& setup, ui32 poolId = 0); - /** + /** * Name service which can be paired with external discovery service. * Copies information from setup on the start (table may be empty). * Handles TEvNodesInfo to change list of known nodes. @@ -147,17 +147,17 @@ namespace NActors { ui32 poolId = 0); /** - * Creates an actor that resolves host/port and replies with either: - * - * - TEvLocalNodeInfo on success - * - TEvResolveError on errors - * - * Optional defaultAddress may be used as fallback. - */ - IActor* CreateResolveActor( - const TString& host, ui16 port, ui32 nodeId, const TString& defaultAddress, - const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline); - + * Creates an actor that resolves host/port and replies with either: + * + * - TEvLocalNodeInfo on success + * - TEvResolveError on errors + * + * Optional defaultAddress may be used as fallback. + */ + IActor* CreateResolveActor( + const TString& host, ui16 port, ui32 nodeId, const TString& defaultAddress, + const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline); + inline IActor* CreateResolveActor( ui32 nodeId, const TTableNameserverSetup::TNodeInfo& nodeInfo, const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) @@ -166,14 +166,14 @@ namespace NActors { replyTo, replyFrom, deadline); } - /** - * Creates an actor that resolves host/port and replies with either: - * - * - TEvAddressInfo on success - * - TEvResolveError on errors - */ - IActor* CreateResolveActor( - const TString& host, ui16 port, - const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline); - + /** + * Creates an actor that resolves host/port and replies with either: + * + * - TEvAddressInfo on success + * - TEvResolveError on errors + */ + IActor* CreateResolveActor( + const TString& host, ui16 port, + const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline); + } diff --git a/library/cpp/actors/interconnect/interconnect_nameserver_table.cpp b/library/cpp/actors/interconnect/interconnect_nameserver_table.cpp index 18f3f4aed3..43419bf70d 100644 --- a/library/cpp/actors/interconnect/interconnect_nameserver_table.cpp +++ b/library/cpp/actors/interconnect/interconnect_nameserver_table.cpp @@ -17,7 +17,7 @@ namespace NActors { return NAMESERVICE; } - TInterconnectNameserverTable(const TIntrusivePtr<TTableNameserverSetup>& setup, ui32 /*resolvePoolId*/) + TInterconnectNameserverTable(const TIntrusivePtr<TTableNameserverSetup>& setup, ui32 /*resolvePoolId*/) : TInterconnectNameserverBase<TInterconnectNameserverTable>(&TInterconnectNameserverTable::StateFunc, setup->StaticNodeTable) , Config(setup) { diff --git a/library/cpp/actors/interconnect/interconnect_resolve.cpp b/library/cpp/actors/interconnect/interconnect_resolve.cpp index 1f0ed84e30..14296194df 100644 --- a/library/cpp/actors/interconnect/interconnect_resolve.cpp +++ b/library/cpp/actors/interconnect/interconnect_resolve.cpp @@ -1,174 +1,174 @@ -#include "interconnect.h" -#include "interconnect_address.h" -#include "events_local.h" - -#include <library/cpp/actors/core/actor_bootstrapped.h> -#include <library/cpp/actors/core/hfunc.h> -#include <library/cpp/actors/dnsresolver/dnsresolver.h> - -namespace NActors { - - using namespace NActors::NDnsResolver; - - class TInterconnectResolveActor : public TActorBootstrapped<TInterconnectResolveActor> { - public: - TInterconnectResolveActor( - const TString& host, ui16 port, ui32 nodeId, const TString& defaultAddress, - const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) - : Host(host) - , NodeId(nodeId) - , Port(port) - , DefaultAddress(defaultAddress) - , ReplyTo(replyTo) - , ReplyFrom(replyFrom) - , Deadline(deadline) - { } - - TInterconnectResolveActor( - const TString& host, ui16 port, - const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) - : Host(host) - , Port(port) - , ReplyTo(replyTo) - , ReplyFrom(replyFrom) - , Deadline(deadline) - { } - - static constexpr EActivityType ActorActivityType() { - return NAMESERVICE; - } - - void Bootstrap() { - TMaybe<TString> errorText; - if (auto addr = ExtractDefaultAddr(errorText)) { - return SendAddrAndDie(std::move(addr)); - } - - if (errorText) { - SendErrorAndDie(*errorText); - } - - auto now = TActivationContext::Now(); - if (Deadline < now) { - SendErrorAndDie("Deadline"); - return; - } - - Send(MakeDnsResolverActorId(), - new TEvDns::TEvGetAddr(Host, AF_UNSPEC), - IEventHandle::FlagTrackDelivery); - - if (Deadline != TInstant::Max()) { - Schedule(Deadline, new TEvents::TEvWakeup); - } - - Become(&TThis::StateWork); - } - - STRICT_STFUNC(StateWork, { - sFunc(TEvents::TEvWakeup, HandleTimeout); - sFunc(TEvents::TEvUndelivered, HandleUndelivered); - hFunc(TEvDns::TEvGetAddrResult, Handle); - }); - - void HandleTimeout() { - SendErrorAndDie("Deadline"); - } - - void HandleUndelivered() { - SendErrorAndDie("Dns resolver is unavailable"); - } - - void Handle(TEvDns::TEvGetAddrResult::TPtr& ev) { - if (auto addr = ExtractAddr(ev->Get())) { - return SendAddrAndDie(std::move(addr)); - } - - SendErrorAndDie(ev->Get()->ErrorText); - } - - void SendAddrAndDie(NAddr::IRemoteAddrPtr addr) { - if (NodeId) { - auto reply = new TEvLocalNodeInfo; - reply->NodeId = *NodeId; - reply->Address = std::move(addr); - TActivationContext::Send(new IEventHandle(ReplyTo, ReplyFrom, reply)); - } else { - auto reply = new TEvAddressInfo; - reply->Address = std::move(addr); - TActivationContext::Send(new IEventHandle(ReplyTo, ReplyFrom, reply)); - } - PassAway(); - } - - void SendErrorAndDie(const TString& errorText) { - auto *event = new TEvResolveError; - event->Explain = errorText; - TActivationContext::Send(new IEventHandle(ReplyTo, ReplyFrom, event)); - PassAway(); - } - - NAddr::IRemoteAddrPtr ExtractAddr(TEvDns::TEvGetAddrResult* msg) { - if (msg->Status == 0) { - if (msg->IsV6()) { - struct sockaddr_in6 sin6; - Zero(sin6); - sin6.sin6_family = AF_INET6; - sin6.sin6_addr = msg->GetAddrV6(); - sin6.sin6_port = HostToInet(Port); - return MakeHolder<NAddr::TIPv6Addr>(sin6); - } - - if (msg->IsV4()) { - return MakeHolder<NAddr::TIPv4Addr>(TIpAddress(msg->GetAddrV4().s_addr, Port)); - } - - Y_FAIL("Unexpected result address family"); - } - - return nullptr; - } - - NAddr::IRemoteAddrPtr ExtractDefaultAddr(TMaybe<TString>& errorText) { - if (DefaultAddress) { - NInterconnect::TAddress address(DefaultAddress.data(), Port); - - switch (address.GetFamily()) { - case AF_INET: - return MakeHolder<NAddr::TIPv4Addr>(*(sockaddr_in*)address.SockAddr()); - case AF_INET6: - return MakeHolder<NAddr::TIPv6Addr>(*(sockaddr_in6*)address.SockAddr()); - default: - errorText = "Unsupported default address: " + DefaultAddress; - break; - } - } - - return nullptr; - } - - private: - const TString Host; - const std::optional<ui32> NodeId; - const ui16 Port; - const TString DefaultAddress; - const TActorId ReplyTo; - const TActorId ReplyFrom; - const TInstant Deadline; - }; - - IActor* CreateResolveActor( - const TString& host, ui16 port, ui32 nodeId, const TString& defaultAddress, - const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) - { - return new TInterconnectResolveActor(host, port, nodeId, defaultAddress, replyTo, replyFrom, deadline); - } - - IActor* CreateResolveActor( - const TString& host, ui16 port, - const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) - { - return new TInterconnectResolveActor(host, port, replyTo, replyFrom, deadline); - } - -} // namespace NActors +#include "interconnect.h" +#include "interconnect_address.h" +#include "events_local.h" + +#include <library/cpp/actors/core/actor_bootstrapped.h> +#include <library/cpp/actors/core/hfunc.h> +#include <library/cpp/actors/dnsresolver/dnsresolver.h> + +namespace NActors { + + using namespace NActors::NDnsResolver; + + class TInterconnectResolveActor : public TActorBootstrapped<TInterconnectResolveActor> { + public: + TInterconnectResolveActor( + const TString& host, ui16 port, ui32 nodeId, const TString& defaultAddress, + const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) + : Host(host) + , NodeId(nodeId) + , Port(port) + , DefaultAddress(defaultAddress) + , ReplyTo(replyTo) + , ReplyFrom(replyFrom) + , Deadline(deadline) + { } + + TInterconnectResolveActor( + const TString& host, ui16 port, + const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) + : Host(host) + , Port(port) + , ReplyTo(replyTo) + , ReplyFrom(replyFrom) + , Deadline(deadline) + { } + + static constexpr EActivityType ActorActivityType() { + return NAMESERVICE; + } + + void Bootstrap() { + TMaybe<TString> errorText; + if (auto addr = ExtractDefaultAddr(errorText)) { + return SendAddrAndDie(std::move(addr)); + } + + if (errorText) { + SendErrorAndDie(*errorText); + } + + auto now = TActivationContext::Now(); + if (Deadline < now) { + SendErrorAndDie("Deadline"); + return; + } + + Send(MakeDnsResolverActorId(), + new TEvDns::TEvGetAddr(Host, AF_UNSPEC), + IEventHandle::FlagTrackDelivery); + + if (Deadline != TInstant::Max()) { + Schedule(Deadline, new TEvents::TEvWakeup); + } + + Become(&TThis::StateWork); + } + + STRICT_STFUNC(StateWork, { + sFunc(TEvents::TEvWakeup, HandleTimeout); + sFunc(TEvents::TEvUndelivered, HandleUndelivered); + hFunc(TEvDns::TEvGetAddrResult, Handle); + }); + + void HandleTimeout() { + SendErrorAndDie("Deadline"); + } + + void HandleUndelivered() { + SendErrorAndDie("Dns resolver is unavailable"); + } + + void Handle(TEvDns::TEvGetAddrResult::TPtr& ev) { + if (auto addr = ExtractAddr(ev->Get())) { + return SendAddrAndDie(std::move(addr)); + } + + SendErrorAndDie(ev->Get()->ErrorText); + } + + void SendAddrAndDie(NAddr::IRemoteAddrPtr addr) { + if (NodeId) { + auto reply = new TEvLocalNodeInfo; + reply->NodeId = *NodeId; + reply->Address = std::move(addr); + TActivationContext::Send(new IEventHandle(ReplyTo, ReplyFrom, reply)); + } else { + auto reply = new TEvAddressInfo; + reply->Address = std::move(addr); + TActivationContext::Send(new IEventHandle(ReplyTo, ReplyFrom, reply)); + } + PassAway(); + } + + void SendErrorAndDie(const TString& errorText) { + auto *event = new TEvResolveError; + event->Explain = errorText; + TActivationContext::Send(new IEventHandle(ReplyTo, ReplyFrom, event)); + PassAway(); + } + + NAddr::IRemoteAddrPtr ExtractAddr(TEvDns::TEvGetAddrResult* msg) { + if (msg->Status == 0) { + if (msg->IsV6()) { + struct sockaddr_in6 sin6; + Zero(sin6); + sin6.sin6_family = AF_INET6; + sin6.sin6_addr = msg->GetAddrV6(); + sin6.sin6_port = HostToInet(Port); + return MakeHolder<NAddr::TIPv6Addr>(sin6); + } + + if (msg->IsV4()) { + return MakeHolder<NAddr::TIPv4Addr>(TIpAddress(msg->GetAddrV4().s_addr, Port)); + } + + Y_FAIL("Unexpected result address family"); + } + + return nullptr; + } + + NAddr::IRemoteAddrPtr ExtractDefaultAddr(TMaybe<TString>& errorText) { + if (DefaultAddress) { + NInterconnect::TAddress address(DefaultAddress.data(), Port); + + switch (address.GetFamily()) { + case AF_INET: + return MakeHolder<NAddr::TIPv4Addr>(*(sockaddr_in*)address.SockAddr()); + case AF_INET6: + return MakeHolder<NAddr::TIPv6Addr>(*(sockaddr_in6*)address.SockAddr()); + default: + errorText = "Unsupported default address: " + DefaultAddress; + break; + } + } + + return nullptr; + } + + private: + const TString Host; + const std::optional<ui32> NodeId; + const ui16 Port; + const TString DefaultAddress; + const TActorId ReplyTo; + const TActorId ReplyFrom; + const TInstant Deadline; + }; + + IActor* CreateResolveActor( + const TString& host, ui16 port, ui32 nodeId, const TString& defaultAddress, + const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) + { + return new TInterconnectResolveActor(host, port, nodeId, defaultAddress, replyTo, replyFrom, deadline); + } + + IActor* CreateResolveActor( + const TString& host, ui16 port, + const TActorId& replyTo, const TActorId& replyFrom, TInstant deadline) + { + return new TInterconnectResolveActor(host, port, replyTo, replyFrom, deadline); + } + +} // namespace NActors diff --git a/library/cpp/actors/interconnect/interconnect_tcp_proxy.cpp b/library/cpp/actors/interconnect/interconnect_tcp_proxy.cpp index a634fd804b..7e2d8ccb94 100644 --- a/library/cpp/actors/interconnect/interconnect_tcp_proxy.cpp +++ b/library/cpp/actors/interconnect/interconnect_tcp_proxy.cpp @@ -43,7 +43,7 @@ namespace NActors { PassAwayTimestamp = TActivationContext::Now() + TDuration::Seconds(15); LOG_INFO_IC("ICP01", "ready to work"); - } + } void TInterconnectProxyTCP::Registered(TActorSystem* sys, const TActorId& owner) { if (!DynamicPtr) { @@ -53,7 +53,7 @@ namespace NActors { if (const auto& mon = Common->RegisterMonPage) { TString path = Sprintf("peer%04" PRIu32, PeerNodeId); TString title = Sprintf("Peer #%04" PRIu32, PeerNodeId); - mon(path, title, sys, SelfId()); + mon(path, title, sys, SelfId()); } } @@ -508,14 +508,14 @@ namespace NActors { switch (ev->GetTypeRewrite()) { case TEvInterconnect::EvForward: if (ev->Flags & IEventHandle::FlagSubscribeOnSession) { - Send(ev->Sender, new TEvInterconnect::TEvNodeDisconnected(PeerNodeId), 0, ev->Cookie); + Send(ev->Sender, new TEvInterconnect::TEvNodeDisconnected(PeerNodeId), 0, ev->Cookie); } TActivationContext::Send(ev->ForwardOnNondelivery(TEvents::TEvUndelivered::Disconnected)); break; case TEvInterconnect::TEvConnectNode::EventType: case TEvents::TEvSubscribe::EventType: - Send(ev->Sender, new TEvInterconnect::TEvNodeDisconnected(PeerNodeId), 0, ev->Cookie); + Send(ev->Sender, new TEvInterconnect::TEvNodeDisconnected(PeerNodeId), 0, ev->Cookie); break; case TEvents::TEvUnsubscribe::EventType: diff --git a/library/cpp/actors/interconnect/interconnect_tcp_session.cpp b/library/cpp/actors/interconnect/interconnect_tcp_session.cpp index 8767bd68ce..2ded7f9f53 100644 --- a/library/cpp/actors/interconnect/interconnect_tcp_session.cpp +++ b/library/cpp/actors/interconnect/interconnect_tcp_session.cpp @@ -82,8 +82,8 @@ namespace NActors { IActor::InvokeOtherActor(*Proxy, &TInterconnectProxyTCP::UnregisterSession, this); ShutdownSocket(std::move(reason)); - for (const auto& kv : Subscribers) { - Send(kv.first, new TEvInterconnect::TEvNodeDisconnected(Proxy->PeerNodeId), 0, kv.second); + for (const auto& kv : Subscribers) { + Send(kv.first, new TEvInterconnect::TEvNodeDisconnected(Proxy->PeerNodeId), 0, kv.second); } Proxy->Metrics->SubSubscribersCount(Subscribers.size()); Subscribers.clear(); @@ -186,13 +186,13 @@ namespace NActors { void TInterconnectSessionTCP::Subscribe(STATEFN_SIG) { LOG_DEBUG_IC_SESSION("ICS04", "subscribe for session state for %s", ev->Sender.ToString().data()); - const auto [it, inserted] = Subscribers.emplace(ev->Sender, ev->Cookie); + const auto [it, inserted] = Subscribers.emplace(ev->Sender, ev->Cookie); if (inserted) { Proxy->Metrics->IncSubscribersCount(); - } else { - it->second = ev->Cookie; + } else { + it->second = ev->Cookie; } - Send(ev->Sender, new TEvInterconnect::TEvNodeConnected(Proxy->PeerNodeId), 0, ev->Cookie); + Send(ev->Sender, new TEvInterconnect::TEvNodeConnected(Proxy->PeerNodeId), 0, ev->Cookie); } void TInterconnectSessionTCP::Unsubscribe(STATEFN_SIG) { diff --git a/library/cpp/actors/interconnect/interconnect_tcp_session.h b/library/cpp/actors/interconnect/interconnect_tcp_session.h index 76357187b0..7fc00dbcc5 100644 --- a/library/cpp/actors/interconnect/interconnect_tcp_session.h +++ b/library/cpp/actors/interconnect/interconnect_tcp_session.h @@ -477,7 +477,7 @@ namespace NActors { ui32 SendBufferSize; ui64 InflightDataAmount = 0; - std::unordered_map<TActorId, ui64, TActorId::THash> Subscribers; + std::unordered_map<TActorId, ui64, TActorId::THash> Subscribers; // time at which we want to send confirmation packet even if there was no outgoing data ui64 UnconfirmedBytes = 0; diff --git a/library/cpp/actors/interconnect/mock/ic_mock.cpp b/library/cpp/actors/interconnect/mock/ic_mock.cpp index 275a02e551..884503e602 100644 --- a/library/cpp/actors/interconnect/mock/ic_mock.cpp +++ b/library/cpp/actors/interconnect/mock/ic_mock.cpp @@ -95,7 +95,7 @@ namespace NActors { class TProxyMockActor : public TActor<TProxyMockActor> { class TSessionMockActor : public TActor<TSessionMockActor> { - std::map<TActorId, ui64> Subscribers; + std::map<TActorId, ui64> Subscribers; TProxyMockActor* const Proxy; std::deque<std::unique_ptr<IEventHandle>> Queue; @@ -113,8 +113,8 @@ namespace NActors { for (auto&& ev : std::exchange(Queue, {})) { TActivationContext::Send(ev->ForwardOnNondelivery(TEvents::TEvUndelivered::Disconnected)); } - for (const auto& kv : Subscribers) { - Send(kv.first, new TEvInterconnect::TEvNodeDisconnected(Proxy->PeerNodeId), 0, kv.second); + for (const auto& kv : Subscribers) { + Send(kv.first, new TEvInterconnect::TEvNodeDisconnected(Proxy->PeerNodeId), 0, kv.second); } Y_VERIFY(Proxy->Session == this); Proxy->Session = nullptr; @@ -123,7 +123,7 @@ namespace NActors { void HandleForward(TAutoPtr<IEventHandle> ev) { if (ev->Flags & IEventHandle::FlagSubscribeOnSession) { - Subscribe(ev->Sender, ev->Cookie); + Subscribe(ev->Sender, ev->Cookie); } if (Queue.empty()) { TActivationContext::Send(new IEventHandle(EvRam, 0, SelfId(), {}, {}, 0)); @@ -140,11 +140,11 @@ namespace NActors { } void Handle(TEvInterconnect::TEvConnectNode::TPtr ev) { - Subscribe(ev->Sender, ev->Cookie); + Subscribe(ev->Sender, ev->Cookie); } void Handle(TEvents::TEvSubscribe::TPtr ev) { - Subscribe(ev->Sender, ev->Cookie); + Subscribe(ev->Sender, ev->Cookie); } void Handle(TEvents::TEvUnsubscribe::TPtr ev) { @@ -165,9 +165,9 @@ namespace NActors { ) private: - void Subscribe(const TActorId& actorId, ui64 cookie) { - Subscribers[actorId] = cookie; - Send(actorId, new TEvInterconnect::TEvNodeConnected(Proxy->PeerNodeId), 0, cookie); + void Subscribe(const TActorId& actorId, ui64 cookie) { + Subscribers[actorId] = cookie; + Send(actorId, new TEvInterconnect::TEvNodeConnected(Proxy->PeerNodeId), 0, cookie); } }; 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 5b3c8f1dcc..2b6d27cd3f 100644 --- a/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h +++ b/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h @@ -25,7 +25,7 @@ private: THashMap<ui32, THolder<TNode>> Nodes; TList<TTrafficInterrupter> interrupters; NActors::TChannelsConfig ChannelsConfig; - TPortManager PortManager; + TPortManager PortManager; public: TTestICCluster(ui32 numNodes = 1, NActors::TChannelsConfig channelsConfig = NActors::TChannelsConfig(), @@ -38,7 +38,7 @@ public: THashMap<ui32, THashMap<ui32, ui16>> specificNodePortMap; for (ui32 i = 1; i <= NumNodes; ++i) { - nodeToPortMap.emplace(i, PortManager.GetPort()); + nodeToPortMap.emplace(i, PortManager.GetPort()); } if (tiSettings) { @@ -48,7 +48,7 @@ public: for (auto& item : nodeToPortMap) { nodeId = item.first; listenPort = item.second; - forwardPort = PortManager.GetPort(); + forwardPort = PortManager.GetPort(); specificNodePortMap[nodeId] = nodeToPortMap; specificNodePortMap[nodeId].at(nodeId) = forwardPort; diff --git a/library/cpp/actors/interconnect/ut/lib/node.h b/library/cpp/actors/interconnect/ut/lib/node.h index 12087fbe32..ff30b1445e 100644 --- a/library/cpp/actors/interconnect/ut/lib/node.h +++ b/library/cpp/actors/interconnect/ut/lib/node.h @@ -4,7 +4,7 @@ #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/dnsresolver/dnsresolver.h> #include <library/cpp/actors/interconnect/interconnect_tcp_server.h> #include <library/cpp/actors/interconnect/interconnect_tcp_proxy.h> @@ -91,7 +91,7 @@ public: for (ui32 i = 1; i <= numNodes; ++i) { names->StaticNodeTable[i] = TTableNameserverSetup::TNodeInfo(address, address, nodeToPort.at(i)); } - setup.LocalServices.emplace_back( + setup.LocalServices.emplace_back( NDnsResolver::MakeDnsResolverActorId(), TActorSetupCmd( NDnsResolver::CreateOnDemandDnsResolver(), @@ -111,7 +111,7 @@ public: } ~TNode() { - ActorSystem->Stop(); + ActorSystem->Stop(); } bool Send(const TActorId& recipient, IEventBase* ev) { diff --git a/library/cpp/actors/interconnect/ya.make b/library/cpp/actors/interconnect/ya.make index 01a57c3aa2..60d29b0fc0 100644 --- a/library/cpp/actors/interconnect/ya.make +++ b/library/cpp/actors/interconnect/ya.make @@ -33,7 +33,7 @@ SRCS( interconnect_nameserver_table.cpp interconnect_proxy_wrapper.cpp interconnect_proxy_wrapper.h - interconnect_resolve.cpp + interconnect_resolve.cpp interconnect_stream.cpp interconnect_stream.h interconnect_tcp_input_session.cpp @@ -76,7 +76,7 @@ PEERDIR( contrib/libs/openssl library/cpp/actors/core library/cpp/actors/dnscachelib - library/cpp/actors/dnsresolver + library/cpp/actors/dnsresolver library/cpp/actors/helpers library/cpp/actors/prof library/cpp/actors/protos diff --git a/library/cpp/actors/testlib/test_runtime.cpp b/library/cpp/actors/testlib/test_runtime.cpp index cf93742d21..6fa25b9965 100644 --- a/library/cpp/actors/testlib/test_runtime.cpp +++ b/library/cpp/actors/testlib/test_runtime.cpp @@ -58,7 +58,7 @@ namespace NActors { TTestActorRuntimeBase::TNodeDataBase::TNodeDataBase() { ActorSystemTimestamp = nullptr; - ActorSystemMonotonic = nullptr; + ActorSystemMonotonic = nullptr; } void TTestActorRuntimeBase::TNodeDataBase::Stop() { @@ -210,10 +210,10 @@ namespace NActors { return Scheduled.empty(); } - TInstant TEventMailBox::GetFirstScheduleDeadline() const { - return Scheduled.begin()->Deadline; - } - + TInstant TEventMailBox::GetFirstScheduleDeadline() const { + return Scheduled.begin()->Deadline; + } + ui64 TEventMailBox::GetSentEventCount() const { return Sent.size(); } @@ -245,7 +245,7 @@ namespace NActors { void Prepare(TActorSystem *actorSystem, volatile ui64 *currentTimestamp, volatile ui64 *currentMonotonic) override { Y_UNUSED(actorSystem); Node->ActorSystemTimestamp = currentTimestamp; - Node->ActorSystemMonotonic = currentMonotonic; + Node->ActorSystemMonotonic = currentMonotonic; } void PrepareSchedules(NSchedulerQueue::TReader **readers, ui32 scheduleReadersCount) override { @@ -299,8 +299,8 @@ namespace NActors { void Schedule(TMonotonic deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie *cookie, TWorkerId workerId) override { DoSchedule(TInstant::FromValue(deadline.GetValue()), ev, cookie, workerId); - } - + } + void Schedule(TDuration delay, TAutoPtr<IEventHandle> ev, ISchedulerCookie *cookie, TWorkerId workerId) override { TInstant deadline = Runtime->GetTimeProvider()->Now() + delay; DoSchedule(deadline, ev, cookie, workerId); @@ -467,7 +467,7 @@ namespace NActors { , LocalId(0) , DispatchCyclesCount(0) , DispatchedEventsCount(0) - , NeedMonitoring(false) + , NeedMonitoring(false) , RandomProvider(CreateDeterministicRandomProvider(DefaultRandomSeed)) , TimeProvider(new TTimeProvider(*this)) , ShouldContinue() @@ -490,7 +490,7 @@ namespace NActors { node->LogSettings = new NActors::NLog::TSettings(loggerActorId, 410 /* NKikimrServices::LOGGER */, NActors::NLog::PRI_WARN, NActors::NLog::PRI_WARN, 0); node->LogSettings->SetAllowDrop(false); - node->LogSettings->SetThrottleDelay(TDuration::Zero()); + node->LogSettings->SetThrottleDelay(TDuration::Zero()); node->DynamicCounters = new NMonitoring::TDynamicCounters; InitNodeImpl(node, nodeIndex); @@ -572,9 +572,9 @@ namespace NActors { bool TTestActorRuntimeBase::NopFilterFunc(TTestActorRuntimeBase& runtime, TAutoPtr<IEventHandle>& event, TDuration delay, TInstant& deadline) { Y_UNUSED(runtime); - Y_UNUSED(delay); + Y_UNUSED(delay); Y_UNUSED(event); - Y_UNUSED(deadline); + Y_UNUSED(deadline); return true; } @@ -843,14 +843,14 @@ namespace NActors { } ui32 TTestActorRuntimeBase::InterconnectPoolId() const { - if (UseRealThreads && NSan::TSanIsOn()) { - // Interconnect coroutines may move across threads - // Use a special single-threaded pool to avoid that - return 4; - } - return 0; - } - + if (UseRealThreads && NSan::TSanIsOn()) { + // Interconnect coroutines may move across threads + // Use a special single-threaded pool to avoid that + return 4; + } + return 0; + } + TString TTestActorRuntimeBase::GetTempDir() { if (!TmpDir) TmpDir.Reset(new TTempDir()); @@ -967,7 +967,7 @@ namespace NActors { Y_VERIFY(nodeIndex < NodeCount); TActorId edgeActor = Register(new TEdgeActor(this), nodeIndex); EdgeActors.insert(edgeActor); - EdgeActorByMailbox[TEventMailboxId(edgeActor.NodeId(), edgeActor.Hint())] = edgeActor; + EdgeActorByMailbox[TEventMailboxId(edgeActor.NodeId(), edgeActor.Hint())] = edgeActor; return edgeActor; } @@ -1161,15 +1161,15 @@ namespace NActors { if (mbox.second->IsActive(TInstant::MicroSeconds(CurrentTimestamp))) { bool isEdgeMailbox = false; - if (EdgeActorByMailbox.FindPtr(TEventMailboxId(mbox.first.NodeId, mbox.first.Hint))) { - isEdgeMailbox = true; - TEventsList events; - mbox.second->Capture(events); - for (auto& ev : events) { - TInverseGuard<TMutex> inverseGuard(Mutex); - ObserverFunc(*this, ev); + if (EdgeActorByMailbox.FindPtr(TEventMailboxId(mbox.first.NodeId, mbox.first.Hint))) { + isEdgeMailbox = true; + TEventsList events; + mbox.second->Capture(events); + for (auto& ev : events) { + TInverseGuard<TMutex> inverseGuard(Mutex); + ObserverFunc(*this, ev); } - mbox.second->PushFront(events); + mbox.second->PushFront(events); } if (!isEdgeMailbox) { @@ -1246,8 +1246,8 @@ namespace NActors { } } - if (localContext.FinalEventFound) { - return true; + if (localContext.FinalEventFound) { + return true; } if (!localContext.FoundNonEmptyMailboxes.empty()) @@ -1287,8 +1287,8 @@ namespace NActors { inspectScheduledEventsAt = dispatchTime + scheduledEventsInspectInterval; bool isEmpty = true; TMaybe<TInstant> nearestMailboxDeadline; - TVector<TIntrusivePtr<TEventMailBox>> nextScheduleMboxes; - TMaybe<TInstant> nextScheduleDeadline; + TVector<TIntrusivePtr<TEventMailBox>> nextScheduleMboxes; + TMaybe<TInstant> nextScheduleDeadline; for (auto& mbox : currentMailboxes) { if (!mbox.second->IsActive(TInstant::MicroSeconds(CurrentTimestamp))) { if (!nearestMailboxDeadline.Defined() || *nearestMailboxDeadline.Get() > mbox.second->GetInactiveUntil()) { @@ -1301,29 +1301,29 @@ namespace NActors { if (mbox.second->IsScheduledEmpty()) continue; - auto firstScheduleDeadline = mbox.second->GetFirstScheduleDeadline(); - if (!nextScheduleDeadline || firstScheduleDeadline < *nextScheduleDeadline) { - nextScheduleMboxes.clear(); - nextScheduleMboxes.emplace_back(mbox.second); - nextScheduleDeadline = firstScheduleDeadline; - } else if (firstScheduleDeadline == *nextScheduleDeadline) { - nextScheduleMboxes.emplace_back(mbox.second); - } - } - - for (const auto& nextScheduleMbox : nextScheduleMboxes) { + auto firstScheduleDeadline = mbox.second->GetFirstScheduleDeadline(); + if (!nextScheduleDeadline || firstScheduleDeadline < *nextScheduleDeadline) { + nextScheduleMboxes.clear(); + nextScheduleMboxes.emplace_back(mbox.second); + nextScheduleDeadline = firstScheduleDeadline; + } else if (firstScheduleDeadline == *nextScheduleDeadline) { + nextScheduleMboxes.emplace_back(mbox.second); + } + } + + for (const auto& nextScheduleMbox : nextScheduleMboxes) { TEventsList selectedEvents; TScheduledEventsList capturedScheduledEvents; - nextScheduleMbox->CaptureScheduled(capturedScheduledEvents); + nextScheduleMbox->CaptureScheduled(capturedScheduledEvents); ScheduledEventsSelectorFunc(*this, capturedScheduledEvents, selectedEvents); - nextScheduleMbox->PushScheduled(capturedScheduledEvents); + nextScheduleMbox->PushScheduled(capturedScheduledEvents); for (auto& event : selectedEvents) { if (verbose && (BlockedOutput.find(event->Sender) == BlockedOutput.end())) { Cerr << "Selected scheduled event at " << TInstant::MicroSeconds(CurrentTimestamp) << ", "; PrintEvent(event, this); } - nextScheduleMbox->Send(event); + nextScheduleMbox->Send(event); isEmpty = false; } } @@ -1371,10 +1371,10 @@ namespace NActors { while (context) { for (const auto& finalEvent : context->Options->FinalEvents) { if (finalEvent.EventCheck(ev)) { - auto& freq = context->FinalEventFrequency[&finalEvent]; - if (++freq >= finalEvent.RequiredCount) { - context->FinalEventFound = true; - } + auto& freq = context->FinalEventFrequency[&finalEvent]; + if (++freq >= finalEvent.RequiredCount) { + context->FinalEventFound = true; + } } } @@ -1424,15 +1424,15 @@ namespace NActors { void TTestActorRuntimeBase::WaitForEdgeEvents(TEventFilter filter, const TSet<TActorId>& edgeFilter, TDuration simTimeout) { TGuard<TMutex> guard(Mutex); ui32 dispatchCount = 0; - if (!edgeFilter.empty()) { - for (auto edgeActor : edgeFilter) { + if (!edgeFilter.empty()) { + for (auto edgeActor : edgeFilter) { Y_VERIFY(EdgeActors.contains(edgeActor), "%s is not an edge actor", ToString(edgeActor).data()); - } - } + } + } const TSet<TActorId>& edgeActors = edgeFilter.empty() ? EdgeActors : edgeFilter; - TInstant deadline = TInstant::MicroSeconds(CurrentTimestamp) + simTimeout; + TInstant deadline = TInstant::MicroSeconds(CurrentTimestamp) + simTimeout; for (;;) { - for (auto edgeActor : edgeActors) { + for (auto edgeActor : edgeActors) { TEventsList events; auto& mbox = GetMailbox(edgeActor.NodeId(), edgeActor.Hint()); bool foundEvent = false; @@ -1451,7 +1451,7 @@ namespace NActors { ++dispatchCount; { - if (!DispatchEventsInternal(TDispatchOptions(), deadline)) { + if (!DispatchEventsInternal(TDispatchOptions(), deadline)) { return; // Timed out; event was not found } } @@ -1522,8 +1522,8 @@ namespace NActors { return node->DynamicCounters; } - void TTestActorRuntimeBase::SetupMonitoring() { - NeedMonitoring = true; + void TTestActorRuntimeBase::SetupMonitoring() { + NeedMonitoring = true; } void TTestActorRuntimeBase::SendInternal(IEventHandle* ev, ui32 nodeIndex, bool viaActorSystem) { @@ -1612,13 +1612,13 @@ namespace NActors { setup->NodeId = FirstNodeId + nodeIndex; if (UseRealThreads) { - setup->ExecutorsCount = 5; - setup->Executors.Reset(new TAutoPtr<IExecutorPool>[5]); + setup->ExecutorsCount = 5; + setup->Executors.Reset(new TAutoPtr<IExecutorPool>[5]); setup->Executors[0].Reset(new TBasicExecutorPool(0, 2, 20)); setup->Executors[1].Reset(new TBasicExecutorPool(1, 2, 20)); setup->Executors[2].Reset(new TIOExecutorPool(2, 1)); setup->Executors[3].Reset(new TBasicExecutorPool(3, 2, 20)); - setup->Executors[4].Reset(new TBasicExecutorPool(4, 1, 20)); + setup->Executors[4].Reset(new TBasicExecutorPool(4, 1, 20)); setup->Scheduler.Reset(new TBasicSchedulerThread(TSchedulerConfig(512, 100))); } else { setup->ExecutorsCount = 1; @@ -1652,14 +1652,14 @@ namespace NActors { common->MonCounters = interconnectCounters; common->TechnicalSelfHostName = "::1"; - if (!UseRealThreads) { + if (!UseRealThreads) { common->Settings.DeadPeer = TDuration::Max(); common->Settings.CloseOnIdle = TDuration::Max(); common->Settings.PingPeriod = TDuration::Max(); common->Settings.ForceConfirmPeriod = TDuration::Max(); common->Settings.Handshake = TDuration::Max(); - } - + } + common->ClusterUUID = ClusterUUID; common->AcceptUUID = {ClusterUUID}; @@ -1677,7 +1677,7 @@ namespace NActors { } setup->Interconnect.ProxyWrapperFactory = CreateProxyWrapperFactory(common, InterconnectPoolId(), &InterconnectMock); - + if (UseRealInterconnect) { setup->LocalServices.emplace_back(MakePollerActorId(), NActors::TActorSetupCmd(CreatePollerActor(), NActors::TMailboxType::Simple, InterconnectPoolId())); diff --git a/library/cpp/actors/testlib/test_runtime.h b/library/cpp/actors/testlib/test_runtime.h index 387b386df3..26e3b45c98 100644 --- a/library/cpp/actors/testlib/test_runtime.h +++ b/library/cpp/actors/testlib/test_runtime.h @@ -16,7 +16,7 @@ #include <util/datetime/base.h> #include <util/folder/tempdir.h> #include <util/generic/deque.h> -#include <util/generic/hash.h> +#include <util/generic/hash.h> #include <util/generic/noncopyable.h> #include <util/generic/ptr.h> #include <util/generic/queue.h> @@ -62,12 +62,12 @@ namespace NActors { return (NodeId == other.NodeId) && (Hint == other.Hint); } - struct THash { - ui64 operator()(const TEventMailboxId& mboxId) const noexcept { - return mboxId.NodeId * 31ULL + mboxId.Hint; - } - }; - + struct THash { + ui64 operator()(const TEventMailboxId& mboxId) const noexcept { + return mboxId.NodeId * 31ULL + mboxId.Hint; + } + }; + ui32 NodeId; ui32 Hint; }; @@ -150,7 +150,7 @@ namespace NActors { TInstant GetInactiveUntil() const; void Schedule(const TScheduledEventQueueItem& item); bool IsScheduledEmpty() const; - TInstant GetFirstScheduleDeadline() const; + TInstant GetFirstScheduleDeadline() const; ui64 GetSentEventCount() const; private: @@ -164,7 +164,7 @@ namespace NActors { #endif }; - typedef THashMap<TEventMailboxId, TIntrusivePtr<TEventMailBox>, TEventMailboxId::THash> TEventMailBoxList; + typedef THashMap<TEventMailboxId, TIntrusivePtr<TEventMailBox>, TEventMailboxId::THash> TEventMailBoxList; class TEmptyEventQueueException : public yexception { public: @@ -237,7 +237,7 @@ namespace NActors { ui32 GetNodeId(ui32 index = 0) const; ui32 GetNodeCount() const; ui64 AllocateLocalId(); - ui32 InterconnectPoolId() const; + ui32 InterconnectPoolId() const; TString GetTempDir(); TActorId Register(IActor* actor, ui32 nodeIndex = 0, ui32 poolId = 0, TMailboxType::EType mailboxType = TMailboxType::Simple, ui64 revolvingCounter = 0, @@ -268,7 +268,7 @@ namespace NActors { void EnableScheduleForActor(const TActorId& actorId, bool allow = true); bool IsScheduleForActorEnabled(const TActorId& actorId) const; TIntrusivePtr<NMonitoring::TDynamicCounters> GetDynamicCounters(ui32 nodeIndex = 0); - void SetupMonitoring(); + void SetupMonitoring(); template<typename T> void AppendToLogSettings(NLog::EComponent minVal, NLog::EComponent maxVal, T func) { @@ -303,7 +303,7 @@ namespace NActors { } return false; - }, {}, simTimeout); + }, {}, simTimeout); if (simTimeout == TDuration::Max()) Y_VERIFY(handle); @@ -315,44 +315,44 @@ namespace NActors { } } - template<class TEvent> - typename TEvent::TPtr GrabEdgeEventIf( + template<class TEvent> + typename TEvent::TPtr GrabEdgeEventIf( const TSet<TActorId>& edgeFilter, - const std::function<bool(const typename TEvent::TPtr&)>& predicate, - TDuration simTimeout = TDuration::Max()) - { - typename TEvent::TPtr handle; - const ui32 eventType = TEvent::EventType; + const std::function<bool(const typename TEvent::TPtr&)>& predicate, + TDuration simTimeout = TDuration::Max()) + { + typename TEvent::TPtr handle; + const ui32 eventType = TEvent::EventType; WaitForEdgeEvents([&](TTestActorRuntimeBase& runtime, TAutoPtr<IEventHandle>& event) { - Y_UNUSED(runtime); - if (event->GetTypeRewrite() != eventType) - return false; - - typename TEvent::TPtr* typedEvent = reinterpret_cast<typename TEvent::TPtr*>(&event); - if (predicate(*typedEvent)) { - handle = *typedEvent; - return true; - } - - return false; - }, edgeFilter, simTimeout); - - if (simTimeout == TDuration::Max()) - Y_VERIFY(handle); - - return handle; - } - - template<class TEvent> - typename TEvent::TPtr GrabEdgeEventIf( + Y_UNUSED(runtime); + if (event->GetTypeRewrite() != eventType) + return false; + + typename TEvent::TPtr* typedEvent = reinterpret_cast<typename TEvent::TPtr*>(&event); + if (predicate(*typedEvent)) { + handle = *typedEvent; + return true; + } + + return false; + }, edgeFilter, simTimeout); + + if (simTimeout == TDuration::Max()) + Y_VERIFY(handle); + + return handle; + } + + template<class TEvent> + typename TEvent::TPtr GrabEdgeEventIf( const TActorId& edgeActor, - const std::function<bool(const typename TEvent::TPtr&)>& predicate, - TDuration simTimeout = TDuration::Max()) - { + const std::function<bool(const typename TEvent::TPtr&)>& predicate, + TDuration simTimeout = TDuration::Max()) + { TSet<TActorId> edgeFilter{edgeActor}; - return GrabEdgeEventIf<TEvent>(edgeFilter, predicate, simTimeout); - } - + return GrabEdgeEventIf<TEvent>(edgeFilter, predicate, simTimeout); + } + template <typename TEvent> TEvent* GrabEdgeEvent(TAutoPtr<IEventHandle>& handle, TDuration simTimeout = TDuration::Max()) { std::function<bool(const TEvent&)> truth = [](const TEvent&) { return true; }; @@ -367,17 +367,17 @@ namespace NActors { return THolder(handle ? handle->Release<TEvent>().Release() : nullptr); } - template<class TEvent> + template<class TEvent> typename TEvent::TPtr GrabEdgeEvent(const TSet<TActorId>& edgeFilter, TDuration simTimeout = TDuration::Max()) { - return GrabEdgeEventIf<TEvent>(edgeFilter, [](const typename TEvent::TPtr&) { return true; }, simTimeout); - } - - template<class TEvent> + return GrabEdgeEventIf<TEvent>(edgeFilter, [](const typename TEvent::TPtr&) { return true; }, simTimeout); + } + + template<class TEvent> typename TEvent::TPtr GrabEdgeEvent(const TActorId& edgeActor, TDuration simTimeout = TDuration::Max()) { TSet<TActorId> edgeFilter{edgeActor}; - return GrabEdgeEvent<TEvent>(edgeFilter, simTimeout); - } - + return GrabEdgeEvent<TEvent>(edgeFilter, simTimeout); + } + // replace with std::variant<> template <typename... TEvents> std::tuple<TEvents*...> GrabEdgeEvents(TAutoPtr<IEventHandle>& handle, TDuration simTimeout = TDuration::Max()) { @@ -388,7 +388,7 @@ namespace NActors { return false; handle = event; return true; - }, {}, simTimeout); + }, {}, simTimeout); if (simTimeout == TDuration::Max()) Y_VERIFY(handle); if (handle) { @@ -408,24 +408,24 @@ namespace NActors { } } - template<class TEvent> + template<class TEvent> typename TEvent::TPtr GrabEdgeEventRethrow(const TSet<TActorId>& edgeFilter, TDuration simTimeout = TDuration::Max()) { - try { - return GrabEdgeEvent<TEvent>(edgeFilter, simTimeout); - } catch (...) { - ythrow TWithBackTrace<yexception>() << "Exception occured while waiting for " << TypeName<TEvent>() << ": " << CurrentExceptionMessage(); - } - } - - template<class TEvent> + try { + return GrabEdgeEvent<TEvent>(edgeFilter, simTimeout); + } catch (...) { + ythrow TWithBackTrace<yexception>() << "Exception occured while waiting for " << TypeName<TEvent>() << ": " << CurrentExceptionMessage(); + } + } + + template<class TEvent> typename TEvent::TPtr GrabEdgeEventRethrow(const TActorId& edgeActor, TDuration simTimeout = TDuration::Max()) { - try { - return GrabEdgeEvent<TEvent>(edgeActor, simTimeout); - } catch (...) { - ythrow TWithBackTrace<yexception>() << "Exception occured while waiting for " << TypeName<TEvent>() << ": " << CurrentExceptionMessage(); - } - } - + try { + return GrabEdgeEvent<TEvent>(edgeActor, simTimeout); + } catch (...) { + ythrow TWithBackTrace<yexception>() << "Exception occured while waiting for " << TypeName<TEvent>() << ": " << CurrentExceptionMessage(); + } + } + template <typename... TEvents> static TString TypeNames() { static TString names[] = { TypeName<TEvents>()... }; @@ -530,7 +530,7 @@ namespace NActors { ui64 DispatcherRandomSeed; TIntrusivePtr<IRandomProvider> DispatcherRandomProvider; TAutoPtr<TLogBackend> LogBackend; - bool NeedMonitoring; + bool NeedMonitoring; TIntrusivePtr<IRandomProvider> RandomProvider; TIntrusivePtr<ITimeProvider> TimeProvider; @@ -558,7 +558,7 @@ namespace NActors { TIntrusivePtr<NActors::NLog::TSettings> LogSettings; TIntrusivePtr<NInterconnect::TPollerThreads> Poller; volatile ui64* ActorSystemTimestamp; - volatile ui64* ActorSystemMonotonic; + volatile ui64* ActorSystemMonotonic; TVector<std::pair<TActorId, TActorSetupCmd> > LocalServices; TMap<TActorId, IActor*> LocalServicesActors; TMap<IActor*, TActorId> ActorToActorId; @@ -607,7 +607,7 @@ namespace NActors { TMap<const TDispatchOptions::TFinalEventCondition*, ui32> FinalEventFrequency; TSet<TEventMailboxId> FoundNonEmptyMailboxes; - bool FinalEventFound = false; + bool FinalEventFound = false; }; TProgramShouldContinue ShouldContinue; diff --git a/library/cpp/actors/util/unordered_cache.h b/library/cpp/actors/util/unordered_cache.h index 6b1b4f838a..76f036c0cf 100644 --- a/library/cpp/actors/util/unordered_cache.h +++ b/library/cpp/actors/util/unordered_cache.h @@ -3,126 +3,126 @@ #include "defs.h" #include "queue_chunk.h" -template <typename T, ui32 Size = 512, ui32 ConcurrencyFactor = 1, typename TChunk = TQueueChunk<T, Size>> +template <typename T, ui32 Size = 512, ui32 ConcurrencyFactor = 1, typename TChunk = TQueueChunk<T, Size>> class TUnorderedCache : TNonCopyable { static_assert(std::is_integral<T>::value || std::is_pointer<T>::value, "expect std::is_integral<T>::value || std::is_pointer<T>::value"); public: - static constexpr ui32 Concurrency = ConcurrencyFactor * 4; + static constexpr ui32 Concurrency = ConcurrencyFactor * 4; private: - struct TReadSlot { - TChunk* volatile ReadFrom; - volatile ui32 ReadPosition; - char Padding[64 - sizeof(TChunk*) - sizeof(ui32)]; // 1 slot per cache line - }; - - struct TWriteSlot { - TChunk* volatile WriteTo; - volatile ui32 WritePosition; - char Padding[64 - sizeof(TChunk*) - sizeof(ui32)]; // 1 slot per cache line - }; - - static_assert(sizeof(TReadSlot) == 64, "expect sizeof(TReadSlot) == 64"); - static_assert(sizeof(TWriteSlot) == 64, "expect sizeof(TWriteSlot) == 64"); - -private: - TReadSlot ReadSlots[Concurrency]; - TWriteSlot WriteSlots[Concurrency]; - + struct TReadSlot { + TChunk* volatile ReadFrom; + volatile ui32 ReadPosition; + char Padding[64 - sizeof(TChunk*) - sizeof(ui32)]; // 1 slot per cache line + }; + + struct TWriteSlot { + TChunk* volatile WriteTo; + volatile ui32 WritePosition; + char Padding[64 - sizeof(TChunk*) - sizeof(ui32)]; // 1 slot per cache line + }; + + static_assert(sizeof(TReadSlot) == 64, "expect sizeof(TReadSlot) == 64"); + static_assert(sizeof(TWriteSlot) == 64, "expect sizeof(TWriteSlot) == 64"); + +private: + TReadSlot ReadSlots[Concurrency]; + TWriteSlot WriteSlots[Concurrency]; + static_assert(sizeof(TChunk*) == sizeof(TAtomic), "expect sizeof(TChunk*) == sizeof(TAtomic)"); -private: - struct TLockedWriter { - TWriteSlot* Slot; - TChunk* WriteTo; - - TLockedWriter() - : Slot(nullptr) - , WriteTo(nullptr) - { } - - TLockedWriter(TWriteSlot* slot, TChunk* writeTo) - : Slot(slot) - , WriteTo(writeTo) - { } - - ~TLockedWriter() noexcept { - Drop(); - } - - void Drop() { - if (Slot) { - AtomicStore(&Slot->WriteTo, WriteTo); - Slot = nullptr; - } - } - - TLockedWriter(const TLockedWriter&) = delete; - TLockedWriter& operator=(const TLockedWriter&) = delete; - - TLockedWriter(TLockedWriter&& rhs) - : Slot(rhs.Slot) - , WriteTo(rhs.WriteTo) - { - rhs.Slot = nullptr; - } - - TLockedWriter& operator=(TLockedWriter&& rhs) { - if (Y_LIKELY(this != &rhs)) { - Drop(); - Slot = rhs.Slot; - WriteTo = rhs.WriteTo; - rhs.Slot = nullptr; - } - return *this; - } - }; - -private: - TLockedWriter LockWriter(ui64 writerRotation) { - ui32 cycle = 0; +private: + struct TLockedWriter { + TWriteSlot* Slot; + TChunk* WriteTo; + + TLockedWriter() + : Slot(nullptr) + , WriteTo(nullptr) + { } + + TLockedWriter(TWriteSlot* slot, TChunk* writeTo) + : Slot(slot) + , WriteTo(writeTo) + { } + + ~TLockedWriter() noexcept { + Drop(); + } + + void Drop() { + if (Slot) { + AtomicStore(&Slot->WriteTo, WriteTo); + Slot = nullptr; + } + } + + TLockedWriter(const TLockedWriter&) = delete; + TLockedWriter& operator=(const TLockedWriter&) = delete; + + TLockedWriter(TLockedWriter&& rhs) + : Slot(rhs.Slot) + , WriteTo(rhs.WriteTo) + { + rhs.Slot = nullptr; + } + + TLockedWriter& operator=(TLockedWriter&& rhs) { + if (Y_LIKELY(this != &rhs)) { + Drop(); + Slot = rhs.Slot; + WriteTo = rhs.WriteTo; + rhs.Slot = nullptr; + } + return *this; + } + }; + +private: + TLockedWriter LockWriter(ui64 writerRotation) { + ui32 cycle = 0; for (;;) { - TWriteSlot* slot = &WriteSlots[writerRotation % Concurrency]; - if (AtomicLoad(&slot->WriteTo) != nullptr) { - if (TChunk* writeTo = AtomicSwap(&slot->WriteTo, nullptr)) { - return TLockedWriter(slot, writeTo); + TWriteSlot* slot = &WriteSlots[writerRotation % Concurrency]; + if (AtomicLoad(&slot->WriteTo) != nullptr) { + if (TChunk* writeTo = AtomicSwap(&slot->WriteTo, nullptr)) { + return TLockedWriter(slot, writeTo); } } - ++writerRotation; - - // Do a spinlock pause after a full cycle - if (++cycle == Concurrency) { - SpinLockPause(); - cycle = 0; - } - } + ++writerRotation; + + // Do a spinlock pause after a full cycle + if (++cycle == Concurrency) { + SpinLockPause(); + cycle = 0; + } + } } - void WriteOne(TLockedWriter& lock, T x) { + void WriteOne(TLockedWriter& lock, T x) { Y_VERIFY_DEBUG(x != 0); - const ui32 pos = AtomicLoad(&lock.Slot->WritePosition); + const ui32 pos = AtomicLoad(&lock.Slot->WritePosition); if (pos != TChunk::EntriesCount) { - AtomicStore(&lock.Slot->WritePosition, pos + 1); - AtomicStore(&lock.WriteTo->Entries[pos], x); + AtomicStore(&lock.Slot->WritePosition, pos + 1); + AtomicStore(&lock.WriteTo->Entries[pos], x); } else { - TChunk* next = new TChunk(); + TChunk* next = new TChunk(); AtomicStore(&next->Entries[0], x); - AtomicStore(&lock.Slot->WritePosition, 1u); - AtomicStore(&lock.WriteTo->Next, next); - lock.WriteTo = next; + AtomicStore(&lock.Slot->WritePosition, 1u); + AtomicStore(&lock.WriteTo->Next, next); + lock.WriteTo = next; } } public: - TUnorderedCache() { - for (ui32 i = 0; i < Concurrency; ++i) { - ReadSlots[i].ReadFrom = new TChunk(); - ReadSlots[i].ReadPosition = 0; + TUnorderedCache() { + for (ui32 i = 0; i < Concurrency; ++i) { + ReadSlots[i].ReadFrom = new TChunk(); + ReadSlots[i].ReadPosition = 0; - WriteSlots[i].WriteTo = ReadSlots[i].ReadFrom; - WriteSlots[i].WritePosition = 0; + WriteSlots[i].WriteTo = ReadSlots[i].ReadFrom; + WriteSlots[i].WritePosition = 0; } } @@ -130,44 +130,44 @@ public: Y_VERIFY(!Pop(0)); for (ui64 i = 0; i < Concurrency; ++i) { - if (ReadSlots[i].ReadFrom) { - delete ReadSlots[i].ReadFrom; - ReadSlots[i].ReadFrom = nullptr; + if (ReadSlots[i].ReadFrom) { + delete ReadSlots[i].ReadFrom; + ReadSlots[i].ReadFrom = nullptr; } - WriteSlots[i].WriteTo = nullptr; + WriteSlots[i].WriteTo = nullptr; } } - T Pop(ui64 readerRotation) noexcept { + T Pop(ui64 readerRotation) noexcept { ui64 readerIndex = readerRotation; const ui64 endIndex = readerIndex + Concurrency; for (; readerIndex != endIndex; ++readerIndex) { - TReadSlot* slot = &ReadSlots[readerIndex % Concurrency]; - if (AtomicLoad(&slot->ReadFrom) != nullptr) { - if (TChunk* readFrom = AtomicSwap(&slot->ReadFrom, nullptr)) { - const ui32 pos = AtomicLoad(&slot->ReadPosition); + TReadSlot* slot = &ReadSlots[readerIndex % Concurrency]; + if (AtomicLoad(&slot->ReadFrom) != nullptr) { + if (TChunk* readFrom = AtomicSwap(&slot->ReadFrom, nullptr)) { + const ui32 pos = AtomicLoad(&slot->ReadPosition); if (pos != TChunk::EntriesCount) { if (T ret = AtomicLoad(&readFrom->Entries[pos])) { - AtomicStore(&slot->ReadPosition, pos + 1); - AtomicStore(&slot->ReadFrom, readFrom); // release lock with same chunk - return ret; // found, return + AtomicStore(&slot->ReadPosition, pos + 1); + AtomicStore(&slot->ReadFrom, readFrom); // release lock with same chunk + return ret; // found, return } else { - AtomicStore(&slot->ReadFrom, readFrom); // release lock with same chunk + AtomicStore(&slot->ReadFrom, readFrom); // release lock with same chunk } } else if (TChunk* next = AtomicLoad(&readFrom->Next)) { if (T ret = AtomicLoad(&next->Entries[0])) { - AtomicStore(&slot->ReadPosition, 1u); - AtomicStore(&slot->ReadFrom, next); // release lock with next chunk + AtomicStore(&slot->ReadPosition, 1u); + AtomicStore(&slot->ReadFrom, next); // release lock with next chunk delete readFrom; return ret; } else { - AtomicStore(&slot->ReadPosition, 0u); - AtomicStore(&slot->ReadFrom, next); // release lock with new chunk + AtomicStore(&slot->ReadPosition, 0u); + AtomicStore(&slot->ReadFrom, next); // release lock with new chunk delete readFrom; } } else { - // nothing in old chunk and no next chunk, just release lock with old chunk - AtomicStore(&slot->ReadFrom, readFrom); + // nothing in old chunk and no next chunk, just release lock with old chunk + AtomicStore(&slot->ReadFrom, readFrom); } } } @@ -177,20 +177,20 @@ public: } void Push(T x, ui64 writerRotation) { - TLockedWriter lock = LockWriter(writerRotation); - WriteOne(lock, x); + TLockedWriter lock = LockWriter(writerRotation); + WriteOne(lock, x); } void PushBulk(T* x, ui32 xcount, ui64 writerRotation) { - for (;;) { - // Fill no more then one queue chunk per round + for (;;) { + // Fill no more then one queue chunk per round const ui32 xround = Min(xcount, (ui32)TChunk::EntriesCount); - { - TLockedWriter lock = LockWriter(writerRotation++); - for (T* end = x + xround; x != end; ++x) - WriteOne(lock, *x); - } + { + TLockedWriter lock = LockWriter(writerRotation++); + for (T* end = x + xround; x != end; ++x) + WriteOne(lock, *x); + } if (xcount <= TChunk::EntriesCount) break; diff --git a/library/cpp/actors/util/unordered_cache_ut.cpp b/library/cpp/actors/util/unordered_cache_ut.cpp index 8f15f9c3d5..37865f2f91 100644 --- a/library/cpp/actors/util/unordered_cache_ut.cpp +++ b/library/cpp/actors/util/unordered_cache_ut.cpp @@ -1,138 +1,138 @@ -#include "unordered_cache.h" - -#include <library/cpp/testing/unittest/registar.h> -#include <util/random/random.h> -#include <util/system/hp_timer.h> -#include <util/system/sanitizers.h> -#include <util/system/thread.h> - -Y_UNIT_TEST_SUITE(UnorderedCache) { - - void DoOnePushOnePop(ui64 count) { - TUnorderedCache<ui64> queue; - - ui64 readRotation = 0; - ui64 writeRotation = 0; - - auto popped = queue.Pop(readRotation++); - UNIT_ASSERT_VALUES_EQUAL(popped, 0u); - - for (ui64 i = 0; i < count; ++i) { - queue.Push(i + 1, writeRotation++); - popped = queue.Pop(readRotation++); - UNIT_ASSERT_VALUES_EQUAL(popped, i + 1); - - popped = queue.Pop(readRotation++); - UNIT_ASSERT_VALUES_EQUAL(popped, 0u); - } - } - - Y_UNIT_TEST(OnePushOnePop) { - DoOnePushOnePop(1); - } - - Y_UNIT_TEST(OnePushOnePop_Repeat1M) { - DoOnePushOnePop(1000000); - } - - /** - * Simplified thread spawning for testing - */ - class TWorkerThread : public ISimpleThread { - private: - std::function<void()> Func; - double Time = 0.0; - - public: - TWorkerThread(std::function<void()> func) - : Func(std::move(func)) - { } - - double GetTime() const { - return Time; - } - - static THolder<TWorkerThread> Spawn(std::function<void()> func) { - THolder<TWorkerThread> thread = MakeHolder<TWorkerThread>(std::move(func)); - thread->Start(); - return thread; - } - - private: - void* ThreadProc() noexcept override { - THPTimer timer; - Func(); - Time = timer.Passed(); - return nullptr; - } - }; - - void DoConcurrentPushPop(size_t threads, ui64 perThreadCount) { - // Concurrency factor 4 is up to 16 threads - TUnorderedCache<ui64, 512, 4> queue; - - auto workerFunc = [&](size_t threadIndex) { - ui64 readRotation = 0; - ui64 writeRotation = 0; - ui64 readsDone = 0; - ui64 writesDone = 0; - for (;;) { - bool canRead = readsDone < writesDone; - bool canWrite = writesDone < perThreadCount; - if (!canRead && !canWrite) { - break; - } - if (canRead && canWrite) { - // Randomly choose between read and write - if (RandomNumber<ui64>(2)) { - canRead = false; - } else { - canWrite = false; - } - } - if (canRead) { - ui64 popped = queue.Pop(readRotation++); - if (popped) { - ++readsDone; - } - } - if (canWrite) { - queue.Push(1 + writesDone * threads + threadIndex, writeRotation++); - ++writesDone; - } - } - }; - - TVector<THolder<TWorkerThread>> workers(threads); - for (size_t i = 0; i < threads; ++i) { - workers[i] = TWorkerThread::Spawn([workerFunc, i]() { - workerFunc(i); - }); - } - - double maxTime = 0; - for (size_t i = 0; i < threads; ++i) { - workers[i]->Join(); - maxTime = Max(maxTime, workers[i]->GetTime()); - } - - auto popped = queue.Pop(0); - UNIT_ASSERT_VALUES_EQUAL(popped, 0u); - - Cerr << "Concurrent with " << threads << " threads: " << maxTime << " seconds" << Endl; - } - - void DoConcurrentPushPop_3times(size_t threads, ui64 perThreadCount) { - for (size_t i = 0; i < 3; ++i) { - DoConcurrentPushPop(threads, perThreadCount); - } - } - - static constexpr ui64 PER_THREAD_COUNT = NSan::PlainOrUnderSanitizer(1000000, 100000); - - Y_UNIT_TEST(ConcurrentPushPop_1thread) { DoConcurrentPushPop_3times(1, PER_THREAD_COUNT); } - Y_UNIT_TEST(ConcurrentPushPop_2threads) { DoConcurrentPushPop_3times(2, PER_THREAD_COUNT); } - Y_UNIT_TEST(ConcurrentPushPop_4threads) { DoConcurrentPushPop_3times(4, PER_THREAD_COUNT); } - Y_UNIT_TEST(ConcurrentPushPop_8threads) { DoConcurrentPushPop_3times(8, PER_THREAD_COUNT); } - Y_UNIT_TEST(ConcurrentPushPop_16threads) { DoConcurrentPushPop_3times(16, PER_THREAD_COUNT); } -} +#include "unordered_cache.h" + +#include <library/cpp/testing/unittest/registar.h> +#include <util/random/random.h> +#include <util/system/hp_timer.h> +#include <util/system/sanitizers.h> +#include <util/system/thread.h> + +Y_UNIT_TEST_SUITE(UnorderedCache) { + + void DoOnePushOnePop(ui64 count) { + TUnorderedCache<ui64> queue; + + ui64 readRotation = 0; + ui64 writeRotation = 0; + + auto popped = queue.Pop(readRotation++); + UNIT_ASSERT_VALUES_EQUAL(popped, 0u); + + for (ui64 i = 0; i < count; ++i) { + queue.Push(i + 1, writeRotation++); + popped = queue.Pop(readRotation++); + UNIT_ASSERT_VALUES_EQUAL(popped, i + 1); + + popped = queue.Pop(readRotation++); + UNIT_ASSERT_VALUES_EQUAL(popped, 0u); + } + } + + Y_UNIT_TEST(OnePushOnePop) { + DoOnePushOnePop(1); + } + + Y_UNIT_TEST(OnePushOnePop_Repeat1M) { + DoOnePushOnePop(1000000); + } + + /** + * Simplified thread spawning for testing + */ + class TWorkerThread : public ISimpleThread { + private: + std::function<void()> Func; + double Time = 0.0; + + public: + TWorkerThread(std::function<void()> func) + : Func(std::move(func)) + { } + + double GetTime() const { + return Time; + } + + static THolder<TWorkerThread> Spawn(std::function<void()> func) { + THolder<TWorkerThread> thread = MakeHolder<TWorkerThread>(std::move(func)); + thread->Start(); + return thread; + } + + private: + void* ThreadProc() noexcept override { + THPTimer timer; + Func(); + Time = timer.Passed(); + return nullptr; + } + }; + + void DoConcurrentPushPop(size_t threads, ui64 perThreadCount) { + // Concurrency factor 4 is up to 16 threads + TUnorderedCache<ui64, 512, 4> queue; + + auto workerFunc = [&](size_t threadIndex) { + ui64 readRotation = 0; + ui64 writeRotation = 0; + ui64 readsDone = 0; + ui64 writesDone = 0; + for (;;) { + bool canRead = readsDone < writesDone; + bool canWrite = writesDone < perThreadCount; + if (!canRead && !canWrite) { + break; + } + if (canRead && canWrite) { + // Randomly choose between read and write + if (RandomNumber<ui64>(2)) { + canRead = false; + } else { + canWrite = false; + } + } + if (canRead) { + ui64 popped = queue.Pop(readRotation++); + if (popped) { + ++readsDone; + } + } + if (canWrite) { + queue.Push(1 + writesDone * threads + threadIndex, writeRotation++); + ++writesDone; + } + } + }; + + TVector<THolder<TWorkerThread>> workers(threads); + for (size_t i = 0; i < threads; ++i) { + workers[i] = TWorkerThread::Spawn([workerFunc, i]() { + workerFunc(i); + }); + } + + double maxTime = 0; + for (size_t i = 0; i < threads; ++i) { + workers[i]->Join(); + maxTime = Max(maxTime, workers[i]->GetTime()); + } + + auto popped = queue.Pop(0); + UNIT_ASSERT_VALUES_EQUAL(popped, 0u); + + Cerr << "Concurrent with " << threads << " threads: " << maxTime << " seconds" << Endl; + } + + void DoConcurrentPushPop_3times(size_t threads, ui64 perThreadCount) { + for (size_t i = 0; i < 3; ++i) { + DoConcurrentPushPop(threads, perThreadCount); + } + } + + static constexpr ui64 PER_THREAD_COUNT = NSan::PlainOrUnderSanitizer(1000000, 100000); + + Y_UNIT_TEST(ConcurrentPushPop_1thread) { DoConcurrentPushPop_3times(1, PER_THREAD_COUNT); } + Y_UNIT_TEST(ConcurrentPushPop_2threads) { DoConcurrentPushPop_3times(2, PER_THREAD_COUNT); } + Y_UNIT_TEST(ConcurrentPushPop_4threads) { DoConcurrentPushPop_3times(4, PER_THREAD_COUNT); } + Y_UNIT_TEST(ConcurrentPushPop_8threads) { DoConcurrentPushPop_3times(8, PER_THREAD_COUNT); } + Y_UNIT_TEST(ConcurrentPushPop_16threads) { DoConcurrentPushPop_3times(16, PER_THREAD_COUNT); } +} diff --git a/library/cpp/actors/util/ut/ya.make b/library/cpp/actors/util/ut/ya.make index 5546525c3a..3b08b77984 100644 --- a/library/cpp/actors/util/ut/ya.make +++ b/library/cpp/actors/util/ut/ya.make @@ -12,7 +12,7 @@ OWNER( SRCS( rope_ut.cpp - unordered_cache_ut.cpp + unordered_cache_ut.cpp ) END() diff --git a/library/cpp/actors/ya.make b/library/cpp/actors/ya.make index 7d7eb1fa85..737c7fbc18 100644 --- a/library/cpp/actors/ya.make +++ b/library/cpp/actors/ya.make @@ -3,7 +3,7 @@ RECURSE_FOR_TESTS(ut) RECURSE( log_backend core - dnsresolver + dnsresolver examples memory_log helpers |