You are viewing a plain text version of this content. The canonical link for it is here.
Posted to cvs@httpd.apache.org by ic...@apache.org on 2016/04/28 14:43:02 UTC
svn commit: r1741419 [3/4] - in /httpd/httpd/branches/2.4.x: ./
modules/http2/
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_stream.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_stream.c?rev=1741419&r1=1741418&r2=1741419&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_stream.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_stream.c Thu Apr 28 12:43:02 2016
@@ -24,6 +24,8 @@
#include <nghttp2/nghttp2.h>
#include "h2_private.h"
+#include "h2.h"
+#include "h2_bucket_beam.h"
#include "h2_conn.h"
#include "h2_config.h"
#include "h2_h2.h"
@@ -36,7 +38,6 @@
#include "h2_stream.h"
#include "h2_task.h"
#include "h2_ctx.h"
-#include "h2_task_input.h"
#include "h2_task.h"
#include "h2_util.h"
@@ -52,6 +53,13 @@ static int state_transition[][7] = {
/*CL*/{ 1, 1, 0, 0, 1, 1, 1 },
};
+#define H2_STREAM_OUT_LOG(lvl,s,msg) \
+ do { \
+ if (APLOG_C_IS_LEVEL((s)->session->c,lvl)) \
+ h2_util_bb_log((s)->session->c,(s)->session->id,lvl,msg,(s)->buffer); \
+ } while(0)
+
+
static int set_state(h2_stream *stream, h2_stream_state_t state)
{
int allowed = state_transition[state][stream->state];
@@ -135,36 +143,59 @@ static int output_open(h2_stream *stream
}
}
-static h2_sos *h2_sos_mplx_create(h2_stream *stream, h2_response *response);
-
-h2_stream *h2_stream_open(int id, apr_pool_t *pool, h2_session *session)
+h2_stream *h2_stream_open(int id, apr_pool_t *pool, h2_session *session,
+ int initiated_on, const h2_request *creq)
{
+ h2_request *req;
h2_stream *stream = apr_pcalloc(pool, sizeof(h2_stream));
+
stream->id = id;
stream->state = H2_STREAM_ST_IDLE;
stream->pool = pool;
stream->session = session;
set_state(stream, H2_STREAM_ST_OPEN);
- stream->request = h2_request_create(id, pool,
- h2_config_geti(session->config, H2_CONF_SER_HEADERS));
+
+ if (creq) {
+ /* take it into out pool and assure correct id's */
+ req = h2_request_clone(pool, creq);
+ req->id = id;
+ req->initiated_on = initiated_on;
+ }
+ else {
+ req = h2_request_create(id, pool,
+ h2_config_geti(session->config, H2_CONF_SER_HEADERS));
+ }
+ stream->request = req;
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03082)
"h2_stream(%ld-%d): opened", session->id, stream->id);
return stream;
}
-apr_status_t h2_stream_destroy(h2_stream *stream)
+void h2_stream_cleanup(h2_stream *stream)
+{
+ AP_DEBUG_ASSERT(stream);
+ if (stream->input) {
+ h2_beam_destroy(stream->input);
+ stream->input = NULL;
+ }
+ if (stream->buffer) {
+ apr_brigade_cleanup(stream->buffer);
+ }
+}
+
+void h2_stream_destroy(h2_stream *stream)
{
AP_DEBUG_ASSERT(stream);
+ h2_stream_cleanup(stream);
if (stream->pool) {
apr_pool_destroy(stream->pool);
}
- return APR_SUCCESS;
}
-void h2_stream_cleanup(h2_stream *stream)
+void h2_stream_eos_destroy(h2_stream *stream)
{
- h2_session_stream_destroy(stream->session, stream);
+ h2_session_stream_done(stream->session, stream);
/* stream is gone */
}
@@ -187,33 +218,7 @@ void h2_stream_rst(h2_stream *stream, in
struct h2_response *h2_stream_get_response(h2_stream *stream)
{
- return stream->sos? stream->sos->response : NULL;
-}
-
-apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
- apr_bucket_brigade *bb)
-{
- apr_status_t status = APR_SUCCESS;
- h2_sos *sos;
-
- if (!output_open(stream)) {
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
- "h2_stream(%ld-%d): output closed",
- stream->session->id, stream->id);
- return APR_ECONNRESET;
- }
-
- sos = h2_sos_mplx_create(stream, response);
- if (sos->response->sos_filter) {
- sos = h2_filter_sos_create(sos->response->sos_filter, sos);
- }
- stream->sos = sos;
-
- status = stream->sos->buffer(stream->sos, bb);
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
- "h2_stream(%ld-%d): set_response(%d)",
- stream->session->id, stream->id, stream->sos->response->http_status);
- return status;
+ return stream->response;
}
apr_status_t h2_stream_set_request(h2_stream *stream, request_rec *r)
@@ -231,14 +236,6 @@ apr_status_t h2_stream_set_request(h2_st
return status;
}
-void h2_stream_set_h2_request(h2_stream *stream, int initiated_on,
- const h2_request *req)
-{
- h2_request_copy(stream->pool, stream->request, req);
- stream->request->initiated_on = initiated_on;
- stream->request->eoh = 0;
-}
-
apr_status_t h2_stream_add_header(h2_stream *stream,
const char *name, size_t nlen,
const char *value, size_t vlen)
@@ -281,15 +278,11 @@ apr_status_t h2_stream_schedule(h2_strea
status = h2_request_end_headers(stream->request, stream->pool,
eos, push_enabled);
if (status == APR_SUCCESS) {
- if (!eos) {
- stream->request->body = 1;
- }
- stream->input_remaining = stream->request->content_length;
-
- status = h2_mplx_process(stream->session->mplx, stream->id,
- stream->request, cmp, ctx);
+ stream->request->body = !eos;
stream->scheduled = 1;
+ stream->input_remaining = stream->request->content_length;
+ status = h2_mplx_process(stream->session->mplx, stream, cmp, ctx);
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
"h2_stream(%ld-%d): scheduled %s %s://%s%s",
stream->session->id, stream->id,
@@ -298,7 +291,7 @@ apr_status_t h2_stream_schedule(h2_strea
}
else {
h2_stream_rst(stream, H2_ERR_INTERNAL_ERROR);
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
"h2_stream(%ld-%d): RST=2 (internal err) %s %s://%s%s",
stream->session->id, stream->id,
stream->request->method, stream->request->scheme,
@@ -326,8 +319,8 @@ apr_status_t h2_stream_close_input(h2_st
return APR_ECONNRESET;
}
- if (close_input(stream)) {
- status = h2_mplx_in_close(stream->session->mplx, stream->id);
+ if (close_input(stream) && stream->input) {
+ status = h2_beam_close(stream->input);
}
return status;
}
@@ -335,25 +328,29 @@ apr_status_t h2_stream_close_input(h2_st
apr_status_t h2_stream_write_data(h2_stream *stream,
const char *data, size_t len, int eos)
{
+ conn_rec *c = stream->session->c;
apr_status_t status = APR_SUCCESS;
AP_DEBUG_ASSERT(stream);
+ if (!stream->input) {
+ return APR_EOF;
+ }
if (input_closed(stream) || !stream->request->eoh) {
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
"h2_stream(%ld-%d): writing denied, closed=%d, eoh=%d",
stream->session->id, stream->id, input_closed(stream),
stream->request->eoh);
return APR_EINVAL;
}
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
"h2_stream(%ld-%d): add %ld input bytes",
stream->session->id, stream->id, (long)len);
if (!stream->request->chunked) {
stream->input_remaining -= len;
if (stream->input_remaining < 0) {
- ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c,
+ ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, c,
APLOGNO(02961)
"h2_stream(%ld-%d): got %ld more content bytes than announced "
"in content-length header: %ld",
@@ -365,10 +362,18 @@ apr_status_t h2_stream_write_data(h2_str
}
}
- status = h2_mplx_in_write(stream->session->mplx, stream->id, data, len, eos);
+ if (!stream->tmp) {
+ stream->tmp = apr_brigade_create(stream->pool, c->bucket_alloc);
+ }
+ apr_brigade_write(stream->tmp, NULL, NULL, data, len);
if (eos) {
+ APR_BRIGADE_INSERT_TAIL(stream->tmp,
+ apr_bucket_eos_create(c->bucket_alloc));
close_input(stream);
}
+
+ status = h2_beam_send(stream->input, stream->tmp, APR_BLOCK_READ);
+ apr_brigade_cleanup(stream->tmp);
return status;
}
@@ -387,44 +392,122 @@ int h2_stream_is_suspended(const h2_stre
return stream->suspended;
}
-apr_status_t h2_stream_out_prepare(h2_stream *stream,
+static apr_status_t fill_buffer(h2_stream *stream, apr_size_t amount)
+{
+ if (!stream->output) {
+ return APR_EOF;
+ }
+ return h2_beam_receive(stream->output, stream->buffer,
+ APR_NONBLOCK_READ, amount);
+}
+
+apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
+ h2_bucket_beam *output)
+{
+ apr_status_t status = APR_SUCCESS;
+ conn_rec *c = stream->session->c;
+
+ if (!output_open(stream)) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
+ "h2_stream(%ld-%d): output closed",
+ stream->session->id, stream->id);
+ return APR_ECONNRESET;
+ }
+
+ stream->response = response;
+ stream->output = output;
+ stream->buffer = apr_brigade_create(stream->pool, c->bucket_alloc);
+
+ h2_stream_filter(stream);
+ if (stream->output) {
+ status = fill_buffer(stream, 0);
+ }
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, c,
+ "h2_stream(%ld-%d): set_response(%d)",
+ stream->session->id, stream->id,
+ stream->response->http_status);
+ return status;
+}
+
+apr_status_t h2_stream_out_prepare(h2_stream *stream,
apr_off_t *plen, int *peos)
{
+ conn_rec *c = stream->session->c;
+ apr_status_t status = APR_SUCCESS;
+ apr_off_t requested = (*plen > 0)? *plen : 32*1024;
+
if (stream->rst_error) {
*plen = 0;
*peos = 1;
return APR_ECONNRESET;
}
- AP_DEBUG_ASSERT(stream->sos);
- return stream->sos->prepare(stream->sos, plen, peos);
+ H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "h2_stream_out_prepare_pre");
+ h2_util_bb_avail(stream->buffer, plen, peos);
+ if (!*peos && !*plen) {
+ /* try to get more data */
+ status = fill_buffer(stream, H2MIN(requested, 32*1024));
+ if (APR_STATUS_IS_EOF(status)) {
+ apr_bucket *eos = apr_bucket_eos_create(c->bucket_alloc);
+ APR_BRIGADE_INSERT_TAIL(stream->buffer, eos);
+ status = APR_SUCCESS;
+ }
+ h2_util_bb_avail(stream->buffer, plen, peos);
+ }
+ H2_STREAM_OUT_LOG(APLOG_TRACE2, stream, "h2_stream_out_prepare_post");
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, c,
+ "h2_stream(%ld-%d): prepare, len=%ld eos=%d, trailers=%s",
+ c->id, stream->id, (long)*plen, *peos,
+ (stream->response && stream->response->trailers)?
+ "yes" : "no");
+ if (!*peos && !*plen && status == APR_SUCCESS) {
+ return APR_EAGAIN;
+ }
+ return status;
}
+
apr_status_t h2_stream_readx(h2_stream *stream,
h2_io_data_cb *cb, void *ctx,
apr_off_t *plen, int *peos)
{
+ conn_rec *c = stream->session->c;
+ apr_status_t status = APR_SUCCESS;
+
if (stream->rst_error) {
return APR_ECONNRESET;
}
- if (!stream->sos) {
- return APR_EGENERAL;
+ status = h2_util_bb_readx(stream->buffer, cb, ctx, plen, peos);
+ if (status == APR_SUCCESS && !*peos && !*plen) {
+ status = APR_EAGAIN;
}
- return stream->sos->readx(stream->sos, cb, ctx, plen, peos);
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, c,
+ "h2_stream(%ld-%d): readx, len=%ld eos=%d",
+ c->id, stream->id, (long)*plen, *peos);
+ return status;
}
+
apr_status_t h2_stream_read_to(h2_stream *stream, apr_bucket_brigade *bb,
apr_off_t *plen, int *peos)
{
+ conn_rec *c = stream->session->c;
+ apr_status_t status = APR_SUCCESS;
+
if (stream->rst_error) {
return APR_ECONNRESET;
}
- if (!stream->sos) {
- return APR_EGENERAL;
+ status = h2_append_brigade(bb, stream->buffer, plen, peos);
+ if (status == APR_SUCCESS && !*peos && !*plen) {
+ status = APR_EAGAIN;
}
- return stream->sos->read_to(stream->sos, bb, plen, peos);
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, c,
+ "h2_stream(%ld-%d): read_to, len=%ld eos=%d",
+ c->id, stream->id, (long)*plen, *peos);
+ return status;
}
+
int h2_stream_input_is_open(const h2_stream *stream)
{
return input_open(stream);
@@ -469,7 +552,7 @@ apr_status_t h2_stream_submit_pushes(h2_
apr_table_t *h2_stream_get_trailers(h2_stream *stream)
{
- return stream->sos? stream->sos->get_trailers(stream->sos) : NULL;
+ return stream->response? stream->response->trailers : NULL;
}
const h2_priority *h2_stream_get_priority(h2_stream *stream)
@@ -486,147 +569,3 @@ const h2_priority *h2_stream_get_priorit
return NULL;
}
-/*******************************************************************************
- * h2_sos_mplx
- ******************************************************************************/
-
-typedef struct h2_sos_mplx {
- h2_mplx *m;
- apr_bucket_brigade *bb;
- apr_bucket_brigade *tmp;
- apr_table_t *trailers;
- apr_off_t buffer_size;
-} h2_sos_mplx;
-
-#define H2_SOS_MPLX_OUT(lvl,msos,msg) \
- do { \
- if (APLOG_C_IS_LEVEL((msos)->m->c,lvl)) \
- h2_util_bb_log((msos)->m->c,(msos)->m->id,lvl,msg,(msos)->bb); \
- } while(0)
-
-
-static apr_status_t mplx_transfer(h2_sos_mplx *msos, int stream_id,
- apr_pool_t *pool)
-{
- apr_status_t status;
- apr_table_t *trailers = NULL;
-
- if (!msos->tmp) {
- msos->tmp = apr_brigade_create(msos->bb->p, msos->bb->bucket_alloc);
- }
- status = h2_mplx_out_get_brigade(msos->m, stream_id, msos->tmp,
- msos->buffer_size-1, &trailers);
- if (!APR_BRIGADE_EMPTY(msos->tmp)) {
- h2_transfer_brigade(msos->bb, msos->tmp, pool);
- }
- if (trailers) {
- msos->trailers = trailers;
- }
- return status;
-}
-
-static apr_status_t h2_sos_mplx_read_to(h2_sos *sos, apr_bucket_brigade *bb,
- apr_off_t *plen, int *peos)
-{
- h2_sos_mplx *msos = sos->ctx;
- apr_status_t status;
-
- status = h2_append_brigade(bb, msos->bb, plen, peos);
- if (status == APR_SUCCESS && !*peos && !*plen) {
- status = APR_EAGAIN;
- ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, msos->m->c,
- "h2_stream(%ld-%d): read_to, len=%ld eos=%d",
- msos->m->id, sos->stream->id, (long)*plen, *peos);
- }
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, msos->m->c,
- "h2_stream(%ld-%d): read_to, len=%ld eos=%d",
- msos->m->id, sos->stream->id, (long)*plen, *peos);
- return status;
-}
-
-static apr_status_t h2_sos_mplx_readx(h2_sos *sos, h2_io_data_cb *cb, void *ctx,
- apr_off_t *plen, int *peos)
-{
- h2_sos_mplx *msos = sos->ctx;
- apr_status_t status = APR_SUCCESS;
-
- status = h2_util_bb_readx(msos->bb, cb, ctx, plen, peos);
- if (status == APR_SUCCESS && !*peos && !*plen) {
- status = APR_EAGAIN;
- }
- ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, msos->m->c,
- "h2_stream(%ld-%d): readx, len=%ld eos=%d",
- msos->m->id, sos->stream->id, (long)*plen, *peos);
- return status;
-}
-
-static apr_status_t h2_sos_mplx_prepare(h2_sos *sos, apr_off_t *plen, int *peos)
-{
- h2_sos_mplx *msos = sos->ctx;
- apr_status_t status = APR_SUCCESS;
-
- H2_SOS_MPLX_OUT(APLOG_TRACE2, msos, "h2_sos_mplx prepare_pre");
-
- if (APR_BRIGADE_EMPTY(msos->bb)) {
- status = mplx_transfer(msos, sos->stream->id, sos->stream->pool);
- }
- h2_util_bb_avail(msos->bb, plen, peos);
-
- H2_SOS_MPLX_OUT(APLOG_TRACE2, msos, "h2_sos_mplx prepare_post");
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, msos->m->c,
- "h2_stream(%ld-%d): prepare, len=%ld eos=%d, trailers=%s",
- msos->m->id, sos->stream->id, (long)*plen, *peos,
- msos->trailers? "yes" : "no");
- if (!*peos && !*plen) {
- status = APR_EAGAIN;
- }
-
- return status;
-}
-
-static apr_table_t *h2_sos_mplx_get_trailers(h2_sos *sos)
-{
- h2_sos_mplx *msos = sos->ctx;
-
- return msos->trailers;
-}
-
-static apr_status_t h2_sos_mplx_buffer(h2_sos *sos, apr_bucket_brigade *bb)
-{
- h2_sos_mplx *msos = sos->ctx;
- apr_status_t status = APR_SUCCESS;
-
- if (bb && !APR_BRIGADE_EMPTY(bb)) {
- H2_SOS_MPLX_OUT(APLOG_TRACE2, msos, "h2_sos_mplx set_response_pre");
- status = mplx_transfer(msos, sos->stream->id, sos->stream->pool);
- H2_SOS_MPLX_OUT(APLOG_TRACE2, msos, "h2_sos_mplx set_response_post");
- }
- return status;
-}
-
-static h2_sos *h2_sos_mplx_create(h2_stream *stream, h2_response *response)
-{
- h2_sos *sos;
- h2_sos_mplx *msos;
-
- msos = apr_pcalloc(stream->pool, sizeof(*msos));
- msos->m = stream->session->mplx;
- msos->bb = apr_brigade_create(stream->pool, msos->m->c->bucket_alloc);
- msos->buffer_size = 32 * 1024;
-
- sos = apr_pcalloc(stream->pool, sizeof(*sos));
- sos->stream = stream;
- sos->response = response;
-
- sos->ctx = msos;
- sos->buffer = h2_sos_mplx_buffer;
- sos->prepare = h2_sos_mplx_prepare;
- sos->readx = h2_sos_mplx_readx;
- sos->read_to = h2_sos_mplx_read_to;
- sos->get_trailers = h2_sos_mplx_get_trailers;
-
- sos->response = response;
-
- return sos;
-}
-
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_stream.h
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_stream.h?rev=1741419&r1=1741418&r2=1741419&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_stream.h (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_stream.h Thu Apr 28 12:43:02 2016
@@ -30,7 +30,6 @@
* The h2_response gives the HEADER frames to sent to the client, followed
* by DATA frames read from the h2_stream until EOS is reached.
*/
-#include "h2_io.h"
struct h2_mplx;
struct h2_priority;
@@ -38,6 +37,7 @@ struct h2_request;
struct h2_response;
struct h2_session;
struct h2_sos;
+struct h2_bucket_beam;
typedef struct h2_stream h2_stream;
@@ -48,8 +48,14 @@ struct h2_stream {
apr_pool_t *pool; /* the memory pool for this stream */
struct h2_request *request; /* the request made in this stream */
+ struct h2_bucket_beam *input;
+
+ struct h2_response *response;
+ struct h2_bucket_beam *output;
+ apr_bucket_brigade *buffer;
+ apr_bucket_brigade *tmp;
+
int rst_error; /* stream error for RST_STREAM */
-
unsigned int aborted : 1; /* was aborted */
unsigned int suspended : 1; /* DATA sending has been suspended */
unsigned int scheduled : 1; /* stream has been scheduled */
@@ -57,7 +63,6 @@ struct h2_stream {
apr_off_t input_remaining; /* remaining bytes on input as advertised via content-length */
- struct h2_sos *sos; /* stream output source, e.g. to read output from */
apr_off_t data_frames_sent; /* # of DATA frames sent out for this stream */
};
@@ -71,15 +76,18 @@ struct h2_stream {
* @param session the session this stream belongs to
* @return the newly opened stream
*/
-h2_stream *h2_stream_open(int id, apr_pool_t *pool, struct h2_session *session);
+h2_stream *h2_stream_open(int id, apr_pool_t *pool, struct h2_session *session,
+ int initiated_on, const struct h2_request *req);
/**
- * Destroy any resources held by this stream. Will destroy memory pool
- * if still owned by the stream.
- *
- * @param stream the stream to destroy
+ * Cleanup any resources still held by the stream, called by last bucket.
*/
-apr_status_t h2_stream_destroy(h2_stream *stream);
+void h2_stream_eos_destroy(h2_stream *stream);
+
+/**
+ * Destroy memory pool if still owned by the stream.
+ */
+void h2_stream_destroy(h2_stream *stream);
/**
* Removes stream from h2_session and destroys it.
@@ -106,15 +114,6 @@ apr_pool_t *h2_stream_detach_pool(h2_str
*/
apr_status_t h2_stream_set_request(h2_stream *stream, request_rec *r);
-/**
- * Initialize stream->request with the given h2_request.
- *
- * @param stream the stream to init the request for
- * @param req the request for initializing, will be copied
- */
-void h2_stream_set_h2_request(h2_stream *stream, int initiated_on,
- const struct h2_request *req);
-
/*
* Add a HTTP/2 header (including pseudo headers) or trailer
* to the given stream, depending on stream state.
@@ -187,7 +186,7 @@ struct h2_response *h2_stream_get_respon
*/
apr_status_t h2_stream_set_response(h2_stream *stream,
struct h2_response *response,
- apr_bucket_brigade *bb);
+ struct h2_bucket_beam *output);
/**
* Do a speculative read on the stream output to determine the
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task.c?rev=1741419&r1=1741418&r2=1741419&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task.c Thu Apr 28 12:43:02 2016
@@ -33,6 +33,8 @@
#include <scoreboard.h>
#include "h2_private.h"
+#include "h2.h"
+#include "h2_bucket_beam.h"
#include "h2_conn.h"
#include "h2_config.h"
#include "h2_ctx.h"
@@ -42,12 +44,371 @@
#include "h2_request.h"
#include "h2_session.h"
#include "h2_stream.h"
-#include "h2_task_input.h"
-#include "h2_task_output.h"
#include "h2_task.h"
-#include "h2_ctx.h"
#include "h2_worker.h"
+#include "h2_util.h"
+
+/*******************************************************************************
+ * task input handling
+ ******************************************************************************/
+
+static int input_ser_header(void *ctx, const char *name, const char *value)
+{
+ h2_task *task = ctx;
+ apr_brigade_printf(task->input.bb, NULL, NULL, "%s: %s\r\n", name, value);
+ return 1;
+}
+
+static apr_status_t input_append_eos(h2_task *task, request_rec *r)
+{
+ apr_status_t status = APR_SUCCESS;
+ apr_bucket_brigade *bb = task->input.bb;
+ apr_table_t *t = task->request->trailers;
+
+ if (task->input.chunked) {
+ if (t && !apr_is_empty_table(t)) {
+ status = apr_brigade_puts(bb, NULL, NULL, "0\r\n");
+ apr_table_do(input_ser_header, task, t, NULL);
+ status = apr_brigade_puts(bb, NULL, NULL, "\r\n");
+ }
+ else {
+ status = apr_brigade_puts(bb, NULL, NULL, "0\r\n\r\n");
+ }
+ }
+ else if (r && t && !apr_is_empty_table(t)){
+ /* trailers passed in directly. */
+ apr_table_overlap(r->trailers_in, t, APR_OVERLAP_TABLES_SET);
+ }
+ task->input.eos_written = 1;
+ APR_BRIGADE_INSERT_TAIL(bb, apr_bucket_eos_create(bb->bucket_alloc));
+ return status;
+}
+
+static apr_status_t input_read(h2_task *task, ap_filter_t* f,
+ apr_bucket_brigade* bb, ap_input_mode_t mode,
+ apr_read_type_e block, apr_off_t readbytes)
+{
+ apr_status_t status = APR_SUCCESS;
+ apr_bucket *b, *next;
+ apr_off_t bblen = 0;
+
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
+ "h2_task(%s): read, mode=%d, block=%d, readbytes=%ld",
+ task->id, mode, block, (long)readbytes);
+
+ if (mode == AP_MODE_INIT) {
+ return ap_get_brigade(f->c->input_filters, bb, mode, block, readbytes);
+ }
+
+ if (f->c->aborted) {
+ return APR_ECONNABORTED;
+ }
+
+ if (task->input.bb) {
+ /* Cleanup brigades from those nasty 0 length non-meta buckets
+ * that apr_brigade_split_line() sometimes produces. */
+ for (b = APR_BRIGADE_FIRST(task->input.bb);
+ b != APR_BRIGADE_SENTINEL(task->input.bb); b = next) {
+ next = APR_BUCKET_NEXT(b);
+ if (b->length == 0 && !APR_BUCKET_IS_METADATA(b)) {
+ apr_bucket_delete(b);
+ }
+ }
+ apr_brigade_length(task->input.bb, 0, &bblen);
+ }
+
+ if (bblen == 0) {
+ if (task->input.eos_written) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, APR_EOF, f->c,
+ "h2_task(%s): read no data", task->id);
+ return APR_EOF;
+ }
+ else if (task->input.eos) {
+ input_append_eos(task, f->r);
+ }
+ }
+
+ while (APR_BRIGADE_EMPTY(task->input.bb)) {
+ /* Get more input data for our request. */
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
+ "h2_task(%s): get more data from mplx, block=%d, "
+ "readbytes=%ld, queued=%ld",
+ task->id, block, (long)readbytes, (long)bblen);
+
+ /* Override the block mode we get called with depending on the input's
+ * setting. */
+ if (task->input.beam) {
+ status = h2_beam_receive(task->input.beam, task->input.bb, block,
+ H2MIN(readbytes, 32*1024));
+ }
+ else {
+ status = APR_EOF;
+ }
+
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
+ "h2_task(%s): read returned", task->id);
+ if (APR_STATUS_IS_EAGAIN(status)
+ && (mode == AP_MODE_GETLINE || block == APR_BLOCK_READ)) {
+ /* chunked input handling does not seem to like it if we
+ * return with APR_EAGAIN from a GETLINE read...
+ * upload 100k test on test-ser.example.org hangs */
+ status = APR_SUCCESS;
+ }
+ else if (APR_STATUS_IS_EOF(status) && !task->input.eos_written) {
+ task->input.eos = 1;
+ }
+ else if (status != APR_SUCCESS) {
+ return status;
+ }
+
+ apr_brigade_length(task->input.bb, 0, &bblen);
+ if (bblen > 0 && task->input.chunked) {
+ /* need to add chunks since request processing expects it */
+ char buffer[128];
+ apr_bucket *b;
+ int len;
+
+ len = apr_snprintf(buffer, H2_ALEN(buffer), "%lx\r\n",
+ (unsigned long)bblen);
+ b = apr_bucket_heap_create(buffer, len, NULL,
+ task->input.bb->bucket_alloc);
+ APR_BRIGADE_INSERT_HEAD(task->input.bb, b);
+ status = apr_brigade_puts(task->input.bb, NULL, NULL, "\r\n");
+ }
+
+ if (h2_util_has_eos(task->input.bb, -1)) {
+ task->input.eos = 1;
+ }
+
+ if (task->input.eos && !task->input.eos_written) {
+ input_append_eos(task, f->r);
+ }
+
+ if (h2_task_logio_add_bytes_in) {
+ h2_task_logio_add_bytes_in(f->c, bblen);
+ }
+ }
+
+ h2_util_bb_log(f->c, task->stream_id, APLOG_TRACE2,
+ "task_input.bb", task->input.bb);
+
+ if (APR_BRIGADE_EMPTY(task->input.bb)) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
+ "h2_task(%s): no data", task->id);
+ return (block == APR_NONBLOCK_READ)? APR_EAGAIN : APR_EOF;
+ }
+
+ if (mode == AP_MODE_EXHAUSTIVE) {
+ /* return all we have */
+ APR_BRIGADE_CONCAT(bb, task->input.bb);
+ }
+ else if (mode == AP_MODE_READBYTES) {
+ status = h2_brigade_concat_length(bb, task->input.bb, readbytes);
+ }
+ else if (mode == AP_MODE_SPECULATIVE) {
+ status = h2_brigade_copy_length(bb, task->input.bb, readbytes);
+ }
+ else if (mode == AP_MODE_GETLINE) {
+ /* we are reading a single LF line, e.g. the HTTP headers.
+ * this has the nasty side effect to split the bucket, even
+ * though it ends with CRLF and creates a 0 length bucket */
+ status = apr_brigade_split_line(bb, task->input.bb, block,
+ HUGE_STRING_LEN);
+ if (APLOGctrace1(f->c)) {
+ char buffer[1024];
+ apr_size_t len = sizeof(buffer)-1;
+ apr_brigade_flatten(bb, buffer, &len);
+ buffer[len] = 0;
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
+ "h2_task(%s): getline: %s",
+ task->id, buffer);
+ }
+ }
+ else {
+ /* Hmm, well. There is mode AP_MODE_EATCRLF, but we chose not
+ * to support it. Seems to work. */
+ ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOTIMPL, f->c,
+ APLOGNO(02942)
+ "h2_task, unsupported READ mode %d", mode);
+ status = APR_ENOTIMPL;
+ }
+
+ if (APLOGctrace1(f->c)) {
+ apr_brigade_length(bb, 0, &bblen);
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
+ "h2_task(%s): return %ld data bytes",
+ task->id, (long)bblen);
+ }
+ return status;
+}
+
+/*******************************************************************************
+ * task output handling
+ ******************************************************************************/
+
+static apr_status_t open_response(h2_task *task)
+{
+ h2_response *response;
+ response = h2_from_h1_get_response(task->output.from_h1);
+ if (!response) {
+ /* This happens currently when ap_die(status, r) is invoked
+ * by a read request filter. */
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c, APLOGNO(03204)
+ "h2_task(%s): write without response for %s %s %s",
+ task->id,
+ task->request->method,
+ task->request->authority,
+ task->request->path);
+ task->c->aborted = 1;
+ return APR_ECONNABORTED;
+ }
+
+ if (h2_task_logio_add_bytes_out) {
+ /* count headers as if we'd do a HTTP/1.1 serialization */
+ task->output.written = h2_util_table_bytes(response->headers, 3)+1;
+ h2_task_logio_add_bytes_out(task->c, task->output.written);
+ }
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c, APLOGNO(03348)
+ "h2_task(%s): open response to %s %s %s",
+ task->id, task->request->method,
+ task->request->authority,
+ task->request->path);
+ return h2_mplx_out_open(task->mplx, task->stream_id, response);
+}
+
+static apr_status_t send_out(h2_task *task, apr_bucket_brigade* bb)
+{
+ apr_off_t written, left;
+ apr_status_t status;
+
+ apr_brigade_length(bb, 0, &written);
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
+ "h2_task(%s): write response body (%ld bytes)",
+ task->id, (long)written);
+
+ status = h2_beam_send(task->output.beam, bb,
+ task->blocking? APR_BLOCK_READ
+ : APR_NONBLOCK_READ);
+ if (APR_STATUS_IS_EAGAIN(status)) {
+ apr_brigade_length(bb, 0, &left);
+ written -= left;
+ status = APR_SUCCESS;
+ }
+ if (status == APR_SUCCESS) {
+ task->output.written += written;
+ if (h2_task_logio_add_bytes_out) {
+ h2_task_logio_add_bytes_out(task->c, written);
+ }
+ }
+ return status;
+}
+
+/* Bring the data from the brigade (which represents the result of the
+ * request_rec out filter chain) into the h2_mplx for further sending
+ * on the master connection.
+ */
+static apr_status_t output_write(h2_task *task, ap_filter_t* f,
+ apr_bucket_brigade* bb)
+{
+ apr_bucket *b;
+ apr_status_t status = APR_SUCCESS;
+ int flush = 0;
+
+ if (APR_BRIGADE_EMPTY(bb)) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
+ "h2_task(%s): empty write", task->id);
+ return APR_SUCCESS;
+ }
+
+ if (task->frozen) {
+ h2_util_bb_log(task->c, task->stream_id, APLOG_TRACE2,
+ "frozen task output write, ignored", bb);
+ while (!APR_BRIGADE_EMPTY(bb)) {
+ b = APR_BRIGADE_FIRST(bb);
+ if (AP_BUCKET_IS_EOR(b)) {
+ APR_BUCKET_REMOVE(b);
+ task->eor = b;
+ }
+ else {
+ apr_bucket_delete(b);
+ }
+ }
+ return APR_SUCCESS;
+ }
+
+ if (!task->output.beam) {
+ h2_beam_create(&task->output.beam, task->pool,
+ task->stream_id, "output", 0);
+ }
+
+ /* Attempt to write saved brigade first */
+ if (task->output.bb && !APR_BRIGADE_EMPTY(task->output.bb)) {
+ status = send_out(task, task->output.bb);
+ if (status != APR_SUCCESS) {
+ return status;
+ }
+ }
+
+ /* If there is nothing saved (anymore), try to write the brigade passed */
+ if ((!task->output.bb || APR_BRIGADE_EMPTY(task->output.bb))
+ && !APR_BRIGADE_EMPTY(bb)) {
+ /* check if we have a flush before the end-of-request */
+ if (!task->output.response_open) {
+ for (b = APR_BRIGADE_FIRST(bb);
+ b != APR_BRIGADE_SENTINEL(bb);
+ b = APR_BUCKET_NEXT(b)) {
+ if (AP_BUCKET_IS_EOR(b)) {
+ break;
+ }
+ else if (APR_BUCKET_IS_FLUSH(b)) {
+ flush = 1;
+ }
+ }
+ }
+
+ status = send_out(task, bb);
+ if (status != APR_SUCCESS) {
+ return status;
+ }
+ }
+
+ /* If the passed brigade is not empty, save it before return */
+ if (!APR_BRIGADE_EMPTY(bb)) {
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, task->c,
+ "h2_task(%s): could not write all, saving brigade",
+ task->id);
+ if (!task->output.bb) {
+ task->output.bb = apr_brigade_create(task->pool,
+ task->c->bucket_alloc);
+ }
+ return ap_save_brigade(f, &task->output.bb, &bb, task->pool);
+ }
+
+ if (!task->output.response_open
+ && (flush || h2_beam_get_mem_used(task->output.beam) > (32*1024))) {
+ /* if we have enough buffered or we got a flush bucket, open
+ * the response now. */
+ status = open_response(task);
+ task->output.response_open = 1;
+ }
+
+ return status;
+}
+static apr_status_t output_finish(h2_task *task)
+{
+ apr_status_t status = APR_SUCCESS;
+
+ if (!task->output.response_open) {
+ status = open_response(task);
+ task->output.response_open = 1;
+ }
+ return status;
+}
+
+/*******************************************************************************
+ * task slave connection filters
+ ******************************************************************************/
static apr_status_t h2_filter_stream_input(ap_filter_t* filter,
apr_bucket_brigade* brigade,
@@ -57,11 +418,7 @@ static apr_status_t h2_filter_stream_inp
{
h2_task *task = h2_ctx_cget_task(filter->c);
AP_DEBUG_ASSERT(task);
- if (!task->input) {
- return APR_ECONNABORTED;
- }
- return h2_task_input_read(task->input, filter, brigade,
- mode, block, readbytes);
+ return input_read(task, filter, brigade, mode, block, readbytes);
}
static apr_status_t h2_filter_stream_output(ap_filter_t* filter,
@@ -69,10 +426,7 @@ static apr_status_t h2_filter_stream_out
{
h2_task *task = h2_ctx_cget_task(filter->c);
AP_DEBUG_ASSERT(task);
- if (!task->output) {
- return APR_ECONNABORTED;
- }
- return h2_task_output_write(task->output, filter, brigade);
+ return output_write(task, filter, brigade);
}
static apr_status_t h2_filter_read_response(ap_filter_t* filter,
@@ -80,10 +434,74 @@ static apr_status_t h2_filter_read_respo
{
h2_task *task = h2_ctx_cget_task(filter->c);
AP_DEBUG_ASSERT(task);
- if (!task->output || !task->output->from_h1) {
+ if (!task->output.from_h1) {
return APR_ECONNABORTED;
}
- return h2_from_h1_read_response(task->output->from_h1, filter, bb);
+ return h2_from_h1_read_response(task->output.from_h1, filter, bb);
+}
+
+/*******************************************************************************
+ * task things
+ ******************************************************************************/
+
+void h2_task_set_response(h2_task *task, h2_response *response)
+{
+ AP_DEBUG_ASSERT(response);
+ AP_DEBUG_ASSERT(!task->response);
+ /* we used to clone the response into out own pool. But
+ * we have much tighter control over the EOR bucket nowadays,
+ * so just use the instance given */
+ task->response = response;
+ if (response->rst_error) {
+ h2_task_rst(task, response->rst_error);
+ }
+}
+
+
+int h2_task_can_redo(h2_task *task) {
+ if (task->submitted
+ || (task->input.beam && h2_beam_was_received(task->input.beam))
+ || !task->request) {
+ /* cannot repeat that. */
+ return 0;
+ }
+ return (!strcmp("GET", task->request->method)
+ || !strcmp("HEAD", task->request->method)
+ || !strcmp("OPTIONS", task->request->method));
+}
+
+void h2_task_redo(h2_task *task)
+{
+ task->response = NULL;
+ task->rst_error = 0;
+}
+
+void h2_task_rst(h2_task *task, int error)
+{
+ task->rst_error = error;
+ if (task->input.beam) {
+ h2_beam_abort(task->input.beam);
+ }
+ if (task->output.beam) {
+ h2_beam_abort(task->output.beam);
+ }
+}
+
+apr_status_t h2_task_shutdown(h2_task *task, int block)
+{
+ if (task->output.beam) {
+ apr_status_t status;
+ status = h2_beam_shutdown(task->output.beam, APR_NONBLOCK_READ);
+ if (block && status == APR_EAGAIN) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, task->c,
+ "h2_task(%s): output shutdown waiting", task->id);
+ status = h2_beam_shutdown(task->output.beam, APR_BLOCK_READ);
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, task->c,
+ "h2_task(%s): output shutdown done", task->id);
+ }
+ return status;
+ }
+ return APR_SUCCESS;
}
/*******************************************************************************
@@ -153,8 +571,8 @@ static int h2_task_pre_conn(conn_rec* c,
return OK;
}
-h2_task *h2_task_create(long session_id, const h2_request *req,
- conn_rec *c, h2_mplx *mplx)
+h2_task *h2_task_create(conn_rec *c, const h2_request *req,
+ h2_bucket_beam *input, h2_mplx *mplx)
{
apr_pool_t *pool;
h2_task *task;
@@ -164,21 +582,22 @@ h2_task *h2_task_create(long session_id,
if (task == NULL) {
ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, c,
APLOGNO(02941) "h2_task(%ld-%d): create stream task",
- session_id, req->id);
- h2_mplx_out_close(mplx, req->id);
+ c->id, req->id);
return NULL;
}
- task->id = apr_psprintf(pool, "%ld-%d", session_id, req->id);
+ task->id = apr_psprintf(pool, "%ld-%d", c->id, req->id);
task->stream_id = req->id;
task->c = c;
task->mplx = mplx;
task->c->keepalives = mplx->c->keepalives;
task->pool = pool;
task->request = req;
- task->input_eos = !req->body;
task->ser_headers = req->serialize;
task->blocking = 1;
+ task->input.beam = input;
+
+ apr_thread_cond_create(&task->cond, pool);
h2_ctx_create_for(c, task);
return task;
@@ -186,6 +605,13 @@ h2_task *h2_task_create(long session_id,
void h2_task_destroy(h2_task *task)
{
+ if (task->output.beam) {
+ h2_beam_destroy(task->output.beam);
+ task->output.beam = NULL;
+ }
+ if (task->eor) {
+ apr_bucket_destroy(task->eor);
+ }
if (task->pool) {
apr_pool_destroy(task->pool);
}
@@ -196,14 +622,37 @@ void h2_task_set_io_blocking(h2_task *ta
task->blocking = blocking;
}
-apr_status_t h2_task_do(h2_task *task, apr_thread_cond_t *cond)
+apr_status_t h2_task_do(h2_task *task)
{
- apr_status_t status;
-
AP_DEBUG_ASSERT(task);
- task->io = cond;
- task->input = h2_task_input_create(task, task->c);
- task->output = h2_task_output_create(task, task->c);
+
+ task->input.block = APR_BLOCK_READ;
+ task->input.chunked = task->request->chunked;
+ task->input.eos = !task->request->body;
+ if (task->input.eos && !task->input.chunked && !task->ser_headers) {
+ /* We do not serialize/chunk and have eos already, no need to
+ * create a bucket brigade. */
+ task->input.bb = NULL;
+ task->input.eos_written = 1;
+ }
+ else {
+ task->input.bb = apr_brigade_create(task->pool, task->c->bucket_alloc);
+ if (task->ser_headers) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
+ "h2_task(%s): serialize request %s %s",
+ task->id, task->request->method, task->request->path);
+ apr_brigade_printf(task->input.bb, NULL,
+ NULL, "%s %s HTTP/1.1\r\n",
+ task->request->method, task->request->path);
+ apr_table_do(input_ser_header, task, task->request->headers, NULL);
+ apr_brigade_puts(task->input.bb, NULL, NULL, "\r\n");
+ }
+ if (task->input.eos) {
+ input_append_eos(task, NULL);
+ }
+ }
+
+ task->output.from_h1 = h2_from_h1_create(task->stream_id, task->pool);
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
"h2_task(%s): process connection", task->id);
@@ -214,15 +663,13 @@ apr_status_t h2_task_do(h2_task *task, a
"h2_task(%s): process_conn returned frozen task",
task->id);
/* cleanup delayed */
- status = APR_EAGAIN;
+ return APR_EAGAIN;
}
else {
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
"h2_task(%s): processing done", task->id);
- status = APR_SUCCESS;
+ return output_finish(task);
}
-
- return status;
}
static apr_status_t h2_task_process_request(h2_task *task, conn_rec *c)
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task.h
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task.h?rev=1741419&r1=1741418&r2=1741419&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task.h (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task.h Thu Apr 28 12:43:02 2016
@@ -38,12 +38,13 @@
*/
struct apr_thread_cond_t;
+struct h2_bucket_beam;
struct h2_conn;
struct h2_mplx;
struct h2_task;
struct h2_req_engine;
struct h2_request;
-struct h2_resp_head;
+struct h2_response;
struct h2_worker;
typedef struct h2_task h2_task;
@@ -52,32 +53,71 @@ struct h2_task {
const char *id;
int stream_id;
conn_rec *c;
- struct h2_mplx *mplx;
apr_pool_t *pool;
+
const struct h2_request *request;
+ struct h2_response *response;
+
+ struct {
+ struct h2_bucket_beam *beam;
+ apr_bucket_brigade *bb;
+ apr_read_type_e block;
+ unsigned int chunked : 1;
+ unsigned int eos : 1;
+ unsigned int eos_written : 1;
+ } input;
+ struct {
+ struct h2_bucket_beam *beam;
+ struct h2_from_h1 *from_h1;
+ unsigned int response_open : 1;
+ apr_off_t written;
+ apr_bucket_brigade *bb;
+ } output;
+
+ struct h2_mplx *mplx;
+ struct apr_thread_cond_t *cond;
+
+ int rst_error; /* h2 related stream abort error */
+ unsigned int filters_set : 1;
+ unsigned int ser_headers : 1;
+ unsigned int frozen : 1;
+ unsigned int blocking : 1;
+ unsigned int detached : 1;
+ unsigned int orphaned : 1; /* h2_stream is gone for this task */
+ unsigned int submitted : 1; /* response has been submitted to client */
+ unsigned int worker_started : 1; /* h2_worker started processing for this io */
+ unsigned int worker_done : 1; /* h2_worker finished for this io */
- unsigned int filters_set : 1;
- unsigned int input_eos : 1;
- unsigned int ser_headers : 1;
- unsigned int frozen : 1;
- unsigned int blocking : 1;
- unsigned int detached : 1;
-
- struct h2_task_input *input;
- struct h2_task_output *output;
- struct apr_thread_cond_t *io; /* used to wait for events on */
+ apr_time_t started_at; /* when processing started */
+ apr_time_t done_at; /* when processing was done */
+ apr_bucket *eor;
struct h2_req_engine *engine; /* engine hosted by this task */
struct h2_req_engine *assigned; /* engine that task has been assigned to */
request_rec *r; /* request being processed in this task */
};
-h2_task *h2_task_create(long session_id, const struct h2_request *req,
- conn_rec *c, struct h2_mplx *mplx);
+h2_task *h2_task_create(conn_rec *c, const struct h2_request *req,
+ struct h2_bucket_beam *input, struct h2_mplx *mplx);
void h2_task_destroy(h2_task *task);
-apr_status_t h2_task_do(h2_task *task, struct apr_thread_cond_t *cond);
+apr_status_t h2_task_do(h2_task *task);
+
+void h2_task_set_response(h2_task *task, struct h2_response *response);
+
+void h2_task_redo(h2_task *task);
+int h2_task_can_redo(h2_task *task);
+
+/**
+ * Reset the task with the given error code, resets all input/output.
+ */
+void h2_task_rst(h2_task *task, int error);
+
+/**
+ * Shuts all input/output down. Clears any buckets buffered and closes.
+ */
+apr_status_t h2_task_shutdown(h2_task *task, int block);
void h2_task_register_hooks(void);
/*
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_util.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_util.c?rev=1741419&r1=1741418&r2=1741419&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_util.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_util.c Thu Apr 28 12:43:02 2016
@@ -286,7 +286,7 @@ size_t h2_ihash_count(h2_ihash_t *ih)
return apr_hash_count(ih->hash);
}
-int h2_ihash_is_empty(h2_ihash_t *ih)
+int h2_ihash_empty(h2_ihash_t *ih)
{
return apr_hash_count(ih->hash) == 0;
}
@@ -332,6 +332,301 @@ void h2_ihash_clear(h2_ihash_t *ih)
}
/*******************************************************************************
+ * ilist - sorted list for structs with int identifier
+ ******************************************************************************/
+
+#define h2_ilist_IDX(list, i) ((int**)(list)->elts)[i]
+
+struct h2_ilist_t {
+ apr_array_header_t *l;
+};
+
+h2_ilist_t *h2_ilist_create(apr_pool_t *pool)
+{
+ h2_ilist_t *list = apr_pcalloc(pool, sizeof(h2_ilist_t));
+ if (list) {
+ list->l = apr_array_make(pool, 100, sizeof(int*));
+ if (!list->l) {
+ return NULL;
+ }
+ }
+ return list;
+}
+
+static int h2_ilist_cmp(const void *s1, const void *s2)
+{
+ int **pi1 = (int **)s1;
+ int **pi2 = (int **)s2;
+ return *(*pi1) - *(*pi2);
+}
+
+void *h2_ilist_get(h2_ilist_t *list, int id)
+{
+ /* we keep the array sorted by id, so lookup can be done
+ * by bsearch.
+ */
+ int **pi;
+ int *pkey = &id;
+
+ pi = bsearch(&pkey, list->l->elts, list->l->nelts,
+ list->l->elt_size, h2_ilist_cmp);
+ return pi? *pi : NULL;
+}
+
+static void h2_ilist_sort(h2_ilist_t *list)
+{
+ qsort(list->l->elts, list->l->nelts, list->l->elt_size, h2_ilist_cmp);
+}
+
+apr_status_t h2_ilist_add(h2_ilist_t *list, void *val)
+{
+ int *pi = val;
+ void *existing = h2_ilist_get(list, *pi);
+ if (!existing) {
+ int last;
+ APR_ARRAY_PUSH(list->l, void*) = val;
+ /* Often, values get added in ascending order of id. We
+ * keep the array sorted, so we just need to check if the newly
+ * appended stream has a lower id than the last one. if not,
+ * sorting is not necessary.
+ */
+ last = list->l->nelts - 1;
+ if (last > 0
+ && *h2_ilist_IDX(list->l, last) < *h2_ilist_IDX(list->l, last-1)) {
+ h2_ilist_sort(list);
+ }
+ }
+ return APR_SUCCESS;
+}
+
+static void remove_idx(h2_ilist_t *list, int idx)
+{
+ int n;
+ --list->l->nelts;
+ n = list->l->nelts - idx;
+ if (n > 0) {
+ /* There are n h2_io* behind idx. Move the rest down */
+ int **selts = (int**)list->l->elts;
+ memmove(selts + idx, selts + idx + 1, n * sizeof(int*));
+ }
+}
+
+void *h2_ilist_remove(h2_ilist_t *list, int id)
+{
+ int i;
+ for (i = 0; i < list->l->nelts; ++i) {
+ int *e = h2_ilist_IDX(list->l, i);
+ if (id == *e) {
+ remove_idx(list, i);
+ return e;
+ }
+ }
+ return NULL;
+}
+
+void *h2_ilist_shift(h2_ilist_t *list)
+{
+ if (list->l->nelts > 0) {
+ int *pi = h2_ilist_IDX(list->l, 0);
+ remove_idx(list, 0);
+ return pi;
+ }
+ return NULL;
+}
+
+int h2_ilist_empty(h2_ilist_t *list)
+{
+ return list->l->nelts == 0;
+}
+
+int h2_ilist_iter(h2_ilist_t *list, h2_ilist_iter_t *iter, void *ctx)
+{
+ int i;
+ for (i = 0; i < list->l->nelts; ++i) {
+ int *pi = h2_ilist_IDX(list->l, i);
+ if (!iter(ctx, pi)) {
+ return 0;
+ }
+ }
+ return 1;
+}
+
+apr_size_t h2_ilist_count(h2_ilist_t *list)
+{
+ return list->l->nelts;
+}
+
+/*******************************************************************************
+ * iqueue - sorted list of int
+ ******************************************************************************/
+
+static void iq_grow(h2_iqueue *q, int nlen);
+static void iq_swap(h2_iqueue *q, int i, int j);
+static int iq_bubble_up(h2_iqueue *q, int i, int top,
+ h2_iq_cmp *cmp, void *ctx);
+static int iq_bubble_down(h2_iqueue *q, int i, int bottom,
+ h2_iq_cmp *cmp, void *ctx);
+
+h2_iqueue *h2_iq_create(apr_pool_t *pool, int capacity)
+{
+ h2_iqueue *q = apr_pcalloc(pool, sizeof(h2_iqueue));
+ if (q) {
+ q->pool = pool;
+ iq_grow(q, capacity);
+ q->nelts = 0;
+ }
+ return q;
+}
+
+int h2_iq_empty(h2_iqueue *q)
+{
+ return q->nelts == 0;
+}
+
+int h2_iq_count(h2_iqueue *q)
+{
+ return q->nelts;
+}
+
+
+void h2_iq_add(h2_iqueue *q, int sid, h2_iq_cmp *cmp, void *ctx)
+{
+ int i;
+
+ if (q->nelts >= q->nalloc) {
+ iq_grow(q, q->nalloc * 2);
+ }
+
+ i = (q->head + q->nelts) % q->nalloc;
+ q->elts[i] = sid;
+ ++q->nelts;
+
+ if (cmp) {
+ /* bubble it to the front of the queue */
+ iq_bubble_up(q, i, q->head, cmp, ctx);
+ }
+}
+
+int h2_iq_remove(h2_iqueue *q, int sid)
+{
+ int i;
+ for (i = 0; i < q->nelts; ++i) {
+ if (sid == q->elts[(q->head + i) % q->nalloc]) {
+ break;
+ }
+ }
+
+ if (i < q->nelts) {
+ ++i;
+ for (; i < q->nelts; ++i) {
+ q->elts[(q->head+i-1)%q->nalloc] = q->elts[(q->head+i)%q->nalloc];
+ }
+ --q->nelts;
+ return 1;
+ }
+ return 0;
+}
+
+void h2_iq_clear(h2_iqueue *q)
+{
+ q->nelts = 0;
+}
+
+void h2_iq_sort(h2_iqueue *q, h2_iq_cmp *cmp, void *ctx)
+{
+ /* Assume that changes in ordering are minimal. This needs,
+ * best case, q->nelts - 1 comparisions to check that nothing
+ * changed.
+ */
+ if (q->nelts > 0) {
+ int i, ni, prev, last;
+
+ /* Start at the end of the queue and create a tail of sorted
+ * entries. Make that tail one element longer in each iteration.
+ */
+ last = i = (q->head + q->nelts - 1) % q->nalloc;
+ while (i != q->head) {
+ prev = (q->nalloc + i - 1) % q->nalloc;
+
+ ni = iq_bubble_up(q, i, prev, cmp, ctx);
+ if (ni == prev) {
+ /* i bubbled one up, bubble the new i down, which
+ * keeps all tasks below i sorted. */
+ iq_bubble_down(q, i, last, cmp, ctx);
+ }
+ i = prev;
+ };
+ }
+}
+
+
+int h2_iq_shift(h2_iqueue *q)
+{
+ int sid;
+
+ if (q->nelts <= 0) {
+ return 0;
+ }
+
+ sid = q->elts[q->head];
+ q->head = (q->head + 1) % q->nalloc;
+ q->nelts--;
+
+ return sid;
+}
+
+static void iq_grow(h2_iqueue *q, int nlen)
+{
+ if (nlen > q->nalloc) {
+ int *nq = apr_pcalloc(q->pool, sizeof(int) * nlen);
+ if (q->nelts > 0) {
+ int l = ((q->head + q->nelts) % q->nalloc) - q->head;
+
+ memmove(nq, q->elts + q->head, sizeof(int) * l);
+ if (l < q->nelts) {
+ /* elts wrapped, append elts in [0, remain] to nq */
+ int remain = q->nelts - l;
+ memmove(nq + l, q->elts, sizeof(int) * remain);
+ }
+ }
+ q->elts = nq;
+ q->nalloc = nlen;
+ q->head = 0;
+ }
+}
+
+static void iq_swap(h2_iqueue *q, int i, int j)
+{
+ int x = q->elts[i];
+ q->elts[i] = q->elts[j];
+ q->elts[j] = x;
+}
+
+static int iq_bubble_up(h2_iqueue *q, int i, int top,
+ h2_iq_cmp *cmp, void *ctx)
+{
+ int prev;
+ while (((prev = (q->nalloc + i - 1) % q->nalloc), i != top)
+ && (*cmp)(q->elts[i], q->elts[prev], ctx) < 0) {
+ iq_swap(q, prev, i);
+ i = prev;
+ }
+ return i;
+}
+
+static int iq_bubble_down(h2_iqueue *q, int i, int bottom,
+ h2_iq_cmp *cmp, void *ctx)
+{
+ int next;
+ while (((next = (q->nalloc + i + 1) % q->nalloc), i != bottom)
+ && (*cmp)(q->elts[i], q->elts[next], ctx) > 0) {
+ iq_swap(q, next, i);
+ i = next;
+ }
+ return i;
+}
+
+/*******************************************************************************
* h2_util for apt_table_t
******************************************************************************/
@@ -368,15 +663,6 @@ apr_size_t h2_util_table_bytes(apr_table
* h2_util for bucket brigades
******************************************************************************/
-/* DEEP_COPY==0 crashes under load. I think the setaside is fine,
- * however buckets moved to another thread will still be
- * free'd against the old bucket_alloc. *And* if the old
- * pool gets destroyed too early, the bucket disappears while
- * still needed.
- */
-static const int DEEP_COPY = 1;
-static const int FILE_MOVE = 1;
-
static apr_status_t last_not_included(apr_bucket_brigade *bb,
apr_off_t maxlen,
int same_alloc,
@@ -434,203 +720,95 @@ static apr_status_t last_not_included(ap
return status;
}
-#define LOG_BUCKETS 0
-#define LOG_LEVEL APLOG_INFO
-
-apr_status_t h2_util_move(apr_bucket_brigade *to, apr_bucket_brigade *from,
- apr_off_t maxlen, apr_size_t *pfile_buckets_allowed,
- const char *msg)
+apr_status_t h2_brigade_concat_length(apr_bucket_brigade *dest,
+ apr_bucket_brigade *src,
+ apr_off_t length)
{
+ apr_bucket *b, *next;
+ apr_off_t remain = length;
apr_status_t status = APR_SUCCESS;
- int same_alloc;
-
- AP_DEBUG_ASSERT(to);
- AP_DEBUG_ASSERT(from);
- same_alloc = (to->bucket_alloc == from->bucket_alloc
- || to->p == from->p);
-
- if (!FILE_MOVE) {
- pfile_buckets_allowed = NULL;
- }
- if (!APR_BRIGADE_EMPTY(from)) {
- apr_bucket *b, *nb, *end;
+ for (b = APR_BRIGADE_FIRST(src);
+ b != APR_BRIGADE_SENTINEL(src);
+ b = next) {
+ next = APR_BUCKET_NEXT(b);
- status = last_not_included(from, maxlen, same_alloc,
- pfile_buckets_allowed, &end);
- if (status != APR_SUCCESS) {
- return status;
+ if (APR_BUCKET_IS_METADATA(b)) {
+ /* fall through */
}
-
- while (!APR_BRIGADE_EMPTY(from) && status == APR_SUCCESS) {
- b = APR_BRIGADE_FIRST(from);
- if (b == end) {
- break;
+ else {
+ if (remain == b->length) {
+ /* fall through */
}
-
- if (same_alloc || (b->list == to->bucket_alloc)) {
- /* both brigades use the same bucket_alloc and auto-cleanups
- * have the same life time. It's therefore safe to just move
- * directly. */
- APR_BUCKET_REMOVE(b);
- APR_BRIGADE_INSERT_TAIL(to, b);
-#if LOG_BUCKETS
- ap_log_perror(APLOG_MARK, LOG_LEVEL, 0, to->p, APLOGNO(03205)
- "h2_util_move: %s, passed bucket(same bucket_alloc) "
- "%ld-%ld, type=%s",
- msg, (long)b->start, (long)b->length,
- APR_BUCKET_IS_METADATA(b)?
- (APR_BUCKET_IS_EOS(b)? "EOS":
- (APR_BUCKET_IS_FLUSH(b)? "FLUSH" : "META")) :
- (APR_BUCKET_IS_FILE(b)? "FILE" : "DATA"));
-#endif
+ else if (remain <= 0) {
+ return status;
}
- else if (DEEP_COPY) {
- /* we have not managed the magic of passing buckets from
- * one thread to another. Any attempts result in
- * cleanup of pools scrambling memory.
- */
- if (APR_BUCKET_IS_METADATA(b)) {
- if (APR_BUCKET_IS_EOS(b)) {
- APR_BRIGADE_INSERT_TAIL(to, apr_bucket_eos_create(to->bucket_alloc));
- }
- else {
- /* ignore */
- }
- }
- else if (pfile_buckets_allowed
- && *pfile_buckets_allowed > 0
- && APR_BUCKET_IS_FILE(b)) {
- /* We do not want to read files when passing buckets, if
- * we can avoid it. However, what we've come up so far
- * is not working corrently, resulting either in crashes or
- * too many open file descriptors.
- */
- apr_bucket_file *f = (apr_bucket_file *)b->data;
- apr_file_t *fd = f->fd;
- int setaside = (f->readpool != to->p);
-#if LOG_BUCKETS
- ap_log_perror(APLOG_MARK, LOG_LEVEL, 0, to->p, APLOGNO(03206)
- "h2_util_move: %s, moving FILE bucket %ld-%ld "
- "from=%lx(p=%lx) to=%lx(p=%lx), setaside=%d",
- msg, (long)b->start, (long)b->length,
- (long)from, (long)from->p,
- (long)to, (long)to->p, setaside);
-#endif
- if (setaside) {
- status = apr_file_setaside(&fd, fd, to->p);
- if (status != APR_SUCCESS) {
- ap_log_perror(APLOG_MARK, APLOG_ERR, status, to->p,
- APLOGNO(02947) "h2_util: %s, setaside FILE",
- msg);
- return status;
- }
+ else {
+ if (b->length == ((apr_size_t)-1)) {
+ const char *ign;
+ apr_size_t ilen;
+ status = apr_bucket_read(b, &ign, &ilen, APR_BLOCK_READ);
+ if (status != APR_SUCCESS) {
+ return status;
}
- nb = apr_brigade_insert_file(to, fd, b->start, b->length,
- to->p);
-#if APR_HAS_MMAP
- apr_bucket_file_enable_mmap(nb, 0);
-#endif
- --(*pfile_buckets_allowed);
}
- else {
- const char *data;
- apr_size_t len;
-
- status = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);
- if (status == APR_SUCCESS && len > 0) {
- status = apr_brigade_write(to, NULL, NULL, data, len);
-#if LOG_BUCKETS
- ap_log_perror(APLOG_MARK, LOG_LEVEL, 0, to->p, APLOGNO(03207)
- "h2_util_move: %s, copied bucket %ld-%ld "
- "from=%lx(p=%lx) to=%lx(p=%lx)",
- msg, (long)b->start, (long)b->length,
- (long)from, (long)from->p,
- (long)to, (long)to->p);
-#endif
- }
+
+ if (remain < b->length) {
+ apr_bucket_split(b, remain);
}
- apr_bucket_delete(b);
- }
- else {
- apr_bucket_setaside(b, to->p);
- APR_BUCKET_REMOVE(b);
- APR_BRIGADE_INSERT_TAIL(to, b);
-#if LOG_BUCKETS
- ap_log_perror(APLOG_MARK, LOG_LEVEL, 0, to->p, APLOGNO(03208)
- "h2_util_move: %s, passed setaside bucket %ld-%ld "
- "from=%lx(p=%lx) to=%lx(p=%lx)",
- msg, (long)b->start, (long)b->length,
- (long)from, (long)from->p,
- (long)to, (long)to->p);
-#endif
}
}
+ APR_BUCKET_REMOVE(b);
+ APR_BRIGADE_INSERT_TAIL(dest, b);
+ remain -= b->length;
}
-
return status;
}
-apr_status_t h2_util_copy(apr_bucket_brigade *to, apr_bucket_brigade *from,
- apr_off_t maxlen, const char *msg)
+apr_status_t h2_brigade_copy_length(apr_bucket_brigade *dest,
+ apr_bucket_brigade *src,
+ apr_off_t length)
{
+ apr_bucket *b, *next;
+ apr_off_t remain = length;
apr_status_t status = APR_SUCCESS;
- int same_alloc;
-
- (void)msg;
- AP_DEBUG_ASSERT(to);
- AP_DEBUG_ASSERT(from);
- same_alloc = (to->bucket_alloc == from->bucket_alloc);
-
- if (!APR_BRIGADE_EMPTY(from)) {
- apr_bucket *b, *end, *cpy;
+
+ for (b = APR_BRIGADE_FIRST(src);
+ b != APR_BRIGADE_SENTINEL(src);
+ b = next) {
+ next = APR_BUCKET_NEXT(b);
- status = last_not_included(from, maxlen, 0, 0, &end);
- if (status != APR_SUCCESS) {
- return status;
+ if (APR_BUCKET_IS_METADATA(b)) {
+ /* fall through */
}
-
- for (b = APR_BRIGADE_FIRST(from);
- b != APR_BRIGADE_SENTINEL(from) && b != end;
- b = APR_BUCKET_NEXT(b))
- {
- if (same_alloc) {
- status = apr_bucket_copy(b, &cpy);
- if (status != APR_SUCCESS) {
- break;
- }
- APR_BRIGADE_INSERT_TAIL(to, cpy);
+ else {
+ if (remain == b->length) {
+ /* fall through */
+ }
+ else if (remain <= 0) {
+ return status;
}
else {
- if (APR_BUCKET_IS_METADATA(b)) {
- if (APR_BUCKET_IS_EOS(b)) {
- APR_BRIGADE_INSERT_TAIL(to, apr_bucket_eos_create(to->bucket_alloc));
- }
- else if (APR_BUCKET_IS_FLUSH(b)) {
- APR_BRIGADE_INSERT_TAIL(to, apr_bucket_flush_create(to->bucket_alloc));
- }
- else {
- /* ignore */
+ if (b->length == ((apr_size_t)-1)) {
+ const char *ign;
+ apr_size_t ilen;
+ status = apr_bucket_read(b, &ign, &ilen, APR_BLOCK_READ);
+ if (status != APR_SUCCESS) {
+ return status;
}
}
- else {
- const char *data;
- apr_size_t len;
- status = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);
- if (status == APR_SUCCESS && len > 0) {
- status = apr_brigade_write(to, NULL, NULL, data, len);
-#if LOG_BUCKETS
- ap_log_perror(APLOG_MARK, LOG_LEVEL, 0, to->p, APLOGNO(03209)
- "h2_util_copy: %s, copied bucket %ld-%ld "
- "from=%lx(p=%lx) to=%lx(p=%lx)",
- msg, (long)b->start, (long)b->length,
- (long)from, (long)from->p,
- (long)to, (long)to->p);
-#endif
- }
+
+ if (remain < b->length) {
+ apr_bucket_split(b, remain);
}
}
}
+ status = apr_bucket_copy(b, &b);
+ if (status != APR_SUCCESS) {
+ return status;
+ }
+ APR_BRIGADE_INSERT_TAIL(dest, b);
+ remain -= b->length;
}
return status;
}
@@ -669,25 +847,6 @@ int h2_util_bb_has_data(apr_bucket_briga
return 0;
}
-int h2_util_bb_has_data_or_eos(apr_bucket_brigade *bb)
-{
- apr_bucket *b;
- for (b = APR_BRIGADE_FIRST(bb);
- b != APR_BRIGADE_SENTINEL(bb);
- b = APR_BUCKET_NEXT(b))
- {
- if (APR_BUCKET_IS_METADATA(b)) {
- if (APR_BUCKET_IS_EOS(b)) {
- return 1;
- }
- }
- else {
- return 1;
- }
- }
- return 0;
-}
-
apr_status_t h2_util_bb_avail(apr_bucket_brigade *bb,
apr_off_t *plen, int *peos)
{
@@ -792,186 +951,102 @@ apr_status_t h2_util_bb_readx(apr_bucket
return status;
}
-void h2_util_bb_log(conn_rec *c, int stream_id, int level,
- const char *tag, apr_bucket_brigade *bb)
+apr_size_t h2_util_bucket_print(char *buffer, apr_size_t bmax,
+ apr_bucket *b, const char *sep)
{
- char buffer[16 * 1024];
- const char *line = "(null)";
- apr_size_t bmax = sizeof(buffer)/sizeof(buffer[0]);
- int off = 0;
- apr_bucket *b;
+ apr_size_t off = 0;
+ if (sep && *sep) {
+ off += apr_snprintf(buffer+off, bmax-off, "%s", sep);
+ }
- if (bb) {
- memset(buffer, 0, bmax--);
- for (b = APR_BRIGADE_FIRST(bb);
- bmax && (b != APR_BRIGADE_SENTINEL(bb));
- b = APR_BUCKET_NEXT(b)) {
-
- if (APR_BUCKET_IS_METADATA(b)) {
- if (APR_BUCKET_IS_EOS(b)) {
- off += apr_snprintf(buffer+off, bmax-off, "eos ");
- }
- else if (APR_BUCKET_IS_FLUSH(b)) {
- off += apr_snprintf(buffer+off, bmax-off, "flush ");
- }
- else if (AP_BUCKET_IS_EOR(b)) {
- off += apr_snprintf(buffer+off, bmax-off, "eor ");
- }
- else {
- off += apr_snprintf(buffer+off, bmax-off, "meta(unknown) ");
- }
- }
- else {
- const char *btype = "data";
- if (APR_BUCKET_IS_FILE(b)) {
- btype = "file";
- }
- else if (APR_BUCKET_IS_PIPE(b)) {
- btype = "pipe";
- }
- else if (APR_BUCKET_IS_SOCKET(b)) {
- btype = "socket";
- }
- else if (APR_BUCKET_IS_HEAP(b)) {
- btype = "heap";
- }
- else if (APR_BUCKET_IS_TRANSIENT(b)) {
- btype = "transient";
- }
- else if (APR_BUCKET_IS_IMMORTAL(b)) {
- btype = "immortal";
- }
-#if APR_HAS_MMAP
- else if (APR_BUCKET_IS_MMAP(b)) {
- btype = "mmap";
- }
-#endif
- else if (APR_BUCKET_IS_POOL(b)) {
- btype = "pool";
- }
-
- off += apr_snprintf(buffer+off, bmax-off, "%s[%ld] ",
- btype,
- (long)(b->length == ((apr_size_t)-1)?
- -1 : b->length));
- }
+ if (APR_BUCKET_IS_METADATA(b)) {
+ if (APR_BUCKET_IS_EOS(b)) {
+ off += apr_snprintf(buffer+off, bmax-off, "eos");
+ }
+ else if (APR_BUCKET_IS_FLUSH(b)) {
+ off += apr_snprintf(buffer+off, bmax-off, "flush");
+ }
+ else if (AP_BUCKET_IS_EOR(b)) {
+ off += apr_snprintf(buffer+off, bmax-off, "eor");
+ }
+ else {
+ off += apr_snprintf(buffer+off, bmax-off, "meta(unknown)");
}
- line = *buffer? buffer : "(empty)";
}
- /* Intentional no APLOGNO */
- ap_log_cerror(APLOG_MARK, level, 0, c, "bb_dump(%ld-%d)-%s: %s",
- c->id, stream_id, tag, line);
-
-}
-
-apr_status_t h2_ltransfer_brigade(apr_bucket_brigade *to,
- apr_bucket_brigade *from,
- apr_pool_t *p,
- apr_off_t *plen,
- int *peos)
-{
- apr_bucket *e;
- apr_off_t len = 0, remain = *plen;
- apr_status_t rv;
-
- *peos = 0;
-
- while (!APR_BRIGADE_EMPTY(from)) {
- e = APR_BRIGADE_FIRST(from);
-
- if (APR_BUCKET_IS_METADATA(e)) {
- if (APR_BUCKET_IS_EOS(e)) {
- *peos = 1;
- }
+ else {
+ const char *btype = "data";
+ if (APR_BUCKET_IS_FILE(b)) {
+ btype = "file";
}
- else {
- if (remain > 0 && e->length == ((apr_size_t)-1)) {
- const char *ign;
- apr_size_t ilen;
- rv = apr_bucket_read(e, &ign, &ilen, APR_BLOCK_READ);
- if (rv != APR_SUCCESS) {
- return rv;
- }
- }
-
- if (remain < e->length) {
- if (remain <= 0) {
- return APR_SUCCESS;
- }
- apr_bucket_split(e, remain);
- }
+ else if (APR_BUCKET_IS_PIPE(b)) {
+ btype = "pipe";
}
-
- rv = apr_bucket_setaside(e, p);
-
- /* If the bucket type does not implement setaside, then
- * (hopefully) morph it into a bucket type which does, and set
- * *that* aside... */
- if (rv == APR_ENOTIMPL) {
- const char *s;
- apr_size_t n;
-
- rv = apr_bucket_read(e, &s, &n, APR_BLOCK_READ);
- if (rv == APR_SUCCESS) {
- rv = apr_bucket_setaside(e, p);
- }
+ else if (APR_BUCKET_IS_SOCKET(b)) {
+ btype = "socket";
}
-
- if (rv != APR_SUCCESS) {
- /* Return an error but still save the brigade if
- * ->setaside() is really not implemented. */
- if (rv != APR_ENOTIMPL) {
- return rv;
- }
+ else if (APR_BUCKET_IS_HEAP(b)) {
+ btype = "heap";
+ }
+ else if (APR_BUCKET_IS_TRANSIENT(b)) {
+ btype = "transient";
+ }
+ else if (APR_BUCKET_IS_IMMORTAL(b)) {
+ btype = "immortal";
+ }
+#if APR_HAS_MMAP
+ else if (APR_BUCKET_IS_MMAP(b)) {
+ btype = "mmap";
+ }
+#endif
+ else if (APR_BUCKET_IS_POOL(b)) {
+ btype = "pool";
}
- APR_BUCKET_REMOVE(e);
- APR_BRIGADE_INSERT_TAIL(to, e);
- len += e->length;
- remain -= e->length;
+ off += apr_snprintf(buffer+off, bmax-off, "%s[%ld]",
+ btype,
+ (long)(b->length == ((apr_size_t)-1)?
+ -1 : b->length));
}
-
- *plen = len;
- return APR_SUCCESS;
+ return off;
}
-apr_status_t h2_transfer_brigade(apr_bucket_brigade *to,
- apr_bucket_brigade *from,
- apr_pool_t *p)
+apr_size_t h2_util_bb_print(char *buffer, apr_size_t bmax,
+ const char *tag, const char *sep,
+ apr_bucket_brigade *bb)
{
- apr_bucket *e;
- apr_status_t rv;
-
- while (!APR_BRIGADE_EMPTY(from)) {
- e = APR_BRIGADE_FIRST(from);
-
- rv = apr_bucket_setaside(e, p);
-
- /* If the bucket type does not implement setaside, then
- * (hopefully) morph it into a bucket type which does, and set
- * *that* aside... */
- if (rv == APR_ENOTIMPL) {
- const char *s;
- apr_size_t n;
+ apr_size_t off = 0;
+ const char *sp = "";
+ apr_bucket *b;
+
+ if (bb) {
+ memset(buffer, 0, bmax--);
+ off += apr_snprintf(buffer+off, bmax-off, "%s(", tag);
+ for (b = APR_BRIGADE_FIRST(bb);
+ bmax && (b != APR_BRIGADE_SENTINEL(bb));
+ b = APR_BUCKET_NEXT(b)) {
- rv = apr_bucket_read(e, &s, &n, APR_BLOCK_READ);
- if (rv == APR_SUCCESS) {
- rv = apr_bucket_setaside(e, p);
- }
+ off += h2_util_bucket_print(buffer+off, bmax-off, b, sp);
+ sp = " ";
}
-
- if (rv != APR_SUCCESS) {
- /* Return an error but still save the brigade if
- * ->setaside() is really not implemented. */
- if (rv != APR_ENOTIMPL) {
- return rv;
- }
- }
-
- APR_BUCKET_REMOVE(e);
- APR_BRIGADE_INSERT_TAIL(to, e);
+ off += apr_snprintf(buffer+off, bmax-off, ")%s", sep);
}
- return APR_SUCCESS;
+ else {
+ off += apr_snprintf(buffer+off, bmax-off, "%s(null)%s", tag, sep);
+ }
+ return off;
+}
+
+void h2_util_bb_log(conn_rec *c, int stream_id, int level,
+ const char *tag, apr_bucket_brigade *bb)
+{
+ char buffer[4 * 1024];
+ const char *line = "(null)";
+ apr_size_t len, bmax = sizeof(buffer)/sizeof(buffer[0]);
+
+ len = h2_util_bb_print(buffer, bmax, tag, "", bb);
+ /* Intentional no APLOGNO */
+ ap_log_cerror(APLOG_MARK, level, 0, c, "bb_dump(%ld-%d): %s",
+ c->id, stream_id, len? buffer : line);
}
apr_status_t h2_append_brigade(apr_bucket_brigade *to,
@@ -1355,3 +1430,123 @@ void h2_push_policy_determine(struct h2_
req->push_policy = policy;
}
+/*******************************************************************************
+ * ap_casecmpstr, when will it be backported?
+ ******************************************************************************/
+#if !APR_CHARSET_EBCDIC
+/*
+ * Provide our own known-fast implementation of str[n]casecmp()
+ * NOTE: Only ASCII alpha characters 41-5A are folded to 61-7A,
+ * other 8-bit latin alphabetics are never case-folded!
+ */
+static const unsigned char ucharmap[] = {
+ 0x0, 0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7,
+ 0x8, 0x9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf,
+ 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17,
+ 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+ 0x20, 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27,
+ 0x28, 0x29, 0x2a, 0x2b, 0x2c, 0x2d, 0x2e, 0x2f,
+ 0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37,
+ 0x38, 0x39, 0x3a, 0x3b, 0x3c, 0x3d, 0x3e, 0x3f,
+ 0x40, 'a', 'b', 'c', 'd', 'e', 'f', 'g',
+ 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o',
+ 'p', 'q', 'r', 's', 't', 'u', 'v', 'w',
+ 'x', 'y', 'z', 0x5b, 0x5c, 0x5d, 0x5e, 0x5f,
+ 0x60, 'a', 'b', 'c', 'd', 'e', 'f', 'g',
+ 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o',
+ 'p', 'q', 'r', 's', 't', 'u', 'v', 'w',
+ 'x', 'y', 'z', 0x7b, 0x7c, 0x7d, 0x7e, 0x7f,
+ 0x80, 0x81, 0x82, 0x83, 0x84, 0x85, 0x86, 0x87,
+ 0x88, 0x89, 0x8a, 0x8b, 0x8c, 0x8d, 0x8e, 0x8f,
+ 0x90, 0x91, 0x92, 0x93, 0x94, 0x95, 0x96, 0x97,
+ 0x98, 0x99, 0x9a, 0x9b, 0x9c, 0x9d, 0x9e, 0x9f,
+ 0xa0, 0xa1, 0xa2, 0xa3, 0xa4, 0xa5, 0xa6, 0xa7,
+ 0xa8, 0xa9, 0xaa, 0xab, 0xac, 0xad, 0xae, 0xaf,
+ 0xb0, 0xb1, 0xb2, 0xb3, 0xb4, 0xb5, 0xb6, 0xb7,
+ 0xb8, 0xb9, 0xba, 0xbb, 0xbc, 0xbd, 0xbe, 0xbf,
+ 0xc0, 0xc1, 0xc2, 0xc3, 0xc4, 0xc5, 0xc6, 0xc7,
+ 0xc8, 0xc9, 0xca, 0xcb, 0xcc, 0xcd, 0xce, 0xcf,
+ 0xd0, 0xd1, 0xd2, 0xd3, 0xd4, 0xd5, 0xd6, 0xd7,
+ 0xd8, 0xd9, 0xda, 0xdb, 0xdc, 0xdd, 0xde, 0xdf,
+ 0xe0, 0xe1, 0xe2, 0xe3, 0xe4, 0xe5, 0xe6, 0xe7,
+ 0xe8, 0xe9, 0xea, 0xeb, 0xec, 0xed, 0xee, 0xef,
+ 0xf0, 0xf1, 0xf2, 0xf3, 0xf4, 0xf5, 0xf6, 0xf7,
+ 0xf8, 0xf9, 0xfa, 0xfb, 0xfc, 0xfd, 0xfe, 0xff
+};
+
+#else /* APR_CHARSET_EBCDIC */
+/* Derived from apr-iconv/ccs/cp037.c for EBCDIC case comparison,
+ provides unique identity of every char value (strict ISO-646
+ conformance, arbitrary election of an ISO-8859-1 ordering, and
+ very arbitrary control code assignments into C1 to achieve
+ identity and a reversible mapping of code points),
+ then folding the equivalences of ASCII 41-5A into 61-7A,
+ presenting comparison results in a somewhat ISO/IEC 10646
+ (ASCII-like) order, depending on the EBCDIC code page in use.
+ */
+static const unsigned char ucharmap[] = {
+ 0x00, 0x01, 0x02, 0x03, 0x9C, 0x09, 0x86, 0x7F,
+ 0x97, 0x8D, 0x8E, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F,
+ 0x10, 0x11, 0x12, 0x13, 0x9D, 0x85, 0x08, 0x87,
+ 0x18, 0x19, 0x92, 0x8F, 0x1C, 0x1D, 0x1E, 0x1F,
+ 0x80, 0x81, 0x82, 0x83, 0x84, 0x0A, 0x17, 0x1B,
+ 0x88, 0x89, 0x8A, 0x8B, 0x8C, 0x05, 0x06, 0x07,
+ 0x90, 0x91, 0x16, 0x93, 0x94, 0x95, 0x96, 0x04,
+ 0x98, 0x99, 0x9A, 0x9B, 0x14, 0x15, 0x9E, 0x1A,
+ 0x20, 0xA0, 0xE2, 0xE4, 0xE0, 0xE1, 0xE3, 0xE5,
+ 0xE7, 0xF1, 0xA2, 0x2E, 0x3C, 0x28, 0x2B, 0x7C,
+ 0x26, 0xE9, 0xEA, 0xEB, 0xE8, 0xED, 0xEE, 0xEF,
+ 0xEC, 0xDF, 0x21, 0x24, 0x2A, 0x29, 0x3B, 0xAC,
+ 0x2D, 0x2F, 0xC2, 0xC4, 0xC0, 0xC1, 0xC3, 0xC5,
+ 0xC7, 0xD1, 0xA6, 0x2C, 0x25, 0x5F, 0x3E, 0x3F,
+ 0xF8, 0xC9, 0xCA, 0xCB, 0xC8, 0xCD, 0xCE, 0xCF,
+ 0xCC, 0x60, 0x3A, 0x23, 0x40, 0x27, 0x3D, 0x22,
+ 0xD8, 0x61, 0x62, 0x63, 0x64, 0x65, 0x66, 0x67,
+ 0x68, 0x69, 0xAB, 0xBB, 0xF0, 0xFD, 0xFE, 0xB1,
+ 0xB0, 0x6A, 0x6B, 0x6C, 0x6D, 0x6E, 0x6F, 0x70,
+ 0x71, 0x72, 0xAA, 0xBA, 0xE6, 0xB8, 0xC6, 0xA4,
+ 0xB5, 0x7E, 0x73, 0x74, 0x75, 0x76, 0x77, 0x78,
+ 0x79, 0x7A, 0xA1, 0xBF, 0xD0, 0xDD, 0xDE, 0xAE,
+ 0x5E, 0xA3, 0xA5, 0xB7, 0xA9, 0xA7, 0xB6, 0xBC,
+ 0xBD, 0xBE, 0x5B, 0x5D, 0xAF, 0xA8, 0xB4, 0xD7,
+ 0x7B, 0x61, 0x62, 0x63, 0x64, 0x65, 0x66, 0x67,
+ 0x68, 0x69, 0xAD, 0xF4, 0xF6, 0xF2, 0xF3, 0xF5,
+ 0x7D, 0x6A, 0x6B, 0x6C, 0x6D, 0x6E, 0x6F, 0x70,
+ 0x71, 0x72, 0xB9, 0xFB, 0xFC, 0xF9, 0xFA, 0xFF,
+ 0x5C, 0xF7, 0x73, 0x74, 0x75, 0x76, 0x77, 0x78,
+ 0x79, 0x7A, 0xB2, 0xD4, 0xD6, 0xD2, 0xD3, 0xD5,
+ 0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37,
+ 0x38, 0x39, 0xB3, 0xDB, 0xDC, 0xD9, 0xDA, 0x9F
+};
+#endif
+
+int h2_casecmpstr(const char *s1, const char *s2)
+{
+ const unsigned char *ps1 = (const unsigned char *) s1;
+ const unsigned char *ps2 = (const unsigned char *) s2;
+
+ while (ucharmap[*ps1] == ucharmap[*ps2]) {
+ if (*ps1++ == '\0') {
+ return (0);
+ }
+ ps2++;
+ }
+ return (ucharmap[*ps1] - ucharmap[*ps2]);
+}
+
+int h2_casecmpstrn(const char *s1, const char *s2, apr_size_t n)
+{
+ const unsigned char *ps1 = (const unsigned char *) s1;
+ const unsigned char *ps2 = (const unsigned char *) s2;
+ while (n--) {
+ if (ucharmap[*ps1] != ucharmap[*ps2]) {
+ return (ucharmap[*ps1] - ucharmap[*ps2]);
+ }
+ if (*ps1++ == '\0') {
+ break;
+ }
+ ps2++;
+ }
+ return (0);
+}
+