HTTP/2 stream manager (#356)

Author: TingDaoK

include/aws/http/connection.h

@@ -567,7 +567,7 @@ void aws_http2_connection_get_local_settings(
  * Get the settings received from remote peer, which we are using to restricts the message to send.
  *
  * @param http2_connection HTTP/2 connection.
- * @param out_settings fixed size array of aws_http2_setting gets set to  the remote settings
+ * @param out_settings fixed size array of aws_http2_setting gets set to the remote settings
  */
 AWS_HTTP_API
 void aws_http2_connection_get_remote_settings(

include/aws/http/connection_manager.h

@@ -16,6 +16,7 @@ struct aws_http_connection_manager;
 struct aws_socket_options;
 struct aws_tls_connection_options;
 struct proxy_env_var_settings;
+struct aws_http2_setting;
 
 typedef void(aws_http_connection_manager_on_connection_setup_fn)(
     struct aws_http_connection *connection,

include/aws/http/http.h

@@ -51,6 +51,9 @@ enum aws_http_errors {
     AWS_ERROR_HTTP_PROXY_CONNECT_FAILED_RETRYABLE,
     AWS_ERROR_HTTP_PROTOCOL_SWITCH_FAILURE,
     AWS_ERROR_HTTP_MAX_CONCURRENT_STREAMS_EXCEEDED,
+    AWS_ERROR_HTTP_STREAM_MANAGER_SHUTTING_DOWN,
+    AWS_ERROR_HTTP_STREAM_MANAGER_CONNECTION_ACQUIRE_FAILURE,
+    AWS_ERROR_HTTP_STREAM_MANAGER_UNEXPECTED_HTTP_VERSION,
 
     AWS_ERROR_HTTP_END_RANGE = AWS_ERROR_ENUM_END_RANGE(AWS_C_HTTP_PACKAGE_ID)
 };
@@ -82,6 +85,7 @@ enum aws_http_log_subject {
     AWS_LS_HTTP_SERVER,
     AWS_LS_HTTP_STREAM,
     AWS_LS_HTTP_CONNECTION_MANAGER,
+    AWS_LS_HTTP_STREAM_MANAGER,
     AWS_LS_HTTP_WEBSOCKET,
     AWS_LS_HTTP_WEBSOCKET_SETUP,
     AWS_LS_HTTP_PROXY_NEGOTIATION,

include/aws/http/http2_stream_manager.h

@@ -0,0 +1,164 @@
+#ifndef AWS_HTTP2_STREAM_MANAGER_H
+#define AWS_HTTP2_STREAM_MANAGER_H
+
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <aws/http/http.h>
+
+struct aws_http2_stream_manager;
+struct aws_client_bootstrap;
+struct aws_http_connection;
+struct aws_http_connection_manager;
+struct aws_socket_options;
+struct aws_tls_connection_options;
+struct proxy_env_var_settings;
+struct aws_http2_setting;
+struct aws_http_make_request_options;
+struct aws_http_stream;
+
+/**
+ * Always invoked asynchronously when the stream was created, successfully or not.
+ * When stream is NULL, error code will be set to indicate what happened.
+ * If there is a stream returned, you own the stream completely.
+ * Invoked on the same thread as other callback of the stream, which will be the thread of the connection, ideally.
+ * If there is no connection made, the callback will be invoked from a sperate thread.
+ */
+typedef void(
+    aws_http2_stream_manager_on_stream_acquired_fn)(struct aws_http_stream *stream, int error_code, void *user_data);
+
+/**
+ * Invoked asynchronously when the stream manager has been shutdown completely.
+ * Never invoked when `aws_http2_stream_manager_new` failed.
+ */
+typedef void(aws_http2_stream_manager_shutdown_complete_fn)(void *user_data);
+
+/**
+ * HTTP/2 stream manager configuration struct.
+ *
+ * Contains all of the configuration needed to create an http2 connection as well as
+ * connection manager under the hood.
+ */
+struct aws_http2_stream_manager_options {
+    /**
+     * basic http connection configuration
+     */
+    struct aws_client_bootstrap *bootstrap;
+    const struct aws_socket_options *socket_options;
+    /**
+     * If TLS options is set, you also need to handle ALPN, otherwise, may not able to get HTTP/2 connection and fail
+     * the stream manager.
+     * If TLS options not set, prior knowledge will be used.
+     */
+    const struct aws_tls_connection_options *tls_connection_options;
+    struct aws_byte_cursor host;
+    uint16_t port;
+
+    /**
+     * HTTP/2 Stream window control.
+     * If set to true, the read back pressure mechanism will be enabled for streams created.
+     * The initial window size can be set through `initial window size`
+     */
+    bool enable_read_back_pressure;
+    /**
+     * Optional.
+     * If set, it will be sent to the peer as the `AWS_HTTP2_SETTINGS_INITIAL_WINDOW_SIZE` in the initial settings for
+     * HTTP/2 connection.
+     * If not set, the default will be used, which is 65,535 (2^16-1)(RFC-7540 6.5.2)
+     * Ignored if enable_read_back_pressure is false.
+     */
+    uint32_t initial_window_size;
+
+    /* Connection monitor for the underlying connections made */
+    const struct aws_http_connection_monitoring_options *monitoring_options;
+
+    /* Optional. Proxy configuration for underlying http connection */
+    const struct aws_http_proxy_options *proxy_options;
+    const struct proxy_env_var_settings *proxy_ev_settings;
+
+    /**
+     * Required.
+     * When the stream manager finishes deleting all the resources, the callback will be invoked.
+     */
+    void *shutdown_complete_user_data;
+    aws_http2_stream_manager_shutdown_complete_fn *shutdown_complete_callback;
+
+    /* TODO: More flexible policy about the connections, but will always has these three values below. */
+    /**
+     * Optional.
+     * 0 will be considered as using a default value.
+     * The ideal number of concurrent streams for a connection. Stream manager will try to create a new connection if
+     * one connection reaches this number. But, if the max connections reaches, manager will reuse connections to create
+     * the acquired steams as much as possible. */
+    size_t ideal_concurrent_streams_per_connection;
+    /**
+     * Optional.
+     * Default is no limit, which will use the limit from the server. 0 will be considered as using the default value.
+     * The real number of concurrent streams per connection will be controlled by the minmal value of the setting from
+     * other end and the value here.
+     */
+    size_t max_concurrent_streams_per_connection;
+    /**
+     * Required.
+     * The max number of connections will be open at same time. If all the connections are full, manager will wait until
+     * available to vender more streams */
+    size_t max_connections;
+};
+
+struct aws_http2_stream_manager_acquire_stream_options {
+    /**
+     * Required.
+     * Invoked when the stream finishes acquiring by stream manager.
+     */
+    aws_http2_stream_manager_on_stream_acquired_fn *callback;
+    /**
+     * Optional.
+     * User data for the callback.
+     */
+    void *user_data;
+    /* Required. see `aws_http_make_request_options` */
+    const struct aws_http_make_request_options *options;
+};
+
+AWS_EXTERN_C_BEGIN
+
+/**
+ * Acquire a refcount from the stream manager, stream manager will start to destroy after the refcount drops to zero.
+ * NULL is acceptable. Initial refcount after new is 1.
+ *
+ * @param manager
+ * @return The same pointer acquiring.
+ */
+AWS_HTTP_API
+struct aws_http2_stream_manager *aws_http2_stream_manager_acquire(struct aws_http2_stream_manager *manager);
+
+/**
+ * Release a refcount from the stream manager, stream manager will start to destroy after the refcount drops to zero.
+ * NULL is acceptable. Initial refcount after new is 1.
+ *
+ * @param manager
+ * @return NULL
+ */
+AWS_HTTP_API
+struct aws_http2_stream_manager *aws_http2_stream_manager_release(struct aws_http2_stream_manager *manager);
+
+AWS_HTTP_API
+struct aws_http2_stream_manager *aws_http2_stream_manager_new(
+    struct aws_allocator *allocator,
+    struct aws_http2_stream_manager_options *options);
+
+/**
+ * Acquire a stream from stream manager asynchronously.
+ *
+ * @param http2_stream_manager
+ * @param acquire_stream_option see `aws_http2_stream_manager_acquire_stream_options`
+ */
+AWS_HTTP_API
+void aws_http2_stream_manager_acquire_stream(
+    struct aws_http2_stream_manager *http2_stream_manager,
+    const struct aws_http2_stream_manager_acquire_stream_options *acquire_stream_option);
+
+AWS_EXTERN_C_END
+#endif /* AWS_HTTP2_STREAM_MANAGER_H */

include/aws/http/private/http2_stream_manager_impl.h

@@ -0,0 +1,182 @@
+#ifndef AWS_HTTP2_STREAM_MANAGER_IMPL_H
+#define AWS_HTTP2_STREAM_MANAGER_IMPL_H
+
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <aws/common/mutex.h>
+#include <aws/common/ref_count.h>
+#include <aws/http/http2_stream_manager.h>
+#include <aws/http/private/random_access_set.h>
+
+enum aws_h2_sm_state_type {
+    AWS_H2SMST_READY,
+    AWS_H2SMST_DESTROYING, /* On zero external ref count, can destroy */
+};
+
+enum aws_h2_sm_connection_state_type {
+    AWS_H2SMCST_IDEAL,
+    AWS_H2SMCST_NEARLY_FULL,
+    AWS_H2SMCST_FULL,
+};
+
+/* Live with the streams opening, and if there no outstanding pending acquisition and no opening streams on the
+ * connection, this structure should die */
+struct aws_h2_sm_connection {
+    struct aws_http2_stream_manager *stream_manager;
+    struct aws_http_connection *connection;
+    uint32_t num_streams_assigned;   /* From a stream assigned to the connection until the stream completed
+                                                     or failed to be created from the connection. */
+    uint32_t max_concurrent_streams; /* lower bound between user configured and the other side */
+
+    enum aws_h2_sm_connection_state_type state;
+};
+
+/* Live from the user request to acquire a stream to the stream completed. */
+struct aws_h2_sm_pending_stream_acquisition {
+    struct aws_allocator *allocator;
+    struct aws_linked_list_node node;
+    struct aws_http_make_request_options options;
+    struct aws_h2_sm_connection *sm_connection; /* The connection to make request to. Keep
+                                               NULL, until find available one and move it to the pending_make_requests
+                                               list. */
+    struct aws_http_message *request;
+    struct aws_channel_task make_request_task;
+    aws_http2_stream_manager_on_stream_acquired_fn *callback;
+    void *user_data;
+};
+
+/* connections_acquiring_count, open_stream_count, pending_make_requests_count AND pending_stream_acquisition_count */
+enum aws_sm_count_type {
+    AWS_SMCT_CONNECTIONS_ACQUIRING,
+    AWS_SMCT_OPEN_STREAM,
+    AWS_SMCT_PENDING_MAKE_REQUESTS,
+    AWS_SMCT_PENDING_ACQUISITION,
+    AWS_SMCT_COUNT,
+};
+
+struct aws_http2_stream_manager {
+    struct aws_allocator *allocator;
+    void *shutdown_complete_user_data;
+    aws_http2_stream_manager_shutdown_complete_fn *shutdown_complete_callback;
+    /**
+     * Underlying connection manager. Always has the same life time with the stream manager who owns it.
+     */
+    struct aws_http_connection_manager *connection_manager;
+    /**
+     * Refcount managed by user. Once this drops to zero, the manager state transitions to shutting down
+     */
+    struct aws_ref_count external_ref_count;
+    /**
+     * Internal refcount that keeps connection manager alive.
+     *
+     * It's a sum of connections_acquiring_count, open_stream_count, pending_make_requests_count and
+     * pending_stream_acquisition_count, besides the number of `struct aws_http2_stream_management_transaction` alive.
+     * And one for external usage.
+     *
+     * Once this refcount drops to zero, stream manager should either be cleaned up all the memory all waiting for
+     * the last task to clean un the memory and do nothing else.
+     */
+    struct aws_ref_count internal_ref_count;
+    struct aws_client_bootstrap *bootstrap;
+
+    size_t max_connections;
+
+    /**
+     * Default is no limit. 0 will be considered as using the default value.
+     * The ideal number of concurrent streams for a connection. Stream manager will try to create a new connection if
+     * one connection reaches this number. But, if the max connections reaches, manager will reuse connections to create
+     * the acquired steams as much as possible. */
+    size_t ideal_concurrent_streams_per_connection;
+    /**
+     * Default is no limit. 0 will be considered as using the default value.
+     * The real number of concurrent streams per connection will be controlled by the minmal value of the setting from
+     * other end and the value here.
+     */
+    size_t max_concurrent_streams_per_connection;
+
+    /**
+     * Task to invoke pending acquisition callbacks asynchronously if stream manager is shutting.
+     */
+    struct aws_event_loop *finish_pending_stream_acquisitions_task_event_loop;
+
+    /* Any thread may touch this data, but the lock must be held (unless it's an atomic) */
+    struct {
+        struct aws_mutex lock;
+        /*
+         * A manager can be in one of two states, READY or SHUTTING_DOWN.  The state transition
+         * takes place when ref_count drops to zero.
+         */
+        enum aws_h2_sm_state_type state;
+
+        /**
+         * A set of all connections that meet all requirement to use. Note: there will be connections not in this set,
+         * but hold by the stream manager, which can be tracked by the streams created on it. Set of `struct
+         * aws_h2_sm_connection *`
+         */
+        struct aws_random_access_set ideal_available_set;
+        /**
+         * A set of all available connections that exceed the soft limits set by users. Note: there will be connections
+         * not in this set, but hold by the stream manager, which can be tracked by the streams created. Set of `struct
+         * aws_h2_sm_connection *`
+         */
+        struct aws_random_access_set nonideal_available_set;
+        /* We don't mantain set for connections that is full or "dead" (Cannot make any new streams). We have streams
+         * opening from the connection tracking them */
+
+        /**
+         * The set of all incomplete stream acquisition requests (haven't decide what connection to make the request
+         * to), list of `struct aws_h2_sm_pending_stream_acquisition*`
+         */
+        struct aws_linked_list pending_stream_acquisitions;
+
+        /**
+         * The number of connections acquired from connection manager and not released yet.
+         */
+        size_t holding_connections_count;
+
+        /**
+         * Counts that contributes to the internal refcount.
+         * When the value changes, s_sm_count_increase/decrease_synced needed.
+         *
+         * AWS_SMCT_CONNECTIONS_ACQUIRING: The number of new connections we acquiring from the connection manager.
+         * AWS_SMCT_OPEN_STREAM: The number of streams that opened and not completed yet.
+         * AWS_SMCT_PENDING_MAKE_REQUESTS: The number of streams that scheduled to be made from a connection but haven't
+         *      been executed yet.
+         * AWS_SMCT_PENDING_ACQUISITION:  The number of all incomplete stream acquisition requests (haven't decide what
+         *      connection to make the request to). So that we don't have compute the size of a linked list every time.
+         */
+        size_t internal_refcount_stats[AWS_SMCT_COUNT];
+
+        bool finish_pending_stream_acquisitions_task_scheduled;
+    } synced_data;
+};
+
+/**
+ * Encompasses all of the external operations that need to be done for various
+ * events:
+ *  - User level:
+ *   stream manager release
+ *   stream acquire
+ *  - Internal eventloop (anther thread):
+ *   connection_acquired
+ *   stream_completed
+ *  - Internal (can happen from any thread):
+ *   connection acquire
+ *   connection release
+ *
+ * The transaction is built under the manager's lock (and the internal state is updated optimistically),
+ * but then executed outside of it.
+ */
+struct aws_http2_stream_management_transaction {
+    struct aws_http2_stream_manager *stream_manager;
+    struct aws_allocator *allocator;
+    size_t new_connections;
+    struct aws_h2_sm_connection *sm_connection_to_release;
+    struct aws_linked_list
+        pending_make_requests; /* List of aws_h2_sm_pending_stream_acquisition with chosen connection */
+};
+
+#endif /* AWS_HTTP2_STREAM_MANAGER_IMPL_H */

source/connection.c

@@ -390,7 +390,9 @@ void aws_http_connection_acquire(struct aws_http_connection *connection) {
 }
 
 void aws_http_connection_release(struct aws_http_connection *connection) {
-    AWS_ASSERT(connection);
+    if (!connection) {
+        return;
+    }
     size_t prev_refcount = aws_atomic_fetch_sub(&connection->refcount, 1);
     if (prev_refcount == 1) {
         AWS_LOGF_TRACE(