1
0
mirror of https://github.com/moparisthebest/curl synced 2024-11-17 06:55:02 -05:00

multi: implement wait using winsock events

This avoids using a pair of TCP ports to provide wakeup functionality
for every multi instance on Windows, where socketpair() is emulated
using a TCP socket on loopback which could in turn lead to socket
resource exhaustion.

Reviewed-by: Gergely Nagy
Reviewed-by: Marc Hörsken

Closes #5397
This commit is contained in:
rcombs 2020-05-13 18:49:57 -05:00 committed by Marc Hoersken
parent 477a4e31d7
commit 8bc25c590e
No known key found for this signature in database
GPG Key ID: 61E03CBED7BC859E
2 changed files with 122 additions and 4 deletions

View File

@ -374,6 +374,11 @@ struct Curl_multi *Curl_multi_handle(int hashsize, /* socket hash */
multi->max_concurrent_streams = 100; multi->max_concurrent_streams = 100;
multi->ipv6_works = Curl_ipv6works(NULL); multi->ipv6_works = Curl_ipv6works(NULL);
#ifdef USE_WINSOCK
multi->wsa_event = WSACreateEvent();
if(multi->wsa_event == WSA_INVALID_EVENT)
goto error;
#else
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
if(Curl_socketpair(AF_UNIX, SOCK_STREAM, 0, multi->wakeup_pair) < 0) { if(Curl_socketpair(AF_UNIX, SOCK_STREAM, 0, multi->wakeup_pair) < 0) {
multi->wakeup_pair[0] = CURL_SOCKET_BAD; multi->wakeup_pair[0] = CURL_SOCKET_BAD;
@ -386,6 +391,7 @@ struct Curl_multi *Curl_multi_handle(int hashsize, /* socket hash */
multi->wakeup_pair[0] = CURL_SOCKET_BAD; multi->wakeup_pair[0] = CURL_SOCKET_BAD;
multi->wakeup_pair[1] = CURL_SOCKET_BAD; multi->wakeup_pair[1] = CURL_SOCKET_BAD;
} }
#endif
#endif #endif
return multi; return multi;
@ -1067,11 +1073,15 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
unsigned int i; unsigned int i;
unsigned int nfds = 0; unsigned int nfds = 0;
unsigned int curlfds; unsigned int curlfds;
bool ufds_malloc = FALSE;
long timeout_internal; long timeout_internal;
int retcode = 0; int retcode = 0;
#ifndef USE_WINSOCK
struct pollfd a_few_on_stack[NUM_POLLS_ON_STACK]; struct pollfd a_few_on_stack[NUM_POLLS_ON_STACK];
struct pollfd *ufds = &a_few_on_stack[0]; struct pollfd *ufds = &a_few_on_stack[0];
bool ufds_malloc = FALSE;
#else
DEBUGASSERT(multi->wsa_event != WSA_INVALID_EVENT);
#endif
if(!GOOD_MULTI_HANDLE(multi)) if(!GOOD_MULTI_HANDLE(multi))
return CURLM_BAD_HANDLE; return CURLM_BAD_HANDLE;
@ -1117,11 +1127,16 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
nfds += extra_nfds; /* add the externally provided ones */ nfds += extra_nfds; /* add the externally provided ones */
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
#ifdef USE_WINSOCK
if(use_wakeup) {
#else
if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) { if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) {
#endif
++nfds; ++nfds;
} }
#endif #endif
#ifndef USE_WINSOCK
if(nfds > NUM_POLLS_ON_STACK) { if(nfds > NUM_POLLS_ON_STACK) {
/* 'nfds' is a 32 bit value and 'struct pollfd' is typically 8 bytes /* 'nfds' is a 32 bit value and 'struct pollfd' is typically 8 bytes
big, so at 2^29 sockets this value might wrap. When a process gets big, so at 2^29 sockets this value might wrap. When a process gets
@ -1132,7 +1147,9 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
return CURLM_OUT_OF_MEMORY; return CURLM_OUT_OF_MEMORY;
ufds_malloc = TRUE; ufds_malloc = TRUE;
} }
nfds = 0; nfds = 0;
#endif
/* only do the second loop if we found descriptors in the first stage run /* only do the second loop if we found descriptors in the first stage run
above */ above */
@ -1145,22 +1162,36 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
for(i = 0; i< MAX_SOCKSPEREASYHANDLE; i++) { for(i = 0; i< MAX_SOCKSPEREASYHANDLE; i++) {
curl_socket_t s = CURL_SOCKET_BAD; curl_socket_t s = CURL_SOCKET_BAD;
#ifdef USE_WINSOCK
long mask = 0;
#endif
if(bitmap & GETSOCK_READSOCK(i)) { if(bitmap & GETSOCK_READSOCK(i)) {
#ifdef USE_WINSOCK
mask |= FD_READ;
#else
ufds[nfds].fd = sockbunch[i]; ufds[nfds].fd = sockbunch[i];
ufds[nfds].events = POLLIN; ufds[nfds].events = POLLIN;
++nfds; ++nfds;
#endif
s = sockbunch[i]; s = sockbunch[i];
} }
if(bitmap & GETSOCK_WRITESOCK(i)) { if(bitmap & GETSOCK_WRITESOCK(i)) {
#ifdef USE_WINSOCK
mask |= FD_WRITE;
#else
ufds[nfds].fd = sockbunch[i]; ufds[nfds].fd = sockbunch[i];
ufds[nfds].events = POLLOUT; ufds[nfds].events = POLLOUT;
++nfds; ++nfds;
#endif
s = sockbunch[i]; s = sockbunch[i];
} }
if(s == CURL_SOCKET_BAD) { if(s == CURL_SOCKET_BAD) {
break; break;
} }
#ifdef USE_WINSOCK
if(WSAEventSelect(s, multi->wsa_event, mask) != 0)
return CURLM_INTERNAL_ERROR;
#endif
} }
data = data->next; /* check next handle */ data = data->next; /* check next handle */
@ -1169,6 +1200,17 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
/* Add external file descriptions from poll-like struct curl_waitfd */ /* Add external file descriptions from poll-like struct curl_waitfd */
for(i = 0; i < extra_nfds; i++) { for(i = 0; i < extra_nfds; i++) {
#ifdef USE_WINSOCK
long events = 0;
if(extra_fds[i].events & CURL_WAIT_POLLIN)
events |= FD_READ;
if(extra_fds[i].events & CURL_WAIT_POLLPRI)
events |= FD_OOB;
if(extra_fds[i].events & CURL_WAIT_POLLOUT)
events |= FD_WRITE;
if(WSAEventSelect(extra_fds[i].fd, multi->wsa_event, events) != 0)
return CURLM_INTERNAL_ERROR;
#else
ufds[nfds].fd = extra_fds[i].fd; ufds[nfds].fd = extra_fds[i].fd;
ufds[nfds].events = 0; ufds[nfds].events = 0;
if(extra_fds[i].events & CURL_WAIT_POLLIN) if(extra_fds[i].events & CURL_WAIT_POLLIN)
@ -1178,28 +1220,61 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
if(extra_fds[i].events & CURL_WAIT_POLLOUT) if(extra_fds[i].events & CURL_WAIT_POLLOUT)
ufds[nfds].events |= POLLOUT; ufds[nfds].events |= POLLOUT;
++nfds; ++nfds;
#endif
} }
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
#ifndef USE_WINSOCK
if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) { if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) {
ufds[nfds].fd = multi->wakeup_pair[0]; ufds[nfds].fd = multi->wakeup_pair[0];
ufds[nfds].events = POLLIN; ufds[nfds].events = POLLIN;
++nfds; ++nfds;
} }
#endif
#endif #endif
if(nfds) { if(nfds) {
int pollrc; int pollrc;
/* wait... */ /* wait... */
#ifdef USE_WINSOCK
DWORD waitrc = WSAWaitForMultipleEvents(1, &multi->wsa_event, FALSE,
timeout_ms, FALSE);
/* WSA_WAIT_EVENT_0 is 0, so waitrc >= WSA_WAIT_EVENT_0 warns */
if(waitrc == WSA_WAIT_EVENT_0)
pollrc = 1;
else
pollrc = -1;
#else
pollrc = Curl_poll(ufds, nfds, timeout_ms); pollrc = Curl_poll(ufds, nfds, timeout_ms);
#endif
if(pollrc > 0) { if(pollrc > 0) {
#ifdef USE_WINSOCK
retcode = 0;
#else
retcode = pollrc; retcode = pollrc;
#endif
/* copy revents results from the poll to the curl_multi_wait poll /* copy revents results from the poll to the curl_multi_wait poll
struct, the bit values of the actual underlying poll() implementation struct, the bit values of the actual underlying poll() implementation
may not be the same as the ones in the public libcurl API! */ may not be the same as the ones in the public libcurl API! */
for(i = 0; i < extra_nfds; i++) { for(i = 0; i < extra_nfds; i++) {
unsigned short mask = 0; unsigned short mask = 0;
#ifdef USE_WINSOCK
WSANETWORKEVENTS events = {0};
if(WSAEnumNetworkEvents(extra_fds[i].fd, multi->wsa_event,
&events) == 0) {
if(events.lNetworkEvents & FD_READ)
mask |= CURL_WAIT_POLLIN;
if(events.lNetworkEvents & FD_WRITE)
mask |= CURL_WAIT_POLLOUT;
if(events.lNetworkEvents & FD_OOB)
mask |= CURL_WAIT_POLLPRI;
if(events.lNetworkEvents != 0)
retcode++;
}
WSAEventSelect(extra_fds[i].fd, multi->wsa_event, 0);
#else
unsigned r = ufds[curlfds + i].revents; unsigned r = ufds[curlfds + i].revents;
if(r & POLLIN) if(r & POLLIN)
@ -1208,10 +1283,39 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
mask |= CURL_WAIT_POLLOUT; mask |= CURL_WAIT_POLLOUT;
if(r & POLLPRI) if(r & POLLPRI)
mask |= CURL_WAIT_POLLPRI; mask |= CURL_WAIT_POLLPRI;
#endif
extra_fds[i].revents = mask; extra_fds[i].revents = mask;
} }
#ifdef USE_WINSOCK
/* Count up all our own sockets that had activity,
and remove them from the event. */
if(curlfds) {
data = multi->easyp;
while(data) {
bitmap = multi_getsock(data, sockbunch);
for(i = 0; i < MAX_SOCKSPEREASYHANDLE; i++) {
if(bitmap & (GETSOCK_READSOCK(i) | GETSOCK_WRITESOCK(i))) {
WSANETWORKEVENTS events = {0};
if(WSAEnumNetworkEvents(sockbunch[i], multi->wsa_event,
&events) == 0) {
if(events.lNetworkEvents != 0)
retcode++;
}
WSAEventSelect(sockbunch[i], multi->wsa_event, 0);
}
else
break;
}
data = data->next;
}
}
WSAResetEvent(multi->wsa_event);
#else
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) { if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) {
if(ufds[curlfds + extra_nfds].revents & POLLIN) { if(ufds[curlfds + extra_nfds].revents & POLLIN) {
@ -1224,10 +1328,8 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
when there is no more data, breaking the loop. */ when there is no more data, breaking the loop. */
nread = sread(multi->wakeup_pair[0], buf, sizeof(buf)); nread = sread(multi->wakeup_pair[0], buf, sizeof(buf));
if(nread <= 0) { if(nread <= 0) {
#ifndef USE_WINSOCK
if(nread < 0 && EINTR == SOCKERRNO) if(nread < 0 && EINTR == SOCKERRNO)
continue; continue;
#endif
break; break;
} }
} }
@ -1235,12 +1337,15 @@ static CURLMcode Curl_multi_wait(struct Curl_multi *multi,
retcode--; retcode--;
} }
} }
#endif
#endif #endif
} }
} }
#ifndef USE_WINSOCK
if(ufds_malloc) if(ufds_malloc)
free(ufds); free(ufds);
#endif
if(ret) if(ret)
*ret = retcode; *ret = retcode;
if(!extrawait || nfds) if(!extrawait || nfds)
@ -1295,6 +1400,10 @@ CURLMcode curl_multi_wakeup(struct Curl_multi *multi)
return CURLM_BAD_HANDLE; return CURLM_BAD_HANDLE;
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
#ifdef USE_WINSOCK
if(WSASetEvent(multi->wsa_event))
return CURLM_OK;
#else
/* the wakeup_pair variable is only written during init and cleanup, /* the wakeup_pair variable is only written during init and cleanup,
making it safe to access from another thread after the init part making it safe to access from another thread after the init part
and before cleanup */ and before cleanup */
@ -1327,6 +1436,7 @@ CURLMcode curl_multi_wakeup(struct Curl_multi *multi)
return CURLM_OK; return CURLM_OK;
} }
} }
#endif
#endif #endif
return CURLM_WAKEUP_FAILURE; return CURLM_WAKEUP_FAILURE;
} }
@ -2485,9 +2595,13 @@ CURLMcode curl_multi_cleanup(struct Curl_multi *multi)
Curl_hash_destroy(&multi->hostcache); Curl_hash_destroy(&multi->hostcache);
Curl_psl_destroy(&multi->psl); Curl_psl_destroy(&multi->psl);
#ifdef USE_WINSOCK
WSACloseEvent(multi->wsa_event);
#else
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
sclose(multi->wakeup_pair[0]); sclose(multi->wakeup_pair[0]);
sclose(multi->wakeup_pair[1]); sclose(multi->wakeup_pair[1]);
#endif
#endif #endif
free(multi); free(multi);

View File

@ -138,9 +138,13 @@ struct Curl_multi {
previous callback */ previous callback */
unsigned int max_concurrent_streams; unsigned int max_concurrent_streams;
#ifdef USE_WINSOCK
WSAEVENT wsa_event; /* winsock event used for waits */
#else
#ifdef ENABLE_WAKEUP #ifdef ENABLE_WAKEUP
curl_socket_t wakeup_pair[2]; /* socketpair() used for wakeup curl_socket_t wakeup_pair[2]; /* socketpair() used for wakeup
0 is used for read, 1 is used for write */ 0 is used for read, 1 is used for write */
#endif
#endif #endif
/* multiplexing wanted */ /* multiplexing wanted */
bool multiplexing; bool multiplexing;