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