Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix windows polling using sockets for the WakeupPipe #5599

Closed
wants to merge 2 commits into from
Closed

fix windows polling using sockets for the WakeupPipe #5599

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a996038
fix windows polling using sockets for the WakeupPipe
blagoev Jun 16, 2022
c98eb03
handle PR feedback
blagoev Jun 17, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 113 additions & 35 deletions src/realm/util/network.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,30 +348,128 @@ class WakeupPipe {

#else // defined _WIN32

/* dumb_socketpair: http://cantrip.org/socketpair.c

* Copyright 2007 by Nathan C. Myers <ncm@cantrip.org>; some rights reserved.
* This code is Free Software. It may be copied freely, in original or
* modified form, subject only to the restrictions that (1) the author is
* relieved from all responsibilities for any use for any purpose, and (2)
* this copyright notice must be retained, unchanged, in its entirety. If
* for any reason the author might be held responsible for any consequences
* of copying or use, license is withheld.
*/
/*
* 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;
}

listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (listener == INVALID_SOCKET)
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;

socks[0] = socks[1] = INVALID_SOCKET;
blagoev marked this conversation as resolved.
Show resolved Hide resolved
do {
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;
if (getsockname(listener, &a.addr, &addrlen) == SOCKET_ERROR)
break;
if (listen(listener, 1) == SOCKET_ERROR)
break;
socks[0] = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
if (socks[0] == INVALID_SOCKET)
break;
if (connect(socks[0], &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm somewhat surprised that this doesn't hang since we haven't set this to non-blocking. I noticed that the python impl sets it to nonblocking prior to connecting then sets it back to blocking. We could skip setting it back to blocking since I think we want it in non-blocking mode anyway.

Edit: This is related to the next comment. It looks like connect() only blocks until we enter the listener's backlog, since the connection is "pre-accepted" by the kernel for us prior to calling accept() as long as there is room in the backlog. Since we have a backlog of 1, we would only hang if some other process beat us to connecting. Maybe it is worth setting nonblocking so that we can cleanly error here rather than hanging indefinitely in this very unlikely case?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess this got reviewed a bunch of times from multiple people so I assume it is fine. Like this one https://git.infradead.org/users/dwmw2/openconnect.git/commitdiff/bdeefa54

I am hesitant to try to improve it since we risk introducing bugs which were not there.

break;
socks[1] = accept(listener, NULL, NULL);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is a slight risk that some other process will connect to this port before we do. We could mitigate that by generating a secure random number and requiring the connecter to send that number and verify that it matches. But it seems like there is a ton of code (including the python stdlib) using some form of this pattern and none of it tries to handle this case, so it is probably overkill. Since this is a localhost listener, at least we don't have to worry about remote port scanners.

Edit: I guess if some other process did manage to beat us at connecting then we would probably have the connect call above either hang or return an error since we call listen with a backlog of 1. So maybe we would only get to this line if we were the one connecting.

if (socks[1] == INVALID_SOCKET)
break;

closesocket(listener);
return 0;

} while (0);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This do {} while(0)+break is basically just a disguised goto ERR;. Personally I think it would be better to just use a goto than hiding it in something that looks like a loop, but maybe others are more anti-goto than I am.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree its weird. I don't like using wholes for something other than a loop especially I don't like writing endless loops.
but I tried not to change anything on this code so we follow the already public version that I assume is tested and reviewed by others


e = WSAGetLastError();
closesocket(listener);
closesocket(socks[0]);
closesocket(socks[1]);
blagoev marked this conversation as resolved.
Show resolved Hide resolved
WSASetLastError(e);
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]);
Comment on lines +481 to +482
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Don't we need to set these to non-blocking? Or is that handled somewhere else?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think these should be blocking, cause signal and acknowledge_signal should complete after the signal has been sent/read

}

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, m_write_fd;
blagoev marked this conversation as resolved.
Show resolved Hide resolved
Mutex m_mutex;
bool m_signaled = false; // Protected by `m_mutex`.
};

#endif // defined _WIN32
Expand Down Expand Up @@ -1122,41 +1220,21 @@ 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;
blagoev marked this conversation as resolved.
Show resolved Hide resolved

using namespace std::chrono;
auto started = steady_clock::now();
blagoev marked this conversation as resolved.
Show resolved Hide resolved
int ret = 0;

do {
if (m_pollfd_slots.size() > 1) {

blagoev marked this conversation as resolved.
Show resolved Hide resolved
// Poll all network sockets
ret = WSAPoll(LPWSAPOLLFD(&m_pollfd_slots[1]), ULONG(m_pollfd_slots.size() - 1),
socket_poll_timeout);
ret = WSAPoll(LPWSAPOLLFD(fds), ULONG(m_pollfd_slots.size()),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if we could get rid of most of the ifdefs in this code by having a poll_wrapper function above that dispatches to WSAPoll on windows and ::poll everywhere else. At the very least, could we narrow the #ifdeffing to just be around the function call or is some of the other setup also platform specific?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I targeted of making as minimal changes as possible. I guess we could refactor the code more

max_wait_millis);
/*if (ret == SOCKET_ERROR) {
int lastErr = WSAGetLastError();
}*/
blagoev marked this conversation as resolved.
Show resolved Hide resolved
REALM_ASSERT(ret != SOCKET_ERROR);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't we try to use the common error handling below (removing the #ifndef on line 1244)?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am hesitant to do this much changes. So I left the error handling as it is on this platform.

}

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));

#else // !defined _WIN32
int ret = ::poll(fds, nfds, max_wait_millis);
Expand Down