mod_http2: more orderly destruction of stream/task pairs

git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@1747531 13f79535-47bb-0310-9956-ffa450edef68

CHANGES

@@ -1,6 +1,9 @@
                                                          -*- coding: utf-8 -*-
 Changes with Apache 2.5.0
 
+  *) mod_http2: more orderly destruction of stream/task pairs, addressing
+     reported crashes on aborted connections. [Stefan Eissing]
+
   *) mod_dav: Add support for childtags to dav_error.
      [Jari Urpalainen <jari.urpalainen nokia.com>]
 

modules/http2/h2_bucket_beam.c

@@ -407,6 +407,12 @@ static apr_status_t beam_cleanup(void *data)
     r_purge_reds(beam);
     h2_blist_cleanup(&beam->red);
     report_consumption(beam, 0);
+    while (!H2_BPROXY_LIST_EMPTY(&beam->proxies)) {
+        h2_beam_proxy *proxy = H2_BPROXY_LIST_FIRST(&beam->proxies);
+        H2_BPROXY_REMOVE(proxy);
+        proxy->beam = NULL;
+        proxy->bred = NULL;
+    }
     h2_blist_cleanup(&beam->purge);
     h2_blist_cleanup(&beam->hold);
 
@@ -534,16 +540,18 @@ apr_status_t h2_beam_close(h2_bucket_beam *beam)
     return beam->aborted? APR_ECONNABORTED : APR_SUCCESS;
 }
 
-apr_status_t h2_beam_shutdown(h2_bucket_beam *beam, apr_read_type_e block)
+apr_status_t h2_beam_shutdown(h2_bucket_beam *beam, apr_read_type_e block,
+                              int clear_buffers)
 {
     apr_status_t status;
     h2_beam_lock bl;
 
     if ((status = enter_yellow(beam, &bl)) == APR_SUCCESS) {
-        r_purge_reds(beam);
-        h2_blist_cleanup(&beam->red);
+        if (clear_buffers) {
+            r_purge_reds(beam);
+            h2_blist_cleanup(&beam->red);
+        }
         beam_close(beam);
-        report_consumption(beam, 0);
 
         while (status == APR_SUCCESS 
                && (!H2_BPROXY_LIST_EMPTY(&beam->proxies)

modules/http2/h2_bucket_beam.h

@@ -282,16 +282,17 @@ void h2_beam_abort(h2_bucket_beam *beam);
 apr_status_t h2_beam_close(h2_bucket_beam *beam);
 
 /**
- * Empty any buffered data and return APR_SUCCESS when all buckets
- * in transit have been handled. When called with APR_BLOCK_READ and
- * with a mutex installed, will wait until this is the case. Otherwise
- * APR_EAGAIN is returned.
+ * Return APR_SUCCESS when all buckets in transit have been handled. 
+ * When called with APR_BLOCK_READ and a mutex set, will wait until the green
+ * side has consumed all data. Otherwise APR_EAGAIN is returned.
+ * With clear_buffers set, any queued data is discarded.
  * If a timeout is set on the beam, waiting might also time out and
  * return APR_ETIMEUP.
  *
  * Call from the red side only.
  */
-apr_status_t h2_beam_shutdown(h2_bucket_beam *beam, apr_read_type_e block);
+apr_status_t h2_beam_shutdown(h2_bucket_beam *beam, apr_read_type_e block,
+                              int clear_buffers);
 
 void h2_beam_mutex_set(h2_bucket_beam *beam, 
                        h2_beam_mutex_enter m_enter,

modules/http2/h2_mplx.c

@@ -200,7 +200,7 @@ static int purge_stream(void *ctx, void *val)
     h2_ihash_remove(m->spurge, stream->id);
     h2_stream_destroy(stream);
     if (task) {
-        task_destroy(m, task, 0);
+        task_destroy(m, task, 1);
     }
     return 0;
 }
@@ -348,15 +348,6 @@ static void task_destroy(h2_mplx *m, h2_task *task, int called_from_master)
 
     ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, m->c, 
                   "h2_task(%s): destroy", task->id);
-    /* cleanup any buffered input */
-    if (task->input.beam) {
-        status = h2_beam_shutdown(task->input.beam, APR_NONBLOCK_READ);
-        if (status != APR_SUCCESS){
-            ap_log_cerror(APLOG_MARK, APLOG_WARNING, status, m->c, APLOGNO(03385) 
-                          "h2_task(%s): input shutdown", task->id);
-        }
-    }
-
     if (called_from_master) {
         /* Process outstanding events before destruction */
         h2_stream *stream = h2_ihash_get(m->streams, task->stream_id);
@@ -372,6 +363,12 @@ static void task_destroy(h2_mplx *m, h2_task *task, int called_from_master)
         m->tx_handles_reserved += 
         h2_beam_get_files_beamed(task->output.beam);
         h2_beam_on_produced(task->output.beam, NULL, NULL);
+        status = h2_beam_shutdown(task->output.beam, APR_NONBLOCK_READ, 1);
+        if (status != APR_SUCCESS){
+            ap_log_cerror(APLOG_MARK, APLOG_WARNING, status, m->c, 
+                          APLOGNO(03385) "h2_task(%s): output shutdown "
+                          "incomplete", task->id);
+        }
     }
 
     slave = task->c;
@@ -438,6 +435,8 @@ static void stream_done(h2_mplx *m, h2_stream *stream, int rst_error)
     h2_ihash_remove(m->streams, stream->id);
     if (stream->input) {
         m->tx_handles_reserved += h2_beam_get_files_beamed(stream->input);
+        h2_beam_on_consumed(stream->input, NULL, NULL);
+        h2_beam_mutex_set(stream->input, NULL, NULL, NULL);
     }
     h2_stream_cleanup(stream);
 
@@ -651,6 +650,7 @@ apr_status_t h2_mplx_stream_done(h2_mplx *m, h2_stream *stream)
                       "h2_mplx(%ld-%d): marking stream as done.", 
                       m->id, stream->id);
         stream_done(m, stream, stream->rst_error);
+        purge_streams(m);
         leave_mutex(m, acquired);
     }
     return status;
@@ -766,6 +766,7 @@ apr_status_t h2_mplx_out_trywait(h2_mplx *m, apr_interval_time_t timeout,
             status = APR_SUCCESS;
         }
         else {
+            purge_streams(m);
             m->added_output = iowait;
             status = apr_thread_cond_timedwait(m->added_output, m->lock, timeout);
             if (APLOGctrace2(m->c)) {
@@ -1021,20 +1022,18 @@ static void task_done(h2_mplx *m, h2_task *task, h2_req_engine *ngn)
             }
         }
         else {
-            /* stream done, was it placed in hold? */
+            /* stream no longer active, was it placed in hold? */
             stream = h2_ihash_get(m->shold, task->stream_id);
             if (stream) {
                 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c,
                               "h2_mplx(%s): task_done, stream in hold", 
                               task->id);
-                stream->response = NULL; /* ref from task memory */
                 /* We cannot destroy the stream here since this is 
                  * called from a worker thread and freeing memory pools
                  * is only safe in the only thread using it (and its
                  * parent pool / allocator) */
                 h2_ihash_remove(m->shold, stream->id);
                 h2_ihash_add(m->spurge, stream);
-                task_destroy(m, task, 0);
             }
             else {
                 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c,

modules/http2/h2_stream.c

@@ -213,12 +213,12 @@ void h2_stream_cleanup(h2_stream *stream)
     }
     if (stream->input) {
         apr_status_t status;
-        status = h2_beam_shutdown(stream->input, APR_NONBLOCK_READ);
+        status = h2_beam_shutdown(stream->input, APR_NONBLOCK_READ, 1);
         if (status == APR_EAGAIN) {
             ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c, 
                           "h2_stream(%ld-%d): wait on input shutdown", 
                           stream->session->id, stream->id);
-            status = h2_beam_shutdown(stream->input, APR_BLOCK_READ);
+            status = h2_beam_shutdown(stream->input, APR_BLOCK_READ, 1);
             ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, stream->session->c, 
                           "h2_stream(%ld-%d): input shutdown returned", 
                           stream->session->id, stream->id);