granicus.if.org Git - apache/blob - modules/http2/h2_filter.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
  18 #include <apr_strings.h>
  19 #include <httpd.h>
  20 #include <http_core.h>
  21 #include <http_log.h>
  22 #include <http_connection.h>
  23 #include <scoreboard.h>
  24
  25 #include "h2_private.h"
  26 #include "h2.h"
  27 #include "h2_config.h"
  28 #include "h2_conn_io.h"
  29 #include "h2_ctx.h"
  30 #include "h2_mplx.h"
  31 #include "h2_push.h"
  32 #include "h2_task.h"
  33 #include "h2_stream.h"
  34 #include "h2_request.h"
  35 #include "h2_response.h"
  36 #include "h2_stream.h"
  37 #include "h2_session.h"
  38 #include "h2_util.h"
  39 #include "h2_version.h"
  40
  41 #include "h2_filter.h"
  42
  43 #define UNSET       -1
  44 #define H2MIN(x,y) ((x) < (y) ? (x) : (y))
  45
  46 static apr_status_t consume_brigade(h2_filter_cin *cin,
  47                                     apr_bucket_brigade *bb,
  48                                     apr_read_type_e block)
  49 {
  50     apr_status_t status = APR_SUCCESS;
  51     apr_size_t readlen = 0;
  52
  53     while (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(bb)) {
  54
  55         apr_bucket* bucket = APR_BRIGADE_FIRST(bb);
  56         if (APR_BUCKET_IS_METADATA(bucket)) {
  57             /* we do nothing regarding any meta here */
  58         }
  59         else {
  60             const char *bucket_data = NULL;
  61             apr_size_t bucket_length = 0;
  62             status = apr_bucket_read(bucket, &bucket_data,
  63                                      &bucket_length, block);
  64
  65             if (status == APR_SUCCESS && bucket_length > 0) {
  66                 apr_size_t consumed = 0;
  67
  68                 status = cin->cb(cin->cb_ctx, bucket_data, bucket_length, &consumed);
  69                 if (status == APR_SUCCESS && bucket_length > consumed) {
  70                     /* We have data left in the bucket. Split it. */
  71                     status = apr_bucket_split(bucket, consumed);
  72                 }
  73                 readlen += consumed;
  74                 cin->start_read = apr_time_now();
  75             }
  76         }
  77         apr_bucket_delete(bucket);
  78     }
  79
  80     if (readlen == 0 && status == APR_SUCCESS && block == APR_NONBLOCK_READ) {
  81         return APR_EAGAIN;
  82     }
  83     return status;
  84 }
  85
  86 h2_filter_cin *h2_filter_cin_create(apr_pool_t *p, h2_filter_cin_cb *cb, void *ctx)
  87 {
  88     h2_filter_cin *cin;
  89
  90     cin = apr_pcalloc(p, sizeof(*cin));
  91     cin->pool      = p;
  92     cin->cb        = cb;
  93     cin->cb_ctx    = ctx;
  94     cin->start_read = UNSET;
  95     return cin;
  96 }
  97
  98 void h2_filter_cin_timeout_set(h2_filter_cin *cin, apr_interval_time_t timeout)
  99 {
 100     cin->timeout = timeout;
 101 }
 102
 103 apr_status_t h2_filter_core_input(ap_filter_t* f,
 104                                   apr_bucket_brigade* brigade,
 105                                   ap_input_mode_t mode,
 106                                   apr_read_type_e block,
 107                                   apr_off_t readbytes)
 108 {
 109     h2_filter_cin *cin = f->ctx;
 110     apr_status_t status = APR_SUCCESS;
 111     apr_interval_time_t saved_timeout = UNSET;
 112
 113     ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,
 114                   "core_input(%ld): read, %s, mode=%d, readbytes=%ld",
 115                   (long)f->c->id, (block == APR_BLOCK_READ)? "BLOCK_READ" : "NONBLOCK_READ",
 116                   mode, (long)readbytes);
 117
 118     if (mode == AP_MODE_INIT || mode == AP_MODE_SPECULATIVE) {
 119         return ap_get_brigade(f->next, brigade, mode, block, readbytes);
 120     }
 121
 122     if (mode != AP_MODE_READBYTES) {
 123         return (block == APR_BLOCK_READ)? APR_SUCCESS : APR_EAGAIN;
 124     }
 125
 126     if (!cin->bb) {
 127         cin->bb = apr_brigade_create(cin->pool, f->c->bucket_alloc);
 128     }
 129
 130     if (!cin->socket) {
 131         cin->socket = ap_get_conn_socket(f->c);
 132     }
 133
 134     cin->start_read = apr_time_now();
 135     if (APR_BRIGADE_EMPTY(cin->bb)) {
 136         /* We only do a blocking read when we have no streams to process. So,
 137          * in httpd scoreboard lingo, we are in a KEEPALIVE connection state.
 138          * When reading non-blocking, we do have streams to process and update
 139          * child with NULL request. That way, any current request information
 140          * in the scoreboard is preserved.
 141          */
 142         if (block == APR_BLOCK_READ) {
 143             if (cin->timeout > 0) {
 144                 apr_socket_timeout_get(cin->socket, &saved_timeout);
 145                 apr_socket_timeout_set(cin->socket, cin->timeout);
 146             }
 147         }
 148         status = ap_get_brigade(f->next, cin->bb, AP_MODE_READBYTES,
 149                                 block, readbytes);
 150         if (saved_timeout != UNSET) {
 151             apr_socket_timeout_set(cin->socket, saved_timeout);
 152         }
 153     }
 154
 155     switch (status) {
 156         case APR_SUCCESS:
 157             status = consume_brigade(cin, cin->bb, block);
 158             break;
 159         case APR_EOF:
 160         case APR_EAGAIN:
 161         case APR_TIMEUP:
 162             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
 163                           "core_input(%ld): read", (long)f->c->id);
 164             break;
 165         default:
 166             ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, f->c, APLOGNO(03046)
 167                           "h2_conn_io: error reading");
 168             break;
 169     }
 170     return status;
 171 }
 172
 173 /*******************************************************************************
 174  * http2 connection status handler + stream out source
 175  ******************************************************************************/
 176
 177 static const char *H2_SOS_H2_STATUS = "http2-status";
 178
 179 int h2_filter_h2_status_handler(request_rec *r)
 180 {
 181     h2_ctx *ctx = h2_ctx_rget(r);
 182     h2_task *task;
 183
 184     if (strcmp(r->handler, "http2-status")) {
 185         return DECLINED;
 186     }
 187     if (r->method_number != M_GET) {
 188         return DECLINED;
 189     }
 190
 191     task = ctx? h2_ctx_get_task(ctx) : NULL;
 192     if (task) {
 193         /* We need to handle the actual output on the main thread, as
 194          * we need to access h2_session information. */
 195         apr_table_setn(r->notes, H2_RESP_SOS_NOTE, H2_SOS_H2_STATUS);
 196         apr_table_setn(r->headers_out, "Content-Type", "application/json");
 197         r->status = 200;
 198         return DONE;
 199     }
 200     return DECLINED;
 201 }
 202
 203 static apr_status_t bbout(apr_bucket_brigade *bb, const char *fmt, ...)
 204 {
 205     va_list args;
 206     apr_status_t rv;
 207
 208     va_start(args, fmt);
 209     rv = apr_brigade_vprintf(bb, NULL, NULL, fmt, args);
 210     va_end(args);
 211
 212     return rv;
 213 }
 214
 215 static void add_settings(apr_bucket_brigade *bb, h2_session *s, int last)
 216 {
 217     h2_mplx *m = s->mplx;
 218
 219     bbout(bb, "  \"settings\": {\n");
 220     bbout(bb, "    \"SETTINGS_MAX_CONCURRENT_STREAMS\": %d,\n", m->max_streams);
 221     bbout(bb, "    \"SETTINGS_MAX_FRAME_SIZE\": %d,\n", 16*1024);
 222     bbout(bb, "    \"SETTINGS_INITIAL_WINDOW_SIZE\": %d,\n",
 223           h2_config_geti(s->config, H2_CONF_WIN_SIZE));
 224     bbout(bb, "    \"SETTINGS_ENABLE_PUSH\": %d\n", h2_session_push_enabled(s));
 225     bbout(bb, "  }%s\n", last? "" : ",");
 226 }
 227
 228 static void add_peer_settings(apr_bucket_brigade *bb, h2_session *s, int last)
 229 {
 230     bbout(bb, "  \"peerSettings\": {\n");
 231     bbout(bb, "    \"SETTINGS_MAX_CONCURRENT_STREAMS\": %d,\n",
 232         nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS));
 233     bbout(bb, "    \"SETTINGS_MAX_FRAME_SIZE\": %d,\n",
 234         nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_MAX_FRAME_SIZE));
 235     bbout(bb, "    \"SETTINGS_INITIAL_WINDOW_SIZE\": %d,\n",
 236         nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE));
 237     bbout(bb, "    \"SETTINGS_ENABLE_PUSH\": %d,\n",
 238         nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_ENABLE_PUSH));
 239     bbout(bb, "    \"SETTINGS_HEADER_TABLE_SIZE\": %d,\n",
 240         nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_HEADER_TABLE_SIZE));
 241     bbout(bb, "    \"SETTINGS_MAX_HEADER_LIST_SIZE\": %d\n",
 242         nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE));
 243     bbout(bb, "  }%s\n", last? "" : ",");
 244 }
 245
 246 typedef struct {
 247     apr_bucket_brigade *bb;
 248     h2_session *s;
 249     int idx;
 250 } stream_ctx_t;
 251
 252 static int add_stream(h2_stream *stream, void *ctx)
 253 {
 254     stream_ctx_t *x = ctx;
 255     int32_t flowIn, flowOut;
 256
 257     flowIn = nghttp2_session_get_stream_effective_local_window_size(x->s->ngh2, stream->id);
 258     flowOut = nghttp2_session_get_stream_remote_window_size(x->s->ngh2, stream->id);
 259     bbout(x->bb, "%s\n    \"%d\": {\n", (x->idx? "," : ""), stream->id);
 260     bbout(x->bb, "    \"state\": \"%s\",\n", h2_stream_state_str(stream));
 261     bbout(x->bb, "    \"created\": %f,\n", ((double)stream->created)/APR_USEC_PER_SEC);
 262     bbout(x->bb, "    \"flowIn\": %d,\n", flowIn);
 263     bbout(x->bb, "    \"flowOut\": %d,\n", flowOut);
 264     bbout(x->bb, "    \"dataIn\": %"APR_UINT64_T_FMT",\n", stream->in_data_octets);
 265     bbout(x->bb, "    \"dataOut\": %"APR_UINT64_T_FMT"\n", stream->out_data_octets);
 266     bbout(x->bb, "    }");
 267
 268     ++x->idx;
 269     return 1;
 270 }
 271
 272 static void add_streams(apr_bucket_brigade *bb, h2_session *s, int last)
 273 {
 274     stream_ctx_t x;
 275
 276     x.bb = bb;
 277     x.s = s;
 278     x.idx = 0;
 279     bbout(bb, "  \"streams\": {");
 280     h2_mplx_stream_do(s->mplx, add_stream, &x);
 281     bbout(bb, "\n  }%s\n", last? "" : ",");
 282 }
 283
 284 static void add_push(apr_bucket_brigade *bb, h2_session *s,
 285                      h2_stream *stream, int last)
 286 {
 287     h2_push_diary *diary;
 288     apr_status_t status;
 289
 290     bbout(bb, "    \"push\": {\n");
 291     diary = s->push_diary;
 292     if (diary) {
 293         const char *data;
 294         const char *base64_digest;
 295         apr_size_t len;
 296
 297         status = h2_push_diary_digest_get(diary, bb->p, 256,
 298                                           stream->request->authority,
 299                                           &data, &len);
 300         if (status == APR_SUCCESS) {
 301             base64_digest = h2_util_base64url_encode(data, len, bb->p);
 302             bbout(bb, "      \"cacheDigest\": \"%s\",\n", base64_digest);
 303         }
 304     }
 305     bbout(bb, "      \"promises\": %d,\n", s->pushes_promised);
 306     bbout(bb, "      \"submits\": %d,\n", s->pushes_submitted);
 307     bbout(bb, "      \"resets\": %d\n", s->pushes_reset);
 308     bbout(bb, "    }%s\n", last? "" : ",");
 309 }
 310
 311 static void add_in(apr_bucket_brigade *bb, h2_session *s, int last)
 312 {
 313     bbout(bb, "    \"in\": {\n");
 314     bbout(bb, "      \"requests\": %d,\n", s->remote.emitted_count);
 315     bbout(bb, "      \"resets\": %d, \n", s->streams_reset);
 316     bbout(bb, "      \"frames\": %ld,\n", (long)s->frames_received);
 317     bbout(bb, "      \"octets\": %"APR_UINT64_T_FMT"\n", s->io.bytes_read);
 318     bbout(bb, "    }%s\n", last? "" : ",");
 319 }
 320
 321 static void add_out(apr_bucket_brigade *bb, h2_session *s, int last)
 322 {
 323     bbout(bb, "    \"out\": {\n");
 324     bbout(bb, "      \"responses\": %d,\n", s->responses_submitted);
 325     bbout(bb, "      \"frames\": %ld,\n", (long)s->frames_sent);
 326     bbout(bb, "      \"octets\": %"APR_UINT64_T_FMT"\n", s->io.bytes_written);
 327     bbout(bb, "    }%s\n", last? "" : ",");
 328 }
 329
 330 static void add_stats(apr_bucket_brigade *bb, h2_session *s,
 331                      h2_stream *stream, int last)
 332 {
 333     bbout(bb, "  \"stats\": {\n");
 334     add_in(bb, s, 0);
 335     add_out(bb, s, 0);
 336     add_push(bb, s, stream, 1);
 337     bbout(bb, "  }%s\n", last? "" : ",");
 338 }
 339
 340 static apr_status_t h2_status_stream_filter(h2_stream *stream)
 341 {
 342     h2_session *s = stream->session;
 343     conn_rec *c = s->c;
 344     apr_bucket_brigade *bb;
 345     int32_t connFlowIn, connFlowOut;
 346
 347     if (!stream->response) {
 348         return APR_EINVAL;
 349     }
 350
 351     if (!stream->buffer) {
 352         stream->buffer = apr_brigade_create(stream->pool, c->bucket_alloc);
 353     }
 354     bb = stream->buffer;
 355
 356     apr_table_unset(stream->response->headers, "Content-Length");
 357     stream->response->content_length = -1;
 358
 359     connFlowIn = nghttp2_session_get_effective_local_window_size(s->ngh2);
 360     connFlowOut = nghttp2_session_get_remote_window_size(s->ngh2);
 361     apr_table_setn(stream->response->headers, "conn-flow-in",
 362                    apr_itoa(stream->pool, connFlowIn));
 363     apr_table_setn(stream->response->headers, "conn-flow-out",
 364                    apr_itoa(stream->pool, connFlowOut));
 365
 366     bbout(bb, "{\n");
 367     bbout(bb, "  \"version\": \"draft-01\",\n");
 368     add_settings(bb, s, 0);
 369     add_peer_settings(bb, s, 0);
 370     bbout(bb, "  \"connFlowIn\": %d,\n", connFlowIn);
 371     bbout(bb, "  \"connFlowOut\": %d,\n", connFlowOut);
 372     bbout(bb, "  \"sentGoAway\": %d,\n", s->local.shutdown);
 373
 374     add_streams(bb, s, 0);
 375
 376     add_stats(bb, s, stream, 1);
 377     bbout(bb, "}\n");
 378
 379     return APR_SUCCESS;
 380 }
 381
 382 apr_status_t h2_stream_filter(h2_stream *stream)
 383 {
 384     const char *fname = stream->response? stream->response->sos_filter : NULL;
 385     if (fname && !strcmp(H2_SOS_H2_STATUS, fname)) {
 386         return h2_status_stream_filter(stream);
 387     }
 388     return APR_SUCCESS;
 389 }
 390