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);
* 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); \
}
* ....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);
*/
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);
}
* 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);
*/
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
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:
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
* 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;
*
* @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;
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
* 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
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
* 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
*/
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
/*
* 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;
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
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);
/*
* 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);
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);
}
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);
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);
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;
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;
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);
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);
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;
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);
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);
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;
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);
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);
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
*/
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)
*/
preq->cancelled = false; /* Reset */
+ TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_CANCEL");
+
/*
* Signal the request should be cancelled
*/
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.
*/
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.
*/
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);
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);
}
/*
* 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);
{ "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
*/
projects / thirdparty / freeradius-server.git / commitdiff
