fix windows polling using sockets for the WakeupPipe

src/realm/util/network.cpp

@@ -348,30 +348,172 @@ class WakeupPipe {
 
 #else // defined _WIN32
 
+/* socketpair.c https://github.com/ncm/selectable-socketpair/blob/master/socketpair.c
+Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org>
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+ Redistributions of source code must retain the above copyright notice, this
+ list of conditions and the following disclaimer.
+ Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+ The name of the author must not be used to endorse or promote products
+ derived from this software without specific prior written permission.
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+/* Changes:
+ * 2014-02-12: merge David Woodhouse, Ger Hobbelt improvements
+ * git.infradead.org/users/dwmw2/openconnect.git/commitdiff/bdeefa54
+ * github.com/GerHobbelt/selectable-socketpair
+ * always init the socks[] to -1/INVALID_SOCKET on error, both on Win32/64
+ * and UNIX/other platforms
+ * 2013-07-18: Change to BSD 3-clause license
+ * 2010-03-31:
+ * set addr to 127.0.0.1 because win32 getsockname does not always set it.
+ * 2010-02-25:
+ * set SO_REUSEADDR option to avoid leaking some windows resource.
+ * Windows System Error 10049, "Event ID 4226 TCP/IP has reached
+ * the security limit imposed on the number of concurrent TCP connect
+ * attempts." Bleah.
+ * 2007-04-25:
+ * preserve value of WSAGetLastError() on all error returns.
+ * 2007-04-22: (Thanks to Matthew Gregan <kinetik@flim.org>)
+ * s/EINVAL/WSAEINVAL/ fix trivial compile failure
+ * s/socket/WSASocket/ enable creation of sockets suitable as stdin/stdout
+ * of a child process.
+ * add argument make_overlapped
+ */
+/*
+ * If make_overlapped is nonzero, both sockets created will be usable for
+ * "overlapped" operations via WSASend etc. If make_overlapped is zero,
+ * socks[0] (only) will be usable with regular ReadFile etc., and thus
+ * suitable for use as stdin or stdout of a child process. Note that the
+ * sockets must be closed with closesocket() regardless.
+ */
+int socketpair(SOCKET socks[2], int make_overlapped)
+{
+ union {
+ struct sockaddr_in inaddr;
+ struct sockaddr addr;
+ } a;
+ SOCKET listener;
+ int e;
+ socklen_t addrlen = sizeof(a.inaddr);
+ DWORD flags = (make_overlapped ? WSA_FLAG_OVERLAPPED : 0);
+ int reuse = 1;
+
+ if (socks == 0) {
+ WSASetLastError(WSAEINVAL);
+ return SOCKET_ERROR;
+ }
+ socks[0] = socks[1] = -1;
+
+ listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+ if (listener == -1)
+ return SOCKET_ERROR;
+
+ memset(&a, 0, sizeof(a));
+ a.inaddr.sin_family = AF_INET;
+ a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ a.inaddr.sin_port = 0;
+
+ for (;;) {
+ if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
+ (char*)&reuse, (socklen_t)sizeof(reuse)) == -1)
+ break;
+ if (bind(listener, &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
+ break;
+
+ memset(&a, 0, sizeof(a));
+ if (getsockname(listener, &a.addr, &addrlen) == SOCKET_ERROR)
+ break;
+ // win32 getsockname may only set the port number, p=0.0005.
+ // ( http://msdn.microsoft.com/library/ms738543.aspx ):
+ a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ a.inaddr.sin_family = AF_INET;
+
+ if (listen(listener, 1) == SOCKET_ERROR)
+ break;
+
+ socks[0] = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
+ if (socks[0] == -1)
+ break;
+ if (connect(socks[0], &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
+ break;
+
+ socks[1] = accept(listener, NULL, NULL);
+ if (socks[1] == -1)
+ break;
+
+ closesocket(listener);
+ return 0;
+ }
+
+ e = WSAGetLastError();
+ closesocket(listener);
+ closesocket(socks[0]);
+ closesocket(socks[1]);
+ WSASetLastError(e);
+ socks[0] = socks[1] = -1;
+ return SOCKET_ERROR;
+}
+
 class WakeupPipe {
 public:
- SOCKET wait_fd() const noexcept
- {
- return INVALID_SOCKET;
+ WakeupPipe() {
+ SOCKET socks[2];
+
+ int res = socketpair(socks, true);
+ if (REALM_UNLIKELY(res == SOCKET_ERROR)) {
+ std::error_code ec = make_basic_system_error_code(errno);
+ throw std::system_error(ec);
+ }
+
+ m_write_fd.reset(socks[0]);
+ m_read_fd.reset(socks[1]);
  }
 
- void signal() noexcept
+ SOCKET wait_fd() const noexcept
  {
- m_signal_count++;
+ return m_read_fd;
  }
 
- bool is_signaled() const noexcept
+ void signal() noexcept
  {
- return m_signal_count > 0;
+ LockGuard lock{m_mutex};
+ if (!m_signaled) {
+ char c = 0;
+ ssize_t ret = ::send(m_write_fd, &c, 1, 0);
+ REALM_ASSERT_RELEASE(ret != SOCKET_ERROR);
+ m_signaled = true;
+ }
  }
 
  void acknowledge_signal() noexcept
  {
- m_signal_count--;
+ LockGuard lock{m_mutex};
+ if (m_signaled) {
+ char c;
+ ssize_t ret = ::recv(m_read_fd, &c, 1, 0);
+ REALM_ASSERT_RELEASE(ret != SOCKET_ERROR);
+ m_signaled = false;
+ }
  }
 
 private:
- std::atomic<uint32_t> m_signal_count = 0;
+ CloseGuard m_read_fd;
+ CloseGuard m_write_fd;
+ Mutex m_mutex;
+ bool m_signaled = false; // Protected by `m_mutex`.
 };
 
 #endif // defined _WIN32
@@ -1122,41 +1264,10 @@ bool Service::IoReactor::wait_and_advance(clock::time_point timeout, clock::time
 #endif
 
 #ifdef _WIN32
- max_wait_millis = 1000;
-
- // Windows does not have a single API call to wait for pipes and
- // sockets with a timeout. So we repeatedly poll them individually
- // in a loop until max_wait_millis has elapsed or an event happend.
- //
- // FIXME: Maybe switch to Windows IOCP instead.
-
- // Following variable is the poll time for the sockets in
- // miliseconds. Adjust it to find a balance between CPU usage and
- // response time:
- constexpr INT socket_poll_timeout = 10;
-
- for (size_t t = 0; t < m_pollfd_slots.size(); t++)
- m_pollfd_slots[t].revents = 0;
-
- using namespace std::chrono;
- auto started = steady_clock::now();
- int ret = 0;
-
- do {
- if (m_pollfd_slots.size() > 1) {
- // Poll all network sockets
- ret = WSAPoll(LPWSAPOLLFD(&m_pollfd_slots[1]), ULONG(m_pollfd_slots.size() - 1),
- socket_poll_timeout);
- REALM_ASSERT(ret != SOCKET_ERROR);
- }
-
- if (m_wakeup_pipe.is_signaled()) {
- m_pollfd_slots[0].revents = POLLIN;
- ret++;
- }
-
- } while (ret == 0 &&
- (duration_cast<milliseconds>(steady_clock::now() - started).count() < max_wait_millis));
+ // Poll all network sockets
+ int ret = WSAPoll(LPWSAPOLLFD(fds), ULONG(m_pollfd_slots.size()),
+ max_wait_millis);
+ REALM_ASSERT(ret != SOCKET_ERROR);
 
 #else // !defined _WIN32
  int ret = ::poll(fds, nfds, max_wait_millis);