Can create streams on h2 connections (#175)

Author: graebm

include/aws/http/http.h

@@ -48,6 +48,7 @@ enum aws_http_errors {
     AWS_ERROR_HTTP_CONNECTION_MANAGER_SHUTTING_DOWN,
     AWS_ERROR_HTTP_PROTOCOL_ERROR,
     AWS_ERROR_HTTP_STREAM_CLOSED,
+    AWS_ERROR_HTTP_STREAM_IDS_EXHAUSTED,
     AWS_ERROR_HTTP_INVALID_FRAME_SIZE,
     AWS_ERROR_HTTP_COMPRESSION,
 

include/aws/http/private/h2_connection.h

@@ -27,18 +27,40 @@ struct aws_h2_decoder;
 struct aws_h2_connection {
     struct aws_http_connection base;
 
+    struct aws_channel_task cross_thread_work_task;
+
     /* Only the event-loop thread may touch this data */
     struct {
         struct aws_h2_decoder *decoder;
         struct aws_h2_frame_encoder encoder;
+
+        /* True when reading/writing has stopped, whether due to errors or normal channel shutdown. */
+        bool is_reading_stopped;
+        bool is_writing_stopped;
+
+        /* Maps stream-id to aws_h2_frame* */
+        struct aws_hash_table active_streams_map;
+
     } thread_data;
 
     /* Any thread may touch this data, but the lock must be held */
     struct {
         struct aws_mutex lock;
 
+        /* New `aws_h2_stream *` that haven't moved to `thread_data` yet */
+        struct aws_linked_list pending_stream_list;
+
         /* Refers to the next stream id to vend */
         uint32_t next_stream_id;
+
+        /* If non-zero, reason to immediately reject new streams. (ex: closing) */
+        int new_stream_error_code;
+
+        bool is_cross_thread_work_task_scheduled;
+
+        /* For checking status from outside the event-loop thread. */
+        bool is_open;
+
     } synced_data;
 };
 
@@ -54,9 +76,6 @@ struct aws_http_connection *aws_http_connection_new_http2_client(
     struct aws_allocator *allocator,
     size_t initial_window_size);
 
-AWS_HTTP_API
-uint32_t aws_h2_connection_get_next_stream_id(struct aws_h2_connection *connection);
-
 AWS_EXTERN_C_END
 
 #endif /* AWS_HTTP_H2_CONNECTION_H */

include/aws/http/private/h2_stream.h

@@ -21,6 +21,18 @@
 
 #include <aws/common/mutex.h>
 
+#include <inttypes.h>
+
+#define AWS_H2_STREAM_LOGF(level, stream, text, ...)                                                                   \
+    AWS_LOGF_##level(                                                                                                  \
+        AWS_LS_HTTP_STREAM,                                                                                            \
+        "id=%" PRIu32 " connection=%p state=%s: " text,                                                                \
+        (stream)->id,                                                                                                  \
+        (void *)(stream)->base.owning_connection,                                                                      \
+        aws_h2_stream_state_to_str((stream)->thread_data.state),                                                       \
+        __VA_ARGS__)
+#define AWS_H2_STREAM_LOG(level, stream, text) AWS_H2_STREAM_LOGF(level, (stream), "%s", (text))
+
 enum aws_h2_stream_state {
     AWS_H2_STREAM_STATE_IDLE,
     AWS_H2_STREAM_STATE_RESERVED_LOCAL,
@@ -36,7 +48,9 @@ enum aws_h2_stream_state {
 struct aws_h2_stream {
     struct aws_http_stream base;
 
-    const uint32_t id;
+    uint32_t id;
+
+    struct aws_linked_list_node node;
 
     /* Only the event-loop thread may touch this data */
     struct {
@@ -60,7 +74,7 @@ AWS_HTTP_API
 const char *aws_h2_stream_state_to_str(enum aws_h2_stream_state state);
 
 AWS_HTTP_API
-struct aws_h2_stream *aws_h1_stream_new_request(
+struct aws_h2_stream *aws_h2_stream_new_request(
     struct aws_http_connection *client_connection,
     const struct aws_http_make_request_options *options);
 

source/h1_connection.c

@@ -1576,6 +1576,11 @@ static int s_handler_process_read_message(
     }
 
     AWS_LOGF_TRACE(AWS_LS_HTTP_CONNECTION, "id=%p: Done processing message.", (void *)&connection->base);
+    if (message) {
+        /* release message back to pool before re-opening window */
+        aws_mem_release(message->allocator, message);
+        message = NULL;
+    }
 
     /* Increment read window */
     if (incoming_message_size > connection->thread_data.incoming_message_window_shrink_size) {
@@ -1593,9 +1598,6 @@ static int s_handler_process_read_message(
         }
     }
 
-    if (message) {
-        aws_mem_release(message->allocator, message);
-    }
     return AWS_OP_SUCCESS;
 
 shutdown:
@@ -1735,7 +1737,7 @@ static int s_handler_shutdown(
         }
 
         /* It's OK to access synced_data.pending_stream_list without holding the lock because
-         * no more streams can be added after s_shutdown_connection() has been invoked. */
+         * no more streams can be added after s_stop() has been invoked. */
         while (!aws_linked_list_empty(&connection->synced_data.pending_stream_list)) {
             struct aws_linked_list_node *node = aws_linked_list_front(&connection->synced_data.pending_stream_list);
             s_stream_complete(AWS_CONTAINER_OF(node, struct aws_h1_stream, node), stream_error_code);