diff --git a/.gitignore b/.gitignore index 493d95c59f9..3a10f04cd74 100644 --- a/.gitignore +++ b/.gitignore @@ -87,6 +87,8 @@ iocore/eventsystem/test_Event proxy/config/records.config.default proxy/config/storage.config.default proxy/http2/test_Huffmancode +proxy/http2/test_Http2DependencyTree +proxy/http2/test_Http2PriorityQueue plugins/header_rewrite/header_rewrite_test plugins/experimental/esi/*_test diff --git a/mgmt/RecordsConfig.cc b/mgmt/RecordsConfig.cc index 0e963c9c855..9204be77472 100644 --- a/mgmt/RecordsConfig.cc +++ b/mgmt/RecordsConfig.cc @@ -1975,6 +1975,8 @@ static const RecordElement RecordsConfig[] = //############ {RECT_CONFIG, "proxy.config.http2.enabled", RECD_INT, "0", RECU_RESTART_TM, RR_NULL, RECC_INT, "[0-1]", RECA_NULL} , + {RECT_CONFIG, "proxy.config.http2.stream_priority_enabled", RECD_INT, "0", RECU_RESTART_TM, RR_NULL, RECC_INT, "[0-1]", RECA_NULL} + , {RECT_CONFIG, "proxy.config.http2.max_concurrent_streams_in", RECD_INT, "100", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL} , {RECT_CONFIG, "proxy.config.http2.initial_window_size_in", RECD_INT, "1048576", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL} diff --git a/proxy/FetchSM.h b/proxy/FetchSM.h index 12ca627ceb3..460cfb41dcd 100644 --- a/proxy/FetchSM.h +++ b/proxy/FetchSM.h @@ -129,6 +129,11 @@ class FetchSM : public Continuation is_internal_request = val; } + VIO* get_read_vio() + { + return read_vio; + } + private: int InvokePlugin(int event, void *data); void InvokePluginExt(int error_event = 0); diff --git a/proxy/http2/HTTP2.cc b/proxy/http2/HTTP2.cc index 5cb9d79048e..b438bf58194 100644 --- a/proxy/http2/HTTP2.cc +++ b/proxy/http2/HTTP2.cc @@ -316,15 +316,19 @@ http2_parse_headers_parameter(IOVec iov, Http2HeadersParameter ¶ms) } bool -http2_parse_priority_parameter(IOVec iov, Http2Priority ¶ms) +http2_parse_priority_parameter(IOVec iov, Http2Priority &priority) { byte_pointer ptr(iov.iov_base); byte_addressable_value dependency; memcpy_and_advance(dependency.bytes, ptr); - memcpy_and_advance(params.weight, ptr); - params.stream_dependency = ntohl(dependency.value); + priority.exclusive_flag = dependency.bytes[0] & 0x80; + + dependency.bytes[0] &= 0x7f; // Clear the highest bit for exclusive flag + priority.stream_dependency = ntohl(dependency.value); + + memcpy_and_advance(priority.weight, ptr); return true; } @@ -727,6 +731,7 @@ http2_decode_header_blocks(HTTPHdr *hdr, const uint8_t *buf_start, const uint8_t } // Initialize this subsystem with librecords configs (for now) +bool Http2::stream_priority_enabled = 0; uint32_t Http2::max_concurrent_streams = 100; uint32_t Http2::initial_window_size = 1048576; uint32_t Http2::max_frame_size = 16384; @@ -740,6 +745,7 @@ uint32_t Http2::active_timeout_in = 0; void Http2::init() { + REC_EstablishStaticConfigBool(stream_priority_enabled, "proxy.config.http2.stream_priority_enabled"); REC_EstablishStaticConfigInt32U(max_concurrent_streams, "proxy.config.http2.max_concurrent_streams_in"); REC_EstablishStaticConfigInt32U(initial_window_size, "proxy.config.http2.initial_window_size_in"); REC_EstablishStaticConfigInt32U(max_frame_size, "proxy.config.http2.max_frame_size"); diff --git a/proxy/http2/HTTP2.h b/proxy/http2/HTTP2.h index 10e22e30d3c..9d74dfd8e0c 100644 --- a/proxy/http2/HTTP2.h +++ b/proxy/http2/HTTP2.h @@ -60,6 +60,10 @@ const uint32_t HTTP2_MAX_FRAME_SIZE = 16384; const uint32_t HTTP2_HEADER_TABLE_SIZE = 4096; const uint32_t HTTP2_MAX_HEADER_LIST_SIZE = UINT_MAX; +// [RFC 7540] 5.3.5 Default Priorities +const uint32_t HTTP2_PRIORITY_DEFAULT_STREAM_DEPENDENCY = 0; +const uint8_t HTTP2_PRIORITY_DEFAULT_WEIGHT = 15; + // Statistics enum { HTTP2_STAT_CURRENT_CLIENT_SESSION_COUNT, // Current # of active HTTP2 @@ -247,10 +251,14 @@ struct Http2SettingsParameter { // [RFC 7540] 6.3 PRIORITY Format struct Http2Priority { - Http2Priority() : stream_dependency(0), weight(15) {} + Http2Priority() + : exclusive_flag(false), weight(HTTP2_PRIORITY_DEFAULT_WEIGHT), stream_dependency(HTTP2_PRIORITY_DEFAULT_STREAM_DEPENDENCY) + { + } - uint32_t stream_dependency; + bool exclusive_flag; uint8_t weight; + uint32_t stream_dependency; }; // [RFC 7540] 6.2 HEADERS Format @@ -342,6 +350,7 @@ int64_t http2_write_header_fragment(HTTPHdr *, MIMEFieldIter &, uint8_t *, uint6 class Http2 { public: + static bool stream_priority_enabled; static uint32_t max_concurrent_streams; static uint32_t initial_window_size; static uint32_t max_frame_size; diff --git a/proxy/http2/Http2ConnectionState.cc b/proxy/http2/Http2ConnectionState.cc index 077f8ea8186..0d2f2a1a6cb 100644 --- a/proxy/http2/Http2ConnectionState.cc +++ b/proxy/http2/Http2ConnectionState.cc @@ -36,7 +36,7 @@ typedef Http2Error (*http2_frame_dispatch)(Http2ConnectionState &, const Http2Frame &); static const int buffer_size_index[HTTP2_FRAME_TYPE_MAX] = { - BUFFER_SIZE_INDEX_8K, // HTTP2_FRAME_TYPE_DATA + BUFFER_SIZE_INDEX_16K, // HTTP2_FRAME_TYPE_DATA BUFFER_SIZE_INDEX_16K, // HTTP2_FRAME_TYPE_HEADERS -1, // HTTP2_FRAME_TYPE_PRIORITY BUFFER_SIZE_INDEX_128, // HTTP2_FRAME_TYPE_RST_STREAM @@ -190,6 +190,8 @@ rcv_headers_frame(Http2ConnectionState &cstate, const Http2Frame &frame) } Http2Stream *stream = NULL; + bool new_stream = false; + if (stream_id <= cstate.get_latest_stream_id()) { stream = cstate.find_stream(stream_id); if (stream == NULL || !stream->has_trailing_header()) { @@ -197,6 +199,7 @@ rcv_headers_frame(Http2ConnectionState &cstate, const Http2Frame &frame) } } else { // Create new stream + new_stream = true; stream = cstate.create_stream(stream_id); if (!stream) { return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); @@ -236,22 +239,29 @@ rcv_headers_frame(Http2ConnectionState &cstate, const Http2Frame &frame) header_block_fragment_length -= (HTTP2_HEADERS_PADLEN_LEN + params.pad_length); } - // NOTE: Parse priority parameters if exists - // TODO: Currently priority is NOT supported. TS-3535 will fix this. - if (frame.header().flags & HTTP2_FLAGS_HEADERS_PRIORITY) { - uint8_t buf[HTTP2_PRIORITY_LEN] = {0}; + if (new_stream) { + // NOTE: Parse priority parameters if exists + if (frame.header().flags & HTTP2_FLAGS_HEADERS_PRIORITY) { + uint8_t buf[HTTP2_PRIORITY_LEN] = {0}; - frame.reader()->memcpy(buf, HTTP2_PRIORITY_LEN, header_block_fragment_offset); - if (!http2_parse_priority_parameter(make_iovec(buf, HTTP2_PRIORITY_LEN), params.priority)) { - return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); - } - // Protocol error if the stream depends on itself - if (stream_id == params.priority.stream_dependency) { - return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); + frame.reader()->memcpy(buf, HTTP2_PRIORITY_LEN, header_block_fragment_offset); + if (!http2_parse_priority_parameter(make_iovec(buf, HTTP2_PRIORITY_LEN), params.priority)) { + return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); + } + // Protocol error if the stream depends on itself + if (stream_id == params.priority.stream_dependency) { + return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); + } + + DebugHttp2Stream(cstate.ua_session, stream_id, "PRIORITY - dep: %d, weight: %d, excl: %d", params.priority.stream_dependency, + params.priority.weight, params.priority.exclusive_flag); + + header_block_fragment_offset += HTTP2_PRIORITY_LEN; + header_block_fragment_length -= HTTP2_PRIORITY_LEN; } - header_block_fragment_offset += HTTP2_PRIORITY_LEN; - header_block_fragment_length -= HTTP2_PRIORITY_LEN; + cstate.dependency_tree->add(params.priority.stream_dependency, stream_id, params.priority.weight, + params.priority.exclusive_flag, stream); } stream->header_blocks = static_cast(ats_malloc(header_block_fragment_length)); @@ -300,25 +310,49 @@ rcv_headers_frame(Http2ConnectionState &cstate, const Http2Frame &frame) return Http2Error(HTTP2_ERROR_CLASS_NONE); } +/* + * [RFC 7540] 6.3 PRIORITY + * + */ static Http2Error rcv_priority_frame(Http2ConnectionState &cstate, const Http2Frame &frame) { - DebugHttp2Stream(cstate.ua_session, frame.header().streamid, "Received PRIORITY frame"); + const Http2StreamId stream_id = frame.header().streamid; + const uint32_t payload_length = frame.header().length; + + DebugHttp2Stream(cstate.ua_session, stream_id, "Received PRIORITY frame"); // If a PRIORITY frame is received with a stream identifier of 0x0, the // recipient MUST respond with a connection error of type PROTOCOL_ERROR. - if (frame.header().streamid == 0) { + if (stream_id == 0) { return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); } // 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. - if (frame.header().length != HTTP2_PRIORITY_LEN) { + if (payload_length != HTTP2_PRIORITY_LEN) { return Http2Error(HTTP2_ERROR_CLASS_STREAM, HTTP2_ERROR_FRAME_SIZE_ERROR); } - // TODO Pick stream dependencies and weight - // Supporting PRIORITY is not essential so its temporarily ignored. + uint8_t buf[HTTP2_PRIORITY_LEN] = {0}; + frame.reader()->memcpy(buf, HTTP2_PRIORITY_LEN, 0); + + Http2Priority priority; + if (!http2_parse_priority_parameter(make_iovec(buf, HTTP2_PRIORITY_LEN), priority)) { + return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); + } + + DebugHttp2Stream(cstate.ua_session, stream_id, "PRIORITY - dep: %d, weight: %d, excl: %d", priority.stream_dependency, + priority.weight, priority.exclusive_flag); + + if (cstate.dependency_tree->find(stream_id) != NULL) { + // [RFC 7540] 5.3.3 Reprioritization + DebugHttp2Stream(cstate.ua_session, stream_id, "Reprioritize"); + + cstate.dependency_tree->reprioritize(stream_id, priority.stream_dependency, priority.exclusive_flag); + } else { + cstate.dependency_tree->add(priority.stream_dependency, stream_id, priority.weight, priority.exclusive_flag, NULL); + } return Http2Error(HTTP2_ERROR_CLASS_NONE); } @@ -531,38 +565,27 @@ rcv_window_update_frame(Http2ConnectionState &cstate, const Http2Frame &frame) { char buf[HTTP2_WINDOW_UPDATE_LEN]; uint32_t size; - const Http2StreamId sid = frame.header().streamid; + const Http2StreamId stream_id = frame.header().streamid; // A WINDOW_UPDATE frame with a length other than 4 octets MUST be // treated as a connection error of type FRAME_SIZE_ERROR. if (frame.header().length != HTTP2_WINDOW_UPDATE_LEN) { - DebugHttp2Stream(cstate.ua_session, sid, "Received WINDOW_UPDATE frame - length incorrect"); + DebugHttp2Stream(cstate.ua_session, stream_id, "Received WINDOW_UPDATE frame - length incorrect"); return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_FRAME_SIZE_ERROR); } - if (sid == 0) { - // Connection level window update - frame.reader()->memcpy(buf, sizeof(buf), 0); - http2_parse_window_update(make_iovec(buf, sizeof(buf)), size); + frame.reader()->memcpy(buf, sizeof(buf), 0); + http2_parse_window_update(make_iovec(buf, sizeof(buf)), size); - DebugHttp2Stream(cstate.ua_session, sid, "Received WINDOW_UPDATE frame - updated to: %zd delta: %u", + if (stream_id == 0) { + // Connection level window update + DebugHttp2Stream(cstate.ua_session, stream_id, "Received WINDOW_UPDATE frame - updated to: %zd delta: %u", (cstate.client_rwnd + size), size); - // A receiver MUST treat the receipt of a WINDOW_UPDATE frame with a - // connection - // flow control window increment of 0 as a connection error of type - // PROTOCOL_ERROR; if (size == 0) { return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); } - // 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. if (size > HTTP2_MAX_WINDOW_SIZE - cstate.client_rwnd) { return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_FLOW_CONTROL_ERROR); } @@ -571,44 +594,36 @@ rcv_window_update_frame(Http2ConnectionState &cstate, const Http2Frame &frame) cstate.restart_streams(); } else { // Stream level window update - Http2Stream *stream = cstate.find_stream(sid); + Http2Stream *stream = cstate.find_stream(stream_id); if (stream == NULL) { - if (sid <= cstate.get_latest_stream_id()) { + if (stream_id <= cstate.get_latest_stream_id()) { return Http2Error(HTTP2_ERROR_CLASS_NONE); } else { return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); } } - frame.reader()->memcpy(buf, sizeof(buf), 0); - http2_parse_window_update(make_iovec(buf, sizeof(buf)), size); - - DebugHttp2Stream(cstate.ua_session, sid, "Received WINDOW_UPDATE frame - updated to: %zd delta: %u", + DebugHttp2Stream(cstate.ua_session, stream_id, "Received WINDOW_UPDATE frame - updated to: %zd delta: %u", (stream->client_rwnd + size), size); - // A receiver MUST treat the receipt of a WINDOW_UPDATE frame with an - // flow control window increment of 0 as a stream error of type - // PROTOCOL_ERROR; if (size == 0) { return Http2Error(HTTP2_ERROR_CLASS_STREAM, HTTP2_ERROR_PROTOCOL_ERROR); } - // 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. if (size > HTTP2_MAX_WINDOW_SIZE - stream->client_rwnd) { return Http2Error(HTTP2_ERROR_CLASS_STREAM, HTTP2_ERROR_FLOW_CONTROL_ERROR); } stream->client_rwnd += size; ssize_t wnd = min(cstate.client_rwnd, stream->client_rwnd); + if (stream->get_state() == HTTP2_STREAM_STATE_HALF_CLOSED_REMOTE && wnd > 0) { - cstate.send_data_frame(stream->get_fetcher()); + if (Http2::stream_priority_enabled) { + cstate.schedule_stream(stream); + } else { + cstate.send_response_body(stream); + } } } @@ -744,6 +759,14 @@ Http2ConnectionState::main_event_handler(int event, void *edata) return 0; } + // Scheduling HTTP/2 Data Frame + case HTTP2_SESSION_EVENT_XMIT: { + SCOPED_MUTEX_LOCK(lock, mutex, this_ethread()); + stream_scheduler(); + + return 0; + } + // Parse received HTTP/2 frames case HTTP2_SESSION_EVENT_RECV: { const Http2Frame *frame = (Http2Frame *)edata; @@ -793,7 +816,15 @@ Http2ConnectionState::main_event_handler(int event, void *edata) // Process a part of response body from origin server case TS_FETCH_EVENT_EXT_BODY_READY: { FetchSM *fetch_sm = reinterpret_cast(edata); - this->send_data_frame(fetch_sm); + Http2Stream *stream = static_cast(fetch_sm->ext_get_user_data()); + ink_release_assert(stream != NULL); + + if (Http2::stream_priority_enabled) { + schedule_stream(stream); + } else { + send_response_body(stream); + } + return 0; } @@ -801,8 +832,16 @@ Http2ConnectionState::main_event_handler(int event, void *edata) case TS_FETCH_EVENT_EXT_BODY_DONE: { FetchSM *fetch_sm = reinterpret_cast(edata); Http2Stream *stream = static_cast(fetch_sm->ext_get_user_data()); + ink_release_assert(stream != NULL); + stream->mark_body_done(); - this->send_data_frame(fetch_sm); + + if (Http2::stream_priority_enabled) { + schedule_stream(stream); + } else { + send_response_body(stream); + } + return 0; } @@ -862,15 +901,18 @@ Http2ConnectionState::find_stream(Http2StreamId id) const void Http2ConnectionState::restart_streams() { - // Currently lookup retryable streams sequentially. - // TODO considering to stream weight and dependencies. - Http2Stream *s = stream_list.head; - while (s) { - Http2Stream *next = s->link.next; - if (s->get_state() == HTTP2_STREAM_STATE_HALF_CLOSED_REMOTE && min(this->client_rwnd, s->client_rwnd) > 0) { - this->send_data_frame(s->get_fetcher()); + if (Http2::stream_priority_enabled) { + SET_HANDLER(&Http2ConnectionState::main_event_handler); + this_ethread()->schedule_imm_local((Continuation *)this, HTTP2_SESSION_EVENT_XMIT); + } else { + Http2Stream *stream = stream_list.head; + while (stream) { + Http2Stream *next = stream->link.next; + if (stream->get_state() == HTTP2_STREAM_STATE_HALF_CLOSED_REMOTE && min(this->client_rwnd, stream->client_rwnd) > 0) { + send_response_body(stream); + } + stream = next; } - s = next; } } @@ -913,79 +955,182 @@ Http2ConnectionState::update_initial_rwnd(Http2WindowSize new_size) } } +// Schedule DATA Frames using priority. void -Http2ConnectionState::send_data_frame(FetchSM *fetch_sm) +Http2ConnectionState::stream_scheduler() { - if (fetch_sm == NULL) { + DependencyTree::Node *node = dependency_tree->top(); + + if (node == NULL) { + DebugHttp2Con(ua_session, "No active stream"); return; } - size_t buf_len = BUFFER_SIZE_FOR_INDEX(buffer_size_index[HTTP2_FRAME_TYPE_DATA]) - HTTP2_FRAME_HEADER_LEN; - uint8_t payload_buffer[buf_len]; + while (node != NULL) { + // No window left (connection level) + if (client_rwnd <= 0) { + break; + } - Http2Stream *stream = static_cast(fetch_sm->ext_get_user_data()); + Http2Stream *stream = node->t; + size_t len = 0; - if (stream->get_state() == HTTP2_STREAM_STATE_CLOSED) { - return; - } + ink_release_assert(stream != NULL); - for (;;) { - uint8_t flags = 0x00; + if (stream->get_state() == HTTP2_STREAM_STATE_CLOSED) { + dependency_tree->deactivate(node, len); + node = dependency_tree->top(); + continue; + } - // Select appropriate payload size - if (this->client_rwnd <= 0 || stream->client_rwnd <= 0) + // No window left (stream level) + if (stream->client_rwnd <= 0) { + dependency_tree->deactivate(node, len); break; - size_t window_size = min(this->client_rwnd, stream->client_rwnd); - size_t send_size = min(buf_len, window_size); + } - size_t payload_length = fetch_sm->ext_read_data(reinterpret_cast(payload_buffer), send_size); + FetchSM *fetch_sm = stream->get_fetcher(); + if (fetch_sm == NULL) { + dependency_tree->deactivate(node, len); + break; + } - // If we break here, we never send the END_STREAM in the case of a - // early terminating OS. Ok if there is no body yet. Otherwise - // continue on to delete the stream - if (payload_length == 0 && !stream->is_body_done()) { + // Lock mutex of PluginVC + SCOPED_MUTEX_LOCK(lock, fetch_sm->get_read_vio()->mutex, this_ethread()); + Http2Error error = send_data_frame(stream, len); + + if (error.cls != HTTP2_ERROR_CLASS_NONE) { + dependency_tree->deactivate(node, len); break; } - // Update window size - this->client_rwnd -= payload_length; - stream->client_rwnd -= payload_length; + // No response body to send yet + if (len == 0 && !stream->is_body_done()) { + dependency_tree->deactivate(node, len); + break; + } - if (stream->is_body_done() && payload_length < send_size) { - flags |= HTTP2_FLAGS_DATA_END_STREAM; + if (stream->get_state() == HTTP2_STREAM_STATE_CLOSED) { + dependency_tree->deactivate(node, len); + delete_stream(stream); + } else { + dependency_tree->update(node, len); } - // Create frame - DebugHttp2Stream(ua_session, stream->get_id(), "Send DATA frame - client window con: %zd stream: %zd payload: %zd", client_rwnd, - stream->client_rwnd, payload_length); - Http2Frame data(HTTP2_FRAME_TYPE_DATA, stream->get_id(), flags); - data.alloc(buffer_size_index[HTTP2_FRAME_TYPE_DATA]); - http2_write_data(payload_buffer, payload_length, data.write()); - data.finalize(payload_length); - - // Change state to 'closed' if its end of DATAs. - if (flags & HTTP2_FLAGS_DATA_END_STREAM) { - DebugHttp2Stream(ua_session, stream->get_id(), "End of DATA frame"); - if (!stream->change_state(data.header().type, data.header().flags)) { - this->send_goaway_frame(stream->get_id(), HTTP2_ERROR_PROTOCOL_ERROR); - } + node = dependency_tree->top(); + } + + _scheduled = false; +} + +void +Http2ConnectionState::schedule_stream(Http2Stream *stream) +{ + DependencyTree::Node *node = dependency_tree->find(stream->get_id()); + ink_release_assert(node != NULL); + + dependency_tree->activate(node); + + if (!_scheduled) { + SET_HANDLER(&Http2ConnectionState::main_event_handler); + this_ethread()->schedule_imm_local((Continuation *)this, HTTP2_SESSION_EVENT_XMIT); + _scheduled = true; + } +} + +void +Http2ConnectionState::send_response_body(Http2Stream *stream) +{ + if (stream->get_state() == HTTP2_STREAM_STATE_CLOSED) { + DebugHttp2Stream(ua_session, stream->get_id(), "Send Response Body - stream is closed"); + return; + } + size_t len = 0; + for (;;) { + if (this->client_rwnd <= 0 || stream->client_rwnd <= 0) { + break; } - // xmit event - SCOPED_MUTEX_LOCK(lock, this->ua_session->mutex, this_ethread()); - this->ua_session->handleEvent(HTTP2_SESSION_EVENT_XMIT, &data); - - if (flags & HTTP2_FLAGS_DATA_END_STREAM) { - // Delete a stream immediately - // TODO its should not be deleted for a several time to handling - // RST_STREAM and WINDOW_UPDATE. - // See 'closed' state written at [RFC 7540] 5.1. - this->delete_stream(stream); + Http2Error error = send_data_frame(stream, len); + + if (error.cls != HTTP2_ERROR_CLASS_NONE) { + break; + } + + if (len == 0 && !stream->is_body_done()) { + break; + } + + if (stream->get_state() == HTTP2_STREAM_STATE_CLOSED) { + delete_stream(stream); break; } } } +Http2Error +Http2ConnectionState::send_data_frame(Http2Stream *stream, size_t &payload_length) +{ + if (stream == NULL) { + DebugHttp2Con(ua_session, "Stream is NULL"); + + return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); + } + + FetchSM *fetch_sm = stream->get_fetcher(); + if (fetch_sm == NULL) { + DebugHttp2Con(ua_session, "FetchSM is NULL"); + + return Http2Error(HTTP2_ERROR_CLASS_CONNECTION, HTTP2_ERROR_PROTOCOL_ERROR); + } + + size_t buf_len = BUFFER_SIZE_FOR_INDEX(buffer_size_index[HTTP2_FRAME_TYPE_DATA]) - HTTP2_FRAME_HEADER_LEN; + uint8_t payload_buffer[buf_len]; + size_t window_size = min(this->client_rwnd, stream->client_rwnd); + size_t send_size = min(buf_len, window_size); + payload_length = fetch_sm->ext_read_data(reinterpret_cast(payload_buffer), send_size); + + // If we break here, we never send the END_STREAM in the case of a + // early terminating OS. Ok if there is no body yet. Otherwise + // continue on to delete the stream + if (payload_length == 0 && !stream->is_body_done()) { + DebugHttp2Stream(ua_session, stream->get_id(), "Send DATA frame - body is not yet - client window con: %zd stream: %zd", + client_rwnd, stream->client_rwnd); + return Http2Error(HTTP2_ERROR_CLASS_NONE); + } + + // Update window size + this->client_rwnd -= payload_length; + stream->client_rwnd -= payload_length; + + uint8_t flags = 0x00; + if (stream->is_body_done() && payload_length < send_size) { + flags |= HTTP2_FLAGS_DATA_END_STREAM; + } + + // Create frame + DebugHttp2Stream(ua_session, stream->get_id(), "Send DATA frame - client window con: %zd stream: %zd payload: %zd", client_rwnd, + stream->client_rwnd, payload_length); + Http2Frame data(HTTP2_FRAME_TYPE_DATA, stream->get_id(), flags); + data.alloc(buffer_size_index[HTTP2_FRAME_TYPE_DATA]); + http2_write_data(payload_buffer, payload_length, data.write()); + data.finalize(payload_length); + + // Change state to 'closed' if its end of DATAs. + if (flags & HTTP2_FLAGS_DATA_END_STREAM) { + DebugHttp2Stream(ua_session, stream->get_id(), "End of DATA frame"); + if (!stream->change_state(data.header().type, data.header().flags)) { + this->send_goaway_frame(stream->get_id(), HTTP2_ERROR_PROTOCOL_ERROR); + } + } + + // xmit event + SCOPED_MUTEX_LOCK(lock, this->ua_session->mutex, this_ethread()); + this->ua_session->handleEvent(HTTP2_SESSION_EVENT_XMIT, &data); + + return Http2Error(HTTP2_ERROR_CLASS_NONE); +} + void Http2ConnectionState::send_headers_frame(FetchSM *fetch_sm) { diff --git a/proxy/http2/Http2ConnectionState.h b/proxy/http2/Http2ConnectionState.h index 9423c4e13f9..73cfb4b5b5b 100644 --- a/proxy/http2/Http2ConnectionState.h +++ b/proxy/http2/Http2ConnectionState.h @@ -28,6 +28,7 @@ #include "HPACK.h" #include "FetchSM.h" #include "Http2Stream.h" +#include "Http2DependencyTree.h" class Http2ClientSession; @@ -104,9 +105,12 @@ class Http2ConnectionSettings class Http2ConnectionState : public Continuation { public: + typedef Http2DependencyTree DependencyTree; + Http2ConnectionState() - : Continuation(NULL), ua_session(NULL), client_rwnd(HTTP2_INITIAL_WINDOW_SIZE), server_rwnd(Http2::initial_window_size), - stream_list(), latest_streamid(0), client_streams_count(0), continued_stream_id(0) + : Continuation(NULL), ua_session(NULL), dependency_tree(NULL), client_rwnd(HTTP2_INITIAL_WINDOW_SIZE), + server_rwnd(Http2::initial_window_size), stream_list(), latest_streamid(0), client_streams_count(0), continued_stream_id(0), + _scheduled(false) { SET_HANDLER(&Http2ConnectionState::main_event_handler); } @@ -114,6 +118,7 @@ class Http2ConnectionState : public Continuation Http2ClientSession *ua_session; Http2IndexingTable *local_indexing_table; Http2IndexingTable *remote_indexing_table; + DependencyTree *dependency_tree; // Settings. Http2ConnectionSettings server_settings; @@ -127,6 +132,8 @@ class Http2ConnectionState : public Continuation continued_buffer.iov_base = NULL; continued_buffer.iov_len = 0; + + dependency_tree = new DependencyTree(); } void @@ -139,6 +146,8 @@ class Http2ConnectionState : public Continuation delete remote_indexing_table; ats_free(continued_buffer.iov_base); + + delete dependency_tree; } // Event handlers @@ -180,8 +189,10 @@ class Http2ConnectionState : public Continuation // Connection level window size ssize_t client_rwnd, server_rwnd; - // HTTP/2 frame sender - void send_data_frame(FetchSM *fetch_sm); + void stream_scheduler(); + void schedule_stream(Http2Stream *stream); + void send_response_body(Http2Stream *stream); + Http2Error send_data_frame(Http2Stream *stream, size_t &payload_length); void send_headers_frame(FetchSM *fetch_sm); void send_rst_stream_frame(Http2StreamId id, Http2ErrorCode ec); void send_settings_frame(const Http2ConnectionSettings &new_settings); @@ -220,6 +231,7 @@ class Http2ConnectionState : public Continuation // another CONTINUATION frame." Http2StreamId continued_stream_id; IOVec continued_buffer; + bool _scheduled; }; #endif // __HTTP2_CONNECTION_STATE_H__ diff --git a/proxy/http2/Http2DependencyTree.h b/proxy/http2/Http2DependencyTree.h new file mode 100644 index 00000000000..f264d790994 --- /dev/null +++ b/proxy/http2/Http2DependencyTree.h @@ -0,0 +1,279 @@ +/** @file + + HTTP/2 Dependency Tree + + The original idea of Stream Priority Algorithm using Weighted Fair Queue (WFQ) + Scheduling is invented by Kazuho Oku (H2O project). + + @section license License + + 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. + */ + +#ifndef __HTTP2_DEP_TREE_H__ +#define __HTTP2_DEP_TREE_H__ + +#include "ts/List.h" +#include "ts/Diags.h" +#include "HTTP2.h" +#include "Http2PriorityQueue.h" + +// TODO: K is a constant, 256 is temporal value. +const static int K = 256; + +template class Http2DependencyTree +{ +public: + class Node + { + public: + Node() + : active(false), queued(false), id(HTTP2_PRIORITY_DEFAULT_STREAM_DEPENDENCY), weight(HTTP2_PRIORITY_DEFAULT_WEIGHT), point(0), + parent(NULL), t(NULL) + { + entry = new Http2PriorityQueueEntry(this); + } + Node(uint32_t i, uint32_t w, uint32_t p, Node *n, T t) + : active(false), queued(false), id(i), weight(w), point(p), parent(n), t(t) + { + entry = new Http2PriorityQueueEntry(this); + } + + ~Node() + { + delete entry; + + // delete all child nodes + if (!children.empty()) { + Node *node = children.head; + Node *next = NULL; + while (node) { + next = node->link.next; + children.remove(node); + delete node; + node = next; + } + } + } + + LINK(Node, link); + + bool operator<(const Node &n) const { return point < n.point; } + bool operator>(const Node &n) const { return point > n.point; } + + bool active; + bool queued; + uint32_t id; + uint32_t weight; + uint32_t point; + Node *parent; + DLL children; + Http2PriorityQueueEntry *entry; + Http2PriorityQueue queue; + T t; + }; + + Http2DependencyTree() { _root = new Node(); } + ~Http2DependencyTree() { delete _root; } + + Node *find(uint32_t id); + void add(uint32_t parent_id, uint32_t id, uint32_t weight, bool exclusive, T t); + void reprioritize(uint32_t new_parent_id, uint32_t id, bool exclusive); + Node *top(); + void activate(Node *node); + void deactivate(Node *node, uint32_t sent); + void update(Node *node, uint32_t sent); + +private: + Node *_find(Node *node, uint32_t id); + Node *_top(Node *node); + void _change_parent(Node *new_parent, Node *node, bool exclusive); + + Node *_root; +}; + +// TODO: Add a link to Node from Http2Stream after TS-3612 to avoid calling this function. +template +typename Http2DependencyTree::Node * +Http2DependencyTree::_find(Node *node, uint32_t id) +{ + if (node->id == id) { + return node; + } + + if (node->children.empty()) { + return NULL; + } + + Node *result = NULL; + for (Node *n = node->children.head; n; n = n->link.next) { + result = _find(n, id); + if (result != NULL) { + break; + } + } + + return result; +} + +template +typename Http2DependencyTree::Node * +Http2DependencyTree::find(uint32_t id) +{ + return _find(_root, id); +} + +template +void +Http2DependencyTree::add(uint32_t parent_id, uint32_t id, uint32_t weight, bool exclusive, T t) +{ + Node *parent = find(parent_id); + if (parent == NULL) { + parent = _root; + } + + Node *node = new Node(id, weight, 0, parent, t); + + if (exclusive) { + while (Node *child = parent->children.pop()) { + node->children.push(child); + child->parent = node; + } + } + + parent->children.push(node); +} + +template +void +Http2DependencyTree::reprioritize(uint32_t id, uint32_t new_parent_id, bool exclusive) +{ + Node *node = find(id); + if (node == NULL) { + return; + } + + Node *old_parent = node->parent; + if (old_parent->id == new_parent_id) { + // Do nothing + return; + } + + Node *new_parent = find(new_parent_id); + if (new_parent == NULL) { + return; + } + _change_parent(new_parent, old_parent, false); + _change_parent(node, new_parent, exclusive); +} + +// Change node's parent to new_parent +template +void +Http2DependencyTree::_change_parent(Node *node, Node *new_parent, bool exclusive) +{ + node->parent->children.remove(node); + node->parent = NULL; + + if (exclusive) { + while (Node *child = new_parent->children.pop()) { + node->children.push(child); + child->parent = node; + } + } + + new_parent->children.push(node); + node->parent = new_parent; +} + +template +typename Http2DependencyTree::Node * +Http2DependencyTree::_top(Node *node) +{ + Node *child = node; + + while (child != NULL) { + if (child->active) { + return child; + } else if (!child->queue.empty()) { + child = child->queue.top()->node; + } else { + return NULL; + } + } + + return child; +} + + +template +typename Http2DependencyTree::Node * +Http2DependencyTree::top() +{ + return _top(_root); +} + +template +void +Http2DependencyTree::activate(Node *node) +{ + node->active = true; + + while (node->parent != NULL && !node->queued) { + node->parent->queue.push(node->entry); + node->queued = true; + node = node->parent; + } +} + +template +void +Http2DependencyTree::deactivate(Node *node, uint32_t sent) +{ + node->active = false; + + while (node->queue.empty() && node->parent != NULL) { + ink_assert(node->parent->queue.top() == node->entry); + + node->parent->queue.pop(); + node->queued = false; + + node = node->parent; + } + + update(node, sent); +} + +template +void +Http2DependencyTree::update(Node *node, uint32_t sent) +{ + while (node->parent != NULL) { + node->point += sent * K / (node->weight + 1); + + if (node->queued) { + node->parent->queue.update(node->entry); + } else { + node->parent->queue.push(node->entry); + node->queued = true; + } + + node = node->parent; + } +} + +#endif // __HTTP2_DEP_TREE_H__ diff --git a/proxy/http2/Http2PriorityQueue.h b/proxy/http2/Http2PriorityQueue.h new file mode 100644 index 00000000000..afa3b9a39b0 --- /dev/null +++ b/proxy/http2/Http2PriorityQueue.h @@ -0,0 +1,194 @@ +/** @file + + HTTP/2 Priority Queue + + Priority Queue Implimentation using Min Heap. + Used by HTTP/2 Dependency Tree for WFQ Scheduling. + + @section license License + + 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. + */ + +#ifndef __HTTP2_PRIORITY_QUEUE_H__ +#define __HTTP2_PRIORITY_QUEUE_H__ + +#include "ts/Diags.h" +#include "ts/Vec.h" + +template struct Http2PriorityQueueEntry { + Http2PriorityQueueEntry(T n) : index(0), node(n) {} + + uint32_t index; + T node; +}; + +template class Http2PriorityQueue +{ +public: + Http2PriorityQueue() {} + ~Http2PriorityQueue() {} + + const bool empty(); + Http2PriorityQueueEntry *top(); + void pop(); + void push(Http2PriorityQueueEntry *); + void update(Http2PriorityQueueEntry *, bool); + const Vec *> &dump() const; + +private: + Vec *> _v; + + void _swap(uint32_t, uint32_t); + void _bubble_up(uint32_t); + void _bubble_down(uint32_t); +}; + +template +const Vec *> & +Http2PriorityQueue::dump() const +{ + return _v; +} + +template +const bool +Http2PriorityQueue::empty() +{ + return _v.length() == 0; +} + +template +void +Http2PriorityQueue::push(Http2PriorityQueueEntry *entry) +{ + ink_release_assert(entry != NULL); + + int len = _v.length(); + _v.push_back(entry); + entry->index = len; + + _bubble_up(len); +} + +template +Http2PriorityQueueEntry * +Http2PriorityQueue::top() +{ + if (empty()) { + return NULL; + } else { + return _v[0]; + } +} + +template +void +Http2PriorityQueue::pop() +{ + if (empty()) { + return; + } + + _v[0] = _v[_v.length() - 1]; + _v.pop(); + _bubble_down(0); +} + +template +void +Http2PriorityQueue::update(Http2PriorityQueueEntry *entry, bool increased = true) +{ + ink_release_assert(entry != NULL); + + if (empty()) { + return; + } + + if (increased) { + _bubble_down(entry->index); + } else { + _bubble_up(entry->index); + } +} + +template +void +Http2PriorityQueue::_swap(uint32_t i, uint32_t j) +{ + Http2PriorityQueueEntry *tmp = _v[i]; + _v[i] = _v[j]; + _v[j] = tmp; + + _v[i]->index = i; + _v[j]->index = j; +} + + +template +void +Http2PriorityQueue::_bubble_up(uint32_t index) +{ + if (empty()) { + ink_release_assert(false); + } + + uint32_t parent; + while (index != 0) { + parent = (index - 1) / 2; + if (*(_v[index]->node) < *(_v[parent]->node)) { + _swap(parent, index); + index = parent; + continue; + } + + break; + } +} + + +template +void +Http2PriorityQueue::_bubble_down(uint32_t index) +{ + if (empty()) { + // Do nothing + return; + } + + uint32_t left, right, smaller; + + while (true) { + if ((left = index * 2 + 1) >= _v.length()) { + break; + } else if ((right = index * 2 + 2) >= _v.length()) { + smaller = left; + } else { + smaller = (*(_v[left]->node) < *(_v[right]->node)) ? left : right; + } + + if (*(_v[smaller]->node) < *(_v[index]->node)) { + _swap(smaller, index); + index = smaller; + continue; + } + + break; + } +} + +#endif // __HTTP2_PRIORITY_QUEUE_H__ diff --git a/proxy/http2/Makefile.am b/proxy/http2/Makefile.am index 5912847e736..bcd96065a6e 100644 --- a/proxy/http2/Makefile.am +++ b/proxy/http2/Makefile.am @@ -42,6 +42,8 @@ libhttp2_a_SOURCES = \ Http2ConnectionState.h \ Http2DebugNames.cc \ Http2DebugNames.h \ + Http2DependencyTree.h \ + Http2PriorityQueue.h \ Http2Stream.cc \ Http2Stream.h \ Http2SessionAccept.cc \ @@ -55,9 +57,14 @@ if BUILD_TESTS endif noinst_PROGRAMS = \ - test_Huffmancode + test_Huffmancode \ + test_Http2DependencyTree \ + test_Http2PriorityQueue -TESTS = test_Huffmancode +TESTS = \ + test_Huffmancode \ + test_Http2DependencyTree \ + test_Http2PriorityQueue test_Huffmancode_LDADD = \ $(top_builddir)/lib/ts/libtsutil.la @@ -66,3 +73,17 @@ test_Huffmancode_SOURCES = \ test_Huffmancode.cc \ HuffmanCodec.cc \ HuffmanCodec.h + +test_Http2DependencyTree_LDADD = \ + $(top_builddir)/lib/ts/libtsutil.la + +test_Http2DependencyTree_SOURCES = \ + test_Http2DependencyTree.cc \ + Http2DependencyTree.h + +test_Http2PriorityQueue_LDADD = \ + $(top_builddir)/lib/ts/libtsutil.la + +test_Http2PriorityQueue_SOURCES = \ + test_Http2PriorityQueue.cc \ + Http2PriorityQueue.h diff --git a/proxy/http2/test_Http2DependencyTree.cc b/proxy/http2/test_Http2DependencyTree.cc new file mode 100644 index 00000000000..8de1f5f7d35 --- /dev/null +++ b/proxy/http2/test_Http2DependencyTree.cc @@ -0,0 +1,367 @@ +/** @file + + Unit tests for Http2DependencyTree + + @section license License + + 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 "Http2DependencyTree.h" + +using namespace std; + + +typedef Http2DependencyTree Tree; + +/** + * Exclusive Dependency Creation + * + * A A + * / \ => | + * B C D + * / \ + * B C + */ +bool +exclusive_dependency_creaion() +{ + Tree *tree = new Tree(); + string a("A"), b("B"), c("C"), d("D"); + + tree->add(0, 1, 0, false, &b); + tree->add(0, 3, 0, false, &c); + + Tree::Node *node_a = tree->find(0); + Tree::Node *node_b = tree->find(1); + Tree::Node *node_c = tree->find(3); + + ink_assert(node_b->parent == node_a); + ink_assert(node_c->parent == node_a); + + // Add node with exclusive flag + tree->add(0, 5, 0, true, &d); + + Tree::Node *node_d = tree->find(5); + + ink_assert(node_d->parent == node_a); + ink_assert(node_b->parent == node_d); + ink_assert(node_c->parent == node_d); + + return true; +} + + +/** + * Reprioritization (non-exclusive) + * + * x x + * | | + * A D + * / \ / \ + * B C ==> F A + * / \ / \ + * D E B C + * | | + * F E + */ +bool +reprioritization() +{ + Tree *tree = new Tree(); + string a("A"), b("B"), c("C"), d("D"), e("E"), f("F"); + + tree->add(0, 1, 0, false, &a); + tree->add(1, 3, 0, false, &b); + tree->add(1, 5, 0, false, &c); + tree->add(5, 7, 0, false, &d); + tree->add(5, 9, 0, false, &e); + tree->add(7, 11, 0, false, &f); + + tree->reprioritize(1, 7, false); + + Tree::Node *node_x = tree->find(0); + Tree::Node *node_a = tree->find(1); + Tree::Node *node_d = tree->find(7); + Tree::Node *node_f = tree->find(11); + + ink_assert(node_a->parent == node_d); + ink_assert(node_d->parent == node_x); + ink_assert(node_f->parent == node_d); + + return true; +} + +/** + * Reprioritization (exclusive) + * + * x x + * | | + * A D + * / \ | + * B C ==> A + * / \ /|\ + * D E B C F + * | | + * F E + */ +bool +exclusive_reprioritization() +{ + Tree *tree = new Tree(); + string a("A"), b("B"), c("C"), d("D"), e("E"), f("F"); + + tree->add(0, 1, 0, false, &a); + tree->add(1, 3, 0, false, &b); + tree->add(1, 5, 0, false, &c); + tree->add(5, 7, 0, false, &d); + tree->add(5, 9, 0, false, &e); + tree->add(7, 11, 0, false, &f); + + tree->reprioritize(1, 7, true); + + Tree::Node *node_x = tree->find(0); + Tree::Node *node_a = tree->find(1); + Tree::Node *node_d = tree->find(7); + Tree::Node *node_f = tree->find(11); + + ink_assert(node_a->parent == node_d); + ink_assert(node_d->parent == node_x); + ink_assert(node_f->parent == node_a); + + return true; +} + + +/** + * Tree + * ROOT + * / + * A(1) + */ +bool +one_node() +{ + Tree *tree = new Tree(); + string a("A"); + tree->add(0, 1, 0, false, &a); + + Tree::Node *node_a = tree->find(1); + + ink_assert(tree->top() == NULL); + + tree->activate(node_a); + ink_assert(tree->top() == node_a); + + tree->deactivate(node_a, 0); + ink_assert(tree->top() == NULL); + + return true; +} + +/** + * Tree + * ROOT + * / + * A(3) + * / + * B(5) + * + */ +bool +active_intermediate_node() +{ + Tree *tree = new Tree(); + string a("A"), b("B"), c("C"); + + tree->add(0, 3, 15, false, &a); + tree->add(3, 5, 15, false, &b); + + Tree::Node *node_a = tree->find(3); + Tree::Node *node_b = tree->find(5); + + ink_assert(tree->top() == NULL); + + tree->activate(node_a); + tree->activate(node_b); + ink_assert(tree->top() == node_a); + + tree->deactivate(node_a, 0); + ink_assert(tree->top() == node_b); + + return true; +} + +/** + * Basic Tree + * ROOT + * / \ + * A(3) D(9) + * / \ + * B(5) C(7) + * + */ +bool +basic_tree() +{ + Tree *tree = new Tree(); + + string a("A"), b("B"), c("C"), d("D"); + + // NOTE, weight is actual weight - 1 + tree->add(0, 3, 0, false, &a); + tree->add(3, 5, 0, false, &b); + tree->add(3, 7, 1, false, &c); + tree->add(0, 9, 1, false, &d); + + Tree::Node *node_b = tree->find(5); + Tree::Node *node_c = tree->find(7); + Tree::Node *node_d = tree->find(9); + + // Activate B, C and D + tree->activate(node_b); + tree->activate(node_c); + tree->activate(node_d); + + ostringstream oss; + + for (int i = 0; i < 30; ++i) { + Tree::Node *node = tree->top(); + oss << node->t->c_str(); + tree->update(node, 100); + } + + bool result = false; + + // TODO: check strictly + if (oss.str() == "BDDDCDDCDDCDDBDDCDDCDDBDDCDDCD") { + result = true; + } else { + cerr << "ERR: " << oss.str() << endl; + result = false; + } + + delete tree; + + return result; +} + +/** + * Chrome's Tree + * ROOT + * / \ + * A(3) C(7) + * / \ + * B(5) D(9) + * + */ +bool +chrome_tree() +{ + Tree *tree = new Tree(); + + string a("A"), b("B"), c("C"), d("D"); + + tree->add(0, 3, 15, false, &a); + tree->add(3, 5, 15, true, &b); + tree->add(0, 7, 15, false, &c); + tree->add(7, 9, 15, true, &d); + + Tree::Node *node_a = tree->find(3); + Tree::Node *node_b = tree->find(5); + Tree::Node *node_c = tree->find(7); + Tree::Node *node_d = tree->find(9); + + // Activate A and B + tree->activate(node_a); + tree->activate(node_b); + tree->activate(node_c); + tree->activate(node_d); + + ostringstream oss; + + for (int i = 0; i < 30; ++i) { + Tree::Node *node = tree->top(); + oss << node->t->c_str(); + + if (i == 14 || i == 15) { + tree->deactivate(node, 100); + } else { + tree->update(node, 100); + } + } + + bool result = false; + + if (oss.str() == "ACCAACCAACCAACCABDDBBDDBBDDBBD") { + result = true; + } else { + cerr << "ERR: " << oss.str() << endl; + result = false; + } + + delete tree; + + return result; +} + +int +main() +{ + cout << "Exclusive Dependency Creation: "; + if (exclusive_dependency_creaion()) { + cout << "OK" << endl; + } + + cout << "Reprioritization: "; + if (reprioritization()) { + cout << "OK" << endl; + } + + cout << "Exclusive Reprioritization: "; + if (exclusive_reprioritization()) { + cout << "OK" << endl; + } + + cout << "One node: "; + if (one_node()) { + cout << "OK" << endl; + } + + cout << "Active Intermidiate Node: "; + if (active_intermediate_node()) { + cout << "OK" << endl; + } + + cout << "Basic Tree: "; + if (basic_tree()) { + cout << "OK" << endl; + } + + cout << "Chrome Tree: "; + if (chrome_tree()) { + cout << "OK" << endl; + } + + return 0; +} diff --git a/proxy/http2/test_Http2PriorityQueue.cc b/proxy/http2/test_Http2PriorityQueue.cc new file mode 100644 index 00000000000..710b4949f56 --- /dev/null +++ b/proxy/http2/test_Http2PriorityQueue.cc @@ -0,0 +1,248 @@ +/** @file + + Unit tests for Http2PriorityQueue + + @section license License + + 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 "Http2PriorityQueue.h" + +using namespace std; + +class N +{ +public: + N(uint32_t w, string c) : weight(w), content(c) {} + + bool operator<(const N &n) const { return weight < n.weight; } + bool operator>(const N &n) const { return weight > n.weight; } + + uint32_t weight; + string content; +}; + +typedef Http2PriorityQueueEntry Entry; +typedef Http2PriorityQueue PQ; + +// For debug +void +dump(PQ *pq) +{ + Vec v = pq->dump(); + + for (uint32_t i = 0; i < v.length(); i++) { + std::cout << v[i]->index << "," << v[i]->node->weight << "," << v[i]->node->content << std::endl; + } + std::cout << "--------" << std::endl; +} + +// Push, top, and pop a entry +void +test_pq_scenario_1() +{ + PQ *pq = new PQ(); + N *a = new N(6, "A"); + Entry *entry_a = new Entry(a); + + pq->push(entry_a); + ink_assert(pq->top() == entry_a); + + pq->pop(); + ink_assert(pq->top() == NULL); +} + +// Update weight +void +test_pq_scenario_2() +{ + PQ *pq = new PQ(); + + N *a = new N(10, "A"); + N *b = new N(20, "B"); + + Entry *entry_a = new Entry(a); + Entry *entry_b = new Entry(b); + + pq->push(entry_a); + pq->push(entry_b); + + ink_assert(pq->top() == entry_a); + + a->weight = 30; + pq->update(entry_a); + + ink_assert(pq->top() == entry_b); +} + +// Push, top, and pop 9 entries +void +test_pq_scenario_3() +{ + PQ *pq = new PQ(); + + ink_assert(pq->empty()); + ink_assert(pq->top() == NULL); + + N *a = new N(6, "A"); + N *b = new N(1, "B"); + N *c = new N(9, "C"); + N *d = new N(8, "D"); + N *e = new N(4, "E"); + N *f = new N(3, "F"); + N *g = new N(2, "G"); + N *h = new N(7, "H"); + N *i = new N(5, "I"); + + Entry *entry_a = new Entry(a); + Entry *entry_b = new Entry(b); + Entry *entry_c = new Entry(c); + Entry *entry_d = new Entry(d); + Entry *entry_e = new Entry(e); + Entry *entry_f = new Entry(f); + Entry *entry_g = new Entry(g); + Entry *entry_h = new Entry(h); + Entry *entry_i = new Entry(i); + + pq->push(entry_a); + pq->push(entry_b); + pq->push(entry_c); + pq->push(entry_d); + pq->push(entry_e); + pq->push(entry_f); + pq->push(entry_g); + pq->push(entry_h); + pq->push(entry_i); + + ink_assert(pq->top() == entry_b); // 1 + pq->pop(); + ink_assert(pq->top() == entry_g); // 2 + pq->pop(); + ink_assert(pq->top() == entry_f); // 3 + pq->pop(); + ink_assert(pq->top() == entry_e); // 4 + pq->pop(); + ink_assert(pq->top() == entry_i); // 5 + pq->pop(); + ink_assert(pq->top() == entry_a); // 6 + pq->pop(); + ink_assert(pq->top() == entry_h); // 7 + pq->pop(); + ink_assert(pq->top() == entry_d); // 8 + pq->pop(); + ink_assert(pq->top() == entry_c); // 9 + pq->pop(); + + ink_assert(pq->top() == NULL); +} + +// Push, top, pop, and update 9 entries +void +test_pq_scenario_4() +{ + PQ *pq = new PQ(); + + ink_assert(pq->empty()); + ink_assert(pq->top() == NULL); + + N *a = new N(6, "A"); + N *b = new N(1, "B"); + N *c = new N(9, "C"); + N *d = new N(8, "D"); + N *e = new N(4, "E"); + N *f = new N(3, "F"); + N *g = new N(2, "G"); + N *h = new N(7, "H"); + N *i = new N(5, "I"); + + Entry *entry_a = new Entry(a); + Entry *entry_b = new Entry(b); + Entry *entry_c = new Entry(c); + Entry *entry_d = new Entry(d); + Entry *entry_e = new Entry(e); + Entry *entry_f = new Entry(f); + Entry *entry_g = new Entry(g); + Entry *entry_h = new Entry(h); + Entry *entry_i = new Entry(i); + + pq->push(entry_a); + pq->push(entry_b); + pq->push(entry_c); + pq->push(entry_d); + pq->push(entry_e); + pq->push(entry_f); + pq->push(entry_g); + pq->push(entry_h); + pq->push(entry_i); + dump(pq); + + // Pop head and push it back again + ink_assert(pq->top() == entry_b); // 1 + pq->pop(); + b->weight += 100; + pq->push(entry_b); + // Update weight + a->weight += 100; + pq->update(entry_a); + c->weight += 100; + pq->update(entry_d); + e->weight += 100; + pq->update(entry_e); + g->weight += 100; + pq->update(entry_g); + dump(pq); + + // Check + ink_assert(pq->top() == entry_f); // 3 + pq->pop(); + ink_assert(pq->top() == entry_i); // 5 + pq->pop(); + ink_assert(pq->top() == entry_h); // 7 + pq->pop(); + ink_assert(pq->top() == entry_d); // 8 + pq->pop(); + ink_assert(pq->top() == entry_b); // 101 + pq->pop(); + ink_assert(pq->top() == entry_g); // 102 + pq->pop(); + ink_assert(pq->top() == entry_e); // 104 + pq->pop(); + ink_assert(pq->top() == entry_a); // 106 + pq->pop(); + ink_assert(pq->top() == entry_c); // 109 + pq->pop(); + + ink_assert(pq->top() == NULL); +} + +int +main() +{ + test_pq_scenario_1(); + test_pq_scenario_2(); + test_pq_scenario_3(); + test_pq_scenario_4(); + + return 0; +}