one shot poller

With WaitReadOneShot:
  - there is no need to do Unwait on connection activation, one less syscall per request;
  - this allows to make several listener threads over one epoll poller.

Turn option on for search daemons (check it turned on by default here https://a.yandex-team.ru/review/4372795/details).

3 files changed, 60 insertions, 40 deletions

diff --git a/library/cpp/http/server/http.cpp b/library/cpp/http/server/http.cpp
index 2dd407dcef..19a16bdcb5 100644
--- a/library/cpp/http/server/http.cpp
+++ b/library/cpp/http/server/http.cpp
@@ -83,12 +83,17 @@ public:
             TGuard<TMutex> g(Mutex_);
 
             Conns_.PushBack(c);
-            Poller_->WaitRead(c->Socket_, (void*)static_cast<const IPollAble*>(c));
+            if (Options.OneShotPoll) {
+                Poller_->WaitReadOneShot(c->Socket_, (void*)static_cast<const IPollAble*>(c));
+            } else {
+                Poller_->WaitRead(c->Socket_, (void*)static_cast<const IPollAble*>(c));
+            }
         }
 
+
         inline void Erase(TClientConnection* c, TInstant now) noexcept {
             TGuard<TMutex> g(Mutex_);
-            EraseUnsafe(c);
+            EraseUnsafe(c, /*removeFromPoller*/!Options.OneShotPoll);
             if (Options.ExpirationTimeout > TDuration::Zero()) {
                 TryRemovingUnsafe(now - Options.ExpirationTimeout);
             }
@@ -118,8 +123,10 @@ public:
             return true;
         }
 
-        void EraseUnsafe(TClientConnection* c) noexcept {
-            Poller_->Unwait(c->Socket_);
+        void EraseUnsafe(TClientConnection* c, bool removeFromPoller = true) noexcept {
+            if (removeFromPoller) {
+                Poller_->Unwait(c->Socket_);
+            }
             c->Unlink();
         }
 
diff --git a/library/cpp/http/server/http_ut.cpp b/library/cpp/http/server/http_ut.cpp
index a74e31c37e..b953a4ef2a 100644
--- a/library/cpp/http/server/http_ut.cpp
+++ b/library/cpp/http/server/http_ut.cpp
@@ -830,31 +830,33 @@ Y_UNIT_TEST_SUITE(THttpServerTest) {
         TShooter shooter(threadCount, port);
 
         TString res = TestData();
-        TEchoServer serverImpl(res);
-        THttpServer server(&serverImpl, THttpServer::TOptions(port).EnableKeepAlive(true));
-        for (size_t i = 0; i < 100; ++i) {
-            UNIT_ASSERT(server.Start());
-            shooter.WaitProgress();
-
-            {
-                auto before = shooter.GetCounters();
+        for (bool oneShot : {true, false}) {
+            TEchoServer serverImpl(res);
+            THttpServer server(&serverImpl, THttpServer::TOptions(port).EnableKeepAlive(true).SetOneShotPoll(oneShot));
+            for (size_t i = 0; i < 100; ++i) {
+                UNIT_ASSERT(server.Start());
                 shooter.WaitProgress();
-                auto after = shooter.GetCounters();
-                for (size_t i = 0; i < before.size(); ++i) {
-                    UNIT_ASSERT(before[i].Success < after[i].Success);
-                    UNIT_ASSERT(before[i].Fail == after[i].Fail);
+
+                {
+                    auto before = shooter.GetCounters();
+                    shooter.WaitProgress();
+                    auto after = shooter.GetCounters();
+                    for (size_t i = 0; i < before.size(); ++i) {
+                        UNIT_ASSERT(before[i].Success < after[i].Success);
+                        UNIT_ASSERT(before[i].Fail == after[i].Fail);
+                    }
                 }
-            }
 
-            server.Stop();
-            shooter.WaitProgress();
-            {
-                auto before = shooter.GetCounters();
+                server.Stop();
                 shooter.WaitProgress();
-                auto after = shooter.GetCounters();
-                for (size_t i = 0; i < before.size(); ++i) {
-                    UNIT_ASSERT(before[i].Success == after[i].Success);
-                    UNIT_ASSERT(before[i].Fail < after[i].Fail);
+                {
+                    auto before = shooter.GetCounters();
+                    shooter.WaitProgress();
+                    auto after = shooter.GetCounters();
+                    for (size_t i = 0; i < before.size(); ++i) {
+                        UNIT_ASSERT(before[i].Success == after[i].Success);
+                        UNIT_ASSERT(before[i].Fail < after[i].Fail);
+                    }
                 }
             }
         }
@@ -881,27 +883,30 @@ Y_UNIT_TEST_SUITE(THttpServerTest) {
         const size_t maxConnections = 5;
 
         TString res = TestData();
-        TMaxConnServer serverImpl(res);
-        THttpServer server(&serverImpl, THttpServer::TOptions(port).EnableKeepAlive(true).SetMaxConnections(maxConnections));
 
-        UNIT_ASSERT(server.Start());
+        for (bool oneShot : {true, false}) {
+            TMaxConnServer serverImpl(res);
+            THttpServer server(&serverImpl, THttpServer::TOptions(port).EnableKeepAlive(true).SetMaxConnections(maxConnections).SetOneShotPoll(oneShot));
 
-        TShooter shooter(maxConnections + 1, port);
+            UNIT_ASSERT(server.Start());
 
-        for (size_t i = 0; i < 100; ++i) {
-            const size_t prev = serverImpl.MaxConns.load();
-            while (serverImpl.MaxConns.load() < prev + 100) {
-                Sleep(TDuration::MilliSeconds(1));
+            TShooter shooter(maxConnections + 1, port);
+
+            for (size_t i = 0; i < 100; ++i) {
+                const size_t prev = serverImpl.MaxConns.load();
+                while (serverImpl.MaxConns.load() < prev + 100) {
+                    Sleep(TDuration::MilliSeconds(1));
+                }
             }
-        }
 
-        shooter.Stop();
-        server.Stop();
+            shooter.Stop();
+            server.Stop();
 
-        for (const auto& c : shooter.GetCounters()) {
-            UNIT_ASSERT(c.Success > 0);
-            UNIT_ASSERT(c.Fail > 0);
-            UNIT_ASSERT(c.Success > c.Fail);
+            for (const auto& c : shooter.GetCounters()) {
+                UNIT_ASSERT(c.Success > 0);
+                UNIT_ASSERT(c.Fail > 0);
+                UNIT_ASSERT(c.Success > c.Fail);
+            }
         }
     }
 }
diff --git a/library/cpp/http/server/options.h b/library/cpp/http/server/options.h
index 5976d58f32..8a83f4f6db 100644
--- a/library/cpp/http/server/options.h
+++ b/library/cpp/http/server/options.h
@@ -139,6 +139,12 @@ public:
         return *this;
     }
 
+    inline THttpServerOptions& SetOneShotPoll(bool v) {
+        OneShotPoll = v;
+
+        return *this;
+    }
+
     struct TAddr {
         TString Addr;
         ui16 Port;
@@ -174,4 +180,6 @@ public:
     TString ListenThreadName = "HttpListen";
     TString RequestsThreadName = "HttpServer";
     TString FailRequestsThreadName = "HttpServer";
+
+    bool OneShotPoll = false;
 };