From: Arran Cudbard-Bell Date: Sat, 7 Dec 2019 11:31:25 +0000 (+0700) Subject: Rebalance the backlogs of connections when a new connection becomes active X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=d27174eb1c670520a6ff07b07beee5400a2b278a;p=thirdparty%2Ffreeradius-server.git Rebalance the backlogs of connections when a new connection becomes active --- diff --git a/src/lib/server/trunk.c b/src/lib/server/trunk.c index 98663c198a8..774f058765d 100644 --- a/src/lib/server/trunk.c +++ b/src/lib/server/trunk.c @@ -571,6 +571,7 @@ static void trunk_connection_enter_inactive(fr_trunk_connection_t *tconn); static void trunk_connection_enter_draining(fr_trunk_connection_t *tconn); static void trunk_connection_enter_active(fr_trunk_connection_t *tconn); +static void trunk_rebalance(fr_trunk_t *trunk); static void trunk_manage(fr_trunk_t *trunk, fr_time_t now); static void _trunk_manage_timer(fr_event_list_t *el, fr_time_t now, void *uctx); static void trunk_backlog_drain(fr_trunk_t *trunk); @@ -1231,13 +1232,21 @@ static fr_trunk_enqueue_t trunk_request_enqueue_existing(fr_trunk_request_t *tre * requests into. * @param[in] tconn to dequeue requests from. * @param[in] states Dequeue request in these states. + * @param[in] max The maximum number of requests to dequeue. 0 for unlimited. */ -static void trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk_connection_t *tconn, int states) +static uint64_t trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk_connection_t *tconn, + int states, uint64_t max) { fr_trunk_request_t *treq; + uint64_t count = 0; + + if (max == 0) max = UINT64_MAX; + +#define OVER_MAX_CHECK if (++count > max) return (count - 1) #define DEQUEUE_ALL(_src_list) \ while ((treq = fr_dlist_head(_src_list))) { \ + OVER_MAX_CHECK; \ trunk_request_enter_unassigned(treq); \ fr_dlist_insert_tail(out, treq); \ } @@ -1257,6 +1266,7 @@ static void trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk_con * ....same with cancel partial */ if (states & FR_TRUNK_REQUEST_CANCEL_PARTIAL) { + OVER_MAX_CHECK; treq = tconn->cancel_partial; if (treq) { trunk_request_enter_unassigned(treq); @@ -1269,6 +1279,7 @@ static void trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk_con */ if (states & FR_TRUNK_REQUEST_PENDING) { while ((treq = fr_heap_peek(tconn->pending))) { + OVER_MAX_CHECK; trunk_request_enter_unassigned(treq); fr_dlist_insert_tail(out, treq); } @@ -1278,6 +1289,7 @@ static void trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk_con * Cancel partially sent requests */ if (states & FR_TRUNK_REQUEST_PARTIAL) { + OVER_MAX_CHECK; treq = tconn->partial; if (treq) { trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_MOVE); @@ -1291,31 +1303,47 @@ static void trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk_con */ if (states & FR_TRUNK_REQUEST_SENT) { while ((treq = fr_dlist_head(&tconn->sent))) { + OVER_MAX_CHECK; trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_MOVE); trunk_request_enter_unassigned(treq); fr_dlist_insert_tail(out, treq); } } + + return count; } /** Remove requests in specified states from a connection, attempting to distribute them to new connections * * @param[in] tconn To remove requests from. * @param[in] states One or more states or'd together. + * @param[in] max The maximum number of requests to dequeue. 0 for unlimited. + * * @return the number of requests re-queued. */ -static uint64_t trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, int states) +static uint64_t trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, int states, uint64_t max) { + fr_trunk_t *trunk = tconn->trunk; fr_dlist_head_t to_process; fr_trunk_request_t *treq = NULL; uint64_t moved = 0; + if (max == 0) max = UINT64_MAX; + fr_dlist_talloc_init(&to_process, fr_trunk_request_t, list); /* * Remove non-cancelled requests from the connection */ - trunk_connection_requests_dequeue(&to_process, tconn, states & ~FR_TRUNK_REQUEST_CANCEL_ALL); + moved += trunk_connection_requests_dequeue(&to_process, tconn, states & ~FR_TRUNK_REQUEST_CANCEL_ALL, max); + + /* + * Prevent requests being requeued on the same trunk + * connection, which would break rebalancing. + * + * This is a bit of a hack. + */ + if (tconn->state == FR_TRUNK_CONN_ACTIVE) fr_heap_extract(trunk->active, tconn); /* * Loop over all the requests we gathered @@ -1324,8 +1352,6 @@ static uint64_t trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, while ((treq = fr_dlist_next(&to_process, treq))) { fr_trunk_request_t *prev; - moved++; - prev = fr_dlist_remove(&to_process, treq); switch (trunk_request_enqueue_existing(treq)) { case TRUNK_ENQUEUE_OK: @@ -1354,6 +1380,13 @@ static uint64_t trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, treq = prev; } + /* + * Add the connection back into the active list + */ + if (tconn->state == FR_TRUNK_CONN_ACTIVE) fr_heap_insert(trunk->active, tconn); + + if (moved >= max) return moved; + /* * Deal with the cancelled requests specially we can't * queue them up again as they were only valid on that @@ -1363,12 +1396,11 @@ static uint64_t trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, * they should already be in the unassigned state, * just means freeing them. */ - trunk_connection_requests_dequeue(&to_process, tconn, states & FR_TRUNK_REQUEST_CANCEL_ALL); + moved += trunk_connection_requests_dequeue(&to_process, tconn, + states & FR_TRUNK_REQUEST_CANCEL_ALL, max - moved); while ((treq = fr_dlist_next(&to_process, treq))) { fr_trunk_request_t *prev; - moved++; - prev = fr_dlist_remove(&to_process, treq); talloc_free(treq); treq = prev; @@ -1390,15 +1422,17 @@ static uint64_t trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, * * @param[in] tconn to move requests off of. * @param[in] states Only move requests in this state. + * @param[in] max The maximum number of requests to dequeue. 0 for unlimited. + * * @return The number of requests requeued. */ -uint64_t fr_trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, int states) +uint64_t fr_trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, int states, uint64_t max) { switch (tconn->state) { case FR_TRUNK_CONN_ACTIVE: case FR_TRUNK_CONN_INACTIVE: case FR_TRUNK_CONN_DRAINING: - return trunk_connection_requests_requeue(tconn, states); + return trunk_connection_requests_requeue(tconn, states, max); default: return 0; @@ -1814,12 +1848,13 @@ static inline void trunk_connection_auto_active(fr_trunk_connection_t *tconn) fr_trunk_t *trunk = tconn->trunk; uint32_t count; - if (!trunk->conf->max_requests_per_conn || - tconn->signalled_inactive || + if (tconn->signalled_inactive || (tconn->state != FR_TRUNK_CONN_INACTIVE)) return; count = fr_trunk_request_count_by_connection(tconn, FR_TRUNK_REQUEST_ALL); - if (count < trunk->conf->max_requests_per_conn) trunk_connection_enter_active(tconn); + if ((trunk->conf->max_requests_per_conn == 0) || (count < trunk->conf->max_requests_per_conn)) { + trunk_connection_enter_active(tconn); + } } /** A connection is readable. Call the request_demux function to read pending requests @@ -1962,7 +1997,7 @@ static void trunk_connection_enter_draining(fr_trunk_connection_t *tconn) * requests, so the connection is drained * quicker. */ - trunk_connection_requests_requeue(tconn, FR_TRUNK_REQUEST_PENDING); + trunk_connection_requests_requeue(tconn, FR_TRUNK_REQUEST_PENDING, 0); } /** Transition a connection back to the active state @@ -1996,6 +2031,16 @@ static void trunk_connection_enter_active(fr_trunk_connection_t *tconn) CONN_STATE_TRANSITION(FR_TRUNK_CONN_ACTIVE); MEM(fr_heap_insert(trunk->active, tconn) == 0); /* re-insert into the active heap*/ + /* + * Reorder the connections + */ + CONN_REORDER(tconn); + + /* + * Rebalance requests + */ + trunk_rebalance(trunk); + /* * We place requests into the backlog * because there were no connections @@ -2139,7 +2184,7 @@ static void _trunk_connection_on_closed(UNUSED fr_connection_t *conn, UNUSED fr_ * removed from the active, pool * re-enqueue the requests. */ - if (need_requeue) trunk_connection_requests_requeue(tconn, FR_TRUNK_REQUEST_ALL); + if (need_requeue) trunk_connection_requests_requeue(tconn, FR_TRUNK_REQUEST_ALL, 0); /* * There should be no requests left on this @@ -2239,7 +2284,7 @@ static void _trunk_connection_on_halted(UNUSED fr_connection_t *conn, UNUSED fr_ */ CONN_STATE_TRANSITION(FR_TRUNK_CONN_HALTED); - if (need_requeue) trunk_connection_requests_requeue(tconn, FR_TRUNK_REQUEST_ALL); + if (need_requeue) trunk_connection_requests_requeue(tconn, FR_TRUNK_REQUEST_ALL, 0); /* * There should be no requests left on this @@ -2279,7 +2324,7 @@ static int _trunk_connection_free(fr_trunk_connection_t *tconn) /* * Remove requests from this connection */ - trunk_connection_requests_dequeue(&to_fail, tconn, FR_TRUNK_REQUEST_ALL); + trunk_connection_requests_dequeue(&to_fail, tconn, FR_TRUNK_REQUEST_ALL, 0); while ((treq = fr_dlist_next(&to_fail, treq))) { fr_trunk_request_t *prev; @@ -2529,6 +2574,43 @@ void fr_trunk_connection_signal_reconnect(fr_trunk_connection_t *tconn) fr_connection_signal_reconnect(tconn->conn); } +/** Rebalance connections across active trunk members when a new connection becomes active + * + * We don't have any visibility into the connection prioritisation algorithm + * it's essentially a black box. + * + * We can however determine when the correct level of requests per connection + * has been reached, by dequeuing and requeing requests up until the point + * where the connection that just had a request dequeued, receives the same + * request back. + * + * + * @param[in] trunk The trunk to rebalance. + */ +static void trunk_rebalance(fr_trunk_t *trunk) +{ + fr_trunk_connection_t *head; + + head = fr_heap_peek(trunk->active); + + /* + * Only rebalance if the top and bottom of + * the heap are not equal. + */ + if (trunk->funcs.connection_prioritise(fr_heap_peek_tail(trunk->active), head) == 0) return; + + DEBUG4("Rebalancing requests"); + + /* + * Keep requeuing requests from the connection + * at the bottom of the heap until the + * connection at the top if shifted from that + * position. + */ + while ((fr_heap_peek(trunk->active) == head) && + trunk_connection_requests_requeue(fr_heap_peek_tail(trunk->active), FR_TRUNK_REQUEST_PENDING, 1)); +} + /** Implements the algorithm we use to manage requests per connection levels * * This is executed periodically using a timer event, and opens/closes @@ -3076,6 +3158,9 @@ fr_trunk_t *fr_trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, char const *log trunk->conf = conf; memcpy(&trunk->funcs, funcs, sizeof(trunk->funcs)); + if (!trunk->funcs.connection_prioritise) { + trunk->funcs.connection_prioritise = _trunk_connection_order_by_shortest_queue; + } memcpy(&trunk->uctx, &uctx, sizeof(trunk->uctx)); talloc_set_destructor(trunk, _trunk_free); @@ -3088,7 +3173,7 @@ fr_trunk_t *fr_trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, char const *log /* * Connection queues and trees */ - MEM(trunk->active = fr_heap_talloc_create(trunk, _trunk_connection_order_by_shortest_queue, + MEM(trunk->active = fr_heap_talloc_create(trunk, trunk->funcs.connection_prioritise, fr_trunk_connection_t, heap_id)); fr_dlist_talloc_init(&trunk->connecting, fr_trunk_connection_t, list); fr_dlist_talloc_init(&trunk->full, fr_trunk_connection_t, list); diff --git a/src/lib/server/trunk.h b/src/lib/server/trunk.h index 554fba0b452..679e0d3ac6b 100644 --- a/src/lib/server/trunk.h +++ b/src/lib/server/trunk.h @@ -332,7 +332,8 @@ typedef struct { fr_trunk_connection_alloc_t connection_alloc; //!< Allocate a new fr_connection_t. fr_trunk_connection_notify_t connection_notify; //!< Update the I/O event registrations for - ///< a connection. + + fr_heap_cmp_t connection_prioritise; //!< Ordering function for connections. fr_heap_cmp_t request_prioritise; //!< Ordering function for requests. Controls ///< where in the outbound queues they're inserted. @@ -389,7 +390,7 @@ void fr_trunk_request_signal_cancel_complete(fr_trunk_request_t *treq); /** @name (R)enqueue requests * @{ */ -uint64_t fr_trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, int states); +uint64_t fr_trunk_connection_requests_requeue(fr_trunk_connection_t *tconn, int states, uint64_t max); int fr_trunk_request_enqueue(fr_trunk_request_t **treq, fr_trunk_t *trunk, REQUEST *request, void *preq, void *rctx); diff --git a/src/lib/server/trunk_tests.c b/src/lib/server/trunk_tests.c index 4583f8ff0f3..30a4745475b 100644 --- a/src/lib/server/trunk_tests.c +++ b/src/lib/server/trunk_tests.c @@ -189,7 +189,7 @@ static void _conn_io_loopback(fr_event_list_t *el, int fd, int flags, void *uctx rad_assert(fd == our_h[1]); slen = read(fd, buff, sizeof(buff)); - printf("Received %zu bytes of data, sending it back\n", slen); + if (test_verbose_level__ >= 3) printf("Received %zu bytes of data, sending it back\n", slen); write(our_h[1], buff, (size_t)slen); } @@ -634,9 +634,7 @@ static void test_enqueue_cancellation_points(void) preq = talloc_zero(NULL, test_proto_request_t); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); - /* - * Cancellation in backlog via trunk free - */ + TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_BACKLOG"); talloc_free(trunk); TEST_CHECK(preq->completed == false); TEST_CHECK(preq->failed == true); @@ -644,9 +642,7 @@ static void test_enqueue_cancellation_points(void) TEST_CHECK(preq->freed == true); talloc_free(preq); - /* - * Cancellation in backlog via signal - */ + TEST_CASE("cancellation via signal - FR_TRUNK_REQUEST_BACKLOG"); trunk = test_setup_trunk(ctx, el, &conf, false); preq = talloc_zero(NULL, test_proto_request_t); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); @@ -661,9 +657,7 @@ static void test_enqueue_cancellation_points(void) talloc_free(preq); talloc_free(trunk); - /* - * Cancellation in partial via trunk free - */ + TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_PARTIAL"); trunk = test_setup_trunk(ctx, el, &conf, false); preq = talloc_zero(NULL, test_proto_request_t); preq->signal_partial = true; @@ -686,9 +680,7 @@ static void test_enqueue_cancellation_points(void) TEST_CHECK(preq->freed == true); talloc_free(preq); - /* - * Cancellation in partial via signal - */ + TEST_CASE("cancellation via signal - FR_TRUNK_REQUEST_PARTIAL"); trunk = test_setup_trunk(ctx, el, &conf, false); preq = talloc_zero(NULL, test_proto_request_t); preq->signal_partial = true; @@ -712,9 +704,7 @@ static void test_enqueue_cancellation_points(void) talloc_free(preq); talloc_free(trunk); - /* - * Cancellation in sent via trunk free - */ + TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_SENT"); trunk = test_setup_trunk(ctx, el, &conf, false); preq = talloc_zero(NULL, test_proto_request_t); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); @@ -735,9 +725,7 @@ static void test_enqueue_cancellation_points(void) TEST_CHECK(preq->freed == true); talloc_free(preq); - /* - * Cancellation in sent via signal - */ + TEST_CASE("cancellation via signal - FR_TRUNK_REQUEST_SENT"); trunk = test_setup_trunk(ctx, el, &conf, false); preq = talloc_zero(NULL, test_proto_request_t); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); @@ -760,9 +748,7 @@ static void test_enqueue_cancellation_points(void) talloc_free(preq); talloc_free(trunk); - /* - * Cancellation in cancel-partial via trunk free - */ + TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_CANCEL_PARTIAL"); trunk = test_setup_trunk(ctx, el, &conf, true); preq = talloc_zero(NULL, test_proto_request_t); preq->signal_cancel_partial = true; @@ -792,9 +778,7 @@ static void test_enqueue_cancellation_points(void) TEST_CHECK(preq->freed == true); talloc_free(preq); - /* - * Cancellation in cancel-sent via trunk free - */ + TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_CANCEL_SENT"); trunk = test_setup_trunk(ctx, el, &conf, true); preq = talloc_zero(NULL, test_proto_request_t); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); @@ -823,9 +807,7 @@ static void test_enqueue_cancellation_points(void) TEST_CHECK(preq->freed == true); talloc_free(preq); - /* - * Cancellation after cancel-complete via trunk free - */ + TEST_CASE("trunk free after FR_TRUNK_REQUEST_CANCEL_COMPLETE"); trunk = test_setup_trunk(ctx, el, &conf, true); preq = talloc_zero(NULL, test_proto_request_t); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); @@ -895,6 +877,8 @@ static void test_partial_to_complete_states(void) preq->signal_partial = true; preq->signal_cancel_partial = true; + TEST_CASE("FR_TRUNK_REQUEST_PARTIAL -> FR_TRUNK_REQUEST_SENT"); + fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); preq->treq = treq; @@ -914,6 +898,8 @@ static void test_partial_to_complete_states(void) fr_trunk_request_signal_cancel(treq); TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_CANCEL) == 1); + TEST_CASE("FR_TRUNK_REQUEST_CANCEL_PARTIAL -> FR_TRUNK_REQUEST_CANCEL_SENT"); + events = fr_event_corral(el, test_time_base, false); /* Send partial cancel request */ fr_event_service(el); @@ -973,6 +959,8 @@ static void test_requeue_on_reconnect(void) events = fr_event_corral(el, test_time_base, false); /* Connect the connection(s) */ fr_event_service(el); + TEST_CASE("dequeue on reconnect - FR_TRUNK_REQUEST_PENDING"); + TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 2); fr_trunk_request_enqueue(&treq, trunk, request, preq, NULL); @@ -996,6 +984,8 @@ static void test_requeue_on_reconnect(void) TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_BACKLOG) == 1); TEST_CHECK(!treq->tconn); + TEST_CASE("cancel on reconnect - FR_TRUNK_REQUEST_PARTIAL"); + /* * Allow the connections to reconnect */ @@ -1033,6 +1023,8 @@ static void test_requeue_on_reconnect(void) TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_PENDING) == 1); TEST_CHECK(tconn != treq->tconn); /* Ensure it moved */ + TEST_CASE("cancel on reconnect - FR_TRUNK_REQUEST_SENT"); + /* * Sent the request (fully) */ @@ -1061,6 +1053,8 @@ static void test_requeue_on_reconnect(void) preq->cancelled = false; /* Reset */ + TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_CANCEL"); + /* * Signal the request should be cancelled */ @@ -1081,6 +1075,16 @@ static void test_requeue_on_reconnect(void) talloc_free(preq); + /* + * Allow the connection we just reconnected + * top open so it doesn't interfere with + * the next test. + */ + events = fr_event_corral(el, test_time_base, false); + fr_event_service(el); + + TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_CANCEL_PARTIAL"); + /* * Queue up a new request, and get it to the cancel-partial state. */ @@ -1122,6 +1126,16 @@ static void test_requeue_on_reconnect(void) talloc_free(preq); + /* + * Allow the connection we just reconnected + * top open so it doesn't interfere with + * the next test. + */ + events = fr_event_corral(el, test_time_base, false); + fr_event_service(el); + + TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_CANCEL_SENT"); + /* * Queue up a new request, and get it to the cancel-sent state. */ @@ -1191,16 +1205,12 @@ static void test_connection_start_on_enqueue(void) trunk = test_setup_trunk(ctx, el, &conf, true); preq = talloc_zero(NULL, test_proto_request_t); - /* - * Queuing a request should start a connection. - */ + TEST_CASE("C0 - Enqueue should spawn"); fr_trunk_request_enqueue(&treq_a, trunk, request, preq, NULL); TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1); - /* - * Queuing another request should *NOT* start another connection - */ + TEST_CASE("C1 connecting, !max_requests_per_conn - Enqueue MUST NOT spawn"); fr_trunk_request_enqueue(&treq_b, trunk, request, preq, NULL); TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1); @@ -1213,13 +1223,69 @@ static void test_connection_start_on_enqueue(void) TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1); + TEST_CASE("C1 active, !max_requests_per_conn - Enqueue MUST NOT spawn"); + fr_trunk_request_enqueue(&treq_c, trunk, request, preq, NULL); + + TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1); + TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_PENDING) == 3); + + talloc_free(ctx); +} + +static void test_connection_rebalance_requests(void) +{ + TALLOC_CTX *ctx = talloc_init("test"); + fr_trunk_t *trunk; + fr_event_list_t *el; + int events; + fr_trunk_conf_t conf = { + .min_connections = 2, /* No connections on start */ + .manage_interval = NSEC * 0.5 + }; + test_proto_request_t *preq; + fr_trunk_connection_t *tconn; + fr_trunk_request_t *treq_a, *treq_b, *treq_c; + REQUEST *request; + + DEBUG_LVL_SET; + + el = fr_event_list_alloc(ctx, NULL, NULL); + fr_event_list_set_time_func(el, test_time); + + request = request_alloc(ctx); + + trunk = test_setup_trunk(ctx, el, &conf, true); + preq = talloc_zero(NULL, test_proto_request_t); + /* - * Queuing another request should *NOT* start another connection + * Allow the connections to open */ + events = fr_event_corral(el, test_time_base, false); + fr_event_service(el); + + /* + * Mark one of the connections as full, and + * enqueue three requests on the other. + */ + tconn = fr_heap_peek(trunk->active); + fr_trunk_connection_signal_inactive(tconn); + + fr_trunk_request_enqueue(&treq_a, trunk, request, preq, NULL); + fr_trunk_request_enqueue(&treq_b, trunk, request, preq, NULL); fr_trunk_request_enqueue(&treq_c, trunk, request, preq, NULL); - TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1); TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_PENDING) == 3); + TEST_CHECK(fr_trunk_request_count_by_connection(tconn, FR_TRUNK_REQUEST_ALL) == 0); + + /* + * Now mark the previous connection as + * active. It should receive at least + * one of the requests. + */ + fr_trunk_connection_signal_active(tconn); + + TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_PENDING) == 3); + TEST_CHECK(fr_trunk_request_count_by_connection(tconn, FR_TRUNK_REQUEST_ALL) >= 1); talloc_free(ctx); } @@ -1256,12 +1322,14 @@ static void test_connection_levels(void) /* * Queuing a request should start a connection. */ + TEST_CASE("C0 - Enqueue should spawn"); TEST_CHECK(fr_trunk_request_enqueue(&treq_a, trunk, request, preq, NULL) == TRUNK_ENQUEUE_IN_BACKLOG); TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1); /* * Queuing another request should *NOT* start another connection */ + TEST_CASE("C1 connecting, max_requests_per_conn 2 - Enqueue MUST NOT spawn"); TEST_CHECK(fr_trunk_request_enqueue(&treq_b, trunk, request, preq, NULL) == TRUNK_ENQUEUE_IN_BACKLOG); TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1); @@ -1327,6 +1395,10 @@ TEST_LIST = { { "Enqueue - Partial state transitions", test_partial_to_complete_states }, { "Requeue - On reconnect", test_requeue_on_reconnect }, + /* + * Rebalance + */ + { "Rebalance - Connection rebalance", test_connection_rebalance_requests }, /* * Connection spawning tests */