You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@subversion.apache.org by rh...@apache.org on 2014/01/06 14:59:45 UTC
svn commit: r1555774 - in /subversion/trunk/subversion/libsvn_ra_serf:
eagain_bucket.c ra_serf.h update.c util.c
Author: rhuijben
Date: Mon Jan 6 13:59:45 2014
New Revision: 1555774
URL: http://svn.apache.org/r1555774
Log:
In libsvn_ra_serf, replace the pause handling for the update response that was
embedded through the old -not transition based- xml parser with a pause
handling response handler wrapper directly in the update logic.
* subversion/libsvn_ra_serf/eagain_bucket.c
New file.
* subversion/libsvn_ra_serf/ra_serf.h
(svn_ra_serf__create_bucket_with_eagain): New function.
* subversion/libsvn_ra_serf/update.c
(SPILLBUF_BLOCKSIZE,
SPILLBUF_MAXBUFFSIZE,
PARSE_CHUNK_SIZE): New defines.
(fetch_file): Remove old pause handling.
(end_report): Remove old pause handling.
(update_delay_baton_t): New struct.
(update_delay_handler): New function.
(finish_report): Wrap xml handler with delay handling. Process remaining
data if any.
* subversion/libsvn_ra_serf/util.c
(inject_to_parser): Remove unused argument.
(svn_ra_serf__process_pending): Update caller.
(svn_ra_serf__handle_xml_parser): Don't fetch status line. Is unused and
requires the response to be a response bucket, which is not always the
case any more.
Added:
subversion/trunk/subversion/libsvn_ra_serf/eagain_bucket.c (with props)
Modified:
subversion/trunk/subversion/libsvn_ra_serf/ra_serf.h
subversion/trunk/subversion/libsvn_ra_serf/update.c
subversion/trunk/subversion/libsvn_ra_serf/util.c
Added: subversion/trunk/subversion/libsvn_ra_serf/eagain_bucket.c
URL: http://svn.apache.org/viewvc/subversion/trunk/subversion/libsvn_ra_serf/eagain_bucket.c?rev=1555774&view=auto
==============================================================================
--- subversion/trunk/subversion/libsvn_ra_serf/eagain_bucket.c (added)
+++ subversion/trunk/subversion/libsvn_ra_serf/eagain_bucket.c Mon Jan 6 13:59:45 2014
@@ -0,0 +1,122 @@
+/*
+ * delay_bucket.c : a serf bucket that handles slowing down data
+ * for specific readers that would have unwanted
+ * behavior if they read everything to fast
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ */
+
+#include <serf.h>
+#include <serf_bucket_util.h>
+
+#include "svn_private_config.h"
+
+#include "ra_serf.h"
+
+typedef struct eagain_baton_t
+{
+ const char *data;
+ apr_size_t remaining;
+} eagain_baton_t;
+
+static apr_status_t
+eagain_bucket_read(serf_bucket_t *bucket,
+ apr_size_t requested,
+ const char **data,
+ apr_size_t *len)
+{
+ eagain_baton_t *eab = bucket->data;
+
+ if (eab->remaining > 0)
+ {
+ *data = eab->data;
+ if (requested > eab->remaining || requested == SERF_READ_ALL_AVAIL)
+ {
+ *len = eab->remaining;
+ eab->data = NULL;
+ eab->remaining = 0;
+ }
+ else
+ {
+ *len = requested;
+ eab->data += requested;
+ eab->remaining -= requested;
+ }
+
+ if (eab->remaining)
+ return APR_SUCCESS;
+ }
+
+ return APR_EAGAIN;
+}
+
+
+static apr_status_t
+eagain_bucket_readline(serf_bucket_t *bucket,
+ int acceptable,
+ int *found,
+ const char **data,
+ apr_size_t *len)
+{
+ /* ### for now, we know callers won't use this function. */
+ svn_error_clear(svn_error__malfunction(TRUE, __FILE__, __LINE__,
+ "Not implemented."));
+ return APR_ENOTIMPL;
+}
+
+
+static apr_status_t
+eagain_bucket_peek(serf_bucket_t *bucket,
+ const char **data,
+ apr_size_t *len)
+{
+ const eagain_baton_t *eab = bucket->data;
+
+ *data = eab->data;
+ *len = eab->remaining;
+
+ return *data == NULL ? APR_EOF : APR_SUCCESS;
+}
+
+
+static const serf_bucket_type_t delay_bucket_vtable = {
+ "BUF-EAGAIN",
+ eagain_bucket_read,
+ eagain_bucket_readline,
+ serf_default_read_iovec,
+ serf_default_read_for_sendfile,
+ serf_default_read_bucket,
+ eagain_bucket_peek,
+ serf_default_destroy_and_data,
+};
+
+
+serf_bucket_t *
+svn_ra_serf__create_bucket_with_eagain(const char *data,
+ apr_size_t len,
+ serf_bucket_alloc_t *allocator)
+{
+ eagain_baton_t *eab;
+
+ eab = serf_bucket_mem_alloc(allocator, sizeof(*eab));
+ eab->data = data;
+ eab->remaining = len;
+
+ return serf_bucket_create(&delay_bucket_vtable, allocator, eab);
+}
Propchange: subversion/trunk/subversion/libsvn_ra_serf/eagain_bucket.c
------------------------------------------------------------------------------
svn:eol-style = native
Modified: subversion/trunk/subversion/libsvn_ra_serf/ra_serf.h
URL: http://svn.apache.org/viewvc/subversion/trunk/subversion/libsvn_ra_serf/ra_serf.h?rev=1555774&r1=1555773&r2=1555774&view=diff
==============================================================================
--- subversion/trunk/subversion/libsvn_ra_serf/ra_serf.h (original)
+++ subversion/trunk/subversion/libsvn_ra_serf/ra_serf.h Mon Jan 6 13:59:45 2014
@@ -1863,6 +1863,14 @@ svn_ra_serf__wrap_err(apr_status_t statu
const char *fmt,
...);
+/* Create a bucket that just returns DATA (with length LEN) and then returns
+ the APR_EAGAIN status */
+serf_bucket_t *
+svn_ra_serf__create_bucket_with_eagain(const char *data,
+ apr_size_t len,
+ serf_bucket_alloc_t *allocator);
+
+
#if defined(SVN_DEBUG)
/* Wrapper macros to collect file and line information */
Modified: subversion/trunk/subversion/libsvn_ra_serf/update.c
URL: http://svn.apache.org/viewvc/subversion/trunk/subversion/libsvn_ra_serf/update.c?rev=1555774&r1=1555773&r2=1555774&view=diff
==============================================================================
--- subversion/trunk/subversion/libsvn_ra_serf/update.c (original)
+++ subversion/trunk/subversion/libsvn_ra_serf/update.c Mon Jan 6 13:59:45 2014
@@ -53,6 +53,11 @@
#define USE_TRANSITION_PARSER
*/
+#define SPILLBUF_BLOCKSIZE 4096
+#define SPILLBUF_MAXBUFFSIZE 131072
+
+#define PARSE_CHUNK_SIZE 8000 /* Copied from utils.c ### Needs tuning */
+
/*
* This enum represents the current state of our XML parsing for a REPORT.
@@ -1599,10 +1604,6 @@ fetch_file(report_context_t *ctx, report
SVN_ERR(handle_propchange_only(info, info->pool));
}
- if (ctx->num_active_fetches + ctx->num_active_propfinds
- > REQUEST_COUNT_TO_PAUSE)
- ctx->parser_ctx->paused = TRUE;
-
return SVN_NO_ERROR;
}
@@ -2209,10 +2210,6 @@ end_report(svn_ra_serf__xml_parser_t *pa
list_item->data = info->dir;
list_item->next = ctx->active_dir_propfinds;
ctx->active_dir_propfinds = list_item;
-
- if (ctx->num_active_fetches + ctx->num_active_propfinds
- > REQUEST_COUNT_TO_PAUSE)
- ctx->parser_ctx->paused = TRUE;
}
else
{
@@ -2835,6 +2832,189 @@ setup_update_report_headers(serf_bucket_
return SVN_NO_ERROR;
}
+/* Baton for update_delay_handler */
+typedef struct update_delay_baton_t
+{
+ report_context_t *report;
+ svn_spillbuf_t *spillbuf;
+ svn_ra_serf__response_handler_t inner_handler;
+ void *inner_handler_baton;
+
+ serf_bucket_t *retry_bucket;
+} update_delay_baton_t;
+
+/* Delaying wrapping reponse handler, to avoid creating too many
+ requests to deliver efficiently */
+static svn_error_t *
+update_delay_handler(serf_request_t *request,
+ serf_bucket_t *response,
+ void *handler_baton,
+ apr_pool_t *scratch_pool)
+{
+ update_delay_baton_t *udb = handler_baton;
+ apr_status_t status;
+ apr_pool_t *iterpool = NULL;
+
+ if (! udb->spillbuf)
+ {
+ if (udb->report->send_all_mode)
+ {
+ /* Easy out... We only have one request, so avoid everything and just
+ call the inner handler.
+
+ We will always get in the loop (below) on the first chunk, as only
+ the server can get us in true send-all mode */
+
+ return svn_error_trace(udb->inner_handler(request, response,
+ udb->inner_handler_baton,
+ scratch_pool));
+ }
+
+ while ((udb->report->num_active_fetches + udb->report->num_active_propfinds)
+ < REQUEST_COUNT_TO_RESUME)
+ {
+ const char *data;
+ apr_size_t len;
+ svn_boolean_t at_eof = FALSE;
+ svn_error_t *err;
+
+ status = serf_bucket_read(response, PARSE_CHUNK_SIZE, &data, &len);
+ if (SERF_BUCKET_READ_ERROR(status))
+ return svn_ra_serf__wrap_err(status, NULL);
+ else if (APR_STATUS_IS_EOF(status))
+ at_eof = TRUE;
+
+ if (!iterpool)
+ iterpool = svn_pool_create(scratch_pool);
+ else
+ svn_pool_clear(iterpool);
+
+ if (len == 0 && !at_eof)
+ return svn_ra_serf__wrap_err(status, NULL);
+
+ /* If not at EOF create a bucket that finishes with EAGAIN, otherwise
+ use a standard bucket with default EOF handling */
+ err = udb->inner_handler(request,
+ at_eof
+ ? serf_bucket_simple_create(
+ data, len, NULL, NULL,
+ serf_request_get_alloc(request))
+ : svn_ra_serf__create_bucket_with_eagain(
+ data, len,
+ serf_request_get_alloc(request)),
+ udb->inner_handler_baton, iterpool);
+
+ if (err && SERF_BUCKET_READ_ERROR(err->apr_err))
+ return svn_error_trace(err);
+ else if (err && APR_STATUS_IS_EAGAIN(err->apr_err))
+ {
+ svn_error_clear(err); /* Throttling is working ok */
+ }
+ else if (err && (APR_STATUS_IS_EOF(err->apr_err)))
+ {
+ svn_pool_destroy(iterpool);
+ return svn_error_trace(err); /* No buffering was necessary */
+ }
+ else
+ {
+ /* SERF_ERROR_WAIT_CONN should be impossible? */
+ SVN_ERR_MALFUNCTION();
+ }
+ }
+
+ /* Let's start using the spill infrastructure */
+ udb->spillbuf = svn_spillbuf__create(SPILLBUF_BLOCKSIZE,
+ SPILLBUF_MAXBUFFSIZE,
+ udb->report->pool);
+ }
+
+ /* Read everything we can to a spillbuffer */
+ do
+ {
+ const char *data;
+ apr_size_t len;
+
+ /* ### What blocksize should we pass? */
+ status = serf_bucket_read(response, 8*PARSE_CHUNK_SIZE, &data, &len);
+
+ if (!SERF_BUCKET_READ_ERROR(status))
+ SVN_ERR(svn_spillbuf__write(udb->spillbuf, data, len, scratch_pool));
+ }
+ while (status == APR_SUCCESS);
+
+ /* We handle feeding the data from the main context loop, which will be right
+ after processing the pending data */
+
+ if (status)
+ return svn_ra_serf__wrap_err(status, NULL);
+ else
+ return SVN_NO_ERROR;
+}
+
+/* Process pending data from the update report, if any */
+static svn_error_t *
+process_pending(update_delay_baton_t *udb,
+ apr_pool_t *scratch_pool)
+{
+ apr_pool_t *iterpool = NULL;
+ serf_bucket_alloc_t *alloc = NULL;
+
+ while ((udb->report->num_active_fetches + udb->report->num_active_propfinds)
+ < REQUEST_COUNT_TO_RESUME)
+ {
+ const char *data;
+ apr_size_t len;
+ svn_boolean_t at_eof = FALSE;
+ svn_error_t *err;
+
+ if (!iterpool)
+ {
+ iterpool = svn_pool_create(scratch_pool);
+ alloc = serf_bucket_allocator_create(iterpool, NULL, NULL);
+ }
+ else
+ svn_pool_clear(iterpool);
+
+ SVN_ERR(svn_spillbuf__read(&data, &len, udb->spillbuf, iterpool));
+
+ if (data == NULL)
+ {
+ data = "";
+ at_eof = TRUE;
+ }
+
+ /* If not at EOF create a bucket that finishes with EAGAIN, otherwise
+ use a standard bucket with default EOF handling */
+ err = udb->inner_handler(NULL /* allowed? */,
+ at_eof
+ ? serf_bucket_simple_create(
+ data, len, NULL, NULL, alloc)
+ : svn_ra_serf__create_bucket_with_eagain(
+ data, len, alloc),
+ udb->inner_handler_baton, iterpool);
+
+ if (err && APR_STATUS_IS_EAGAIN(err->apr_err))
+ {
+ svn_error_clear(err); /* Throttling is working */
+ }
+ else if (err && APR_STATUS_IS_EOF(err->apr_err))
+ {
+ svn_error_clear(err);
+
+ svn_pool_destroy(iterpool);
+ udb->spillbuf = NULL;
+ return SVN_NO_ERROR;
+ }
+ else if (err)
+ return svn_error_trace(err);
+ }
+
+ if (iterpool)
+ svn_pool_destroy(iterpool);
+
+ return SVN_NO_ERROR;
+}
+
static svn_error_t *
finish_report(void *report_baton,
apr_pool_t *pool)
@@ -2851,6 +3031,7 @@ finish_report(void *report_baton,
apr_pool_t *iterpool = svn_pool_create(pool);
svn_error_t *err;
apr_interval_time_t waittime_left = sess->timeout;
+ update_delay_baton_t *ud;
svn_xml_make_close_tag(&buf, iterpool, "S:update-report");
SVN_ERR(svn_spillbuf__write(report->body_sb, buf->data, buf->len, iterpool));
@@ -2899,6 +3080,16 @@ finish_report(void *report_baton,
report->parser_ctx = parser_ctx;
#endif
+ /* Now wrap the response handler with delay support to avoid sending
+ out too many requests at once */
+ ud = apr_pcalloc(pool, sizeof(*ud));
+ ud->report = report;
+ ud->inner_handler = handler->response_handler;
+ ud->inner_handler_baton = handler->response_baton;
+
+ handler->response_handler = update_delay_handler;
+ handler->response_baton = ud;
+
svn_ra_serf__request_create(handler);
/* Open the first extra connection. */
@@ -3102,20 +3293,9 @@ finish_report(void *report_baton,
}
report->done_dir_propfinds = NULL;
- /* If the parser is paused, and the number of active requests has
- dropped far enough, then resume parsing. */
- if (parser_ctx->paused
- && (report->num_active_fetches + report->num_active_propfinds
- < REQUEST_COUNT_TO_RESUME))
- parser_ctx->paused = FALSE;
-
- /* If we have not paused the parser and it looks like data MAY be
- present (we can't know for sure because of the private structure),
- then go process the pending content. */
- if (!parser_ctx->paused && parser_ctx->pending != NULL)
- SVN_ERR(svn_ra_serf__process_pending(parser_ctx,
- &report->report_received,
- iterpool));
+ /* If there is pending REPORT data, process it now. */
+ if (ud->spillbuf)
+ SVN_ERR(process_pending(ud, iterpool));
/* Debugging purposes only! */
for (i = 0; i < sess->num_conns; i++)
Modified: subversion/trunk/subversion/libsvn_ra_serf/util.c
URL: http://svn.apache.org/viewvc/subversion/trunk/subversion/libsvn_ra_serf/util.c?rev=1555774&r1=1555773&r2=1555774&view=diff
==============================================================================
--- subversion/trunk/subversion/libsvn_ra_serf/util.c (original)
+++ subversion/trunk/subversion/libsvn_ra_serf/util.c Mon Jan 6 13:59:45 2014
@@ -1296,8 +1296,7 @@ parse_xml(XML_Parser parser, const char
static svn_error_t *
inject_to_parser(svn_ra_serf__xml_parser_t *ctx,
const char *data,
- apr_size_t len,
- const serf_status_line *sl)
+ apr_size_t len)
{
svn_error_t *xml_err;
@@ -1361,7 +1360,7 @@ svn_ra_serf__process_pending(svn_ra_serf
if (data)
{
/* Inject the content into the XML parser. */
- SVN_ERR(inject_to_parser(parser, data, len, NULL));
+ SVN_ERR(inject_to_parser(parser, data, len));
/* If the XML parsing callbacks paused us, then we're done for now. */
if (parser->paused)
@@ -1411,18 +1410,10 @@ svn_ra_serf__handle_xml_parser(serf_requ
void *baton,
apr_pool_t *pool)
{
- serf_status_line sl;
apr_status_t status;
svn_ra_serf__xml_parser_t *ctx = baton;
svn_error_t *err;
- /* ### get the HANDLER rather than fetching this. */
- status = serf_bucket_response_status(response, &sl);
- if (SERF_BUCKET_READ_ERROR(status))
- {
- return svn_ra_serf__wrap_err(status, NULL);
- }
-
if (!ctx->xmlp)
{
ctx->xmlp = XML_ParserCreate(NULL);
@@ -1462,7 +1453,7 @@ svn_ra_serf__handle_xml_parser(serf_requ
}
else
{
- err = inject_to_parser(ctx, data, len, &sl);
+ err = inject_to_parser(ctx, data, len);
}
if (err)
{
Re: svn commit: r1555774 - in /subversion/trunk/subversion/libsvn_ra_serf:
eagain_bucket.c ra_serf.h update.c util.c
Posted by Ivan Zhakov <iv...@visualsvn.com>.
On 6 January 2014 17:59, <rh...@apache.org> wrote:
> Author: rhuijben
> Date: Mon Jan 6 13:59:45 2014
> New Revision: 1555774
>
> URL: http://svn.apache.org/r1555774
> Log:
> In libsvn_ra_serf, replace the pause handling for the update response that was
> embedded through the old -not transition based- xml parser with a pause
> handling response handler wrapper directly in the update logic.
>
Hi Bert, see my comments below.
> +
> +static apr_status_t
> +eagain_bucket_read(serf_bucket_t *bucket,
> + apr_size_t requested,
> + const char **data,
> + apr_size_t *len)
> +{
> + eagain_baton_t *eab = bucket->data;
> +
> + if (eab->remaining > 0)
> + {
> + *data = eab->data;
> + if (requested > eab->remaining || requested == SERF_READ_ALL_AVAIL)
> + {
> + *len = eab->remaining;
> + eab->data = NULL;
> + eab->remaining = 0;
> + }
> + else
> + {
> + *len = requested;
> + eab->data += requested;
> + eab->remaining -= requested;
> + }
> +
> + if (eab->remaining)
> + return APR_SUCCESS;
> + }
> +
> + return APR_EAGAIN;
> +}
> +
Alternative approach is reuse SIMPLE bucket at make eagain_bucket
wrapper around any inner bucket transforming APR_EOF to APR_EAGAIN.
[...]
> +
> +static apr_status_t
> +eagain_bucket_peek(serf_bucket_t *bucket,
> + const char **data,
> + apr_size_t *len)
> +{
> + const eagain_baton_t *eab = bucket->data;
> +
> + *data = eab->data;
> + *len = eab->remaining;
> +
> + return *data == NULL ? APR_EOF : APR_SUCCESS;
> +}
Are you sure about returning APR_EOF here?
[...]
>
> Modified: subversion/trunk/subversion/libsvn_ra_serf/update.c
> URL: http://svn.apache.org/viewvc/subversion/trunk/subversion/libsvn_ra_serf/update.c?rev=1555774&r1=1555773&r2=1555774&view=diff
> ==============================================================================
> --- subversion/trunk/subversion/libsvn_ra_serf/update.c (original)
> +++ subversion/trunk/subversion/libsvn_ra_serf/update.c Mon Jan 6 13:59:45 2014
[...]
> +/* Delaying wrapping reponse handler, to avoid creating too many
> + requests to deliver efficiently */
> +static svn_error_t *
> +update_delay_handler(serf_request_t *request,
> + serf_bucket_t *response,
> + void *handler_baton,
> + apr_pool_t *scratch_pool)
> +{
[...]
> + if (len == 0 && !at_eof)
> + return svn_ra_serf__wrap_err(status, NULL);
> +
> + /* If not at EOF create a bucket that finishes with EAGAIN, otherwise
> + use a standard bucket with default EOF handling */
> + err = udb->inner_handler(request,
> + at_eof
> + ? serf_bucket_simple_create(
> + data, len, NULL, NULL,
> + serf_request_get_alloc(request))
> + : svn_ra_serf__create_bucket_with_eagain(
> + data, len,
> + serf_request_get_alloc(request)),
With my suggestion above this code will be converted to something like:
[[[
bucket = serf_bucket_simple_create();
if (!at_eof)
bucket = svn_ra_serf__create_bucket_with_eagain(bucket);
err = udb->inner_handler(request, bucket, udb->inner_handler_baton, iterpool);
]]]
--
Ivan Zhakov
CTO | VisualSVN | http://www.visualsvn.com