Support poll in windows (#57)

resetius · Nov 23, 2024 · 61f47c1 · 61f47c1
1 parent 16c8e4b
commit 61f47c1
Show file tree

Hide file tree

Showing 10 changed files with 50 additions and 36 deletions.
diff --git a/coroio/all.hpp b/coroio/all.hpp
@@ -14,10 +14,7 @@
 #include "poller.hpp"
 
 #include "select.hpp"
-
-#ifndef _WIN32
 #include "poll.hpp"
-#endif
 
 #ifdef __linux__
 #include "epoll.hpp"

diff --git a/coroio/poll.cpp b/coroio/poll.cpp
@@ -1,10 +1,8 @@
-#ifndef _WIN32
 #include "poll.hpp"
 
 namespace {
 
 #if defined(__APPLE__) || defined(__FreeBSD__)
-#define POLLRDHUP POLLHUP
 int ppoll(struct pollfd* fds, nfds_t nfds, const struct timespec* ts, const sigset_t* /*sigmask*/) {
     int timeout = 0;
     if (ts) {
@@ -14,13 +12,44 @@ int ppoll(struct pollfd* fds, nfds_t nfds, const struct timespec* ts, const sigs
     // TODO: support sigmask
     return poll(fds, nfds, timeout);
 }
+#elif defined(_WIN32)
+int ppoll(struct pollfd* fds, int nfds, const struct timespec* ts, const void* /*sigmask*/) {
+    int timeout = 0;
+    if (ts) {
+        timeout  = ts->tv_sec;
+        timeout += ts->tv_nsec / 1000000;
+    }
+    // TODO: support sigmask
+    int ret = WSAPoll(fds, nfds, timeout);
+    if (ret < 0) { process_errno(); }
+    return ret;
+}
 #endif
 
 } // namespace
 
 namespace NNet {
 
-void TPoll::Poll() {
+#ifdef _WIN32
+TPoll::TPoll()
+    : DummySocket_({}, *this)
+{
+    int fd = DummySocket_.Fd();
+    MaxFd_ = std::max<int>(MaxFd_, fd);
+    if (static_cast<int>(InEvents_.size()) <= MaxFd_) {
+        InEvents_.resize(MaxFd_+1, std::make_tuple(THandlePair{}, -1));
+    }
+    auto& [_, idx] = InEvents_[fd];
+    idx = Fds_.size();
+    Fds_.emplace_back(pollfd{});
+    pollfd& pev = Fds_[idx];
+    pev.fd = fd;
+    pev.events |= POLLIN;
+}
+#endif
+
+void TPoll::Poll()
+{
     auto ts = GetTimeout();
 
     if (static_cast<int>(InEvents_.size()) <= MaxFd_) {
@@ -45,10 +74,12 @@ void TPoll::Poll() {
                 pev.events |= POLLOUT;
                 ev.Write = ch.Handle;
             }
+#ifdef __linux__
             if (ch.Type & TEvent::RHUP) {
                 pev.events |= POLLRDHUP;
                 ev.Write = ch.Handle;
             }
+#endif
         } else if (idx != -1) {
             if (ch.Type & TEvent::READ) {
                 Fds_[idx].events &= ~POLLIN;
@@ -58,10 +89,12 @@ void TPoll::Poll() {
                 Fds_[idx].events &= ~POLLOUT;
                 ev.Write = {};
             }
+#ifdef __linux__
             if (ch.Type & TEvent::RHUP) {
                 Fds_[idx].events &= ~POLLRDHUP;
                 ev.RHup = {};
             }
+#endif
             if (Fds_[idx].events == 0) {
                 std::swap(Fds_[idx], Fds_.back());
                 std::get<1>(InEvents_[Fds_[idx].fd]) = idx;
@@ -83,20 +116,21 @@ void TPoll::Poll() {
     for (auto& pev : Fds_) {
         auto [ev, _] = InEvents_[pev.fd];
         if (pev.revents & POLLIN) {
-            ReadyEvents_.emplace_back(TEvent{pev.fd, TEvent::READ, ev.Read}); ev.Read = {};
+            ReadyEvents_.emplace_back(TEvent{(int)pev.fd, TEvent::READ, ev.Read}); ev.Read = {};
         }
         if (pev.revents & POLLOUT) {
-            ReadyEvents_.emplace_back(TEvent{pev.fd, TEvent::WRITE, ev.Write}); ev.Write = {};
+            ReadyEvents_.emplace_back(TEvent{(int)pev.fd, TEvent::WRITE, ev.Write}); ev.Write = {};
         }
         if (pev.revents & POLLHUP) {
             if (ev.Read) {
-                ReadyEvents_.emplace_back(TEvent{pev.fd, TEvent::READ, ev.Read});
+                ReadyEvents_.emplace_back(TEvent{(int)pev.fd, TEvent::READ, ev.Read});
             }
             if (ev.Write) {
-                ReadyEvents_.emplace_back(TEvent{pev.fd, TEvent::WRITE, ev.Write});
+                ReadyEvents_.emplace_back(TEvent{(int)pev.fd, TEvent::WRITE, ev.Write});
             }
             pev.revents = pev.revents & ~POLLHUP;
         }
+#ifdef __linux__
         if (pev.revents & POLLRDHUP) {
             if (ev.Read) {
                 ReadyEvents_.emplace_back(TEvent{pev.fd, TEvent::READ, ev.Read});
@@ -108,10 +142,11 @@ void TPoll::Poll() {
                 ReadyEvents_.emplace_back(TEvent{pev.fd, TEvent::RHUP, ev.RHup});
             }
         }
+#endif
     }
 
     ProcessTimers();
 }
 
 } // namespace NNet
-#endif
+
diff --git a/coroio/poll.hpp b/coroio/poll.hpp
@@ -21,11 +21,18 @@ class TPoll: public TPollerBase {
     using TSocket = NNet::TSocket;
     using TFileHandle = NNet::TFileHandle;
 
+#ifdef _WIN32
+    TPoll();
+#endif
+
     void Poll();
 
 private:
     std::vector<std::tuple<THandlePair,int>> InEvents_; // event + index in Fds
     std::vector<pollfd> Fds_;
+#ifdef _WIN32
+    TSocket DummySocket_; // for timeouts
+#endif
 };
 
 } // namespace NNet
diff --git a/examples/bench.cpp b/examples/bench.cpp
@@ -162,11 +162,9 @@ int main(int argc, char** argv) {
     if (!strcmp(method, "select")) {
         run_test<TSelect>(num_pipes, num_writes, num_active);
     }
-#ifndef _WIN32
     else if (!strcmp(method, "poll")) {
         run_test<TPoll>(num_pipes, num_writes, num_active);
     }
-#endif
 #ifdef __linux__
     else if (!strcmp(method, "epoll")) {
         run_test<TEPoll>(num_pipes, num_writes, num_active);

diff --git a/examples/echoclient.cpp b/examples/echoclient.cpp
@@ -1,4 +1,3 @@
-#include "coroio/sockutils.hpp"
 #include <coroio/all.hpp>
 
 #include <signal.h>
@@ -76,11 +75,9 @@ int main(int argc, char** argv) {
     if (method == "select") {
         run<TSelect>(debug, address);
     }
-#ifndef _WIN32
     else if (method == "poll") {
         run<TPoll>(debug, address);
     }
-#endif
 #ifdef __linux__
     else if (method == "epoll") {
         run<TEPoll>(debug, address);

diff --git a/examples/echoserver.cpp b/examples/echoserver.cpp
@@ -3,10 +3,7 @@
 using NNet::TVoidTask;
 using NNet::TAddress;
 using NNet::TSelect;
-
-#ifndef _WIN32
 using NNet::TPoll;
-#endif
 
 #ifdef __linux__
 using NNet::TEPoll;
@@ -90,11 +87,9 @@ int main(int argc, char** argv) {
     if (method == "select") {
         run<TSelect>(debug, address, buffer_size);
     }
-#ifndef _WIN32
     else if (method == "poll") {
         run<TPoll>(debug, address, buffer_size);
     }
-#endif
 #ifdef __linux__
     else if (method == "epoll") {
         run<TEPoll>(debug, address, buffer_size);

diff --git a/examples/resolver.cpp b/examples/resolver.cpp
@@ -62,11 +62,9 @@ int main(int argc, char** argv) {
     if (method == "select") {
         run<TSelect>(type);
     }
-#ifndef _WIN32
     else if (method == "poll") {
         run<TPoll>(type);
     }
-#endif
 #ifdef __linux__
     else if (method == "epoll") {
         run<TEPoll>(type);

diff --git a/examples/sslechoclient.cpp b/examples/sslechoclient.cpp
@@ -75,11 +75,9 @@ int main(int argc, char** argv) {
     if (method == "select") {
         run<TSelect>(debug, address);
     }
-#ifndef _WIN32
     else if (method == "poll") {
         run<TPoll>(debug, address);
     }
-#endif
 #ifdef __linux__
     else if (method == "epoll") {
         run<TEPoll>(debug, address);

diff --git a/examples/sslechoserver.cpp b/examples/sslechoserver.cpp
@@ -3,10 +3,7 @@
 using NNet::TVoidTask;
 using NNet::TAddress;
 using NNet::TSelect;
-
-#ifndef _WIN32
 using NNet::TPoll;
-#endif
 
 #ifdef __linux__
 using NNet::TEPoll;
@@ -104,11 +101,9 @@ int main(int argc, char** argv) {
     if (method == "select") {
         run<TSelect>(debug, address, buffer_size);
     }
-#ifndef _WIN32
     else if (method == "poll") {
         run<TPoll>(debug, address, buffer_size);
     }
-#endif
 #ifdef __linux__
     else if (method == "epoll") {
         run<TEPoll>(debug, address, buffer_size);

diff --git a/tests/tests.cpp b/tests/tests.cpp
@@ -1202,15 +1202,9 @@ int main() {
         my_unit_poller(test_futures_any_result),
         my_unit_poller(test_futures_any_same_wakeup),
         my_unit_poller(test_futures_all),
-#ifdef _WIN32
-        my_unit_test(test_read_write_full_ssl, TSelect),
-        my_unit_test(test_resolver, TSelect),
-        my_unit_test(test_resolve_bad_name, TSelect),
-#else
         my_unit_test2(test_read_write_full_ssl, TSelect, TPoll),
         my_unit_test2(test_resolver, TSelect, TPoll),
         my_unit_test2(test_resolve_bad_name, TSelect, TPoll),
-#endif
 #ifdef __linux__
         my_unit_test2(test_remote_disconnect, TPoll, TEPoll),
         cmocka_unit_test(test_uring_create),