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);