Control rate of HTTP/1 writes (#180)

* HTTP/1 only writes 1 message at a time.
This avoids bloating memory with unsent messages.
This also helps improve the accuracy of our upload stats.
Still needs tweaking to deal with websocket upgrade edge-cases.

* Install websocket handler the moment 101 response comes in, do not wait for stream to complete.
Remove hacks where 101 response completed a stream, no 1xx response should complete a stream.

* Simplify confusing list logic.
Remove the server-only `waiting_stream_list`.
Do a simple scan over available streams to find the next one to work on. This is O(N) when pipelining, but I hear that no one does HTTP/1 pipelining, so in that case O(N=1). We can always re-optimize if we have data that pipelining is super popular and this is a hotspot.

Author: graebm

source/h1_connection.c

@@ -636,8 +636,7 @@ static int s_linestate_request(struct aws_h1_decoder *decoder, struct aws_byte_c
 }
 
 static bool s_check_info_response_status_code(int code_val) {
-    /* TODO: 101 is an info_response, we need to revise the 101 behaviour. */
-    return code_val >= 100 && code_val < 200 && code_val != 101;
+    return code_val >= 100 && code_val < 200;
 }
 
 static int s_linestate_response(struct aws_h1_decoder *decoder, struct aws_byte_cursor input) {

source/h1_decoder.c

@@ -99,7 +99,11 @@ static int s_ws_bootstrap_on_handshake_response_headers(
     const struct aws_http_header *header_array,
     size_t num_headers,
     void *user_data);
-static void s_ws_bootstrap_on_handshake_complete(struct aws_http_stream *stream, int error_code, void *user_data);
+static int s_ws_bootstrap_on_handshake_response_header_block_done(
+    struct aws_http_stream *stream,
+    enum aws_http_header_block header_block,
+    void *user_data);
+static void s_ws_bootstrap_on_stream_complete(struct aws_http_stream *stream, int error_code, void *user_data);
 
 int aws_websocket_client_connect(const struct aws_websocket_client_connection_options *options) {
     aws_http_fatal_assert_library_initialized();
@@ -307,7 +311,8 @@ static void s_ws_bootstrap_on_http_setup(struct aws_http_connection *http_connec
         .request = ws_bootstrap->handshake_request,
         .user_data = ws_bootstrap,
         .on_response_headers = s_ws_bootstrap_on_handshake_response_headers,
-        .on_complete = s_ws_bootstrap_on_handshake_complete,
+        .on_response_header_block_done = s_ws_bootstrap_on_handshake_response_header_block_done,
+        .on_complete = s_ws_bootstrap_on_stream_complete,
     };
 
     struct aws_http_stream *handshake_stream =
@@ -393,6 +398,7 @@ static void s_ws_bootstrap_on_http_shutdown(
     s_ws_bootstrap_destroy(ws_bootstrap);
 }
 
+/* Invoked repeatedly as handshake response headers arrive */
 static int s_ws_bootstrap_on_handshake_response_headers(
     struct aws_http_stream *stream,
     enum aws_http_header_block header_block,
@@ -444,15 +450,25 @@ static int s_ws_bootstrap_on_handshake_response_headers(
     return AWS_OP_ERR;
 }
 
-static void s_ws_bootstrap_on_handshake_complete(struct aws_http_stream *stream, int error_code, void *user_data) {
+/**
+ * Invoked each time we reach the end of a block of response headers.
+ * If we got a valid 101 Switching Protocols response, we insert the websocket handler.
+ * Note:
+ *      In HTTP, 1xx responses are "interim" responses. So a 101 Switching Protocols
+ *      response does not "complete" the stream. Once the connection has switched
+ *      protocols, the stream does not end until the whole connection is closed.
+ */
+static int s_ws_bootstrap_on_handshake_response_header_block_done(
+    struct aws_http_stream *stream,
+    enum aws_http_header_block header_block,
+    void *user_data) {
+
+    (void)header_block;
+
     struct aws_websocket_client_bootstrap *ws_bootstrap = user_data;
     struct aws_http_connection *http_connection = s_system_vtable->aws_http_stream_get_connection(stream);
     AWS_ASSERT(http_connection);
 
-    if (error_code) {
-        goto error;
-    }
-
     /* Get data from stream */
     s_system_vtable->aws_http_stream_get_incoming_response_status(stream, &ws_bootstrap->response_status);
 
@@ -522,13 +538,20 @@ static void s_ws_bootstrap_on_handshake_complete(struct aws_http_stream *stream,
     /* Clear setup callback so that we know that it's been invoked. */
     ws_bootstrap->websocket_setup_callback = NULL;
 
-    s_system_vtable->aws_http_stream_release(stream);
-    return;
+    return AWS_OP_SUCCESS;
 
 error:
-    if (!error_code) {
-        error_code = aws_last_error();
-    }
-    s_ws_bootstrap_cancel_setup_due_to_err(ws_bootstrap, http_connection, error_code);
+    s_ws_bootstrap_cancel_setup_due_to_err(ws_bootstrap, http_connection, aws_last_error());
+    /* Returning error stops HTTP from processing any further data */
+    return AWS_OP_ERR;
+}
+
+static void s_ws_bootstrap_on_stream_complete(struct aws_http_stream *stream, int error_code, void *user_data) {
+    /* Not checking error_code because a stream error ends the HTTP connection.
+     * We'll deal with finishing setup and/or shutdown from the http-shutdown callback */
+    (void)error_code;
+    (void)user_data;
+
+    /* Done with stream, let it be cleaned up */
     s_system_vtable->aws_http_stream_release(stream);
 }

source/websocket_bootstrap.c

@@ -50,7 +50,7 @@ add_test_case(h1_client_request_send_large_head)
 add_test_case(h1_client_request_content_length_0_ok)
 add_test_case(h1_client_request_content_length_too_small_is_error)
 add_test_case(h1_client_request_content_length_too_large_is_error)
-add_test_case(h1_client_request_send_multiple_in_1_io_message)
+add_test_case(h1_client_request_send_multiple)
 add_test_case(h1_client_request_close_header_ends_connection)
 add_test_case(h1_client_request_close_header_with_pipelining)
 add_test_case(h1_client_response_get_1liner)
@@ -162,7 +162,8 @@ add_test_case(websocket_boot_golden_path)
 add_test_case(websocket_boot_fail_at_http_connect)
 add_test_case(websocket_boot_fail_at_http_connect_error)
 add_test_case(websocket_boot_fail_at_new_request)
-add_test_case(websocket_boot_fail_at_response_error)
+add_test_case(websocket_boot_fail_before_response_headers)
+add_test_case(websocket_boot_fail_before_response_headers_done)
 add_test_case(websocket_boot_fail_at_response_status)
 add_test_case(websocket_boot_fail_at_new_handler)
 add_test_case(websocket_boot_report_unexpected_http_shutdown)

tests/CMakeLists.txt

@@ -315,7 +315,7 @@ int proxy_tester_send_connect_response(struct proxy_tester *tester) {
     }
 
     /* send response */
-    ASSERT_SUCCESS(testing_channel_send_response_str(tester->testing_channel, response_string));
+    ASSERT_SUCCESS(testing_channel_push_read_str(tester->testing_channel, response_string));
 
     testing_channel_drain_queued_tasks(tester->testing_channel);
 

tests/proxy_test_helper.c

@@ -1109,7 +1109,7 @@ static void s_add_outgoing_stream(struct test_http_stats_event *event) {
 }
 
 static void s_add_response_data(struct test_http_stats_event *event) {
-    testing_channel_send_response_str(&s_test_context.test_channel, event->response_stream_data);
+    testing_channel_push_read_str(&s_test_context.test_channel, event->response_stream_data);
 }
 
 static int s_do_http_statistics_test(