granicus.if.org Git - apache/blob - modules/http2/h2_task.c

   1 /* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at
   6  *
   7  * http://www.apache.org/licenses/LICENSE-2.0
   8
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <assert.h>
  17 #include <stddef.h>
  18
  19 #include <apr_atomic.h>
  20 #include <apr_thread_cond.h>
  21 #include <apr_strings.h>
  22
  23 #include <httpd.h>
  24 #include <http_core.h>
  25 #include <http_connection.h>
  26 #include <http_protocol.h>
  27 #include <http_request.h>
  28 #include <http_log.h>
  29 #include <http_vhost.h>
  30 #include <util_filter.h>
  31 #include <ap_mpm.h>
  32 #include <mod_core.h>
  33 #include <scoreboard.h>
  34
  35 #include "h2_private.h"
  36 #include "h2.h"
  37 #include "h2_bucket_beam.h"
  38 #include "h2_conn.h"
  39 #include "h2_config.h"
  40 #include "h2_ctx.h"
  41 #include "h2_from_h1.h"
  42 #include "h2_h2.h"
  43 #include "h2_mplx.h"
  44 #include "h2_request.h"
  45 #include "h2_response.h"
  46 #include "h2_session.h"
  47 #include "h2_stream.h"
  48 #include "h2_task.h"
  49 #include "h2_worker.h"
  50 #include "h2_util.h"
  51
  52 /*******************************************************************************
  53  * task input handling
  54  ******************************************************************************/
  55
  56 static int input_ser_header(void *ctx, const char *name, const char *value)
  57 {
  58     h2_task *task = ctx;
  59     apr_brigade_printf(task->input.bb, NULL, NULL, "%s: %s\r\n", name, value);
  60     return 1;
  61 }
  62
  63 static void make_chunk(h2_task *task, apr_bucket_brigade *bb,
  64                        apr_bucket *first, apr_uint64_t chunk_len,
  65                        apr_bucket *tail)
  66 {
  67     /* Surround the buckets [first, tail[ with new buckets carrying the
  68      * HTTP/1.1 chunked encoding format. If tail is NULL, the chunk extends
  69      * to the end of the brigade. */
  70     char buffer[128];
  71     apr_bucket *c;
  72     int len;
  73
  74     len = apr_snprintf(buffer, H2_ALEN(buffer),
  75                        "%"APR_UINT64_T_HEX_FMT"\r\n", chunk_len);
  76     c = apr_bucket_heap_create(buffer, len, NULL, bb->bucket_alloc);
  77     APR_BUCKET_INSERT_BEFORE(first, c);
  78     c = apr_bucket_heap_create("\r\n", 2, NULL, bb->bucket_alloc);
  79     if (tail) {
  80         APR_BUCKET_INSERT_BEFORE(tail, c);
  81     }
  82     else {
  83         APR_BRIGADE_INSERT_TAIL(bb, c);
  84     }
  85 }
  86
  87 static apr_status_t input_handle_eos(h2_task *task, request_rec *r,
  88                                      apr_bucket *b)
  89 {
  90     apr_status_t status = APR_SUCCESS;
  91     apr_bucket_brigade *bb = task->input.bb;
  92     apr_table_t *t = task->request? task->request->trailers : NULL;
  93
  94     if (task->input.chunked) {
  95         apr_bucket_brigade *tmp = apr_brigade_split_ex(bb, b, NULL);
  96         if (t && !apr_is_empty_table(t)) {
  97             status = apr_brigade_puts(bb, NULL, NULL, "0\r\n");
  98             apr_table_do(input_ser_header, task, t, NULL);
  99             status = apr_brigade_puts(bb, NULL, NULL, "\r\n");
 100         }
 101         else {
 102             status = apr_brigade_puts(bb, NULL, NULL, "0\r\n\r\n");
 103         }
 104         APR_BRIGADE_CONCAT(bb, tmp);
 105         apr_brigade_destroy(tmp);
 106     }
 107     else if (r && t && !apr_is_empty_table(t)){
 108         /* trailers passed in directly. */
 109         apr_table_overlap(r->trailers_in, t, APR_OVERLAP_TABLES_SET);
 110     }
 111     task->input.eos_written = 1;
 112     return status;
 113 }
 114
 115 static apr_status_t input_append_eos(h2_task *task, request_rec *r)
 116 {
 117     apr_status_t status = APR_SUCCESS;
 118     apr_bucket_brigade *bb = task->input.bb;
 119     apr_table_t *t = task->request? task->request->trailers : NULL;
 120
 121     if (task->input.chunked) {
 122         if (t && !apr_is_empty_table(t)) {
 123             status = apr_brigade_puts(bb, NULL, NULL, "0\r\n");
 124             apr_table_do(input_ser_header, task, t, NULL);
 125             status = apr_brigade_puts(bb, NULL, NULL, "\r\n");
 126         }
 127         else {
 128             status = apr_brigade_puts(bb, NULL, NULL, "0\r\n\r\n");
 129         }
 130     }
 131     else if (r && t && !apr_is_empty_table(t)){
 132         /* trailers passed in directly. */
 133         apr_table_overlap(r->trailers_in, t, APR_OVERLAP_TABLES_SET);
 134     }
 135     APR_BRIGADE_INSERT_TAIL(bb, apr_bucket_eos_create(bb->bucket_alloc));
 136     task->input.eos_written = 1;
 137     return status;
 138 }
 139
 140 static apr_status_t input_read(h2_task *task, ap_filter_t* f,
 141                                apr_bucket_brigade* bb, ap_input_mode_t mode,
 142                                apr_read_type_e block, apr_off_t readbytes)
 143 {
 144     apr_status_t status = APR_SUCCESS;
 145     apr_bucket *b, *next, *first_data;
 146     apr_off_t bblen = 0;
 147
 148     ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
 149                   "h2_task(%s): read, mode=%d, block=%d, readbytes=%ld",
 150                   task->id, mode, block, (long)readbytes);
 151
 152     if (mode == AP_MODE_INIT) {
 153         return ap_get_brigade(f->c->input_filters, bb, mode, block, readbytes);
 154     }
 155
 156     if (f->c->aborted || !task->request) {
 157         return APR_ECONNABORTED;
 158     }
 159
 160     if (!task->input.bb) {
 161         if (!task->input.eos_written) {
 162             input_append_eos(task, f->r);
 163             return APR_SUCCESS;
 164         }
 165         return APR_EOF;
 166     }
 167
 168     /*
 169     if (f->r && f->r->expecting_100) {
 170         ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, f->c,
 171                       "h2_task(%s): need to send 100 Continue here",
 172                       task->id);
 173         f->r->expecting_100 = 0;
 174     }
 175     if (task->r && task->r->expecting_100) {
 176         ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, f->c,
 177                       "h2_task2(%s): need to send 100 Continue here",
 178                       task->id);
 179         task->r->expecting_100 = 0;
 180     }
 181     */
 182
 183     /* Cleanup brigades from those nasty 0 length non-meta buckets
 184      * that apr_brigade_split_line() sometimes produces. */
 185     for (b = APR_BRIGADE_FIRST(task->input.bb);
 186          b != APR_BRIGADE_SENTINEL(task->input.bb); b = next) {
 187         next = APR_BUCKET_NEXT(b);
 188         if (b->length == 0 && !APR_BUCKET_IS_METADATA(b)) {
 189             apr_bucket_delete(b);
 190         }
 191     }
 192
 193     while (APR_BRIGADE_EMPTY(task->input.bb) && !task->input.eos) {
 194         /* Get more input data for our request. */
 195         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
 196                       "h2_task(%s): get more data from mplx, block=%d, "
 197                       "readbytes=%ld, queued=%ld",
 198                       task->id, block, (long)readbytes, (long)bblen);
 199
 200         /* Override the block mode we get called with depending on the input's
 201          * setting. */
 202         if (task->input.beam) {
 203             status = h2_beam_receive(task->input.beam, task->input.bb, block,
 204                                      H2MIN(readbytes, 32*1024));
 205         }
 206         else {
 207             status = APR_EOF;
 208         }
 209
 210         ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, f->c,
 211                       "h2_task(%s): read returned", task->id);
 212         if (APR_STATUS_IS_EAGAIN(status)
 213             && (mode == AP_MODE_GETLINE || block == APR_BLOCK_READ)) {
 214             /* chunked input handling does not seem to like it if we
 215              * return with APR_EAGAIN from a GETLINE read...
 216              * upload 100k test on test-ser.example.org hangs */
 217             status = APR_SUCCESS;
 218         }
 219         else if (APR_STATUS_IS_EOF(status)) {
 220             task->input.eos = 1;
 221         }
 222         else if (status != APR_SUCCESS) {
 223             return status;
 224         }
 225
 226         /* Inspect the buckets received, detect EOS and apply
 227          * chunked encoding if necessary */
 228         h2_util_bb_log(f->c, task->stream_id, APLOG_TRACE2,
 229                        "input.beam recv raw", task->input.bb);
 230         first_data = NULL;
 231         bblen = 0;
 232         for (b = APR_BRIGADE_FIRST(task->input.bb);
 233              b != APR_BRIGADE_SENTINEL(task->input.bb); b = next) {
 234             next = APR_BUCKET_NEXT(b);
 235             if (APR_BUCKET_IS_METADATA(b)) {
 236                 if (first_data && task->input.chunked) {
 237                     make_chunk(task, task->input.bb, first_data, bblen, b);
 238                     first_data = NULL;
 239                     bblen = 0;
 240                 }
 241                 if (APR_BUCKET_IS_EOS(b)) {
 242                     task->input.eos = 1;
 243                     input_handle_eos(task, f->r, b);
 244                     h2_util_bb_log(f->c, task->stream_id, APLOG_TRACE2,
 245                                    "input.bb after handle eos",
 246                                    task->input.bb);
 247                 }
 248             }
 249             else if (b->length == 0) {
 250                 apr_bucket_delete(b);
 251             }
 252             else {
 253                 if (!first_data) {
 254                     first_data = b;
 255                 }
 256                 bblen += b->length;
 257             }
 258         }
 259         if (first_data && task->input.chunked) {
 260             make_chunk(task, task->input.bb, first_data, bblen, NULL);
 261         }
 262
 263         if (h2_task_logio_add_bytes_in) {
 264             h2_task_logio_add_bytes_in(f->c, bblen);
 265         }
 266     }
 267
 268     if (task->input.eos) {
 269         if (!task->input.eos_written) {
 270             input_append_eos(task, f->r);
 271         }
 272         if (APR_BRIGADE_EMPTY(task->input.bb)) {
 273             return APR_EOF;
 274         }
 275     }
 276
 277     h2_util_bb_log(f->c, task->stream_id, APLOG_TRACE2,
 278                    "task_input.bb", task->input.bb);
 279
 280     if (APR_BRIGADE_EMPTY(task->input.bb)) {
 281         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
 282                       "h2_task(%s): no data", task->id);
 283         return (block == APR_NONBLOCK_READ)? APR_EAGAIN : APR_EOF;
 284     }
 285
 286     if (mode == AP_MODE_EXHAUSTIVE) {
 287         /* return all we have */
 288         APR_BRIGADE_CONCAT(bb, task->input.bb);
 289     }
 290     else if (mode == AP_MODE_READBYTES) {
 291         status = h2_brigade_concat_length(bb, task->input.bb, readbytes);
 292     }
 293     else if (mode == AP_MODE_SPECULATIVE) {
 294         status = h2_brigade_copy_length(bb, task->input.bb, readbytes);
 295     }
 296     else if (mode == AP_MODE_GETLINE) {
 297         /* we are reading a single LF line, e.g. the HTTP headers.
 298          * this has the nasty side effect to split the bucket, even
 299          * though it ends with CRLF and creates a 0 length bucket */
 300         status = apr_brigade_split_line(bb, task->input.bb, block,
 301                                         HUGE_STRING_LEN);
 302         if (APLOGctrace1(f->c)) {
 303             char buffer[1024];
 304             apr_size_t len = sizeof(buffer)-1;
 305             apr_brigade_flatten(bb, buffer, &len);
 306             buffer[len] = 0;
 307             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
 308                           "h2_task(%s): getline: %s",
 309                           task->id, buffer);
 310         }
 311     }
 312     else {
 313         /* Hmm, well. There is mode AP_MODE_EATCRLF, but we chose not
 314          * to support it. Seems to work. */
 315         ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOTIMPL, f->c,
 316                       APLOGNO(02942)
 317                       "h2_task, unsupported READ mode %d", mode);
 318         status = APR_ENOTIMPL;
 319     }
 320
 321     if (APLOGctrace1(f->c)) {
 322         apr_brigade_length(bb, 0, &bblen);
 323         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
 324                       "h2_task(%s): return %ld data bytes",
 325                       task->id, (long)bblen);
 326     }
 327     return status;
 328 }
 329
 330 /*******************************************************************************
 331  * task output handling
 332  ******************************************************************************/
 333
 334 static apr_status_t open_response(h2_task *task, h2_response *response)
 335 {
 336     if (!response) {
 337         /* This happens currently when ap_die(status, r) is invoked
 338          * by a read request filter. */
 339         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c, APLOGNO(03204)
 340                       "h2_task(%s): write without response for %s %s %s",
 341                       task->id,
 342                       task->request->method,
 343                       task->request->authority,
 344                       task->request->path);
 345         task->c->aborted = 1;
 346         return APR_ECONNABORTED;
 347     }
 348
 349     if (h2_task_logio_add_bytes_out) {
 350         /* count headers as if we'd do a HTTP/1.1 serialization */
 351         task->output.written = h2_util_table_bytes(response->headers, 3)+1;
 352         h2_task_logio_add_bytes_out(task->c, task->output.written);
 353     }
 354     ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c, APLOGNO(03348)
 355                   "h2_task(%s): open response to %s %s %s",
 356                   task->id, task->request->method,
 357                   task->request->authority,
 358                   task->request->path);
 359     return h2_mplx_out_open(task->mplx, task->stream_id, response);
 360 }
 361
 362 static apr_status_t send_out(h2_task *task, apr_bucket_brigade* bb)
 363 {
 364     apr_off_t written, left;
 365     apr_status_t status;
 366
 367     apr_brigade_length(bb, 0, &written);
 368     ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 369                   "h2_task(%s): write response body (%ld bytes)",
 370                   task->id, (long)written);
 371
 372     status = h2_beam_send(task->output.beam, bb,
 373                           task->blocking? APR_BLOCK_READ
 374                           : APR_NONBLOCK_READ);
 375     if (APR_STATUS_IS_EAGAIN(status)) {
 376         apr_brigade_length(bb, 0, &left);
 377         written -= left;
 378         status = APR_SUCCESS;
 379     }
 380     if (status == APR_SUCCESS) {
 381         task->output.written += written;
 382         if (h2_task_logio_add_bytes_out) {
 383             h2_task_logio_add_bytes_out(task->c, written);
 384         }
 385     }
 386     else {
 387         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, task->c,
 388                       "h2_task(%s): send_out (%ld bytes)",
 389                       task->id, (long)written);
 390     }
 391     return status;
 392 }
 393
 394 /* Bring the data from the brigade (which represents the result of the
 395  * request_rec out filter chain) into the h2_mplx for further sending
 396  * on the master connection.
 397  */
 398 static apr_status_t output_write(h2_task *task, ap_filter_t* f,
 399                                  apr_bucket_brigade* bb)
 400 {
 401     apr_bucket *b;
 402     apr_status_t status = APR_SUCCESS;
 403     int flush = 0;
 404
 405     if (APR_BRIGADE_EMPTY(bb)) {
 406         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 407                       "h2_task(%s): empty write", task->id);
 408         return APR_SUCCESS;
 409     }
 410
 411     if (task->frozen) {
 412         h2_util_bb_log(task->c, task->stream_id, APLOG_TRACE2,
 413                        "frozen task output write, ignored", bb);
 414         while (!APR_BRIGADE_EMPTY(bb)) {
 415             b = APR_BRIGADE_FIRST(bb);
 416             if (AP_BUCKET_IS_EOR(b)) {
 417                 APR_BUCKET_REMOVE(b);
 418                 task->eor = b;
 419             }
 420             else {
 421                 apr_bucket_delete(b);
 422             }
 423         }
 424         return APR_SUCCESS;
 425     }
 426
 427     if (!task->output.beam) {
 428         h2_beam_create(&task->output.beam, task->pool,
 429                        task->stream_id, "output", 0);
 430         if (task->output.copy_files) {
 431             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 432                           "h2_task(%s): copy_files on", task->id);
 433             h2_beam_on_file_beam(task->output.beam, h2_beam_no_files, NULL);
 434         }
 435     }
 436
 437     /* Attempt to write saved brigade first */
 438     if (task->output.bb && !APR_BRIGADE_EMPTY(task->output.bb)) {
 439         status = send_out(task, task->output.bb);
 440         if (status != APR_SUCCESS) {
 441             return status;
 442         }
 443     }
 444
 445     /* If there is nothing saved (anymore), try to write the brigade passed */
 446     if ((!task->output.bb || APR_BRIGADE_EMPTY(task->output.bb))
 447         && !APR_BRIGADE_EMPTY(bb)) {
 448         /* check if we have a flush before the end-of-request */
 449         if (!task->output.response_open) {
 450             for (b = APR_BRIGADE_FIRST(bb);
 451                  b != APR_BRIGADE_SENTINEL(bb);
 452                  b = APR_BUCKET_NEXT(b)) {
 453                 if (AP_BUCKET_IS_EOR(b)) {
 454                     break;
 455                 }
 456                 else if (APR_BUCKET_IS_FLUSH(b)) {
 457                     flush = 1;
 458                 }
 459             }
 460         }
 461
 462         status = send_out(task, bb);
 463         if (status != APR_SUCCESS) {
 464             return status;
 465         }
 466     }
 467
 468     /* If the passed brigade is not empty, save it before return */
 469     if (!APR_BRIGADE_EMPTY(bb)) {
 470         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, task->c, APLOGNO(03405)
 471                       "h2_task(%s): could not write all, saving brigade",
 472                       task->id);
 473         if (!task->output.bb) {
 474             task->output.bb = apr_brigade_create(task->pool,
 475                                           task->c->bucket_alloc);
 476         }
 477         return ap_save_brigade(f, &task->output.bb, &bb, task->pool);
 478     }
 479
 480     if (!task->output.response_open
 481         && (flush || h2_beam_get_mem_used(task->output.beam) > (32*1024))) {
 482         /* if we have enough buffered or we got a flush bucket, open
 483         * the response now. */
 484         status = open_response(task,
 485             h2_from_h1_get_response(task->output.from_h1));
 486         task->output.response_open = 1;
 487     }
 488
 489     return status;
 490 }
 491
 492 static apr_status_t output_finish(h2_task *task)
 493 {
 494     apr_status_t status = APR_SUCCESS;
 495
 496     if (!task->output.response_open) {
 497         status = open_response(task,
 498             h2_from_h1_get_response(task->output.from_h1));
 499         task->output.response_open = 1;
 500     }
 501     return status;
 502 }
 503
 504 /*******************************************************************************
 505  * task slave connection filters
 506  ******************************************************************************/
 507
 508 static apr_status_t h2_filter_stream_input(ap_filter_t* filter,
 509                                            apr_bucket_brigade* brigade,
 510                                            ap_input_mode_t mode,
 511                                            apr_read_type_e block,
 512                                            apr_off_t readbytes)
 513 {
 514     h2_task *task = h2_ctx_cget_task(filter->c);
 515     AP_DEBUG_ASSERT(task);
 516     return input_read(task, filter, brigade, mode, block, readbytes);
 517 }
 518
 519 static apr_status_t h2_filter_continue(ap_filter_t* f,
 520                                        apr_bucket_brigade* brigade,
 521                                        ap_input_mode_t mode,
 522                                        apr_read_type_e block,
 523                                        apr_off_t readbytes)
 524 {
 525     h2_task *task = h2_ctx_cget_task(f->c);
 526     apr_status_t status;
 527
 528     AP_DEBUG_ASSERT(task);
 529     if (f->r->expecting_100 && ap_is_HTTP_SUCCESS(f->r->status)) {
 530         h2_response *response;
 531
 532         response = h2_response_rcreate(task->stream_id, f->r, HTTP_CONTINUE,
 533                                        NULL, f->r->pool);
 534         ap_log_rerror(APLOG_MARK, APLOG_TRACE1, 0, f->r,
 535                       "h2_task(%s): send 100 Continue", task->id);
 536         status = open_response(task, response);
 537         if (status != APR_SUCCESS) {
 538             return status;
 539         }
 540         f->r->expecting_100 = 0;
 541         apr_table_clear(f->r->headers_out);
 542     }
 543     return ap_get_brigade(f->next, brigade, mode, block, readbytes);
 544 }
 545
 546 static apr_status_t h2_filter_stream_output(ap_filter_t* filter,
 547                                             apr_bucket_brigade* brigade)
 548 {
 549     h2_task *task = h2_ctx_cget_task(filter->c);
 550     apr_status_t status;
 551
 552     ap_assert(task);
 553     status = output_write(task, filter, brigade);
 554     if (status != APR_SUCCESS) {
 555         h2_task_rst(task, H2_ERR_INTERNAL_ERROR);
 556     }
 557     return status;
 558 }
 559
 560 static apr_status_t h2_filter_read_response(ap_filter_t* filter,
 561                                             apr_bucket_brigade* bb)
 562 {
 563     h2_task *task = h2_ctx_cget_task(filter->c);
 564     AP_DEBUG_ASSERT(task);
 565     if (!task->output.from_h1) {
 566         return APR_ECONNABORTED;
 567     }
 568     return h2_from_h1_read_response(task->output.from_h1, filter, bb);
 569 }
 570
 571 /*******************************************************************************
 572  * task things
 573  ******************************************************************************/
 574
 575 apr_status_t h2_task_add_response(h2_task *task, h2_response *response)
 576 {
 577     AP_DEBUG_ASSERT(response);
 578     /* we used to clone the response into out own pool. But
 579      * we have much tighter control over the EOR bucket nowadays,
 580      * so just use the instance given */
 581     response->next = task->response;
 582     task->response = response;
 583     if (response->rst_error) {
 584         h2_task_rst(task, response->rst_error);
 585     }
 586     return APR_SUCCESS;
 587 }
 588
 589
 590 int h2_task_can_redo(h2_task *task) {
 591     if (task->response_sent
 592         || (task->input.beam && h2_beam_was_received(task->input.beam))
 593         || !task->request) {
 594         /* cannot repeat that. */
 595         return 0;
 596     }
 597     return (!strcmp("GET", task->request->method)
 598             || !strcmp("HEAD", task->request->method)
 599             || !strcmp("OPTIONS", task->request->method));
 600 }
 601
 602 void h2_task_redo(h2_task *task)
 603 {
 604     task->response = NULL;
 605     task->rst_error = 0;
 606 }
 607
 608 void h2_task_rst(h2_task *task, int error)
 609 {
 610     task->rst_error = error;
 611     if (task->input.beam) {
 612         h2_beam_abort(task->input.beam);
 613     }
 614     if (task->output.beam) {
 615         h2_beam_abort(task->output.beam);
 616     }
 617     if (task->c) {
 618         task->c->aborted = 1;
 619     }
 620 }
 621
 622 /*******************************************************************************
 623  * Register various hooks
 624  */
 625 static const char *const mod_ssl[]        = { "mod_ssl.c", NULL};
 626 static int h2_task_pre_conn(conn_rec* c, void *arg);
 627 static int h2_task_process_conn(conn_rec* c);
 628
 629 APR_OPTIONAL_FN_TYPE(ap_logio_add_bytes_in) *h2_task_logio_add_bytes_in;
 630 APR_OPTIONAL_FN_TYPE(ap_logio_add_bytes_out) *h2_task_logio_add_bytes_out;
 631
 632 void h2_task_register_hooks(void)
 633 {
 634     /* This hook runs on new connections before mod_ssl has a say.
 635      * Its purpose is to prevent mod_ssl from touching our pseudo-connections
 636      * for streams.
 637      */
 638     ap_hook_pre_connection(h2_task_pre_conn,
 639                            NULL, mod_ssl, APR_HOOK_FIRST);
 640     /* When the connection processing actually starts, we might
 641      * take over, if the connection is for a task.
 642      */
 643     ap_hook_process_connection(h2_task_process_conn,
 644                                NULL, NULL, APR_HOOK_FIRST);
 645
 646     ap_register_output_filter("H2_RESPONSE", h2_response_output_filter,
 647                               NULL, AP_FTYPE_PROTOCOL);
 648     ap_register_input_filter("H2_TO_H1", h2_filter_stream_input,
 649                              NULL, AP_FTYPE_NETWORK);
 650     ap_register_input_filter("H2_CONTINUE", h2_filter_continue,
 651                              NULL, AP_FTYPE_PROTOCOL);
 652     ap_register_output_filter("H1_TO_H2", h2_filter_stream_output,
 653                               NULL, AP_FTYPE_NETWORK);
 654     ap_register_output_filter("H1_TO_H2_RESP", h2_filter_read_response,
 655                               NULL, AP_FTYPE_PROTOCOL);
 656     ap_register_output_filter("H2_TRAILERS", h2_response_trailers_filter,
 657                               NULL, AP_FTYPE_PROTOCOL);
 658 }
 659
 660 /* post config init */
 661 apr_status_t h2_task_init(apr_pool_t *pool, server_rec *s)
 662 {
 663     h2_task_logio_add_bytes_in = APR_RETRIEVE_OPTIONAL_FN(ap_logio_add_bytes_in);
 664     h2_task_logio_add_bytes_out = APR_RETRIEVE_OPTIONAL_FN(ap_logio_add_bytes_out);
 665
 666     return APR_SUCCESS;
 667 }
 668
 669 static int h2_task_pre_conn(conn_rec* c, void *arg)
 670 {
 671     h2_ctx *ctx;
 672
 673     if (!c->master) {
 674         return OK;
 675     }
 676
 677     ctx = h2_ctx_get(c, 0);
 678     (void)arg;
 679     if (h2_ctx_is_task(ctx)) {
 680         ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c,
 681                       "h2_h2, pre_connection, found stream task");
 682
 683         /* Add our own, network level in- and output filters.
 684          */
 685         ap_add_input_filter("H2_TO_H1", NULL, NULL, c);
 686         ap_add_output_filter("H1_TO_H2", NULL, NULL, c);
 687     }
 688     return OK;
 689 }
 690
 691 h2_task *h2_task_create(conn_rec *c, const h2_request *req,
 692                         h2_bucket_beam *input, h2_mplx *mplx)
 693 {
 694     apr_pool_t *pool;
 695     h2_task *task;
 696
 697     apr_pool_create(&pool, c->pool);
 698     task = apr_pcalloc(pool, sizeof(h2_task));
 699     if (task == NULL) {
 700         ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, c,
 701                       APLOGNO(02941) "h2_task(%ld-%d): create stream task",
 702                       c->id, req->id);
 703         return NULL;
 704     }
 705
 706     task->id          = apr_psprintf(pool, "%ld-%d", c->id, req->id);
 707     task->stream_id   = req->id;
 708     task->c           = c;
 709     task->mplx        = mplx;
 710     task->c->keepalives = mplx->c->keepalives;
 711     task->pool        = pool;
 712     task->request     = req;
 713     task->ser_headers = req->serialize;
 714     task->blocking    = 1;
 715     task->input.beam  = input;
 716
 717     apr_thread_cond_create(&task->cond, pool);
 718
 719     h2_ctx_create_for(c, task);
 720     return task;
 721 }
 722
 723 void h2_task_destroy(h2_task *task)
 724 {
 725     if (task->output.beam) {
 726         h2_beam_destroy(task->output.beam);
 727         task->output.beam = NULL;
 728     }
 729     if (task->eor) {
 730         apr_bucket_destroy(task->eor);
 731     }
 732     if (task->pool) {
 733         apr_pool_destroy(task->pool);
 734     }
 735 }
 736
 737 void h2_task_set_io_blocking(h2_task *task, int blocking)
 738 {
 739     task->blocking = blocking;
 740 }
 741
 742 apr_status_t h2_task_do(h2_task *task, apr_thread_t *thread)
 743 {
 744     AP_DEBUG_ASSERT(task);
 745
 746     task->input.block = APR_BLOCK_READ;
 747     task->input.chunked = task->request->chunked;
 748     task->input.eos = !task->request->body;
 749     if (task->input.eos && !task->input.chunked && !task->ser_headers) {
 750         /* We do not serialize/chunk and have eos already, no need to
 751          * create a bucket brigade. */
 752         task->input.bb = NULL;
 753         task->input.eos_written = 1;
 754     }
 755     else {
 756         task->input.bb = apr_brigade_create(task->pool, task->c->bucket_alloc);
 757         if (task->ser_headers) {
 758             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 759                           "h2_task(%s): serialize request %s %s",
 760                           task->id, task->request->method, task->request->path);
 761             apr_brigade_printf(task->input.bb, NULL,
 762                                NULL, "%s %s HTTP/1.1\r\n",
 763                                task->request->method, task->request->path);
 764             apr_table_do(input_ser_header, task, task->request->headers, NULL);
 765             apr_brigade_puts(task->input.bb, NULL, NULL, "\r\n");
 766         }
 767         if (task->input.eos) {
 768             input_append_eos(task, NULL);
 769         }
 770     }
 771
 772     task->output.from_h1 = h2_from_h1_create(task->stream_id, task->pool);
 773
 774     ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 775                   "h2_task(%s): process connection", task->id);
 776     task->c->current_thread = thread;
 777     ap_run_process_connection(task->c);
 778
 779     if (task->frozen) {
 780         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 781                       "h2_task(%s): process_conn returned frozen task",
 782                       task->id);
 783         /* cleanup delayed */
 784         return APR_EAGAIN;
 785     }
 786     else {
 787         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
 788                       "h2_task(%s): processing done", task->id);
 789         return output_finish(task);
 790     }
 791 }
 792
 793 static apr_status_t h2_task_process_request(h2_task *task, conn_rec *c)
 794 {
 795     const h2_request *req = task->request;
 796     conn_state_t *cs = c->cs;
 797     request_rec *r;
 798
 799     ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 800                   "h2_task(%s): create request_rec", task->id);
 801     r = h2_request_create_rec(req, c);
 802     if (r && (r->status == HTTP_OK)) {
 803         ap_update_child_status(c->sbh, SERVER_BUSY_WRITE, r);
 804
 805         if (cs) {
 806             cs->state = CONN_STATE_HANDLER;
 807         }
 808         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 809                       "h2_task(%s): start process_request", task->id);
 810         task->r = r;
 811
 812         ap_process_request(r);
 813         if (task->frozen) {
 814             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 815                           "h2_task(%s): process_request frozen", task->id);
 816         }
 817         else {
 818             task->r = NULL;
 819         }
 820         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 821                       "h2_task(%s): process_request done", task->id);
 822
 823         /* After the call to ap_process_request, the
 824          * request pool will have been deleted.  We set
 825          * r=NULL here to ensure that any dereference
 826          * of r that might be added later in this function
 827          * will result in a segfault immediately instead
 828          * of nondeterministic failures later.
 829          */
 830         if (cs)
 831             cs->state = CONN_STATE_WRITE_COMPLETION;
 832         r = NULL;
 833     }
 834     else if (!r) {
 835         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 836                       "h2_task(%s): create request_rec failed, r=NULL", task->id);
 837     }
 838     else {
 839         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 840                       "h2_task(%s): create request_rec failed, r->status=%d",
 841                       task->id, r->status);
 842     }
 843
 844     return APR_SUCCESS;
 845 }
 846
 847 static int h2_task_process_conn(conn_rec* c)
 848 {
 849     h2_ctx *ctx;
 850
 851     if (!c->master) {
 852         return DECLINED;
 853     }
 854
 855     ctx = h2_ctx_get(c, 0);
 856     if (h2_ctx_is_task(ctx)) {
 857         if (!ctx->task->ser_headers) {
 858             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 859                           "h2_h2, processing request directly");
 860             h2_task_process_request(ctx->task, c);
 861             return DONE;
 862         }
 863         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 864                       "h2_task(%s), serialized handling", ctx->task->id);
 865     }
 866     else {
 867         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
 868                       "slave_conn(%ld): has no task", c->id);
 869     }
 870     return DECLINED;
 871 }
 872
 873 apr_status_t h2_task_freeze(h2_task *task)
 874 {
 875     if (!task->frozen) {
 876         task->frozen = 1;
 877         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c, APLOGNO(03406)
 878                       "h2_task(%s), frozen", task->id);
 879     }
 880     return APR_SUCCESS;
 881 }
 882
 883 apr_status_t h2_task_thaw(h2_task *task)
 884 {
 885     if (task->frozen) {
 886         task->frozen = 0;
 887         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c, APLOGNO(03407)
 888                       "h2_task(%s), thawed", task->id);
 889     }
 890     task->detached = 1;
 891     return APR_SUCCESS;
 892 }
 893
 894 int h2_task_is_detached(h2_task *task)
 895 {
 896     return task->detached;
 897 }