intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

10 files changed, 495 insertions, 0 deletions

diff --git a/library/cpp/dns/README.md b/library/cpp/dns/README.md
new file mode 100644
index 00000000000..6659d73ea2f
--- /dev/null
+++ b/library/cpp/dns/README.md
@@ -0,0 +1,9 @@
+Overview
+===
+Библиотека кеширующего resolving-а - изначально писалась для имплементации neh http протокола, использующей корутины.
+Для предотвращения пробоя короткого стека корутин есть метод, предусматривающий вынос в отдельный тред собственно вызов функции резолвинга.
+Для предотвращения обращения к DNS серверам (использования вместо этого заранее заданных ip-адресов),
+предусмотрена ручка добавления alias-ов hosname -> ip-address (требование от метапоискового движка).
+
+Из-за того, что библиотка разрабатывалась под задачу максимально быстрого резолвинга добавлены слои кеширования результатов
+resoving-а, - возможности сбросить кеш для того, чтобы получить более свежие адреса для указанного host-а _нет_.
diff --git a/library/cpp/dns/cache.cpp b/library/cpp/dns/cache.cpp
new file mode 100644
index 00000000000..05c14e82fce
--- /dev/null
+++ b/library/cpp/dns/cache.cpp
@@ -0,0 +1,198 @@
+#include "cache.h"
+
+#include "thread.h"
+
+#include <util/system/tls.h>
+#include <util/system/info.h>
+#include <util/system/rwlock.h>
+#include <util/thread/singleton.h>
+#include <util/generic/singleton.h>
+#include <util/generic/hash.h>
+
+using namespace NDns;
+
+namespace {
+    struct TResolveTask {
+        enum EMethod {
+            Normal,
+            Threaded
+        };
+
+        inline TResolveTask(const TResolveInfo& info, EMethod method)
+            : Info(info)
+            , Method(method)
+        {
+        }
+
+        const TResolveInfo& Info;
+        const EMethod Method;
+    };
+
+    class IDns {
+    public:
+        virtual ~IDns() = default;
+        virtual const TResolvedHost* Resolve(const TResolveTask&) = 0;
+    };
+
+    typedef TAtomicSharedPtr<TResolvedHost> TResolvedHostPtr;
+
+    struct THashResolveInfo {
+        inline size_t operator()(const TResolveInfo& ri) const {
+            return ComputeHash(ri.Host) ^ ri.Port;
+        }
+    };
+
+    struct TCompareResolveInfo {
+        inline bool operator()(const NDns::TResolveInfo& x, const NDns::TResolveInfo& y) const {
+            return x.Host == y.Host && x.Port == y.Port;
+        }
+    };
+
+    class TGlobalCachedDns: public IDns, public TNonCopyable {
+    public:
+        const TResolvedHost* Resolve(const TResolveTask& rt) override {
+            //2. search host in cache
+            {
+                TReadGuard guard(L_);
+
+                TCache::const_iterator it = C_.find(rt.Info);
+
+                if (it != C_.end()) {
+                    return it->second.Get();
+                }
+            }
+
+            TResolvedHostPtr res = ResolveA(rt);
+
+            //update cache
+            {
+                TWriteGuard guard(L_);
+
+                std::pair<TCache::iterator, bool> updateResult = C_.insert(std::make_pair(TResolveInfo(res->Host, rt.Info.Port), res));
+                TResolvedHost* rh = updateResult.first->second.Get();
+
+                if (updateResult.second) {
+                    //fresh resolved host, set cache record id for it
+                    rh->Id = C_.size() - 1;
+                }
+
+                return rh;
+            }
+        }
+
+        void AddAlias(const TString& host, const TString& alias) noexcept {
+            TWriteGuard guard(LA_);
+
+            A_[host] = alias;
+        }
+
+        static inline TGlobalCachedDns* Instance() {
+            return SingletonWithPriority<TGlobalCachedDns, 65530>();
+        }
+
+    private:
+        inline TResolvedHostPtr ResolveA(const TResolveTask& rt) {
+            TString originalHost(rt.Info.Host);
+            TString host(originalHost);
+
+            //3. replace host to alias, if exist
+            if (A_.size()) {
+                TReadGuard guard(LA_);
+                TStringBuf names[] = {"*", host};
+
+                for (const auto& name : names) {
+                    TAliases::const_iterator it = A_.find(name);
+
+                    if (it != A_.end()) {
+                        host = it->second;
+                    }
+                }
+            }
+
+            if (host.length() > 2 && host[0] == '[') {
+                TString unbracedIpV6(host.data() + 1, host.size() - 2);
+                host.swap(unbracedIpV6);
+            }
+
+            TAutoPtr<TNetworkAddress> na;
+
+            //4. getaddrinfo (direct or in separate thread)
+            if (rt.Method == TResolveTask::Normal) {
+                na.Reset(new TNetworkAddress(host, rt.Info.Port));
+            } else if (rt.Method == TResolveTask::Threaded) {
+                na = ThreadedResolve(host, rt.Info.Port);
+            } else {
+                Y_ASSERT(0);
+                throw yexception() << TStringBuf("invalid resolve method");
+            }
+
+            return new TResolvedHost(originalHost, *na);
+        }
+
+        typedef THashMap<TResolveInfo, TResolvedHostPtr, THashResolveInfo, TCompareResolveInfo> TCache;
+        TCache C_;
+        TRWMutex L_;
+        typedef THashMap<TString, TString> TAliases;
+        TAliases A_;
+        TRWMutex LA_;
+    };
+
+    class TCachedDns: public IDns {
+    public:
+        inline TCachedDns(IDns* slave)
+            : S_(slave)
+        {
+        }
+
+        const TResolvedHost* Resolve(const TResolveTask& rt) override {
+            //1. search in local thread cache
+            {
+                TCache::const_iterator it = C_.find(rt.Info);
+
+                if (it != C_.end()) {
+                    return it->second;
+                }
+            }
+
+            const TResolvedHost* res = S_->Resolve(rt);
+
+            C_[TResolveInfo(res->Host, rt.Info.Port)] = res;
+
+            return res;
+        }
+
+    private:
+        typedef THashMap<TResolveInfo, const TResolvedHost*, THashResolveInfo, TCompareResolveInfo> TCache;
+        TCache C_;
+        IDns* S_;
+    };
+
+    struct TThreadedDns: public TCachedDns {
+        inline TThreadedDns()
+            : TCachedDns(TGlobalCachedDns::Instance())
+        {
+        }
+    };
+
+    inline IDns* ThrDns() {
+        return FastTlsSingleton<TThreadedDns>();
+    }
+}
+
+namespace NDns {
+    const TResolvedHost* CachedResolve(const TResolveInfo& ri) {
+        TResolveTask rt(ri, TResolveTask::Normal);
+
+        return ThrDns()->Resolve(rt);
+    }
+
+    const TResolvedHost* CachedThrResolve(const TResolveInfo& ri) {
+        TResolveTask rt(ri, TResolveTask::Threaded);
+
+        return ThrDns()->Resolve(rt);
+    }
+
+    void AddHostAlias(const TString& host, const TString& alias) {
+        TGlobalCachedDns::Instance()->AddAlias(host, alias);
+    }
+}
diff --git a/library/cpp/dns/cache.h b/library/cpp/dns/cache.h
new file mode 100644
index 00000000000..eda5dc40702
--- /dev/null
+++ b/library/cpp/dns/cache.h
@@ -0,0 +1,45 @@
+#pragma once
+
+#include <util/network/socket.h>
+#include <util/generic/strbuf.h>
+#include <util/generic/string.h>
+
+namespace NDns {
+    struct TResolveInfo {
+        inline TResolveInfo(const TStringBuf& host, ui16 port)
+            : Host(host)
+            , Port(port)
+        {
+        }
+
+        TStringBuf Host;
+        ui16 Port;
+    };
+
+    struct TResolvedHost {
+        inline TResolvedHost(const TString& host, const TNetworkAddress& addr) noexcept
+            : Host(host)
+            , Addr(addr)
+            , Id(0)
+        {
+        }
+
+        TString Host; //resolved hostname (from TResolveInfo, - before aliasing)
+        TNetworkAddress Addr;
+        size_t Id; //cache record id
+    };
+
+    // Resolving order:
+    //   1. check local thread cache, return if found
+    //   2. check global cache, return if found
+    //   3. search alias for hostname, if found, continue resolving alias
+    //   4. normal resolver
+    const TResolvedHost* CachedResolve(const TResolveInfo& ri);
+
+    //like previous, but at stage 4 use separate thread for resolving (created on first usage)
+    //useful in green-threads with tiny stack
+    const TResolvedHost* CachedThrResolve(const TResolveInfo& ri);
+
+    //create alias for host, which can be used for static resolving (when alias is ip address)
+    void AddHostAlias(const TString& host, const TString& alias);
+}
diff --git a/library/cpp/dns/magic.cpp b/library/cpp/dns/magic.cpp
new file mode 100644
index 00000000000..dfcb1bf2028
--- /dev/null
+++ b/library/cpp/dns/magic.cpp
@@ -0,0 +1,29 @@
+#include "magic.h"
+
+#include <util/network/socket.h>
+#include <util/generic/yexception.h>
+
+using namespace NDns;
+
+namespace {
+    namespace NX {
+        struct TError: public IError {
+            inline TError()
+                : E_(std::current_exception())
+            {
+            }
+
+            void Raise() override {
+                std::rethrow_exception(E_);
+            }
+
+            std::exception_ptr E_;
+        };
+    }
+}
+
+IErrorRef NDns::SaveError() {
+    using namespace NX;
+
+    return new NX::TError();
+}
diff --git a/library/cpp/dns/magic.h b/library/cpp/dns/magic.h
new file mode 100644
index 00000000000..d52cde0a6c8
--- /dev/null
+++ b/library/cpp/dns/magic.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include <util/generic/yexception.h>
+#include <util/generic/ptr.h>
+
+namespace NDns {
+    class IError {
+    public:
+        virtual ~IError() = default;
+
+        virtual void Raise() = 0;
+    };
+
+    typedef TAutoPtr<IError> IErrorRef;
+
+    IErrorRef SaveError();
+}
diff --git a/library/cpp/dns/thread.cpp b/library/cpp/dns/thread.cpp
new file mode 100644
index 00000000000..8b27d2d527c
--- /dev/null
+++ b/library/cpp/dns/thread.cpp
@@ -0,0 +1,133 @@
+#include "thread.h"
+
+#include "magic.h"
+
+#include <util/network/socket.h>
+#include <util/thread/factory.h>
+#include <util/thread/lfqueue.h>
+#include <util/system/event.h>
+#include <util/generic/vector.h>
+#include <util/generic/singleton.h>
+
+using namespace NDns;
+
+namespace {
+    class TThreadedResolver: public IThreadFactory::IThreadAble, public TNonCopyable {
+        struct TResolveRequest {
+            inline TResolveRequest(const TString& host, ui16 port)
+                : Host(host)
+                , Port(port)
+            {
+            }
+
+            inline TNetworkAddressPtr Wait() {
+                E.Wait();
+
+                if (!Error) {
+                    if (!Result) {
+                        ythrow TNetworkResolutionError(EAI_AGAIN) << TStringBuf(": resolver down");
+                    }
+
+                    return Result;
+                }
+
+                Error->Raise();
+
+                ythrow TNetworkResolutionError(EAI_FAIL) << TStringBuf(": shit happen");
+            }
+
+            inline void Resolve() noexcept {
+                try {
+                    Result = new TNetworkAddress(Host, Port);
+                } catch (...) {
+                    Error = SaveError();
+                }
+
+                Wake();
+            }
+
+            inline void Wake() noexcept {
+                E.Signal();
+            }
+
+            TString Host;
+            ui16 Port;
+            TManualEvent E;
+            TNetworkAddressPtr Result;
+            IErrorRef Error;
+        };
+
+    public:
+        inline TThreadedResolver()
+            : E_(TSystemEvent::rAuto)
+        {
+            T_.push_back(SystemThreadFactory()->Run(this));
+        }
+
+        inline ~TThreadedResolver() override {
+            Schedule(nullptr);
+
+            for (size_t i = 0; i < T_.size(); ++i) {
+                T_[i]->Join();
+            }
+
+            {
+                TResolveRequest* rr = nullptr;
+
+                while (Q_.Dequeue(&rr)) {
+                    if (rr) {
+                        rr->Wake();
+                    }
+                }
+            }
+        }
+
+        static inline TThreadedResolver* Instance() {
+            return Singleton<TThreadedResolver>();
+        }
+
+        inline TNetworkAddressPtr Resolve(const TString& host, ui16 port) {
+            TResolveRequest rr(host, port);
+
+            Schedule(&rr);
+
+            return rr.Wait();
+        }
+
+    private:
+        inline void Schedule(TResolveRequest* rr) {
+            Q_.Enqueue(rr);
+            E_.Signal();
+        }
+
+        void DoExecute() override {
+            while (true) {
+                TResolveRequest* rr = nullptr;
+
+                while (!Q_.Dequeue(&rr)) {
+                    E_.Wait();
+                }
+
+                if (rr) {
+                    rr->Resolve();
+                } else {
+                    break;
+                }
+            }
+
+            Schedule(nullptr);
+        }
+
+    private:
+        TLockFreeQueue<TResolveRequest*> Q_;
+        TSystemEvent E_;
+        typedef TAutoPtr<IThreadFactory::IThread> IThreadRef;
+        TVector<IThreadRef> T_;
+    };
+}
+
+namespace NDns {
+    TNetworkAddressPtr ThreadedResolve(const TString& host, ui16 port) {
+        return TThreadedResolver::Instance()->Resolve(host, port);
+    }
+}
diff --git a/library/cpp/dns/thread.h b/library/cpp/dns/thread.h
new file mode 100644
index 00000000000..06b41d78ced
--- /dev/null
+++ b/library/cpp/dns/thread.h
@@ -0,0 +1,12 @@
+#pragma once
+
+#include <util/network/socket.h>
+
+#include <util/generic/string.h>
+#include <util/generic/ptr.h>
+
+namespace NDns {
+    typedef TAutoPtr<TNetworkAddress> TNetworkAddressPtr;
+
+    TNetworkAddressPtr ThreadedResolve(const TString& host, ui16 port);
+}
diff --git a/library/cpp/dns/ut/dns_ut.cpp b/library/cpp/dns/ut/dns_ut.cpp
new file mode 100644
index 00000000000..aae05a742c4
--- /dev/null
+++ b/library/cpp/dns/ut/dns_ut.cpp
@@ -0,0 +1,25 @@
+#include <library/cpp/testing/unittest/registar.h>
+#include <library/cpp/dns/cache.h>
+#include <util/network/address.h>
+
+Y_UNIT_TEST_SUITE(TestDNS) {
+    using namespace NDns;
+
+    Y_UNIT_TEST(TestMagic) {
+        UNIT_ASSERT_EXCEPTION(CachedThrResolve(TResolveInfo("?", 80)), yexception);
+    }
+
+    Y_UNIT_TEST(TestAsteriskAlias) {
+        AddHostAlias("*", "localhost");
+        const TResolvedHost* rh = CachedThrResolve(TResolveInfo("yandex.ru", 80));
+        UNIT_ASSERT(rh != nullptr);
+
+        const TNetworkAddress& addr = rh->Addr;
+        for (TNetworkAddress::TIterator ai = addr.Begin(); ai != addr.End(); ai++) {
+            if (ai->ai_family == AF_INET || ai->ai_family == AF_INET6) {
+                NAddr::TAddrInfo info(&*ai);
+                UNIT_ASSERT(IsLoopback(info));
+            }
+        }
+    }
+}
diff --git a/library/cpp/dns/ut/ya.make b/library/cpp/dns/ut/ya.make
new file mode 100644
index 00000000000..7cfd0c4c325
--- /dev/null
+++ b/library/cpp/dns/ut/ya.make
@@ -0,0 +1,16 @@
+UNITTEST()
+
+OWNER(
+    and42
+    pg
+)
+
+PEERDIR(
+    library/cpp/dns
+)
+
+SRCS(
+    dns_ut.cpp
+)
+
+END()
diff --git a/library/cpp/dns/ya.make b/library/cpp/dns/ya.make
new file mode 100644
index 00000000000..0cd83332aa6
--- /dev/null
+++ b/library/cpp/dns/ya.make
@@ -0,0 +1,11 @@
+LIBRARY()
+
+OWNER(and42)
+
+SRCS(
+    cache.cpp
+    thread.cpp
+    magic.cpp
+)
+
+END()