Streams now have to be activated. This works around a race condition … (#200)

* Streams now have to be activated. This works around a race condition where the request kicks off before the stream container itself has had a chance to be seated by the user.

* Reworked backpressure to be a connection-wide setting. Removed the request level option altogether. Added tests to verify unactivated streams still properly clean-up.

Author: JonathanHenson

bin/elasticurl/main.c

@@ -504,6 +504,7 @@ static void s_on_signing_complete(struct aws_http_message *request, int error_co
         fprintf(stderr, "failed to create request.");
         exit(1);
     }
+    aws_http_stream_activate(stream);
 
     /* Connection will stay alive until stream completes */
     aws_http_connection_release(app_ctx->connection);

include/aws/http/connection.h

@@ -210,9 +210,15 @@ struct aws_http_client_connection_options {
     aws_http_on_client_connection_shutdown_fn *on_shutdown;
 
     /**
-     * If set to true, read back pressure mechanism will be enabled.
-     */
-    bool enable_read_back_pressure;
+     * Set to true to manually manage the read window size.
+     *
+     * If this is false, the connection will maintain a constant window size.
+     *
+     * If this is true, the caller must manually increment the window size using aws_http_stream_update_window().
+     * If the window is not incremented, it will shrink by the amount of body data received. If the window size
+     * reaches 0, no further data will be received.
+     **/
+    bool manual_window_management;
 };
 
 /**

include/aws/http/private/connection_impl.h

@@ -76,7 +76,7 @@ struct aws_http_connection {
     struct aws_atomic_var refcount;
 
     /* Starts at either 1 or 2, increments by two with each new stream */
-    struct aws_atomic_var next_stream_id;
+    uint32_t next_stream_id;
 
     union {
         struct aws_http_connection_client_data {
@@ -93,13 +93,16 @@ struct aws_http_connection {
      * Opposite is true on server connections */
     struct aws_http_connection_client_data *client_data;
     struct aws_http_connection_server_data *server_data;
+
+    bool manual_window_management;
 };
 
 /* Gets a client connection up and running.
  * Responsible for firing on_setup and on_shutdown callbacks. */
 struct aws_http_client_bootstrap {
     struct aws_allocator *alloc;
     bool is_using_tls;
+    bool manual_window_management;
     size_t initial_window_size;
     struct aws_http_connection_monitoring_options monitoring_options;
     void *user_data;
@@ -135,6 +138,8 @@ struct aws_crt_statistics_http1_channel *aws_h1_connection_get_statistics(struct
 /**
  * Gets the next available stream id within the connection.  Valid for creating both h1 and h2 streams.
  *
+ * This function is not thread-safe.
+ *
  * Returns 0 if there was an error.
  */
 AWS_HTTP_API

include/aws/http/private/h1_connection.h

@@ -23,11 +23,13 @@ AWS_EXTERN_C_BEGIN
 AWS_HTTP_API
 struct aws_http_connection *aws_http_connection_new_http1_1_server(
     struct aws_allocator *allocator,
+    bool manual_window_management,
     size_t initial_window_size);
 
 AWS_HTTP_API
 struct aws_http_connection *aws_http_connection_new_http1_1_client(
     struct aws_allocator *allocator,
+    bool manual_window_management,
     size_t initial_window_size);
 
 AWS_EXTERN_C_END

include/aws/http/private/h1_stream.h

@@ -58,4 +58,8 @@ struct aws_h1_stream *aws_h1_stream_new_request_handler(const struct aws_http_re
 
 AWS_EXTERN_C_END
 
+/* we don't want this exported. We just want it to have external linkage between h1_stream and h1_connection compilation
+ * units. it is defined in h1_connection.c */
+int aws_h1_stream_activate(struct aws_http_stream *stream);
+
 #endif /* AWS_HTTP_H1_STREAM_H */

include/aws/http/private/h2_connection.h

@@ -84,11 +84,13 @@ AWS_EXTERN_C_BEGIN
 AWS_HTTP_API
 struct aws_http_connection *aws_http_connection_new_http2_server(
     struct aws_allocator *allocator,
+    bool manual_window_management,
     size_t initial_window_size);
 
 AWS_HTTP_API
 struct aws_http_connection *aws_http_connection_new_http2_client(
     struct aws_allocator *allocator,
+    bool manual_window_management,
     size_t initial_window_size);
 
 AWS_EXTERN_C_END

include/aws/http/private/h2_stream.h

@@ -75,4 +75,6 @@ enum aws_h2_stream_state aws_h2_stream_get_state(const struct aws_h2_stream *str
 /* Connection is ready to send frames from stream now */
 int aws_h2_stream_on_activated(struct aws_h2_stream *stream, bool *out_has_outgoing_data);
 
+int aws_h2_stream_activate(struct aws_http_stream *stream);
+
 #endif /* AWS_HTTP_H2_STREAM_H */

include/aws/http/private/request_response_impl.h

@@ -25,6 +25,7 @@
 struct aws_http_stream_vtable {
     void (*destroy)(struct aws_http_stream *stream);
     void (*update_window)(struct aws_http_stream *stream, size_t increment_size);
+    int (*activate)(struct aws_http_stream *stream);
 };
 
 /**

include/aws/http/private/websocket_impl.h

@@ -85,6 +85,7 @@ struct aws_websocket_client_bootstrap_system_vtable {
     struct aws_http_stream *(*aws_http_connection_make_request)(
         struct aws_http_connection *client_connection,
         const struct aws_http_make_request_options *options);
+    int (*aws_http_stream_activate)(struct aws_http_stream *stream);
     void (*aws_http_stream_release)(struct aws_http_stream *stream);
     struct aws_http_connection *(*aws_http_stream_get_connection)(const struct aws_http_stream *stream);
     int (*aws_http_stream_get_incoming_response_status)(const struct aws_http_stream *stream, int *out_status);

include/aws/http/request_response.h

@@ -171,7 +171,7 @@ typedef int(aws_http_on_incoming_header_block_done_fn)(
  * The data must be copied immediately if you wish to preserve it.
  * This is always invoked on the HTTP connection's event-loop thread.
  *
- * Note that, if the stream is using manual_window_management then the window
+ * Note that, if the connection is using manual_window_management then the window
  * size has shrunk by the amount of body data received. If the window size
  * reaches 0 no further data will be received. Increment the window size with
  * aws_http_stream_update_window().
@@ -243,17 +243,6 @@ struct aws_http_make_request_options {
      * See `aws_http_on_stream_complete_fn`.
      */
     aws_http_on_stream_complete_fn *on_complete;
-
-    /**
-     * Set to true to manually manage the read window size.
-     *
-     * If this is false, the connection will maintain a constant window size.
-     *
-     * If this is true, the caller must manually increment the window size using aws_http_stream_update_window().
-     * If the window is not incremented, it will shrink by the amount of body data received. If the window size
-     * reaches 0, no further data will be received.
-     */
-    bool manual_window_management;
 };
 
 struct aws_http_request_handler_options {
@@ -305,17 +294,6 @@ struct aws_http_request_handler_options {
      * See `aws_http_on_stream_complete_fn`.
      */
     aws_http_on_stream_complete_fn *on_complete;
-
-    /**
-     * Set to true to manually manage the read window size.
-     *
-     * If this is false, the connection will maintain a constant window size.
-     *
-     * If this is true, the caller must manually increment the window size using aws_http_stream_update_window().
-     * If the window is not incremented, it will shrink by the amount of body data received. If the window size
-     * reaches 0, no further data will be received.
-     */
-    bool manual_window_management;
 };
 
 #define AWS_HTTP_REQUEST_HANDLER_OPTIONS_INIT                                                                          \
@@ -634,7 +612,9 @@ int aws_http_message_erase_header(struct aws_http_message *message, size_t index
 
 /**
  * Create a stream, with a client connection sending a request.
- * The request starts sending automatically once the stream is created.
+ * The request does not start sending automatically once the stream is created. You must call
+ * aws_http_stream_activate to begin execution of the request.
+ *
  * The `options` are copied during this call.
  *
  * Tip for language bindings: Do not bind the `options` struct. Use something more natural for your language,
@@ -664,6 +644,13 @@ struct aws_http_stream *aws_http_stream_new_server_request_handler(
 AWS_HTTP_API
 void aws_http_stream_release(struct aws_http_stream *stream);
 
+/**
+ * Only used for client initiated streams (immediately following a call to aws_http_connection_make_request).
+ *
+ * Activates the request's outgoing stream processing.
+ */
+AWS_HTTP_API int aws_http_stream_activate(struct aws_http_stream *stream);
+
 AWS_HTTP_API
 struct aws_http_connection *aws_http_stream_get_connection(const struct aws_http_stream *stream);
 
@@ -697,7 +684,8 @@ void aws_http_stream_update_window(struct aws_http_stream *stream, size_t increm
 
 /**
  * Gets the Http/2 id associated with a stream.  Even h1 streams have an id (using the same allocation procedure
- * as http/2) for easier tracking purposes.
+ * as http/2) for easier tracking purposes. For client streams, this will only be non-zero after a successful call
+ * to aws_http_stream_activate()
  */
 uint32_t aws_http_stream_get_id(struct aws_http_stream *stream);
 

include/aws/http/server.h

@@ -105,9 +105,15 @@ struct aws_http_server_options {
     aws_http_server_on_destroy_fn *on_destroy_complete;
 
     /**
-     * If set to true, read back pressure mechanism will be enabled.
-     */
-    bool enable_read_back_pressure;
+     * Set to true to manually manage the read window size.
+     *
+     * If this is false, the connection will maintain a constant window size.
+     *
+     * If this is true, the caller must manually increment the window size using aws_http_stream_update_window().
+     * If the window is not incremented, it will shrink by the amount of body data received. If the window size
+     * reaches 0, no further data will be received.
+     **/
+    bool manual_window_management;
 };
 
 /**

source/connection.c

@@ -51,6 +51,7 @@ struct aws_http_server {
     struct aws_allocator *alloc;
     struct aws_server_bootstrap *bootstrap;
     bool is_using_tls;
+    bool manual_window_management;
     size_t initial_window_size;
     void *user_data;
     aws_http_server_on_incoming_connection_fn *on_incoming_connection;
@@ -83,6 +84,7 @@ static struct aws_http_connection *s_connection_new(
     struct aws_channel *channel,
     bool is_server,
     bool is_using_tls,
+    bool manual_window_management,
     size_t initial_window_size) {
 
     struct aws_channel_slot *connection_slot = NULL;
@@ -146,17 +148,19 @@ static struct aws_http_connection *s_connection_new(
     switch (version) {
         case AWS_HTTP_VERSION_1_1:
             if (is_server) {
-                connection = aws_http_connection_new_http1_1_server(alloc, initial_window_size);
+                connection =
+                    aws_http_connection_new_http1_1_server(alloc, manual_window_management, initial_window_size);
             } else {
-                connection = aws_http_connection_new_http1_1_client(alloc, initial_window_size);
+                connection =
+                    aws_http_connection_new_http1_1_client(alloc, manual_window_management, initial_window_size);
             }
             break;
         case AWS_HTTP_VERSION_2:
             AWS_FATAL_ASSERT(false && "H2 is not currently supported"); /* lol nice try */
             if (is_server) {
-                connection = aws_http_connection_new_http2_server(alloc, initial_window_size);
+                connection = aws_http_connection_new_http2_server(alloc, manual_window_management, initial_window_size);
             } else {
-                connection = aws_http_connection_new_http2_client(alloc, initial_window_size);
+                connection = aws_http_connection_new_http2_client(alloc, manual_window_management, initial_window_size);
             }
             break;
         default:
@@ -293,7 +297,13 @@ static void s_server_bootstrap_on_accept_channel_setup(
         goto error;
     }
     /* Create connection */
-    connection = s_connection_new(server->alloc, channel, true, server->is_using_tls, server->initial_window_size);
+    connection = s_connection_new(
+        server->alloc,
+        channel,
+        true,
+        server->is_using_tls,
+        server->manual_window_management,
+        server->initial_window_size);
     if (!connection) {
         AWS_LOGF_ERROR(
             AWS_LS_HTTP_SERVER,
@@ -466,6 +476,7 @@ struct aws_http_server *aws_http_server_new(const struct aws_http_server_options
     server->user_data = options->server_user_data;
     server->on_incoming_connection = options->on_incoming_connection;
     server->on_destroy_complete = options->on_destroy_complete;
+    server->manual_window_management = options->manual_window_management;
 
     int err = aws_mutex_init(&server->synced_data.lock);
     if (err) {
@@ -490,7 +501,7 @@ struct aws_http_server *aws_http_server_new(const struct aws_http_server_options
     }
 
     struct aws_server_socket_channel_bootstrap_options bootstrap_options = {
-        .enable_read_back_pressure = options->enable_read_back_pressure,
+        .enable_read_back_pressure = options->manual_window_management,
         .tls_options = options->tls_options,
         .bootstrap = options->bootstrap,
         .socket_options = options->socket_options,
@@ -613,7 +624,12 @@ static void s_client_bootstrap_on_channel_setup(
     AWS_LOGF_TRACE(AWS_LS_HTTP_CONNECTION, "static: Socket connected, creating client connection object.");
 
     http_bootstrap->connection = s_connection_new(
-        http_bootstrap->alloc, channel, false, http_bootstrap->is_using_tls, http_bootstrap->initial_window_size);
+        http_bootstrap->alloc,
+        channel,
+        false,
+        http_bootstrap->is_using_tls,
+        http_bootstrap->manual_window_management,
+        http_bootstrap->initial_window_size);
     if (!http_bootstrap->connection) {
         AWS_LOGF_ERROR(
             AWS_LS_HTTP_CONNECTION,
@@ -743,6 +759,7 @@ int aws_http_client_connect_internal(
 
     http_bootstrap->alloc = options->allocator;
     http_bootstrap->is_using_tls = options->tls_options != NULL;
+    http_bootstrap->manual_window_management = options->manual_window_management;
     http_bootstrap->initial_window_size = options->initial_window_size;
     http_bootstrap->user_data = options->user_data;
     http_bootstrap->on_setup = options->on_setup;
@@ -766,7 +783,7 @@ int aws_http_client_connect_internal(
         .tls_options = options->tls_options,
         .setup_callback = s_client_bootstrap_on_channel_setup,
         .shutdown_callback = s_client_bootstrap_on_channel_shutdown,
-        .enable_read_back_pressure = options->enable_read_back_pressure,
+        .enable_read_back_pressure = options->manual_window_management,
         .user_data = http_bootstrap,
     };
 
@@ -845,13 +862,15 @@ static const uint32_t MAX_STREAM_ID = UINT32_MAX >> 1;
 
 uint32_t aws_http_connection_get_next_stream_id(struct aws_http_connection *connection) {
 
-    uint32_t next_id = (uint32_t)aws_atomic_fetch_add(&connection->next_stream_id, 2);
-    /* If next fetch would overflow next_stream_id, set it to 0 */
+    uint32_t next_id = connection->next_stream_id;
+
     if (AWS_UNLIKELY(next_id > MAX_STREAM_ID)) {
         AWS_LOGF_INFO(AWS_LS_HTTP_CONNECTION, "id=%p: All available stream ids are gone", (void *)connection);
 
         next_id = 0;
         aws_raise_error(AWS_ERROR_HTTP_STREAM_IDS_EXHAUSTED);
+    } else {
+        connection->next_stream_id += 2;
     }
 
     return next_id;

source/connection_manager.c

@@ -691,7 +691,7 @@ static int s_aws_http_connection_manager_new_connection(struct aws_http_connecti
     options.socket_options = &manager->socket_options;
     options.on_setup = s_aws_http_connection_manager_on_connection_setup;
     options.on_shutdown = s_aws_http_connection_manager_on_connection_shutdown;
-    options.enable_read_back_pressure = manager->enable_read_back_pressure;
+    options.manual_window_management = manager->enable_read_back_pressure;
 
     if (aws_http_connection_monitoring_options_is_valid(&manager->monitoring_options)) {
         options.monitoring_options = &manager->monitoring_options;