lib: move client writer into own source
Refactoring of the client writer that passes the data to the client/application's callback functions. - split out into own source cw-out.[ch] from sendf.c - move tempwrite and tempcount from data->state into the context of the client writer - redesign the 3 tempwrite dynbufs as a linked list of dynbufs. On paused transfers, this allows to "record" interleaved HEADER/BODY chunks to be "played back" in the same order on unpausing. - keep the overall size limit of all buffered data to DYN_PAUSE_BUFFER. On exceeding that, return CURLE_TOO_LARGE instead of CURLE_OUT_OF_MEMORY as before. - add method to be called when a transfer is DONE to allow writing of any data still buffered - when paused, record HEADER writes exactly as they come for later playback. HEADERs are documented to be written one-by-one. Closes #12898
This commit is contained in:
Stefan Eissing
Daniel Stenberg
Родитель
2254551403
Коммит
463472a2d6
|
|
@ -138,6 +138,7 @@ LIB_CFILES = \
|
|||
curl_sspi.c \
|
||||
curl_threads.c \
|
||||
curl_trc.c \
|
||||
cw-out.c \
|
||||
dict.c \
|
||||
doh.c \
|
||||
dynbuf.c \
|
||||
|
|
@ -283,6 +284,7 @@ LIB_HFILES = \
|
|||
curl_threads.h \
|
||||
curl_trc.h \
|
||||
curlx.h \
|
||||
cw-out.h \
|
||||
dict.h \
|
||||
doh.h \
|
||||
dynbuf.h \
|
||||
|
|
|
|||
|
|
@ -0,0 +1,437 @@
|
|||
/***************************************************************************
|
||||
* _ _ ____ _
|
||||
* Project ___| | | | _ \| |
|
||||
* / __| | | | |_) | |
|
||||
* | (__| |_| | _ <| |___
|
||||
* \___|\___/|_| \_\_____|
|
||||
*
|
||||
* Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
|
||||
*
|
||||
* This software is licensed as described in the file COPYING, which
|
||||
* you should have received as part of this distribution. The terms
|
||||
* are also available at https://curl.se/docs/copyright.html.
|
||||
*
|
||||
* You may opt to use, copy, modify, merge, publish, distribute and/or sell
|
||||
* copies of the Software, and permit persons to whom the Software is
|
||||
* furnished to do so, under the terms of the COPYING file.
|
||||
*
|
||||
* This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
|
||||
* KIND, either express or implied.
|
||||
*
|
||||
* SPDX-License-Identifier: curl
|
||||
*
|
||||
***************************************************************************/
|
||||
|
||||
#include "curl_setup.h"
|
||||
|
||||
#include <curl/curl.h>
|
||||
|
||||
#include "urldata.h"
|
||||
#include "cfilters.h"
|
||||
#include "headers.h"
|
||||
#include "multiif.h"
|
||||
#include "sendf.h"
|
||||
#include "cw-out.h"
|
||||
|
||||
/* The last 3 #include files should be in this order */
|
||||
#include "curl_printf.h"
|
||||
#include "curl_memory.h"
|
||||
#include "memdebug.h"
|
||||
|
||||
|
||||
/**
|
||||
* OVERALL DESIGN of this client writer
|
||||
*
|
||||
* The 'cw-out' writer is supposed to be the last writer in a transfer's
|
||||
* stack. It is always added when that stack is initialized. Its purpose
|
||||
* is to pass BODY and HEADER bytes to the client-installed callback
|
||||
* functions.
|
||||
*
|
||||
* These callback may return `CURL_WRITEFUNC_PAUSE` to indicate that the
|
||||
* data had not been written and the whole transfer should stop receiving
|
||||
* new data. Or at least, stop calling the functions. When the transfer
|
||||
* is "unpaused" by the client, the previous data shall be passed as
|
||||
* if nothing happened.
|
||||
*
|
||||
* The `cw-out` writer therefore manages buffers for bytes that could
|
||||
* not be written. Data that was already in flight from the server also
|
||||
* needs buffering on paused transfer when it arrives.
|
||||
*
|
||||
* In addition, the writer allows buffering of "small" body writes,
|
||||
* so client functions are called less often. That is only enabled on a
|
||||
* number of conditions.
|
||||
*
|
||||
* HEADER and BODY data may arrive in any order. For paused transfers,
|
||||
* a list of `struct cw_out_buf` is kept for `cw_out_type` types. The
|
||||
* list may be: [BODY]->[HEADER]->[BODY]->[HEADER]....
|
||||
* When unpausing, this list is "played back" to the client callbacks.
|
||||
*
|
||||
* The amount of bytes being buffered is limited by `DYN_PAUSE_BUFFER`
|
||||
* and when that is exceeded `CURLE_TOO_LARGE` is returned as error.
|
||||
*/
|
||||
typedef enum {
|
||||
CW_OUT_NONE,
|
||||
CW_OUT_BODY,
|
||||
CW_OUT_HDS
|
||||
} cw_out_type;
|
||||
|
||||
struct cw_out_buf {
|
||||
struct cw_out_buf *next;
|
||||
struct dynbuf b;
|
||||
cw_out_type type;
|
||||
};
|
||||
|
||||
static struct cw_out_buf *cw_out_buf_create(cw_out_type otype)
|
||||
{
|
||||
struct cw_out_buf *cwbuf = calloc(1, sizeof(*cwbuf));
|
||||
if(cwbuf) {
|
||||
cwbuf->type = otype;
|
||||
Curl_dyn_init(&cwbuf->b, DYN_PAUSE_BUFFER);
|
||||
}
|
||||
return cwbuf;
|
||||
}
|
||||
|
||||
static void cw_out_buf_free(struct cw_out_buf *cwbuf)
|
||||
{
|
||||
if(cwbuf) {
|
||||
Curl_dyn_free(&cwbuf->b);
|
||||
free(cwbuf);
|
||||
}
|
||||
}
|
||||
|
||||
struct cw_out_ctx {
|
||||
struct Curl_cwriter super;
|
||||
struct cw_out_buf *buf;
|
||||
};
|
||||
|
||||
static CURLcode cw_out_write(struct Curl_easy *data,
|
||||
struct Curl_cwriter *writer, int type,
|
||||
const char *buf, size_t nbytes);
|
||||
static void cw_out_close(struct Curl_easy *data, struct Curl_cwriter *writer);
|
||||
static CURLcode cw_out_init(struct Curl_easy *data,
|
||||
struct Curl_cwriter *writer);
|
||||
|
||||
struct Curl_cwtype Curl_cwt_out = {
|
||||
"cw-out",
|
||||
NULL,
|
||||
cw_out_init,
|
||||
cw_out_write,
|
||||
cw_out_close,
|
||||
sizeof(struct cw_out_ctx)
|
||||
};
|
||||
|
||||
static CURLcode cw_out_init(struct Curl_easy *data,
|
||||
struct Curl_cwriter *writer)
|
||||
{
|
||||
struct cw_out_ctx *ctx = (struct cw_out_ctx *)writer;
|
||||
(void)data;
|
||||
ctx->buf = NULL;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
static void cw_out_bufs_free(struct cw_out_ctx *ctx)
|
||||
{
|
||||
while(ctx->buf) {
|
||||
struct cw_out_buf *next = ctx->buf->next;
|
||||
cw_out_buf_free(ctx->buf);
|
||||
ctx->buf = next;
|
||||
}
|
||||
}
|
||||
|
||||
static size_t cw_out_bufs_len(struct cw_out_ctx *ctx)
|
||||
{
|
||||
struct cw_out_buf *cwbuf = ctx->buf;
|
||||
size_t len = 0;
|
||||
while(cwbuf) {
|
||||
len += Curl_dyn_len(&cwbuf->b);
|
||||
cwbuf = cwbuf->next;
|
||||
}
|
||||
return len;
|
||||
}
|
||||
|
||||
static void cw_out_close(struct Curl_easy *data, struct Curl_cwriter *writer)
|
||||
{
|
||||
struct cw_out_ctx *ctx = (struct cw_out_ctx *)writer;
|
||||
|
||||
(void)data;
|
||||
cw_out_bufs_free(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the current curl_write_callback and user_data for the buf type
|
||||
*/
|
||||
static void cw_get_writefunc(struct Curl_easy *data, cw_out_type otype,
|
||||
curl_write_callback *pwcb, void **pwcb_data,
|
||||
size_t *pmax_write, size_t *pmin_write)
|
||||
{
|
||||
switch(otype) {
|
||||
case CW_OUT_BODY:
|
||||
*pwcb = data->set.fwrite_func;
|
||||
*pwcb_data = data->set.out;
|
||||
*pmax_write = CURL_MAX_WRITE_SIZE;
|
||||
/* if we ever want buffering of BODY output, we can set `min_write`
|
||||
* the preferred size. The default should always be to pass data
|
||||
* to the client as it comes without delay */
|
||||
*pmin_write = 0;
|
||||
break;
|
||||
case CW_OUT_HDS:
|
||||
*pwcb = data->set.fwrite_header? data->set.fwrite_header :
|
||||
(data->set.writeheader? data->set.fwrite_func : NULL);
|
||||
*pwcb_data = data->set.writeheader;
|
||||
*pmax_write = 0; /* do not chunk-write headers, write them as they are */
|
||||
*pmin_write = 0;
|
||||
break;
|
||||
default:
|
||||
*pwcb = NULL;
|
||||
*pwcb_data = NULL;
|
||||
*pmax_write = CURL_MAX_WRITE_SIZE;
|
||||
*pmin_write = 0;
|
||||
}
|
||||
}
|
||||
|
||||
static CURLcode cw_out_ptr_flush(struct cw_out_ctx *ctx,
|
||||
struct Curl_easy *data,
|
||||
cw_out_type otype,
|
||||
bool flush_all,
|
||||
const char *buf, size_t blen,
|
||||
size_t *pconsumed)
|
||||
{
|
||||
curl_write_callback wcb;
|
||||
void *wcb_data;
|
||||
size_t max_write, min_write;
|
||||
size_t wlen, nwritten;
|
||||
|
||||
(void)ctx;
|
||||
/* write callbacks may get NULLed by the client between calls. */
|
||||
cw_get_writefunc(data, otype, &wcb, &wcb_data, &max_write, &min_write);
|
||||
if(!wcb) {
|
||||
*pconsumed = blen;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
*pconsumed = 0;
|
||||
while(blen && !(data->req.keepon & KEEP_RECV_PAUSE)) {
|
||||
if(!flush_all && blen < min_write)
|
||||
break;
|
||||
wlen = max_write? CURLMIN(blen, max_write) : blen;
|
||||
Curl_set_in_callback(data, TRUE);
|
||||
nwritten = wcb((char *)buf, 1, wlen, wcb_data);
|
||||
Curl_set_in_callback(data, FALSE);
|
||||
if(CURL_WRITEFUNC_PAUSE == nwritten) {
|
||||
if(data->conn && data->conn->handler->flags & PROTOPT_NONETWORK) {
|
||||
/* Protocols that work without network cannot be paused. This is
|
||||
actually only FILE:// just now, and it can't pause since the
|
||||
transfer isn't done using the "normal" procedure. */
|
||||
failf(data, "Write callback asked for PAUSE when not supported");
|
||||
return CURLE_WRITE_ERROR;
|
||||
}
|
||||
/* mark the connection as RECV paused */
|
||||
data->req.keepon |= KEEP_RECV_PAUSE;
|
||||
break;
|
||||
}
|
||||
if(nwritten != wlen) {
|
||||
failf(data, "Failure writing output to destination, "
|
||||
"passed %zu returned %zu", wlen, nwritten);
|
||||
return CURLE_WRITE_ERROR;
|
||||
}
|
||||
*pconsumed += nwritten;
|
||||
blen -= nwritten;
|
||||
buf += nwritten;
|
||||
}
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
static CURLcode cw_out_buf_flush(struct cw_out_ctx *ctx,
|
||||
struct Curl_easy *data,
|
||||
struct cw_out_buf *cwbuf,
|
||||
bool flush_all)
|
||||
{
|
||||
CURLcode result = CURLE_OK;
|
||||
|
||||
if(Curl_dyn_len(&cwbuf->b)) {
|
||||
size_t consumed;
|
||||
|
||||
result = cw_out_ptr_flush(ctx, data, cwbuf->type, flush_all,
|
||||
Curl_dyn_ptr(&cwbuf->b),
|
||||
Curl_dyn_len(&cwbuf->b),
|
||||
&consumed);
|
||||
if(result)
|
||||
return result;
|
||||
|
||||
if(consumed) {
|
||||
if(consumed == Curl_dyn_len(&cwbuf->b)) {
|
||||
Curl_dyn_free(&cwbuf->b);
|
||||
}
|
||||
else {
|
||||
DEBUGASSERT(consumed < Curl_dyn_len(&cwbuf->b));
|
||||
result = Curl_dyn_tail(&cwbuf->b, Curl_dyn_len(&cwbuf->b) - consumed);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
static CURLcode cw_out_flush_chain(struct cw_out_ctx *ctx,
|
||||
struct Curl_easy *data,
|
||||
struct cw_out_buf **pcwbuf,
|
||||
bool flush_all)
|
||||
{
|
||||
struct cw_out_buf *cwbuf = *pcwbuf;
|
||||
CURLcode result;
|
||||
|
||||
if(!cwbuf)
|
||||
return CURLE_OK;
|
||||
if(data->req.keepon & KEEP_RECV_PAUSE)
|
||||
return CURLE_OK;
|
||||
|
||||
/* write the end of the chain until it blocks or gets empty */
|
||||
while(cwbuf->next) {
|
||||
struct cw_out_buf **plast = &cwbuf->next;
|
||||
while((*plast)->next)
|
||||
plast = &(*plast)->next;
|
||||
result = cw_out_flush_chain(ctx, data, plast, flush_all);
|
||||
if(result)
|
||||
return result;
|
||||
if(*plast) {
|
||||
/* could not write last, paused again? */
|
||||
DEBUGASSERT(data->req.keepon & KEEP_RECV_PAUSE);
|
||||
return CURLE_OK;
|
||||
}
|
||||
}
|
||||
|
||||
result = cw_out_buf_flush(ctx, data, cwbuf, flush_all);
|
||||
if(result)
|
||||
return result;
|
||||
if(!Curl_dyn_len(&cwbuf->b)) {
|
||||
cw_out_buf_free(cwbuf);
|
||||
*pcwbuf = NULL;
|
||||
}
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
static CURLcode cw_out_append(struct cw_out_ctx *ctx,
|
||||
cw_out_type otype,
|
||||
const char *buf, size_t blen)
|
||||
{
|
||||
if(cw_out_bufs_len(ctx) + blen > DYN_PAUSE_BUFFER)
|
||||
return CURLE_TOO_LARGE;
|
||||
|
||||
/* if we do not have a buffer, or it is of another type, make a new one.
|
||||
* And for CW_OUT_HDS always make a new one, so we "replay" headers
|
||||
* exactly as they came in */
|
||||
if(!ctx->buf || (ctx->buf->type != otype) || (otype == CW_OUT_HDS)) {
|
||||
struct cw_out_buf *cwbuf = cw_out_buf_create(otype);
|
||||
if(!cwbuf)
|
||||
return CURLE_OUT_OF_MEMORY;
|
||||
cwbuf->next = ctx->buf;
|
||||
ctx->buf = cwbuf;
|
||||
}
|
||||
DEBUGASSERT(ctx->buf && (ctx->buf->type == otype));
|
||||
return Curl_dyn_addn(&ctx->buf->b, buf, blen);
|
||||
}
|
||||
|
||||
static CURLcode cw_out_do_write(struct cw_out_ctx *ctx,
|
||||
struct Curl_easy *data,
|
||||
cw_out_type otype,
|
||||
bool flush_all,
|
||||
const char *buf, size_t blen)
|
||||
{
|
||||
CURLcode result;
|
||||
|
||||
/* if we have buffered data and it is a different type than what
|
||||
* we are writing now, try to flush all */
|
||||
if(ctx->buf && ctx->buf->type != otype) {
|
||||
result = cw_out_flush_chain(ctx, data, &ctx->buf, TRUE);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
|
||||
if(ctx->buf) {
|
||||
/* still have buffered data, append and flush */
|
||||
result = cw_out_append(ctx, otype, buf, blen);
|
||||
if(result)
|
||||
return result;
|
||||
result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
else {
|
||||
/* nothing buffered, try direct write */
|
||||
size_t consumed;
|
||||
result = cw_out_ptr_flush(ctx, data, otype, flush_all,
|
||||
buf, blen, &consumed);
|
||||
if(result)
|
||||
return result;
|
||||
if(consumed < blen) {
|
||||
/* did not write all, append the rest */
|
||||
result = cw_out_append(ctx, otype, buf + consumed, blen - consumed);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
}
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
static CURLcode cw_out_write(struct Curl_easy *data,
|
||||
struct Curl_cwriter *writer, int type,
|
||||
const char *buf, size_t blen)
|
||||
{
|
||||
struct cw_out_ctx *ctx = (struct cw_out_ctx *)writer;
|
||||
CURLcode result;
|
||||
bool flush_all;
|
||||
|
||||
flush_all = (type & CLIENTWRITE_EOS)? TRUE:FALSE;
|
||||
if((type & CLIENTWRITE_BODY) ||
|
||||
((type & CLIENTWRITE_HEADER) && data->set.include_header)) {
|
||||
result = cw_out_do_write(ctx, data, CW_OUT_BODY, flush_all, buf, blen);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
|
||||
if(type & (CLIENTWRITE_HEADER|CLIENTWRITE_INFO)) {
|
||||
result = cw_out_do_write(ctx, data, CW_OUT_HDS, flush_all, buf, blen);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
bool Curl_cw_out_is_paused(struct Curl_easy *data)
|
||||
{
|
||||
struct Curl_cwriter *cw_out;
|
||||
struct cw_out_ctx *ctx;
|
||||
|
||||
cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out);
|
||||
if(!cw_out)
|
||||
return FALSE;
|
||||
|
||||
ctx = (struct cw_out_ctx *)cw_out;
|
||||
return cw_out_bufs_len(ctx) > 0;
|
||||
}
|
||||
|
||||
static CURLcode cw_out_flush(struct Curl_easy *data, bool flush_all)
|
||||
{
|
||||
struct Curl_cwriter *cw_out;
|
||||
CURLcode result = CURLE_OK;
|
||||
|
||||
cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out);
|
||||
if(cw_out) {
|
||||
struct cw_out_ctx *ctx = (struct cw_out_ctx *)cw_out;
|
||||
|
||||
result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
CURLcode Curl_cw_out_flush(struct Curl_easy *data)
|
||||
{
|
||||
return cw_out_flush(data, FALSE);
|
||||
}
|
||||
|
||||
CURLcode Curl_cw_out_done(struct Curl_easy *data)
|
||||
{
|
||||
return cw_out_flush(data, TRUE);
|
||||
}
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
#ifndef HEADER_CURL_CW_OUT_H
|
||||
#define HEADER_CURL_CW_OUT_H
|
||||
/***************************************************************************
|
||||
* _ _ ____ _
|
||||
* Project ___| | | | _ \| |
|
||||
* / __| | | | |_) | |
|
||||
* | (__| |_| | _ <| |___
|
||||
* \___|\___/|_| \_\_____|
|
||||
*
|
||||
* Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
|
||||
*
|
||||
* This software is licensed as described in the file COPYING, which
|
||||
* you should have received as part of this distribution. The terms
|
||||
* are also available at https://curl.se/docs/copyright.html.
|
||||
*
|
||||
* You may opt to use, copy, modify, merge, publish, distribute and/or sell
|
||||
* copies of the Software, and permit persons to whom the Software is
|
||||
* furnished to do so, under the terms of the COPYING file.
|
||||
*
|
||||
* This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
|
||||
* KIND, either express or implied.
|
||||
*
|
||||
* SPDX-License-Identifier: curl
|
||||
*
|
||||
***************************************************************************/
|
||||
|
||||
#include "curl_setup.h"
|
||||
|
||||
#include "sendf.h"
|
||||
|
||||
/**
|
||||
* The client writer type "cw-out" that does the actual writing to
|
||||
* the client callbacks. Intended to be the last installed in the
|
||||
* client writer stack of a transfer.
|
||||
*/
|
||||
extern struct Curl_cwtype Curl_cwt_out;
|
||||
|
||||
/**
|
||||
* Return TRUE iff 'cw-out' client write has paused data.
|
||||
*/
|
||||
bool Curl_cw_out_is_paused(struct Curl_easy *data);
|
||||
|
||||
/**
|
||||
* Flush any buffered date to the client, chunk collation still applies.
|
||||
*/
|
||||
CURLcode Curl_cw_out_flush(struct Curl_easy *data);
|
||||
|
||||
/**
|
||||
* Mark EndOfStream reached and flush ALL data to the client.
|
||||
*/
|
||||
CURLcode Curl_cw_out_done(struct Curl_easy *data);
|
||||
|
||||
#endif /* HEADER_CURL_CW_OUT_H */
|
||||
|
|
@ -58,6 +58,7 @@
|
|||
#include "multiif.h"
|
||||
#include "select.h"
|
||||
#include "cfilters.h"
|
||||
#include "cw-out.h"
|
||||
#include "sendf.h" /* for failf function prototype */
|
||||
#include "connect.h" /* for Curl_getconnectinfo */
|
||||
#include "slist.h"
|
||||
|
|
@ -1117,7 +1118,7 @@ CURLcode curl_easy_pause(struct Curl_easy *data, int action)
|
|||
|
||||
if(!(newstate & KEEP_RECV_PAUSE)) {
|
||||
Curl_conn_ev_data_pause(data, FALSE);
|
||||
result = Curl_client_unpause(data);
|
||||
result = Curl_cw_out_flush(data);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
|
|
@ -1141,7 +1142,7 @@ CURLcode curl_easy_pause(struct Curl_easy *data, int action)
|
|||
/* reset the too-slow time keeper */
|
||||
data->state.keeps_speed.tv_sec = 0;
|
||||
|
||||
if(!data->state.tempcount)
|
||||
if(!Curl_cw_out_is_paused(data))
|
||||
/* if not pausing again, force a recv/send check of this connection as
|
||||
the data might've been read off the socket already */
|
||||
data->state.select_bits = CURL_CSELECT_IN | CURL_CSELECT_OUT;
|
||||
|
|
|
|||
|
|
@ -645,7 +645,7 @@ static CURLcode multi_done(struct Curl_easy *data,
|
|||
after an error was detected */
|
||||
bool premature)
|
||||
{
|
||||
CURLcode result;
|
||||
CURLcode result, r2;
|
||||
struct connectdata *conn = data->conn;
|
||||
|
||||
#if defined(DEBUGBUILD) && !defined(CURL_DISABLE_VERBOSE_STRINGS)
|
||||
|
|
@ -696,6 +696,11 @@ static CURLcode multi_done(struct Curl_easy *data,
|
|||
result = CURLE_ABORTED_BY_CALLBACK;
|
||||
}
|
||||
|
||||
/* Make sure that transfer client writes are really done now. */
|
||||
r2 = Curl_xfer_write_done(data, premature);
|
||||
if(r2 && !result)
|
||||
result = r2;
|
||||
|
||||
/* Inform connection filters that this transfer is done */
|
||||
Curl_conn_ev_data_done(data, premature);
|
||||
|
||||
|
|
|
|||
220
lib/sendf.c
220
lib/sendf.c
|
|
@ -41,6 +41,7 @@
|
|||
#include "cfilters.h"
|
||||
#include "connect.h"
|
||||
#include "content_encoding.h"
|
||||
#include "cw-out.h"
|
||||
#include "vtls/vtls.h"
|
||||
#include "vssh/ssh.h"
|
||||
#include "easyif.h"
|
||||
|
|
@ -133,147 +134,6 @@ CURLcode Curl_write(struct Curl_easy *data,
|
|||
return Curl_nwrite(data, num, mem, len, written);
|
||||
}
|
||||
|
||||
static CURLcode pausewrite(struct Curl_easy *data,
|
||||
int type, /* what type of data */
|
||||
bool paused_body,
|
||||
const char *ptr,
|
||||
size_t len)
|
||||
{
|
||||
/* signalled to pause sending on this connection, but since we have data
|
||||
we want to send we need to dup it to save a copy for when the sending
|
||||
is again enabled */
|
||||
struct SingleRequest *k = &data->req;
|
||||
struct UrlState *s = &data->state;
|
||||
unsigned int i;
|
||||
bool newtype = TRUE;
|
||||
|
||||
Curl_conn_ev_data_pause(data, TRUE);
|
||||
|
||||
if(s->tempcount) {
|
||||
for(i = 0; i< s->tempcount; i++) {
|
||||
if(s->tempwrite[i].type == type &&
|
||||
!!s->tempwrite[i].paused_body == !!paused_body) {
|
||||
/* data for this type exists */
|
||||
newtype = FALSE;
|
||||
break;
|
||||
}
|
||||
}
|
||||
DEBUGASSERT(i < 3);
|
||||
if(i >= 3)
|
||||
/* There are more types to store than what fits: very bad */
|
||||
return CURLE_OUT_OF_MEMORY;
|
||||
}
|
||||
else
|
||||
i = 0;
|
||||
|
||||
if(newtype) {
|
||||
/* store this information in the state struct for later use */
|
||||
Curl_dyn_init(&s->tempwrite[i].b, DYN_PAUSE_BUFFER);
|
||||
s->tempwrite[i].type = type;
|
||||
s->tempwrite[i].paused_body = paused_body;
|
||||
s->tempcount++;
|
||||
}
|
||||
|
||||
if(Curl_dyn_addn(&s->tempwrite[i].b, (unsigned char *)ptr, len))
|
||||
return CURLE_OUT_OF_MEMORY;
|
||||
|
||||
/* mark the connection as RECV paused */
|
||||
k->keepon |= KEEP_RECV_PAUSE;
|
||||
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
|
||||
/* chop_write() writes chunks of data not larger than CURL_MAX_WRITE_SIZE via
|
||||
* client write callback(s) and takes care of pause requests from the
|
||||
* callbacks.
|
||||
*/
|
||||
static CURLcode chop_write(struct Curl_easy *data,
|
||||
int type,
|
||||
bool skip_body_write,
|
||||
char *optr,
|
||||
size_t olen)
|
||||
{
|
||||
struct connectdata *conn = data->conn;
|
||||
curl_write_callback writeheader = NULL;
|
||||
curl_write_callback writebody = NULL;
|
||||
char *ptr = optr;
|
||||
size_t len = olen;
|
||||
void *writebody_ptr = data->set.out;
|
||||
|
||||
if(!len)
|
||||
return CURLE_OK;
|
||||
|
||||
/* If reading is paused, append this data to the already held data for this
|
||||
type. */
|
||||
if(data->req.keepon & KEEP_RECV_PAUSE)
|
||||
return pausewrite(data, type, !skip_body_write, ptr, len);
|
||||
|
||||
/* Determine the callback(s) to use. */
|
||||
if(!skip_body_write &&
|
||||
((type & CLIENTWRITE_BODY) ||
|
||||
((type & CLIENTWRITE_HEADER) && data->set.include_header))) {
|
||||
writebody = data->set.fwrite_func;
|
||||
}
|
||||
if((type & (CLIENTWRITE_HEADER|CLIENTWRITE_INFO)) &&
|
||||
(data->set.fwrite_header || data->set.writeheader)) {
|
||||
/*
|
||||
* Write headers to the same callback or to the especially setup
|
||||
* header callback function (added after version 7.7.1).
|
||||
*/
|
||||
writeheader =
|
||||
data->set.fwrite_header? data->set.fwrite_header: data->set.fwrite_func;
|
||||
}
|
||||
|
||||
/* Chop data, write chunks. */
|
||||
while(len) {
|
||||
size_t chunklen = len <= CURL_MAX_WRITE_SIZE? len: CURL_MAX_WRITE_SIZE;
|
||||
|
||||
if(writebody) {
|
||||
size_t wrote;
|
||||
Curl_set_in_callback(data, true);
|
||||
wrote = writebody(ptr, 1, chunklen, writebody_ptr);
|
||||
Curl_set_in_callback(data, false);
|
||||
|
||||
if(CURL_WRITEFUNC_PAUSE == wrote) {
|
||||
if(conn->handler->flags & PROTOPT_NONETWORK) {
|
||||
/* Protocols that work without network cannot be paused. This is
|
||||
actually only FILE:// just now, and it can't pause since the
|
||||
transfer isn't done using the "normal" procedure. */
|
||||
failf(data, "Write callback asked for PAUSE when not supported");
|
||||
return CURLE_WRITE_ERROR;
|
||||
}
|
||||
return pausewrite(data, type, TRUE, ptr, len);
|
||||
}
|
||||
if(wrote != chunklen) {
|
||||
failf(data, "Failure writing output to destination");
|
||||
return CURLE_WRITE_ERROR;
|
||||
}
|
||||
}
|
||||
|
||||
ptr += chunklen;
|
||||
len -= chunklen;
|
||||
}
|
||||
|
||||
if(writeheader) {
|
||||
size_t wrote;
|
||||
|
||||
Curl_set_in_callback(data, true);
|
||||
wrote = writeheader(optr, 1, olen, data->set.writeheader);
|
||||
Curl_set_in_callback(data, false);
|
||||
|
||||
if(CURL_WRITEFUNC_PAUSE == wrote)
|
||||
return pausewrite(data, type, FALSE, optr, olen);
|
||||
if(wrote != olen) {
|
||||
failf(data, "Failed writing header");
|
||||
return CURLE_WRITE_ERROR;
|
||||
}
|
||||
}
|
||||
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
|
||||
/* Curl_client_write() sends data to the write callback(s)
|
||||
|
||||
The bit pattern defines to what "streams" to write to. Body and/or header.
|
||||
|
|
@ -303,42 +163,9 @@ CURLcode Curl_client_write(struct Curl_easy *data,
|
|||
return Curl_cwriter_write(data, data->req.writer_stack, type, buf, blen);
|
||||
}
|
||||
|
||||
CURLcode Curl_client_unpause(struct Curl_easy *data)
|
||||
{
|
||||
CURLcode result = CURLE_OK;
|
||||
|
||||
if(data->state.tempcount) {
|
||||
/* there are buffers for sending that can be delivered as the receive
|
||||
pausing is lifted! */
|
||||
unsigned int i;
|
||||
unsigned int count = data->state.tempcount;
|
||||
struct tempbuf writebuf[3]; /* there can only be three */
|
||||
|
||||
/* copy the structs to allow for immediate re-pausing */
|
||||
for(i = 0; i < data->state.tempcount; i++) {
|
||||
writebuf[i] = data->state.tempwrite[i];
|
||||
Curl_dyn_init(&data->state.tempwrite[i].b, DYN_PAUSE_BUFFER);
|
||||
}
|
||||
data->state.tempcount = 0;
|
||||
|
||||
for(i = 0; i < count; i++) {
|
||||
/* even if one function returns error, this loops through and frees
|
||||
all buffers */
|
||||
if(!result)
|
||||
result = chop_write(data, writebuf[i].type,
|
||||
!writebuf[i].paused_body,
|
||||
Curl_dyn_ptr(&writebuf[i].b),
|
||||
Curl_dyn_len(&writebuf[i].b));
|
||||
Curl_dyn_free(&writebuf[i].b);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
void Curl_client_cleanup(struct Curl_easy *data)
|
||||
{
|
||||
struct Curl_cwriter *writer = data->req.writer_stack;
|
||||
size_t i;
|
||||
|
||||
while(writer) {
|
||||
data->req.writer_stack = writer->next;
|
||||
|
|
@ -347,10 +174,6 @@ void Curl_client_cleanup(struct Curl_easy *data)
|
|||
writer = data->req.writer_stack;
|
||||
}
|
||||
|
||||
for(i = 0; i < data->state.tempcount; i++) {
|
||||
Curl_dyn_free(&data->state.tempwrite[i].b);
|
||||
}
|
||||
data->state.tempcount = 0;
|
||||
data->req.bytecount = 0;
|
||||
data->req.headerline = 0;
|
||||
}
|
||||
|
|
@ -388,26 +211,6 @@ void Curl_cwriter_def_close(struct Curl_easy *data,
|
|||
(void) writer;
|
||||
}
|
||||
|
||||
/* Real client writer to installed callbacks. */
|
||||
static CURLcode cw_client_write(struct Curl_easy *data,
|
||||
struct Curl_cwriter *writer, int type,
|
||||
const char *buf, size_t nbytes)
|
||||
{
|
||||
(void)writer;
|
||||
if(!nbytes)
|
||||
return CURLE_OK;
|
||||
return chop_write(data, type, FALSE, (char *)buf, nbytes);
|
||||
}
|
||||
|
||||
static const struct Curl_cwtype cw_client = {
|
||||
"client",
|
||||
NULL,
|
||||
Curl_cwriter_def_init,
|
||||
cw_client_write,
|
||||
Curl_cwriter_def_close,
|
||||
sizeof(struct Curl_cwriter)
|
||||
};
|
||||
|
||||
static size_t get_max_body_write_len(struct Curl_easy *data, curl_off_t limit)
|
||||
{
|
||||
if(limit != -1) {
|
||||
|
|
@ -496,14 +299,14 @@ static CURLcode cw_download_write(struct Curl_easy *data,
|
|||
}
|
||||
}
|
||||
|
||||
/* Update stats, write and report progress */
|
||||
data->req.bytecount += nwrite;
|
||||
++data->req.bodywrites;
|
||||
if(!data->req.ignorebody && nwrite) {
|
||||
if(!data->req.ignorebody && (nwrite || (type & CLIENTWRITE_EOS))) {
|
||||
result = Curl_cwriter_write(data, writer->next, type, buf, nwrite);
|
||||
if(result)
|
||||
return result;
|
||||
}
|
||||
/* Update stats, write and report progress */
|
||||
data->req.bytecount += nwrite;
|
||||
++data->req.bodywrites;
|
||||
result = Curl_pgrsSetDownloadCounter(data, data->req.bytecount);
|
||||
if(result)
|
||||
return result;
|
||||
|
|
@ -615,7 +418,7 @@ static CURLcode do_init_stack(struct Curl_easy *data)
|
|||
|
||||
DEBUGASSERT(!data->req.writer_stack);
|
||||
result = Curl_cwriter_create(&data->req.writer_stack,
|
||||
data, &cw_client, CURL_CW_CLIENT);
|
||||
data, &Curl_cwt_out, CURL_CW_CLIENT);
|
||||
if(result)
|
||||
return result;
|
||||
|
||||
|
|
@ -669,6 +472,17 @@ struct Curl_cwriter *Curl_cwriter_get_by_name(struct Curl_easy *data,
|
|||
return NULL;
|
||||
}
|
||||
|
||||
struct Curl_cwriter *Curl_cwriter_get_by_type(struct Curl_easy *data,
|
||||
const struct Curl_cwtype *cwt)
|
||||
{
|
||||
struct Curl_cwriter *writer;
|
||||
for(writer = data->req.writer_stack; writer; writer = writer->next) {
|
||||
if(writer->cwt == cwt)
|
||||
return writer;
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
void Curl_cwriter_remove_by_name(struct Curl_easy *data,
|
||||
const char *name)
|
||||
{
|
||||
|
|
|
|||
14
lib/sendf.h
14
lib/sendf.h
|
|
@ -58,13 +58,6 @@
|
|||
CURLcode Curl_client_write(struct Curl_easy *data, int type, const char *ptr,
|
||||
size_t len) WARN_UNUSED_RESULT;
|
||||
|
||||
/**
|
||||
* For a paused transfer, there might be buffered data held back.
|
||||
* Attempt to flush this data to the client. This *may* trigger
|
||||
* another pause of the transfer.
|
||||
*/
|
||||
CURLcode Curl_client_unpause(struct Curl_easy *data);
|
||||
|
||||
/**
|
||||
* Free all resources related to client writing.
|
||||
*/
|
||||
|
|
@ -148,6 +141,13 @@ size_t Curl_cwriter_count(struct Curl_easy *data, Curl_cwriter_phase phase);
|
|||
CURLcode Curl_cwriter_add(struct Curl_easy *data,
|
||||
struct Curl_cwriter *writer);
|
||||
|
||||
/**
|
||||
* Look up an installed client writer on `data` by its type.
|
||||
* @return first writer with that type or NULL
|
||||
*/
|
||||
struct Curl_cwriter *Curl_cwriter_get_by_type(struct Curl_easy *data,
|
||||
const struct Curl_cwtype *cwt);
|
||||
|
||||
void Curl_cwriter_remove_by_name(struct Curl_easy *data,
|
||||
const char *name);
|
||||
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@
|
|||
#include "content_encoding.h"
|
||||
#include "hostip.h"
|
||||
#include "cfilters.h"
|
||||
#include "cw-out.h"
|
||||
#include "transfer.h"
|
||||
#include "sendf.h"
|
||||
#include "speedcheck.h"
|
||||
|
|
@ -1720,3 +1721,9 @@ CURLcode Curl_xfer_write_resp(struct Curl_easy *data,
|
|||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
CURLcode Curl_xfer_write_done(struct Curl_easy *data, bool premature)
|
||||
{
|
||||
(void)premature;
|
||||
return Curl_cw_out_done(data);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -85,4 +85,10 @@ Curl_setup_transfer (struct Curl_easy *data,
|
|||
disables */
|
||||
);
|
||||
|
||||
/**
|
||||
* Multi has set transfer to DONE. Last chance to trigger
|
||||
* missing response things like writing an EOS to the client.
|
||||
*/
|
||||
CURLcode Curl_xfer_write_done(struct Curl_easy *data, bool premature);
|
||||
|
||||
#endif /* HEADER_CURL_TRANSFER_H */
|
||||
|
|
|
|||
|
|
@ -1273,18 +1273,6 @@ struct Curl_data_priority {
|
|||
#endif
|
||||
};
|
||||
|
||||
/*
|
||||
* This struct is for holding data that was attempted to get sent to the user's
|
||||
* callback but is held due to pausing. One instance per type (BOTH, HEADER,
|
||||
* BODY).
|
||||
*/
|
||||
struct tempbuf {
|
||||
struct dynbuf b;
|
||||
int type; /* type of the 'tempwrite' buffer as a bitmask that is used with
|
||||
Curl_client_write() */
|
||||
BIT(paused_body); /* if PAUSE happened before/during BODY write */
|
||||
};
|
||||
|
||||
/* Timers */
|
||||
typedef enum {
|
||||
EXPIRE_100_TIMEOUT,
|
||||
|
|
@ -1362,8 +1350,6 @@ struct UrlState {
|
|||
int retrycount; /* number of retries on a new connection */
|
||||
struct Curl_ssl_session *session; /* array of 'max_ssl_sessions' size */
|
||||
long sessionage; /* number of the most recent session */
|
||||
struct tempbuf tempwrite[3]; /* BOTH, HEADER, BODY */
|
||||
unsigned int tempcount; /* number of entries in use in tempwrite, 0 - 3 */
|
||||
int os_errno; /* filled in with errno whenever an error occurs */
|
||||
char *scratch; /* huge buffer[set.buffer_size*2] for upload CRLF replacing */
|
||||
long followlocation; /* redirect counter */
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче