romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-08-26 12:43:34 +03:00
Code Issues Projects Releases Packages Wiki Activity

multi: event based rework

Browse source 
Rework the event based handling of transfers and connections to
be "localized" into a single source file with clearer dependencies.

- add multi_ev.c and multi_ev.h
- add docs/internal/MULTI-EV.md to explain the overall workings
- only do event handling book keeping when the socket callback
  is set
- add handling for "connection only" event tracking, when internal
  easy handles are used that are not really tied to a connection.
  Used in connection pool.
- remove transfer member "last_poll" and connections "shutdown_poll"
  and keep all that internal to multi_ev.c
- add CURL_TRC_M() for tracing of "multi" related things, including
  event handling and connection pool operations. Add new trace
  feature "multi" for trace config.
  multi traces will show exactly what is going on in regard to
  event handling.
- multi: trace transfers "mstate" in every CURL_TRC_M() call
- make internal trace buffer 2048 bytes and end the silliness
  with +n here -m there. Adjust test 1652 expectations of resulting
  length and input edge cases.
- add trace feature "lib-ids" to perfix libcurl traces with transfer
  and connection ids. Useful for debugging libcurl applications.

Closes #16308
This commit is contained in:
Stefan Eissing 2025-02-22 12:46:42 +01:00 • committed by Daniel Stenberg
parent 886f5dea80
commit cfc657a48d
No known key found for this signature in database
GPG key ID: 5CC908FDB71E12C2
24 changed files with 1305 additions and 764 deletions
Download patch file Download diff file Expand all files Collapse all files

668
lib/multi.c
View file

@ -36,6 +36,7 @@
#include "share.h"
#include "psl.h"
#include "multiif.h"
#include "multi_ev.h"
#include "sendf.h"
#include "timeval.h"
#include "http.h"
@ -94,8 +95,6 @@
static void move_pending_to_connect(struct Curl_multi *multi,
struct Curl_easy *data);
static CURLMcode singlesocket(struct Curl_multi *multi,
struct Curl_easy *data);
static CURLMcode add_next_timeout(struct curltime now,
struct Curl_multi *multi,
struct Curl_easy *d);
@ -104,31 +103,6 @@ static CURLMcode multi_timeout(struct Curl_multi *multi,
long *timeout_ms);
static void process_pending_handles(struct Curl_multi *multi);
static void multi_xfer_bufs_free(struct Curl_multi *multi);
static void expire_ex(struct Curl_easy *data, const struct curltime *nowp,
timediff_t milli, expire_id id);
#if defined( DEBUGBUILD) && !defined(CURL_DISABLE_VERBOSE_STRINGS)
static const char * const multi_statename[]={
"INIT",
"PENDING",
"SETUP",
"CONNECT",
"RESOLVING",
"CONNECTING",
"TUNNELING",
"PROTOCONNECT",
"PROTOCONNECTING",
"DO",
"DOING",
"DOING_MORE",
"DID",
"PERFORMING",
"RATELIMITING",
"DONE",
"COMPLETED",
"MSGSENT",
};
#endif
/* function pointer called once when switching TO a state */
typedef void (*init_multistate_func)(struct Curl_easy *data);
@ -179,26 +153,18 @@ static void mstate(struct Curl_easy *data, CURLMstate state
NULL /* MSGSENT */
};
#if defined(DEBUGBUILD) && defined(CURL_DISABLE_VERBOSE_STRINGS)
(void) lineno;
#endif
if(oldstate == state)
/* do not bother when the new state is the same as the old state */
return;
data->mstate = state;
#if defined(DEBUGBUILD) && !defined(CURL_DISABLE_VERBOSE_STRINGS)
if(data->mstate >= MSTATE_PENDING &&
data->mstate < MSTATE_COMPLETED) {
infof(data,
"STATE: %s => %s handle %p; line %d",
multi_statename[oldstate], multi_statename[data->mstate],
(void *)data, lineno);
}
#ifdef DEBUGBUILD
CURL_TRC_M(data, "-> [%s] (line %d)", CURL_MSTATE_NAME(state), lineno);
#else
CURL_TRC_M(data, "-> [%s]", CURL_MSTATE_NAME(state));
#endif
data->mstate = state;
if(state == MSTATE_COMPLETED) {
/* changing to COMPLETED means there is one less easy handle 'alive' */
DEBUGASSERT(data->multi->num_alive > 0);
@ -220,163 +186,6 @@ static void mstate(struct Curl_easy *data, CURLMstate state
#define multistate(x,y) mstate(x,y, __LINE__)
#endif
/*
* We add one of these structs to the sockhash for each socket
*/
struct Curl_sh_entry {
struct Curl_hash transfers; /* hash of transfers using this socket */
unsigned int action; /* what combined action READ/WRITE this socket waits
for */
unsigned int users; /* number of transfers using this */
void *socketp; /* settable by users with curl_multi_assign() */
unsigned int readers; /* this many transfers want to read */
unsigned int writers; /* this many transfers want to write */
};
/* look up a given socket in the socket hash, skip invalid sockets */
static struct Curl_sh_entry *sh_getentry(struct Curl_hash *sh,
curl_socket_t s)
{
if(s != CURL_SOCKET_BAD) {
/* only look for proper sockets */
return Curl_hash_pick(sh, (char *)&s, sizeof(curl_socket_t));
}
return NULL;
}
#define TRHASH_SIZE 13
/* the given key here is a struct Curl_easy pointer */
static size_t trhash(void *key, size_t key_length, size_t slots_num)
{
unsigned char bytes = ((unsigned char *)key)[key_length - 1] ^
((unsigned char *)key)[0];
return (bytes % slots_num);
}
static size_t trhash_compare(void *k1, size_t k1_len, void *k2, size_t k2_len)
{
(void)k2_len;
return !memcmp(k1, k2, k1_len);
}
static void trhash_dtor(void *nada)
{
(void)nada;
}
/*
* The sockhash has its own separate subhash in each entry that need to be
* safely destroyed first.
*/
static void sockhash_destroy(struct Curl_hash *h)
{
struct Curl_hash_iterator iter;
struct Curl_hash_element *he;
DEBUGASSERT(h);
Curl_hash_start_iterate(h, &iter);
he = Curl_hash_next_element(&iter);
while(he) {
struct Curl_sh_entry *sh = (struct Curl_sh_entry *)he->ptr;
Curl_hash_destroy(&sh->transfers);
he = Curl_hash_next_element(&iter);
}
Curl_hash_destroy(h);
}
/* make sure this socket is present in the hash for this handle */
static struct Curl_sh_entry *sh_addentry(struct Curl_hash *sh,
curl_socket_t s)
{
struct Curl_sh_entry *there = sh_getentry(sh, s);
struct Curl_sh_entry *check;
if(there) {
/* it is present, return fine */
return there;
}
/* not present, add it */
check = calloc(1, sizeof(struct Curl_sh_entry));
if(!check)
return NULL; /* major failure */
Curl_hash_init(&check->transfers, TRHASH_SIZE, trhash, trhash_compare,
trhash_dtor);
/* make/add new hash entry */
if(!Curl_hash_add(sh, (char *)&s, sizeof(curl_socket_t), check)) {
Curl_hash_destroy(&check->transfers);
free(check);
return NULL; /* major failure */
}
return check; /* things are good in sockhash land */
}
/* delete the given socket + handle from the hash */
static void sh_delentry(struct Curl_sh_entry *entry,
struct Curl_hash *sh, curl_socket_t s)
{
Curl_hash_destroy(&entry->transfers);
/* We remove the hash entry. This will end up in a call to
sh_freeentry(). */
Curl_hash_delete(sh, (char *)&s, sizeof(curl_socket_t));
}
/*
* free a sockhash entry
*/
static void sh_freeentry(void *freethis)
{
struct Curl_sh_entry *p = (struct Curl_sh_entry *) freethis;
free(p);
}
static size_t fd_key_compare(void *k1, size_t k1_len, void *k2, size_t k2_len)
{
(void) k1_len; (void) k2_len;
return (*((curl_socket_t *) k1)) == (*((curl_socket_t *) k2));
}
static size_t hash_fd(void *key, size_t key_length, size_t slots_num)
{
curl_socket_t fd = *((curl_socket_t *) key);
(void) key_length;
return (fd % (curl_socket_t)slots_num);
}
/*
* sh_init() creates a new socket hash and returns the handle for it.
*
* Quote from README.multi_socket:
*
* "Some tests at 7000 and 9000 connections showed that the socket hash lookup
* is somewhat of a bottle neck. Its current implementation may be a bit too
* limiting. It simply has a fixed-size array, and on each entry in the array
* it has a linked list with entries. The hash only checks which list to scan
* through. The code I had used so for used a list with merely 7 slots (as
* that is what the DNS hash uses) but with 7000 connections that would make
* an average of 1000 nodes in each list to run through. I upped that to 97
* slots (I believe a prime is suitable) and noticed a significant speed
* increase. I need to reconsider the hash implementation or use a rather
* large default value like this. At 9000 connections I was still below 10us
* per call."
*
*/
static void sh_init(struct Curl_hash *hash, size_t hashsize)
{
Curl_hash_init(hash, hashsize, hash_fd, fd_key_compare,
sh_freeentry);
}
/* multi->proto_hash destructor. Should never be called as elements
* MUST be added with their own destructor */
@ -400,7 +209,7 @@ static void multi_addmsg(struct Curl_multi *multi, struct Curl_message *msg)
Curl_llist_append(&multi->msglist, msg, &msg->list);
}
struct Curl_multi *Curl_multi_handle(size_t hashsize, /* socket hash */
struct Curl_multi *Curl_multi_handle(size_t ev_hashsize, /* event hash */
size_t chashsize, /* connection hash */
size_t dnssize, /* dns hash */
size_t sesssize) /* TLS session cache */
@ -413,8 +222,7 @@ struct Curl_multi *Curl_multi_handle(size_t hashsize, /* socket hash */
multi->magic = CURL_MULTI_HANDLE;
Curl_init_dnscache(&multi->hostcache, dnssize);
sh_init(&multi->sockhash, hashsize);
Curl_multi_ev_init(multi, ev_hashsize);
Curl_hash_init(&multi->proto_hash, 23,
Curl_hash_str, Curl_str_key_compare, ph_freeentry);
@ -452,7 +260,7 @@ struct Curl_multi *Curl_multi_handle(size_t hashsize, /* socket hash */
error:
sockhash_destroy(&multi->sockhash);
Curl_multi_ev_cleanup(multi);
Curl_hash_destroy(&multi->proto_hash);
Curl_hash_destroy(&multi->hostcache);
Curl_cpool_destroy(&multi->cpool);
@ -588,6 +396,7 @@ CURLMcode curl_multi_add_handle(CURLM *m, CURL *d)
Curl_cpool_xfer_init(data);
multi_warn_debug(multi, data);
CURL_TRC_M(data, "added, transfers=%u", multi->num_easy);
return CURLM_OK;
}
@ -621,9 +430,8 @@ static void multi_done_locked(struct connectdata *conn,
if(CONN_INUSE(conn)) {
/* Stop if still used. */
DEBUGF(infof(data, "Connection still in use %zu, "
"no more multi_done now!",
Curl_llist_count(&conn->easyq)));
CURL_TRC_M(data, "Connection still in use %zu, no more multi_done now!",
Curl_llist_count(&conn->easyq));
return;
}
@ -660,12 +468,12 @@ static void multi_done_locked(struct connectdata *conn,
#endif
) || conn->bits.close
|| (mdctx->premature && !Curl_conn_is_multiplex(conn, FIRSTSOCKET))) {
DEBUGF(infof(data, "multi_done, not reusing connection=%"
FMT_OFF_T ", forbid=%d"
", close=%d, premature=%d, conn_multiplex=%d",
conn->connection_id, data->set.reuse_forbid,
conn->bits.close, mdctx->premature,
Curl_conn_is_multiplex(conn, FIRSTSOCKET)));
CURL_TRC_M(data, "multi_done, not reusing connection=%"
FMT_OFF_T ", forbid=%d"
", close=%d, premature=%d, conn_multiplex=%d",
conn->connection_id, data->set.reuse_forbid,
conn->bits.close, mdctx->premature,
Curl_conn_is_multiplex(conn, FIRSTSOCKET));
connclose(conn, "disconnecting");
Curl_cpool_disconnect(data, conn, mdctx->premature);
}
@ -703,14 +511,8 @@ static CURLcode multi_done(struct Curl_easy *data,
memset(&mdctx, 0, sizeof(mdctx));
#if defined(DEBUGBUILD) && !defined(CURL_DISABLE_VERBOSE_STRINGS)
DEBUGF(infof(data, "multi_done[%s]: status: %d prem: %d done: %d",
multi_statename[data->mstate],
(int)status, (int)premature, data->state.done));
#else
DEBUGF(infof(data, "multi_done: status: %d prem: %d done: %d",
(int)status, (int)premature, data->state.done));
#endif
CURL_TRC_M(data, "multi_done: status: %d prem: %d done: %d",
(int)status, (int)premature, data->state.done);
if(data->state.done)
/* Stop if multi_done() has already been called */
@ -861,18 +663,14 @@ CURLMcode curl_multi_remove_handle(CURLM *m, CURL *d)
Curl_wildcard_dtor(&data->wildcard);
/* change state without using multistate(), only to make singlesocket() do
what we want */
data->mstate = MSTATE_COMPLETED;
/* This ignores the return code even in case of problems because there is
nothing more to do about that, here */
(void)singlesocket(multi, data); /* to let the application know what sockets
that vanish with this handle */
/* Remove the association between the connection and the handle */
Curl_detach_connection(data);
/* Tell event handling that this transfer is definitely going away */
Curl_multi_ev_xfer_done(multi, data);
if(data->set.connect_only && !data->multi_easy) {
/* This removes a handle that was part the multi interface that used
CONNECT_ONLY, that connection is now left alive but since this handle
@ -927,6 +725,8 @@ CURLMcode curl_multi_remove_handle(CURLM *m, CURL *d)
if(rc)
return rc;
}
CURL_TRC_M(data, "removed, transfers=%u", multi->num_easy);
return CURLM_OK;
}
@ -1068,13 +868,15 @@ static int perform_getsock(struct Curl_easy *data, curl_socket_t *sock)
/* Initializes `poll_set` with the current socket poll actions needed
* for transfer `data`. */
static void multi_getsock(struct Curl_easy *data,
struct easy_pollset *ps)
void Curl_multi_getsock(struct Curl_easy *data,
struct easy_pollset *ps,
const char *caller)
{
bool expect_sockets = TRUE;
/* The no connection case can happen when this is called from
curl_multi_remove_handle() => singlesocket() => multi_getsock().
*/
/* If the transfer has no connection, this is fine. Happens when
called via curl_multi_remove_handle() => Curl_multi_ev_assess() =>
Curl_multi_getsock(). */
Curl_pollset_reset(data, ps);
if(!data->conn)
return;
@ -1098,30 +900,30 @@ static void multi_getsock(struct Curl_easy *data,
case MSTATE_CONNECTING:
case MSTATE_TUNNELING:
Curl_pollset_add_socks(data, ps, connecting_getsock);
Curl_conn_adjust_pollset(data, ps);
Curl_conn_adjust_pollset(data, data->conn, ps);
break;
case MSTATE_PROTOCONNECT:
case MSTATE_PROTOCONNECTING:
Curl_pollset_add_socks(data, ps, protocol_getsock);
Curl_conn_adjust_pollset(data, ps);
Curl_conn_adjust_pollset(data, data->conn, ps);
break;
case MSTATE_DO:
case MSTATE_DOING:
Curl_pollset_add_socks(data, ps, doing_getsock);
Curl_conn_adjust_pollset(data, ps);
Curl_conn_adjust_pollset(data, data->conn, ps);
break;
case MSTATE_DOING_MORE:
Curl_pollset_add_socks(data, ps, domore_getsock);
Curl_conn_adjust_pollset(data, ps);
Curl_conn_adjust_pollset(data, data->conn, ps);
break;
case MSTATE_DID: /* same as PERFORMING in regard to polling */
case MSTATE_PERFORMING:
Curl_pollset_add_socks(data, ps, perform_getsock);
Curl_conn_adjust_pollset(data, ps);
Curl_conn_adjust_pollset(data, data->conn, ps);
break;
case MSTATE_RATELIMITING:
@ -1143,6 +945,35 @@ static void multi_getsock(struct Curl_easy *data,
break;
}
switch(ps->num) {
case 0:
CURL_TRC_M(data, "%s pollset[], timeouts=%zu, paused %d/%d (r/w)",
caller, Curl_llist_count(&data->state.timeoutlist),
Curl_creader_is_paused(data), Curl_cwriter_is_paused(data));
break;
case 1:
CURL_TRC_M(data, "%s pollset[fd=%" FMT_SOCKET_T " %s%s], timeouts=%zu",
caller, ps->sockets[0],
(ps->actions[0] & CURL_POLL_IN) ? "IN" : "",
(ps->actions[0] & CURL_POLL_OUT) ? "OUT" : "",
Curl_llist_count(&data->state.timeoutlist));
break;
case 2:
CURL_TRC_M(data, "%s pollset[fd=%" FMT_SOCKET_T " %s%s, "
"fd=%" FMT_SOCKET_T " %s%s], timeouts=%zu",
caller, ps->sockets[0],
(ps->actions[0] & CURL_POLL_IN) ? "IN" : "",
(ps->actions[0] & CURL_POLL_OUT) ? "OUT" : "",
ps->sockets[1],
(ps->actions[1] & CURL_POLL_IN) ? "IN" : "",
(ps->actions[1] & CURL_POLL_OUT) ? "OUT" : "",
Curl_llist_count(&data->state.timeoutlist));
break;
default:
CURL_TRC_M(data, "%s pollset[fds=%u], timeouts=%zu",
caller, ps->num, Curl_llist_count(&data->state.timeoutlist));
break;
}
if(expect_sockets && !ps->num &&
!Curl_llist_count(&data->state.timeoutlist) &&
!Curl_cwriter_is_paused(data) && !Curl_creader_is_paused(data) &&
@ -1177,26 +1008,27 @@ CURLMcode curl_multi_fdset(CURLM *m,
for(e = Curl_llist_head(&multi->process); e; e = Curl_node_next(e)) {
struct Curl_easy *data = Curl_node_elem(e);
struct easy_pollset ps;
multi_getsock(data, &data->last_poll);
Curl_multi_getsock(data, &ps, "curl_multi_fdset");
for(i = 0; i < data->last_poll.num; i++) {
if(!FDSET_SOCK(data->last_poll.sockets[i]))
for(i = 0; i < ps.num; i++) {
if(!FDSET_SOCK(ps.sockets[i]))
/* pretend it does not exist */
continue;
#if defined(__DJGPP__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Warith-conversion"
#endif
if(data->last_poll.actions[i] & CURL_POLL_IN)
FD_SET(data->last_poll.sockets[i], read_fd_set);
if(data->last_poll.actions[i] & CURL_POLL_OUT)
FD_SET(data->last_poll.sockets[i], write_fd_set);
if(ps.actions[i] & CURL_POLL_IN)
FD_SET(ps.sockets[i], read_fd_set);
if(ps.actions[i] & CURL_POLL_OUT)
FD_SET(ps.sockets[i], write_fd_set);
#if defined(__DJGPP__)
#pragma GCC diagnostic pop
#endif
if((int)data->last_poll.sockets[i] > this_max_fd)
this_max_fd = (int)data->last_poll.sockets[i];
if((int)ps.sockets[i] > this_max_fd)
this_max_fd = (int)ps.sockets[i];
}
}
@ -1230,8 +1062,10 @@ CURLMcode curl_multi_waitfds(CURLM *m,
Curl_waitfds_init(&cwfds, ufds, size);
for(e = Curl_llist_head(&multi->process); e; e = Curl_node_next(e)) {
struct Curl_easy *data = Curl_node_elem(e);
multi_getsock(data, &data->last_poll);
need += Curl_waitfds_add_ps(&cwfds, &data->last_poll);
struct easy_pollset ps;
Curl_multi_getsock(data, &ps, "curl_multi_waitfds");
need += Curl_waitfds_add_ps(&cwfds, &ps);
}
need += Curl_cpool_add_waitfds(&multi->cpool, &cwfds);
@ -1303,9 +1137,10 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
/* Add the curl handles to our pollfds first */
for(e = Curl_llist_head(&multi->process); e; e = Curl_node_next(e)) {
struct Curl_easy *data = Curl_node_elem(e);
struct easy_pollset ps;
multi_getsock(data, &data->last_poll);
if(Curl_pollfds_add_ps(&cpfds, &data->last_poll)) {
Curl_multi_getsock(data, &ps, "multi_wait");
if(Curl_pollfds_add_ps(&cpfds, &ps)) {
result = CURLM_OUT_OF_MEMORY;
goto out;
}
@ -1440,23 +1275,14 @@ static CURLMcode multi_wait(struct Curl_multi *multi,
#ifdef USE_WINSOCK
/* Count up all our own sockets that had activity,
and remove them from the event. */
if(curl_nfds) {
for(e = Curl_llist_head(&multi->process); e && !result;
e = Curl_node_next(e)) {
struct Curl_easy *data = Curl_node_elem(e);
for(i = 0; i < data->last_poll.num; i++) {
wsa_events.lNetworkEvents = 0;
if(WSAEnumNetworkEvents(data->last_poll.sockets[i], NULL,
&wsa_events) == 0) {
if(ret && !pollrc && wsa_events.lNetworkEvents)
retcode++;
}
WSAEventSelect(data->last_poll.sockets[i], multi->wsa_event, 0);
}
for(i = 0; i < curl_nfds; ++i) {
wsa_events.lNetworkEvents = 0;
if(WSAEnumNetworkEvents(cpfds.pfds[i].fd, NULL, &wsa_events) == 0) {
if(ret && !pollrc && wsa_events.lNetworkEvents)
retcode++;
}
WSAEventSelect(cpfds.pfds[i].fd, multi->wsa_event, 0);
}
WSAResetEvent(multi->wsa_event);
#else
#ifdef ENABLE_WAKEUP
@ -2246,7 +2072,7 @@ static CURLMcode state_resolving(struct Curl_multi *multi,
socket(s) will again be used further down. If the name has not yet been
resolved, it is likely that new sockets have been opened in an attempt to
contact another resolver. */
rc = singlesocket(multi, data);
rc = Curl_multi_ev_assess_xfer(multi, data);
if(rc)
return rc;
@ -2365,7 +2191,7 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
rc = CURLM_OK;
if(multi_ischanged(multi, TRUE)) {
DEBUGF(infof(data, "multi changed, check CONNECT_PEND queue"));
CURL_TRC_M(data, "multi changed, check CONNECT_PEND queue");
process_pending_handles(multi); /* multiplexed */
}
@ -2731,6 +2557,7 @@ CURLMcode curl_multi_perform(CURLM *m, int *running_handles)
struct Curl_llist_node *e;
struct Curl_llist_node *n = NULL;
struct Curl_multi *multi = m;
bool first = TRUE;
SIGPIPE_VARIABLE(pipe_st);
if(!GOOD_MULTI_HANDLE(multi))
@ -2745,12 +2572,16 @@ CURLMcode curl_multi_perform(CURLM *m, int *running_handles)
CURLMcode result;
/* Do the loop and only alter the signal ignore state if the next handle
has a different NO_SIGNAL state than the previous */
if(first) {
CURL_TRC_M(data, "multi_perform(running=%u)", multi->num_alive);
first = FALSE;
}
/* the current node might be unlinked in multi_runsingle(), get the next
pointer now */
n = Curl_node_next(e);
if(data != multi->cpool.idata) {
if(data && data != multi->cpool.idata) {
/* connection pool handle is processed below */
sigpipe_apply(data, &pipe_st);
result = multi_runsingle(multi, &now, data);
@ -2760,7 +2591,7 @@ CURLMcode curl_multi_perform(CURLM *m, int *running_handles)
}
sigpipe_apply(multi->cpool.idata, &pipe_st);
Curl_cpool_multi_perform(multi);
Curl_cpool_multi_perform(multi, CURL_SOCKET_TIMEOUT);
sigpipe_restore(&pipe_st);
if(multi_ischanged(m, TRUE))
@ -2862,7 +2693,7 @@ CURLMcode curl_multi_cleanup(CURLM *m)
multi->magic = 0; /* not good anymore */
sockhash_destroy(&multi->sockhash);
Curl_multi_ev_cleanup(multi);
Curl_hash_destroy(&multi->proto_hash);
Curl_hash_destroy(&multi->hostcache);
Curl_psl_destroy(&multi->psl);
@ -2925,233 +2756,14 @@ CURLMsg *curl_multi_info_read(CURLM *m, int *msgs_in_queue)
return NULL;
}
/*
* singlesocket() checks what sockets we deal with and their "action state"
* and if we have a different state in any of those sockets from last time we
* call the callback accordingly.
*/
static CURLMcode singlesocket(struct Curl_multi *multi,
struct Curl_easy *data)
{
struct easy_pollset cur_poll;
CURLMcode mresult;
/* Fill in the 'current' struct with the state as it is now: what sockets to
supervise and for what actions */
multi_getsock(data, &cur_poll);
mresult = Curl_multi_pollset_ev(multi, data, &cur_poll, &data->last_poll);
if(!mresult) /* Remember for next time */
memcpy(&data->last_poll, &cur_poll, sizeof(cur_poll));
return mresult;
}
CURLMcode Curl_multi_pollset_ev(struct Curl_multi *multi,
struct Curl_easy *data,
struct easy_pollset *ps,
struct easy_pollset *last_ps)
{
unsigned int i;
struct Curl_sh_entry *entry;
curl_socket_t s;
int rc;
/* We have 0 .. N sockets already and we get to know about the 0 .. M
sockets we should have from now on. Detect the differences, remove no
longer supervised ones and add new ones */
/* walk over the sockets we got right now */
for(i = 0; i < ps->num; i++) {
unsigned char cur_action = ps->actions[i];
unsigned char last_action = 0;
int comboaction;
s = ps->sockets[i];
/* get it from the hash */
entry = sh_getentry(&multi->sockhash, s);
if(entry) {
/* check if new for this transfer */
unsigned int j;
for(j = 0; j < last_ps->num; j++) {
if(s == last_ps->sockets[j]) {
last_action = last_ps->actions[j];
break;
}
}
}
else {
/* this is a socket we did not have before, add it to the hash! */
entry = sh_addentry(&multi->sockhash, s);
if(!entry)
/* fatal */
return CURLM_OUT_OF_MEMORY;
}
if(last_action && (last_action != cur_action)) {
/* Socket was used already, but different action now */
if(last_action & CURL_POLL_IN) {
DEBUGASSERT(entry->readers);
entry->readers--;
}
if(last_action & CURL_POLL_OUT) {
DEBUGASSERT(entry->writers);
entry->writers--;
}
if(cur_action & CURL_POLL_IN) {
entry->readers++;
}
if(cur_action & CURL_POLL_OUT)
entry->writers++;
}
else if(!last_action &&
!Curl_hash_pick(&entry->transfers, (char *)&data, /* hash key */
sizeof(struct Curl_easy *))) {
DEBUGASSERT(entry->users < 100000); /* detect weird values */
/* a new transfer using this socket */
entry->users++;
if(cur_action & CURL_POLL_IN)
entry->readers++;
if(cur_action & CURL_POLL_OUT)
entry->writers++;
/* add 'data' to the transfer hash on this socket! */
if(!Curl_hash_add(&entry->transfers, (char *)&data, /* hash key */
sizeof(struct Curl_easy *), data)) {
Curl_hash_destroy(&entry->transfers);
return CURLM_OUT_OF_MEMORY;
}
}
comboaction = (entry->writers ? CURL_POLL_OUT : 0) |
(entry->readers ? CURL_POLL_IN : 0);
/* socket existed before and has the same action set as before */
if(last_action && ((int)entry->action == comboaction))
/* same, continue */
continue;
if(multi->socket_cb) {
set_in_callback(multi, TRUE);
rc = multi->socket_cb(data, s, comboaction, multi->socket_userp,
entry->socketp);
set_in_callback(multi, FALSE);
if(rc == -1) {
multi->dead = TRUE;
return CURLM_ABORTED_BY_CALLBACK;
}
}
/* store the current action state */
entry->action = (unsigned int)comboaction;
}
/* Check for last_poll.sockets that no longer appear in ps->sockets.
* Need to remove the easy handle from the multi->sockhash->transfers and
* remove multi->sockhash entry when this was the last transfer */
for(i = 0; i < last_ps->num; i++) {
unsigned int j;
bool stillused = FALSE;
s = last_ps->sockets[i];
for(j = 0; j < ps->num; j++) {
if(s == ps->sockets[j]) {
/* this is still supervised */
stillused = TRUE;
break;
}
}
if(stillused)
continue;
entry = sh_getentry(&multi->sockhash, s);
/* if this is NULL here, the socket has been closed and notified so
already by Curl_multi_closed() */
if(entry) {
unsigned char oldactions = last_ps->actions[i];
/* this socket has been removed. Decrease user count */
DEBUGASSERT(entry->users);
entry->users--;
if(oldactions & CURL_POLL_OUT)
entry->writers--;
if(oldactions & CURL_POLL_IN)
entry->readers--;
if(!entry->users) {
bool dead = FALSE;
if(multi->socket_cb) {
set_in_callback(multi, TRUE);
rc = multi->socket_cb(data, s, CURL_POLL_REMOVE,
multi->socket_userp, entry->socketp);
set_in_callback(multi, FALSE);
if(rc == -1)
dead = TRUE;
}
sh_delentry(entry, &multi->sockhash, s);
if(dead) {
multi->dead = TRUE;
return CURLM_ABORTED_BY_CALLBACK;
}
}
else {
/* still users, but remove this handle as a user of this socket */
if(Curl_hash_delete(&entry->transfers, (char *)&data,
sizeof(struct Curl_easy *))) {
DEBUGASSERT(NULL);
}
}
}
} /* for loop over num */
return CURLM_OK;
}
CURLcode Curl_updatesocket(struct Curl_easy *data)
{
if(singlesocket(data->multi, data))
return CURLE_ABORTED_BY_CALLBACK;
return CURLE_OK;
}
/*
* Curl_multi_closed()
*
* Used by the connect code to tell the multi_socket code that one of the
* sockets we were using is about to be closed. This function will then
* remove it from the sockethash for this handle to make the multi_socket API
* behave properly, especially for the case when libcurl will create another
* socket again and it gets the same file descriptor number.
*/
void Curl_multi_closed(struct Curl_easy *data, curl_socket_t s)
void Curl_multi_will_close(struct Curl_easy *data, curl_socket_t s)
{
if(data) {
/* if there is still an easy handle associated with this connection */
struct Curl_multi *multi = data->multi;
DEBUGF(infof(data, "Curl_multi_closed, fd=%" FMT_SOCKET_T
" multi is %p", s, (void *)multi));
if(multi) {
/* this is set if this connection is part of a handle that is added to
a multi handle, and only then this is necessary */
struct Curl_sh_entry *entry = sh_getentry(&multi->sockhash, s);
DEBUGF(infof(data, "Curl_multi_closed, fd=%" FMT_SOCKET_T
" entry is %p", s, (void *)entry));
if(entry) {
int rc = 0;
if(multi->socket_cb) {
set_in_callback(multi, TRUE);
rc = multi->socket_cb(data, s, CURL_POLL_REMOVE,
multi->socket_userp, entry->socketp);
set_in_callback(multi, FALSE);
}
/* now remove it from the socket hash */
sh_delentry(entry, &multi->sockhash, s);
if(rc == -1)
/* This just marks the multi handle as "dead" without returning an
error code primarily because this function is used from many
places where propagating an error back is tricky. */
multi->dead = TRUE;
}
CURL_TRC_M(data, "Curl_multi_will_close fd=%" FMT_SOCKET_T, s);
Curl_multi_ev_socket_done(multi, data, s);
}
}
}
@ -3253,9 +2865,8 @@ static CURLMcode multi_run_expired(struct multi_run_ctx *mrc)
result = multi_runsingle(multi, &mrc->now, data);
if(CURLM_OK >= result) {
/* get the socket(s) and check if the state has been changed since
last */
result = singlesocket(multi, data);
/* reassess event handling of data */
result = Curl_multi_ev_assess_xfer(multi, data);
if(result)
goto out;
}
@ -3271,7 +2882,6 @@ static CURLMcode multi_socket(struct Curl_multi *multi,
int *running_handles)
{
CURLMcode result = CURLM_OK;
struct Curl_easy *data = NULL;
struct multi_run_ctx mrc;
(void)ev_bitmask;
@ -3281,56 +2891,19 @@ static CURLMcode multi_socket(struct Curl_multi *multi,
sigpipe_init(&mrc.pipe_st);
if(checkall) {
struct Curl_llist_node *e;
/* *perform() deals with running_handles on its own */
result = curl_multi_perform(multi, running_handles);
/* walk through each easy handle and do the socket state change magic
and callbacks */
if(result != CURLM_BAD_HANDLE) {
for(e = Curl_llist_head(&multi->process); e && !result;
e = Curl_node_next(e)) {
result = singlesocket(multi, Curl_node_elem(e));
}
/* Reassess event status of all active transfers */
result = Curl_multi_ev_assess_xfer_list(multi, &multi->process);
}
mrc.run_cpool = TRUE;
goto out;
}
if(s != CURL_SOCKET_TIMEOUT) {
struct Curl_sh_entry *entry = sh_getentry(&multi->sockhash, s);
if(!entry) {
/* Unmatched socket, we cannot act on it but we ignore this fact. In
real-world tests it has been proved that libevent can in fact give
the application actions even though the socket was just previously
asked to get removed, so thus we better survive stray socket actions
and just move on. */
/* The socket might come from a connection that is being shut down
* by the multi's connection pool. */
Curl_cpool_multi_socket(multi, s, ev_bitmask);
}
else {
struct Curl_hash_iterator iter;
struct Curl_hash_element *he;
/* the socket can be shared by many transfers, iterate */
Curl_hash_start_iterate(&entry->transfers, &iter);
for(he = Curl_hash_next_element(&iter); he;
he = Curl_hash_next_element(&iter)) {
data = (struct Curl_easy *)he->ptr;
DEBUGASSERT(data);
DEBUGASSERT(data->magic == CURLEASY_MAGIC_NUMBER);
if(data == multi->cpool.idata)
mrc.run_cpool = TRUE;
else {
/* Expire with out current now, so we will get it below when
* asking the splaytree for expired transfers. */
expire_ex(data, &mrc.now, 0, EXPIRE_RUN_NOW);
}
}
}
Curl_multi_ev_expire_xfers(multi, s, &mrc.now, &mrc.run_cpool);
}
else {
/* Asked to run due to time-out. Clear the 'last_expire_ts' variable to
@ -3339,6 +2912,7 @@ static CURLMcode multi_socket(struct Curl_multi *multi,
handles the case when the application asks libcurl to run the timeout
prematurely. */
memset(&multi->last_expire_ts, 0, sizeof(multi->last_expire_ts));
mrc.run_cpool = TRUE;
}
result = multi_run_expired(&mrc);
@ -3358,7 +2932,7 @@ static CURLMcode multi_socket(struct Curl_multi *multi,
out:
if(mrc.run_cpool) {
sigpipe_apply(multi->cpool.idata, &mrc.pipe_st);
Curl_cpool_multi_perform(multi);
Curl_cpool_multi_perform(multi, s);
}
sigpipe_restore(&mrc.pipe_st);
@ -3675,9 +3249,9 @@ multi_addtimeout(struct Curl_easy *data,
return CURLM_OK;
}
static void expire_ex(struct Curl_easy *data,
const struct curltime *nowp,
timediff_t milli, expire_id id)
void Curl_expire_ex(struct Curl_easy *data,
const struct curltime *nowp,
timediff_t milli, expire_id id)
{
struct Curl_multi *multi = data->multi;
struct curltime *curr_expire = &data->state.expiretime;
@ -3749,7 +3323,7 @@ static void expire_ex(struct Curl_easy *data,
void Curl_expire(struct Curl_easy *data, timediff_t milli, expire_id id)
{
struct curltime now = Curl_now();
expire_ex(data, &now, milli, id);
Curl_expire_ex(data, &now, milli, id);
}
/*
@ -3793,9 +3367,7 @@ bool Curl_expire_clear(struct Curl_easy *data)
/* clear the timeout list too */
Curl_llist_destroy(list, NULL);
#ifdef DEBUGBUILD
infof(data, "Expire cleared");
#endif
CURL_TRC_M(data, "Expire cleared");
nowp->tv_sec = 0;
nowp->tv_usec = 0;
return TRUE;
@ -3806,19 +3378,11 @@ bool Curl_expire_clear(struct Curl_easy *data)
CURLMcode curl_multi_assign(CURLM *m, curl_socket_t s,
void *hashp)
{
struct Curl_sh_entry *there = NULL;
struct Curl_multi *multi = m;
if(!GOOD_MULTI_HANDLE(multi))
return CURLM_BAD_HANDLE;
there = sh_getentry(&multi->sockhash, s);
if(!there)
return CURLM_BAD_SOCKET;
there->socketp = hashp;
return CURLM_OK;
return Curl_multi_ev_assign(multi, s, hashp);
}
static void move_pending_to_connect(struct Curl_multi *multi,