/*
 *   This program is is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or (at
 *   your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */

/**
 * $Id: 71e25681e7b29de253e73c326fcfef05cfeb1b2b $
 *
 * @file src/lib/server/trunk.c
 * @brief A management API for bonding multiple connections together.
 *
 * @copyright 2019-2020 Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 * @copyright 2019-2020 The FreeRADIUS server project
 */

#define LOG_PREFIX trunk->log_prefix

#ifdef NDEBUG
#  define TALLOC_GET_TYPE_ABORT_NOOP 1
#endif

typedef struct trunk_request_s trunk_request_t;
typedef struct trunk_connection_s trunk_connection_t;
typedef struct trunk_s trunk_t;
#define _TRUNK_PRIVATE 1
#include <freeradius-devel/server/trunk.h>

#include <freeradius-devel/server/connection.h>
#include <freeradius-devel/server/trigger.h>
#include <freeradius-devel/util/misc.h>
#include <freeradius-devel/util/syserror.h>
#include <freeradius-devel/util/table.h>
#include <freeradius-devel/util/minmax_heap.h>

#ifdef HAVE_STDATOMIC_H
#  include <stdatomic.h>
#  ifndef ATOMIC_VAR_INIT
#    define ATOMIC_VAR_INIT(_x) (_x)
#  endif
#else
#  include <freeradius-devel/util/stdatomic.h>
#endif

static atomic_uint_fast64_t request_counter = ATOMIC_VAR_INIT(1);

#ifdef TESTING_TRUNK
static fr_time_t test_time_base = fr_time_wrap(1);

static fr_time_t test_time(void)
{
	return test_time_base;
}

#define fr_time test_time
#endif

#ifndef NDEBUG
/** The maximum number of state logs to record per request
 *
 */
#define TRUNK_REQUEST_STATE_LOG_MAX	20

/** Trace state machine changes for a particular request
 *
 */
typedef struct {
	fr_dlist_head_t			*log_head;	//!< To allow the log entry to remove itself on free.
	fr_dlist_t			entry;		//!< Entry in the linked list.
	trunk_request_state_t	from;		//!< What state we transitioned from.
	trunk_request_state_t	to;		//!< What state we transitioned to.

	trunk_connection_t		*tconn;		//!< The request was associated with.
							///< Pointer may now be invalid, do no de-reference.

	uint64_t			tconn_id;	//!< If the treq was associated with a connection
							///< the connection ID.
	trunk_connection_state_t	tconn_state;	//!< If the treq was associated with a connection
							///< the connection state at the time of the
							///< state transition.

	char const		        *function;	//!< State change occurred in.
	int				line;		//!< Line change occurred on.
} trunk_request_state_log_t;
#endif

/** Wraps a normal request
 *
 */
struct trunk_request_s {
	struct trunk_request_pub_s	pub;		//!< Public fields in the trunk request.
							///< This *MUST* be the first field in this
							///< structure.

	uint64_t 		id;			//!< Trunk request ID.

	fr_heap_index_t		heap_id;		//!< Used to track the request conn->pending heap.

	fr_dlist_t		entry;			//!< Used to track the trunk request in the conn->sent
							///< or trunk->backlog request.

	trunk_cancel_reason_t cancel_reason;		//!< Why this request was cancelled.

	fr_time_t		last_freed;		//!< Last time this request was freed.

	bool			bound_to_conn;		//!< Fail the request if there's an attempt to
							///< re-enqueue it.

	bool			sent;			//!< Trunk request has been sent at least once.
							///< Used so that re-queueing doesn't increase trunk
							///< `sent` count.

#ifndef NDEBUG
	fr_dlist_head_t		log;			//!< State change log.
#endif
};


/** Associates request queues with a connection
 *
 * @dotfile src/lib/server/trunk_conn.gv "Trunk connection state machine"
 * @dotfile src/lib/server/trunk_req.gv "Trunk request state machine"
 */
struct trunk_connection_s {
	struct trunk_connection_pub_s pub;		//!< Public fields in the trunk connection.
							///< This *MUST* be the first field in this
							///< structure.

	fr_heap_index_t		heap_id;		//!< Used to track the connection in the connected
							///< heap.

	fr_dlist_t		entry;			//!< Used to track the connection in the connecting,
							///< full and failed lists.

	/** @name State
	 * @{
 	 */
	trunk_connection_event_t events;		//!< The current events we expect to be notified on.
	/** @} */

	/** @name Request lists
	 * @{
 	 */
	fr_heap_t		*pending;		//!< Requests waiting to be sent.

	trunk_request_t	*partial;		//!< Partially written request.

	fr_dlist_head_t		sent;			//!< Sent request.

	fr_dlist_head_t		reapable;			//!< Idle request.

	fr_dlist_head_t		cancel;			//!< Requests in the cancel state.

	trunk_request_t	*cancel_partial;	//!< Partially written cancellation request.

	fr_dlist_head_t		cancel_sent;		//!< Sent cancellation request.
	/** @} */

	/** @name Statistics
	 * @{
 	 */
 	uint64_t		sent_count;		//!< The number of requests that have been sent using
 							///< this connection.
 	/** @} */

	/** @name Timers
	 * @{
 	 */
  	fr_event_timer_t const	*lifetime_ev;		//!< Maximum time this connection can be open.
  	/** @} */
};

/** An entry in a trunk watch function list
 *
 */
typedef struct trunk_watch_entry_s {
	fr_dlist_t		entry;			//!< List entry.
	trunk_watch_t	func;			//!< Function to call when a trunk enters
							///< the state this list belongs to
	bool			oneshot;		//!< Remove the function after it's called once.
	bool			enabled;		//!< Whether the watch entry is enabled.
	void			*uctx;			//!< User data to pass to the function.
} trunk_watch_entry_t;

/** Main trunk management handle
 *
 */
struct trunk_s {
	struct trunk_pub_s	pub;			//!< Public fields in the trunk connection.
							///< This *MUST* be the first field in this
							///< structure.

	char const		*log_prefix;		//!< What to prepend to messages.

	fr_event_list_t		*el;			//!< Event list used by this trunk and the connection.

	trunk_conf_t		conf;			//!< Trunk common configuration.

	fr_dlist_head_t		free_requests;		//!< Requests in the unassigned state.  Waiting to be
							///< enqueued.

	fr_heap_t		*backlog;		//!< The request backlog.  Requests we couldn't
							///< immediately assign to a connection.

	/** @name Connection lists
	 *
	 * A connection must always be in exactly one of these lists
	 * or trees.
	 *
	 * @{
 	 */
 	fr_dlist_head_t		init;			//!< Connections which have not yet started
 							///< connecting.

 	fr_dlist_head_t		connecting;		//!< Connections which are not yet in the open state.

	fr_minmax_heap_t	*active;		//!< Connections which can service requests.

	fr_dlist_head_t		full;			//!< Connections which have too many outstanding
							///< requests.

	fr_dlist_head_t		inactive;		//!< Connections which have been signalled to be
							///< inactive by the API client.

	fr_dlist_head_t		inactive_draining;	//!< Connections which have been signalled to be
							///< inactive by the API client, which the trunk
							///< manager is draining to close.

	fr_dlist_head_t		failed;			//!< Connections that'll be reconnected shortly.

	fr_dlist_head_t		closed;			//!< Connections that have closed. Either due to
							///< shutdown, reconnection or failure.

	fr_dlist_head_t		draining;		//!< Connections that will be freed once all their
							///< requests are complete, but can be reactivated.

	fr_dlist_head_t		draining_to_free;	//!< Connections that will be freed once all their
							///< requests are complete.

	fr_dlist_head_t		to_free;		//!< Connections we're done with and will free on
							//!< the next call to trunk_manage.
							//!< This prevents connections from being freed
							//!< whilst we're inside callbacks.
	/** @} */

	/** @name Callbacks
	 * @{
 	 */
	trunk_io_funcs_t	funcs;			//!< I/O functions.

	void			*in_handler;		//!< Which handler we're inside.

	void			*uctx;			//!< Uctx data to pass to alloc.

	fr_dlist_head_t		watch[TRUNK_STATE_MAX];	//!< To be called when trunk changes state.

	trunk_watch_entry_t	*next_watcher;		//!< Watcher about to be run. Used to prevent nested watchers.
	/** @} */

	/** @name Timers
	 * @{
 	 */
 	fr_event_timer_t const	*manage_ev;		//!< Periodic connection management event.
	/** @} */

	/** @name Log rate limiting entries
	 * @{
 	 */
	fr_rate_limit_t		limit_max_requests_alloc_log;	//!< Rate limit on "Refusing to alloc requests - Limit of * requests reached"

	fr_rate_limit_t		limit_last_failure_log;	//!< Rate limit on "Refusing to enqueue requests - No active conns"
 	/** @} */

	/** @name State
	 * @{
 	 */
	bool			freeing;		//!< Trunk is being freed, don't spawn new
							///< connections or re-enqueue.

	bool			started;		//!< Has the trunk been started.

	bool			managing_connections;	//!< Whether the trunk is allowed to manage
							///< (open/close) connections.

	uint64_t		last_req_per_conn;	//!< The last request to connection ratio we calculated.
	/** @} */
};

static conf_parser_t const trunk_config_request[] = {
	{ FR_CONF_OFFSET("per_connection_max", trunk_conf_t, max_req_per_conn), .dflt = "2000" },
	{ FR_CONF_OFFSET("per_connection_target", trunk_conf_t, target_req_per_conn), .dflt = "1000" },
	{ FR_CONF_OFFSET("free_delay", trunk_conf_t, req_cleanup_delay), .dflt = "10.0" },

	CONF_PARSER_TERMINATOR
};

static conf_parser_t const trunk_config_connection[] = {
	{ FR_CONF_OFFSET("connect_timeout", connection_conf_t, connection_timeout), .dflt = "3.0" },
	{ FR_CONF_OFFSET("reconnect_delay", connection_conf_t, reconnection_delay), .dflt = "1" },

	CONF_PARSER_TERMINATOR
};

#ifndef TRUNK_TESTS
conf_parser_t const trunk_config[] = {
	{ FR_CONF_OFFSET("start", trunk_conf_t, start), .dflt = "1" },
	{ FR_CONF_OFFSET("min", trunk_conf_t, min), .dflt = "1" },
	{ FR_CONF_OFFSET("max", trunk_conf_t, max), .dflt = "5" },
	{ FR_CONF_OFFSET("connecting", trunk_conf_t, connecting), .dflt = "2" },
	{ FR_CONF_OFFSET("uses", trunk_conf_t, max_uses), .dflt = "0" },
	{ FR_CONF_OFFSET("lifetime", trunk_conf_t, lifetime), .dflt = "0" },

	{ FR_CONF_OFFSET("open_delay", trunk_conf_t, open_delay), .dflt = "0.2" },
	{ FR_CONF_OFFSET("close_delay", trunk_conf_t, close_delay), .dflt = "10.0" },

	{ FR_CONF_OFFSET("manage_interval", trunk_conf_t, manage_interval), .dflt = "0.2" },

	{ FR_CONF_OFFSET("max_backlog", trunk_conf_t, max_backlog), .dflt = "1000" },

	{ FR_CONF_OFFSET_SUBSECTION("connection", 0, trunk_conf_t, conn_conf, trunk_config_connection), .subcs_size = sizeof(trunk_config_connection) },
	{ FR_CONF_POINTER("request", 0, CONF_FLAG_SUBSECTION, NULL), .subcs = (void const *) trunk_config_request },

	CONF_PARSER_TERMINATOR
};
#endif

#ifndef NDEBUG
/** Map request states to trigger names
 *
 * Must stay in the same order as #trunk_connection_state_t
 */
static fr_table_num_indexed_bit_pos_t const trunk_req_trigger_names[] = {
	{ L("pool.request_init"),			TRUNK_REQUEST_STATE_INIT		},	/* 0x0000 - bit 0 */
	{ L("pool.request_unassigned"),			TRUNK_REQUEST_STATE_UNASSIGNED		},	/* 0x0001 - bit 1 */
	{ L("pool.request_backlog"),			TRUNK_REQUEST_STATE_BACKLOG		},	/* 0x0002 - bit 2 */
	{ L("pool.request_pending"),			TRUNK_REQUEST_STATE_PENDING		},	/* 0x0004 - bit 3 */
	{ L("pool.request_partial"),			TRUNK_REQUEST_STATE_PARTIAL		},	/* 0x0008 - bit 4 */
	{ L("pool.request_sent"),			TRUNK_REQUEST_STATE_SENT		},	/* 0x0010 - bit 5 */
	{ L("pool.request_state_reapable"),		TRUNK_REQUEST_STATE_REAPABLE		},	/* 0x0020 - bit 6 */
	{ L("pool.request_complete"),			TRUNK_REQUEST_STATE_COMPLETE		},	/* 0x0040 - bit 7 */
	{ L("pool.request_state_failed"),		TRUNK_REQUEST_STATE_FAILED		},	/* 0x0080 - bit 8 */
	{ L("pool.request_state_cancel"),		TRUNK_REQUEST_STATE_CANCEL		},	/* 0x0100 - bit 9 */
	{ L("pool.request_state_cancel_sent"),		TRUNK_REQUEST_STATE_CANCEL_SENT		},	/* 0x0200 - bit 10 */
	{ L("pool.request_state_cancel_partial"),	TRUNK_REQUEST_STATE_CANCEL_PARTIAL	},	/* 0x0400 - bit 11 */
	{ L("pool.request_state_cancel_complete"),	TRUNK_REQUEST_STATE_CANCEL_COMPLETE	},	/* 0x0800 - bit 12 */
};
static size_t trunk_req_trigger_names_len = NUM_ELEMENTS(trunk_req_trigger_names);
#endif

static fr_table_num_ordered_t const trunk_request_states[] = {
	{ L("INIT"),					TRUNK_REQUEST_STATE_INIT		},
	{ L("UNASSIGNED"),				TRUNK_REQUEST_STATE_UNASSIGNED		},
	{ L("BACKLOG"),					TRUNK_REQUEST_STATE_BACKLOG		},
	{ L("PENDING"),					TRUNK_REQUEST_STATE_PENDING		},
	{ L("PARTIAL"),					TRUNK_REQUEST_STATE_PARTIAL		},
	{ L("SENT"),					TRUNK_REQUEST_STATE_SENT		},
	{ L("REAPABLE"),				TRUNK_REQUEST_STATE_REAPABLE		},
	{ L("COMPLETE"),				TRUNK_REQUEST_STATE_COMPLETE		},
	{ L("FAILED"),					TRUNK_REQUEST_STATE_FAILED		},
	{ L("CANCEL"),					TRUNK_REQUEST_STATE_CANCEL		},
	{ L("CANCEL-SENT"),				TRUNK_REQUEST_STATE_CANCEL_SENT		},
	{ L("CANCEL-PARTIAL"),				TRUNK_REQUEST_STATE_CANCEL_PARTIAL	},
	{ L("CANCEL-COMPLETE"),				TRUNK_REQUEST_STATE_CANCEL_COMPLETE	}
};
static size_t trunk_request_states_len = NUM_ELEMENTS(trunk_request_states);

/** Map connection states to trigger names
 *
 * Must stay in the same order as #trunk_connection_state_t
 */
static fr_table_num_indexed_bit_pos_t const trunk_conn_trigger_names[] = {
	{ L("pool.connection_halted"),			TRUNK_CONN_HALTED			},	/* 0x0000 - bit 0 */
	{ L("pool.connection_init"),			TRUNK_CONN_INIT				},	/* 0x0001 - bit 1 */
	{ L("pool.connection_connecting"),		TRUNK_CONN_CONNECTING			},	/* 0x0002 - bit 2 */
	{ L("pool.connection_active"),			TRUNK_CONN_ACTIVE			},	/* 0x0004 - bit 3 */
	{ L("pool.connection_closed"),			TRUNK_CONN_CLOSED			},	/* 0x0008 - bit 4 */
	{ L("pool.connection_full"),			TRUNK_CONN_FULL				},	/* 0x0010 - bit 5 */
	{ L("pool.connection_inactive"),		TRUNK_CONN_INACTIVE			},	/* 0x0020 - bit 6 */
	{ L("pool.connection_inactive_draining"),	TRUNK_CONN_INACTIVE_DRAINING		},	/* 0x0040 - bit 7 */
	{ L("pool.connection_draining"),		TRUNK_CONN_DRAINING			},	/* 0x0080 - bit 8 */
	{ L("pool.connection_draining_to_free"),	TRUNK_CONN_DRAINING_TO_FREE		}	/* 0x0100 - bit 9 */
};
static size_t trunk_conn_trigger_names_len = NUM_ELEMENTS(trunk_conn_trigger_names);

static fr_table_num_ordered_t const trunk_states[] = {
	{ L("IDLE"),					TRUNK_STATE_IDLE			},
	{ L("ACTIVE"),					TRUNK_STATE_ACTIVE			},
	{ L("PENDING"),					TRUNK_STATE_PENDING			}
};
static size_t trunk_states_len = NUM_ELEMENTS(trunk_states);

static fr_table_num_ordered_t const trunk_connection_states[] = {
	{ L("INIT"),					TRUNK_CONN_INIT				},
	{ L("HALTED"),					TRUNK_CONN_HALTED			},
	{ L("CONNECTING"),				TRUNK_CONN_CONNECTING			},
	{ L("ACTIVE"),					TRUNK_CONN_ACTIVE			},
	{ L("CLOSED"),					TRUNK_CONN_CLOSED			},
	{ L("FULL"),					TRUNK_CONN_FULL				},
	{ L("INACTIVE"),				TRUNK_CONN_INACTIVE			},
	{ L("INACTIVE-DRAINING"),			TRUNK_CONN_INACTIVE_DRAINING		},
	{ L("DRAINING"),				TRUNK_CONN_DRAINING			},
	{ L("DRAINING-TO-FREE"),			TRUNK_CONN_DRAINING_TO_FREE		}
};
static size_t trunk_connection_states_len = NUM_ELEMENTS(trunk_connection_states);

static fr_table_num_ordered_t const trunk_cancellation_reasons[] = {
	{ L("TRUNK_CANCEL_REASON_NONE"),		TRUNK_CANCEL_REASON_NONE		},
	{ L("TRUNK_CANCEL_REASON_SIGNAL"),		TRUNK_CANCEL_REASON_SIGNAL		},
	{ L("TRUNK_CANCEL_REASON_MOVE"),		TRUNK_CANCEL_REASON_MOVE		},
	{ L("TRUNK_CANCEL_REASON_REQUEUE"),		TRUNK_CANCEL_REASON_REQUEUE		}
};
static size_t trunk_cancellation_reasons_len = NUM_ELEMENTS(trunk_cancellation_reasons);

static fr_table_num_ordered_t const trunk_connection_events[] = {
	{ L("TRUNK_CONN_EVENT_NONE"),		TRUNK_CONN_EVENT_NONE 		},
	{ L("TRUNK_CONN_EVENT_READ"),		TRUNK_CONN_EVENT_READ		},
	{ L("TRUNK_CONN_EVENT_WRITE"),		TRUNK_CONN_EVENT_WRITE		},
	{ L("TRUNK_CONN_EVENT_BOTH"),		TRUNK_CONN_EVENT_BOTH		},
};
static size_t trunk_connection_events_len = NUM_ELEMENTS(trunk_connection_events);

#define CONN_TRIGGER(_state) do { \
	if (trunk->pub.triggers) { \
		trigger_exec(unlang_interpret_get_thread_default(), \
			     NULL, fr_table_str_by_value(trunk_conn_trigger_names, _state, \
							 "<INVALID>"), true, NULL); \
	} \
} while (0)

#define CONN_STATE_TRANSITION(_new, _log) \
do { \
	_log("[%" PRIu64 "] Trunk connection changed state %s -> %s", \
	     tconn->pub.conn->id, \
	     fr_table_str_by_value(trunk_connection_states, tconn->pub.state, "<INVALID>"), \
	     fr_table_str_by_value(trunk_connection_states, _new, "<INVALID>")); \
	tconn->pub.state = _new; \
	CONN_TRIGGER(_new); \
	trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false); \
} while (0)

#define CONN_BAD_STATE_TRANSITION(_new) \
do { \
	if (!fr_cond_assert_msg(0, "[%" PRIu64 "] Trunk connection invalid transition %s -> %s", \
				tconn->pub.conn->id, \
				fr_table_str_by_value(trunk_connection_states, tconn->pub.state, "<INVALID>"),	\
				fr_table_str_by_value(trunk_connection_states, _new, "<INVALID>"))) return;	\
} while (0)

#ifndef NDEBUG
void trunk_request_state_log_entry_add(char const *function, int line,
				       trunk_request_t *treq, trunk_request_state_t new) CC_HINT(nonnull);

#define REQUEST_TRIGGER(_state) do { \
	if (trunk->pub.triggers) { \
		trigger_exec(unlang_interpret_get_thread_default(), \
			     NULL, fr_table_str_by_value(trunk_req_trigger_names, _state, \
							 "<INVALID>"), true, NULL); \
	} \
} while (0)

/** Record a request state transition and log appropriate output
 *
 */
#define REQUEST_STATE_TRANSITION(_new) \
do { \
	request_t *request = treq->pub.request; \
	ROPTIONAL(RDEBUG3, DEBUG3, "Trunk request %" PRIu64 " changed state %s -> %s", \
		  treq->id, \
		  fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
		  fr_table_str_by_value(trunk_request_states, _new, "<INVALID>")); \
	trunk_request_state_log_entry_add(__FUNCTION__, __LINE__, treq, _new); \
	treq->pub.state = _new; \
	REQUEST_TRIGGER(_new); \
} while (0)
#define REQUEST_BAD_STATE_TRANSITION(_new) \
do { \
	trunk_request_state_log(&default_log, L_ERR, __FILE__, __LINE__, treq); \
	if (!fr_cond_assert_msg(0, "Trunk request %" PRIu64 " invalid transition %s -> %s", \
				treq->id, \
				fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
				fr_table_str_by_value(trunk_request_states, _new, "<INVALID>"))) return; \
} while (0)
#else
/** Record a request state transition
 *
 */
#define REQUEST_STATE_TRANSITION(_new) \
do { \
	request_t *request = treq->pub.request; \
	ROPTIONAL(RDEBUG3, DEBUG3, "Trunk request %" PRIu64 " changed state %s -> %s", \
		  treq->id, \
		  fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
		  fr_table_str_by_value(trunk_request_states, _new, "<INVALID>")); \
	treq->pub.state = _new; \
} while (0)
#define REQUEST_BAD_STATE_TRANSITION(_new) \
do { \
	if (!fr_cond_assert_msg(0, "Trunk request %" PRIu64 " invalid transition %s -> %s", \
				treq->id, \
				fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
				fr_table_str_by_value(trunk_request_states, _new, "<INVALID>"))) return; \
} while (0)
#endif


/** Call the cancel callback if set
 *
 */
#define DO_REQUEST_CANCEL(_treq, _reason) \
do { \
	if ((_treq)->pub.trunk->funcs.request_cancel) { \
		request_t *request = (_treq)->pub.request; \
		void *_prev = (_treq)->pub.trunk->in_handler; \
		(_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_cancel; \
		ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_cancel(conn=%p, preq=%p, reason=%s, uctx=%p)", \
			  (_treq)->pub.tconn->pub.conn, \
			  (_treq)->pub.preq, \
			  fr_table_str_by_value(trunk_cancellation_reasons, \
			  (_reason), \
			  "<INVALID>"), \
			  (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->funcs.request_cancel((_treq)->pub.tconn->pub.conn, (_treq)->pub.preq, (_reason), (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

/** Call the "conn_release" callback (if set)
 *
 */
#define DO_REQUEST_CONN_RELEASE(_treq) \
do { \
	if ((_treq)->pub.trunk->funcs.request_conn_release) { \
		request_t *request = (_treq)->pub.request; \
		void *_prev = (_treq)->pub.trunk->in_handler; \
		(_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_conn_release; \
		ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_conn_release(conn=%p, preq=%p, uctx=%p)", \
			  (_treq)->pub.tconn->pub.conn, \
			  (_treq)->pub.preq, \
			  (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->funcs.request_conn_release((_treq)->pub.tconn->pub.conn, (_treq)->pub.preq, (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

/** Call the complete callback (if set)
 *
 */
#define DO_REQUEST_COMPLETE(_treq) \
do { \
	if ((_treq)->pub.trunk->funcs.request_complete) { \
		request_t *request = (_treq)->pub.request; \
		void *_prev = (_treq)->pub.trunk->in_handler; \
		ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_complete(request=%p, preq=%p, rctx=%p, uctx=%p)", \
			  (_treq)->pub.request, \
			  (_treq)->pub.preq, \
			  (_treq)->pub.rctx, \
			  (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_complete; \
		(_treq)->pub.trunk->funcs.request_complete((_treq)->pub.request, (_treq)->pub.preq, (_treq)->pub.rctx, (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

/** Call the fail callback (if set)
 *
 */
#define DO_REQUEST_FAIL(_treq, _prev_state) \
do { \
	if ((_treq)->pub.trunk->funcs.request_fail) { \
		request_t *request = (_treq)->pub.request; \
		void *_prev = (_treq)->pub.trunk->in_handler; \
		ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_fail(request=%p, preq=%p, rctx=%p, state=%s uctx=%p)", \
			  (_treq)->pub.request, \
			  (_treq)->pub.preq, \
			  (_treq)->pub.rctx, \
			  fr_table_str_by_value(trunk_request_states, (_prev_state), "<INVALID>"), \
			  (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_fail; \
		(_treq)->pub.trunk->funcs.request_fail((_treq)->pub.request, (_treq)->pub.preq, (_treq)->pub.rctx, _prev_state, (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

/** Call the free callback (if set)
 *
 */
#define DO_REQUEST_FREE(_treq) \
do { \
	if ((_treq)->pub.trunk->funcs.request_free) { \
		request_t *request = (_treq)->pub.request; \
		void *_prev = (_treq)->pub.trunk->in_handler; \
		ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_free(request=%p, preq=%p, uctx=%p)", \
			  (_treq)->pub.request, \
			  (_treq)->pub.preq, \
			  (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_free; \
		(_treq)->pub.trunk->funcs.request_free((_treq)->pub.request, (_treq)->pub.preq, (_treq)->pub.trunk->uctx); \
		(_treq)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

/** Write one or more requests to a connection
 *
 */
#define DO_REQUEST_MUX(_tconn) \
do { \
	void *_prev = (_tconn)->pub.trunk->in_handler; \
	DEBUG3("[%" PRIu64 "] Calling request_mux(el=%p, tconn=%p, conn=%p, uctx=%p)", \
	       (_tconn)->pub.conn->id, \
	       (_tconn)->pub.trunk->el, \
	       (_tconn), \
	       (_tconn)->pub.conn, \
	       (_tconn)->pub.trunk->uctx); \
	(_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.request_mux; \
	(_tconn)->pub.trunk->funcs.request_mux((_tconn)->pub.trunk->el, (_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->uctx); \
	(_tconn)->pub.trunk->in_handler = _prev; \
} while(0)

/** Read one or more requests from a connection
 *
 */
#define DO_REQUEST_DEMUX(_tconn) \
do { \
	void *_prev = (_tconn)->pub.trunk->in_handler; \
	DEBUG3("[%" PRIu64 "] Calling request_demux(tconn=%p, conn=%p, uctx=%p)", \
	       (_tconn)->pub.conn->id, \
	       (_tconn), \
	       (_tconn)->pub.conn, \
	       (_tconn)->pub.trunk->uctx); \
	(_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.request_demux; \
	(_tconn)->pub.trunk->funcs.request_demux((_tconn)->pub.trunk->el, (_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->uctx); \
	(_tconn)->pub.trunk->in_handler = _prev; \
} while(0)

/** Write one or more cancellation requests to a connection
 *
 */
#define DO_REQUEST_CANCEL_MUX(_tconn) \
do { \
	if ((_tconn)->pub.trunk->funcs.request_cancel_mux) { \
		void *_prev = (_tconn)->pub.trunk->in_handler; \
		DEBUG3("[%" PRIu64 "] Calling request_cancel_mux(tconn=%p, conn=%p, uctx=%p)", \
		       (_tconn)->pub.conn->id, \
		       (_tconn), \
		       (_tconn)->pub.conn, \
		       (_tconn)->pub.trunk->uctx); \
		(_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.request_cancel_mux; \
		(_tconn)->pub.trunk->funcs.request_cancel_mux((_tconn)->pub.trunk->el, (_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->uctx); \
		(_tconn)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

/** Allocate a new connection
 *
 */
#define DO_CONNECTION_ALLOC(_tconn) \
do { \
	void *_prev = trunk->in_handler; \
	DEBUG3("Calling connection_alloc(tconn=%p, el=%p, conf=%p, log_prefix=\"%s\", uctx=%p)", \
	       (_tconn), \
	       (_tconn)->pub.trunk->el, \
	       (_tconn)->pub.trunk->conf.conn_conf, \
	       trunk->log_prefix, \
	       (_tconn)->pub.trunk->uctx); \
	(_tconn)->pub.trunk->in_handler = (void *) (_tconn)->pub.trunk->funcs.connection_alloc; \
	(_tconn)->pub.conn = trunk->funcs.connection_alloc((_tconn), (_tconn)->pub.trunk->el, (_tconn)->pub.trunk->conf.conn_conf, (_tconn)->pub.trunk->log_prefix, trunk->uctx); \
	(_tconn)->pub.trunk->in_handler = _prev; \
	if (!(_tconn)->pub.conn) { \
		ERROR("Failed creating new connection"); \
		talloc_free(tconn); \
		return -1; \
	} \
} while(0)

/** Change what events the connection should be notified about
 *
 */
#define DO_CONNECTION_NOTIFY(_tconn, _events) \
do { \
	if ((_tconn)->pub.trunk->funcs.connection_notify) { \
		void *_prev = (_tconn)->pub.trunk->in_handler; \
		DEBUG3("[%" PRIu64 "] Calling connection_notify(tconn=%p, conn=%p, el=%p, events=%s, uctx=%p)", \
		       (_tconn)->pub.conn->id, \
		       (_tconn), \
		       (_tconn)->pub.conn, \
		       (_tconn)->pub.trunk->el, \
		       fr_table_str_by_value(trunk_connection_events, (_events), "<INVALID>"), \
		       (_tconn)->pub.trunk->uctx); \
		(_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.connection_notify; \
		(_tconn)->pub.trunk->funcs.connection_notify((_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->el, (_events), (_tconn)->pub.trunk->uctx); \
		(_tconn)->pub.trunk->in_handler = _prev; \
	} \
} while(0)

#define IN_HANDLER(_trunk)		(((_trunk)->in_handler) != NULL)
#define IN_REQUEST_MUX(_trunk)		(((_trunk)->funcs.request_mux) && ((_trunk)->in_handler == (void *)(_trunk)->funcs.request_mux))
#define IN_REQUEST_DEMUX(_trunk)	(((_trunk)->funcs.request_demux) && ((_trunk)->in_handler == (void *)(_trunk)->funcs.request_demux))
#define IN_REQUEST_CANCEL_MUX(_trunk)	(((_trunk)->funcs.request_cancel_mux) && ((_trunk)->in_handler == (void *)(_trunk)->funcs.request_cancel_mux))

#define IS_SERVICEABLE(_tconn)		((_tconn)->pub.state & TRUNK_CONN_SERVICEABLE)
#define IS_PROCESSING(_tconn)		((tconn)->pub.state & TRUNK_CONN_PROCESSING)

/** Remove the current request from the backlog
 *
 */
#define REQUEST_EXTRACT_BACKLOG(_treq) \
do { \
	int _ret; \
	_ret = fr_heap_extract(&(_treq)->pub.trunk->backlog, _treq); \
	if (!fr_cond_assert_msg(_ret == 0, "Failed extracting conn from backlog heap: %s", fr_strerror())) break; \
} while (0)

/** Remove the current request from the pending list
 *
 */
#define REQUEST_EXTRACT_PENDING(_treq) \
do { \
	int _ret; \
	_ret = fr_heap_extract(&(_treq)->pub.tconn->pending, _treq); \
	if (!fr_cond_assert_msg(_ret == 0, "Failed extracting conn from pending heap: %s", fr_strerror())) break; \
} while (0)

/** Remove the current request from the partial slot
 *
 */
#define REQUEST_EXTRACT_PARTIAL(_treq) \
do { \
	fr_assert((_treq)->pub.tconn->partial == treq); \
	tconn->partial = NULL; \
} while (0)

/** Remove the current request from the sent list
 *
 */
#define REQUEST_EXTRACT_SENT(_treq) fr_dlist_remove(&tconn->sent, treq)

/** Remove the current request from the reapable list
 *
 */
#define REQUEST_EXTRACT_REAPABLE(_treq) fr_dlist_remove(&tconn->reapable, treq)

/** Remove the current request from the cancel list
 *
 */
#define REQUEST_EXTRACT_CANCEL(_treq) fr_dlist_remove(&tconn->cancel, treq)

/** Remove the current request from the cancel_partial slot
 *
 */
#define REQUEST_EXTRACT_CANCEL_PARTIAL(_treq) \
do { \
	fr_assert((_treq)->pub.tconn->cancel_partial == treq); \
	tconn->cancel_partial = NULL; \
} while (0)

/** Remove the current request from the cancel sent list
 *
 */
#define REQUEST_EXTRACT_CANCEL_SENT(_treq) fr_dlist_remove(&tconn->cancel_sent, treq)

/** Reorder the connections in the active heap
 *
 * fr_heap_extract will also error out if heap_id is bad - no need for assert
 */
#define CONN_REORDER(_tconn) \
do { \
	int _ret; \
	if ((fr_minmax_heap_num_elements((_tconn)->pub.trunk->active) == 1)) break; \
	if (!fr_cond_assert((_tconn)->pub.state == TRUNK_CONN_ACTIVE)) break; \
	_ret = fr_minmax_heap_extract((_tconn)->pub.trunk->active, (_tconn)); \
	if (!fr_cond_assert_msg(_ret == 0, "Failed extracting conn from active heap: %s", fr_strerror())) break; \
	fr_minmax_heap_insert((_tconn)->pub.trunk->active, (_tconn)); \
} while (0)

/** Call a list of watch functions associated with a state
 *
 */
static inline void trunk_watch_call(trunk_t *trunk, fr_dlist_head_t *list, trunk_state_t state)
{
	/*
	 *	Nested watcher calls are not allowed
	 *	and shouldn't be possible because of
	 *	deferred signal processing.
	 */
	fr_assert(trunk->next_watcher == NULL);

	while ((trunk->next_watcher = fr_dlist_next(list, trunk->next_watcher))) {
		trunk_watch_entry_t	*entry = trunk->next_watcher;
		bool			oneshot = entry->oneshot;	/* Watcher could be freed, so store now */

		if (!entry->enabled) continue;
		if (oneshot) trunk->next_watcher = fr_dlist_remove(list, entry);

		entry->func(trunk, trunk->pub.state, state, entry->uctx);

		if (oneshot) talloc_free(entry);
	}
	trunk->next_watcher = NULL;
}

/** Call the state change watch functions
 *
 */
#define CALL_WATCHERS(_trunk, _state) \
do { \
	if (fr_dlist_empty(&(_trunk)->watch[_state])) break; \
	trunk_watch_call((_trunk), &(_trunk)->watch[_state], _state); \
} while(0)

/** Remove a watch function from a trunk state list
 *
 * @param[in] trunk	The trunk to remove the watcher from.
 * @param[in] state	to remove the watch from.
 * @param[in] watch	Function to remove.
 * @return
 *	- 0 if the function was removed successfully.
 *	- -1 if the function wasn't present in the watch list.
 *	- -2 if an invalid state was passed.
 */
int trunk_del_watch(trunk_t *trunk, trunk_state_t state, trunk_watch_t watch)
{
	trunk_watch_entry_t	*entry = NULL;
	fr_dlist_head_t	        *list;

	if (state >= TRUNK_STATE_MAX) return -2;

	list = &trunk->watch[state];
	while ((entry = fr_dlist_next(list, entry))) {
		if (entry->func == watch) {
			if (trunk->next_watcher == entry) {
				trunk->next_watcher = fr_dlist_remove(list, entry);
			} else {
				fr_dlist_remove(list, entry);
			}
			talloc_free(entry);
			return 0;
		}
	}

	return -1;
}

/** Add a watch entry to the trunk state list
 *
 * @param[in] trunk	The trunk to add the watcher to.
 * @param[in] state	to watch for.
 * @param[in] watch	Function to add.
 * @param[in] oneshot	Should this watcher only be run once.
 * @param[in] uctx	Context to pass to function.
 * @return
 *	- NULL if an invalid state is passed.
 *	- A new watch entry handle on success.
 */
trunk_watch_entry_t *trunk_add_watch(trunk_t *trunk, trunk_state_t state,
					   trunk_watch_t watch, bool oneshot, void const *uctx)
{
	trunk_watch_entry_t	*entry;
	fr_dlist_head_t		*list;

	if (state >= TRUNK_STATE_MAX) return NULL;

	list = &trunk->watch[state];
	MEM(entry = talloc_zero(trunk, trunk_watch_entry_t));

	entry->func = watch;
	entry->oneshot = oneshot;
	entry->enabled = true;
	memcpy(&entry->uctx, &uctx, sizeof(entry->uctx));
	fr_dlist_insert_tail(list, entry);

	return entry;
}

#define TRUNK_STATE_TRANSITION(_new) \
do { \
	DEBUG3("Trunk changed state %s -> %s", \
	       fr_table_str_by_value(trunk_states, trunk->pub.state, "<INVALID>"), \
	       fr_table_str_by_value(trunk_states, _new, "<INVALID>")); \
	CALL_WATCHERS(trunk, _new); \
	trunk->pub.state = _new; \
} while (0)

static void trunk_request_enter_backlog(trunk_request_t *treq, bool new);
static void trunk_request_enter_pending(trunk_request_t *treq, trunk_connection_t *tconn, bool new);
static void trunk_request_enter_partial(trunk_request_t *treq);
static void trunk_request_enter_sent(trunk_request_t *treq);
static void trunk_request_enter_reapable(trunk_request_t *treq);
static void trunk_request_enter_failed(trunk_request_t *treq);
static void trunk_request_enter_complete(trunk_request_t *treq);
static void trunk_request_enter_cancel(trunk_request_t *treq, trunk_cancel_reason_t reason);
static void trunk_request_enter_cancel_sent(trunk_request_t *treq);
static void trunk_request_enter_cancel_complete(trunk_request_t *treq);

static uint64_t trunk_requests_per_connection(uint16_t *conn_count_out, uint32_t *req_conn_out,
					      trunk_t *trunk, fr_time_t now, NDEBUG_UNUSED bool verify);

static int trunk_connection_spawn(trunk_t *trunk, fr_time_t now);
static inline void trunk_connection_auto_full(trunk_connection_t *tconn);
static inline void trunk_connection_auto_unfull(trunk_connection_t *tconn);
static inline void trunk_connection_readable(trunk_connection_t *tconn);
static inline void trunk_connection_writable(trunk_connection_t *tconn);
static void trunk_connection_event_update(trunk_connection_t *tconn);
static void trunk_connection_enter_full(trunk_connection_t *tconn);
static void trunk_connection_enter_inactive(trunk_connection_t *tconn);
static void trunk_connection_enter_inactive_draining(trunk_connection_t *tconn);
static void trunk_connection_enter_draining(trunk_connection_t *tconn);
static void trunk_connection_enter_draining_to_free(trunk_connection_t *tconn);
static void trunk_connection_enter_active(trunk_connection_t *tconn);

static void trunk_rebalance(trunk_t *trunk);
static void trunk_manage(trunk_t *trunk, fr_time_t now);
static void _trunk_timer(fr_event_list_t *el, fr_time_t now, void *uctx);
static void trunk_backlog_drain(trunk_t *trunk);

/** Compare two protocol requests
 *
 * Allows protocol requests to be prioritised with a function
 * specified by the API client.  Defaults to by pointer address
 * if no function is specified.
 *
 * @param[in] a	treq to compare to b.
 * @param[in] b treq to compare to a.
 * @return
 *	- +1 if a > b.
 *	- 0 if a == b.
 *	- -1 if a < b.
 */
static int8_t _trunk_request_prioritise(void const *a, void const *b)
{
	trunk_request_t const *treq_a = talloc_get_type_abort_const(a, trunk_request_t);
	trunk_request_t const *treq_b = talloc_get_type_abort_const(b, trunk_request_t);

	fr_assert(treq_a->pub.trunk == treq_b->pub.trunk);

	return treq_a->pub.trunk->funcs.request_prioritise(treq_a->pub.preq, treq_b->pub.preq);
}

/** Remove a request from all connection lists
 *
 * A common function used by init, fail, complete state functions to disassociate
 * a request from a connection in preparation for freeing or reassignment.
 *
 * Despite its unassuming name, this function is *the* place to put calls to
 * functions which need to be called when the number of requests associated with
 * a connection changes.
 *
 * Trunk requests will always be passed to this function before they're removed
 * from a connection, even if the requests are being freed.
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_remove_from_conn(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_UNASSIGNED:
		return;	/* Not associated with connection */

	case TRUNK_REQUEST_STATE_PENDING:
		REQUEST_EXTRACT_PENDING(treq);
		break;

	case TRUNK_REQUEST_STATE_PARTIAL:
		REQUEST_EXTRACT_PARTIAL(treq);
		break;

	case TRUNK_REQUEST_STATE_SENT:
		REQUEST_EXTRACT_SENT(treq);
		break;

	case TRUNK_REQUEST_STATE_REAPABLE:
		REQUEST_EXTRACT_REAPABLE(treq);
		break;

	case TRUNK_REQUEST_STATE_CANCEL:
		REQUEST_EXTRACT_CANCEL(treq);
		break;

	case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
		REQUEST_EXTRACT_CANCEL_PARTIAL(treq);
		break;

	case TRUNK_REQUEST_STATE_CANCEL_SENT:
		REQUEST_EXTRACT_CANCEL_SENT(treq);
		break;

	default:
		fr_assert(0);
		break;
	}

	/*
	 *	If the request wasn't associated with a
	 *	connection, then there's nothing more
	 *	to do.
	 */
	if (!tconn) return;

	{
		request_t *request = treq->pub.request;

		ROPTIONAL(RDEBUG3, DEBUG3, "%s Trunk connection released request %" PRIu64,
			  tconn->pub.conn->name, treq->id);
	}
	/*
	 *	Release any connection specific resources the
	 *	treq holds.
	 */
	DO_REQUEST_CONN_RELEASE(treq);

	switch (tconn->pub.state){
	case TRUNK_CONN_FULL:
		trunk_connection_auto_unfull(tconn);		/* Check if we can switch back to active */
		if (tconn->pub.state == TRUNK_CONN_FULL) break;	/* Only fallthrough if conn is now active */
		FALL_THROUGH;

	case TRUNK_CONN_ACTIVE:
		CONN_REORDER(tconn);
		break;

	default:
		break;
	}

	treq->pub.tconn = NULL;

	/*
	 *	Request removed from the connection
	 *	see if we need up deregister I/O events.
	 */
	trunk_connection_event_update(tconn);
}

/** Transition a request to the unassigned state, in preparation for re-assignment
 *
 * @note treq->tconn may be inviable after calling
 *	if treq->conn and connection_signals_pause are not used.
 *	This is due to call to trunk_request_remove_from_conn.
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_enter_unassigned(trunk_request_t *treq)
{
	trunk_t		*trunk = treq->pub.trunk;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_UNASSIGNED:
		return;

	case TRUNK_REQUEST_STATE_BACKLOG:
		REQUEST_EXTRACT_BACKLOG(treq);
		break;

	case TRUNK_REQUEST_STATE_PENDING:
	case TRUNK_REQUEST_STATE_CANCEL:
	case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
	case TRUNK_REQUEST_STATE_CANCEL_SENT:
		trunk_request_remove_from_conn(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_UNASSIGNED);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_UNASSIGNED);
}

/** Transition a request to the backlog state, adding it to the backlog of the trunk
 *
 * @note treq->tconn and treq may be inviable after calling
 *	if treq->conn and connection_signals_pause are not used.
 *	This is due to call to trunk_manage.
 *
 * @param[in] treq	to trigger a state change for.
 * @param[in] new	Whether this is a new request.
 */
static void trunk_request_enter_backlog(trunk_request_t *treq, bool new)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_INIT:
	case TRUNK_REQUEST_STATE_UNASSIGNED:
		break;

	case TRUNK_REQUEST_STATE_PENDING:
		REQUEST_EXTRACT_PENDING(treq);
		break;

	case TRUNK_REQUEST_STATE_CANCEL:
		REQUEST_EXTRACT_CANCEL(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_BACKLOG);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_BACKLOG);
	fr_heap_insert(&trunk->backlog, treq);	/* Insert into the backlog heap */

	/*
	 *	A new request has entered the trunk.
	 *	Re-calculate request/connection ratios.
	 */
	if (new) trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false);

	/*
	 *	To reduce latency, if there's no connections
	 *      in the connecting state, call the trunk manage
	 *	function immediately.
	 *
	 *	Likewise, if there's draining connections
	 *	which could be moved back to active call
	 *	the trunk manage function.
	 *
	 *	Remember requests only enter the backlog if
	 *	there's no connections which can service them.
	 */
	if ((trunk_connection_count_by_state(treq->pub.trunk, TRUNK_CONN_CONNECTING) == 0) ||
	    (trunk_connection_count_by_state(treq->pub.trunk, TRUNK_CONN_DRAINING) > 0)) {
		trunk_connection_manage_schedule(treq->pub.trunk);
	}
}

/** Transition a request to the pending state, adding it to the backlog of an active connection
 *
 * All trunk requests being added to a connection get passed to this function.
 * All trunk requests being removed from a connection get passed to #trunk_request_remove_from_conn.
 *
 * @note treq->tconn and treq may be inviable after calling
 *	if treq->conn and connection_signals_pause is not used.
 *	This is due to call to trunk_connection_event_update.
 *
 * @param[in] treq	to trigger a state change for.
 * @param[in] tconn	to enqueue the request on.
 * @param[in] new	Whether this is a new request.
 */
static void trunk_request_enter_pending(trunk_request_t *treq, trunk_connection_t *tconn, bool new)
{
	trunk_t		*trunk = treq->pub.trunk;

	fr_assert(tconn->pub.trunk == trunk);
	fr_assert(IS_PROCESSING(tconn));

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_INIT:
	case TRUNK_REQUEST_STATE_UNASSIGNED:
		fr_assert(!treq->pub.tconn);
		break;

	case TRUNK_REQUEST_STATE_BACKLOG:
		fr_assert(!treq->pub.tconn);
		REQUEST_EXTRACT_BACKLOG(treq);
		break;

	case TRUNK_REQUEST_STATE_CANCEL:	/* Moved from another connection */
		REQUEST_EXTRACT_CANCEL(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_PENDING);
	}

	/*
	 *	Assign the new connection first this first so
	 *      it appears in the state log.
	 */
	treq->pub.tconn = tconn;

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_PENDING);

	{
		request_t *request = treq->pub.request;

		ROPTIONAL(RDEBUG, DEBUG3, "%s Trunk connection assigned request %"PRIu64,
			  tconn->pub.conn->name, treq->id);
	}
	fr_heap_insert(&tconn->pending, treq);

	/*
	 *	A new request has entered the trunk.
	 *	Re-calculate request/connection ratios.
	 */
	if (new) trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false);

	/*
	 *	Check if we need to automatically transition the
	 *	connection to full.
	 */
	trunk_connection_auto_full(tconn);

	/*
	 *	Reorder the connection in the heap now it has an
	 *	additional request.
	 */
	if (tconn->pub.state == TRUNK_CONN_ACTIVE) CONN_REORDER(tconn);

	/*
	 *	We have a new request, see if we need to register
	 *	for I/O events.
	 */
	trunk_connection_event_update(tconn);
}

/** Transition a request to the partial state, indicating that is has been partially sent
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_enter_partial(trunk_request_t *treq)
{
	trunk_connection_t *tconn = treq->pub.tconn;
	trunk_t	*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PENDING:	/* All requests go through pending, even requeued ones */
		REQUEST_EXTRACT_PENDING(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_PARTIAL);
	}

	fr_assert(!tconn->partial);
	tconn->partial = treq;

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_PARTIAL);
}

/** Transition a request to the sent state, indicating that it's been sent in its entirety
 *
 * @note treq->tconn and treq may be inviable after calling
 *	if treq->conn and connection_signals_pause is not used.
 *	This is due to call to trunk_connection_event_update.
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_enter_sent(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PENDING:
		REQUEST_EXTRACT_PENDING(treq);
		break;

	case TRUNK_REQUEST_STATE_PARTIAL:
		REQUEST_EXTRACT_PARTIAL(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_SENT);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_SENT);
	fr_dlist_insert_tail(&tconn->sent, treq);

	/*
	 *	Update the connection's sent stats if this is the
	 *	first time this request is being sent.
	 */
	if (!treq->sent) {
		tconn->sent_count++;
		treq->sent = true;

		/*
		 *	Enforces max_uses
		 */
		if ((trunk->conf.max_uses > 0) && (tconn->sent_count >= trunk->conf.max_uses)) {
			trunk_connection_enter_draining_to_free(tconn);
		}
	}

	/*
	 *	We just sent a request, we probably need
	 *	to tell the event loop we want to be
	 *	notified if there's data available.
	 */
	trunk_connection_event_update(tconn);
}

/** Transition a request to the reapable state, indicating that it's been sent in its entirety, but no response is expected
 *
 * @note Largely a replica of trunk_request_enter_sent.
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_enter_reapable(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PENDING:
		REQUEST_EXTRACT_PENDING(treq);
		break;

	case TRUNK_REQUEST_STATE_PARTIAL:
		REQUEST_EXTRACT_PARTIAL(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_SENT);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_REAPABLE);
	fr_dlist_insert_tail(&tconn->reapable, treq);

	if (!treq->sent) {
		tconn->sent_count++;
		treq->sent = true;

		if ((trunk->conf.max_uses > 0) && (tconn->sent_count >= trunk->conf.max_uses)) {
			trunk_connection_enter_draining_to_free(tconn);
		}
	}

	trunk_connection_event_update(tconn);
}

/** Transition a request to the cancel state, placing it in a connection's cancellation list
 *
 * If a request_cancel_send callback is provided, that callback will
 * be called periodically for requests which were cancelled due to
 * a signal.
 *
 * The request_cancel_send callback will dequeue cancelled requests
 * and inform a remote server that the result is no longer required.
 *
 * A request must enter this state before being added to the backlog
 * of another connection if it's been sent or partially sent.
 *
 * @note treq->tconn and treq may be inviable after calling
 *	if treq->conn and connection_signals_pause is not used.
 *	This is due to call to trunk_connection_event_update.
 *
 * @param[in] treq	to trigger a state change for.
 * @param[in] reason	Why the request was cancelled.
 *			Should be one of:
 *			- TRUNK_CANCEL_REASON_SIGNAL request cancelled
 *			  because of a signal from the interpreter.
 *			- TRUNK_CANCEL_REASON_MOVE request cancelled
 *			  because the connection failed and it needs
 *			  to be assigned to a new connection.
 *			- TRUNK_CANCEL_REASON_REQUEUE request cancelled
 *			  as it needs to be resent on the same connection.
 */
static void trunk_request_enter_cancel(trunk_request_t *treq, trunk_cancel_reason_t reason)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PARTIAL:
		REQUEST_EXTRACT_PARTIAL(treq);
		break;

	case TRUNK_REQUEST_STATE_SENT:
		REQUEST_EXTRACT_SENT(treq);
		break;

	case TRUNK_REQUEST_STATE_REAPABLE:
		REQUEST_EXTRACT_REAPABLE(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL);
	fr_dlist_insert_tail(&tconn->cancel, treq);
	treq->cancel_reason = reason;

	DO_REQUEST_CANCEL(treq, reason);

	/*
	 *	Our treq is no longer bound to an actual
	 *      request_t *, as we can't guarantee the
	 *	lifetime of the original request_t *.
	 */
	if (treq->cancel_reason == TRUNK_CANCEL_REASON_SIGNAL) treq->pub.request = NULL;

	/*
	 *	Register for I/O write events if we need to.
	 */
	trunk_connection_event_update(treq->pub.tconn);
}

/** Transition a request to the cancel_partial state, placing it in a connection's cancel_partial slot
 *
 * The request_demux function is then responsible for signalling
 * that the cancel request is complete when the remote server
 * acknowledges the cancellation request.
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_enter_cancel_partial(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
	fr_assert(trunk->funcs.request_cancel_mux);
	fr_assert(treq->cancel_reason == TRUNK_CANCEL_REASON_SIGNAL);

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_CANCEL:	/* The only valid state cancel_sent can be reached from */
		REQUEST_EXTRACT_CANCEL(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
	fr_assert(!tconn->partial);
	tconn->cancel_partial = treq;
}

/** Transition a request to the cancel_sent state, placing it in a connection's cancel_sent list
 *
 * The request_demux function is then responsible for signalling
 * that the cancel request is complete when the remote server
 * acknowledges the cancellation request.
 *
 * @note treq->tconn and treq may be inviable after calling
 *	if treq->conn and connection_signals_pause is not used.
 *	This is due to call to trunk_connection_event_update.
 *
 * @param[in] treq	to trigger a state change for.
 */
static void trunk_request_enter_cancel_sent(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
	fr_assert(trunk->funcs.request_cancel_mux);
	fr_assert(treq->cancel_reason == TRUNK_CANCEL_REASON_SIGNAL);

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
		REQUEST_EXTRACT_CANCEL_PARTIAL(treq);
		break;

	case TRUNK_REQUEST_STATE_CANCEL:
		REQUEST_EXTRACT_CANCEL(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_SENT);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_SENT);
	fr_dlist_insert_tail(&tconn->cancel_sent, treq);

	/*
	 *	De-register for I/O write events
	 *	and register the read events
	 *	to drain the cancel ACKs.
	 */
	trunk_connection_event_update(treq->pub.tconn);
}

/** Cancellation was acked, the request is complete, free it
 *
 * The API client will not be informed, as the original request_t *
 * will likely have been freed by this point.
 *
 * @note treq will be inviable after a call to this function.
 *      treq->tconn may be inviable after calling
 *	if treq->conn and connection_signals_pause is not used.
 *	This is due to call to trunk_request_remove_from_conn.
 *
 * @param[in] treq	to mark as complete.
 */
static void trunk_request_enter_cancel_complete(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
	if (!fr_cond_assert(!treq->pub.request)) return;	/* Only a valid state for request_t * which have been cancelled */

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_CANCEL_SENT:
	case TRUNK_REQUEST_STATE_CANCEL:
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_COMPLETE);
	}

	trunk_request_remove_from_conn(treq);

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_COMPLETE);
	trunk_request_free(&treq);	/* Free the request */
}

/** Request completed successfully, inform the API client and free the request
 *
 * @note treq will be inviable after a call to this function.
 *	treq->tconn may also be inviable due to call to
 *	trunk_request_remove_from_conn.
 *
 * @param[in] treq	to mark as complete.
 */
static void trunk_request_enter_complete(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;
	trunk_t		*trunk = treq->pub.trunk;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_SENT:
	case TRUNK_REQUEST_STATE_PENDING:
	case TRUNK_REQUEST_STATE_REAPABLE:
		trunk_request_remove_from_conn(treq);
		break;

	default:
		REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_COMPLETE);
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_COMPLETE);
	DO_REQUEST_COMPLETE(treq);
	trunk_request_free(&treq);	/* Free the request */
}

/** Request failed, inform the API client and free the request
 *
 * @note treq will be inviable after a call to this function.
 *	treq->tconn may also be inviable due to call to
 *	trunk_request_remove_from_conn.
 *
 * @param[in] treq	to mark as failed.
 */
static void trunk_request_enter_failed(trunk_request_t *treq)
{
	trunk_connection_t		*tconn = treq->pub.tconn;
	trunk_t			*trunk = treq->pub.trunk;
	trunk_request_state_t	prev = treq->pub.state;

	if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_BACKLOG:
		REQUEST_EXTRACT_BACKLOG(treq);
		break;

	default:
		trunk_request_remove_from_conn(treq);
		break;
	}

	REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_FAILED);
	DO_REQUEST_FAIL(treq, prev);
	trunk_request_free(&treq);	/* Free the request */
}

/** Check to see if a trunk request can be enqueued
 *
 * @param[out] tconn_out	Connection the request may be enqueued on.
 * @param[in] trunk		To enqueue requests on.
 * @param[in] request		associated with the treq (if any).
 * @return
 *	- TRUNK_ENQUEUE_OK			caller should enqueue request on provided tconn.
 *	- TRUNK_ENQUEUE_IN_BACKLOG		Request should be queued in the backlog.
 *	- TRUNK_ENQUEUE_NO_CAPACITY		Unable to enqueue request as we have no spare
 *						connections or backlog space.
 *	- TRUNK_ENQUEUE_DST_UNAVAILABLE	Can't enqueue because the destination is
 *						unreachable.
 */
static trunk_enqueue_t trunk_request_check_enqueue(trunk_connection_t **tconn_out, trunk_t *trunk,
						      request_t *request)
{
	trunk_connection_t	*tconn;
	/*
	 *	If we have an active connection then
	 *	return that.
	 */
	tconn = fr_minmax_heap_min_peek(trunk->active);
	if (tconn) {
		*tconn_out = tconn;
		return TRUNK_ENQUEUE_OK;
	}

	/*
	 *	Unlike the connection pool, we don't need
	 *	to drive any internal processes by feeding
	 *	it requests.
	 *
	 *	If the last event to occur was a failure
	 *	we refuse to enqueue new requests until
	 *	one or more connections comes online.
	 */
	if (!trunk->conf.backlog_on_failed_conn &&
	    fr_time_gt(trunk->pub.last_failed, fr_time_wrap(0)) &&
	    fr_time_lt(trunk->pub.last_connected, trunk->pub.last_failed)) {
	    	RATE_LIMIT_LOCAL_ROPTIONAL(&trunk->limit_last_failure_log,
					   RWARN, WARN, "Refusing to enqueue requests - "
					   "No active connections and last event was a connection failure");

		return TRUNK_ENQUEUE_DST_UNAVAILABLE;
	}


	/*
	 *	Only enforce if we're limiting maximum
	 *	number of connections, and maximum
	 *	number of requests per connection.
	 *
	 *	The alloc function also checks this
	 *	which is why this is only done for
	 *	debug builds.
	 */
	if (trunk->conf.max_req_per_conn && trunk->conf.max) {
		uint64_t	limit;

		limit = trunk->conf.max * (uint64_t)trunk->conf.max_req_per_conn;
		if (limit > 0) {
			uint64_t	total_reqs;

			total_reqs = trunk_request_count_by_state(trunk, TRUNK_CONN_ALL,
								     TRUNK_REQUEST_STATE_ALL);
			if (total_reqs >= (limit + trunk->conf.max_backlog)) {
				RATE_LIMIT_LOCAL_ROPTIONAL(&trunk->limit_max_requests_alloc_log,
							   RWARN, WARN, "Refusing to alloc requests - "
							   "Limit of %"PRIu64" (max = %u * per_connection_max = %u) "
							   "plus %u backlog requests reached",
							   limit, trunk->conf.max, trunk->conf.max_req_per_conn,
							   trunk->conf.max_backlog);
				return TRUNK_ENQUEUE_NO_CAPACITY;
			}
		}
	}

	return TRUNK_ENQUEUE_IN_BACKLOG;
}

/** Enqueue a request which has never been assigned to a connection or was previously cancelled
 *
 * @param[in] treq	to re enqueue.  Must have been removed
 *			from its existing connection with
 *			#trunk_connection_requests_dequeue.
 * @return
 *	- TRUNK_ENQUEUE_OK			Request was re-enqueued.
 *	- TRUNK_ENQUEUE_NO_CAPACITY		Request enqueueing failed because we're at capacity.
 *	- TRUNK_ENQUEUE_DST_UNAVAILABLE	Enqueuing failed for some reason.
 *      					Usually because the connection to the resource is down.
 */
static trunk_enqueue_t trunk_request_enqueue_existing(trunk_request_t *treq)
{
	trunk_t		*trunk = treq->pub.trunk;
	trunk_connection_t	*tconn = NULL;
	trunk_enqueue_t	ret;

	/*
	 *	Must *NOT* still be assigned to another connection
	 */
	fr_assert(!treq->pub.tconn);

	ret = trunk_request_check_enqueue(&tconn, trunk, treq->pub.request);
	switch (ret) {
	case TRUNK_ENQUEUE_OK:
		if (trunk->conf.always_writable) {
			connection_signals_pause(tconn->pub.conn);
			trunk_request_enter_pending(treq, tconn, false);
			trunk_connection_writable(tconn);
			connection_signals_resume(tconn->pub.conn);
		} else {
			trunk_request_enter_pending(treq, tconn, false);
		}
		break;

	case TRUNK_ENQUEUE_IN_BACKLOG:
		/*
		 *	No more connections and request
		 *	is already in the backlog.
		 *
		 *	Signal our caller it should stop
		 *	trying to drain the backlog.
		 */
		if (treq->pub.state == TRUNK_REQUEST_STATE_BACKLOG) return TRUNK_ENQUEUE_NO_CAPACITY;
		trunk_request_enter_backlog(treq, false);
		break;

	default:
		break;
	}

	return ret;
}

/** Shift requests in the specified states onto new connections
 *
 * This function will blindly dequeue any requests in the specified state and get
 * them back to the unassigned state, cancelling any sent or partially sent requests.
 *
 * This function does not check that dequeuing a request in a particular state is a
 * sane or sensible thing to do, that's up to the caller!
 *
 * @param[out] out	A list to insert the newly dequeued and unassigned
 *			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 uint64_t trunk_connection_requests_dequeue(fr_dlist_head_t *out, trunk_connection_t *tconn,
						  int states, uint64_t max)
{
	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, _state) do { \
	while ((treq = fr_dlist_head(_src_list))) { \
		OVER_MAX_CHECK; \
		fr_assert(treq->pub.state == (_state)); \
		trunk_request_enter_unassigned(treq); \
		fr_dlist_insert_tail(out, treq); \
	} } while (0)

	/*
	 *	Don't need to do anything with
	 *	cancellation requests.
	 */
	if (states & TRUNK_REQUEST_STATE_CANCEL) DEQUEUE_ALL(&tconn->cancel,
								TRUNK_REQUEST_STATE_CANCEL);

	/*
	 *	...same with cancel inform
	 */
	if (states & TRUNK_REQUEST_STATE_CANCEL_SENT) DEQUEUE_ALL(&tconn->cancel_sent,
								     TRUNK_REQUEST_STATE_CANCEL_SENT);

	/*
	 *	....same with cancel partial
	 */
	if (states & TRUNK_REQUEST_STATE_CANCEL_PARTIAL) {
		OVER_MAX_CHECK;
		treq = tconn->cancel_partial;
		if (treq) {
			fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
			trunk_request_enter_unassigned(treq);
			fr_dlist_insert_tail(out, treq);
		}
	}

	/*
	 *	...and pending.
	 */
	if (states & TRUNK_REQUEST_STATE_PENDING) {
		while ((treq = fr_heap_peek(tconn->pending))) {
			OVER_MAX_CHECK;
			fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_PENDING);
			trunk_request_enter_unassigned(treq);
			fr_dlist_insert_tail(out, treq);
		}
	}

	/*
	 *	Cancel partially sent requests
	 */
	if (states & TRUNK_REQUEST_STATE_PARTIAL) {
		OVER_MAX_CHECK;
		treq = tconn->partial;
		if (treq) {
			fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_PARTIAL);

			/*
			 *	Don't allow the connection to change state whilst
			 *	we're draining requests from it.
			 */
			connection_signals_pause(tconn->pub.conn);
			trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_MOVE);
			trunk_request_enter_unassigned(treq);
			fr_dlist_insert_tail(out, treq);
			connection_signals_resume(tconn->pub.conn);
		}
	}

	/*
	 *	Cancel sent requests
	 */
	if (states & TRUNK_REQUEST_STATE_SENT) {
		/*
		 *	Don't allow the connection to change state whilst
		 *	we're draining requests from it.
		 */
		connection_signals_pause(tconn->pub.conn);
		while ((treq = fr_dlist_head(&tconn->sent))) {
			OVER_MAX_CHECK;
			fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_SENT);

			trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_MOVE);
			trunk_request_enter_unassigned(treq);
			fr_dlist_insert_tail(out, treq);
		}
		connection_signals_resume(tconn->pub.conn);
	}

	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.
 * @param[in] fail_bound	If true causes any requests bound to the connection to fail.
 *      			If false bound requests will not be moved.
 *
 * @return the number of requests re-queued.
 */
static uint64_t trunk_connection_requests_requeue_priv(trunk_connection_t *tconn, int states, uint64_t max, bool fail_bound)
{
	trunk_t			*trunk = tconn->pub.trunk;
	fr_dlist_head_t			to_process;
	trunk_request_t		*treq = NULL;
	uint64_t			moved = 0;

	if (max == 0) max = UINT64_MAX;

	fr_dlist_talloc_init(&to_process, trunk_request_t, entry);

	/*
	 *	Prevent the connection changing state whilst we're
	 *	working with it.
	 *
	 *	There's a user callback that can be called by
	 *	trunk_request_enqueue_existing which can reconnect
	 *	the connection.
	 */
	connection_signals_pause(tconn->pub.conn);

	/*
	 *	Remove non-cancelled requests from the connection
	 */
	moved += trunk_connection_requests_dequeue(&to_process, tconn, states & ~TRUNK_REQUEST_STATE_CANCEL_ALL, max);

	/*
	 *	Prevent requests being requeued on the same trunk
	 *	connection, which would break rebalancing.
	 *
	 *	This is a bit of a hack, but nothing should test
	 *	for connection/list consistency in this code,
	 *      and if something is added later, it'll be flagged
	 *	by the tests.
	 */
	if (tconn->pub.state == TRUNK_CONN_ACTIVE) {
		int ret;

		ret = fr_minmax_heap_extract(trunk->active, tconn);
		if (!fr_cond_assert_msg(ret == 0,
					"Failed extracting conn from active heap: %s", fr_strerror())) goto done;

	}

	/*
	 *	Loop over all the requests we gathered and
	 *	redistribute them to new connections.
	 */
	while ((treq = fr_dlist_next(&to_process, treq))) {
		trunk_request_t *prev;

		prev = fr_dlist_remove(&to_process, treq);

		/*
		 *	Attempts to re-queue a request
		 *	that's bound to a connection
		 *	results in a failure.
		 */
		if (treq->bound_to_conn) {
			if (fail_bound || !IS_SERVICEABLE(tconn)) {
				trunk_request_enter_failed(treq);
			} else {
				trunk_request_enter_pending(treq, tconn, false);
			}
			goto next;
		}

		switch (trunk_request_enqueue_existing(treq)) {
		case TRUNK_ENQUEUE_OK:
			break;

		/*
		 *	A connection failed, and
		 *	there's no other connections
		 *	available to deal with the
		 *	load, it's been placed back
		 *	in the backlog.
		 */
		case TRUNK_ENQUEUE_IN_BACKLOG:
			break;

		/*
		 *	If we fail to re-enqueue then
		 *	there's nothing to do except
		 *	fail the request.
		 */
		case TRUNK_ENQUEUE_DST_UNAVAILABLE:
		case TRUNK_ENQUEUE_NO_CAPACITY:
		case TRUNK_ENQUEUE_FAIL:
			trunk_request_enter_failed(treq);
			break;
		}
	next:
		treq = prev;
	}

	/*
	 *	Add the connection back into the active list
	 */
	if (tconn->pub.state == TRUNK_CONN_ACTIVE) {
		int ret;

		ret = fr_minmax_heap_insert(trunk->active, tconn);
		if (!fr_cond_assert_msg(ret == 0,
				        "Failed re-inserting conn into active heap: %s", fr_strerror())) goto done;
	}
	if (moved >= max) goto done;

	/*
	 *	Deal with the cancelled requests specially we can't
	 *      queue them up again as they were only valid on that
	 *	specific connection.
	 *
	 *	We just need to run them to completion which, as
	 *	they should already be in the unassigned state,
	 *	just means freeing them.
	 */
	moved += trunk_connection_requests_dequeue(&to_process, tconn,
						   states & TRUNK_REQUEST_STATE_CANCEL_ALL, max - moved);
	while ((treq = fr_dlist_next(&to_process, treq))) {
		trunk_request_t *prev;

		prev = fr_dlist_remove(&to_process, treq);
		trunk_request_free(&treq);
		treq = prev;
	}

done:

	/*
	 *	Always re-calculate the request/connection
	 *	ratio at the end.
	 *
	 *	This avoids having the state transition
	 *	functions do it.
	 *
	 *	The ratio would be wrong when they calculated
	 *	it anyway, because a bunch of requests are
	 *	dequeued from the connection and temporarily
	 *	cease to exist from the perspective of the
	 *	trunk_requests_per_connection code.
	 */
	trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false);

	connection_signals_resume(tconn->pub.conn);
	return moved;
}

/** Move requests off of a connection and requeue elsewhere
 *
 * @note We don't re-queue on draining or draining to free, as requests should have already been
 *	 moved off of the connection.  It's also dangerous as the trunk management code main
 *	 clean up a connection in this state when it's run on re-queue, and then the caller
 *	 may try and access a now freed connection.
 *
 * @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.
 * @param[in] fail_bound	If true causes any requests bound to the connection to fail.
 *      			If false bound requests will not be moved.
 * @return The number of requests requeued.
 */
uint64_t trunk_connection_requests_requeue(trunk_connection_t *tconn, int states, uint64_t max, bool fail_bound)
{
	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
		return trunk_connection_requests_requeue_priv(tconn, states, max, fail_bound);

	default:
		return 0;
	}
}

/** Signal a partial write
 *
 * Where there's high load, and the outbound write buffer is full
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_partial(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(IN_REQUEST_MUX(treq->pub.trunk),
				"%s can only be called from within request_mux handler", __FUNCTION__)) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PENDING:
		trunk_request_enter_partial(treq);
		break;

	default:
		return;
	}
}

/** Signal that the request was written to a connection successfully
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_sent(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(IN_REQUEST_MUX(treq->pub.trunk),
				"%s can only be called from within request_mux handler", __FUNCTION__)) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PENDING:
	case TRUNK_REQUEST_STATE_PARTIAL:
		trunk_request_enter_sent(treq);
		break;

	default:
		return;
	}
}

/** Signal that the request was written to a connection successfully, but no response is expected
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_reapable(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(IN_REQUEST_MUX(treq->pub.trunk),
				"%s can only be called from within request_mux handler", __FUNCTION__)) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PENDING:
	case TRUNK_REQUEST_STATE_PARTIAL:
		trunk_request_enter_reapable(treq);
		break;

	default:
		return;
	}
}

/** Signal that a trunk request is complete
 *
 * The API client will be informed that the request is now complete.
 */
void trunk_request_signal_complete(trunk_request_t *treq)
{
	trunk_t *trunk = treq->pub.trunk;

	if (!fr_cond_assert_msg(trunk, "treq not associated with trunk")) return;

	/*
	 *	We assume that if the request is being signalled
	 *	as complete from the demux function, that it was
	 *	a successful read.
	 *
	 *	If this assumption turns out to be incorrect
	 *	then we need to add an argument to signal_complete
	 *	to indicate if this is a successful read.
	 */
	if (IN_REQUEST_DEMUX(trunk)) trunk->pub.last_read_success = fr_time();

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_SENT:
	case TRUNK_REQUEST_STATE_PENDING:	/* Got immediate response, i.e. cached */
	case TRUNK_REQUEST_STATE_REAPABLE:
		trunk_request_enter_complete(treq);
		break;

	default:
		return;
	}
}

/** Signal that a trunk request failed
 *
 * The API client will be informed that the request has failed.
 */
void trunk_request_signal_fail(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	trunk_request_enter_failed(treq);
}

/** Cancel a trunk request
 *
 * treq can be in any state, but requests to cancel if the treq is not in
 * the TRUNK_REQUEST_STATE_PARTIAL or TRUNK_REQUEST_STATE_SENT state will be ignored.
 *
 * The complete or failed callbacks will not be called here, as it's assumed the request_t *
 * is now inviable as it's being cancelled.
 *
 * The free function however, is called, and that should be used to perform necessary
 * cleanup.
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_cancel(trunk_request_t *treq)
{
	trunk_t	*trunk;

	/*
	 *	Ensure treq hasn't been freed
	 */
	(void)talloc_get_type_abort(treq, trunk_request_t);

	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(!IN_HANDLER(treq->pub.trunk),
				"%s cannot be called within a handler", __FUNCTION__)) return;

 	trunk = treq->pub.trunk;

	switch (treq->pub.state) {
	/*
	 *	We don't call the complete or failed callbacks
	 *	as the request and rctx are no longer viable.
	 */
	case TRUNK_REQUEST_STATE_PARTIAL:
	case TRUNK_REQUEST_STATE_SENT:
	{
		trunk_connection_t *tconn = treq->pub.tconn;

		/*
		 *	Don't allow connection state changes
		 */
		connection_signals_pause(tconn->pub.conn);
		trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_SIGNAL);
		if (!fr_cond_assert_msg(treq->pub.state == TRUNK_REQUEST_STATE_CANCEL,
					"Bad state %s after cancellation",
					fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"))) {
			connection_signals_resume(tconn->pub.conn);
			return;
		}
		/*
		 *	No cancel muxer.  We're done.
		 *
		 *	If we do have a cancel mux function,
		 *	the next time this connection becomes
		 *	writable, we'll call the cancel mux
		 *	function.
		 *
		 *	We don't run the complete or failed
		 *	callbacks here as the request is
		 *	being cancelled.
		 */
		if (!trunk->funcs.request_cancel_mux) {
			trunk_request_enter_unassigned(treq);
			trunk_request_free(&treq);
		}
		connection_signals_resume(tconn->pub.conn);
	}
		break;

	/*
	 *	We're already in the process of cancelling a
	 *	request, so ignore duplicate signals.
	 */
	case TRUNK_REQUEST_STATE_CANCEL:
	case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
	case TRUNK_REQUEST_STATE_CANCEL_SENT:
	case TRUNK_REQUEST_STATE_CANCEL_COMPLETE:
		break;

	/*
	 *	For any other state, we just release the request
	 *	from its current connection and free it.
	 */
	default:
		trunk_request_enter_unassigned(treq);
		trunk_request_free(&treq);
		break;
	}
}

/** Signal a partial cancel write
 *
 * Where there's high load, and the outbound write buffer is full
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_cancel_partial(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(IN_REQUEST_CANCEL_MUX(treq->pub.trunk),
				"%s can only be called from within request_cancel_mux handler", __FUNCTION__)) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_CANCEL:
		trunk_request_enter_cancel_partial(treq);
		break;

	default:
		return;
	}
}

/** Signal that a remote server has been notified of the cancellation
 *
 * Called from request_cancel_mux to indicate that the datastore has been informed
 * that the response is no longer needed.
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_cancel_sent(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(IN_REQUEST_CANCEL_MUX(treq->pub.trunk),
				"%s can only be called from within request_cancel_mux handler", __FUNCTION__)) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_CANCEL:
	case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
		trunk_request_enter_cancel_sent(treq);
		break;

	default:
		break;
	}
}

/** Signal that a remote server acked our cancellation
 *
 * Called from request_demux to indicate that it got an ack for the cancellation.
 *
 * @param[in] treq	to signal state change for.
 */
void trunk_request_signal_cancel_complete(trunk_request_t *treq)
{
	if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;

	if (!fr_cond_assert_msg(IN_REQUEST_DEMUX(treq->pub.trunk) || IN_REQUEST_CANCEL_MUX(treq->pub.trunk),
				"%s can only be called from within request_demux or request_cancel_mux handlers",
				__FUNCTION__)) return;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_CANCEL_SENT:
	/*
	 *	This is allowed, as we may not need to wait
	 *	for the database to ACK our cancellation
	 *	request.
	 *
	 *	Note: TRUNK_REQUEST_STATE_CANCEL_PARTIAL
	 *	is not allowed here, as that'd mean we'd half
	 *	written the cancellation request out to the
	 *	socket, and then decided to abandon it.
	 *
	 *	That'd leave the socket in an unusable state.
	 */
	case TRUNK_REQUEST_STATE_CANCEL:
		trunk_request_enter_cancel_complete(treq);
		break;

	default:
		break;
	}
}

/** If the trunk request is freed then update the target requests
 *
 * gperftools showed calling the request free function directly was slightly faster
 * than using talloc_free.
 *
 * @param[in] treq_to_free	request.
 */
void trunk_request_free(trunk_request_t **treq_to_free)
{
	trunk_request_t	*treq = *treq_to_free;
	trunk_t		*trunk = treq->pub.trunk;

	if (unlikely(!treq)) return;

	/*
	 *	The only valid states a trunk request can be
	 *	freed from.
	 */
	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_INIT:
	case TRUNK_REQUEST_STATE_UNASSIGNED:
	case TRUNK_REQUEST_STATE_COMPLETE:
	case TRUNK_REQUEST_STATE_FAILED:
	case TRUNK_REQUEST_STATE_CANCEL_COMPLETE:
		break;

	default:
		if (!fr_cond_assert(0)) return;
	}

	/*
	 *	Zero out the pointer to prevent double frees
	 */
	*treq_to_free = NULL;

	/*
	 *	Call the API client callback to free
	 *	any associated memory.
	 */
	DO_REQUEST_FREE(treq);

	/*
	 *	Update the last above/below target stats
	 *	We only do this when we alloc or free
	 *	connections, or on connection
	 *      state changes.
	 */
	trunk_requests_per_connection(NULL, NULL, treq->pub.trunk, fr_time(), false);

	/*
	 *	This tracks the total number of requests
	 *	allocated and not freed or returned to
	 *	the free list.
	 */
	if (fr_cond_assert(trunk->pub.req_alloc > 0)) trunk->pub.req_alloc--;

	/*
	 *	No cleanup delay, means cleanup immediately
	 */
	if (!fr_time_delta_ispos(trunk->conf.req_cleanup_delay)) {
		treq->pub.state = TRUNK_REQUEST_STATE_INIT;

#ifndef NDEBUG
		/*
		 *	Ensure anything parented off the treq
		 *	is freed.  We do this to trigger
		 *	the destructors for the log entries.
		 */
		talloc_free_children(treq);

		/*
		 *	State log should now be empty as entries
		 *	remove themselves from the dlist
		 *	on free.
		 */
		fr_assert_msg(fr_dlist_num_elements(&treq->log) == 0,
			      "Should have 0 remaining log entries, have %u", fr_dlist_num_elements(&treq->log));
#endif

		talloc_free(treq);
		return;
	}

	/*
	 *	Ensure anything parented off the treq
	 *	is freed.
	 */
	talloc_free_children(treq);

#ifndef NDEBUG
	/*
	 *	State log should now be empty as entries
	 *	remove themselves from the dlist
	 *	on free.
	 */
	fr_assert_msg(fr_dlist_num_elements(&treq->log) == 0,
		      "Should have 0 remaining log entries, have %u", fr_dlist_num_elements(&treq->log));
#endif

	/*
	 *
	 *  Return the trunk request back to the init state.
	 */
	*treq = (trunk_request_t){
		.pub = {
			.state = TRUNK_REQUEST_STATE_INIT,
			.trunk = treq->pub.trunk,
		},
		.cancel_reason = TRUNK_CANCEL_REASON_NONE,
		.last_freed = fr_time(),
#ifndef NDEBUG
		.log = treq->log        /* Keep the list head, to save reinitialisation */
#endif
	};

	/*
	 *	Insert at the head, so that we can free
	 *	requests that have been unused for N
	 *	seconds from the tail.
	 */
	fr_dlist_insert_tail(&trunk->free_requests, treq);
}

/** Actually free the trunk request
 *
 */
static int _trunk_request_free(trunk_request_t *treq)
{
	trunk_t	*trunk = treq->pub.trunk;

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_INIT:
	case TRUNK_REQUEST_STATE_UNASSIGNED:
		break;

	default:
		fr_assert(0);
		break;
	}

	fr_dlist_remove(&trunk->free_requests, treq);

	return 0;
}

/** (Pre-)Allocate a new trunk request
 *
 * If trunk->conf.req_pool_headers or trunk->conf.req_pool_size are not zero then the
 * request will be a talloc pool, which can be used to hold the preq.
 *
 * @note Do not use MEM to check the result of this allocated as it may fail for
 * non-fatal reasons.
 *
 * @param[in] trunk	to add request to.
 * @param[in] request	to wrap in a trunk request (treq).
 * @return
 *	- A newly allocated request.
 *	- NULL if too many requests are allocated.
 */
trunk_request_t *trunk_request_alloc(trunk_t *trunk, request_t *request)
{
	trunk_request_t *treq;

	/*
	 *	The number of treqs currently allocated
	 *	exceeds the maximum number allowed.
	 */
	if (trunk->conf.max_req_per_conn && trunk->conf.max) {
		uint64_t limit;

		limit = (uint64_t) trunk->conf.max_req_per_conn * trunk->conf.max;
		if (trunk->pub.req_alloc >= (limit + trunk->conf.max_backlog)) {
			RATE_LIMIT_LOCAL_ROPTIONAL(&trunk->limit_max_requests_alloc_log,
						   RWARN, WARN, "Refusing to alloc requests - "
						   "Limit of %"PRIu64" (max = %u * per_connection_max = %u) "
						   "plus %u backlog requests reached",
						   limit, trunk->conf.max, trunk->conf.max_req_per_conn,
						   trunk->conf.max_backlog);
			return NULL;
		}
	}

	/*
	 *	Allocate or reuse an existing request
	 */
	treq = fr_dlist_head(&trunk->free_requests);
	if (treq) {
		fr_dlist_remove(&trunk->free_requests, treq);
		fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_INIT);
		fr_assert(treq->pub.trunk == trunk);
		fr_assert(treq->pub.tconn == NULL);
		fr_assert(treq->cancel_reason == TRUNK_CANCEL_REASON_NONE);
		fr_assert(fr_time_gt(treq->last_freed, fr_time_wrap(0)));
		trunk->pub.req_alloc_reused++;
	} else {
		MEM(treq = talloc_pooled_object(trunk, trunk_request_t,
						trunk->conf.req_pool_headers, trunk->conf.req_pool_size));
		talloc_set_destructor(treq, _trunk_request_free);

		*treq = (trunk_request_t){
			.pub = {
				.state = TRUNK_REQUEST_STATE_INIT,
				.trunk = trunk
			},
			.cancel_reason = TRUNK_CANCEL_REASON_NONE
		};
		trunk->pub.req_alloc_new++;
#ifndef NDEBUG
		fr_dlist_init(&treq->log, trunk_request_state_log_t, entry);
#endif
	}

	trunk->pub.req_alloc++;
	treq->id = atomic_fetch_add_explicit(&request_counter, 1, memory_order_relaxed);
	/* heap_id	- initialised when treq inserted into pending */
	/* list		- empty */
	/* preq		- populated later */
	/* rctx		- populated later */
	treq->pub.request = request;

	return treq;
}

/** Enqueue a request that needs data written to the trunk
 *
 * When a request_t * needs to make an asynchronous request to an external datastore
 * it should call this function, specifying a preq (protocol request) containing
 * the data necessary to request information from the external datastore, and an
 * rctx (resume ctx) used to hold the decoded response and/or any error codes.
 *
 * After a treq is successfully enqueued it will either be assigned immediately
 * to the pending queue of a connection, or if no connections are available,
 * (depending on the trunk configuration) the treq will be placed in the trunk's
 * global backlog.
 *
 * After receiving a positive return code from this function the caller should
 * immediately yield, to allow the various timers and I/O handlers that drive tconn
 * (trunk connection) and treq state changes to be called.
 *
 * When a tconn becomes writable (or the trunk is configured to be always writable)
 * the #trunk_request_mux_t callback will be called to dequeue, encode and
 * send any pending requests for that tconn.  The #trunk_request_mux_t callback
 * is also responsible for tracking the outbound requests to allow the
 * #trunk_request_demux_t callback to match inbound responses with the original
 * treq.  Once the #trunk_request_mux_t callback is done processing the treq
 * it signals what state the treq should enter next using one of the
 * trunk_request_signal_* functions.
 *
 * When a tconn becomes readable the user specified #trunk_request_demux_t
 * callback is called to process any responses, match them with the original treq.
 * and signal what state they should enter next using one of the
 * trunk_request_signal_* functions.
 *
 * @param[in,out] treq_out	A trunk request handle.  If the memory pointed to
 *				is NULL, a new treq will be allocated.
 *				Otherwise treq should point to memory allocated
 *				with trunk_request_alloc.
 * @param[in] trunk		to enqueue request on.
 * @param[in] request		to enqueue.
 * @param[in] preq		Protocol request to write out.  Will be freed when
 *				treq is freed. Should ideally be parented by the
 *				treq if possible.
 *				Use #trunk_request_alloc for pre-allocation of
 *				the treq.
 * @param[in] rctx		The resume context to write any result to.
 * @return
 *	- TRUNK_ENQUEUE_OK.
 *	- TRUNK_ENQUEUE_IN_BACKLOG.
 *	- TRUNK_ENQUEUE_NO_CAPACITY.
 *	- TRUNK_ENQUEUE_DST_UNAVAILABLE
 *	- TRUNK_ENQUEUE_FAIL
 */
trunk_enqueue_t trunk_request_enqueue(trunk_request_t **treq_out, trunk_t *trunk,
					    request_t *request, void *preq, void *rctx)
{
	trunk_connection_t	*tconn = NULL;
	trunk_request_t	*treq;
	trunk_enqueue_t	ret;

	if (!fr_cond_assert_msg(!IN_HANDLER(trunk),
				"%s cannot be called within a handler", __FUNCTION__)) return TRUNK_ENQUEUE_FAIL;

	if (!fr_cond_assert_msg(!*treq_out || ((*treq_out)->pub.state == TRUNK_REQUEST_STATE_INIT),
				"%s requests must be in \"init\" state", __FUNCTION__)) return TRUNK_ENQUEUE_FAIL;

	/*
	 *	If delay_start was set, we may need
	 *	to insert the timer for the connection manager.
	 */
	if (unlikely(!trunk->started)) {
		if (trunk_start(trunk) < 0) return TRUNK_ENQUEUE_FAIL;
	}

	ret = trunk_request_check_enqueue(&tconn, trunk, request);
	switch (ret) {
	case TRUNK_ENQUEUE_OK:
		if (*treq_out) {
			treq = *treq_out;
		} else {
			*treq_out = treq = trunk_request_alloc(trunk, request);
			if (!treq) return TRUNK_ENQUEUE_FAIL;
		}
		treq->pub.preq = preq;
		treq->pub.rctx = rctx;
		if (trunk->conf.always_writable) {
			connection_signals_pause(tconn->pub.conn);
			trunk_request_enter_pending(treq, tconn, true);
			trunk_connection_writable(tconn);
			connection_signals_resume(tconn->pub.conn);
		} else {
			trunk_request_enter_pending(treq, tconn, true);
		}
		break;

	case TRUNK_ENQUEUE_IN_BACKLOG:
		if (*treq_out) {
			treq = *treq_out;
		} else {
			*treq_out = treq = trunk_request_alloc(trunk, request);
			if (!treq) return TRUNK_ENQUEUE_FAIL;
		}
		treq->pub.preq = preq;
		treq->pub.rctx = rctx;
		trunk_request_enter_backlog(treq, true);
		break;

	default:
		/*
		 *	If a trunk request was provided
		 *	populate the preq and rctx fields
		 *	so that if it's freed with
		 *	trunk_request_free, the free
		 *	function works as intended.
		 */
		if (*treq_out) {
			treq = *treq_out;
			treq->pub.preq = preq;
			treq->pub.rctx = rctx;
		}
		return ret;
	}

	return ret;
}

/** Re-enqueue a request on the same connection
 *
 * If the treq has been sent, we assume that we're being signalled to requeue
 * because something outside of the trunk API has determined that a retransmission
 * is required.  The easiest way to perform that retransmission is to clean up
 * any tracking information for the request, and the requeue it for transmission.
 *
 * IF re-queueing fails, the request will enter the fail state.  It should not be
 * accessed if this occurs.
 *
 * @param[in] treq	to requeue (retransmit).
 * @return
 *	- TRUNK_ENQUEUE_OK.
 *	- TRUNK_ENQUEUE_DST_UNAVAILABLE - Connection cannot service requests.
 *	- TRUNK_ENQUEUE_FAIL - Request isn't in a valid state to be reassigned.
 */
trunk_enqueue_t trunk_request_requeue(trunk_request_t *treq)
{
	trunk_connection_t	*tconn = treq->pub.tconn;	/* Existing conn */

	if (!tconn) return TRUNK_ENQUEUE_FAIL;

	if (!IS_PROCESSING(tconn)) {
		trunk_request_enter_failed(treq);
		return TRUNK_ENQUEUE_DST_UNAVAILABLE;
	}

	switch (treq->pub.state) {
	case TRUNK_REQUEST_STATE_PARTIAL:
	case TRUNK_REQUEST_STATE_SENT:
	case TRUNK_REQUEST_STATE_REAPABLE:
		connection_signals_pause(tconn->pub.conn);
		trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_REQUEUE);
		trunk_request_enter_pending(treq, tconn, false);
		if (treq->pub.trunk->conf.always_writable) {
			trunk_connection_writable(tconn);
		}
		connection_signals_resume(tconn->pub.conn);
		break;

	case TRUNK_REQUEST_STATE_BACKLOG:	/* Do nothing.... */
	case TRUNK_REQUEST_STATE_PENDING:	/* Do nothing.... */
		break;

	default:
		trunk_request_enter_failed(treq);
		return TRUNK_ENQUEUE_FAIL;
	}

	return TRUNK_ENQUEUE_OK;
}

/** Enqueue additional requests on a specific connection
 *
 * This may be used to create a series of requests on a single connection, or to generate
 * in-band status checks.
 *
 * @note If conf->always_writable, then the muxer will be called immediately.  The caller
 *	 must be able to handle multiple calls to its muxer gracefully.
 *
 * @param[in,out] treq_out	A trunk request handle.  If the memory pointed to
 *				is NULL, a new treq will be allocated.
 *				Otherwise treq should point to memory allocated
 *				with trunk_request_alloc.
 * @param[in] tconn		to enqueue request on.
 * @param[in] request		to enqueue.
 * @param[in] preq		Protocol request to write out.  Will be freed when
 *				treq is freed. Should ideally be parented by the
 *				treq if possible.
 *				Use #trunk_request_alloc for pre-allocation of
 *				the treq.
 * @param[in] rctx		The resume context to write any result to.
 * @param[in] ignore_limits	Ignore max_req_per_conn.  Useful to force status
 *				checks through even if the connection is at capacity.
 *				Will also allow enqueuing on "inactive", "draining",
 *				"draining-to-free" connections.
 * @return
 *	- TRUNK_ENQUEUE_OK.
 *	- TRUNK_ENQUEUE_NO_CAPACITY - At max_req_per_conn_limit
 *	- TRUNK_ENQUEUE_DST_UNAVAILABLE - Connection cannot service requests.
 */
trunk_enqueue_t trunk_request_enqueue_on_conn(trunk_request_t **treq_out, trunk_connection_t *tconn,
						    request_t *request, void *preq, void *rctx,
						    bool ignore_limits)
{
	trunk_request_t	*treq;
	trunk_t		*trunk = tconn->pub.trunk;

	if (!fr_cond_assert_msg(!*treq_out || ((*treq_out)->pub.state == TRUNK_REQUEST_STATE_INIT),
				"%s requests must be in \"init\" state", __FUNCTION__)) return TRUNK_ENQUEUE_FAIL;

	if (!IS_SERVICEABLE(tconn)) return TRUNK_ENQUEUE_DST_UNAVAILABLE;

	/*
	 *	Limits check
	 */
	if (!ignore_limits) {
		if (trunk->conf.max_req_per_conn &&
		    (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) >=
		     trunk->conf.max_req_per_conn)) return TRUNK_ENQUEUE_NO_CAPACITY;

		if (tconn->pub.state != TRUNK_CONN_ACTIVE) return TRUNK_ENQUEUE_NO_CAPACITY;
	}

	if (*treq_out) {
		treq = *treq_out;
	} else {
		MEM(*treq_out = treq = trunk_request_alloc(trunk, request));
	}

	treq->pub.preq = preq;
	treq->pub.rctx = rctx;
	treq->bound_to_conn = true;	/* Don't let the request be transferred */

	if (trunk->conf.always_writable) {
		connection_signals_pause(tconn->pub.conn);
		trunk_request_enter_pending(treq, tconn, true);
		trunk_connection_writable(tconn);
		connection_signals_resume(tconn->pub.conn);
	} else {
		trunk_request_enter_pending(treq, tconn, true);
	}

	return TRUNK_ENQUEUE_OK;
}

#ifndef NDEBUG
/** Used for sanity checks to ensure all log entries have been freed
 *
 */
static int _state_log_entry_free(trunk_request_state_log_t *slog)
{
	fr_dlist_remove(slog->log_head, slog);

	return 0;
}

void trunk_request_state_log_entry_add(char const *function, int line,
				       trunk_request_t *treq, trunk_request_state_t new)
{
	trunk_request_state_log_t	*slog = NULL;

	if (fr_dlist_num_elements(&treq->log) >= TRUNK_REQUEST_STATE_LOG_MAX) {
		slog = fr_dlist_head(&treq->log);
		fr_assert_msg(slog, "slog list head NULL but element counter was %u",
			      fr_dlist_num_elements(&treq->log));
		(void)fr_dlist_remove(&treq->log, slog);	/* Returns NULL when removing the list head */
		memset(slog, 0, sizeof(*slog));
	} else {
		MEM(slog = talloc_zero(treq, trunk_request_state_log_t));
		talloc_set_destructor(slog, _state_log_entry_free);
	}

	slog->log_head = &treq->log;
	slog->from = treq->pub.state;
	slog->to = new;
	slog->function = function;
	slog->line = line;
	if (treq->pub.tconn) {
		slog->tconn = treq->pub.tconn;
		slog->tconn_id = treq->pub.tconn->pub.conn->id;
		slog->tconn_state = treq->pub.tconn->pub.state;
	}

	fr_dlist_insert_tail(&treq->log, slog);

}

void trunk_request_state_log(fr_log_t const *log, fr_log_type_t log_type, char const *file, int line,
				trunk_request_t const *treq)
{
	trunk_request_state_log_t	*slog = NULL;

	int i;

	for (slog = fr_dlist_head(&treq->log), i = 0;
	     slog;
	     slog = fr_dlist_next(&treq->log, slog), i++) {
		fr_log(log, log_type, file, line, "[%u] %s:%i - in conn %"PRIu64" in state %s - %s -> %s",
		       i, slog->function, slog->line,
		       slog->tconn_id,
		       slog->tconn ? fr_table_str_by_value(trunk_connection_states,
		       					    slog->tconn_state, "<INVALID>") : "none",
		       fr_table_str_by_value(trunk_request_states, slog->from, "<INVALID>"),
		       fr_table_str_by_value(trunk_request_states, slog->to, "<INVALID>"));
	}
}
#endif

/** Return the count number of connections in the specified states
 *
 * @param[in] trunk		to retrieve counts for.
 * @param[in] conn_state	One or more #trunk_connection_state_t states or'd together.
 * @return The number of connections in the specified states.
 */
uint16_t trunk_connection_count_by_state(trunk_t *trunk, int conn_state)
{
	uint16_t count = 0;

	if (conn_state & TRUNK_CONN_INIT) count += fr_dlist_num_elements(&trunk->init);
	if (conn_state & TRUNK_CONN_CONNECTING) count += fr_dlist_num_elements(&trunk->connecting);
	if (conn_state & TRUNK_CONN_ACTIVE) count += fr_minmax_heap_num_elements(trunk->active);
	if (conn_state & TRUNK_CONN_FULL) count += fr_dlist_num_elements(&trunk->full);
	if (conn_state & TRUNK_CONN_INACTIVE) count += fr_dlist_num_elements(&trunk->inactive);
	if (conn_state & TRUNK_CONN_INACTIVE_DRAINING) count += fr_dlist_num_elements(&trunk->inactive_draining);
	if (conn_state & TRUNK_CONN_CLOSED) count += fr_dlist_num_elements(&trunk->closed);
	if (conn_state & TRUNK_CONN_DRAINING) count += fr_dlist_num_elements(&trunk->draining);
	if (conn_state & TRUNK_CONN_DRAINING_TO_FREE) count += fr_dlist_num_elements(&trunk->draining_to_free);

	return count;
}

/** Return the count number of requests associated with a trunk connection
 *
 * @param[in] tconn		to return request count for.
 * @param[in] req_state		One or more request states or'd together.
 *
 * @return The number of requests in the specified states, associated with a tconn.
 */
uint32_t trunk_request_count_by_connection(trunk_connection_t const *tconn, int req_state)
{
	uint32_t count = 0;

	if (req_state & TRUNK_REQUEST_STATE_PENDING) count += fr_heap_num_elements(tconn->pending);
	if (req_state & TRUNK_REQUEST_STATE_PARTIAL) count += tconn->partial ? 1 : 0;
	if (req_state & TRUNK_REQUEST_STATE_SENT) count += fr_dlist_num_elements(&tconn->sent);
	if (req_state & TRUNK_REQUEST_STATE_REAPABLE) count += fr_dlist_num_elements(&tconn->reapable);
	if (req_state & TRUNK_REQUEST_STATE_CANCEL) count += fr_dlist_num_elements(&tconn->cancel);
	if (req_state & TRUNK_REQUEST_STATE_CANCEL_PARTIAL) count += tconn->cancel_partial ? 1 : 0;
	if (req_state & TRUNK_REQUEST_STATE_CANCEL_SENT) count += fr_dlist_num_elements(&tconn->cancel_sent);

	return count;
}

/** Automatically mark a connection as inactive
 *
 * @param[in] tconn	to potentially mark as inactive.
 */
static inline void trunk_connection_auto_full(trunk_connection_t *tconn)
{
	trunk_t	*trunk = tconn->pub.trunk;
	uint32_t	count;

	if (tconn->pub.state != TRUNK_CONN_ACTIVE) return;

	/*
	 *	Enforces max_req_per_conn
	 */
	if (trunk->conf.max_req_per_conn > 0) {
		count = trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL);
		if (count >= trunk->conf.max_req_per_conn) trunk_connection_enter_full(tconn);
	}
}

/** Return whether a trunk connection should currently be considered full
 *
 * @param[in] tconn	to check.
 * @return
 *	- true if the connection is full.
 *	- false if the connection is not full.
 */
static inline bool trunk_connection_is_full(trunk_connection_t *tconn)
{
	trunk_t	*trunk = tconn->pub.trunk;
	uint32_t	count;

	/*
	 *	Enforces max_req_per_conn
	 */
	count = trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL);
	if ((trunk->conf.max_req_per_conn == 0) || (count < trunk->conf.max_req_per_conn)) return false;

	return true;
}

/** Automatically mark a connection as active or reconnect it
 *
 * @param[in] tconn	to potentially mark as active or reconnect.
 */
static inline void trunk_connection_auto_unfull(trunk_connection_t *tconn)
{
	if (tconn->pub.state != TRUNK_CONN_FULL) return;

	/*
	 *	Enforces max_req_per_conn
	 */
	if (!trunk_connection_is_full(tconn)) trunk_connection_enter_active(tconn);
}

/** A connection is readable.  Call the request_demux function to read pending requests
 *
 */
static inline void trunk_connection_readable(trunk_connection_t *tconn)
{
	trunk_t *trunk = tconn->pub.trunk;

	DO_REQUEST_DEMUX(tconn);
}

/** A connection is writable.  Call the request_mux function to write pending requests
 *
 */
static inline void trunk_connection_writable(trunk_connection_t *tconn)
{
	trunk_t *trunk = tconn->pub.trunk;

	/*
	 *	Call the cancel_sent function (if we have one)
	 *      to inform a backend datastore we no longer
	 *	care about the result
	 */
	if (trunk->funcs.request_cancel_mux && trunk_request_count_by_connection(tconn,
										    TRUNK_REQUEST_STATE_CANCEL |
										    TRUNK_REQUEST_STATE_CANCEL_PARTIAL)) {
		DO_REQUEST_CANCEL_MUX(tconn);
	}
	if (!trunk_request_count_by_connection(tconn,
						  TRUNK_REQUEST_STATE_PENDING |
						  TRUNK_REQUEST_STATE_PARTIAL)) return;
	DO_REQUEST_MUX(tconn);
}

/** Update the registrations for I/O events we're interested in
 *
 */
static void trunk_connection_event_update(trunk_connection_t *tconn)
{
	trunk_t			*trunk = tconn->pub.trunk;
	trunk_connection_event_t	events = TRUNK_CONN_EVENT_NONE;

	switch (tconn->pub.state) {
	/*
	 *	We only register I/O events if the trunk connection is
	 *	in one of these states.
	 *
	 *	For the other states the trunk shouldn't be processing
	 *	requests.
	 */
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_INACTIVE_DRAINING:
	case TRUNK_CONN_DRAINING:
	case TRUNK_CONN_DRAINING_TO_FREE:
		/*
		 *	If the connection is always writable,
		 *	then we don't care about write events.
		 */
		if (!trunk->conf.always_writable &&
		    trunk_request_count_by_connection(tconn,
							 TRUNK_REQUEST_STATE_PARTIAL |
						       	 TRUNK_REQUEST_STATE_PENDING |
							 (trunk->funcs.request_cancel_mux ?
							 TRUNK_REQUEST_STATE_CANCEL |
							 TRUNK_REQUEST_STATE_CANCEL_PARTIAL : 0)) > 0) {
			events |= TRUNK_CONN_EVENT_WRITE;
		}

		if (trunk_request_count_by_connection(tconn,
							 TRUNK_REQUEST_STATE_SENT |
							 (trunk->funcs.request_cancel_mux ?
							 TRUNK_REQUEST_STATE_CANCEL_SENT : 0)) > 0) {
			events |= TRUNK_CONN_EVENT_READ;
		}
		break;

	default:
		break;
	}

	if (tconn->events != events) {
		/*
		 *	There may be a fatal error which results
		 *	in the connection being freed.
		 *
		 *	Stop that from happening until after
		 *	we're done using it.
		 */
		connection_signals_pause(tconn->pub.conn);
		DO_CONNECTION_NOTIFY(tconn, events);
		tconn->events = events;
		connection_signals_resume(tconn->pub.conn);
	}
}

/** Remove a trunk connection from whichever list it's currently in
 *
 * @param[in] tconn to remove.
 */
static void trunk_connection_remove(trunk_connection_t *tconn)
{
	trunk_t *trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	{
		int ret;

		ret = fr_minmax_heap_extract(trunk->active, tconn);
		if (!fr_cond_assert_msg(ret == 0, "Failed extracting conn from active heap: %s", fr_strerror())) return;
	}
		return;

	case TRUNK_CONN_INIT:
		fr_dlist_remove(&trunk->init, tconn);
		break;

	case TRUNK_CONN_CONNECTING:
		fr_dlist_remove(&trunk->connecting, tconn);
		return;

	case TRUNK_CONN_CLOSED:
		fr_dlist_remove(&trunk->closed, tconn);
		return;

	case TRUNK_CONN_FULL:
		fr_dlist_remove(&trunk->full, tconn);
		return;

	case TRUNK_CONN_INACTIVE:
		fr_dlist_remove(&trunk->inactive, tconn);
		return;

	case TRUNK_CONN_INACTIVE_DRAINING:
		fr_dlist_remove(&trunk->inactive_draining, tconn);
		return;

	case TRUNK_CONN_DRAINING:
		fr_dlist_remove(&trunk->draining, tconn);
		return;

	case TRUNK_CONN_DRAINING_TO_FREE:
		fr_dlist_remove(&trunk->draining_to_free, tconn);
		return;

	case TRUNK_CONN_HALTED:
		return;
	}
}

/** Transition a connection to the full state
 *
 * Called whenever a trunk connection is at the maximum number of requests.
 * Removes the connection from the connected heap, and places it in the full list.
 */
static void trunk_connection_enter_full(trunk_connection_t *tconn)
{
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_FULL);
	}

	fr_dlist_insert_head(&trunk->full, tconn);
	CONN_STATE_TRANSITION(TRUNK_CONN_FULL, DEBUG2);
}

/** Transition a connection to the inactive state
 *
 * Called whenever the API client wants to stop new requests being enqueued
 * on a trunk connection.
 */
static void trunk_connection_enter_inactive(trunk_connection_t *tconn)
{
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_INACTIVE);
	}

	fr_dlist_insert_head(&trunk->inactive, tconn);
	CONN_STATE_TRANSITION(TRUNK_CONN_INACTIVE, DEBUG2);
}

/** Transition a connection to the inactive-draining state
 *
 * Called whenever the trunk manager wants to drain an inactive connection
 * of its requests.
 */
static void trunk_connection_enter_inactive_draining(trunk_connection_t *tconn)
{
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_DRAINING:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_INACTIVE_DRAINING);
	}

	fr_dlist_insert_head(&trunk->inactive_draining, tconn);
	CONN_STATE_TRANSITION(TRUNK_CONN_INACTIVE_DRAINING, INFO);

	/*
	 *	Immediately re-enqueue all pending
	 *	requests, so the connection is drained
	 *	quicker.
	 */
	trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_PENDING, 0, false);
}

/** Transition a connection to the draining state
 *
 * Removes the connection from the active heap so it won't be assigned any new
 * connections.
 */
static void trunk_connection_enter_draining(trunk_connection_t *tconn)
{
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_INACTIVE_DRAINING:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_DRAINING);
	}

	fr_dlist_insert_head(&trunk->draining, tconn);
	CONN_STATE_TRANSITION(TRUNK_CONN_DRAINING, INFO);

	/*
	 *	Immediately re-enqueue all pending
	 *	requests, so the connection is drained
	 *	quicker.
	 */
	trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_PENDING, 0, false);
}

/** Transition a connection to the draining-to-reconnect state
 *
 * Removes the connection from the active heap so it won't be assigned any new
 * connections.
 */
static void trunk_connection_enter_draining_to_free(trunk_connection_t *tconn)
{
	trunk_t		*trunk = tconn->pub.trunk;

	if (tconn->lifetime_ev) fr_event_timer_delete(&tconn->lifetime_ev);

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_INACTIVE_DRAINING:
	case TRUNK_CONN_DRAINING:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_DRAINING_TO_FREE);
	}

	fr_dlist_insert_head(&trunk->draining_to_free, tconn);
	CONN_STATE_TRANSITION(TRUNK_CONN_DRAINING_TO_FREE, INFO);

	/*
	 *	Immediately re-enqueue all pending
	 *	requests, so the connection is drained
	 *	quicker.
	 */
	trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_PENDING, 0, false);
}


/** Transition a connection back to the active state
 *
 * This should only be called on a connection which is in the full state,
 * inactive state, draining state or connecting state.
 */
static void trunk_connection_enter_active(trunk_connection_t *tconn)
{
	trunk_t		*trunk = tconn->pub.trunk;
	int			ret;

	switch (tconn->pub.state) {
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_INACTIVE_DRAINING:
	case TRUNK_CONN_DRAINING:
		trunk_connection_remove(tconn);
		break;

	case TRUNK_CONN_INIT:
	case TRUNK_CONN_CONNECTING:
		trunk_connection_remove(tconn);
		fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_ACTIVE);
	}

	ret = fr_minmax_heap_insert(trunk->active, tconn);	/* re-insert into the active heap*/
	if (!fr_cond_assert_msg(ret == 0, "Failed inserting connection into active heap: %s", fr_strerror())) {
		trunk_connection_enter_inactive_draining(tconn);
		return;
	}

	CONN_STATE_TRANSITION(TRUNK_CONN_ACTIVE, DEBUG2);

	/*
	 *	Reorder the connections
	 */
	CONN_REORDER(tconn);

	/*
	 *	Rebalance requests
	 */
	trunk_rebalance(trunk);

	/*
	 *	We place requests into the backlog
	 *      because there were no connections
	 *	available to handle them.
	 *
	 *	If a connection has become active
	 *	chances are those backlogged requests
	 *      can now be enqueued, so try and do
	 *	that now.
	 *
	 *	If there's requests sitting in the
	 *	backlog indefinitely, it's because
	 *	they were inserted there erroneously
	 *	when there were active connections
	 *	which could have handled them.
	 */
	trunk_backlog_drain(trunk);
}

/** Connection transitioned to the the init state
 *
 * Reflect the connection state change in the lists we use to track connections.
 *
 * @note This function is only called from the connection API as a watcher.
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_init(UNUSED connection_t *conn,
				      UNUSED connection_state_t prev,
				      UNUSED connection_state_t state,
				      void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_HALTED:
		break;

	case TRUNK_CONN_CLOSED:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_INIT);
	}

	fr_dlist_insert_head(&trunk->init, tconn);
	CONN_STATE_TRANSITION(TRUNK_CONN_INIT, DEBUG2);
}

/** Connection transitioned to the connecting state
 *
 * Reflect the connection state change in the lists we use to track connections.
 *
 * @note This function is only called from the connection API as a watcher.
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_connecting(UNUSED connection_t *conn,
					    UNUSED connection_state_t prev,
					    UNUSED connection_state_t state,
					    void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_INIT:
	case TRUNK_CONN_CLOSED:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_CONNECTING);
	}

	/*
	 *	If a connection just entered the
	 *	connecting state, it should have
	 *	no requests associated with it.
	 */
	fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);

	fr_dlist_insert_head(&trunk->connecting, tconn);	/* MUST remain a head insertion for reconnect logic */
	CONN_STATE_TRANSITION(TRUNK_CONN_CONNECTING, INFO);
}

/** Connection transitioned to the shutdown state
 *
 * If we're not already in the draining-to-free state, transition there now.
 *
 * The idea is that if something signalled the connection to shutdown, we need
 * to reflect that by dequeuing any pending requests, not accepting new ones,
 * and waiting for the existing requests to complete.
 *
 * @note This function is only called from the connection API as a watcher.
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_shutdown(UNUSED connection_t *conn,
					  UNUSED connection_state_t prev,
					  UNUSED connection_state_t state,
					  void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);

	switch (tconn->pub.state) {
	case TRUNK_CONN_DRAINING_TO_FREE:	/* Do Nothing */
		return;

	case TRUNK_CONN_ACTIVE:		/* Transition to draining-to-free */
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_INACTIVE_DRAINING:
	case TRUNK_CONN_DRAINING:
		break;

	case TRUNK_CONN_INIT:
	case TRUNK_CONN_CONNECTING:
	case TRUNK_CONN_CLOSED:
	case TRUNK_CONN_HALTED:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_DRAINING_TO_FREE);
	}

	trunk_connection_enter_draining_to_free(tconn);
}

/** Trigger a reconnection of the trunk connection
 *
 * @param[in] el	Event list the timer was inserted into.
 * @param[in] now	Current time.
 * @param[in] uctx	The tconn.
 */
static void  _trunk_connection_lifetime_expire(UNUSED fr_event_list_t *el, UNUSED fr_time_t now, void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);

	trunk_connection_enter_draining_to_free(tconn);
}

/** Connection transitioned to the connected state
 *
 * Reflect the connection state change in the lists we use to track connections.
 *
 * @note This function is only called from the connection API as a watcher.
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_connected(UNUSED connection_t *conn,
					   UNUSED connection_state_t prev,
					   UNUSED connection_state_t state,
					   void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);
	trunk_t		*trunk = tconn->pub.trunk;

	/*
	 *	If a connection was just connected,
	 *	it should have no requests associated
	 *	with it.
	 */
	fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);

 	/*
	 *	Set here, as the active state can
	 *	be transitioned to from full and
	 *	draining too.
	 */
	trunk->pub.last_connected = fr_time();

	/*
	 *	Insert a timer to reconnect the
	 *	connection periodically.
	 */
	if (fr_time_delta_ispos(trunk->conf.lifetime)) {
		if (fr_event_timer_in(tconn, trunk->el, &tconn->lifetime_ev,
				       trunk->conf.lifetime, _trunk_connection_lifetime_expire, tconn) < 0) {
			PERROR("Failed inserting connection reconnection timer event, halting connection");
			connection_signal_shutdown(tconn->pub.conn);
			return;
		}
	}

 	trunk_connection_enter_active(tconn);
}

/** Connection failed after it was connected
 *
 * Reflect the connection state change in the lists we use to track connections.
 *
 * @note This function is only called from the connection API as a watcher.
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_closed(UNUSED connection_t *conn,
			  		UNUSED connection_state_t prev,
					UNUSED connection_state_t state,
					void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);
	trunk_t		*trunk = tconn->pub.trunk;
	bool			need_requeue = false;

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
	case TRUNK_CONN_INACTIVE:
	case TRUNK_CONN_INACTIVE_DRAINING:
	case TRUNK_CONN_DRAINING:
	case TRUNK_CONN_DRAINING_TO_FREE:
		need_requeue = true;
		trunk_connection_remove(tconn);
		break;

	case TRUNK_CONN_INIT:			/* Initialisation failed */
	case TRUNK_CONN_CONNECTING:
		trunk_connection_remove(tconn);
		fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
		break;

	case TRUNK_CONN_CLOSED:
	case TRUNK_CONN_HALTED:	/* Can't move backwards? */
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_CLOSED);
	}

	fr_dlist_insert_head(&trunk->closed, tconn);	/* MUST remain a head insertion for reconnect logic */
	CONN_STATE_TRANSITION(TRUNK_CONN_CLOSED, INFO);

	/*
	 *	Now *AFTER* the connection has been
	 *	removed from the active, pool
	 *	re-enqueue the requests.
	 */
	if (need_requeue) trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_ALL, 0, true);

	/*
	 *	There should be no requests left on this
	 *	connection.  They should have all been
	 *	moved off or failed.
	 */
	fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);

	/*
	 *	Clear statistics and flags
	 */
	tconn->sent_count = 0;

	/*
	 *	Remove the reconnect event
	 */
	if (fr_time_delta_ispos(trunk->conf.lifetime)) fr_event_timer_delete(&tconn->lifetime_ev);

	/*
	 *	Remove the I/O events
	 */
	trunk_connection_event_update(tconn);
}

/** Connection failed
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_failed(connection_t *conn,
					connection_state_t prev,
					connection_state_t state,
					void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);
	trunk_t		*trunk = tconn->pub.trunk;

	/*
	 *	Need to set this first as it
	 *	determines whether requests are
	 *	re-queued or fail outright.
	 */
	trunk->pub.last_failed = fr_time();

	/*
	 *	Failed in the init state, transition the
	 *	connection to closed, else we get an
	 *	INIT -> INIT transition which triggers
	 *	an assert.
	 */
	if (prev == CONNECTION_STATE_INIT) _trunk_connection_on_closed(conn, prev, state, uctx);

	/*
	 *	See what the state of the trunk is
	 *	if there are no connections that could
	 *	potentially accept requests in the near
	 *	future, then fail all the requests in the
	 *	trunk backlog.
	 */
	if ((state == CONNECTION_STATE_CONNECTED) &&
	    (trunk_connection_count_by_state(trunk,
						(TRUNK_CONN_ACTIVE |
						 TRUNK_CONN_FULL |
						 TRUNK_CONN_DRAINING)) == 0)) trunk_backlog_drain(trunk);
}

/** Connection transitioned to the halted state
 *
 * Remove the connection remove all lists, as it's likely about to be freed.
 *
 * Setting the trunk back to the init state ensures that if the code is ever
 * refactored and #connection_signal_reconnect is used after a connection
 * is halted, then everything is maintained in a valid state.
 *
 * @note This function is only called from the connection API as a watcher.
 *
 * @param[in] conn	The connection which changes state.
 * @param[in] prev	The connection is was in.
 * @param[in] state	The connection is now in.
 * @param[in] uctx	The trunk_connection_t wrapping the connection.
 */
static void _trunk_connection_on_halted(UNUSED connection_t *conn,
					UNUSED connection_state_t prev,
					UNUSED connection_state_t state,
					void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);
	trunk_t		*trunk = tconn->pub.trunk;

	switch (tconn->pub.state) {
	case TRUNK_CONN_INIT:
	case TRUNK_CONN_CLOSED:
		trunk_connection_remove(tconn);
		break;

	default:
		CONN_BAD_STATE_TRANSITION(TRUNK_CONN_HALTED);
	}

	/*
	 *	It began life in the halted state,
	 *	and will end life in the halted state.
	 */
	CONN_STATE_TRANSITION(TRUNK_CONN_HALTED, DEBUG2);

	/*
	 *	There should be no requests left on this
	 *	connection.  They should have all been
	 *	moved off or failed.
	 */
	fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);

	/*
	 *	And free the connection...
	 */
	if (trunk->in_handler) {
		/*
		 *	...later.
		 */
		fr_dlist_insert_tail(&trunk->to_free, tconn);
		return;
	}
	talloc_free(tconn);
}

/** Free a connection
 *
 * Enforces orderly free order of children of the tconn
 */
static int _trunk_connection_free(trunk_connection_t *tconn)
{
	fr_assert(tconn->pub.state == TRUNK_CONN_HALTED);
	fr_assert(!fr_dlist_entry_in_list(&tconn->entry));	/* Should not be in a list */

	/*
	 *	Loop over all the requests we gathered
	 *	and transition them to the failed state,
	 *	freeing them.
	 *
	 *	Usually, requests will be re-queued when
	 *	a connection enters the closed state,
	 *	but in this case because the whole trunk
	 *	is being freed, we don't bother, and
	 *	just signal to the API client that the
	 *	requests failed.
	 */
	if (tconn->pub.trunk->freeing) {
		fr_dlist_head_t	to_fail;
		trunk_request_t *treq = NULL;

		fr_dlist_talloc_init(&to_fail, trunk_request_t, entry);

		/*
		 *	Remove requests from this connection
		 */
		trunk_connection_requests_dequeue(&to_fail, tconn, TRUNK_REQUEST_STATE_ALL, 0);
		while ((treq = fr_dlist_next(&to_fail, treq))) {
			trunk_request_t *prev;

			prev = fr_dlist_remove(&to_fail, treq);
			trunk_request_enter_failed(treq);
			treq = prev;
		}
	}

	/*
	 *	Ensure we're not signalled by the connection
	 *	as it processes its backlog of state changes,
	 *	as we are about to be freed.
	 */
	connection_del_watch_pre(tconn->pub.conn, CONNECTION_STATE_INIT, _trunk_connection_on_init);
	connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTING, _trunk_connection_on_connecting);
	connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTED, _trunk_connection_on_connected);
	connection_del_watch_pre(tconn->pub.conn, CONNECTION_STATE_CLOSED, _trunk_connection_on_closed);
	connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_SHUTDOWN, _trunk_connection_on_shutdown);
	connection_del_watch_pre(tconn->pub.conn, CONNECTION_STATE_FAILED, _trunk_connection_on_failed);
	connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_HALTED, _trunk_connection_on_halted);

	/*
	 *	This may return -1, indicating the free was deferred
	 *	this is fine.  It just means the conn will be freed
	 *	after all the handlers have exited.
	 */
	(void)talloc_free(tconn->pub.conn);
	tconn->pub.conn = NULL;

	return 0;
}

/** Attempt to spawn a new connection
 *
 * Calls the API client's alloc() callback to create a new connection_t,
 * then inserts the connection into the 'connecting' list.
 *
 * @param[in] trunk	to spawn connection in.
 * @param[in] now	The current time.
 */
static int trunk_connection_spawn(trunk_t *trunk, fr_time_t now)
{
	trunk_connection_t	*tconn;


	/*
	 *	Call the API client's callback to create
	 *	a new connection_t.
	 */
	MEM(tconn = talloc_zero(trunk, trunk_connection_t));
	tconn->pub.trunk = trunk;
	tconn->pub.state = TRUNK_CONN_HALTED;	/* All connections start in the halted state */

	/*
	 *	Allocate a new connection_t or fail.
	 */
	DO_CONNECTION_ALLOC(tconn);

	MEM(tconn->pending = fr_heap_talloc_alloc(tconn, _trunk_request_prioritise, trunk_request_t, heap_id, 0));
	fr_dlist_talloc_init(&tconn->sent, trunk_request_t, entry);
	fr_dlist_talloc_init(&tconn->reapable, trunk_request_t, entry);
	fr_dlist_talloc_init(&tconn->cancel, trunk_request_t, entry);
	fr_dlist_talloc_init(&tconn->cancel_sent, trunk_request_t, entry);

	/*
	 *	OK, we have the connection, now setup watch
	 *	points so we know when it changes state.
	 *
	 *	This lets us automatically move the tconn
	 *	between the different lists in the trunk
	 *	with minimum extra code.
	 */
	connection_add_watch_pre(tconn->pub.conn, CONNECTION_STATE_INIT,
				    _trunk_connection_on_init, false, tconn);		/* Before init() has been called */

	connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTING,
				     _trunk_connection_on_connecting, false, tconn);	/* After init() has been called */

	connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTED,
				     _trunk_connection_on_connected, false, tconn);	/* After open() has been called */

	connection_add_watch_pre(tconn->pub.conn, CONNECTION_STATE_CLOSED,
				    _trunk_connection_on_closed, false, tconn);		/* Before close() has been called */

	connection_add_watch_pre(tconn->pub.conn, CONNECTION_STATE_FAILED,
				    _trunk_connection_on_failed, false, tconn);		/* Before failed() has been called */

	connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_SHUTDOWN,
				     _trunk_connection_on_shutdown, false, tconn);	/* After shutdown() has been called */

	connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_HALTED,
				     _trunk_connection_on_halted, false, tconn);	/* About to be freed */

	talloc_set_destructor(tconn, _trunk_connection_free);

	connection_signal_init(tconn->pub.conn);	/* annnnd GO! */

	trunk->pub.last_open = now;

	return 0;
}

/** Pop a cancellation request off a connection's cancellation queue
 *
 * The request we return is advanced by the request moving out of the
 * cancel state and into the cancel_sent or cancel_complete state.
 *
 * One of these signalling functions must be called after the request
 * has been popped:
 *
 * - #trunk_request_signal_cancel_sent
 *   The remote datastore has been informed, but we need to wait for acknowledgement.
 *   The #trunk_request_demux_t callback must handle the acks calling
 *   #trunk_request_signal_cancel_complete when an ack is received.
 *
 * - #trunk_request_signal_cancel_complete
 *   The request was cancelled and we don't need to wait, clean it up immediately.
 *
 * @param[out] treq_out	to process
 * @param[in] tconn	Connection to drain cancellation request from.
 * @return
 *	- 1 if no more requests.
 *	- 0 if a new request was written to treq_out.
 *	- -1 if the connection was previously freed.  Caller *MUST NOT* touch any
 *	  memory or requests associated with the connection.
 *	- -2 if called outside of the cancel muxer.
 */
int trunk_connection_pop_cancellation(trunk_request_t **treq_out, trunk_connection_t *tconn)
{
	if (unlikely(tconn->pub.state == TRUNK_CONN_HALTED)) return -1;

	if (!fr_cond_assert_msg(IN_REQUEST_CANCEL_MUX(tconn->pub.trunk),
				"%s can only be called from within request_cancel_mux handler",
				__FUNCTION__)) return -2;

	*treq_out = tconn->cancel_partial ? tconn->cancel_partial : fr_dlist_head(&tconn->cancel);
	if (!*treq_out) return 1;

	return 0;
}

/** Pop a request off a connection's pending queue
 *
 * The request we return is advanced by the request moving out of the partial or
 * pending states, when the mux function signals us.
 *
 * If the same request is returned again and again, it means the muxer isn't actually
 * doing anything with the request we returned, and it's and error in the muxer code.
 *
 * One of these signalling functions must be used after the request has been popped:
 *
 * - #trunk_request_signal_complete
 *   The request was completed. Either we got a synchronous response, or we knew the
 *   response without contacting an external server (cache).
 *
 * - #trunk_request_signal_fail
 *   Failed muxing the request due to a permanent issue, i.e. an invalid request.
 *
 * - #trunk_request_signal_partial
 *   Wrote part of a request.  This request will be returned on the next call to this
 *   function so that the request_mux function can finish writing it. Only useful
 *   for stream type connections.  Datagram type connections cannot have partial
 *   writes.
 *
 * - #trunk_request_signal_sent Successfully sent a request.
 *
 * @param[out] treq_out	to process
 * @param[in] tconn	to pop a request from.
 * @return
 *	- 1 if no more requests.
 *	- 0 if a new request was written to treq_out.
 *	- -1 if the connection was previously freed.  Caller *MUST NOT* touch any
 *	  memory or requests associated with the connection.
 *	- -2 if called outside of the muxer.
 */
int trunk_connection_pop_request(trunk_request_t **treq_out, trunk_connection_t *tconn)
{
	if (unlikely(tconn->pub.state == TRUNK_CONN_HALTED)) return -1;

	if (!fr_cond_assert_msg(IN_REQUEST_MUX(tconn->pub.trunk),
				"%s can only be called from within request_mux handler",
				__FUNCTION__)) return -2;

	*treq_out = tconn->partial ? tconn->partial : fr_heap_peek(tconn->pending);
	if (!*treq_out) return 1;

	return 0;
}

/** Signal that a trunk connection is writable
 *
 * Should be called from the 'write' I/O handler to signal that requests can be enqueued.
 *
 * @param[in] tconn to signal.
 */
void trunk_connection_signal_writable(trunk_connection_t *tconn)
{
	trunk_t *trunk = tconn->pub.trunk;

	if (!fr_cond_assert_msg(!IN_HANDLER(tconn->pub.trunk),
				"%s cannot be called within a handler", __FUNCTION__)) return;

	DEBUG3("[%" PRIu64 "] Signalled writable", tconn->pub.conn->id);

	trunk_connection_writable(tconn);
}

/** Signal that a trunk connection is readable
 *
 * Should be called from the 'read' I/O handler to signal that requests should be dequeued.
 *
 * @param[in] tconn to signal.
 */
void trunk_connection_signal_readable(trunk_connection_t *tconn)
{
	trunk_t *trunk = tconn->pub.trunk;

	if (!fr_cond_assert_msg(!IN_HANDLER(tconn->pub.trunk),
				"%s cannot be called within a handler", __FUNCTION__)) return;

	DEBUG3("[%" PRIu64 "] Signalled readable", tconn->pub.conn->id);

	trunk_connection_readable(tconn);
}

/** Signal a trunk connection cannot accept more requests
 *
 * @param[in] tconn to signal.
 */
void trunk_connection_signal_inactive(trunk_connection_t *tconn)
{
	/* Can be called anywhere */

	switch (tconn->pub.state) {
	case TRUNK_CONN_ACTIVE:
	case TRUNK_CONN_FULL:
		trunk_connection_enter_inactive(tconn);
		break;

	case TRUNK_CONN_DRAINING:
		trunk_connection_enter_inactive_draining(tconn);
		break;

	default:
		return;
	}
}

/** Signal a trunk connection is no longer full
 *
 * @param[in] tconn to signal.
 */
void trunk_connection_signal_active(trunk_connection_t *tconn)
{
	switch (tconn->pub.state) {
	case TRUNK_CONN_FULL:
		trunk_connection_auto_unfull(tconn);	/* Mark as active if it should be active */
		break;

	case TRUNK_CONN_INACTIVE:
		/*
		 *	Do the appropriate state transition based on
		 *	how many requests the trunk connection is
		 *	currently servicing.
		 */
		if (trunk_connection_is_full(tconn)) {
			trunk_connection_enter_full(tconn);
			break;
		}
		trunk_connection_enter_active(tconn);
		break;

	/*
	 *	Unsetting the active flag just moves
	 *	the connection back to the normal
	 *	draining state.
	 */
	case TRUNK_CONN_INACTIVE_DRAINING:		/* Only an external signal can trigger this transition */
		trunk_connection_enter_draining(tconn);
		break;

	default:
		return;
	}
}

/** Signal a trunk connection is no longer viable
 *
 * @param[in] tconn	to signal.
 * @param[in] reason	the connection is being reconnected.
 */
void trunk_connection_signal_reconnect(trunk_connection_t *tconn, connection_reason_t reason)
{
	connection_signal_reconnect(tconn->pub.conn, reason);
}

/** Standard I/O read function
 *
 * Underlying FD in now readable, so call the trunk to read any pending requests
 * from this connection.
 *
 * @param[in] el	The event list signalling.
 * @param[in] fd	that's now readable.
 * @param[in] flags	describing the read event.
 * @param[in] uctx	The trunk connection handle (tconn).
 */
void trunk_connection_callback_readable(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);

	trunk_connection_signal_readable(tconn);
}

/** Standard I/O write function
 *
 * Underlying FD is now writable, so call the trunk to write any pending requests
 * to this connection.
 *
 * @param[in] el	The event list signalling.
 * @param[in] fd	that's now writable.
 * @param[in] flags	describing the write event.
 * @param[in] uctx	The trunk connection handle (tcon).
 */
void trunk_connection_callback_writable(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
{
	trunk_connection_t	*tconn = talloc_get_type_abort(uctx, trunk_connection_t);

	trunk_connection_signal_writable(tconn);
}


/** Returns true if the trunk connection is in one of the specified states
 *
 * @param[in] tconn	To check state for.
 * @param[in] state	to check
 * @return
 *	- True if trunk connection is in a particular state.
 *	- False if trunk connection is not in a particular state.
 */
bool trunk_connection_in_state(trunk_connection_t *tconn, int state)
{
	return (bool)(tconn->pub.state & state);
}

/** Close connections in a particular connection list if they have no requests associated with them
 *
 * @param[in] trunk	containing connections we want to close.
 * @param[in] head	of list of connections to examine.
 */
static void trunk_connection_close_if_empty(trunk_t *trunk, fr_dlist_head_t *head)
{
	trunk_connection_t *tconn = NULL;

	while ((tconn = fr_dlist_next(head, tconn))) {
		trunk_connection_t *prev;

		if (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) != 0) continue;

		prev = fr_dlist_prev(head, tconn);

		DEBUG3("Closing %s connection with no requests",
		       fr_table_str_by_value(trunk_connection_states, tconn->pub.state, "<INVALID>"));
		/*
		 *	Close the connection as gracefully
		 *	as possible by signalling it should
		 *	shutdown.
		 *
		 *	The connection, should, if serviced
		 *	correctly by the underlying library,
		 *	automatically transition to halted after
		 *	all pending reads/writes are
		 *	complete at which point we'll be informed
		 *	and free our tconn wrapper.
		 */
		connection_signal_shutdown(tconn->pub.conn);
		tconn = prev;
	}
}

/** 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(trunk_t *trunk)
{
	trunk_connection_t	*head;

	head = fr_minmax_heap_min_peek(trunk->active);

	/*
	 *	Only rebalance if the top and bottom of
	 *	the heap are not equal.
	 */
	if (trunk->funcs.connection_prioritise(fr_minmax_heap_max_peek(trunk->active), head) == 0) return;

	DEBUG3("Rebalancing requests");

	/*
	 *	Keep requeuing requests from the connection
	 *	at the bottom of the heap until the
	 *	connection at the top is shifted from that
	 *	position.
	 */
	while ((fr_minmax_heap_min_peek(trunk->active) == head) &&
	       trunk_connection_requests_requeue_priv(fr_minmax_heap_max_peek(trunk->active),
	       					      TRUNK_REQUEST_STATE_PENDING, 1, false));
}

/** Implements the algorithm we use to manage requests per connection levels
 *
 * This is executed periodically using a timer event, and opens/closes
 * connections.
 *
 * The aim is to try and keep the request per connection level in a sweet spot,
 * where there's enough outstanding work for the connection/pipelining to work
 * efficiently, but not so much so that we encounter increased latency.
 *
 * In the request enqueue and dequeue functions we record every time the
 * average number of requests per connection goes above the target count
 * and record every time the average number of requests per connection goes
 * below the target count.
 *
 * This may sound expensive, but in all cases we're just summing counters.
 * CPU time required does not increase with additional requests, only with
 * large numbers of connections.
 *
 * If we do encounter scaling issues, we can always maintain the counters
 * as aggregates as an optimisation later.
 *
 * If when the management function runs, the trunk was above the target
 * most recently, we:
 * - Return if we've been in this state for a shorter period than 'open_delay'.
 * - Return if we're at max.
 * - Return if opening a new connection will take us below the load target.
 * - Return if we last opened a connection within 'open_delay'.
 * - Otherwise we attempt to open a new connection.
 *
 * If the trunk we below the target most recently, we:
 * - Return if we've been in this state for a shorter period than 'close_delay'.
 * - Return if we're at min.
 * - Return if we have no connections.
 * - Close a connection if min is 0, and we have no outstanding
 *   requests.  Then return.
 * - Return if closing a new connection will take us above the load target.
 * - Return if we last closed a connection within 'closed_delay'.
 * - Otherwise we move a connection to draining state.
 */
static void trunk_manage(trunk_t *trunk, fr_time_t now)
{
	trunk_connection_t	*tconn = NULL;
	trunk_request_t	*treq;
	uint32_t		average = 0;
	uint32_t		req_count;
	uint16_t		conn_count;
	trunk_state_t	new_state;

	DEBUG4("Managing trunk");

	/*
	 *	Cleanup requests in our request cache which
	 *	have been reapable for too long.
	 */
	while ((treq = fr_dlist_tail(&trunk->free_requests)) &&
	       fr_time_lteq(fr_time_add(treq->last_freed, trunk->conf.req_cleanup_delay), now)) talloc_free(treq);

	/*
	 *	Free any connections which have drained
	 *	and we didn't reactivate during the last
	 *	round of management.
	 */
	trunk_connection_close_if_empty(trunk, &trunk->inactive_draining);
	trunk_connection_close_if_empty(trunk, &trunk->draining);
	trunk_connection_close_if_empty(trunk, &trunk->draining_to_free);

	/*
	 *	Process deferred connection freeing
	 */
	if (!trunk->in_handler) {
		while ((tconn = fr_dlist_head(&trunk->to_free))) talloc_free(fr_dlist_remove(&trunk->to_free, tconn));
	}

	/*
	 *	Update the state of the trunk
	 */
	if (trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE)) {
		new_state = TRUNK_STATE_ACTIVE;
	} else {
		/*
		 *	INIT / CONNECTING / FULL mean connections will become active
		 *	so the trunk is PENDING
		 */
		new_state = trunk_connection_count_by_state(trunk, TRUNK_CONN_INIT |
							       TRUNK_CONN_CONNECTING |
							       TRUNK_CONN_FULL) ?
			     TRUNK_STATE_PENDING : TRUNK_STATE_IDLE;
	}

	if (new_state != trunk->pub.state) TRUNK_STATE_TRANSITION(new_state);

	/*
	 *	A trunk can be signalled to not proactively
	 *	manage connections if a destination is known
	 *	to be unreachable, and doing so would result
	 *	in spurious connections still being opened.
	 *
	 *	We still run other connection management
	 *	functions and just short circuit the function
	 *	here.
	 */
	if (!trunk->managing_connections) return;

	/*
	 *	We're above the target requests per connection
	 *	spawn more connections!
	 */
	if (fr_time_gteq(trunk->pub.last_above_target, trunk->pub.last_below_target)) {
		/*
		 *	If connecting is provided, check we
		 *	wouldn't have too many connections in
		 *	the connecting state.
		 *
		 *	This is a throttle in the case of transitory
		 *	load spikes, or a backend becoming
		 *	unavailable.
		 */
		if ((trunk->conf.connecting > 0) &&
		    (trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) >=
		     trunk->conf.connecting)) {
			DEBUG4("Not opening connection - Too many (%u) connections in the connecting state",
			       trunk->conf.connecting);
			return;
		}

		trunk_requests_per_connection(&conn_count, &req_count, trunk, now, true);

		/*
		 *	Only apply hysteresis if we have at least
		 *	one available connection.
		 */
		if (conn_count && fr_time_gt(fr_time_add(trunk->pub.last_above_target, trunk->conf.open_delay), now)) {
			DEBUG4("Not opening connection - Need to be above target for %pVs.  It's been %pVs",
			       fr_box_time_delta(trunk->conf.open_delay),
			       fr_box_time_delta(fr_time_sub(now, trunk->pub.last_above_target)));
			return;	/* too soon */
		}

		/*
		 *	We don't consider 'draining' connections
		 *	in the max calculation, as if we do
		 *	determine that we need to spawn a new
		 *	request, then we'd move all 'draining'
		 *	connections to active before spawning
		 *	any new connections.
		 */
		if ((trunk->conf.max > 0) && (conn_count >= trunk->conf.max)) {
			DEBUG4("Not opening connection - Have %u connections, need %u or below",
			       conn_count, trunk->conf.max);
			return;
		}

		/*
		 *	We consider requests pending on all connections
		 *      and the trunk's backlog as that's the current count
		 *	load.
		 */
		if (!req_count) {
			DEBUG4("Not opening connection - No outstanding requests");
			return;
		}

		/*
		 *	Do the n+1 check, i.e. if we open one connection
		 *	will that take us below our target threshold.
		 */
		if (conn_count > 0) {
			average = ROUND_UP_DIV(req_count, (conn_count + 1));
			if (average < trunk->conf.target_req_per_conn) {
				DEBUG4("Not opening connection - Would leave us below our target requests "
				       "per connection (now %u, after open %u)",
				       ROUND_UP_DIV(req_count, conn_count), average);
				return;
			}
		} else {
			(void)trunk_connection_spawn(trunk, now);
			return;
		}

		/*
		 *	If we've got a connection in the draining list
		 *      move it back into the active list if we've
		 *      been requested to add a connection back in.
		 */
		tconn = fr_dlist_head(&trunk->draining);
		if (tconn) {
			if (trunk_connection_is_full(tconn)) {
				trunk_connection_enter_full(tconn);
			} else {
				trunk_connection_enter_active(tconn);
			}
			return;
		}

		/*
		 *	Implement delay if there's no connections that
		 *	could be immediately re-activated.
		 */
		if (fr_time_gt(fr_time_add(trunk->pub.last_open, trunk->conf.open_delay), now)) {
			DEBUG4("Not opening connection - Need to wait %pVs before opening another connection.  "
			       "It's been %pVs",
			       fr_box_time_delta(trunk->conf.open_delay),
			       fr_box_time_delta(fr_time_sub(now, trunk->pub.last_open)));
			return;
		}

		DEBUG4("Opening connection - Above target requests per connection (now %u, target %u)",
		       ROUND_UP_DIV(req_count, conn_count), trunk->conf.target_req_per_conn);
		/* last_open set by trunk_connection_spawn */
		(void)trunk_connection_spawn(trunk, now);
	}

	/*
	 *	We're below the target requests per connection.
	 *	Free some connections...
	 */
	else if (fr_time_gt(trunk->pub.last_below_target, trunk->pub.last_above_target)) {
		if (fr_time_gt(fr_time_add(trunk->pub.last_below_target, trunk->conf.close_delay), now)) {
			DEBUG4("Not closing connection - Need to be below target for %pVs. It's been %pVs",
			       fr_box_time_delta(trunk->conf.close_delay),
			       fr_box_time_delta(fr_time_sub(now, trunk->pub.last_below_target)));
			return;	/* too soon */
		}

		trunk_requests_per_connection(&conn_count, &req_count, trunk, now, true);

		if (!conn_count) {
			DEBUG4("Not closing connection - No connections to close!");
			return;
		}

		if ((trunk->conf.min > 0) && ((conn_count - 1) < trunk->conf.min)) {
			DEBUG4("Not closing connection - Have %u connections, need %u or above",
			       conn_count, trunk->conf.min);
			return;
		}

		if (!req_count) {
			DEBUG4("Closing connection - No outstanding requests");
			goto close;
		}

		/*
		 *	The minimum number of connections must be set
		 *	to zero for this to work.
		 *	min == 0, no requests, close all the connections.
		 *      This is useful for backup databases, when
		 *	maintaining the connection would lead to lots of
		 *	log file churn.
		 */
		if (conn_count == 1) {
			DEBUG4("Not closing connection - Would leave connections "
			       "and there are still %u outstanding requests", req_count);
			return;
		}

		/*
		 *	Do the n-1 check, i.e. if we close one connection
		 *	will that take us above our target threshold.
		 */
		average = ROUND_UP_DIV(req_count, (conn_count - 1));
		if (average > trunk->conf.target_req_per_conn) {
			DEBUG4("Not closing connection - Would leave us above our target requests per connection "
			       "(now %u, after close %u)", ROUND_UP_DIV(req_count, conn_count), average);
			return;
		}

		DEBUG4("Closing connection - Below target requests per connection (now %u, target %u)",
		       ROUND_UP_DIV(req_count, conn_count), trunk->conf.target_req_per_conn);

	close:
		if (fr_time_gt(fr_time_add(trunk->pub.last_closed, trunk->conf.close_delay), now)) {
			DEBUG4("Not closing connection - Need to wait %pVs before closing another connection.  "
			       "It's been %pVs",
			       fr_box_time_delta(trunk->conf.close_delay),
			       fr_box_time_delta(fr_time_sub(now, trunk->pub.last_closed)));
			return;
		}

		/*
		 *	If the last event on the trunk was a connection failure and
		 *	there is only one connection, this may well be a reconnect
		 *	attempt after a failure - and needs to persist otherwise
		 *	the last event will be a failure and no new connection will
		 *	be made, leading to no new requests being enqueued.
		 */
		if (fr_time_gt(trunk->pub.last_failed, fr_time_wrap(0)) &&
		    fr_time_lt(trunk->pub.last_connected, trunk->pub.last_failed) && (conn_count == 1)) {
			DEBUG4("Not closing remaining connection - last event was a failure");
			return;
		}

		/*
		 *	Inactive connections get counted in the
		 *	set of viable connections, but are likely
		 *	to be congested or dead, so we drain
		 *	(and possibly eventually free) those first.
		 */
		if ((tconn = fr_dlist_tail(&trunk->inactive))) {
			/*
			 *	If the connection has no requests associated
			 *	with it then immediately free.
			 */
			if (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0) {
				connection_signal_halt(tconn->pub.conn);	/* Also frees the tconn */
			} else {
				trunk_connection_enter_inactive_draining(tconn);
			}
		/*
		 *	It is possible to have too may connecting
		 *	connections when the connections are
		 *	taking a while to open and the number
		 *	of requests decreases.
		 */
		} else if ((tconn = fr_dlist_tail(&trunk->connecting))) {
			connection_signal_halt(tconn->pub.conn);	/* Also frees the tconn */

		/*
		 *	Finally if there are no "connecting"
		 *	connections to close, and no "inactive"
		 *	connections, start draining "active"
		 *	connections.
		 */
		} else if ((tconn = fr_minmax_heap_max_peek(trunk->active))) {
			/*
			 *	If the connection has no requests associated
			 *	with it then immediately free.
			 */
			if (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0) {
				connection_signal_halt(tconn->pub.conn);	/* Also frees the tconn */
			} else {
				trunk_connection_enter_draining(tconn);
			}
		}

		trunk->pub.last_closed = now;


		return;
	}
}

/** Event to periodically call the connection management function
 *
 * @param[in] el	this event belongs to.
 * @param[in] now	current time.
 * @param[in] uctx	The trunk.
 */
static void _trunk_timer(fr_event_list_t *el, fr_time_t now, void *uctx)
{
	trunk_t *trunk = talloc_get_type_abort(uctx, trunk_t);

	trunk_manage(trunk, now);

	if (fr_time_delta_ispos(trunk->conf.manage_interval)) {
		if (fr_event_timer_in(trunk, el, &trunk->manage_ev, trunk->conf.manage_interval,
				      _trunk_timer, trunk) < 0) {
			PERROR("Failed inserting trunk management event");
			/* Not much we can do, hopefully the trunk will be freed soon */
		}
	}
}

/** Return a count of requests on a connection in a specific state
 *
 * @param[in] trunk		to retrieve counts for.
 * @param[in] conn_state	One or more connection states or'd together.
 * @param[in] req_state		One or more request states or'd together.
 * @return The number of requests in a particular state, on connection in a particular state.
 */
uint64_t trunk_request_count_by_state(trunk_t *trunk, int conn_state, int req_state)
{
	uint64_t		count = 0;
	trunk_connection_t	*tconn = NULL;
	fr_minmax_heap_iter_t	iter;

#define COUNT_BY_STATE(_state, _list) \
do { \
	if (conn_state & (_state)) { \
		tconn = NULL; \
		while ((tconn = fr_dlist_next(&trunk->_list, tconn))) { \
			count += trunk_request_count_by_connection(tconn, req_state); \
		} \
	} \
} while (0)

	if (conn_state & TRUNK_CONN_ACTIVE) {
		for (tconn = fr_minmax_heap_iter_init(trunk->active, &iter);
		     tconn;
		     tconn = fr_minmax_heap_iter_next(trunk->active, &iter)) {
			count += trunk_request_count_by_connection(tconn, req_state);
		}
	}

	COUNT_BY_STATE(TRUNK_CONN_FULL, full);
	COUNT_BY_STATE(TRUNK_CONN_INACTIVE, inactive);
	COUNT_BY_STATE(TRUNK_CONN_INACTIVE_DRAINING, inactive_draining);
	COUNT_BY_STATE(TRUNK_CONN_DRAINING, draining);
	COUNT_BY_STATE(TRUNK_CONN_DRAINING_TO_FREE, draining_to_free);

	if (req_state & TRUNK_REQUEST_STATE_BACKLOG) count += fr_heap_num_elements(trunk->backlog);

	return count;
}

/** Update timestamps for when we last had a transition from above target to below target or vice versa
 *
 * Should be called on every time a connection or request is allocated or freed.
 *
 * @param[out] conn_count_out	How many connections we considered.
 * @param[out] req_count_out	How many requests we considered.
 * @param[in] trunk		to operate on.
 * @param[in] now		The current time.
 * @param[in] verify		if true (and this is a debug build), then assert if req_per_conn
 *				has changed.
 * @return
 *	- 0 if the average couldn't be calculated (no requests or no connections).
 *	- The average number of requests per connection.
 */
static uint64_t trunk_requests_per_connection(uint16_t *conn_count_out, uint32_t *req_count_out,
					      trunk_t *trunk, fr_time_t now,
					      NDEBUG_UNUSED bool verify)
{
	uint32_t req_count = 0;
	uint16_t conn_count = 0;
	uint64_t req_per_conn = 0;

	fr_assert(fr_time_gt(now, fr_time_wrap(0)));

	/*
	 *	No need to update these as the trunk is being freed
	 */
	if (trunk->freeing) goto done;

	/*
	 *	Count all connections except draining and draining to free.
	 *
	 *	Omitting these connection states artificially raises the
	 *	request to connection ratio, so that we can preemptively spawn
	 *	new connections.
	 *
	 *	In the case of TRUNK_CONN_DRAINING | TRUNK_CONN_INACTIVE_DRAINING
	 *	the trunk management code has enough hysteresis to not
	 *	immediately reactivate the connection.
	 *
	 *	In the case of TRUNK_CONN_DRAINING_TO_FREE the trunk
	 *	management code should spawn a new connection to takes its place.
	 *
	 *	Connections placed in the DRAINING_TO_FREE state are being
	 *	closed preemptively to deal with bugs on the server we're
	 *	talking to, or misconfigured firewalls which are trashing
	 *	TCP/UDP connection states.
	 */
	conn_count = trunk_connection_count_by_state(trunk, TRUNK_CONN_ALL ^
							(TRUNK_CONN_DRAINING |
							 TRUNK_CONN_INACTIVE_DRAINING |
							 TRUNK_CONN_DRAINING_TO_FREE));

	/*
	 *	Requests on all connections
	 */
	req_count = trunk_request_count_by_state(trunk,
						    TRUNK_CONN_ALL ^
						    TRUNK_CONN_DRAINING_TO_FREE, TRUNK_REQUEST_STATE_ALL);

	/*
	 *	No connections, but we do have requests
	 */
	if (conn_count == 0) {
		if ((req_count > 0) && (trunk->conf.target_req_per_conn > 0)) goto above_target;
		goto done;
	}

	if (req_count == 0) {
		if (trunk->conf.target_req_per_conn > 0) goto below_target;
		goto done;
	}

	/*
	 *	Calculate the req_per_conn
	 */
	req_per_conn = ROUND_UP_DIV(req_count, conn_count);
	if (req_per_conn > trunk->conf.target_req_per_conn) {
	above_target:
		/*
		 *	Edge - Below target to above target (too many requests per conn - spawn more)
		 *
		 *	The equality check is correct here as both values start at 0.
		 */
		if (fr_time_lteq(trunk->pub.last_above_target, trunk->pub.last_below_target)) trunk->pub.last_above_target = now;
	} else if (req_per_conn < trunk->conf.target_req_per_conn) {
	below_target:
		/*
		 *	Edge - Above target to below target (too few requests per conn - close some)
		 *
		 *	The equality check is correct here as both values start at 0.
		 */
		if (fr_time_lteq(trunk->pub.last_below_target, trunk->pub.last_above_target)) trunk->pub.last_below_target = now;
	}

done:
	if (conn_count_out) *conn_count_out = conn_count;
	if (req_count_out) *req_count_out = req_count;

	/*
	 *	Check we haven't missed a call to trunk_requests_per_connection
	 */
	fr_assert(!verify || (trunk->last_req_per_conn == 0) || (req_per_conn == trunk->last_req_per_conn));

	trunk->last_req_per_conn = req_per_conn;

	return req_per_conn;
}

/** Drain the backlog of as many requests as possible
 *
 * @param[in] trunk	To drain backlog requests for.
 */
static void trunk_backlog_drain(trunk_t *trunk)
{
	trunk_request_t *treq;

	if (fr_heap_num_elements(trunk->backlog) == 0) return;

	/*
	 *	If it's always writable, this isn't
	 *	really a noteworthy event.
	 */
	if (!trunk->conf.always_writable) DEBUG3("Draining backlog of requests");

	/*
	 *	Do *NOT* add an artificial limit
	 *	here.  We rely on all available
	 *	connections entering the full
	 *	state and transitioning back to
	 *	active in order to drain the
	 *	backlog.
	 */
	while ((treq = fr_heap_peek(trunk->backlog))) {
		switch (trunk_request_enqueue_existing(treq)) {
		case TRUNK_ENQUEUE_OK:
			continue;

		/*
		 *	Signal to stop
		 */
		case TRUNK_ENQUEUE_IN_BACKLOG:
			break;

		/*
		 *	Failed enqueueing the request,
		 *	have it enter the failed state
		 *	which will free it and
		 *	re-enliven the yielded request.
		 */
		case TRUNK_ENQUEUE_DST_UNAVAILABLE:
		case TRUNK_ENQUEUE_FAIL:
			trunk_request_enter_failed(treq);
			continue;

		case TRUNK_ENQUEUE_NO_CAPACITY:
			fr_assert(fr_minmax_heap_num_elements(trunk->active) == 0);
			return;
		}
	}
}

/** Force the trunk to re-establish its connections
 *
 * @param[in] trunk		to signal.
 * @param[in] states		One or more states or'd together.
 * @param[in] reason		Why the connections are being signalled to reconnect.
 */
void trunk_reconnect(trunk_t *trunk, int states, connection_reason_t reason)
{

#define RECONNECT_BY_STATE(_state, _list) \
do { \
	if (states & (_state)) { \
		size_t i; \
		for (i = fr_dlist_num_elements(&trunk->_list); i > 0; i--) { \
			connection_signal_reconnect(((trunk_connection_t *)fr_dlist_tail(&trunk->_list))->pub.conn, reason); \
		} \
	} \
} while (0)

	/*
	 *	Connections in the 'connecting' state
	 *	may re-enter that state, so we need to
	 *	be careful not to enter an infinite
	 *	loop, as we iterate over the list
	 *	again and again.
	 */
	RECONNECT_BY_STATE(TRUNK_CONN_CONNECTING, connecting);

	if (states & TRUNK_CONN_ACTIVE) {
		trunk_connection_t *tconn;
		while ((tconn = fr_minmax_heap_min_peek(trunk->active))) connection_signal_reconnect(tconn->pub.conn, reason);
	}

	RECONNECT_BY_STATE(TRUNK_CONN_INIT, init);
	RECONNECT_BY_STATE(TRUNK_CONN_FULL, full);
	RECONNECT_BY_STATE(TRUNK_CONN_INACTIVE, inactive);
	RECONNECT_BY_STATE(TRUNK_CONN_INACTIVE_DRAINING, inactive_draining);
	RECONNECT_BY_STATE(TRUNK_CONN_CLOSED, closed);
	RECONNECT_BY_STATE(TRUNK_CONN_DRAINING, draining);
	RECONNECT_BY_STATE(TRUNK_CONN_DRAINING_TO_FREE, draining_to_free);
}

/** Start the trunk running
 *
 */
int trunk_start(trunk_t *trunk)
{
	uint16_t i;

	if (unlikely(trunk->started)) return 0;

	/*
	 *	Spawn the initial set of connections
	 */
	for (i = 0; i < trunk->conf.start; i++) {
		DEBUG("[%i] Starting initial connection", i);
		if (trunk_connection_spawn(trunk, fr_time()) != 0) return -1;
	}

	if (fr_time_delta_ispos(trunk->conf.manage_interval)) {
		/*
		 *	Insert the event timer to manage
		 *	the interval between managing connections.
		 */
		if (fr_event_timer_in(trunk, trunk->el, &trunk->manage_ev, trunk->conf.manage_interval,
				      _trunk_timer, trunk) < 0) {
			PERROR("Failed inserting trunk management event");
			return -1;
		}
	}
	trunk->started = true;
	trunk->managing_connections = true;

	return 0;
}

/** Allow the trunk to open and close connections in response to load
 *
 */
void trunk_connection_manage_start(trunk_t *trunk)
{
	if (!trunk->started || trunk->managing_connections) return;

	DEBUG3("Connection management enabled");
	trunk->managing_connections = true;
}

/** Stop the trunk from opening and closing connections in response to load
 *
 */
void trunk_connection_manage_stop(trunk_t *trunk)
{
	if (!trunk->started || !trunk->managing_connections) return;

	DEBUG3("Connection management disabled");
	trunk->managing_connections = false;
}

/** Schedule a trunk management event for the next time the event loop is executed
 */
int trunk_connection_manage_schedule(trunk_t *trunk)
{
	if (!trunk->started || !trunk->managing_connections) return 0;

	if (fr_event_timer_in(trunk, trunk->el, &trunk->manage_ev, fr_time_delta_wrap(0), _trunk_timer, trunk) < 0) {
		PERROR("Failed inserting trunk management event");
		return -1;
	}

	return 0;
}

/** Order connections by queue depth
 *
 */
static int8_t _trunk_connection_order_by_shortest_queue(void const *one, void const *two)
{
	trunk_connection_t const	*a = talloc_get_type_abort_const(one, trunk_connection_t);
	trunk_connection_t const	*b = talloc_get_type_abort_const(two, trunk_connection_t);

	uint32_t			a_count = trunk_request_count_by_connection(a, TRUNK_REQUEST_STATE_ALL);
	uint32_t			b_count = trunk_request_count_by_connection(b, TRUNK_REQUEST_STATE_ALL);

	/*
	 *	Add a fudge factor of 1 to reduce spurious rebalancing
	 */
	return ((a_count > b_count) && ((a_count - b_count) > 1)) - ((b_count > a_count) && ((b_count - a_count) > 1));
}

/** Free a trunk, gracefully closing all connections.
 *
 */
static int _trunk_free(trunk_t *trunk)
{
	trunk_connection_t	*tconn;
	trunk_request_t	*treq;
	trunk_watch_entry_t	*watch;
	size_t			i;

	DEBUG4("Trunk free %p", trunk);

	trunk->freeing = true;	/* Prevent re-enqueuing */

	/*
	 *	We really don't want this firing after
	 *	we've freed everything.
	 */
	fr_event_timer_delete(&trunk->manage_ev);

	/*
	 *	Now free the connections in each of the lists.
	 *
	 *	Each time a connection is freed it removes itself from the list
	 *	its in, which means the head should keep advancing automatically.
	 */
	while ((tconn = fr_minmax_heap_min_peek(trunk->active))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->init))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->connecting))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->full))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->inactive))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->inactive_draining))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->closed))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->draining))) connection_signal_halt(tconn->pub.conn);
	while ((tconn = fr_dlist_head(&trunk->draining_to_free))) connection_signal_halt(tconn->pub.conn);

	/*
	 *	Process any deferred connection frees
	 */
	while ((tconn = fr_dlist_head(&trunk->to_free))) talloc_free(fr_dlist_remove(&trunk->to_free, tconn));

	/*
	 *	Free any requests left in the backlog
	 */
	while ((treq = fr_heap_peek(trunk->backlog))) trunk_request_enter_failed(treq);

	/*
	 *	Free any requests in our request cache
	 */
	while ((treq = fr_dlist_head(&trunk->free_requests))) talloc_free(treq);

	/*
	 *	Free any entries in the watch lists
	 */
	for (i = 0; i < NUM_ELEMENTS(trunk->watch); i++) {
		while ((watch = fr_dlist_pop_head(&trunk->watch[i]))) talloc_free(watch);
	}

	return 0;
}

/** Allocate a new collection of connections
 *
 * This function should be called first to allocate a new trunk connection.
 *
 * After the trunk has been allocated, #trunk_request_alloc and
 * #trunk_request_enqueue should be used to allocate memory for trunk
 * requests, and pass a preq (protocol request) to the trunk for
 * processing.
 *
 * The trunk will then asynchronously process the request, writing the result
 * to a specified rctx.  See #trunk_request_enqueue for more details.
 *
 * @note Trunks may not be shared between multiple threads under any circumstances.
 *
 * @param[in] ctx		To use for any memory allocations.  Must be thread local.
 * @param[in] el		to use for I/O and timer events.
 * @param[in] funcs		Callback functions.
 * @param[in] conf		Common user configurable parameters.
 * @param[in] log_prefix	To prepend to global messages.
 * @param[in] uctx		User data to pass to the alloc function.
 * @param[in] delay_start	If true, then we will not spawn any connections
 *				until the first request is enqueued.
 * @return
 *	- New trunk handle on success.
 *	- NULL on error.
 */
trunk_t *trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el,
			   trunk_io_funcs_t const *funcs, trunk_conf_t const *conf,
			   char const *log_prefix, void const *uctx, bool delay_start)
{
	trunk_t	*trunk;
	size_t		i;

	/*
	 *	Check we have the functions we need
	 */
	if (!fr_cond_assert(funcs->connection_alloc)) return NULL;

	MEM(trunk = talloc_zero(ctx, trunk_t));
	trunk->el = el;
	trunk->log_prefix = talloc_strdup(trunk, log_prefix);

	memcpy(&trunk->funcs, funcs, sizeof(trunk->funcs));
	if (!trunk->funcs.connection_prioritise) {
		trunk->funcs.connection_prioritise = _trunk_connection_order_by_shortest_queue;
	}
	if (!trunk->funcs.request_prioritise) trunk->funcs.request_prioritise = fr_pointer_cmp;

	memcpy(&trunk->conf, conf, sizeof(trunk->conf));

	memcpy(&trunk->uctx, &uctx, sizeof(trunk->uctx));
	talloc_set_destructor(trunk, _trunk_free);

	/*
	 *	Unused request list...
	 */
	fr_dlist_talloc_init(&trunk->free_requests, trunk_request_t, entry);

	/*
	 *	Request backlog queue
	 */
	MEM(trunk->backlog = fr_heap_talloc_alloc(trunk, _trunk_request_prioritise,
						   trunk_request_t, heap_id, 0));

	/*
	 *	Connection queues and trees
	 */
	MEM(trunk->active = fr_minmax_heap_talloc_alloc(trunk, trunk->funcs.connection_prioritise,
						  trunk_connection_t, heap_id, 0));
	fr_dlist_talloc_init(&trunk->init, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->connecting, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->full, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->inactive, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->inactive_draining, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->closed, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->draining, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->draining_to_free, trunk_connection_t, entry);
	fr_dlist_talloc_init(&trunk->to_free, trunk_connection_t, entry);

	/*
	 *	Watch lists
	 */
	for (i = 0; i < NUM_ELEMENTS(trunk->watch); i++) {
		fr_dlist_talloc_init(&trunk->watch[i], trunk_watch_entry_t, entry);
	}

	DEBUG4("Trunk allocated %p", trunk);

	if (!delay_start) {
		if (trunk_start(trunk) < 0) {
			talloc_free(trunk);
			return NULL;
		}
	}

	return trunk;
}

#ifndef TALLOC_GET_TYPE_ABORT_NOOP
/** Verify a trunk
 *
 * A trunk has some number of connections, which each have some number of requests. The connections and
 * requests are in differing kinds of containers depending on their state and how they are used, and may
 * have fields that can only be validated by comparison with a parent.  We had planned on passing a "context"
 * down with the ancestral values, but that breaks the foo_verify() API. Each foo_verify() will only verify the
 * foo's children.
 */
void trunk_verify(char const *file, int line, trunk_t *trunk)
{
	fr_fatal_assert_msg(trunk, "CONSISTENCY CHECK FAILED %s[%i]: trunk_t pointer was NULL", file, line);
	(void) talloc_get_type_abort(trunk, trunk_t);

	for (size_t i = 0; i < NUM_ELEMENTS(trunk->watch); i++) {
		_fr_dlist_verify(file, line, &trunk->watch[i]);
	}

#define IO_FUNC_VERIFY(_func) \
	fr_fatal_assert_msg(trunk->funcs._func, "CONSISTENCY_CHECK_FAILED %s[%i}: " #_func " was NULL", file, line)

	/*
	 *	Only a few of the function pointers *must* be non-NULL..
	 */
	IO_FUNC_VERIFY(connection_alloc);
	IO_FUNC_VERIFY(connection_prioritise);
	IO_FUNC_VERIFY(request_prioritise);

#define TRUNK_TCONN_CHECKS(_tconn, _state) \
do { \
	fr_fatal_assert_msg(trunk == _tconn->pub.trunk, \
			    "CONSISTENCY_CHECK_FAILED %s[%i}: connection-trunk mismatch", file, line); \
	fr_fatal_assert_msg(_state == _tconn->pub.state, \
			    "CONSISTENCY_CHECK_FAILED %s[%i}: connection-state mismatch", file, line); \
} while (0)

#define TCONN_DLIST_VERIFY(_dlist, _state) \
do { \
	_fr_dlist_verify(file, line, &(trunk->_dlist)); \
	fr_dlist_foreach(&(trunk->_dlist), trunk_connection_t, tconn)  { \
		trunk_connection_verify(file, line, tconn); \
		TRUNK_TCONN_CHECKS(tconn, _state); \
	} \
} while (0)

#define TCONN_MINMAX_HEAP_VERIFY(_heap, _state) \
do {\
	fr_minmax_heap_verify(file, line, trunk->_heap); \
	fr_minmax_heap_foreach(trunk->_heap, trunk_connection_t, tconn) { \
		trunk_connection_verify(file, line, tconn); \
		TRUNK_TCONN_CHECKS(tconn, _state); \
	}} \
} while (0)

	fr_dlist_verify(&(trunk->free_requests));
	FR_HEAP_VERIFY(trunk->backlog);

	TCONN_DLIST_VERIFY(init, TRUNK_CONN_INIT);
	TCONN_DLIST_VERIFY(connecting, TRUNK_CONN_CONNECTING);
	TCONN_MINMAX_HEAP_VERIFY(active, TRUNK_CONN_ACTIVE);
	TCONN_DLIST_VERIFY(full, TRUNK_CONN_FULL);
	TCONN_DLIST_VERIFY(inactive, TRUNK_CONN_INACTIVE);
	TCONN_DLIST_VERIFY(inactive_draining, TRUNK_CONN_INACTIVE_DRAINING);
	/* TCONN_DLIST_VERIFY(failed, ???); */
	TCONN_DLIST_VERIFY(closed, TRUNK_CONN_CLOSED);
	TCONN_DLIST_VERIFY(draining, TRUNK_CONN_DRAINING);
	TCONN_DLIST_VERIFY(draining_to_free, TRUNK_CONN_DRAINING_TO_FREE);
	TCONN_DLIST_VERIFY(to_free, TRUNK_CONN_HALTED);
}

void trunk_connection_verify(char const *file, int line, trunk_connection_t *tconn)
{
	fr_fatal_assert_msg(tconn, "CONSISTENCY CHECK FAILED %s[%i]: trunk_connection_t pointer was NULL", file, line);
	(void) talloc_get_type_abort(tconn, trunk_connection_t);

	(void) talloc_get_type_abort(tconn->pub.trunk, trunk_t);

	/*
	 *	shouldn't be both in heap and on list--but it doesn't look like moves
	 *	to active heap wipe the dlist pointers.
	 */

#define TCONN_TREQ_CHECKS(_treq, _state) \
do { \
	fr_fatal_assert_msg(tconn == _treq->pub.tconn, \
			    "CONSISTENCY_CHECK_FAILED %s[%i}: trunk request-tconn mismatch", file, line); \
	fr_fatal_assert_msg(tconn->pub.trunk == _treq->pub.trunk, \
			    "CONSISTENCY_CHECK_FAILED %s[%i}: trunk request-trunk mismatch", file, line); \
	fr_fatal_assert_msg(_state == _treq->pub.state, \
			    "CONSISTENCY_CHECK_FAILED %s[%i}: trunk request-state mismatch", file, line); \
} while (0)

#define TREQ_DLIST_VERIFY(_dlist, _state) \
do { \
	_fr_dlist_verify(file, line, &(tconn->_dlist)); \
	fr_dlist_foreach(&(tconn->_dlist), trunk_request_t, treq)  { \
		trunk_request_verify(file, line, treq); \
		TCONN_TREQ_CHECKS(treq, _state); \
	} \
} while (0)

#define TREQ_HEAP_VERIFY(_heap, _state) \
do { \
	fr_heap_iter_t _iter; \
	fr_heap_verify(file, line, tconn->_heap); \
	for (trunk_request_t *treq = fr_heap_iter_init(tconn->_heap, &_iter); \
	     treq; \
	     treq = fr_heap_iter_next(tconn->_heap, &_iter)) { \
		trunk_request_verify(file, line, treq); \
		TCONN_TREQ_CHECKS(treq, _state); \
	} \
} while (0)

#define TREQ_OPTION_VERIFY(_option, _state) \
do { \
	if (tconn->_option) { \
		trunk_request_verify(file, line, tconn->_option); \
		TCONN_TREQ_CHECKS(tconn->_option, _state); \
	} \
} while (0)

	/* verify associated requests */
	TREQ_HEAP_VERIFY(pending, TRUNK_REQUEST_STATE_PENDING);
	TREQ_DLIST_VERIFY(sent, TRUNK_REQUEST_STATE_SENT);
	TREQ_DLIST_VERIFY(cancel, TRUNK_REQUEST_STATE_CANCEL);
	TREQ_DLIST_VERIFY(cancel_sent, TRUNK_REQUEST_STATE_CANCEL_SENT);
	TREQ_OPTION_VERIFY(partial, TRUNK_REQUEST_STATE_PARTIAL);
	TREQ_OPTION_VERIFY(cancel_partial, TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
}

void trunk_request_verify(char const *file, int line, trunk_request_t *treq)
{
	fr_fatal_assert_msg(treq, "CONSISTENCY CHECK FAILED %s[%i]: trunk_request_t pointer was NULL", file, line);
	(void) talloc_get_type_abort(treq, trunk_request_t);

#ifdef WITH_VERIFY_PTR
	if (treq->pub.request) request_verify(file, line, treq->pub.request);
#endif
}


bool trunk_search(trunk_t *trunk, void *ptr)
{
#define TCONN_DLIST_SEARCH(_dlist) \
do { \
	fr_dlist_foreach(&(trunk->_dlist), trunk_connection_t, tconn)  { \
		if (ptr == tconn) { \
			fr_fprintf(stderr, "trunk_search: tconn %p on " #_dlist "\n", ptr); \
			return true; \
		} \
		if (trunk_connection_search(tconn, ptr)) { \
			fr_fprintf(stderr, " in tconn %p on " #_dlist "\n", tconn); \
			return true; \
		} \
	} \
} while (0)

#define TCONN_MINMAX_HEAP_SEARCH(_heap) \
do { \
	fr_minmax_heap_foreach(trunk->_heap, trunk_connection_t, tconn) { \
		if (ptr == tconn) { \
			fr_fprintf(stderr, "trunk_search: tconn %p on " #_heap "\n", ptr); \
			return true; \
		} \
		if (trunk_connection_search(tconn, ptr)) { \
			fr_fprintf(stderr, " on tconn %p on " #_heap "\n", tconn); \
			return true; \
		} \
	}}\
} while (0)

	TCONN_DLIST_SEARCH(init);
	TCONN_DLIST_SEARCH(connecting);
	TCONN_MINMAX_HEAP_SEARCH(active);
	TCONN_DLIST_SEARCH(full);
	TCONN_DLIST_SEARCH(inactive);
	TCONN_DLIST_SEARCH(inactive_draining);
	TCONN_DLIST_SEARCH(failed);
	TCONN_DLIST_SEARCH(closed);
	TCONN_DLIST_SEARCH(draining);
	TCONN_DLIST_SEARCH(draining_to_free);
	TCONN_DLIST_SEARCH(to_free);

	return false;
}

bool trunk_connection_search(trunk_connection_t *tconn, void *ptr)
{
#define TREQ_DLIST_SEARCH(_dlist) \
do { \
	fr_dlist_foreach(&(tconn->_dlist), trunk_request_t, treq)  { \
		if (ptr == treq) { \
			fr_fprintf(stderr, "trunk_search: treq %p on " #_dlist "\n", ptr); \
			return true; \
		} \
		if (trunk_request_search(treq, ptr)) { \
			fr_fprintf(stderr, "trunk_search: preq %p found on " #_dlist, ptr); \
			return true; \
		} \
	} \
} while (0)

#define TREQ_HEAP_SEARCH(_heap) \
do { \
	fr_heap_iter_t _iter; \
	for (trunk_request_t *treq = fr_heap_iter_init(tconn->_heap, &_iter); \
	     treq; \
	     treq = fr_heap_iter_next(tconn->_heap, &_iter)) { \
		if (ptr == treq) { \
			fr_fprintf(stderr, "trunk_search: treq %p in " #_heap "\n", ptr); \
			return true; \
		} \
		if (trunk_request_search(treq, ptr)) { \
			fr_fprintf(stderr, "trunk_search: preq %p found in " #_heap, ptr); \
			return true; \
		} \
	} \
} while (0)

#define TREQ_OPTION_SEARCH(_option) \
do { \
	if (tconn->_option) { \
		if (ptr == tconn->_option) { \
			fr_fprintf(stderr, "trunk_search: treq %p is " #_option "\n", ptr); \
			return true; \
		} \
		if (trunk_request_search(tconn->_option, ptr)) { \
			fr_fprintf(stderr, "trunk_search: preq %p found in " #_option, ptr); \
			return true; \
		} \
	} \
} while (0)

	/* search associated requests */
	TREQ_HEAP_SEARCH(pending);
	TREQ_DLIST_SEARCH(sent);
	TREQ_DLIST_SEARCH(cancel);
	TREQ_DLIST_SEARCH(cancel_sent);
	TREQ_OPTION_SEARCH(partial);
	TREQ_OPTION_SEARCH(cancel_partial);

	return false;
}

bool trunk_request_search(trunk_request_t *treq, void *ptr)
{
	return treq->pub.preq == ptr;
}
#endif
