Skip to content

Commit

Permalink
Revert "Revert 'multi: implement wait using winsock events'"
Browse files Browse the repository at this point in the history 
This reverts commit 2260e0e,
also restoring previous follow up changes which were reverted.

Authored-by: rcombs on github
Authored-by: Marc Hörsken

Restores curl#5634
Reverts curl#6281
Part of curl#6245
  • Loading branch information
@mback2k
mback2k committed Apr 20, 2021
1 parent 6e3f2fe commit 769eb5d
Show file tree
Hide file tree
Showing 2 changed files with 153 additions and 3 deletions.
152 changes: 149 additions & 3 deletions lib/multi.c
View file
Expand Up @@ -381,6 +381,11 @@ struct Curl_multi *Curl_multi_handle(int hashsize, /* socket hash */
multi->max_concurrent_streams = 100;
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
if(Curl_socketpair(AF_UNIX, SOCK_STREAM, 0, multi->wakeup_pair) < 0) {
multi->wakeup_pair[0] = CURL_SOCKET_BAD;
Expand All @@ -393,6 +398,7 @@ struct Curl_multi *Curl_multi_handle(int hashsize, /* socket hash */
multi->wakeup_pair[0] = CURL_SOCKET_BAD;
multi->wakeup_pair[1] = CURL_SOCKET_BAD;
}
#endif
#endif

return multi;
Expand Down Expand Up @@ -1080,9 +1086,15 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
unsigned int curlfds;
long timeout_internal;
int retcode = 0;
#ifndef USE_WINSOCK
struct pollfd a_few_on_stack[NUM_POLLS_ON_STACK];
struct pollfd *ufds = &a_few_on_stack[0];
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))
return CURLM_BAD_HANDLE;
Expand Down Expand Up @@ -1128,11 +1140,16 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
nfds += extra_nfds; /* add the externally provided ones */

#ifdef ENABLE_WAKEUP
#ifdef USE_WINSOCK
if(use_wakeup) {
#else
if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) {
#endif
++nfds;
}
#endif

#ifndef USE_WINSOCK
if(nfds > NUM_POLLS_ON_STACK) {
/* '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
Expand All @@ -1143,7 +1160,9 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
return CURLM_OUT_OF_MEMORY;
ufds_malloc = TRUE;
}

nfds = 0;
#endif

/* only do the second loop if we found descriptors in the first stage run
above */
Expand All @@ -1156,22 +1175,40 @@ static CURLMcode multi_wait(struct Curl_multi *multi,

for(i = 0; i < MAX_SOCKSPEREASYHANDLE; i++) {
curl_socket_t s = CURL_SOCKET_BAD;

#ifdef USE_WINSOCK
long mask = 0;
#endif
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].events = POLLIN;
++nfds;
#endif
s = sockbunch[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].events = POLLOUT;
++nfds;
#endif
s = sockbunch[i];
}
if(s == CURL_SOCKET_BAD) {
break;
}
#ifdef USE_WINSOCK
if(WSAEventSelect(s, multi->wsa_event, mask) != 0)
return CURLM_INTERNAL_ERROR;
#endif
}

data = data->next; /* check next handle */
Expand All @@ -1180,6 +1217,37 @@ static CURLMcode multi_wait(struct Curl_multi *multi,

/* Add external file descriptions from poll-like struct curl_waitfd */
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].events = 0;
if(extra_fds[i].events & CURL_WAIT_POLLIN)
Expand All @@ -1189,26 +1257,57 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
if(extra_fds[i].events & CURL_WAIT_POLLOUT)
ufds[nfds].events |= POLLOUT;
++nfds;
#endif
}

#ifdef ENABLE_WAKEUP
#ifndef USE_WINSOCK
if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) {
ufds[nfds].fd = multi->wakeup_pair[0];
ufds[nfds].events = POLLIN;
++nfds;
}
#endif
#endif

if(nfds) {
/* wait... */
#ifdef USE_WINSOCK
WSAWaitForMultipleEvents(1, &multi->wsa_event, FALSE, timeout_ms, FALSE);
#else
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) {
retcode = pollrc;
#endif
/* copy revents results from the poll to the curl_multi_wait poll
struct, the bit values of the actual underlying poll() implementation
may not be the same as the ones in the public libcurl API! */
for(i = 0; i < extra_nfds; i++) {
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;

if(r & POLLIN)
Expand All @@ -1217,9 +1316,46 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
mask |= CURL_WAIT_POLLOUT;
if(r & POLLPRI)
mask |= CURL_WAIT_POLLPRI;
#endif
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
if(use_wakeup && multi->wakeup_pair[0] != CURL_SOCKET_BAD) {
if(ufds[curlfds + extra_nfds].revents & POLLIN) {
Expand All @@ -1232,23 +1368,24 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
when there is no more data, breaking the loop. */
nread = sread(multi->wakeup_pair[0], buf, sizeof(buf));
if(nread <= 0) {
#ifndef USE_WINSOCK
if(nread < 0 && EINTR == SOCKERRNO)
continue;
#endif
break;
}
}
/* do not count the wakeup socket into the returned value */
retcode--;
}
}
#endif
#endif
}
}

#ifndef USE_WINSOCK
if(ufds_malloc)
free(ufds);
#endif
if(ret)
*ret = retcode;
if(!extrawait || nfds)
Expand Down Expand Up @@ -1303,6 +1440,10 @@ CURLMcode curl_multi_wakeup(struct Curl_multi *multi)
return CURLM_BAD_HANDLE;

#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,
making it safe to access from another thread after the init part
and before cleanup */
Expand Down Expand Up @@ -1335,6 +1476,7 @@ CURLMcode curl_multi_wakeup(struct Curl_multi *multi)
return CURLM_OK;
}
}
#endif
#endif
return CURLM_WAKEUP_FAILURE;
}
Expand Down Expand Up @@ -2478,9 +2620,13 @@ CURLMcode curl_multi_cleanup(struct Curl_multi *multi)
Curl_hash_destroy(&multi->hostcache);
Curl_psl_destroy(&multi->psl);

#ifdef USE_WINSOCK
WSACloseEvent(multi->wsa_event);
#else
#ifdef ENABLE_WAKEUP
sclose(multi->wakeup_pair[0]);
sclose(multi->wakeup_pair[1]);
#endif
#endif
free(multi);

Expand Down
4 changes: 4 additions & 0 deletions lib/multihandle.h
View file
Expand Up @@ -140,9 +140,13 @@ struct Curl_multi {
previous callback */
unsigned int max_concurrent_streams;

#ifdef USE_WINSOCK
WSAEVENT wsa_event; /* winsock event used for waits */
#else
#ifdef ENABLE_WAKEUP
curl_socket_t wakeup_pair[2]; /* socketpair() used for wakeup
0 is used for read, 1 is used for write */
#endif
#endif
/* multiplexing wanted */
bool multiplexing;
Expand Down

0 comments on commit 769eb5d

Please sign in to comment.