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 2015/11/20 16:13:11 UTC
svn commit: r1715371 [5/6] - in /httpd/httpd/branches/2.4.x: ./
docs/manual/mod/ 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=1715371&r1=1715370&r2=1715371&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 Fri Nov 20 15:13:11 2015
@@ -16,8 +16,6 @@
#include <assert.h>
#include <stddef.h>
-#define APR_POOL_DEBUG 7
-
#include <httpd.h>
#include <http_core.h>
#include <http_connection.h>
@@ -27,9 +25,12 @@
#include "h2_private.h"
#include "h2_conn.h"
+#include "h2_h2.h"
#include "h2_mplx.h"
+#include "h2_push.h"
#include "h2_request.h"
#include "h2_response.h"
+#include "h2_session.h"
#include "h2_stream.h"
#include "h2_task.h"
#include "h2_ctx.h"
@@ -38,57 +39,153 @@
#include "h2_util.h"
-static void set_state(h2_stream *stream, h2_stream_state_t state)
+#define H2_STREAM_OUT(lvl,s,msg) \
+ do { \
+ if (APLOG_C_IS_LEVEL((s)->session->c,lvl)) \
+ h2_util_bb_log((s)->session->c,(s)->id,lvl,msg,(s)->bbout); \
+ } while(0)
+#define H2_STREAM_IN(lvl,s,msg) \
+ do { \
+ if (APLOG_C_IS_LEVEL((s)->session->c,lvl)) \
+ h2_util_bb_log((s)->session->c,(s)->id,lvl,msg,(s)->bbin); \
+ } while(0)
+
+
+static int state_transition[][7] = {
+ /* ID OP RL RR CI CO CL */
+/*ID*/{ 1, 0, 0, 0, 0, 0, 0 },
+/*OP*/{ 1, 1, 0, 0, 0, 0, 0 },
+/*RL*/{ 0, 0, 1, 0, 0, 0, 0 },
+/*RR*/{ 0, 0, 0, 1, 0, 0, 0 },
+/*CI*/{ 1, 1, 0, 0, 1, 0, 0 },
+/*CO*/{ 1, 1, 0, 0, 0, 1, 0 },
+/*CL*/{ 1, 1, 0, 0, 1, 1, 1 },
+};
+
+static int set_state(h2_stream *stream, h2_stream_state_t state)
{
- AP_DEBUG_ASSERT(stream);
- if (stream->state != state) {
+ int allowed = state_transition[state][stream->state];
+ if (allowed) {
stream->state = state;
+ return 1;
}
+
+ ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c,
+ "h2_stream(%ld-%d): invalid state transition from %d to %d",
+ stream->session->id, stream->id, stream->state, state);
+ return 0;
}
-h2_stream *h2_stream_create(int id, apr_pool_t *pool, struct h2_mplx *m)
+static int close_input(h2_stream *stream)
{
- h2_stream *stream = apr_pcalloc(pool, sizeof(h2_stream));
- if (stream != NULL) {
- stream->id = id;
- stream->state = H2_STREAM_ST_IDLE;
- stream->pool = pool;
- stream->m = m;
- stream->request = h2_request_create(id, pool, m->c->bucket_alloc);
- ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c,
- "h2_stream(%ld-%d): created", m->id, stream->id);
+ switch (stream->state) {
+ case H2_STREAM_ST_CLOSED_INPUT:
+ case H2_STREAM_ST_CLOSED:
+ return 0; /* ignore, idempotent */
+ case H2_STREAM_ST_CLOSED_OUTPUT:
+ /* both closed now */
+ set_state(stream, H2_STREAM_ST_CLOSED);
+ break;
+ default:
+ /* everything else we jump to here */
+ set_state(stream, H2_STREAM_ST_CLOSED_INPUT);
+ break;
}
- return stream;
+ return 1;
}
-static void h2_stream_cleanup(h2_stream *stream)
+static int input_closed(h2_stream *stream)
{
- if (stream->request) {
- h2_request_destroy(stream->request);
- stream->request = NULL;
+ switch (stream->state) {
+ case H2_STREAM_ST_OPEN:
+ case H2_STREAM_ST_CLOSED_OUTPUT:
+ return 0;
+ default:
+ return 1;
}
}
+static int close_output(h2_stream *stream)
+{
+ switch (stream->state) {
+ case H2_STREAM_ST_CLOSED_OUTPUT:
+ case H2_STREAM_ST_CLOSED:
+ return 0; /* ignore, idempotent */
+ case H2_STREAM_ST_CLOSED_INPUT:
+ /* both closed now */
+ set_state(stream, H2_STREAM_ST_CLOSED);
+ break;
+ default:
+ /* everything else we jump to here */
+ set_state(stream, H2_STREAM_ST_CLOSED_OUTPUT);
+ break;
+ }
+ return 1;
+}
+
+static int input_open(h2_stream *stream)
+{
+ switch (stream->state) {
+ case H2_STREAM_ST_OPEN:
+ case H2_STREAM_ST_CLOSED_OUTPUT:
+ return 1;
+ default:
+ return 0;
+ }
+}
+
+static int output_open(h2_stream *stream)
+{
+ switch (stream->state) {
+ case H2_STREAM_ST_OPEN:
+ case H2_STREAM_ST_CLOSED_INPUT:
+ return 1;
+ default:
+ return 0;
+ }
+}
+
+h2_stream *h2_stream_create(int id, apr_pool_t *pool, h2_session *session)
+{
+ h2_stream *stream = apr_pcalloc(pool, sizeof(h2_stream));
+ stream->id = id;
+ stream->state = H2_STREAM_ST_IDLE;
+ stream->pool = pool;
+ stream->session = session;
+ return stream;
+}
+
+h2_stream *h2_stream_open(int id, apr_pool_t *pool, h2_session *session)
+{
+ h2_stream *stream = h2_stream_create(id, pool, session);
+ set_state(stream, H2_STREAM_ST_OPEN);
+ stream->request = h2_request_create(id, pool);
+ stream->bbout = apr_brigade_create(stream->pool,
+ stream->session->c->bucket_alloc);
+
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+ "h2_stream(%ld-%d): opened", session->id, stream->id);
+ return stream;
+}
+
apr_status_t h2_stream_destroy(h2_stream *stream)
{
AP_DEBUG_ASSERT(stream);
- ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->m->c,
- "h2_stream(%ld-%d): destroy", stream->m->id, stream->id);
- h2_stream_cleanup(stream);
-
- if (stream->task) {
- h2_task_destroy(stream->task);
- stream->task = NULL;
+ if (stream->request) {
+ h2_request_destroy(stream->request);
+ stream->request = NULL;
}
+
if (stream->pool) {
apr_pool_destroy(stream->pool);
}
return APR_SUCCESS;
}
-void h2_stream_attach_pool(h2_stream *stream, apr_pool_t *pool)
+void h2_stream_cleanup(h2_stream *stream)
{
- stream->pool = pool;
+ h2_session_stream_destroy(stream->session, stream);
+ /* stream is gone */
}
apr_pool_t *h2_stream_detach_pool(h2_stream *stream)
@@ -98,171 +195,402 @@ apr_pool_t *h2_stream_detach_pool(h2_str
return pool;
}
-void h2_stream_abort(h2_stream *stream)
+void h2_stream_rst(h2_stream *stream, int error_code)
{
- AP_DEBUG_ASSERT(stream);
- stream->aborted = 1;
+ stream->rst_error = error_code;
+ close_input(stream);
+ close_output(stream);
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
+ "h2_stream(%ld-%d): reset, error=%d",
+ stream->session->id, stream->id, error_code);
}
apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
apr_bucket_brigade *bb)
{
+ apr_status_t status = APR_SUCCESS;
+ if (!output_open(stream)) {
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
+ "h2_stream(%ld-%d): output closed",
+ stream->session->id, stream->id);
+ return APR_ECONNRESET;
+ }
+
stream->response = response;
if (bb && !APR_BRIGADE_EMPTY(bb)) {
- if (!stream->bbout) {
- stream->bbout = apr_brigade_create(stream->pool,
- stream->m->c->bucket_alloc);
- }
- return h2_util_move(stream->bbout, bb, 16 * 1024, NULL,
- "h2_stream_set_response");
+ int move_all = INT_MAX;
+ /* we can move file handles from h2_mplx into this h2_stream as many
+ * as we want, since the lifetimes are the same and we are not freeing
+ * the ones in h2_mplx->io before this stream is done. */
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream set_response_pre");
+ status = h2_util_move(stream->bbout, bb, -1, &move_all,
+ "h2_stream_set_response");
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream set_response_post");
}
- return APR_SUCCESS;
+
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, stream->session->c,
+ "h2_stream(%ld-%d): set_response(%d)",
+ stream->session->id, stream->id, response->http_status);
+ return status;
}
-static int set_closed(h2_stream *stream)
+apr_status_t h2_stream_set_request(h2_stream *stream, request_rec *r)
{
- switch (stream->state) {
- case H2_STREAM_ST_CLOSED_INPUT:
- case H2_STREAM_ST_CLOSED:
- return 0; /* ignore, idempotent */
- case H2_STREAM_ST_CLOSED_OUTPUT:
- /* both closed now */
- set_state(stream, H2_STREAM_ST_CLOSED);
- break;
- default:
- /* everything else we jump to here */
- set_state(stream, H2_STREAM_ST_CLOSED_INPUT);
- break;
+ apr_status_t status;
+ AP_DEBUG_ASSERT(stream);
+ if (stream->rst_error) {
+ return APR_ECONNRESET;
}
- return 1;
+ set_state(stream, H2_STREAM_ST_OPEN);
+ status = h2_request_rwrite(stream->request, r);
+ return status;
}
-apr_status_t h2_stream_rwrite(h2_stream *stream, request_rec *r)
+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->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)
{
- apr_status_t status;
AP_DEBUG_ASSERT(stream);
- set_state(stream, H2_STREAM_ST_OPEN);
- status = h2_request_rwrite(stream->request, r, stream->m);
- return status;
+ if (h2_stream_is_scheduled(stream)) {
+ return h2_request_add_trailer(stream->request, stream->pool,
+ name, nlen, value, vlen);
+ }
+ else {
+ if (!input_open(stream)) {
+ return APR_ECONNRESET;
+ }
+ return h2_request_add_header(stream->request, stream->pool,
+ name, nlen, value, vlen);
+ }
}
-apr_status_t h2_stream_write_eoh(h2_stream *stream, int eos)
+apr_status_t h2_stream_schedule(h2_stream *stream, int eos,
+ h2_stream_pri_cmp *cmp, void *ctx)
{
apr_status_t status;
AP_DEBUG_ASSERT(stream);
+ AP_DEBUG_ASSERT(stream->session);
+ AP_DEBUG_ASSERT(stream->session->mplx);
+
+ if (!output_open(stream)) {
+ return APR_ECONNRESET;
+ }
+ if (stream->scheduled) {
+ return APR_EINVAL;
+ }
+ if (eos) {
+ close_input(stream);
+ }
/* Seeing the end-of-headers, we have everything we need to
* start processing it.
*/
- status = h2_mplx_create_task(stream->m, stream);
+ status = h2_request_end_headers(stream->request, stream->pool, eos);
if (status == APR_SUCCESS) {
- status = h2_request_end_headers(stream->request,
- stream->m, stream->task, eos);
- if (status == APR_SUCCESS) {
- status = h2_mplx_do_task(stream->m, stream->task);
- }
- if (eos) {
- status = h2_stream_write_eos(stream);
+ if (!eos) {
+ stream->bbin = apr_brigade_create(stream->pool,
+ stream->session->c->bucket_alloc);
}
- ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, stream->m->c,
- "h2_mplx(%ld-%d): start stream, task %s %s (%s)",
- stream->m->id, stream->id,
- stream->request->method, stream->request->path,
- stream->request->authority);
+ stream->input_remaining = stream->request->content_length;
+ status = h2_mplx_process(stream->session->mplx, stream->id,
+ stream->request, eos, cmp, ctx);
+ stream->scheduled = 1;
+
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
+ "h2_stream(%ld-%d): scheduled %s %s://%s%s",
+ stream->session->id, stream->id,
+ stream->request->method, stream->request->scheme,
+ stream->request->authority, stream->request->path);
}
+ else {
+ h2_stream_rst(stream, H2_ERR_INTERNAL_ERROR);
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, 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,
+ stream->request->authority, stream->request->path);
+ }
+
return status;
}
-apr_status_t h2_stream_write_eos(h2_stream *stream)
+int h2_stream_is_scheduled(h2_stream *stream)
{
- AP_DEBUG_ASSERT(stream);
- ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->m->c,
- "h2_stream(%ld-%d): closing input",
- stream->m->id, stream->id);
- if (set_closed(stream)) {
- return h2_request_close(stream->request);
+ return stream->scheduled;
+}
+
+static apr_status_t h2_stream_input_flush(h2_stream *stream)
+{
+ apr_status_t status = APR_SUCCESS;
+ if (stream->bbin && !APR_BRIGADE_EMPTY(stream->bbin)) {
+
+ status = h2_mplx_in_write(stream->session->mplx, stream->id, stream->bbin);
+ if (status != APR_SUCCESS) {
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, stream->session->mplx->c,
+ "h2_stream(%ld-%d): flushing input data",
+ stream->session->id, stream->id);
+ }
}
- return APR_SUCCESS;
+ return status;
+}
+
+static apr_status_t input_flush(apr_bucket_brigade *bb, void *ctx)
+{
+ (void)bb;
+ return h2_stream_input_flush(ctx);
}
-apr_status_t h2_stream_write_header(h2_stream *stream,
- const char *name, size_t nlen,
- const char *value, size_t vlen)
+static apr_status_t input_add_data(h2_stream *stream,
+ const char *data, size_t len, int chunked)
{
+ apr_status_t status = APR_SUCCESS;
+
+ if (chunked) {
+ status = apr_brigade_printf(stream->bbin, input_flush, stream,
+ "%lx\r\n", (unsigned long)len);
+ if (status == APR_SUCCESS) {
+ status = apr_brigade_write(stream->bbin, input_flush, stream, data, len);
+ if (status == APR_SUCCESS) {
+ status = apr_brigade_puts(stream->bbin, input_flush, stream, "\r\n");
+ }
+ }
+ }
+ else {
+ status = apr_brigade_write(stream->bbin, input_flush, stream, data, len);
+ }
+ return status;
+}
+
+static int input_add_header(void *str, const char *key, const char *value)
+{
+ h2_stream *stream = str;
+ apr_status_t status = input_add_data(stream, key, strlen(key), 0);
+ if (status == APR_SUCCESS) {
+ status = input_add_data(stream, ": ", 2, 0);
+ if (status == APR_SUCCESS) {
+ status = input_add_data(stream, value, strlen(value), 0);
+ if (status == APR_SUCCESS) {
+ status = input_add_data(stream, "\r\n", 2, 0);
+ }
+ }
+ }
+ return (status == APR_SUCCESS);
+}
+
+apr_status_t h2_stream_close_input(h2_stream *stream)
+{
+ apr_status_t status = APR_SUCCESS;
+
AP_DEBUG_ASSERT(stream);
- switch (stream->state) {
- case H2_STREAM_ST_IDLE:
- set_state(stream, H2_STREAM_ST_OPEN);
- break;
- case H2_STREAM_ST_OPEN:
- break;
- default:
- return APR_EINVAL;
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
+ "h2_stream(%ld-%d): closing input",
+ stream->session->id, stream->id);
+
+ if (stream->rst_error) {
+ return APR_ECONNRESET;
+ }
+
+ H2_STREAM_IN(APLOG_TRACE2, stream, "close_pre");
+ if (close_input(stream) && stream->bbin) {
+ if (stream->request->chunked) {
+ apr_table_t *trailers = stream->request->trailers;
+ if (trailers && !apr_is_empty_table(trailers)) {
+ status = input_add_data(stream, "0\r\n", 3, 0);
+ apr_table_do(input_add_header, stream, trailers, NULL);
+ status = input_add_data(stream, "\r\n", 2, 0);
+ }
+ else {
+ status = input_add_data(stream, "0\r\n\r\n", 5, 0);
+ }
+ }
+
+ if (status == APR_SUCCESS) {
+ status = h2_stream_input_flush(stream);
+ }
+ if (status == APR_SUCCESS) {
+ status = h2_mplx_in_close(stream->session->mplx, stream->id);
+ }
}
- return h2_request_write_header(stream->request, name, nlen,
- value, vlen, stream->m);
+ H2_STREAM_IN(APLOG_TRACE2, stream, "close_post");
+ return status;
}
apr_status_t h2_stream_write_data(h2_stream *stream,
const char *data, size_t len)
{
+ apr_status_t status = APR_SUCCESS;
+
AP_DEBUG_ASSERT(stream);
- AP_DEBUG_ASSERT(stream);
- switch (stream->state) {
- case H2_STREAM_ST_OPEN:
- break;
- default:
- return APR_EINVAL;
+ if (input_closed(stream) || !stream->request->eoh || !stream->bbin) {
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
+ "h2_stream(%ld-%d): writing denied, closed=%d, eoh=%d, bbin=%d",
+ stream->session->id, stream->id, input_closed(stream),
+ stream->request->eoh, !!stream->bbin);
+ return APR_EINVAL;
+ }
+
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->session->c,
+ "h2_stream(%ld-%d): add %ld input bytes",
+ stream->session->id, stream->id, (long)len);
+
+ H2_STREAM_IN(APLOG_TRACE2, stream, "write_data_pre");
+ if (stream->request->chunked) {
+ /* if input may have a body and we have not seen any
+ * content-length header, we need to chunk the input data.
+ */
+ status = input_add_data(stream, data, len, 1);
}
- return h2_request_write_data(stream->request, data, len);
+ else {
+ stream->input_remaining -= len;
+ if (stream->input_remaining < 0) {
+ ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, stream->session->c,
+ APLOGNO(02961)
+ "h2_stream(%ld-%d): got %ld more content bytes than announced "
+ "in content-length header: %ld",
+ stream->session->id, stream->id,
+ (long)stream->request->content_length,
+ -(long)stream->input_remaining);
+ h2_stream_rst(stream, H2_ERR_PROTOCOL_ERROR);
+ return APR_ECONNABORTED;
+ }
+ status = input_add_data(stream, data, len, 0);
+ }
+ if (status == APR_SUCCESS) {
+ status = h2_stream_input_flush(stream);
+ }
+ H2_STREAM_IN(APLOG_TRACE2, stream, "write_data_post");
+ return status;
}
apr_status_t h2_stream_prep_read(h2_stream *stream,
- apr_size_t *plen, int *peos)
+ apr_off_t *plen, int *peos)
{
apr_status_t status = APR_SUCCESS;
const char *src;
+ int test_read = (*plen == 0);
- if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
+ if (stream->rst_error) {
+ return APR_ECONNRESET;
+ }
+
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream prep_read_pre");
+ if (!APR_BRIGADE_EMPTY(stream->bbout)) {
src = "stream";
status = h2_util_bb_avail(stream->bbout, plen, peos);
- if (status == APR_SUCCESS && !*peos && !*plen) {
+ if (!test_read && status == APR_SUCCESS && !*peos && !*plen) {
apr_brigade_cleanup(stream->bbout);
return h2_stream_prep_read(stream, plen, peos);
}
}
else {
src = "mplx";
- status = h2_mplx_out_readx(stream->m, stream->id,
+ status = h2_mplx_out_readx(stream->session->mplx, stream->id,
NULL, NULL, plen, peos);
}
- if (status == APR_SUCCESS && !*peos && !*plen) {
+ if (!test_read && status == APR_SUCCESS && !*peos && !*plen) {
status = APR_EAGAIN;
}
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->m->c,
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream prep_read_post");
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
"h2_stream(%ld-%d): prep_read %s, len=%ld eos=%d",
- stream->m->id, stream->id,
- src, (long)*plen, *peos);
+ stream->session->id, stream->id, src, (long)*plen, *peos);
return status;
}
apr_status_t h2_stream_readx(h2_stream *stream,
h2_io_data_cb *cb, void *ctx,
- apr_size_t *plen, int *peos)
+ apr_off_t *plen, int *peos)
{
- if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
- return h2_util_bb_readx(stream->bbout, cb, ctx, plen, peos);
+ apr_status_t status = APR_SUCCESS;
+ const char *src;
+
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream readx_pre");
+ if (stream->rst_error) {
+ return APR_ECONNRESET;
+ }
+ *peos = 0;
+ if (!APR_BRIGADE_EMPTY(stream->bbout)) {
+ apr_off_t origlen = *plen;
+
+ src = "stream";
+ status = h2_util_bb_readx(stream->bbout, cb, ctx, plen, peos);
+ if (status == APR_SUCCESS && !*peos && !*plen) {
+ apr_brigade_cleanup(stream->bbout);
+ *plen = origlen;
+ return h2_stream_readx(stream, cb, ctx, plen, peos);
+ }
+ }
+ else {
+ src = "mplx";
+ status = h2_mplx_out_readx(stream->session->mplx, stream->id,
+ cb, ctx, plen, peos);
+ }
+
+ if (status == APR_SUCCESS && !*peos && !*plen) {
+ status = APR_EAGAIN;
}
- return h2_mplx_out_readx(stream->m, stream->id,
- cb, ctx, plen, peos);
+
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream prep_readx_post");
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
+ "h2_stream(%ld-%d): readx %s, len=%ld eos=%d",
+ stream->session->id, stream->id, src, (long)*plen, *peos);
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream readx_post");
+
+ return status;
}
+apr_status_t h2_stream_read_to(h2_stream *stream, apr_bucket_brigade *bb,
+ apr_off_t *plen, int *peos)
+{
+ apr_status_t status = APR_SUCCESS;
+
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream read_to_pre");
+ if (stream->rst_error) {
+ return APR_ECONNRESET;
+ }
+
+ if (APR_BRIGADE_EMPTY(stream->bbout)) {
+ apr_off_t tlen = *plen;
+ int eos;
+ status = h2_mplx_out_read_to(stream->session->mplx, stream->id,
+ stream->bbout, &tlen, &eos);
+ }
+
+ if (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(stream->bbout)) {
+ status = h2_transfer_brigade(bb, stream->bbout, stream->pool,
+ plen, peos);
+ }
+ else {
+ *plen = 0;
+ *peos = 0;
+ }
+
+ if (status == APR_SUCCESS && !*peos && !*plen) {
+ status = APR_EAGAIN;
+ }
+ H2_STREAM_OUT(APLOG_TRACE2, stream, "h2_stream read_to_post");
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->session->c,
+ "h2_stream(%ld-%d): read_to, len=%ld eos=%d",
+ stream->session->id, stream->id, (long)*plen, *peos);
+ return status;
+}
void h2_stream_set_suspended(h2_stream *stream, int suspended)
{
AP_DEBUG_ASSERT(stream);
stream->suspended = !!suspended;
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c,
+ "h2_stream(%ld-%d): suspended=%d",
+ stream->session->id, stream->id, stream->suspended);
}
int h2_stream_is_suspended(h2_stream *stream)
@@ -271,3 +599,49 @@ int h2_stream_is_suspended(h2_stream *st
return stream->suspended;
}
+int h2_stream_input_is_open(h2_stream *stream)
+{
+ return input_open(stream);
+}
+
+int h2_stream_needs_submit(h2_stream *stream)
+{
+ switch (stream->state) {
+ case H2_STREAM_ST_OPEN:
+ case H2_STREAM_ST_CLOSED_INPUT:
+ case H2_STREAM_ST_CLOSED_OUTPUT:
+ case H2_STREAM_ST_CLOSED:
+ return !stream->submitted;
+ default:
+ return 0;
+ }
+}
+
+apr_status_t h2_stream_submit_pushes(h2_stream *stream)
+{
+ apr_status_t status = APR_SUCCESS;
+ apr_array_header_t *pushes;
+ int i;
+
+ pushes = h2_push_collect(stream->pool, stream->request, stream->response);
+ if (pushes && !apr_is_empty_array(pushes)) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, stream->session->c,
+ "h2_stream(%ld-%d): found %d push candidates",
+ stream->session->id, stream->id, pushes->nelts);
+ for (i = 0; i < pushes->nelts; ++i) {
+ h2_push *push = APR_ARRAY_IDX(pushes, i, h2_push*);
+ h2_stream *s = h2_session_push(stream->session, stream, push);
+ if (!s) {
+ status = APR_ECONNRESET;
+ break;
+ }
+ }
+ }
+ return status;
+}
+
+apr_table_t *h2_stream_get_trailers(h2_stream *stream)
+{
+ /* TODO */
+ return NULL;
+}
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=1715371&r1=1715370&r2=1715371&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 Fri Nov 20 15:13:11 2015
@@ -19,19 +19,14 @@
/**
* A HTTP/2 stream, e.g. a client request+response in HTTP/1.1 terms.
*
- * Ok, not quite, but close enough, since we do not implement server
- * pushes yet.
- *
* A stream always belongs to a h2_session, the one managing the
* connection to the client. The h2_session writes to the h2_stream,
* adding HEADERS and DATA and finally an EOS. When headers are done,
- * h2_stream can create a h2_task that can be scheduled to fullfill the
- * request.
+ * h2_stream is scheduled for handling, which is expected to produce
+ * a h2_response.
*
- * This response headers are added directly to the h2_mplx of the session,
- * but the response DATA can be read via h2_stream. Reading data will
- * never block but return APR_EAGAIN when there currently is no data (and
- * no eos) in the multiplexer for this stream.
+ * 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"
@@ -48,60 +43,261 @@ typedef enum {
struct h2_mplx;
struct h2_request;
struct h2_response;
+struct h2_session;
struct h2_task;
typedef struct h2_stream h2_stream;
struct h2_stream {
int id; /* http2 stream id */
+ int initiated_on; /* http2 stream id this was initiated on or 0 */
h2_stream_state_t state; /* http/2 state of this stream */
- struct h2_mplx *m; /* the multiplexer to work with */
+ struct h2_session *session; /* the session this stream belongs to */
+
+ apr_pool_t *pool; /* the memory pool for this stream */
+ struct h2_request *request; /* the request made in this stream */
+ struct h2_response *response; /* the response, once ready */
int aborted; /* was aborted */
int suspended; /* DATA sending has been suspended */
+ int rst_error; /* stream error for RST_STREAM */
+ int scheduled; /* stream has been scheduled */
+ int submitted; /* response HEADER has been sent */
- apr_pool_t *pool; /* the memory pool for this stream */
- struct h2_request *request; /* the request made in this stream */
+ apr_off_t input_remaining; /* remaining bytes on input as advertised via content-length */
+ apr_bucket_brigade *bbin; /* input DATA */
- struct h2_task *task; /* task created for this stream */
- struct h2_response *response; /* the response, once ready */
apr_bucket_brigade *bbout; /* output DATA */
+ apr_off_t data_frames_sent; /* # of DATA frames sent out for this stream */
};
-h2_stream *h2_stream_create(int id, apr_pool_t *pool, struct h2_mplx *m);
+#define H2_STREAM_RST(s, def) (s->rst_error? s->rst_error : (def))
+
+/**
+ * Create a stream in IDLE state.
+ * @param id the stream identifier
+ * @param pool the memory pool to use for this stream
+ * @param session the session this stream belongs to
+ * @return the newly created IDLE stream
+ */
+h2_stream *h2_stream_create(int id, apr_pool_t *pool, struct h2_session *session);
+
+/**
+ * Create a stream in OPEN state.
+ * @param id the stream identifier
+ * @param pool the memory pool to use for this 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);
+/**
+ * Destroy any resources held by this stream. Will destroy memory pool
+ * if still owned by the stream.
+ *
+ * @param stream the stream to destroy
+ */
apr_status_t h2_stream_destroy(h2_stream *stream);
+/**
+ * Removes stream from h2_session and destroys it.
+ *
+ * @param stream the stream to cleanup
+ */
+void h2_stream_cleanup(h2_stream *stream);
+
+/**
+ * Detach the memory pool from the stream. Will prevent stream
+ * destruction to take the pool with it.
+ *
+ * @param stream the stream to detach the pool from
+ * @param the detached memmory pool or NULL if stream no longer has one
+ */
apr_pool_t *h2_stream_detach_pool(h2_stream *stream);
-void h2_stream_attach_pool(h2_stream *stream, apr_pool_t *pool);
-void h2_stream_abort(h2_stream *stream);
-apr_status_t h2_stream_rwrite(h2_stream *stream, request_rec *r);
+/**
+ * Initialize stream->request with the given request_rec.
+ *
+ * @param stream stream to write request to
+ * @param r the request with all the meta data
+ */
+apr_status_t h2_stream_set_request(h2_stream *stream, request_rec *r);
-apr_status_t h2_stream_write_eos(h2_stream *stream);
+/**
+ * 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);
-apr_status_t h2_stream_write_header(h2_stream *stream,
- const char *name, size_t nlen,
- const char *value, size_t vlen);
+/*
+ * Add a HTTP/2 header (including pseudo headers) or trailer
+ * to the given stream, depending on stream state.
+ *
+ * @param stream stream to write the header to
+ * @param name the name of the HTTP/2 header
+ * @param nlen the number of characters in name
+ * @param value the header value
+ * @param vlen the number of characters in value
+ */
+apr_status_t h2_stream_add_header(h2_stream *stream,
+ const char *name, size_t nlen,
+ const char *value, size_t vlen);
-apr_status_t h2_stream_write_eoh(h2_stream *stream, int eos);
+/**
+ * Closes the stream's input.
+ *
+ * @param stream stream to close intput of
+ */
+apr_status_t h2_stream_close_input(h2_stream *stream);
+/*
+ * Write a chunk of DATA to the stream.
+ *
+ * @param stream stream to write the data to
+ * @param data the beginning of the bytes to write
+ * @param len the number of bytes to write
+ */
apr_status_t h2_stream_write_data(h2_stream *stream,
const char *data, size_t len);
+/**
+ * Reset the stream. Stream write/reads will return errors afterwards.
+ *
+ * @param stream the stream to reset
+ * @param error_code the HTTP/2 error code
+ */
+void h2_stream_rst(h2_stream *streamm, int error_code);
+
+/**
+ * Schedule the stream for execution. All header information must be
+ * present. Use the given priority comparision callback to determine
+ * order in queued streams.
+ *
+ * @param stream the stream to schedule
+ * @param eos != 0 iff no more input will arrive
+ * @param cmp priority comparision
+ * @param ctx context for comparision
+ */
+apr_status_t h2_stream_schedule(h2_stream *stream, int eos,
+ h2_stream_pri_cmp *cmp, void *ctx);
+
+/**
+ * Determine if stream has been scheduled already.
+ * @param stream the stream to check on
+ * @return != 0 iff stream has been scheduled
+ */
+int h2_stream_is_scheduled(h2_stream *stream);
+
+/**
+ * Set the response for this stream. Invoked when all meta data for
+ * the stream response has been collected.
+ *
+ * @param stream the stream to set the response for
+ * @param resonse the response data for the stream
+ * @param bb bucket brigade with output data for the stream. Optional,
+ * may be incomplete.
+ */
apr_status_t h2_stream_set_response(h2_stream *stream,
struct h2_response *response,
apr_bucket_brigade *bb);
+/**
+ * Do a speculative read on the stream output to determine the
+ * amount of data that can be read.
+ *
+ * @param stream the stream to speculatively read from
+ * @param plen (in-/out) number of bytes requested and on return amount of bytes that
+ * may be read without blocking
+ * @param peos (out) != 0 iff end of stream will be reached when reading plen
+ * bytes (out value).
+ * @return APR_SUCCESS if out information was computed successfully.
+ * APR_EAGAIN if not data is available and end of stream has not been
+ * reached yet.
+ */
apr_status_t h2_stream_prep_read(h2_stream *stream,
- apr_size_t *plen, int *peos);
+ apr_off_t *plen, int *peos);
+/**
+ * Read data from the stream output.
+ *
+ * @param stream the stream to read from
+ * @param cb callback to invoke for byte chunks read. Might be invoked
+ * multiple times (with different values) for one read operation.
+ * @param ctx context data for callback
+ * @param plen (in-/out) max. number of bytes to read and on return actual
+ * number of bytes read
+ * @param peos (out) != 0 iff end of stream has been reached while reading
+ * @return APR_SUCCESS if out information was computed successfully.
+ * APR_EAGAIN if not data is available and end of stream has not been
+ * reached yet.
+ */
apr_status_t h2_stream_readx(h2_stream *stream, h2_io_data_cb *cb,
- void *ctx, apr_size_t *plen, int *peos);
+ void *ctx, apr_off_t *plen, int *peos);
+
+/**
+ * Read a maximum number of bytes into the bucket brigade.
+ *
+ * @param stream the stream to read from
+ * @param bb the brigade to append output to
+ * @param plen (in-/out) max. number of bytes to append and on return actual
+ * number of bytes appended to brigade
+ * @param peos (out) != 0 iff end of stream has been reached while reading
+ * @return APR_SUCCESS if out information was computed successfully.
+ * APR_EAGAIN if not data is available and end of stream has not been
+ * reached yet.
+ */
+apr_status_t h2_stream_read_to(h2_stream *stream, apr_bucket_brigade *bb,
+ apr_off_t *plen, int *peos);
+/**
+ * Set the suspended state of the stream.
+ * @param stream the stream to change state on
+ * @param suspended boolean value if stream is suspended
+ */
void h2_stream_set_suspended(h2_stream *stream, int suspended);
+
+/**
+ * Check if the stream has been suspended.
+ * @param stream the stream to check
+ * @return != 0 iff stream is suspended.
+ */
int h2_stream_is_suspended(h2_stream *stream);
+/**
+ * Check if the stream has open input.
+ * @param stream the stream to check
+ * @return != 0 iff stream has open input.
+ */
+int h2_stream_input_is_open(h2_stream *stream);
+
+/**
+ * Check if the stream has not submitted a response or RST yet.
+ * @param stream the stream to check
+ * @return != 0 iff stream has not submitted a response or RST.
+ */
+int h2_stream_needs_submit(h2_stream *stream);
+
+/**
+ * Submit any server push promises on this stream and schedule
+ * the tasks connection with these.
+ *
+ * @param stream the stream for which to submit
+ */
+apr_status_t h2_stream_submit_pushes(h2_stream *stream);
+
+/**
+ * Get optional trailers for this stream, may be NULL. Meaningful
+ * results can only be expected when the end of the response body has
+ * been reached.
+ *
+ * @param stream to ask for trailers
+ * @return trailers for NULL
+ */
+apr_table_t *h2_stream_get_trailers(h2_stream *stream);
+
#endif /* defined(__mod_h2__h2_stream__) */
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.c?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.c Fri Nov 20 15:13:11 2015
@@ -16,34 +16,31 @@
#include <assert.h>
#include <stddef.h>
+#include <apr_hash.h>
#include <apr_strings.h>
#include <httpd.h>
-#include <http_core.h>
-#include <http_connection.h>
#include <http_log.h>
#include "h2_private.h"
-#include "h2_session.h"
#include "h2_stream.h"
-#include "h2_task.h"
#include "h2_stream_set.h"
-#define H2_STREAM_IDX(list, i) ((h2_stream**)(list)->elts)[i]
struct h2_stream_set {
- apr_array_header_t *list;
+ apr_hash_t *hash;
};
-h2_stream_set *h2_stream_set_create(apr_pool_t *pool)
+static unsigned int stream_hash(const char *key, apr_ssize_t *klen)
+{
+ return (unsigned int)(*((int*)key));
+}
+
+h2_stream_set *h2_stream_set_create(apr_pool_t *pool, int max)
{
h2_stream_set *sp = apr_pcalloc(pool, sizeof(h2_stream_set));
- if (sp) {
- sp->list = apr_array_make(pool, 100, sizeof(h2_stream*));
- if (!sp->list) {
- return NULL;
- }
- }
+ sp->hash = apr_hash_make_custom(pool, stream_hash);
+
return sp;
}
@@ -52,113 +49,97 @@ void h2_stream_set_destroy(h2_stream_set
(void)sp;
}
-static int h2_stream_id_cmp(const void *s1, const void *s2)
+h2_stream *h2_stream_set_get(h2_stream_set *sp, int stream_id)
{
- h2_stream **pstream1 = (h2_stream **)s1;
- h2_stream **pstream2 = (h2_stream **)s2;
- return (*pstream1)->id - (*pstream2)->id;
+ return apr_hash_get(sp->hash, &stream_id, sizeof(stream_id));
}
-h2_stream *h2_stream_set_get(h2_stream_set *sp, int stream_id)
+void h2_stream_set_add(h2_stream_set *sp, h2_stream *stream)
{
- /* we keep the array sorted by id, so lookup can be done
- * by bsearch.
- */
- h2_stream key;
- h2_stream *pkey, **ps;
- memset(&key, 0, sizeof(key));
- key.id = stream_id;
- pkey = &key;
- ps = bsearch(&pkey, sp->list->elts, sp->list->nelts,
- sp->list->elt_size, h2_stream_id_cmp);
- return ps? *ps : NULL;
-}
-
-static void h2_stream_set_sort(h2_stream_set *sp)
-{
- qsort(sp->list->elts, sp->list->nelts, sp->list->elt_size,
- h2_stream_id_cmp);
-}
-
-apr_status_t h2_stream_set_add(h2_stream_set *sp, h2_stream *stream)
-{
- h2_stream *existing = h2_stream_set_get(sp, stream->id);
- if (!existing) {
- int last;
- APR_ARRAY_PUSH(sp->list, h2_stream*) = stream;
- /* Normally, streams get added in ascending order if id. We
- * keep the array sorted, so we just need to check of the newly
- * appended stream has a lower id than the last one. if not,
- * sorting is not necessary.
- */
- last = sp->list->nelts - 1;
- if (last > 0
- && (H2_STREAM_IDX(sp->list, last)->id
- < H2_STREAM_IDX(sp->list, last-1)->id)) {
- h2_stream_set_sort(sp);
- }
- }
- return APR_SUCCESS;
+ apr_hash_set(sp->hash, &stream->id, sizeof(stream->id), stream);
}
-h2_stream *h2_stream_set_remove(h2_stream_set *sp, h2_stream *stream)
+void h2_stream_set_remove(h2_stream_set *sp, int stream_id)
{
- int i;
- for (i = 0; i < sp->list->nelts; ++i) {
- h2_stream *s = H2_STREAM_IDX(sp->list, i);
- if (s == stream) {
- int n;
- --sp->list->nelts;
- n = sp->list->nelts - i;
- if (n > 0) {
- /* Close the hole in the array by moving the upper
- * parts down one step.
- */
- h2_stream **selts = (h2_stream**)sp->list->elts;
- memmove(selts+i, selts+i+1, n * sizeof(h2_stream*));
- }
- return s;
- }
- }
- return NULL;
+ apr_hash_set(sp->hash, &stream_id, sizeof(stream_id), NULL);
}
-void h2_stream_set_remove_all(h2_stream_set *sp)
+int h2_stream_set_is_empty(h2_stream_set *sp)
{
- sp->list->nelts = 0;
+ return apr_hash_count(sp->hash) == 0;
}
-int h2_stream_set_is_empty(h2_stream_set *sp)
+apr_size_t h2_stream_set_size(h2_stream_set *sp)
{
- AP_DEBUG_ASSERT(sp);
- return sp->list->nelts == 0;
+ return apr_hash_count(sp->hash);
}
-h2_stream *h2_stream_set_find(h2_stream_set *sp,
- h2_stream_set_match_fn *match, void *ctx)
+typedef struct {
+ h2_stream_set_iter_fn *iter;
+ void *ctx;
+} iter_ctx;
+
+static int hash_iter(void *ctx, const void *key, apr_ssize_t klen,
+ const void *val)
{
- h2_stream *s = NULL;
- int i;
- for (i = 0; !s && i < sp->list->nelts; ++i) {
- s = match(ctx, H2_STREAM_IDX(sp->list, i));
- }
- return s;
+ iter_ctx *ictx = ctx;
+ return ictx->iter(ictx->ctx, (h2_stream*)val);
}
void h2_stream_set_iter(h2_stream_set *sp,
h2_stream_set_iter_fn *iter, void *ctx)
{
- int i;
- for (i = 0; i < sp->list->nelts; ++i) {
- h2_stream *s = H2_STREAM_IDX(sp->list, i);
- if (!iter(ctx, s)) {
- break;
- }
+ iter_ctx ictx;
+ ictx.iter = iter;
+ ictx.ctx = ctx;
+ apr_hash_do(hash_iter, &ictx, sp->hash);
+}
+
+static int unsubmitted_iter(void *ctx, h2_stream *stream)
+{
+ if (h2_stream_needs_submit(stream)) {
+ *((int *)ctx) = 1;
+ return 0;
}
+ return 1;
}
-apr_size_t h2_stream_set_size(h2_stream_set *sp)
+int h2_stream_set_has_unsubmitted(h2_stream_set *sp)
+{
+ int has_unsubmitted = 0;
+ h2_stream_set_iter(sp, unsubmitted_iter, &has_unsubmitted);
+ return has_unsubmitted;
+}
+
+static int input_open_iter(void *ctx, h2_stream *stream)
+{
+ if (h2_stream_input_is_open(stream)) {
+ *((int *)ctx) = 1;
+ return 0;
+ }
+ return 1;
+}
+
+int h2_stream_set_has_open_input(h2_stream_set *sp)
+{
+ int has_input_open = 0;
+ h2_stream_set_iter(sp, input_open_iter, &has_input_open);
+ return has_input_open;
+}
+
+static int suspended_iter(void *ctx, h2_stream *stream)
+{
+ if (h2_stream_is_suspended(stream)) {
+ *((int *)ctx) = 1;
+ return 0;
+ }
+ return 1;
+}
+
+int h2_stream_set_has_suspended(h2_stream_set *sp)
{
- return sp->list->nelts;
+ int has_suspended = 0;
+ h2_stream_set_iter(sp, suspended_iter, &has_suspended);
+ return has_suspended;
}
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.h
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.h?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.h (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_stream_set.h Fri Nov 20 15:13:11 2015
@@ -27,26 +27,25 @@ typedef int h2_stream_set_iter_fn(void *
typedef struct h2_stream_set h2_stream_set;
-h2_stream_set *h2_stream_set_create(apr_pool_t *pool);
+h2_stream_set *h2_stream_set_create(apr_pool_t *pool, int max);
void h2_stream_set_destroy(h2_stream_set *sp);
-apr_status_t h2_stream_set_add(h2_stream_set *sp, h2_stream *stream);
+void h2_stream_set_add(h2_stream_set *sp, h2_stream *stream);
h2_stream *h2_stream_set_get(h2_stream_set *sp, int stream_id);
-h2_stream *h2_stream_set_remove(h2_stream_set *sp,h2_stream *stream);
+void h2_stream_set_remove(h2_stream_set *sp, int stream_id);
-void h2_stream_set_remove_all(h2_stream_set *sp);
+void h2_stream_set_iter(h2_stream_set *sp,
+ h2_stream_set_iter_fn *iter, void *ctx);
int h2_stream_set_is_empty(h2_stream_set *sp);
apr_size_t h2_stream_set_size(h2_stream_set *sp);
-h2_stream *h2_stream_set_find(h2_stream_set *sp,
- h2_stream_set_match_fn *match, void *ctx);
-
-void h2_stream_set_iter(h2_stream_set *sp,
- h2_stream_set_iter_fn *iter, void *ctx);
+int h2_stream_set_has_unsubmitted(h2_stream_set *sp);
+int h2_stream_set_has_open_input(h2_stream_set *sp);
+int h2_stream_set_has_suspended(h2_stream_set *sp);
#endif /* defined(__mod_h2__h2_stream_set__) */
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_switch.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_switch.c?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_switch.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_switch.c Fri Nov 20 15:13:11 2015
@@ -35,24 +35,12 @@
#include "h2_switch.h"
/*******************************************************************************
- * SSL var lookup
- */
-APR_DECLARE_OPTIONAL_FN(char *, ssl_var_lookup,
- (apr_pool_t *, server_rec *,
- conn_rec *, request_rec *,
- char *));
-static char *(*opt_ssl_var_lookup)(apr_pool_t *, server_rec *,
- conn_rec *, request_rec *,
- char *);
-
-/*******************************************************************************
* Once per lifetime init, retrieve optional functions
*/
apr_status_t h2_switch_init(apr_pool_t *pool, server_rec *s)
{
(void)pool;
ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s, "h2_switch init");
- opt_ssl_var_lookup = APR_RETRIEVE_OPTIONAL_FN(ssl_var_lookup);
return APR_SUCCESS;
}
@@ -63,7 +51,8 @@ static int h2_protocol_propose(conn_rec
apr_array_header_t *proposals)
{
int proposed = 0;
- const char **protos = h2_h2_is_tls(c)? h2_tls_protos : h2_clear_protos;
+ int is_tls = h2_h2_is_tls(c);
+ const char **protos = is_tls? h2_tls_protos : h2_clear_protos;
(void)s;
if (strcmp(AP_PROTOCOL_HTTP1, ap_get_protocol(c))) {
@@ -74,12 +63,23 @@ static int h2_protocol_propose(conn_rec
return DECLINED;
}
+ if (!h2_is_acceptable_connection(c, 0)) {
+ ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, c,
+ "protocol propose: connection requirements not met");
+ return DECLINED;
+ }
+
if (r) {
- const char *p;
/* So far, this indicates an HTTP/1 Upgrade header initiated
* protocol switch. For that, the HTTP2-Settings header needs
* to be present and valid for the connection.
*/
+ const char *p;
+
+ if (!h2_allows_h2_upgrade(c)) {
+ return DECLINED;
+ }
+
p = apr_table_get(r->headers_in, "HTTP2-Settings");
if (!p) {
ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, r,
@@ -154,7 +154,7 @@ static int h2_protocol_switch(conn_rec *
ap_remove_input_filter_byhandle(r->input_filters, "reqtimeout");
/* Ok, start an h2_conn on this one. */
- status = h2_conn_rprocess(r);
+ status = h2_conn_process(r->connection, r);
if (status != DONE) {
/* Nothing really to do about this. */
ap_log_rerror(APLOG_MARK, APLOG_DEBUG, status, r,
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=1715371&r1=1715370&r2=1715371&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 Fri Nov 20 15:13:11 2015
@@ -38,6 +38,7 @@
#include "h2_from_h1.h"
#include "h2_h2.h"
#include "h2_mplx.h"
+#include "h2_request.h"
#include "h2_session.h"
#include "h2_stream.h"
#include "h2_task_input.h"
@@ -52,35 +53,37 @@ static apr_status_t h2_filter_stream_inp
ap_input_mode_t mode,
apr_read_type_e block,
apr_off_t readbytes) {
- h2_task_env *env = filter->ctx;
- AP_DEBUG_ASSERT(env);
- if (!env->input) {
+ h2_task *task = filter->ctx;
+ AP_DEBUG_ASSERT(task);
+ if (!task->input) {
return APR_ECONNABORTED;
}
- return h2_task_input_read(env->input, filter, brigade,
+ return h2_task_input_read(task->input, filter, brigade,
mode, block, readbytes);
}
static apr_status_t h2_filter_stream_output(ap_filter_t* filter,
apr_bucket_brigade* brigade) {
- h2_task_env *env = filter->ctx;
- AP_DEBUG_ASSERT(env);
- if (!env->output) {
+ h2_task *task = filter->ctx;
+ AP_DEBUG_ASSERT(task);
+ if (!task->output) {
return APR_ECONNABORTED;
}
- return h2_task_output_write(env->output, filter, brigade);
+ return h2_task_output_write(task->output, filter, brigade);
}
static apr_status_t h2_filter_read_response(ap_filter_t* f,
apr_bucket_brigade* bb) {
- h2_task_env *env = f->ctx;
- AP_DEBUG_ASSERT(env);
- if (!env->output || !env->output->from_h1) {
+ h2_task *task = f->ctx;
+ AP_DEBUG_ASSERT(task);
+ if (!task->output || !task->output->from_h1) {
return APR_ECONNABORTED;
}
- return h2_from_h1_read_response(env->output->from_h1, f, bb);
+ return h2_from_h1_read_response(task->output->from_h1, f, bb);
}
+static apr_status_t h2_task_process_request(h2_task *task);
+
/*******************************************************************************
* Register various hooks
*/
@@ -119,21 +122,15 @@ static int h2_task_pre_conn(conn_rec* c,
(void)arg;
if (h2_ctx_is_task(ctx)) {
- h2_task_env *env = h2_ctx_get_task(ctx);
-
- /* This connection is a pseudo-connection used for a h2_task.
- * Since we read/write directly from it ourselves, we need
- * to disable a possible ssl connection filter.
- */
- h2_tls_disable(c);
+ h2_task *task = h2_ctx_get_task(ctx);
ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c,
"h2_h2, pre_connection, found stream task");
/* Add our own, network level in- and output filters.
*/
- ap_add_input_filter("H2_TO_H1", env, NULL, c);
- ap_add_output_filter("H1_TO_H2", env, NULL, c);
+ ap_add_input_filter("H2_TO_H1", task, NULL, c);
+ ap_add_output_filter("H1_TO_H2", task, NULL, c);
}
return OK;
}
@@ -143,61 +140,43 @@ static int h2_task_process_conn(conn_rec
h2_ctx *ctx = h2_ctx_get(c);
if (h2_ctx_is_task(ctx)) {
- if (!ctx->task_env->serialize_headers) {
+ if (!ctx->task->serialize_headers) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c,
"h2_h2, processing request directly");
- h2_task_process_request(ctx->task_env);
+ h2_task_process_request(ctx->task);
return DONE;
}
ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c,
- "h2_task(%s), serialized handling", ctx->task_env->id);
+ "h2_task(%s), serialized handling", ctx->task->id);
}
return DECLINED;
}
-h2_task *h2_task_create(long session_id,
- int stream_id,
- apr_pool_t *stream_pool,
- h2_mplx *mplx, conn_rec *c)
+h2_task *h2_task_create(long session_id, const h2_request *req,
+ apr_pool_t *pool, h2_mplx *mplx, int eos)
{
- h2_task *task = apr_pcalloc(stream_pool, sizeof(h2_task));
+ h2_task *task = apr_pcalloc(pool, sizeof(h2_task));
if (task == NULL) {
- ap_log_perror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, stream_pool,
+ ap_log_perror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, pool,
APLOGNO(02941) "h2_task(%ld-%d): create stream task",
- session_id, stream_id);
- h2_mplx_out_close(mplx, stream_id);
+ session_id, req->id);
+ h2_mplx_out_close(mplx, req->id);
return NULL;
}
- APR_RING_ELEM_INIT(task, link);
-
- task->id = apr_psprintf(stream_pool, "%ld-%d", session_id, stream_id);
- task->stream_id = stream_id;
+ task->id = apr_psprintf(pool, "%ld-%d", session_id, req->id);
+ task->stream_id = req->id;
+ task->pool = pool;
task->mplx = mplx;
+ task->c = h2_conn_create(mplx->c, task->pool);
+
+ task->request = req;
+ task->input_eos = eos;
- task->c = c;
-
- ap_log_perror(APLOG_MARK, APLOG_DEBUG, 0, stream_pool,
- "h2_task(%s): created", task->id);
return task;
}
-void h2_task_set_request(h2_task *task,
- const char *method,
- const char *scheme,
- const char *authority,
- const char *path,
- apr_table_t *headers, int eos)
-{
- task->method = method;
- task->scheme = scheme;
- task->authority = authority;
- task->path = path;
- task->headers = headers;
- task->input_eos = eos;
-}
-
apr_status_t h2_task_destroy(h2_task *task)
{
(void)task;
@@ -208,121 +187,56 @@ apr_status_t h2_task_do(h2_task *task, h
{
apr_status_t status = APR_SUCCESS;
h2_config *cfg = h2_config_get(task->mplx->c);
- h2_task_env env;
AP_DEBUG_ASSERT(task);
- memset(&env, 0, sizeof(env));
-
- env.id = task->id;
- env.stream_id = task->stream_id;
- env.mplx = task->mplx;
- task->mplx = NULL;
-
- env.input_eos = task->input_eos;
- env.serialize_headers = h2_config_geti(cfg, H2_CONF_SER_HEADERS);
-
- /* Create a subpool from the worker one to be used for all things
- * with life-time of this task_env execution.
- */
- apr_pool_create(&env.pool, h2_worker_get_pool(worker));
+ task->serialize_headers = h2_config_geti(cfg, H2_CONF_SER_HEADERS);
- /* Link the env to the worker which provides useful things such
- * as mutex, a socket etc. */
- env.io = h2_worker_get_cond(worker);
-
- /* Clone fields, so that lifetimes become (more) independent. */
- env.method = apr_pstrdup(env.pool, task->method);
- env.scheme = apr_pstrdup(env.pool, task->scheme);
- env.authority = apr_pstrdup(env.pool, task->authority);
- env.path = apr_pstrdup(env.pool, task->path);
- env.headers = apr_table_clone(env.pool, task->headers);
-
- /* Setup the pseudo connection to use our own pool and bucket_alloc */
- env.c = *task->c;
- task->c = NULL;
- status = h2_conn_setup(&env, worker);
-
- /* save in connection that this one is a pseudo connection, prevents
- * other hooks from messing with it. */
- h2_ctx_create_for(&env.c, &env);
+ status = h2_worker_setup_task(worker, task);
+
+ /* save in connection that this one is a pseudo connection */
+ h2_ctx_create_for(task->c, task);
if (status == APR_SUCCESS) {
- env.input = h2_task_input_create(&env, env.pool,
- env.c.bucket_alloc);
- env.output = h2_task_output_create(&env, env.pool,
- env.c.bucket_alloc);
- status = h2_conn_process(&env.c, h2_worker_get_socket(worker));
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, &env.c,
- "h2_task(%s): processing done", env.id);
+ task->input = h2_task_input_create(task, task->pool,
+ task->c->bucket_alloc);
+ task->output = h2_task_output_create(task, task->pool);
+
+ ap_process_connection(task->c, h2_worker_get_socket(worker));
+
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
+ "h2_task(%s): processing done", task->id);
}
else {
- ap_log_cerror(APLOG_MARK, APLOG_WARNING, status, &env.c,
- APLOGNO(02957) "h2_task(%s): error setting up h2_task_env",
- env.id);
+ ap_log_cerror(APLOG_MARK, APLOG_WARNING, status, task->c,
+ APLOGNO(02957) "h2_task(%s): error setting up h2_task",
+ task->id);
}
- if (env.input) {
- h2_task_input_destroy(env.input);
- env.input = NULL;
+ if (task->input) {
+ h2_task_input_destroy(task->input);
+ task->input = NULL;
}
- if (env.output) {
- h2_task_output_close(env.output);
- h2_task_output_destroy(env.output);
- env.output = NULL;
+ if (task->output) {
+ h2_task_output_close(task->output);
+ h2_task_output_destroy(task->output);
+ task->output = NULL;
}
- h2_task_set_finished(task);
- if (env.io) {
- apr_thread_cond_signal(env.io);
- }
-
- if (env.pool) {
- apr_pool_destroy(env.pool);
- env.pool = NULL;
+ if (task->io) {
+ apr_thread_cond_signal(task->io);
}
- if (env.c.id) {
- h2_conn_post(&env.c, worker);
- }
-
- h2_mplx_task_done(env.mplx, env.stream_id);
+ h2_mplx_task_done(task->mplx, task->stream_id);
+ h2_worker_release_task(worker, task);
return status;
}
-int h2_task_has_started(h2_task *task)
-{
- AP_DEBUG_ASSERT(task);
- return apr_atomic_read32(&task->has_started);
-}
-
-void h2_task_set_started(h2_task *task)
+static request_rec *h2_task_create_request(h2_task *task)
{
- AP_DEBUG_ASSERT(task);
- apr_atomic_set32(&task->has_started, 1);
-}
-
-int h2_task_has_finished(h2_task *task)
-{
- return apr_atomic_read32(&task->has_finished);
-}
-
-void h2_task_set_finished(h2_task *task)
-{
- apr_atomic_set32(&task->has_finished, 1);
-}
-
-void h2_task_die(h2_task_env *env, int status, request_rec *r)
-{
- (void)env;
- ap_die(status, r);
-}
-
-static request_rec *h2_task_create_request(h2_task_env *env)
-{
- conn_rec *conn = &env->c;
+ conn_rec *conn = task->c;
request_rec *r;
apr_pool_t *p;
int access_status = HTTP_OK;
@@ -340,7 +254,7 @@ static request_rec *h2_task_create_reque
r->allowed_methods = ap_make_method_list(p, 2);
- r->headers_in = apr_table_copy(r->pool, env->headers);
+ r->headers_in = apr_table_copy(r->pool, task->request->headers);
r->trailers_in = apr_table_make(r->pool, 5);
r->subprocess_env = apr_table_make(r->pool, 25);
r->headers_out = apr_table_make(r->pool, 12);
@@ -379,18 +293,19 @@ static request_rec *h2_task_create_reque
/* Time to populate r with the data we have. */
r->request_time = apr_time_now();
- r->the_request = apr_psprintf(r->pool, "%s %s HTTP/1.1",
- env->method, env->path);
- r->method = env->method;
+ r->method = task->request->method;
/* Provide quick information about the request method as soon as known */
r->method_number = ap_method_number_of(r->method);
if (r->method_number == M_GET && r->method[0] == 'H') {
r->header_only = 1;
}
- ap_parse_uri(r, env->path);
+ ap_parse_uri(r, task->request->path);
r->protocol = (char*)"HTTP/2";
r->proto_num = HTTP_VERSION(2, 0);
+
+ r->the_request = apr_psprintf(r->pool, "%s %s %s",
+ r->method, task->request->path, r->protocol);
/* update what we think the virtual host is based on the headers we've
* now read. may update status.
@@ -417,7 +332,7 @@ static request_rec *h2_task_create_reque
/* Request check post hooks failed. An example of this would be a
* request for a vhost where h2 is disabled --> 421.
*/
- h2_task_die(env, access_status, r);
+ ap_die(access_status, r);
ap_update_child_status(conn->sbh, SERVER_BUSY_LOG, r);
ap_run_log_transaction(r);
r = NULL;
@@ -434,13 +349,13 @@ traceout:
}
-apr_status_t h2_task_process_request(h2_task_env *env)
+static apr_status_t h2_task_process_request(h2_task *task)
{
- conn_rec *c = &env->c;
+ conn_rec *c = task->c;
request_rec *r;
conn_state_t *cs = c->cs;
- r = h2_task_create_request(env);
+ r = h2_task_create_request(task);
if (r && (r->status == HTTP_OK)) {
ap_update_child_status(c->sbh, SERVER_BUSY_READ, r);
@@ -454,6 +369,8 @@ apr_status_t h2_task_process_request(h2_
* will result in a segfault immediately instead
* of nondeterministic failures later.
*/
+ if (cs)
+ cs->state = CONN_STATE_WRITE_COMPLETION;
r = NULL;
}
ap_update_child_status(c->sbh, SERVER_BUSY_WRITE, NULL);
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=1715371&r1=1715370&r2=1715371&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 Fri Nov 20 15:13:11 2015
@@ -39,149 +39,39 @@ struct apr_thread_cond_t;
struct h2_conn;
struct h2_mplx;
struct h2_task;
+struct h2_request;
struct h2_resp_head;
struct h2_worker;
typedef struct h2_task h2_task;
struct h2_task {
- /** Links to the rest of the tasks */
- APR_RING_ENTRY(h2_task) link;
-
const char *id;
int stream_id;
struct h2_mplx *mplx;
- volatile apr_uint32_t has_started;
- volatile apr_uint32_t has_finished;
-
- const char *method;
- const char *scheme;
- const char *authority;
- const char *path;
- apr_table_t *headers;
+ const struct h2_request *request;
int input_eos;
- struct conn_rec *c;
-};
+ int serialize_headers;
-typedef struct h2_task_env h2_task_env;
+ struct conn_rec *c;
-struct h2_task_env {
- const char *id;
- int stream_id;
- struct h2_mplx *mplx;
-
apr_pool_t *pool; /* pool for task lifetime things */
apr_bucket_alloc_t *bucket_alloc;
-
- const char *method;
- const char *scheme;
- const char *authority;
- const char *path;
- apr_table_t *headers;
- int input_eos;
-
- int serialize_headers;
-
- struct conn_rec c;
struct h2_task_input *input;
struct h2_task_output *output;
struct apr_thread_cond_t *io; /* used to wait for events on */
};
-/**
- * The magic pointer value that indicates the head of a h2_task list
- * @param b The task list
- * @return The magic pointer value
- */
-#define H2_TASK_LIST_SENTINEL(b) APR_RING_SENTINEL((b), h2_task, link)
-
-/**
- * Determine if the task list is empty
- * @param b The list to check
- * @return true or false
- */
-#define H2_TASK_LIST_EMPTY(b) APR_RING_EMPTY((b), h2_task, link)
-
-/**
- * Return the first task in a list
- * @param b The list to query
- * @return The first task in the list
- */
-#define H2_TASK_LIST_FIRST(b) APR_RING_FIRST(b)
-
-/**
- * Return the last task in a list
- * @param b The list to query
- * @return The last task int he list
- */
-#define H2_TASK_LIST_LAST(b) APR_RING_LAST(b)
-
-/**
- * Insert a single task at the front of a list
- * @param b The list to add to
- * @param e The task to insert
- */
-#define H2_TASK_LIST_INSERT_HEAD(b, e) do { \
- h2_task *ap__b = (e); \
- APR_RING_INSERT_HEAD((b), ap__b, h2_task, link); \
-} while (0)
-
-/**
- * Insert a single task at the end of a list
- * @param b The list to add to
- * @param e The task to insert
- */
-#define H2_TASK_LIST_INSERT_TAIL(b, e) do { \
- h2_task *ap__b = (e); \
- APR_RING_INSERT_TAIL((b), ap__b, h2_task, link); \
-} while (0)
-
-/**
- * Get the next task in the list
- * @param e The current task
- * @return The next task
- */
-#define H2_TASK_NEXT(e) APR_RING_NEXT((e), link)
-/**
- * Get the previous task in the list
- * @param e The current task
- * @return The previous task
- */
-#define H2_TASK_PREV(e) APR_RING_PREV((e), link)
-
-/**
- * Remove a task from its list
- * @param e The task to remove
- */
-#define H2_TASK_REMOVE(e) APR_RING_REMOVE((e), link)
-
-
-h2_task *h2_task_create(long session_id, int stream_id,
- apr_pool_t *pool, struct h2_mplx *mplx,
- conn_rec *c);
+h2_task *h2_task_create(long session_id, const struct h2_request *req,
+ apr_pool_t *pool, struct h2_mplx *mplx, int eos);
apr_status_t h2_task_destroy(h2_task *task);
-void h2_task_set_request(h2_task *task,
- const char *method,
- const char *scheme,
- const char *authority,
- const char *path,
- apr_table_t *headers, int eos);
-
-
apr_status_t h2_task_do(h2_task *task, struct h2_worker *worker);
-apr_status_t h2_task_process_request(h2_task_env *env);
-
-int h2_task_has_started(h2_task *task);
-void h2_task_set_started(h2_task *task);
-int h2_task_has_finished(h2_task *task);
-void h2_task_set_finished(h2_task *task);
void h2_task_register_hooks(void);
-void h2_task_die(h2_task_env *env, int status, request_rec *r);
#endif /* defined(__mod_h2__h2_task__) */
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.c?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.c Fri Nov 20 15:13:11 2015
@@ -23,6 +23,7 @@
#include "h2_private.h"
#include "h2_conn.h"
#include "h2_mplx.h"
+#include "h2_request.h"
#include "h2_session.h"
#include "h2_stream.h"
#include "h2_task_input.h"
@@ -42,49 +43,34 @@ static int ser_header(void *ctx, const c
return 1;
}
-h2_task_input *h2_task_input_create(h2_task_env *env, apr_pool_t *pool,
+h2_task_input *h2_task_input_create(h2_task *task, apr_pool_t *pool,
apr_bucket_alloc_t *bucket_alloc)
{
h2_task_input *input = apr_pcalloc(pool, sizeof(h2_task_input));
if (input) {
- input->env = env;
+ input->task = task;
input->bb = NULL;
- if (env->serialize_headers) {
- ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, &env->c,
+ if (task->serialize_headers) {
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
"h2_task_input(%s): serialize request %s %s",
- env->id, env->method, env->path);
+ task->id, task->request->method, task->request->path);
input->bb = apr_brigade_create(pool, bucket_alloc);
apr_brigade_printf(input->bb, NULL, NULL, "%s %s HTTP/1.1\r\n",
- env->method, env->path);
- apr_table_do(ser_header, input, env->headers, NULL);
+ task->request->method, task->request->path);
+ apr_table_do(ser_header, input, task->request->headers, NULL);
apr_brigade_puts(input->bb, NULL, NULL, "\r\n");
- if (input->env->input_eos) {
+ if (input->task->input_eos) {
APR_BRIGADE_INSERT_TAIL(input->bb, apr_bucket_eos_create(bucket_alloc));
}
}
- else if (!input->env->input_eos) {
+ else if (!input->task->input_eos) {
input->bb = apr_brigade_create(pool, bucket_alloc);
}
else {
/* We do not serialize and have eos already, no need to
* create a bucket brigade. */
}
-
- if (APLOGcdebug(&env->c)) {
- char buffer[1024];
- apr_size_t len = sizeof(buffer)-1;
- if (input->bb) {
- apr_brigade_flatten(input->bb, buffer, &len);
- }
- else {
- len = 0;
- }
- buffer[len] = 0;
- ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, &env->c,
- "h2_task_input(%s): request is: %s",
- env->id, buffer);
- }
}
return input;
}
@@ -104,32 +90,31 @@ apr_status_t h2_task_input_read(h2_task_
apr_status_t status = APR_SUCCESS;
apr_off_t bblen = 0;
- ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, f->c,
+ ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
"h2_task_input(%s): read, block=%d, mode=%d, readbytes=%ld",
- input->env->id, block, mode, (long)readbytes);
+ input->task->id, block, mode, (long)readbytes);
+
+ if (mode == AP_MODE_INIT) {
+ return ap_get_brigade(f->c->input_filters, bb, mode, block, readbytes);
+ }
if (is_aborted(f)) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
- "h2_task_input(%s): is aborted",
- input->env->id);
+ "h2_task_input(%s): is aborted", input->task->id);
return APR_ECONNABORTED;
}
- if (mode == AP_MODE_INIT) {
- return APR_SUCCESS;
- }
-
if (input->bb) {
status = apr_brigade_length(input->bb, 1, &bblen);
if (status != APR_SUCCESS) {
ap_log_cerror(APLOG_MARK, APLOG_WARNING, status, f->c,
APLOGNO(02958) "h2_task_input(%s): brigade length fail",
- input->env->id);
+ input->task->id);
return status;
}
}
- if ((bblen == 0) && input->env->input_eos) {
+ if ((bblen == 0) && input->task->input_eos) {
return APR_EOF;
}
@@ -139,19 +124,19 @@ apr_status_t h2_task_input_read(h2_task_
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
"h2_task_input(%s): get more data from mplx, block=%d, "
"readbytes=%ld, queued=%ld",
- input->env->id, block,
+ input->task->id, block,
(long)readbytes, (long)bblen);
/* Although we sometimes get called with APR_NONBLOCK_READs,
we seem to fill our buffer blocking. Otherwise we get EAGAIN,
return that to our caller and everyone throws up their hands,
never calling us again. */
- status = h2_mplx_in_read(input->env->mplx, APR_BLOCK_READ,
- input->env->stream_id, input->bb,
- input->env->io);
+ status = h2_mplx_in_read(input->task->mplx, APR_BLOCK_READ,
+ input->task->stream_id, input->bb,
+ input->task->io);
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
"h2_task_input(%s): mplx in read returned",
- input->env->id);
+ input->task->id);
if (status != APR_SUCCESS) {
return status;
}
@@ -160,27 +145,27 @@ apr_status_t h2_task_input_read(h2_task_
return status;
}
if ((bblen == 0) && (block == APR_NONBLOCK_READ)) {
- return h2_util_has_eos(input->bb, 0)? APR_EOF : APR_EAGAIN;
+ return h2_util_has_eos(input->bb, -1)? APR_EOF : APR_EAGAIN;
}
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
"h2_task_input(%s): mplx in read, %ld bytes in brigade",
- input->env->id, (long)bblen);
+ input->task->id, (long)bblen);
}
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
"h2_task_input(%s): read, mode=%d, block=%d, "
"readbytes=%ld, queued=%ld",
- input->env->id, mode, block,
+ input->task->id, mode, block,
(long)readbytes, (long)bblen);
if (!APR_BRIGADE_EMPTY(input->bb)) {
if (mode == AP_MODE_EXHAUSTIVE) {
/* return all we have */
- return h2_util_move(bb, input->bb, readbytes, 0,
+ return h2_util_move(bb, input->bb, readbytes, NULL,
"task_input_read(exhaustive)");
}
else if (mode == AP_MODE_READBYTES) {
- return h2_util_move(bb, input->bb, readbytes, 0,
+ return h2_util_move(bb, input->bb, readbytes, NULL,
"task_input_read(readbytes)");
}
else if (mode == AP_MODE_SPECULATIVE) {
@@ -199,7 +184,7 @@ apr_status_t h2_task_input_read(h2_task_
buffer[len] = 0;
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
"h2_task_input(%s): getline: %s",
- input->env->id, buffer);
+ input->task->id, buffer);
}
return status;
}
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.h
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.h?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.h (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task_input.h Fri Nov 20 15:13:11 2015
@@ -22,16 +22,16 @@
*/
struct apr_thread_cond_t;
struct h2_mplx;
-struct h2_task_env;
+struct h2_task;
typedef struct h2_task_input h2_task_input;
struct h2_task_input {
- struct h2_task_env *env;
+ struct h2_task *task;
apr_bucket_brigade *bb;
};
-h2_task_input *h2_task_input_create(struct h2_task_env *env, apr_pool_t *pool,
+h2_task_input *h2_task_input_create(struct h2_task *task, apr_pool_t *pool,
apr_bucket_alloc_t *bucket_alloc);
void h2_task_input_destroy(h2_task_input *input);
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.c?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.c Fri Nov 20 15:13:11 2015
@@ -24,6 +24,7 @@
#include "h2_private.h"
#include "h2_conn.h"
#include "h2_mplx.h"
+#include "h2_request.h"
#include "h2_session.h"
#include "h2_stream.h"
#include "h2_from_h1.h"
@@ -33,16 +34,14 @@
#include "h2_util.h"
-h2_task_output *h2_task_output_create(h2_task_env *env, apr_pool_t *pool,
- apr_bucket_alloc_t *bucket_alloc)
+h2_task_output *h2_task_output_create(h2_task *task, apr_pool_t *pool)
{
h2_task_output *output = apr_pcalloc(pool, sizeof(h2_task_output));
- (void)bucket_alloc;
if (output) {
- output->env = env;
+ output->task = task;
output->state = H2_TASK_OUT_INIT;
- output->from_h1 = h2_from_h1_create(env->stream_id, pool);
+ output->from_h1 = h2_from_h1_create(task->stream_id, pool);
if (!output->from_h1) {
return NULL;
}
@@ -73,18 +72,19 @@ static apr_status_t open_if_needed(h2_ta
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, f->c,
"h2_task_output(%s): write without response "
"for %s %s %s",
- output->env->id, output->env->method,
- output->env->authority, output->env->path);
+ output->task->id, output->task->request->method,
+ output->task->request->authority,
+ output->task->request->path);
f->c->aborted = 1;
}
- if (output->env->io) {
- apr_thread_cond_broadcast(output->env->io);
+ if (output->task->io) {
+ apr_thread_cond_broadcast(output->task->io);
}
return APR_ECONNABORTED;
}
- return h2_mplx_out_open(output->env->mplx, output->env->stream_id,
- response, f, bb, output->env->io);
+ return h2_mplx_out_open(output->task->mplx, output->task->stream_id,
+ response, f, bb, output->task->io);
}
return APR_EOF;
}
@@ -93,7 +93,7 @@ void h2_task_output_close(h2_task_output
{
open_if_needed(output, NULL, NULL);
if (output->state != H2_TASK_OUT_DONE) {
- h2_mplx_out_close(output->env->mplx, output->env->stream_id);
+ h2_mplx_out_close(output->task->mplx, output->task->stream_id);
output->state = H2_TASK_OUT_DONE;
}
}
@@ -113,7 +113,7 @@ apr_status_t h2_task_output_write(h2_tas
apr_status_t status;
if (APR_BRIGADE_EMPTY(bb)) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
- "h2_task_output(%s): empty write", output->env->id);
+ "h2_task_output(%s): empty write", output->task->id);
return APR_SUCCESS;
}
@@ -121,12 +121,12 @@ apr_status_t h2_task_output_write(h2_tas
if (status != APR_EOF) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
"h2_task_output(%s): opened and passed brigade",
- output->env->id);
+ output->task->id);
return status;
}
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
- "h2_task_output(%s): write brigade", output->env->id);
- return h2_mplx_out_write(output->env->mplx, output->env->stream_id,
- f, bb, output->env->io);
+ "h2_task_output(%s): write brigade", output->task->id);
+ return h2_mplx_out_write(output->task->mplx, output->task->stream_id,
+ f, bb, output->task->io);
}
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.h
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.h?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.h (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task_output.h Fri Nov 20 15:13:11 2015
@@ -23,7 +23,7 @@
*/
struct apr_thread_cond_t;
struct h2_mplx;
-struct h2_task_env;
+struct h2_task;
struct h2_from_h1;
typedef enum {
@@ -35,13 +35,12 @@ typedef enum {
typedef struct h2_task_output h2_task_output;
struct h2_task_output {
- struct h2_task_env *env;
+ struct h2_task *task;
h2_task_output_state_t state;
struct h2_from_h1 *from_h1;
};
-h2_task_output *h2_task_output_create(struct h2_task_env *env, apr_pool_t *pool,
- apr_bucket_alloc_t *bucket_alloc);
+h2_task_output *h2_task_output_create(struct h2_task *task, apr_pool_t *pool);
void h2_task_output_destroy(h2_task_output *output);
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_task_queue.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_task_queue.c?rev=1715371&r1=1715370&r2=1715371&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_task_queue.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_task_queue.c Fri Nov 20 15:13:11 2015
@@ -23,65 +23,158 @@
#include "h2_task_queue.h"
-h2_task_queue *h2_tq_create(long id, apr_pool_t *pool)
+static void tq_grow(h2_task_queue *q, int nlen);
+static void tq_swap(h2_task_queue *q, int i, int j);
+static int tq_bubble_up(h2_task_queue *q, int i, int top,
+ h2_tq_cmp *cmp, void *ctx);
+static int tq_bubble_down(h2_task_queue *q, int i, int bottom,
+ h2_tq_cmp *cmp, void *ctx);
+
+h2_task_queue *h2_tq_create(apr_pool_t *pool, int capacity)
{
h2_task_queue *q = apr_pcalloc(pool, sizeof(h2_task_queue));
if (q) {
- q->id = id;
- APR_RING_ELEM_INIT(q, link);
- APR_RING_INIT(&q->tasks, h2_task, link);
+ q->pool = pool;
+ tq_grow(q, capacity);
+ q->nelts = 0;
}
return q;
}
-void h2_tq_destroy(h2_task_queue *q)
+int h2_tq_empty(h2_task_queue *q)
{
- while (!H2_TASK_LIST_EMPTY(&q->tasks)) {
- h2_task *task = H2_TASK_LIST_FIRST(&q->tasks);
- H2_TASK_REMOVE(task);
- }
+ return q->nelts == 0;
}
-static int in_list(h2_task_queue *q, h2_task *task)
+void h2_tq_add(h2_task_queue *q, int sid, h2_tq_cmp *cmp, void *ctx)
{
- h2_task *e;
- for (e = H2_TASK_LIST_FIRST(&q->tasks);
- e != H2_TASK_LIST_SENTINEL(&q->tasks);
- e = H2_TASK_NEXT(e)) {
- if (e == task) {
- return 1;
+ int i;
+
+ if (q->nelts >= q->nalloc) {
+ tq_grow(q, q->nalloc * 2);
+ }
+
+ i = (q->head + q->nelts) % q->nalloc;
+ q->elts[i] = sid;
+ ++q->nelts;
+
+ /* bubble it to the front of the queue */
+ tq_bubble_up(q, i, q->head, cmp, ctx);
+}
+
+int h2_tq_remove(h2_task_queue *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;
}
-int h2_tq_empty(h2_task_queue *q)
+void h2_tq_sort(h2_task_queue *q, h2_tq_cmp *cmp, void *ctx)
{
- return H2_TASK_LIST_EMPTY(&q->tasks);
+ /* 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 = tq_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. */
+ tq_bubble_down(q, i, last, cmp, ctx);
+ }
+ i = prev;
+ };
+ }
+}
+
+
+int h2_tq_shift(h2_task_queue *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 tq_grow(h2_task_queue *q, int nlen)
+{
+ AP_DEBUG_ASSERT(q->nalloc <= nlen);
+ if (nlen > q->nalloc) {
+ int *nq = apr_pcalloc(q->pool, sizeof(h2_task *) * 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;
+ }
}
-void h2_tq_append(h2_task_queue *q, struct h2_task *task)
+static void tq_swap(h2_task_queue *q, int i, int j)
{
- H2_TASK_LIST_INSERT_TAIL(&q->tasks, task);
+ int x = q->elts[i];
+ q->elts[i] = q->elts[j];
+ q->elts[j] = x;
}
-apr_status_t h2_tq_remove(h2_task_queue *q, struct h2_task *task)
+static int tq_bubble_up(h2_task_queue *q, int i, int top,
+ h2_tq_cmp *cmp, void *ctx)
{
- if (in_list(q, task)) {
- H2_TASK_REMOVE(task);
- return APR_SUCCESS;
+ int prev;
+ while (((prev = (q->nalloc + i - 1) % q->nalloc), i != top)
+ && (*cmp)(q->elts[i], q->elts[prev], ctx) < 0) {
+ tq_swap(q, prev, i);
+ i = prev;
}
- return APR_NOTFOUND;
+ return i;
}
-h2_task *h2_tq_pop_first(h2_task_queue *q)
+static int tq_bubble_down(h2_task_queue *q, int i, int bottom,
+ h2_tq_cmp *cmp, void *ctx)
{
- if (!H2_TASK_LIST_EMPTY(&q->tasks)) {
- h2_task *task = H2_TASK_LIST_FIRST(&q->tasks);
- H2_TASK_REMOVE(task);
- return task;
+ int next;
+ while (((next = (q->nalloc + i + 1) % q->nalloc), i != bottom)
+ && (*cmp)(q->elts[i], q->elts[next], ctx) > 0) {
+ tq_swap(q, next, i);
+ i = next;
}
- return NULL;
+ return i;
}