Revert "Revert 'multi: implement wait using winsock events'"

This reverts commit 2260e0ebe6,
also restoring previous follow up changes which were reverted.

Authored-by: rcombs on github
Authored-by: Marc Hörsken
Reviewed-by: Jay Satiro
Reviewed-by: Marcel Raad

Restores #5634
Reverts #6281
Part of #6245
This commit is contained in:
Marc Hoersken 2020-12-09 07:25:15 +01:00
parent 3057c6cd73
commit bcca174cfa
No known key found for this signature in database
GPG Key ID: 61E03CBED7BC859E
2 changed files with 153 additions and 3 deletions

View File

@ -381,6 +381,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;
@ -393,6 +398,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;
@ -1080,9 +1086,15 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
unsigned int curlfds; unsigned int curlfds;
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; bool ufds_malloc = FALSE;
#else
struct pollfd pre_poll;
WSANETWORKEVENTS wsa_events;
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;
@ -1128,11 +1140,16 @@ static CURLMcode 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
@ -1143,7 +1160,9 @@ static CURLMcode 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 */
@ -1156,22 +1175,40 @@ static CURLMcode 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
if(timeout_ms && SOCKET_READABLE(sockbunch[i], 0) > 0)
timeout_ms = 0;
mask |= FD_READ|FD_ACCEPT|FD_CLOSE;
#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
if(timeout_ms && SOCKET_WRITABLE(sockbunch[i], 0) > 0)
timeout_ms = 0;
mask |= FD_WRITE|FD_CONNECT|FD_CLOSE;
#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 */
@ -1180,6 +1217,37 @@ static CURLMcode 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 mask = 0;
extra_fds[i].revents = 0;
pre_poll.fd = extra_fds[i].fd;
pre_poll.events = 0;
pre_poll.revents = 0;
if(extra_fds[i].events & CURL_WAIT_POLLIN) {
mask |= FD_READ|FD_ACCEPT|FD_CLOSE;
pre_poll.events |= POLLIN;
}
if(extra_fds[i].events & CURL_WAIT_POLLPRI) {
mask |= FD_OOB;
pre_poll.events |= POLLPRI;
}
if(extra_fds[i].events & CURL_WAIT_POLLOUT) {
mask |= FD_WRITE|FD_CONNECT|FD_CLOSE;
pre_poll.events |= POLLOUT;
}
if(Curl_poll(&pre_poll, 1, 0) > 0) {
if(pre_poll.revents & POLLIN)
extra_fds[i].revents |= CURL_WAIT_POLLIN;
if(pre_poll.revents & POLLPRI)
extra_fds[i].revents |= CURL_WAIT_POLLPRI;
if(pre_poll.revents & POLLOUT)
extra_fds[i].revents |= CURL_WAIT_POLLOUT;
if(extra_fds[i].revents)
timeout_ms = 0;
}
if(WSAEventSelect(extra_fds[i].fd, multi->wsa_event, mask) != 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)
@ -1189,26 +1257,57 @@ static CURLMcode 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) {
/* wait... */ /* wait... */
#ifdef USE_WINSOCK
WSAWaitForMultipleEvents(1, &multi->wsa_event, FALSE, timeout_ms, FALSE);
#else
int pollrc = Curl_poll(ufds, nfds, timeout_ms); int pollrc = Curl_poll(ufds, nfds, timeout_ms);
#endif
#ifdef USE_WINSOCK
/* With Winsock, we have to run this unconditionally to call
WSAEventSelect(fd, event, 0) on all the sockets */
{
retcode = 0;
#else
if(pollrc > 0) { if(pollrc > 0) {
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
wsa_events.lNetworkEvents = 0;
mask = extra_fds[i].revents;
if(WSAEnumNetworkEvents(extra_fds[i].fd, multi->wsa_event,
&wsa_events) == 0) {
if(wsa_events.lNetworkEvents & (FD_READ|FD_ACCEPT|FD_CLOSE))
mask |= CURL_WAIT_POLLIN;
if(wsa_events.lNetworkEvents & (FD_WRITE|FD_CONNECT|FD_CLOSE))
mask |= CURL_WAIT_POLLOUT;
if(wsa_events.lNetworkEvents & FD_OOB)
mask |= CURL_WAIT_POLLPRI;
if(ret && wsa_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)
@ -1217,9 +1316,46 @@ static CURLMcode 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))) {
wsa_events.lNetworkEvents = 0;
if(WSAEnumNetworkEvents(sockbunch[i], multi->wsa_event,
&wsa_events) == 0) {
if(ret && wsa_events.lNetworkEvents != 0)
retcode++;
}
if(ret && !timeout_ms && wsa_events.lNetworkEvents == 0) {
if((bitmap & GETSOCK_READSOCK(i)) &&
SOCKET_READABLE(sockbunch[i], 0) > 0)
retcode++;
else if((bitmap & GETSOCK_WRITESOCK(i)) &&
SOCKET_WRITABLE(sockbunch[i], 0) > 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) {
@ -1232,10 +1368,8 @@ static CURLMcode 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;
} }
} }
@ -1243,12 +1377,15 @@ static CURLMcode 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)
@ -1303,6 +1440,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 */
@ -1335,6 +1476,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;
} }
@ -2478,9 +2620,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

@ -140,9 +140,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;