/* ==================================================================== * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. * ==================================================================== */ #include #include #include #include #include #include "serf.h" #include "serf_bucket_util.h" #include "serf_private.h" #include "http2_buckets.h" #include "http2_protocol.h" static apr_status_t http2_outgoing_read(serf_connection_t *conn); static apr_status_t http2_outgoing_write(serf_connection_t *conn); static apr_status_t http2_outgoing_hangup(serf_connection_t *conn); static void http2_outgoing_teardown(serf_connection_t *conn); static void http2_outgoing_pre_teardown(serf_connection_t *conn); static apr_status_t http2_incoming_read(serf_incoming_t *client); static apr_status_t http2_incoming_write(serf_incoming_t *client); static apr_status_t http2_incoming_hangup(serf_incoming_t *client); static void http2_incoming_teardown(serf_incoming_t *conn); static void http2_incoming_pre_teardown(serf_incoming_t *conn); static apr_status_t http2_process(serf_http2_protocol_t *h2); static void http2_cancel_request(serf_request_t *rq, apr_status_t reason); static void http2_prioritize_request(serf_request_t *rq, bool exclusive); static apr_status_t http2_write_data(serf_http2_protocol_t *h2); static serf_bucket_t * serf_bucket_create_numberv(serf_bucket_alloc_t *allocator, const char *format, ...) { va_list argp; const char *c; char *buffer; apr_size_t sz = 0; unsigned char *r; va_start(argp, format); for (c = format; *c; c++) { SERF_H2_assert(*c >= '1' && *c <= '4'); if (*c >= '1' && *c <= '4') sz += (*c - '0'); } buffer = serf_bucket_mem_alloc(allocator, sz); r = (void*)buffer; for (c = format; *c; c++) { apr_uint32_t tmp; switch (*c) { case '1': *r++ = va_arg(argp, int) & 0xFF; break; case '2': tmp = va_arg(argp, int); *r++ = (tmp >> 8) & 0xFF; *r++ = tmp & 0xFF; break; case '3': tmp = va_arg(argp, apr_uint32_t); *r++ = (tmp >> 16) & 0xFF; *r++ = (tmp >> 8) & 0xFF; *r++ = tmp & 0xFF; break; case '4': tmp = va_arg(argp, apr_uint32_t); *r++ = (tmp >> 24) & 0xFF; *r++ = (tmp >> 16) & 0xFF; *r++ = (tmp >> 8) & 0xFF; *r++ = tmp & 0xFF; break; } } va_end(argp); return serf_bucket_simple_own_create(buffer, sz, allocator); } struct serf_http2_protocol_t { apr_pool_t *pool; serf_connection_t *conn; /* Either CONN or CLIENT is set */ serf_incoming_t *client; serf_io_baton_t *io; /* Low level connection */ serf_pump_t *pump; serf_bucket_alloc_t *allocator; serf_http2_processor_t processor; void *processor_baton; serf_bucket_t *read_frame; /* Frame currently being read */ bool in_frame; apr_size_t prefix_left; serf_hpack_table_t *hpack_tbl; serf_config_t *config; /* Local -> Remote. Settings provided by other side */ apr_uint32_t lr_default_window; apr_uint32_t lr_window; apr_uint32_t lr_max_framesize; apr_uint32_t lr_max_headersize; apr_uint32_t lr_max_concurrent; apr_uint32_t lr_hpack_table_size; apr_int32_t lr_next_streamid; bool lr_push_enabled; /* Remote -> Local. Settings set by us. Acknowledged by other side */ apr_uint32_t rl_default_window; apr_uint32_t rl_window; apr_uint32_t rl_max_framesize; apr_uint32_t rl_max_headersize; apr_uint32_t rl_max_concurrent; apr_uint32_t rl_hpack_table_size; apr_int32_t rl_next_streamid; bool rl_push_enabled; apr_uint32_t rl_window_upd_below; apr_uint32_t rl_window_upd_to; serf_http2_stream_t *first; serf_http2_stream_t *last; int setting_acks; bool enforce_flow_control; serf_bucket_t *continuation_bucket; apr_int32_t continuation_streamid; serf_http2_stream_t *first_writable, *last_writable, *cur_writable; }; /* Forward definition */ static apr_status_t http2_bucket_processor(void *baton, serf_http2_protocol_t *h2, serf_bucket_t *frame_bucket); static apr_status_t http2_protocol_cleanup(void *state) { serf_http2_protocol_t *h2 = state; serf_connection_t *conn = h2->conn; serf_incoming_t *client = h2->client; serf_http2_stream_t *stream, *next; /* First clean out all streams */ for (stream = h2->first; stream; stream = next) { next = stream->next; serf_http2__stream_cleanup(stream); } h2->first = h2->last = NULL; if (h2->processor != NULL) { h2->read_frame = NULL; if (h2->processor == http2_bucket_processor) { serf_bucket_t *payload = h2->processor_baton; if (payload) serf_bucket_destroy(payload); h2->processor = NULL; h2->processor_baton = NULL; } /* Else: The processor (probably a stream) needs to handle this. It usually does that by adding frames to an aggregate to allow reading multiple frames as a stream. */ } else if (h2->read_frame) { serf_bucket_destroy(h2->read_frame); h2->read_frame = NULL; } h2->in_frame = FALSE; if (conn) conn->protocol_baton = NULL; if (client) client->protocol_baton = NULL; return APR_SUCCESS; } static void http2_send_window_update(serf_http2_protocol_t *h2, serf_http2_stream_t *stream) { apr_uint32_t increase; apr_uint32_t *window; apr_int32_t *stream_id; serf_bucket_t *bkt; apr_status_t status; struct window_update_t { unsigned char v3, v2, v1, v0; } window_update; if (!stream) { if (h2->rl_window >= h2->rl_window_upd_to) return; increase = h2->rl_window_upd_to - h2->rl_window; window = &h2->rl_window; stream_id = NULL; } else { if (stream->rl_window >= stream->rl_window_upd_to) return; increase = stream->rl_window_upd_to - stream->rl_window; window = &stream->rl_window; stream_id = &stream->streamid; } window_update.v3 = (increase >> 24) & 0xFF; window_update.v2 = (increase >> 16) & 0xFF; window_update.v1 = (increase >> 8) & 0xFF; window_update.v0 = increase & 0xFF; bkt = serf_bucket_simple_copy_create((void *)&window_update, sizeof(window_update), h2->allocator); bkt = serf__bucket_http2_frame_create(bkt, HTTP2_FRAME_TYPE_WINDOW_UPDATE, 0, stream_id, NULL, NULL/* stream */, h2->lr_max_framesize, h2->allocator); status = serf_http2__enqueue_frame(h2, bkt, FALSE); if (!status) { /* Update our administration */ (*window) += increase; } /* Ignore connection broken statee. Move along */ } void serf__http2_protocol_init(serf_connection_t *conn) { serf_http2_protocol_t *h2; apr_pool_t *protocol_pool; serf_bucket_t *tmp; const bool WE_ARE_CLIENT = true; apr_pool_create(&protocol_pool, conn->pool); h2 = apr_pcalloc(protocol_pool, sizeof(*h2)); h2->pool = protocol_pool; h2->conn = conn; h2->io = &conn->io; h2->pump = &conn->pump; h2->allocator = conn->allocator; h2->config = conn->config; /* Defaults until negotiated */ h2->rl_default_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->rl_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->rl_next_streamid = WE_ARE_CLIENT ? 2 : 1; h2->rl_max_framesize = HTTP2_DEFAULT_MAX_FRAMESIZE; h2->rl_max_headersize = APR_UINT32_MAX; h2->rl_max_concurrent = HTTP2_DEFAULT_MAX_CONCURRENT; h2->rl_hpack_table_size = HTTP2_DEFAULT_HPACK_TABLE_SIZE; h2->rl_push_enabled = TRUE; h2->lr_default_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->lr_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->lr_next_streamid = WE_ARE_CLIENT ? 1 : 2; h2->lr_max_framesize = HTTP2_DEFAULT_MAX_FRAMESIZE; h2->lr_max_headersize = APR_UINT32_MAX; h2->lr_max_concurrent = HTTP2_DEFAULT_MAX_CONCURRENT; h2->lr_hpack_table_size = HTTP2_DEFAULT_HPACK_TABLE_SIZE; h2->lr_push_enabled = TRUE; h2->rl_window_upd_below = 16 * 1024 * 1024; /* 16 MB*/ h2->rl_window_upd_to = 128 * 1024 * 1024; /* 128 MB */ h2->setting_acks = 0; h2->enforce_flow_control = TRUE; h2->continuation_bucket = NULL; h2->continuation_streamid = 0; h2->first = h2->last = NULL; h2->hpack_tbl = serf__hpack_table_create(TRUE, HTTP2_DEFAULT_HPACK_TABLE_SIZE, protocol_pool); apr_pool_cleanup_register(protocol_pool, h2, http2_protocol_cleanup, apr_pool_cleanup_null); conn->perform_read = http2_outgoing_read; conn->perform_write = http2_outgoing_write; conn->perform_hangup = http2_outgoing_hangup; conn->perform_teardown = http2_outgoing_teardown; conn->perform_pre_teardown = http2_outgoing_pre_teardown; conn->perform_cancel_request = http2_cancel_request; conn->perform_prioritize_request = http2_prioritize_request; conn->protocol_baton = h2; /* Disable HTTP/1.1 guessing that affects writability */ conn->probable_keepalive_limit = 0; conn->max_outstanding_requests = 0; /* Send the HTTP/2 Connection Preface */ tmp = SERF_BUCKET_SIMPLE_STRING(HTTP2_CONNECTION_PREFIX, h2->allocator); serf_pump__add_output(h2->pump, tmp, false); /* And now a settings frame */ tmp = serf__bucket_http2_frame_create(NULL, HTTP2_FRAME_TYPE_SETTINGS, 0, NULL, NULL, NULL, /* stream: 0 */ h2->lr_max_framesize, h2->allocator); serf_http2__enqueue_frame(h2, tmp, FALSE); /* And an initial window update */ http2_send_window_update(h2, NULL); } void serf__http2_protocol_init_server(serf_incoming_t *client) { serf_http2_protocol_t *h2; apr_pool_t *protocol_pool; serf_bucket_t *tmp; const int WE_ARE_CLIENT = false; apr_pool_create(&protocol_pool, client->pool); h2 = apr_pcalloc(protocol_pool, sizeof(*h2)); h2->pool = protocol_pool; h2->client = client; h2->io = &client->io; h2->pump = &client->pump; h2->allocator = client->allocator; h2->config = client->config; h2->prefix_left = sizeof(HTTP2_CONNECTION_PREFIX) - 1; /* Defaults until negotiated */ h2->rl_default_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->rl_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->rl_next_streamid = WE_ARE_CLIENT ? 2 : 1; h2->rl_max_framesize = HTTP2_DEFAULT_MAX_FRAMESIZE; h2->rl_max_headersize = APR_UINT32_MAX; h2->rl_max_concurrent = HTTP2_DEFAULT_MAX_CONCURRENT; h2->rl_hpack_table_size = HTTP2_DEFAULT_HPACK_TABLE_SIZE; h2->rl_push_enabled = TRUE; h2->lr_default_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->lr_window = HTTP2_DEFAULT_WINDOW_SIZE; h2->lr_next_streamid = WE_ARE_CLIENT ? 1 : 2; h2->lr_max_framesize = HTTP2_DEFAULT_MAX_FRAMESIZE; h2->lr_max_headersize = APR_UINT32_MAX; h2->lr_max_concurrent = HTTP2_DEFAULT_MAX_CONCURRENT; h2->lr_hpack_table_size = HTTP2_DEFAULT_HPACK_TABLE_SIZE; h2->lr_push_enabled = TRUE; h2->rl_window_upd_below = 16 * 1024 * 1024; /* 16 MB*/ h2->rl_window_upd_to = 128 * 1024 * 1024; /* 128 MB */ h2->setting_acks = 0; h2->enforce_flow_control = TRUE; h2->continuation_bucket = NULL; h2->continuation_streamid = 0; h2->first = h2->last = NULL; h2->hpack_tbl = serf__hpack_table_create(TRUE, HTTP2_DEFAULT_HPACK_TABLE_SIZE, protocol_pool); apr_pool_cleanup_register(protocol_pool, h2, http2_protocol_cleanup, apr_pool_cleanup_null); client->perform_read = http2_incoming_read; client->perform_write = http2_incoming_write; client->perform_hangup = http2_incoming_hangup; client->perform_teardown = http2_incoming_teardown; client->perform_pre_teardown = http2_incoming_pre_teardown; client->protocol_baton = h2; /* Send a settings frame */ tmp = serf__bucket_http2_frame_create(NULL, HTTP2_FRAME_TYPE_SETTINGS, 0, NULL, NULL, NULL, /* stream: 0 */ h2->lr_max_framesize, h2->allocator); serf_http2__enqueue_frame(h2, tmp, FALSE); /* And an initial window update*/ http2_send_window_update(h2, NULL); } /* Creates a HTTP/2 request from a serf request */ static apr_status_t enqueue_http2_request(serf_http2_protocol_t *h2) { serf_http2_stream_t *stream; stream = serf_http2__stream_create(h2, -1, h2->lr_default_window, h2->rl_default_window, h2->allocator); if (h2->first) { stream->next = h2->first; h2->first->prev = stream; h2->first = stream; } else h2->last = h2->first = stream; return serf_http2__stream_setup_next_request(stream, h2->conn, h2->lr_max_framesize, h2->hpack_tbl); } apr_status_t serf_http2__enqueue_frame(serf_http2_protocol_t *h2, serf_bucket_t *frame, bool flush) { return serf_pump__add_output(h2->pump, frame, flush); } /* Implements serf_bucket_prefix_handler_t. Handles PRIORITY frames and the priority prefix of HEADERS frames */ static apr_status_t http2_handle_priority(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_stream_t *stream = baton; if (len != HTTP2_PRIORITY_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; if (stream == NULL) return APR_SUCCESS; /* Nothing to record this on */ /* ### TODO: Store priority information on stream */ SERF_H2_assert(stream->h2 != NULL); return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles the promise prefix of PUSH_PROMISE frames */ static apr_status_t http2_handle_promise(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_stream_t *parent_stream = baton; serf_http2_protocol_t *h2 = parent_stream->h2; serf_http2_stream_t *promised_stream; apr_int32_t streamid; const struct promise_t { unsigned char s3, s2, s1, s0; } *promise; if (len != HTTP2_PROMISE_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; SERF_H2_assert(h2 != NULL); promise = (const void *)data; /* Highest bit is reserved */ streamid = ((promise->s3 & 0x7F) << 24) | (promise->s2 << 16) | (promise->s1 << 8) | promise->s0; if (streamid == 0 || (streamid < h2->rl_next_streamid) || (streamid & 0x01) != (h2->rl_next_streamid & 0x01)) { /* The promised stream identifier MUST bet a valid choice for the next stream sent by the sender */ /* A receiver MUST treat the receipt of a PUSH_PROMISE that promises an illegal stream identifier (Section 5.1.1) as a connection error (Section 5.4.1) of type PROTOCOL_ERROR. Note that an illegal stream identifier is an identifier for a stream that is not currently in the "idle" state.*/ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } else if (parent_stream->status != H2S_OPEN && parent_stream->status != H2S_HALFCLOSED_LOCAL) { /* PUSH_PROMISE frames MUST only be sent on a peer-initiated stream that is in either the "open" or "half-closed (remote)" state. The stream identifier of a PUSH_PROMISE frame indicates the stream it is associated with. If the stream identifier field specifies the value 0x0, a recipient MUST respond with a connection error (Section 5.4.1) of type PROTOCOL_ERROR.*/ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } promised_stream = serf_http2__stream_get(h2, streamid, TRUE, FALSE); if (!promised_stream || promised_stream->status != H2S_IDLE) return SERF_ERROR_HTTP2_PROTOCOL_ERROR; promised_stream->status = H2S_RESERVED_REMOTE; /* Store data to allow stream to handle the promise */ parent_stream->new_reserved_stream = promised_stream; return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles the promise prefix of FRAME_RSET frames */ static apr_status_t http2_handle_frame_reset(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_stream_t *stream = baton; if (len != HTTP2_RST_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; SERF_H2_assert(stream->h2 != NULL); /* ### TODO: Handle error code, etc. */ stream->status = H2S_CLOSED; return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles WINDOW_UPDATE frames when they apply to a stream */ static apr_status_t http2_handle_stream_window_update(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_stream_t *stream = baton; apr_uint32_t value; const struct window_update_t { unsigned char v3, v2, v1, v0; } *window_update; if (len != HTTP2_WINDOW_UPDATE_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; window_update = (const void *)data; value = (window_update->v3 << 24) | (window_update->v2 << 16) | (window_update->v1 << 8) | window_update->v0; value &= HTTP2_WINDOW_MAX_ALLOWED; /* The highest bit is reserved */ if (value == 0) { /* A receiver MUST treat the receipt of a WINDOW_UPDATE frame with an flow - control window increment of 0 as a stream error(Section 5.4.2) of type PROTOCOL_ERROR; errors on the connection flow - control window MUST be treated as a connection error(Section 5.4.1). */ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } stream->lr_window += value; if (stream->lr_window > HTTP2_WINDOW_MAX_ALLOWED ) { /* A sender MUST NOT allow a flow-control window to exceed 2^31-1 octets. If a sender receives a WINDOW_UPDATE that causes a flow- control window to exceed this maximum, it MUST terminate either the stream or the connection, as appropriate. For streams, the sender sends a RST_STREAM with an error code of FLOW_CONTROL_ERROR; for the connection, a GOAWAY frame with an error code of FLOW_CONTROL_ERROR is sent.*/ return SERF_ERROR_HTTP2_FLOW_CONTROL_ERROR; } serf__log(LOGLVL_INFO, SERF_LOGHTTP2, stream->h2->config, "Increasing window on stream %d with 0x%x to 0x%x\n", stream->streamid, value, stream->lr_window); return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles WINDOW_UPDATE frames when they apply to the connection */ static apr_status_t http2_handle_connection_window_update(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_protocol_t *h2 = baton; apr_uint32_t value; bool was0; const struct window_update_t { unsigned char v3, v2, v1, v0; } *window_update; if (len != HTTP2_WINDOW_UPDATE_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; SERF_H2_assert(h2 != NULL); window_update = (const void *)data; value = (window_update->v3 << 24) | (window_update->v2 << 16) | (window_update->v1 << 8) | window_update->v0; value &= HTTP2_WINDOW_MAX_ALLOWED; /* The highest bit is reserved */ if (value == 0) { /* A receiver MUST treat the receipt of a WINDOW_UPDATE frame with an flow - control window increment of 0 as a stream error(Section 5.4.2) of type PROTOCOL_ERROR; errors on the connection flow - control window MUST be treated as a connection error(Section 5.4.1). */ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } was0 = (h2->lr_window == 0); h2->lr_window += value; if (h2->lr_window > HTTP2_WINDOW_MAX_ALLOWED) { /* A sender MUST NOT allow a flow-control window to exceed 2^31-1 octets. If a sender receives a WINDOW_UPDATE that causes a flow- control window to exceed this maximum, it MUST terminate either the stream or the connection, as appropriate. For streams, the sender sends a RST_STREAM with an error code of FLOW_CONTROL_ERROR; for the connection, a GOAWAY frame with an error code of FLOW_CONTROL_ERROR is sent.*/ return SERF_ERROR_HTTP2_FLOW_CONTROL_ERROR; } serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Increasing window on connection with 0x%x to 0x%x\n", value, h2->lr_window); if (was0) return http2_write_data(h2); else return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles PING frames for pings initiated remotely */ static apr_status_t http2_handle_ping(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_protocol_t *h2 = baton; serf_bucket_t *body; apr_status_t status; if (len != HTTP2_PING_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; SERF_H2_assert(h2 != NULL); /* Reply with a PONG (=PING + ACK) with the same data*/ body = serf_bucket_simple_copy_create(data, len, h2->allocator); status = serf_http2__enqueue_frame( h2, serf__bucket_http2_frame_create(body, HTTP2_FRAME_TYPE_PING, HTTP2_FLAG_ACK, NULL, NULL, NULL, h2->lr_max_framesize, h2->allocator), TRUE /* pump */); if (SERF_BUCKET_READ_ERROR(status)) return status; return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles PING frames for pings initiated locally */ static apr_status_t http2_handle_ping_ack(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { /* serf_http2_protocol_t *h2 = baton; */ if (len != HTTP2_PING_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; SERF_H2_assert(baton != NULL); /* Did we send a ping? */ return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles SETTINGS frames */ static apr_status_t http2_handle_settings(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_protocol_t *h2 = baton; apr_size_t i; const struct setting_t { unsigned char s1, s0; unsigned char v3, v2, v1, v0; } *setting; if ((len % HTTP2_SETTING_SIZE) != 0) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; /* ### TODO: Handle settings */ setting = (const void *)data; for (i = 0, setting = (const void *)data; i < len; i += sizeof(*setting), setting++) { apr_uint16_t id = (setting->s1 << 8) | setting->s0; apr_uint32_t value = (setting->v3 << 24) | (setting->v2 << 16) | (setting->v1 << 8) | setting->v0; switch (id) { case HTTP2_SETTING_HEADER_TABLE_SIZE: serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Setting HPACK Table size to %u\n", value); serf__hpack_table_set_max_table_size(h2->hpack_tbl, h2->rl_hpack_table_size, value); break; case HTTP2_SETTING_ENABLE_PUSH: serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Setting Push enabled: %u\n", value); h2->lr_push_enabled = (value != 0); break; case HTTP2_SETTING_MAX_CONCURRENT_STREAMS: serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Setting Max Concurrent %u\n", value); h2->lr_max_concurrent = value; break; case HTTP2_SETTING_INITIAL_WINDOW_SIZE: /* Sanitize? */ serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Setting Initial Window Size %u\n", value); /* This only affects the default window size for new streams (the connection window size is left unchanged): Both endpoints can adjust the initial window size for new streams by including a value for SETTINGS_INITIAL_WINDOW_SIZE in the SETTINGS frame that forms part of the connection preface. The connection flow-control window can only be changed using WINDOW_UPDATE frames. */ h2->lr_default_window = value; break; case HTTP2_SETTING_MAX_FRAME_SIZE: /* Sanitize? */ serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Setting Max framesize %u\n", value); h2->lr_max_framesize = value; break; case HTTP2_SETTING_MAX_HEADER_LIST_SIZE: serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Setting Max header list size %u\n", value); h2->lr_max_headersize = value; break; default: /* An endpoint that receives a SETTINGS frame with any unknown or unsupported identifier MUST ignore that setting. */ serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Ignoring unknown setting %d, value %u\n", id, value); break; } } /* Always ack settings */ serf_http2__enqueue_frame( h2, serf__bucket_http2_frame_create( NULL, HTTP2_FRAME_TYPE_SETTINGS, HTTP2_FLAG_ACK, NULL, NULL, NULL, h2->lr_max_framesize, h2->allocator), TRUE); return APR_SUCCESS; } /* Implements serf_bucket_prefix_handler_t. Handles GOAWAY frames */ static apr_status_t http2_handle_goaway(void *baton, serf_bucket_t *bucket, const char *data, apr_size_t len) { serf_http2_protocol_t *h2 = baton; apr_int32_t last_streamid; apr_uint32_t error_code; apr_uint32_t loglevel; const struct goaway_t { unsigned char s3, s2, s1, s0; unsigned char e3, e2, e1, e0; } *goaway; if (len < HTTP2_GOWAWAY_DATA_SIZE) return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; SERF_H2_assert(h2 != NULL); goaway = (const void *)data; last_streamid = ((goaway->s3 & 0x7F) << 24) | (goaway->s2 << 16) | (goaway->s1 << 8) | goaway->s0; error_code = (goaway->e3 << 24) | (goaway->e2 << 16) | (goaway->e1 << 8) | goaway->e0; switch (error_code + SERF_ERROR_HTTP2_NO_ERROR) { case SERF_ERROR_HTTP2_PROTOCOL_ERROR: case SERF_ERROR_HTTP2_FLOW_CONTROL_ERROR: case SERF_ERROR_HTTP2_SETTINGS_TIMEOUT: case SERF_ERROR_HTTP2_FRAME_SIZE_ERROR: case SERF_ERROR_HTTP2_COMPRESSION_ERROR: case SERF_ERROR_HTTP2_INADEQUATE_SECURITY: loglevel = LOGLVL_ERROR; break; case SERF_ERROR_HTTP2_HTTP_1_1_REQUIRED: case SERF_ERROR_HTTP2_ENHANCE_YOUR_CALM: loglevel = LOGLVL_WARNING; break; case SERF_ERROR_HTTP2_REFUSED_STREAM: case SERF_ERROR_HTTP2_CANCEL: case SERF_ERROR_HTTP2_CONNECT_ERROR: case SERF_ERROR_HTTP2_STREAM_CLOSED: /* These errors should have been sent as a stream error. This usually tells us that we have an http/2 implementation on the other side that doesn't implement full stream state handling. (See HTTP/2 RFC)*/ loglevel = LOGLVL_ERROR; break; case SERF_ERROR_HTTP2_NO_ERROR: loglevel = LOGLVL_INFO; break; case SERF_ERROR_HTTP2_INTERNAL_ERROR: default: loglevel = LOGLVL_WARNING; break; } if (len > HTTP2_GOWAWAY_DATA_SIZE) { char *goaway_text; /* The server produced additional information in the error frame Usually this is some literal text explaining what went wrong. Copy the text to make it 0 terminated and then log it. */ /* If this value appears truncated, that may be caused by the limit set in http2_process */ goaway_text = serf_bstrmemdup(h2->allocator, data + HTTP2_GOWAWAY_DATA_SIZE, len - HTTP2_GOWAWAY_DATA_SIZE); serf__log(loglevel, SERF_LOGHTTP2, h2->config, "Received GOAWAY, last-stream=0x%x, error=%u: %s\n", last_streamid, error_code, goaway_text); serf_bucket_mem_free(h2->allocator, goaway_text); } else { serf__log(loglevel, SERF_LOGHTTP2, h2->config, "Received GOAWAY, last-stream=0x%x, error=%u.\n", last_streamid, error_code); } /* ### TODO: If the error is not critical stop creating new frames on this connection, while still going forward with the existing frames. We may receive a new error later, signalling a more important problem */ return APR_SUCCESS; } /* Implements serf_bucket_aggregate_eof_t */ static apr_status_t http2_handle_continuation(void *baton, serf_bucket_t *aggregate_bucket) { serf_http2_protocol_t *h2 = baton; apr_status_t status; const char *data; apr_size_t len; if (h2->continuation_bucket != aggregate_bucket) return APR_EOF; /* This is all we have */ SERF_H2_assert(h2->read_frame == NULL); SERF_H2_assert(h2->continuation_bucket == aggregate_bucket); status = http2_process(h2); if (status) return status; if (h2->continuation_bucket == aggregate_bucket) { /* We expect more data in the future. Something was done in http2_process() or it didn't return APR_SUCCESS */ return APR_SUCCESS; } /* As h2->continuation_bucket is no longer attached we don't recurse on peeking. Just check if there is more */ return serf_bucket_peek(aggregate_bucket, &data, &len); } /* Implements the serf__bucket_http2_unframe_set_eof callback */ static apr_status_t http2_end_of_frame(void *baton, serf_bucket_t *frame) { serf_http2_protocol_t *h2 = baton; SERF_H2_assert(h2->read_frame == frame); h2->read_frame = NULL; h2->in_frame = FALSE; h2->processor = NULL; h2->processor_baton = NULL; return APR_SUCCESS; } /* Implements serf_http2_processor_t */ static apr_status_t http2_bucket_processor(void *baton, serf_http2_protocol_t *h2, serf_bucket_t *frame_bucket) { struct iovec vecs[SERF__STD_IOV_COUNT]; int vecs_used; serf_bucket_t *payload = baton; apr_status_t status; status = serf_bucket_read_iovec(payload, SERF_READ_ALL_AVAIL, COUNT_OF(vecs), vecs, &vecs_used); if (APR_STATUS_IS_EOF(status)) { SERF_H2_assert(!h2->in_frame && !h2->read_frame); serf_bucket_destroy(payload); } return status; } /* Processes incoming HTTP2 data */ static apr_status_t http2_process(serf_http2_protocol_t *h2) { while (TRUE) { apr_status_t status; serf_bucket_t *body; if (h2->processor) { status = h2->processor(h2->processor_baton, h2, h2->read_frame); if (SERF_BUCKET_READ_ERROR(status)) return status; else if (APR_STATUS_IS_EOF(status)) { /* ### frame ended */ SERF_H2_assert(h2->read_frame == NULL); h2->processor = NULL; h2->processor_baton = NULL; } else if (h2->in_frame) { if (status) return status; else continue; } } else { SERF_H2_assert(!h2->in_frame); } body = h2->read_frame; if (!body) { SERF_H2_assert(!h2->in_frame); body = serf__bucket_http2_unframe_create( h2->pump->stream, h2->rl_max_framesize, h2->allocator); serf__bucket_http2_unframe_set_eof(body, http2_end_of_frame, h2); serf_bucket_set_config(body, h2->config); h2->read_frame = body; } if (!h2->in_frame) { apr_int32_t sid; unsigned char frametype; unsigned char frameflags; apr_size_t remaining; serf_http2_processor_t process_handler = NULL; void *process_baton = NULL; serf_bucket_t *process_bucket = NULL; serf_http2_stream_t *stream; apr_uint32_t reset_reason; status = serf__bucket_http2_unframe_read_info(body, &sid, &frametype, &frameflags); if (APR_STATUS_IS_EOF(status)) { /* Entire frame is already read (just header) */ SERF_H2_assert(h2->read_frame == NULL); SERF_H2_assert(!h2->in_frame); } else if (status) { SERF_H2_assert(h2->read_frame != NULL); SERF_H2_assert(!h2->in_frame); return (status == SERF_ERROR_EMPTY_READ) ? APR_SUCCESS : status; } else { h2->in_frame = TRUE; SERF_H2_assert(h2->read_frame != NULL); } serf__log(LOGLVL_INFO, SERF_LOGHTTP2, h2->config, "Reading 0x%x frame, stream=0x%x, flags=0x%x\n", frametype, sid, frameflags); /* If status is EOF then the frame doesn't have/declare a body */ switch (frametype) { /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_DATA: case HTTP2_FRAME_TYPE_HEADERS: case HTTP2_FRAME_TYPE_PUSH_PROMISE: if (h2->continuation_bucket) { h2->continuation_bucket = NULL; h2->continuation_streamid = 0; return APR_EAGAIN; } stream = serf_http2__stream_get(h2, sid, TRUE, TRUE); if (sid == 0) { /* DATA, HEADERS and PUSH_PROMISE: These frames MUST be associated with a stream. If a XXX frame is received whose stream identifier field is 0x0, the recipient MUST respond with a connection error (Section 5.4.1) of type PROTOCOL_ERROR. */ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } reset_reason = 0; if (frametype == HTTP2_FRAME_TYPE_DATA) { /* Windowing is applied above padding! */ remaining = (apr_size_t)serf_bucket_get_remaining(body); if (h2->rl_window < remaining) { if (h2->enforce_flow_control) { reset_reason = SERF_ERROR_HTTP2_FLOW_CONTROL_ERROR; } h2->rl_window = 0; } else h2->rl_window -= remaining; if (h2->rl_window < h2->rl_window_upd_below) http2_send_window_update(h2, NULL); if (stream) { if (stream->rl_window < remaining) { if (h2->enforce_flow_control) { reset_reason = SERF_ERROR_HTTP2_FLOW_CONTROL_ERROR; } stream->rl_window = 0; } else stream->rl_window -= remaining; /* If the stream is not at the end, perhaps we should allow it to send more data */ if (!(frameflags & HTTP2_FLAG_END_STREAM) && stream->rl_window < stream->rl_window_upd_below) { http2_send_window_update(h2, stream); } } } /* DATA, HEADERS and PUSH_PROMISE can have padding */ if (frameflags & HTTP2_FLAG_PADDED) { body = serf__bucket_http2_unpad_create(body, h2->allocator); } /* An HEADERS frame can have an included priority 'frame' */ if (frametype == HTTP2_FRAME_TYPE_HEADERS && (frameflags & HTTP2_FLAG_PRIORITY)) { body = serf_bucket_prefix_create(body, HTTP2_PRIORITY_DATA_SIZE, http2_handle_priority, stream, h2->allocator); } else if (frametype == HTTP2_FRAME_TYPE_PUSH_PROMISE) { body = serf_bucket_prefix_create(body, HTTP2_PROMISE_DATA_SIZE, http2_handle_promise, stream, h2->allocator); } if (!stream) { if (!reset_reason) reset_reason = SERF_ERROR_HTTP2_STREAM_CLOSED; } else switch (frametype) { case HTTP2_FRAME_TYPE_DATA: if (stream->status != H2S_OPEN && stream->status != H2S_HALFCLOSED_LOCAL) { reset_reason = SERF_ERROR_HTTP2_STREAM_CLOSED; } break; case HTTP2_FRAME_TYPE_HEADERS: if (stream->status != H2S_OPEN && stream->status != H2S_HALFCLOSED_LOCAL && stream->status != H2S_IDLE && stream->status != H2S_RESERVED_REMOTE) { reset_reason = SERF_ERROR_HTTP2_STREAM_CLOSED; } break; case HTTP2_FRAME_TYPE_PUSH_PROMISE: if (stream->status != H2S_OPEN && stream->status != H2S_HALFCLOSED_LOCAL) { reset_reason = SERF_ERROR_HTTP2_STREAM_CLOSED; } break; } if (reset_reason) { if (stream) { serf_http2__stream_reset(stream, reset_reason, TRUE); } else { serf_http2__enqueue_stream_reset(h2, sid, reset_reason); } } if (frametype == HTTP2_FRAME_TYPE_HEADERS || frametype == HTTP2_FRAME_TYPE_PUSH_PROMISE) { if (!(frameflags & HTTP2_FLAG_END_HEADERS)) { /* This header frame is *directly* followed by continuation frames... We hide this from the stream code, by providing an aggregate that will read through the body of multiple frames */ h2->continuation_bucket = serf_bucket_aggregate_create(h2->allocator); h2->continuation_streamid = sid; serf_bucket_aggregate_append( h2->continuation_bucket, body); serf_bucket_aggregate_hold_open( h2->continuation_bucket, http2_handle_continuation, h2); body = h2->continuation_bucket; } if (stream && !reset_reason) { body = serf_http2__stream_handle_hpack( stream, body, frametype, (frameflags & HTTP2_FLAG_END_STREAM), HTTP2_MAX_HEADER_ENTRYSIZE, h2->hpack_tbl, h2->config, h2->allocator); } else { /* Even when we don't want to process the headers we must read them to update the HPACK state */ body = serf__bucket_hpack_decode_create( body, HTTP2_MAX_HEADER_ENTRYSIZE, h2->hpack_tbl, h2->allocator); } } else if (!reset_reason) { /* We have a data bucket */ body = serf_http2__stream_handle_data( stream, body, frametype, (frameflags & HTTP2_FLAG_END_STREAM), h2->config, h2->allocator); } if (body) { /* We will take care of discarding */ process_bucket = body; } else { /* The stream wants to handle the reading itself */ process_handler = serf_http2__stream_processor; process_baton = stream; } break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_PRIORITY: if (sid == 0) { /* The PRIORITY frame always identifies a stream. If a PRIORITY frame is received with a stream identifier of 0x0, the recipient MUST respond with a connection error (Section 5.4.1) of type PROTOCOL_ERROR.*/ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } else if (serf_bucket_get_remaining(body) != HTTP2_PRIORITY_DATA_SIZE) { /* A PRIORITY frame with a length other than 5 octets MUST be treated as a stream error (Section 5.4.2) of type FRAME_SIZE_ERROR.*/ /* ### But we currently upgrade this to a connection error */ return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; } stream = serf_http2__stream_get(h2, sid, TRUE, TRUE); if (stream) { body = serf_bucket_prefix_create( body, HTTP2_PRIORITY_DATA_SIZE, http2_handle_priority, stream, h2->allocator); } /* Just reading will do the right thing now */ process_bucket = body; break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_RST_STREAM: if (sid == 0) { /* RST_STREAM frames MUST be associated with a stream. If a RST_STREAM frame is received with a stream identifier of 0x0, the recipient MUST treat this as a connection error (Section 5.4.1) of type PROTOCOL_ERROR. */ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } else if (serf_bucket_get_remaining(body) != HTTP2_RST_DATA_SIZE) { /* A RST_STREAM frame with a length other than 4 octets MUST be treated as a connection error (Section 5.4.1) of type FRAME_SIZE_ERROR. */ return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; } stream = serf_http2__stream_get(h2, sid, TRUE, TRUE); if (stream) { body = serf_bucket_prefix_create( body, HTTP2_RST_DATA_SIZE, http2_handle_frame_reset, stream, h2->allocator); } /* Just reading will do the right thing now */ process_bucket = body; break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_SETTINGS: if (sid != 0) { /* SETTINGS frames always apply to a connection, never a single stream. The stream identifier for a SETTINGS frame MUST be zero (0x0). If an endpoint receives a SETTINGS frame whose stream identifier field is anything other than 0x0, the endpoint MUST respond with a connection error (Section 5.4.1) of type PROTOCOL_ERROR. */ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } remaining = (apr_size_t)serf_bucket_get_remaining(body); if (frameflags & HTTP2_FLAG_ACK) { if (remaining != 0) { /* When this bit is set, the payload of the SETTINGS frame MUST be empty. Receipt of a SETTINGS frame with the ACK flag set and a length field value other than 0 MUST be treated as a connection error (Section 5.4.1) of type FRAME_SIZE_ERROR. */ return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; } h2->setting_acks++; } else if ((remaining % HTTP2_SETTING_SIZE) != 0) { /* A SETTINGS frame with a length other than a multiple of 6 octets MUST be treated as a connection error (Section 5.4.1) of type FRAME_SIZE_ERROR. */ return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; } else { /* Just read everything... We checked it against our max-framesize */ body = serf_bucket_prefix_create(body, remaining, http2_handle_settings, h2, h2->allocator); } /* Just reading will do the right thing now */ process_bucket = body; break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_PING: if (sid != 0) { /* PING frames are not associated with any individual stream. If a PING frame is received with a stream identifier field value other than 0x0, the recipient MUST respond with a connection error (Section 5.4.1) of type PROTOCOL_ERROR.*/ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } else if (serf_bucket_get_remaining(body) != HTTP2_PING_DATA_SIZE) { /* Receipt of a PING frame with a length field value other than 8 MUST be treated as a connection error (Section 5.4.1) of type FRAME_SIZE_ERROR.. */ return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; } body = serf_bucket_prefix_create( body, HTTP2_PING_DATA_SIZE, (frameflags & HTTP2_FLAG_ACK) ? http2_handle_ping : http2_handle_ping_ack, h2, h2->allocator); /* Just reading will do the right thing now */ process_bucket = body; break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_GOAWAY: if (sid != 0) { /* The GOAWAY frame applies to the connection, not a specific stream. An endpoint MUST treat a GOAWAY frame with a stream identifier other than 0x0 as a connection error (Section 5.4.1) of type PROTOCOL_ERROR. */ return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } /* As the final go-away frame is best effort only we are not checking the bodysize against HTTP2_GOWAWAY_DATA_SIZE here. We'll see what we get in the goaway handler. Go away frames may contain additional opaque debug information at the end, so instead of reading HTTP2_GOWAWAY_DATA_SIZE bytes, we just read the whole frame. */ remaining = (apr_size_t)serf_bucket_get_remaining(body); body = serf_bucket_prefix_create(body, MIN(remaining, HTTP2_GOWAWAY_DATA_SIZE + 256), http2_handle_goaway, h2, h2->allocator); /* Just reading will do the right thing now */ process_bucket = body; break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_WINDOW_UPDATE: if (serf_bucket_get_remaining(body) != HTTP2_WINDOW_UPDATE_DATA_SIZE) { /* A WINDOW_UPDATE frame with a length other than 4 octets MUST be treated as a connection error (Section 5.4.1) of type FRAME_SIZE_ERROR. */ return SERF_ERROR_HTTP2_FRAME_SIZE_ERROR; } if (sid == 0) { body = serf_bucket_prefix_create( body, HTTP2_WINDOW_UPDATE_DATA_SIZE, http2_handle_connection_window_update, h2, h2->allocator); } else { stream = serf_http2__stream_get(h2, sid, TRUE, TRUE); if (stream) body = serf_bucket_prefix_create( body, HTTP2_WINDOW_UPDATE_DATA_SIZE, http2_handle_stream_window_update, stream, h2->allocator); } /* Just reading will do the right thing now */ process_bucket = body; break; /* ---------------------------------------------------- */ case HTTP2_FRAME_TYPE_CONTINUATION: if (!h2->continuation_bucket || sid != h2->continuation_streamid) { /* A CONTINUATION frame MUST be preceded by a HEADERS, PUSH_PROMISE or CONTINUATION frame without the END_HEADERS flag set. A recipient that observes violation of this rule MUST respond with a connection error(Section 5.4.1) of type PROTOCOL_ERROR. */ h2->continuation_bucket = NULL; h2->continuation_streamid = 0; return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } serf_bucket_aggregate_append(h2->continuation_bucket, body); if (frameflags & HTTP2_FLAG_END_HEADERS) { h2->continuation_bucket = NULL; h2->continuation_streamid = 0; } return APR_SUCCESS; /* ---------------------------------------------------- */ default: /* We explicitly ignore all other frames as required, so reading will do the right thing now */ process_bucket = body; } /* switch */ if (body) serf_bucket_set_config(body, h2->config); SERF_H2_assert(h2->processor == NULL); if (process_handler) { h2->processor = process_handler; h2->processor_baton = process_baton; } else { SERF_H2_assert(process_bucket != NULL); h2->processor = http2_bucket_processor; h2->processor_baton = process_bucket; } } } /* while(TRUE) */ } static apr_status_t http2_write_data(serf_http2_protocol_t *h2) { serf_http2_stream_t *stream = h2->cur_writable; while (h2->lr_window > 0) { apr_status_t status; if (!stream) stream = h2->first_writable; if (!stream) return APR_SUCCESS; if (stream->status != H2S_OPEN && stream->status != H2S_HALFCLOSED_REMOTE) { /* This stream is NOT writable. Remove it */ if (stream->prev_writable) stream->prev_writable->next_writable = stream->next_writable; else h2->first_writable = stream->next_writable; if (stream->next_writable) stream->next_writable->prev_writable = stream->prev_writable; else h2->last_writable = stream->prev_writable; stream->prev_writable = stream->next_writable = NULL; stream = NULL; continue; } status = serf_http2__stream_write_data(stream); if (status || stream->lr_window == 0) h2->cur_writable = stream->next_writable; return status ? status : APR_EAGAIN; } return APR_SUCCESS; /* Done for now */ } static apr_status_t http2_outgoing_read(serf_connection_t *conn) { serf_http2_protocol_t *h2 = conn->protocol_baton; apr_status_t status; status = http2_process(h2); if (!status) return APR_SUCCESS; else if (APR_STATUS_IS_EOF(status)) { /* TODO: Teardown connection, reset if necessary, etc. */ return status; } else if (APR_STATUS_IS_EAGAIN(status) || status == SERF_ERROR_WAIT_CONN) { /* Update pollset, etc. etc. */ return APR_SUCCESS; } else return status; } static apr_status_t http2_outgoing_write(serf_connection_t *conn) { serf_http2_protocol_t *h2 = conn->protocol_baton; apr_status_t status; if (conn->unwritten_reqs && conn->nr_of_written_reqs < h2->lr_max_concurrent) { status = enqueue_http2_request(h2); if (status) return status; } status = serf_pump__write(h2->pump, true); if (!status) status = http2_write_data(h2); if (APR_STATUS_IS_EAGAIN(status)) return APR_SUCCESS; else if (status) return status; /* Probably nothing to write. Connection will check new requests */ serf_io__set_pollset_dirty(&conn->io); return APR_SUCCESS; } static apr_status_t http2_outgoing_hangup(serf_connection_t *conn) { /* serf_http2_protocol_t *ctx = conn->protocol_baton; */ return APR_EGENERAL; } static void http2_outgoing_pre_teardown(serf_connection_t *conn) { serf_http2_protocol_t *h2 = conn->protocol_baton; serf_http2_stream_t *s = h2->first; while (s) { serf_http2__stream_pre_cleanup(s); s = s->next; } } static void http2_outgoing_teardown(serf_connection_t *conn) { serf_http2_protocol_t *ctx = conn->protocol_baton; apr_pool_destroy(ctx->pool); conn->protocol_baton = NULL; } static void http2_cancel_request(serf_request_t *rq, apr_status_t reason) { serf_connection_t *conn = rq->conn; if (!conn || !conn->protocol_baton || !rq->protocol_baton) return; serf_http2__stream_cancel_request(rq->protocol_baton, rq, reason); } static void http2_prioritize_request(serf_request_t *rq, bool exclusive) { serf_connection_t *conn = rq->conn; if (!conn || !conn->protocol_baton || !rq->protocol_baton) return; serf_http2__stream_prioritize_request(rq->protocol_baton, rq, exclusive); } static apr_status_t http2_incoming_read(serf_incoming_t *client) { apr_status_t status; serf_http2_protocol_t *h2 = client->protocol_baton; if (h2->prefix_left) { serf_bucket_t *stream; if (client->proto_peek_bkt) stream = client->proto_peek_bkt; else stream = client->pump.stream; do { const char *data; apr_size_t len; status = serf_bucket_read(stream, h2->prefix_left, &data, &len); if (!SERF_BUCKET_READ_ERROR(status)) { if (len && memcmp(data, HTTP2_CONNECTION_PREFIX - h2->prefix_left - 1 + sizeof(HTTP2_CONNECTION_PREFIX), len) != 0) { return SERF_ERROR_HTTP2_PROTOCOL_ERROR; } h2->prefix_left -= len; } } while (status == APR_SUCCESS && h2->prefix_left); if (!h2->prefix_left && client->proto_peek_bkt) { /* Peek buffer is now empty. Use actual stream */ serf_bucket_destroy(client->proto_peek_bkt); client->proto_peek_bkt = NULL; } if (APR_STATUS_IS_EAGAIN(status) || status == SERF_ERROR_WAIT_CONN) { return APR_SUCCESS; } else if (status) { return status; } } status = http2_process(h2); if (!status) return APR_SUCCESS; else if (APR_STATUS_IS_EOF(status)) { /* TODO: Teardown connection, reset if necessary, etc. */ return status; } else if (APR_STATUS_IS_EAGAIN(status) || status == SERF_ERROR_WAIT_CONN) { /* Update pollset, etc. etc. */ return APR_SUCCESS; } else return status; } static apr_status_t http2_incoming_write(serf_incoming_t *client) { serf_http2_protocol_t *h2 = client->protocol_baton; apr_status_t status; status = serf_pump__write(h2->pump, true); if (APR_STATUS_IS_EAGAIN(status)) return APR_SUCCESS; else if (status) return status; status = http2_write_data(h2); if (status) return status; /* Probably nothing to write. Connection will check new requests */ serf_io__set_pollset_dirty(&client->io); return APR_SUCCESS; } static apr_status_t http2_incoming_hangup(serf_incoming_t *client) { /* serf_http2_protocol_t *ctx = conn->protocol_baton; */ return APR_EGENERAL; } static void http2_incoming_teardown(serf_incoming_t *client) { serf_http2_protocol_t *ctx = client->protocol_baton; apr_pool_destroy(ctx->pool); client->protocol_baton = NULL; } static void http2_incoming_pre_teardown(serf_incoming_t *conn) { serf_http2_protocol_t *h2 = conn->protocol_baton; serf_http2_stream_t *s = h2->first; while (s) { serf_http2__stream_pre_cleanup(s); s = s->next; } } void serf_http2__allocate_stream_id(void *baton, apr_int32_t *streamid) { serf_http2_stream_t *stream = baton; SERF_H2_assert(streamid == &stream->streamid); /* Do we need to assign a new id? We do this when converting the frame to on-wire data, to avoid creating frames out of order... which would make the other side deny our frame. */ if (stream->streamid < 0) { stream->streamid = stream->h2->lr_next_streamid; stream->h2->lr_next_streamid += 2; if (stream->status == H2S_INIT) stream->status = H2S_IDLE; } } static void move_to_head(serf_http2_stream_t *stream) { /* Not implemented yet */ } serf_http2_stream_t * serf_http2__stream_get(serf_http2_protocol_t *h2, apr_int32_t streamid, bool create_for_remote, bool move_first) { serf_http2_stream_t *stream; if (streamid < 0) return NULL; for (stream = h2->first; stream; stream = stream->next) { if (stream->streamid == streamid) { if (move_first && stream != h2->first) move_to_head(stream); return stream; } } if (create_for_remote && (streamid & 0x01) == (h2->rl_next_streamid & 0x01)) { stream = serf_http2__stream_create(h2, streamid, h2->lr_default_window, h2->rl_default_window, h2->allocator); if (h2->first) { stream->next = h2->first; h2->first->prev = stream; h2->first = stream; } else h2->last = h2->first = stream; if (streamid < h2->rl_next_streamid) { /* https://tools.ietf.org/html/rfc7540#section-5.1.1 The first use of a new stream identifier implicitly closes all streams in the "idle" state that might have been initiated by that peer with a lower-valued stream identifier. */ stream->status = H2S_CLOSED; } else h2->rl_next_streamid = (streamid + 2); return stream; } return NULL; } apr_status_t serf_http2__enqueue_stream_reset(serf_http2_protocol_t *h2, apr_int32_t streamid, apr_status_t reason) { serf_bucket_t *bkt; apr_int32_t http_reason; if (reason >= SERF_ERROR_HTTP2_NO_ERROR && reason <= SERF_ERROR_HTTP2_HTTP_1_1_REQUIRED) { http_reason = (reason - SERF_ERROR_HTTP2_NO_ERROR); } else http_reason = SERF_ERROR_HTTP2_INTERNAL_ERROR; bkt = serf_bucket_create_numberv(h2->allocator, "4", http_reason); return serf_http2__enqueue_frame( h2, serf__bucket_http2_frame_create(bkt, HTTP2_FRAME_TYPE_RST_STREAM, 0, &streamid, NULL, NULL, h2->lr_max_framesize, h2->allocator), TRUE); } apr_status_t serf_http2__setup_incoming_request(serf_incoming_request_t **in_request, serf_incoming_request_setup_t *req_setup, void **req_setup_baton, serf_http2_protocol_t *h2) { if (!h2->client) return SERF_ERROR_HTTP2_PROTOCOL_ERROR; *in_request = serf__incoming_request_create(h2->client); *req_setup = h2->client->req_setup; *req_setup_baton = h2->client->req_setup_baton; return APR_SUCCESS; } apr_size_t serf_http2__max_payload_size(serf_http2_protocol_t *h2) { return h2->lr_max_framesize; } apr_size_t serf_http2__alloc_window(serf_http2_protocol_t *h2, serf_http2_stream_t *stream, apr_size_t requested) { if (requested > h2->lr_max_framesize) requested = h2->lr_max_framesize; if (requested > h2->lr_window) requested = h2->lr_window; if (requested > stream->lr_window) requested = stream->lr_window; if (requested) { h2->lr_window -= requested; stream->lr_window -= requested; } return requested; } void serf_http2__return_window(serf_http2_protocol_t *h2, serf_http2_stream_t *stream, apr_size_t returned) { SERF_H2_assert(h2->lr_window + returned <= HTTP2_WINDOW_MAX_ALLOWED); SERF_H2_assert(stream->lr_window + returned <= HTTP2_WINDOW_MAX_ALLOWED); h2->lr_window += returned; stream->lr_window += returned; } void serf_http2__ensure_writable(serf_http2_stream_t *stream) { serf_http2_protocol_t *h2 = stream->h2; SERF_H2_assert(stream->status == H2S_OPEN || stream->status == H2S_HALFCLOSED_REMOTE); if (stream->next_writable || stream->prev_writable) return; stream->prev_writable = h2->last_writable; h2->last_writable = stream; if (stream->prev_writable) stream->prev_writable->next_writable = stream; else h2->first_writable = stream; }