Ping frame for H2

include/aws/http/private/h2_frames.h

@@ -115,6 +115,10 @@ struct aws_h2_frame {
     struct aws_linked_list_node node;
     enum aws_h2_frame_type type;
     uint32_t stream_id;
+
+    /* If true, frame will be sent before those with normal priority.
+     * Useful for frames like PING ACK where low latency is important. */
+    bool high_priority;
 };
 
 /* A h2 setting and its value, used in SETTINGS frame */

source/h2_connection.c

@@ -61,6 +61,8 @@ static struct aws_http_stream *s_connection_make_request(
 static void s_cross_thread_work_task(struct aws_channel_task *task, void *arg, enum aws_task_status status);
 static void s_outgoing_frames_task(struct aws_channel_task *task, void *arg, enum aws_task_status status);
 
+static int s_decoder_on_ping(uint8_t opaque_data[AWS_H2_PING_DATA_SIZE], void *userdata);
+
 static struct aws_http_connection_vtable s_h2_connection_vtable = {
     .channel_handler_vtable =
         {
@@ -84,6 +86,7 @@ static struct aws_http_connection_vtable s_h2_connection_vtable = {
 
 static const struct aws_h2_decoder_vtable s_h2_decoder_vtable = {
     .on_data = NULL,
+    .on_ping = s_decoder_on_ping,
 };
 
 static void s_lock_synced_data(struct aws_h2_connection *connection) {
@@ -294,7 +297,22 @@ void aws_h2_connection_enqueue_outgoing_frame(struct aws_h2_connection *connecti
     AWS_PRECONDITION(frame->type != AWS_H2_FRAME_T_DATA);
     AWS_PRECONDITION(aws_channel_thread_is_callers_thread(connection->base.channel_slot->channel));
 
-    aws_linked_list_push_back(&connection->thread_data.outgoing_frames_queue, &frame->node);
+    if (frame->high_priority) {
+        /* Check from the head of the queue, and find a node with normal priority, and insert before it */
+        struct aws_linked_list_node *iter = aws_linked_list_begin(&connection->thread_data.outgoing_frames_queue);
+        /* one past the last element */
+        const struct aws_linked_list_node *end = aws_linked_list_end(&connection->thread_data.outgoing_frames_queue);
+        while (iter != end) {
+            struct aws_h2_frame *frame_i = AWS_CONTAINER_OF(iter, struct aws_h2_frame, node);
+            if (!frame_i->high_priority) {
+                break;
+            }
+            iter = iter->next;
+        }
+        aws_linked_list_insert_before(iter, &frame->node);
+    } else {
+        aws_linked_list_push_back(&connection->thread_data.outgoing_frames_queue, &frame->node);
+    }
 }
 
 static void s_on_channel_write_complete(
@@ -492,6 +510,24 @@ static void s_try_write_outgoing_frames(struct aws_h2_connection *connection) {
     s_outgoing_frames_task(&connection->outgoing_frames_task, connection, AWS_TASK_STATUS_RUN_READY);
 }
 
+/* Decoder callbacks */
+static int s_decoder_on_ping(uint8_t opaque_data[AWS_H2_PING_DATA_SIZE], void *userdata) {
+    struct aws_h2_connection *connection = userdata;
+
+    /* send a PING frame with the ACK flag set in response, with an identical payload. */
+    struct aws_h2_frame *ping_ack_frame = aws_h2_frame_new_ping(connection->base.alloc, true, opaque_data);
+    if (!ping_ack_frame) {
+        goto error;
+    }
+
+    aws_h2_connection_enqueue_outgoing_frame(connection, ping_ack_frame);
+    s_try_write_outgoing_frames(connection);
+    return AWS_OP_SUCCESS;
+error:
+    CONNECTION_LOGF(ERROR, connection, "Ping ACK frame failed to be sent, error %s", aws_error_name(aws_last_error()));
+    return AWS_OP_ERR;
+}
+
 static int s_send_connection_preface_client_string(struct aws_h2_connection *connection) {
 
     /* Just send the magic string on its own aws_io_message. */
@@ -735,16 +771,51 @@ static int s_handler_process_read_message(
     struct aws_channel_handler *handler,
     struct aws_channel_slot *slot,
     struct aws_io_message *message) {
-
-    (void)handler;
     (void)slot;
-    (void)message;
+    struct aws_h2_connection *connection = handler->impl;
+
+    CONNECTION_LOGF(TRACE, connection, "Begin processing message of size %zu.", message->message_data.len);
+
+    if (connection->thread_data.is_reading_stopped) {
+        CONNECTION_LOG(ERROR, connection, "Cannot process message because connection is shutting down.");
+        aws_raise_error(AWS_ERROR_HTTP_CONNECTION_CLOSED);
+        goto shutdown;
+    }
+
+    struct aws_byte_cursor message_cursor = aws_byte_cursor_from_buf(&message->message_data);
+    if (aws_h2_decode(connection->thread_data.decoder, &message_cursor)) {
+        CONNECTION_LOGF(
+            ERROR,
+            connection,
+            "Decoding message failed, error %d (%s). Closing connection",
+            aws_last_error(),
+            aws_error_name(aws_last_error()));
+    }
 
     /* HTTP/2 protocol uses WINDOW_UPDATE frames to coordinate data rates with peer,
      * so we can just keep the aws_channel's read-window wide open */
-    /* #TODO update read window by however much we just read */
+    if (aws_channel_slot_increment_read_window(slot, message->message_data.len)) {
+        CONNECTION_LOGF(
+            ERROR,
+            connection,
+            "Incrementing read window failed, error %d (%s). Closing connection",
+            aws_last_error(),
+            aws_error_name(aws_last_error()));
+    }
 
-    return aws_raise_error(AWS_ERROR_UNIMPLEMENTED);
+    /* release message */
+    if (message) {
+        aws_mem_release(message->allocator, message);
+        message = NULL;
+    }
+    return AWS_OP_SUCCESS;
+shutdown:
+    if (message) {
+        aws_mem_release(message->allocator, message);
+    }
+    /* Stop reading, because the reading error happans here */
+    s_stop(connection, true /*stop_reading*/, false /*stop_writing*/, true /*schedule_shutdown*/, aws_last_error());
+    return AWS_OP_SUCCESS;
 }
 
 static int s_handler_process_write_message(

source/h2_frames.c

@@ -985,6 +985,8 @@ struct aws_h2_frame *aws_h2_frame_new_ping(
     writes_ok &= aws_byte_buf_write(&frame->encoded_buf, opaque_data, AWS_H2_PING_DATA_SIZE);
     AWS_ASSERT(writes_ok);
 
+    /* PING responses SHOULD be given higher priority than any other frame */
+    frame->base.high_priority = ack;
     return &frame->base;
 }
 

tests/CMakeLists.txt

@@ -300,6 +300,7 @@ add_h2_decoder_test_set(h2_decoder_err_bad_preface_from_client_3)
 add_test_case(h2_client_sanity_check)
 add_test_case(h2_client_request_create)
 add_test_case(h2_client_connection_preface_sent)
+add_test_case(h2_client_ping_ack)
 
 add_test_case(server_new_destroy)
 add_test_case(connection_setup_shutdown)

tests/test_h2_client.c

@@ -126,3 +126,32 @@ TEST_CASE(h2_client_connection_preface_sent) {
 
     return s_tester_clean_up();
 }
+
+/* Test that client will automatically send the PING ACK frame back, when the PING frame is received */
+TEST_CASE(h2_client_ping_ack) {
+    ASSERT_SUCCESS(s_tester_init(allocator, ctx));
+
+    /* Connection preface requires that SETTINGS be sent first (RFC-7540 3.5). */
+    ASSERT_SUCCESS(h2_fake_peer_send_connection_preface_default_settings(&s_tester.peer));
+
+    uint8_t opaque_data[AWS_H2_PING_DATA_SIZE] = {0, 1, 2, 3, 4, 5, 6, 7};
+
+    struct aws_h2_frame *frame = aws_h2_frame_new_ping(allocator, false /*ack*/, opaque_data);
+    ASSERT_NOT_NULL(frame);
+
+    ASSERT_SUCCESS(h2_fake_peer_send_frame(&s_tester.peer, frame));
+
+    /* Have the fake peer to run its decoder on what the client has written.
+     * The decoder will raise an error if it doesn't receive the "client connection preface string" first. */
+    ASSERT_SUCCESS(h2_fake_peer_decode_messages_from_testing_channel(&s_tester.peer));
+
+    /* Now check that client sent PING ACK frame, it should be the latest frame received by peer
+     * The last frame should be a ping type with ack on, and identical payload */
+    struct h2_decoded_frame *latest_frame = h2_decode_tester_latest_frame(&s_tester.peer.decode);
+    ASSERT_UINT_EQUALS(AWS_H2_FRAME_T_PING, latest_frame->type);
+    ASSERT_TRUE(latest_frame->ack);
+    ASSERT_BIN_ARRAYS_EQUALS(opaque_data, AWS_H2_PING_DATA_SIZE, latest_frame->ping_opaque_data, AWS_H2_PING_DATA_SIZE);
+
+    return s_tester_clean_up();
+}
+/* TODO: test that ping response is sent with higher priority than any other frame */