Skip to content

Commit 30abcb3

Browse files
waahm7quinnj
waahm7
authored and
quinnj
committed
Fix Error Handling For Connection Manager (awslabs#507)
1 parent 532e4f1 commit 30abcb3

File tree

3 files changed

+171
-35
lines changed

3 files changed

+171
-35
lines changed

‎source/connection_manager.c‎

Lines changed: 30 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1164,7 +1164,6 @@ static void s_aws_http_connection_manager_execute_transaction(struct aws_connect
11641164

11651165
struct aws_http_connection_manager *manager = work->manager;
11661166

1167-
int representative_error = 0;
11681167
size_t new_connection_failures = 0;
11691168

11701169
/*
@@ -1202,29 +1201,28 @@ static void s_aws_http_connection_manager_execute_transaction(struct aws_connect
12021201
*/
12031202
struct aws_array_list errors;
12041203
AWS_ZERO_STRUCT(errors);
1205-
/* Even if we can't init this array, we still need to invoke error callbacks properly */
1206-
bool push_errors = false;
12071204

12081205
if (work->new_connections > 0) {
12091206
AWS_LOGF_INFO(
12101207
AWS_LS_HTTP_CONNECTION_MANAGER,
12111208
"id=%p: Requesting %zu new connections from http",
12121209
(void *)manager,
12131210
work->new_connections);
1214-
push_errors = aws_array_list_init_dynamic(&errors, work->allocator, work->new_connections, sizeof(int)) ==
1215-
AWS_ERROR_SUCCESS;
1211+
AWS_FATAL_ASSERT(
1212+
aws_array_list_init_dynamic(&errors, work->allocator, work->new_connections, sizeof(int)) ==
1213+
AWS_OP_SUCCESS);
12161214
}
12171215

12181216
for (size_t i = 0; i < work->new_connections; ++i) {
12191217
if (s_aws_http_connection_manager_new_connection(manager)) {
12201218
++new_connection_failures;
1221-
representative_error = aws_last_error();
1222-
if (push_errors) {
1223-
AWS_FATAL_ASSERT(aws_array_list_push_back(&errors, &representative_error) == AWS_OP_SUCCESS);
1224-
}
1219+
int error = aws_last_error();
1220+
AWS_FATAL_ASSERT(aws_array_list_push_back(&errors, &error) == AWS_OP_SUCCESS);
12251221
}
12261222
}
12271223

1224+
bool has_pending_acquisitions = false;
1225+
struct aws_connection_management_transaction pending_acquisitions_work;
12281226
if (new_connection_failures > 0) {
12291227
/*
12301228
* We failed and aren't going to receive a callback, but the current state assumes we will receive
@@ -1235,44 +1233,45 @@ static void s_aws_http_connection_manager_execute_transaction(struct aws_connect
12351233
AWS_FATAL_ASSERT(manager->internal_ref[AWS_HCMCT_PENDING_CONNECTIONS] >= new_connection_failures);
12361234
s_connection_manager_internal_ref_decrease(manager, AWS_HCMCT_PENDING_CONNECTIONS, new_connection_failures);
12371235

1238-
/*
1239-
* Rather than failing one acquisition for each connection failure, if there's at least one
1240-
* connection failure, we instead fail all excess acquisitions, since there's no pending
1241-
* connect that will necessarily resolve them.
1242-
*
1243-
* Try to correspond an error with the acquisition failure, but as a fallback just use the
1244-
* representative error.
1245-
*/
1246-
size_t i = 0;
1247-
while (manager->pending_acquisition_count > manager->internal_ref[AWS_HCMCT_PENDING_CONNECTIONS]) {
1248-
int error = representative_error;
1249-
if (i < aws_array_list_length(&errors)) {
1250-
aws_array_list_get_at(&errors, &error, i);
1251-
}
1252-
1236+
for (size_t i = 0; i < new_connection_failures && manager->pending_acquisition_count > 0; i++) {
1237+
int error;
1238+
aws_array_list_get_at(&errors, &error, i);
12531239
AWS_LOGF_DEBUG(
12541240
AWS_LS_HTTP_CONNECTION_MANAGER,
1255-
"id=%p: Failing excess connection acquisition with error code %d",
1241+
"id=%p: Failing connection acquisition with error code %d",
12561242
(void *)manager,
12571243
(int)error);
12581244
s_aws_http_connection_manager_move_front_acquisition(manager, NULL, error, &work->completions);
1259-
++i;
12601245
}
1261-
1246+
has_pending_acquisitions =
1247+
manager->internal_ref[AWS_HCMCT_PENDING_CONNECTIONS] + manager->pending_settings_count <
1248+
manager->pending_acquisition_count;
1249+
if (has_pending_acquisitions) {
1250+
/* If there are pending acquisitions, schedule work again to try acquiring them */
1251+
s_aws_connection_management_transaction_init(&pending_acquisitions_work, manager);
1252+
s_aws_http_connection_manager_build_transaction(&pending_acquisitions_work);
1253+
}
12621254
aws_mutex_unlock(&manager->lock);
12631255
}
12641256

12651257
/*
12661258
* Step 4 - Perform acquisition callbacks
12671259
*/
12681260
s_aws_http_connection_manager_complete_acquisitions(&work->completions, work->allocator);
1269-
12701261
aws_array_list_clean_up(&errors);
12711262

12721263
/*
12731264
* Step 5 - Clean up work. Do this here rather than at the end of every caller. Destroy the manager if necessary
12741265
*/
12751266
s_aws_connection_management_transaction_clean_up(work);
1267+
1268+
/*
1269+
* Step 6 - If some connection acquisition requests failed and we still have more pending acquisitions, try to
1270+
* acquire them.
1271+
*/
1272+
if (has_pending_acquisitions) {
1273+
s_aws_http_connection_manager_execute_transaction(&pending_acquisitions_work);
1274+
}
12761275
}
12771276

12781277
void aws_http_connection_manager_acquire_connection(
@@ -1479,12 +1478,11 @@ static void s_cm_on_connection_ready_or_failed(
14791478
work->connection_to_release = connection;
14801479
}
14811480
} else {
1482-
/* fail acquisition as one connection cannot be used any more */
1483-
while (manager->pending_acquisition_count >
1484-
manager->internal_ref[AWS_HCMCT_PENDING_CONNECTIONS] + manager->pending_settings_count) {
1481+
if (manager->pending_acquisition_count > 0) {
1482+
/* fail acquisition as connection acquire failed */
14851483
AWS_LOGF_DEBUG(
14861484
AWS_LS_HTTP_CONNECTION_MANAGER,
1487-
"id=%p: Failing excess connection acquisition with error code %d",
1485+
"id=%p: Failing connection acquisition with error code %d",
14881486
(void *)manager,
14891487
(int)error_code);
14901488
s_aws_http_connection_manager_move_front_acquisition(manager, NULL, error_code, &work->completions);

‎tests/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -529,6 +529,8 @@ add_net_test_case(connection_manager_setup_shutdown)
529529
add_net_test_case(connection_manager_acquire_release_mix_synchronous)
530530
add_net_test_case(connection_manager_acquisition_timeout)
531531
add_net_test_case(connection_manager_connect_callback_failure)
532+
add_net_test_case(test_connection_manager_connect_callback_async_failure)
533+
add_net_test_case(test_connection_manager_connect_callback_async_with_immediate_failure)
532534
add_net_test_case(connection_manager_connect_immediate_failure)
533535
add_net_test_case(connection_manager_proxy_setup_shutdown)
534536
add_net_test_case(connection_manager_idle_culling_single)

‎tests/test_connection_manager.c‎

Lines changed: 139 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -815,7 +815,7 @@ AWS_TEST_CASE(
815815
connection_manager_max_pending_acquisitions_with_vended_connections,
816816
s_test_connection_manager_max_pending_acquisitions_with_vended_connections);
817817

818-
static int s_aws_http_connection_manager_create_connection_sync_mock(
818+
static int s_aws_http_connection_manager_create_connection_validate(
819819
const struct aws_http_client_connection_options *options) {
820820
struct cm_tester *tester = &s_tester;
821821

@@ -826,8 +826,6 @@ static int s_aws_http_connection_manager_create_connection_sync_mock(
826826
ASSERT_TRUE(aws_byte_cursor_eq_c_str(&interface_name, options->socket_options->network_interface_name));
827827
}
828828

829-
size_t next_connection_id = aws_atomic_fetch_add(&tester->next_connection_id, 1);
830-
831829
ASSERT_SUCCESS(aws_mutex_lock(&tester->lock));
832830
tester->release_connection_fn = options->on_shutdown;
833831
ASSERT_SUCCESS(aws_mutex_unlock(&tester->lock));
@@ -843,7 +841,15 @@ static int s_aws_http_connection_manager_create_connection_sync_mock(
843841
ASSERT_UINT_EQUALS(options->proxy_options->connection_type, tester->verify_proxy_options->connection_type);
844842
}
845843

844+
return AWS_OP_SUCCESS;
845+
}
846+
static int s_aws_http_connection_manager_create_connection_sync_mock(
847+
const struct aws_http_client_connection_options *options) {
848+
s_aws_http_connection_manager_create_connection_validate(options);
849+
struct cm_tester *tester = &s_tester;
850+
846851
struct mock_connection *connection = NULL;
852+
size_t next_connection_id = aws_atomic_fetch_add(&tester->next_connection_id, 1);
847853

848854
if (next_connection_id < aws_array_list_length(&tester->mock_connections)) {
849855
aws_array_list_get_at(&tester->mock_connections, &connection, next_connection_id);
@@ -864,6 +870,63 @@ static int s_aws_http_connection_manager_create_connection_sync_mock(
864870
return aws_raise_error(AWS_ERROR_HTTP_UNKNOWN);
865871
}
866872

873+
struct connect_task_args {
874+
void *user_data;
875+
struct mock_connection *connection;
876+
aws_http_on_client_connection_setup_fn *on_setup;
877+
};
878+
879+
static void s_aws_http_connection_manager_connect_task(
880+
struct aws_task *task,
881+
void *user_data,
882+
enum aws_task_status status) {
883+
(void)status;
884+
struct cm_tester *tester = &s_tester;
885+
886+
struct connect_task_args *task_args = user_data;
887+
struct mock_connection *connection = task_args->connection;
888+
if (connection) {
889+
if (connection->result == AWS_NCRT_SUCCESS) {
890+
task_args->on_setup((struct aws_http_connection *)connection, AWS_ERROR_SUCCESS, task_args->user_data);
891+
} else if (connection->result == AWS_NCRT_ERROR_VIA_CALLBACK) {
892+
task_args->on_setup(NULL, AWS_ERROR_HTTP_UNKNOWN, task_args->user_data);
893+
} else {
894+
AWS_FATAL_ASSERT(0 && "Unexpected connection->result");
895+
}
896+
}
897+
898+
aws_mem_release(tester->allocator, task);
899+
aws_mem_release(tester->allocator, task_args);
900+
}
901+
902+
static int s_aws_http_connection_manager_connect_async_mock(const struct aws_http_client_connection_options *options) {
903+
s_aws_http_connection_manager_create_connection_validate(options);
904+
struct cm_tester *tester = &s_tester;
905+
906+
struct mock_connection *connection = NULL;
907+
size_t next_connection_id = aws_atomic_fetch_add(&tester->next_connection_id, 1);
908+
if (next_connection_id < aws_array_list_length(&tester->mock_connections)) {
909+
aws_array_list_get_at(&tester->mock_connections, &connection, next_connection_id);
910+
}
911+
912+
if (connection->result == AWS_NCRT_ERROR_FROM_CREATE) {
913+
return aws_raise_error(AWS_ERROR_HTTP_UNKNOWN);
914+
}
915+
916+
struct aws_task *task = aws_mem_calloc(options->allocator, 1, sizeof(struct aws_task));
917+
struct connect_task_args *task_args = aws_mem_calloc(options->allocator, 1, sizeof(struct connect_task_args));
918+
task_args->connection = connection;
919+
task_args->user_data = options->user_data;
920+
task_args->on_setup = options->on_setup;
921+
aws_task_init(task, s_aws_http_connection_manager_connect_task, task_args, "create_connection_task");
922+
923+
struct aws_event_loop *event_loop = aws_event_loop_group_get_next_loop(tester->event_loop_group);
924+
uint64_t now;
925+
ASSERT_SUCCESS(aws_event_loop_current_clock_time(event_loop, &now));
926+
aws_event_loop_schedule_task_future(event_loop, task, now + 1000000000);
927+
return AWS_OP_SUCCESS;
928+
}
929+
867930
static void s_aws_http_connection_manager_release_connection_sync_mock(struct aws_http_connection *connection) {
868931
(void)connection;
869932

@@ -916,6 +979,17 @@ static struct aws_http_connection_manager_system_vtable s_synchronous_mocks = {
916979
.aws_http_connection_get_version = s_aws_http_connection_manager_connection_get_version_sync_mock,
917980
};
918981

982+
static struct aws_http_connection_manager_system_vtable s_async_connect_mock = {
983+
.aws_http_client_connect = s_aws_http_connection_manager_connect_async_mock,
984+
.aws_http_connection_release = s_aws_http_connection_manager_release_connection_sync_mock,
985+
.aws_http_connection_close = s_aws_http_connection_manager_close_connection_sync_mock,
986+
.aws_http_connection_new_requests_allowed = s_aws_http_connection_manager_is_connection_available_sync_mock,
987+
.aws_high_res_clock_get_ticks = aws_high_res_clock_get_ticks,
988+
.aws_http_connection_get_channel = s_aws_http_connection_manager_connection_get_channel_sync_mock,
989+
.aws_channel_thread_is_callers_thread = s_aws_http_connection_manager_is_callers_thread_sync_mock,
990+
.aws_http_connection_get_version = s_aws_http_connection_manager_connection_get_version_sync_mock,
991+
};
992+
919993
static int s_test_connection_manager_with_network_interface_list(struct aws_allocator *allocator, void *ctx) {
920994
(void)ctx;
921995
struct aws_byte_cursor *interface_names_array = aws_mem_calloc(allocator, 3, sizeof(struct aws_byte_cursor));
@@ -1051,6 +1125,68 @@ static int s_test_connection_manager_connect_callback_failure(struct aws_allocat
10511125
}
10521126
AWS_TEST_CASE(connection_manager_connect_callback_failure, s_test_connection_manager_connect_callback_failure);
10531127

1128+
static int s_test_connection_manager_connect_callback_async_failure(struct aws_allocator *allocator, void *ctx) {
1129+
(void)ctx;
1130+
int error_connections = 5;
1131+
int success_connections = 5;
1132+
struct cm_tester_options options = {
1133+
.allocator = allocator,
1134+
.max_connections = error_connections,
1135+
.mock_table = &s_async_connect_mock,
1136+
};
1137+
1138+
ASSERT_SUCCESS(s_cm_tester_init(&options));
1139+
1140+
s_add_mock_connections(error_connections, AWS_NCRT_ERROR_VIA_CALLBACK, false);
1141+
s_add_mock_connections(success_connections, AWS_NCRT_SUCCESS, true);
1142+
1143+
s_acquire_connections(error_connections + success_connections);
1144+
1145+
ASSERT_SUCCESS(s_wait_on_connection_reply_count(error_connections + success_connections));
1146+
1147+
ASSERT_UINT_EQUALS(s_tester.connection_errors, error_connections);
1148+
ASSERT_SUCCESS(s_release_connections(success_connections, false));
1149+
1150+
ASSERT_SUCCESS(s_cm_tester_clean_up());
1151+
1152+
return AWS_OP_SUCCESS;
1153+
}
1154+
AWS_TEST_CASE(
1155+
test_connection_manager_connect_callback_async_failure,
1156+
s_test_connection_manager_connect_callback_async_failure);
1157+
1158+
static int s_test_connection_manager_connect_callback_async_with_immediate_failure(
1159+
struct aws_allocator *allocator,
1160+
void *ctx) {
1161+
(void)ctx;
1162+
int error_connections = 5;
1163+
int success_connections = 5;
1164+
struct cm_tester_options options = {
1165+
.allocator = allocator,
1166+
.max_connections = error_connections + success_connections,
1167+
.mock_table = &s_async_connect_mock,
1168+
};
1169+
1170+
ASSERT_SUCCESS(s_cm_tester_init(&options));
1171+
1172+
s_add_mock_connections(success_connections, AWS_NCRT_SUCCESS, true);
1173+
s_add_mock_connections(error_connections, AWS_NCRT_ERROR_VIA_CALLBACK, false);
1174+
1175+
s_acquire_connections(error_connections + success_connections);
1176+
1177+
ASSERT_SUCCESS(s_wait_on_connection_reply_count(error_connections + success_connections));
1178+
1179+
ASSERT_UINT_EQUALS(s_tester.connection_errors, error_connections);
1180+
ASSERT_SUCCESS(s_release_connections(success_connections, false));
1181+
1182+
ASSERT_SUCCESS(s_cm_tester_clean_up());
1183+
1184+
return AWS_OP_SUCCESS;
1185+
}
1186+
AWS_TEST_CASE(
1187+
test_connection_manager_connect_callback_async_with_immediate_failure,
1188+
s_test_connection_manager_connect_callback_async_with_immediate_failure);
1189+
10541190
static int s_test_connection_manager_connect_immediate_failure(struct aws_allocator *allocator, void *ctx) {
10551191
(void)ctx;
10561192

0 commit comments

Comments
 (0)