mirror of
https://github.com/moparisthebest/curl
synced 2024-08-13 17:03:50 -04:00
pipeline: switch some code over to functions
... to "compartmentalize" a bit and make it easier to change behavior when multiplexing is used instead of good old pipelining.
This commit is contained in:
parent
7d0df07e48
commit
d6f1c74330
@ -1211,7 +1211,7 @@ CURLcode Curl_add_buffer_send(Curl_send_buffer *in,
|
|||||||
*/
|
*/
|
||||||
return CURLE_SEND_ERROR;
|
return CURLE_SEND_ERROR;
|
||||||
else
|
else
|
||||||
conn->writechannel_inuse = FALSE;
|
Curl_pipeline_leave_write(conn);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Curl_add_buffer_free(in);
|
Curl_add_buffer_free(in);
|
||||||
|
63
lib/multi.c
63
lib/multi.c
@ -69,8 +69,6 @@ static void singlesocket(struct Curl_multi *multi,
|
|||||||
struct SessionHandle *data);
|
struct SessionHandle *data);
|
||||||
static int update_timer(struct Curl_multi *multi);
|
static int update_timer(struct Curl_multi *multi);
|
||||||
|
|
||||||
static bool isHandleAtHead(struct SessionHandle *handle,
|
|
||||||
struct curl_llist *pipeline);
|
|
||||||
static CURLMcode add_next_timeout(struct timeval now,
|
static CURLMcode add_next_timeout(struct timeval now,
|
||||||
struct Curl_multi *multi,
|
struct Curl_multi *multi,
|
||||||
struct SessionHandle *d);
|
struct SessionHandle *d);
|
||||||
@ -1269,19 +1267,8 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
|
|||||||
|
|
||||||
case CURLM_STATE_WAITDO:
|
case CURLM_STATE_WAITDO:
|
||||||
/* Wait for our turn to DO when we're pipelining requests */
|
/* Wait for our turn to DO when we're pipelining requests */
|
||||||
#ifdef DEBUGBUILD
|
if(Curl_pipeline_checkget_write(data, data->easy_conn)) {
|
||||||
infof(data, "WAITDO: Conn %ld send pipe %zu inuse %s athead %s\n",
|
/* Grabbed the channel */
|
||||||
data->easy_conn->connection_id,
|
|
||||||
data->easy_conn->send_pipe->size,
|
|
||||||
data->easy_conn->writechannel_inuse?"TRUE":"FALSE",
|
|
||||||
isHandleAtHead(data,
|
|
||||||
data->easy_conn->send_pipe)?"TRUE":"FALSE");
|
|
||||||
#endif
|
|
||||||
if(!data->easy_conn->writechannel_inuse &&
|
|
||||||
isHandleAtHead(data,
|
|
||||||
data->easy_conn->send_pipe)) {
|
|
||||||
/* Grab the channel */
|
|
||||||
data->easy_conn->writechannel_inuse = TRUE;
|
|
||||||
multistate(data, CURLM_STATE_DO);
|
multistate(data, CURLM_STATE_DO);
|
||||||
rc = CURLM_CALL_MULTI_PERFORM;
|
rc = CURLM_CALL_MULTI_PERFORM;
|
||||||
}
|
}
|
||||||
@ -1464,25 +1451,11 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
|
|||||||
|
|
||||||
case CURLM_STATE_WAITPERFORM:
|
case CURLM_STATE_WAITPERFORM:
|
||||||
/* Wait for our turn to PERFORM */
|
/* Wait for our turn to PERFORM */
|
||||||
if((!data->easy_conn->readchannel_inuse &&
|
if(Curl_pipeline_checkget_read(data, data->easy_conn)) {
|
||||||
isHandleAtHead(data,
|
/* Grabbed the channel */
|
||||||
data->easy_conn->recv_pipe)) ||
|
|
||||||
data->easy_conn->bits.multiplex) {
|
|
||||||
/* Grab the channel */
|
|
||||||
data->easy_conn->readchannel_inuse = TRUE;
|
|
||||||
multistate(data, CURLM_STATE_PERFORM);
|
multistate(data, CURLM_STATE_PERFORM);
|
||||||
rc = CURLM_CALL_MULTI_PERFORM;
|
rc = CURLM_CALL_MULTI_PERFORM;
|
||||||
}
|
}
|
||||||
#ifdef DEBUGBUILD
|
|
||||||
else {
|
|
||||||
infof(data, "WAITPERFORM: Conn %ld recv pipe %zu inuse %s athead %s\n",
|
|
||||||
data->easy_conn->connection_id,
|
|
||||||
data->easy_conn->recv_pipe->size,
|
|
||||||
data->easy_conn->readchannel_inuse?"TRUE":"FALSE",
|
|
||||||
isHandleAtHead(data,
|
|
||||||
data->easy_conn->recv_pipe)?"TRUE":"FALSE");
|
|
||||||
}
|
|
||||||
#endif
|
|
||||||
break;
|
break;
|
||||||
|
|
||||||
case CURLM_STATE_TOOFAST: /* limit-rate exceeded in either direction */
|
case CURLM_STATE_TOOFAST: /* limit-rate exceeded in either direction */
|
||||||
@ -1543,15 +1516,13 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
|
|||||||
|
|
||||||
k = &data->req;
|
k = &data->req;
|
||||||
|
|
||||||
if(!(k->keepon & KEEP_RECV)) {
|
if(!(k->keepon & KEEP_RECV))
|
||||||
/* We're done receiving */
|
/* We're done receiving */
|
||||||
data->easy_conn->readchannel_inuse = FALSE;
|
Curl_pipeline_leave_read(data->easy_conn);
|
||||||
}
|
|
||||||
|
|
||||||
if(!(k->keepon & KEEP_SEND)) {
|
if(!(k->keepon & KEEP_SEND))
|
||||||
/* We're done sending */
|
/* We're done sending */
|
||||||
data->easy_conn->writechannel_inuse = FALSE;
|
Curl_pipeline_leave_write(data->easy_conn);
|
||||||
}
|
|
||||||
|
|
||||||
if(done || (result == CURLE_RECV_ERROR)) {
|
if(done || (result == CURLE_RECV_ERROR)) {
|
||||||
/* If CURLE_RECV_ERROR happens early enough, we assume it was a race
|
/* If CURLE_RECV_ERROR happens early enough, we assume it was a race
|
||||||
@ -1733,8 +1704,8 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
|
|||||||
|
|
||||||
if(data->easy_conn) {
|
if(data->easy_conn) {
|
||||||
/* if this has a connection, unsubscribe from the pipelines */
|
/* if this has a connection, unsubscribe from the pipelines */
|
||||||
data->easy_conn->writechannel_inuse = FALSE;
|
Curl_pipeline_leave_write(data->easy_conn);
|
||||||
data->easy_conn->readchannel_inuse = FALSE;
|
Curl_pipeline_leave_read(data->easy_conn);
|
||||||
Curl_removeHandleFromPipeline(data, data->easy_conn->send_pipe);
|
Curl_removeHandleFromPipeline(data, data->easy_conn->send_pipe);
|
||||||
Curl_removeHandleFromPipeline(data, data->easy_conn->recv_pipe);
|
Curl_removeHandleFromPipeline(data, data->easy_conn->recv_pipe);
|
||||||
|
|
||||||
@ -2081,7 +2052,7 @@ static void singlesocket(struct Curl_multi *multi,
|
|||||||
for the recv_pipe, or the first (in case this particular easy
|
for the recv_pipe, or the first (in case this particular easy
|
||||||
isn't already) */
|
isn't already) */
|
||||||
if(entry->easy == data) {
|
if(entry->easy == data) {
|
||||||
if(isHandleAtHead(data, easy_conn->recv_pipe))
|
if(Curl_recvpipe_head(data, easy_conn))
|
||||||
entry->easy = easy_conn->recv_pipe->head->next->ptr;
|
entry->easy = easy_conn->recv_pipe->head->next->ptr;
|
||||||
else
|
else
|
||||||
entry->easy = easy_conn->recv_pipe->head->ptr;
|
entry->easy = easy_conn->recv_pipe->head->ptr;
|
||||||
@ -2095,7 +2066,7 @@ static void singlesocket(struct Curl_multi *multi,
|
|||||||
for the send_pipe, or the first (in case this particular easy
|
for the send_pipe, or the first (in case this particular easy
|
||||||
isn't already) */
|
isn't already) */
|
||||||
if(entry->easy == data) {
|
if(entry->easy == data) {
|
||||||
if(isHandleAtHead(data, easy_conn->send_pipe))
|
if(Curl_sendpipe_head(data, easy_conn))
|
||||||
entry->easy = easy_conn->send_pipe->head->next->ptr;
|
entry->easy = easy_conn->send_pipe->head->next->ptr;
|
||||||
else
|
else
|
||||||
entry->easy = easy_conn->send_pipe->head->ptr;
|
entry->easy = easy_conn->send_pipe->head->ptr;
|
||||||
@ -2533,16 +2504,6 @@ static int update_timer(struct Curl_multi *multi)
|
|||||||
return multi->timer_cb((CURLM*)multi, timeout_ms, multi->timer_userp);
|
return multi->timer_cb((CURLM*)multi, timeout_ms, multi->timer_userp);
|
||||||
}
|
}
|
||||||
|
|
||||||
static bool isHandleAtHead(struct SessionHandle *handle,
|
|
||||||
struct curl_llist *pipeline)
|
|
||||||
{
|
|
||||||
struct curl_llist_element *curr = pipeline->head;
|
|
||||||
if(curr)
|
|
||||||
return (curr->ptr == handle) ? TRUE : FALSE;
|
|
||||||
|
|
||||||
return FALSE;
|
|
||||||
}
|
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* multi_freetimeout()
|
* multi_freetimeout()
|
||||||
*
|
*
|
||||||
|
@ -113,7 +113,7 @@ CURLcode Curl_add_handle_to_pipeline(struct SessionHandle *handle,
|
|||||||
|
|
||||||
if(pipeline == conn->send_pipe && sendhead != conn->send_pipe->head) {
|
if(pipeline == conn->send_pipe && sendhead != conn->send_pipe->head) {
|
||||||
/* this is a new one as head, expire it */
|
/* this is a new one as head, expire it */
|
||||||
conn->writechannel_inuse = FALSE; /* not in use yet */
|
Curl_pipeline_leave_write(conn); /* not in use yet */
|
||||||
Curl_expire(conn->send_pipe->head->ptr, 1);
|
Curl_expire(conn->send_pipe->head->ptr, 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -144,7 +144,7 @@ void Curl_move_handle_from_send_to_recv_pipe(struct SessionHandle *handle,
|
|||||||
if(conn->send_pipe->head) {
|
if(conn->send_pipe->head) {
|
||||||
/* Since there's a new easy handle at the start of the send pipeline,
|
/* Since there's a new easy handle at the start of the send pipeline,
|
||||||
set its timeout value to 1ms to make it trigger instantly */
|
set its timeout value to 1ms to make it trigger instantly */
|
||||||
conn->writechannel_inuse = FALSE; /* not used now */
|
Curl_pipeline_leave_write(conn); /* not used now */
|
||||||
#ifdef DEBUGBUILD
|
#ifdef DEBUGBUILD
|
||||||
infof(conn->data, "%p is at send pipe head B!\n",
|
infof(conn->data, "%p is at send pipe head B!\n",
|
||||||
(void *)conn->send_pipe->head->ptr);
|
(void *)conn->send_pipe->head->ptr);
|
||||||
@ -320,6 +320,93 @@ CURLMcode Curl_pipeline_set_server_blacklist(char **servers,
|
|||||||
return CURLM_OK;
|
return CURLM_OK;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static bool pipe_head(struct SessionHandle *data,
|
||||||
|
struct curl_llist *pipeline)
|
||||||
|
{
|
||||||
|
struct curl_llist_element *curr = pipeline->head;
|
||||||
|
if(curr)
|
||||||
|
return (curr->ptr == data) ? TRUE : FALSE;
|
||||||
|
|
||||||
|
return FALSE;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* returns TRUE if the given handle is head of the recv pipe */
|
||||||
|
bool Curl_recvpipe_head(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn)
|
||||||
|
{
|
||||||
|
return pipe_head(data, conn->recv_pipe);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* returns TRUE if the given handle is head of the send pipe */
|
||||||
|
bool Curl_sendpipe_head(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn)
|
||||||
|
{
|
||||||
|
return pipe_head(data, conn->send_pipe);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Check if the write channel is available and this handle as at the head,
|
||||||
|
* then grab the channel and return TRUE.
|
||||||
|
*
|
||||||
|
* If not available, return FALSE.
|
||||||
|
*/
|
||||||
|
|
||||||
|
bool Curl_pipeline_checkget_write(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn)
|
||||||
|
{
|
||||||
|
if(conn->bits.multiplex)
|
||||||
|
/* when multiplexing, we can use it at once */
|
||||||
|
return TRUE;
|
||||||
|
|
||||||
|
if(!conn->writechannel_inuse && Curl_sendpipe_head(data, conn)) {
|
||||||
|
/* Grab the channel */
|
||||||
|
conn->writechannel_inuse = TRUE;
|
||||||
|
return TRUE;
|
||||||
|
}
|
||||||
|
return FALSE;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Check if the read channel is available and this handle as at the head, then
|
||||||
|
* grab the channel and return TRUE.
|
||||||
|
*
|
||||||
|
* If not available, return FALSE.
|
||||||
|
*/
|
||||||
|
|
||||||
|
bool Curl_pipeline_checkget_read(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn)
|
||||||
|
{
|
||||||
|
if(conn->bits.multiplex)
|
||||||
|
/* when multiplexing, we can use it at once */
|
||||||
|
return TRUE;
|
||||||
|
|
||||||
|
if(!conn->readchannel_inuse && Curl_recvpipe_head(data, conn)) {
|
||||||
|
/* Grab the channel */
|
||||||
|
conn->readchannel_inuse = TRUE;
|
||||||
|
return TRUE;
|
||||||
|
}
|
||||||
|
return FALSE;
|
||||||
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
* The current user of the pipeline write channel gives it up.
|
||||||
|
*/
|
||||||
|
void Curl_pipeline_leave_write(struct connectdata *conn)
|
||||||
|
{
|
||||||
|
conn->writechannel_inuse = FALSE;
|
||||||
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
* The current user of the pipeline read channel gives it up.
|
||||||
|
*/
|
||||||
|
void Curl_pipeline_leave_read(struct connectdata *conn)
|
||||||
|
{
|
||||||
|
conn->readchannel_inuse = FALSE;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
#if 0
|
#if 0
|
||||||
void print_pipeline(struct connectdata *conn)
|
void print_pipeline(struct connectdata *conn)
|
||||||
{
|
{
|
||||||
|
@ -7,6 +7,7 @@
|
|||||||
* | (__| |_| | _ <| |___
|
* | (__| |_| | _ <| |___
|
||||||
* \___|\___/|_| \_\_____|
|
* \___|\___/|_| \_\_____|
|
||||||
*
|
*
|
||||||
|
* Copyright (C) 2015, Daniel Stenberg, <daniel@haxx.se>, et al.
|
||||||
* Copyright (C) 2013 - 2014, Linus Nielsen Feltzing, <linus@haxx.se>
|
* Copyright (C) 2013 - 2014, Linus Nielsen Feltzing, <linus@haxx.se>
|
||||||
*
|
*
|
||||||
* This software is licensed as described in the file COPYING, which
|
* This software is licensed as described in the file COPYING, which
|
||||||
@ -41,4 +42,15 @@ bool Curl_pipeline_server_blacklisted(struct SessionHandle *handle,
|
|||||||
CURLMcode Curl_pipeline_set_server_blacklist(char **servers,
|
CURLMcode Curl_pipeline_set_server_blacklist(char **servers,
|
||||||
struct curl_llist **list_ptr);
|
struct curl_llist **list_ptr);
|
||||||
|
|
||||||
|
bool Curl_pipeline_checkget_write(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn);
|
||||||
|
bool Curl_pipeline_checkget_read(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn);
|
||||||
|
void Curl_pipeline_leave_write(struct connectdata *conn);
|
||||||
|
void Curl_pipeline_leave_read(struct connectdata *conn);
|
||||||
|
bool Curl_recvpipe_head(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn);
|
||||||
|
bool Curl_sendpipe_head(struct SessionHandle *data,
|
||||||
|
struct connectdata *conn);
|
||||||
|
|
||||||
#endif /* HEADER_CURL_PIPELINE_H */
|
#endif /* HEADER_CURL_PIPELINE_H */
|
||||||
|
@ -2885,15 +2885,14 @@ void Curl_getoff_all_pipelines(struct SessionHandle *data,
|
|||||||
struct connectdata *conn)
|
struct connectdata *conn)
|
||||||
{
|
{
|
||||||
bool recv_head = (conn->readchannel_inuse &&
|
bool recv_head = (conn->readchannel_inuse &&
|
||||||
(gethandleathead(conn->recv_pipe) == data)) ? TRUE : FALSE;
|
Curl_recvpipe_head(data, conn));
|
||||||
|
|
||||||
bool send_head = (conn->writechannel_inuse &&
|
bool send_head = (conn->writechannel_inuse &&
|
||||||
(gethandleathead(conn->send_pipe) == data)) ? TRUE : FALSE;
|
Curl_sendpipe_head(data, conn));
|
||||||
|
|
||||||
if(Curl_removeHandleFromPipeline(data, conn->recv_pipe) && recv_head)
|
if(Curl_removeHandleFromPipeline(data, conn->recv_pipe) && recv_head)
|
||||||
conn->readchannel_inuse = FALSE;
|
Curl_pipeline_leave_read(conn);
|
||||||
if(Curl_removeHandleFromPipeline(data, conn->send_pipe) && send_head)
|
if(Curl_removeHandleFromPipeline(data, conn->send_pipe) && send_head)
|
||||||
conn->writechannel_inuse = FALSE;
|
Curl_pipeline_leave_write(conn);
|
||||||
}
|
}
|
||||||
|
|
||||||
static void signalPipeClose(struct curl_llist *pipeline, bool pipe_broke)
|
static void signalPipeClose(struct curl_llist *pipeline, bool pipe_broke)
|
||||||
|
Loading…
Reference in New Issue
Block a user