pool.c

Go to the documentation of this file.

/*
 *   This program 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
 */
 
/**
 * @file pool.c
 * @brief Handle pools of connections (threads, sockets, etc.)
 * @note This API must be used by all modules in the public distribution that
 * maintain pools of connections.
 *
 * @copyright 2012 The FreeRADIUS server project
 * @copyright 2012 Alan DeKok (aland@deployingradius.com)
 */
RCSID("$Id: dc3679204298804cd2fcd222838ddfd44dc6bc20 $")
 
#define LOG_PREFIX pool->log_prefix
 
#include <freeradius-devel/server/main_config.h>
#include <freeradius-devel/server/modpriv.h>
#include <freeradius-devel/server/trigger.h>
 
#include <freeradius-devel/util/debug.h>
 
#include <freeradius-devel/util/misc.h>
 
 
typedef struct fr_pool_connection_s fr_pool_connection_t;
 
static int connection_check(fr_pool_t *pool, request_t *request);
static int max_dflt(CONF_PAIR **out, void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule);
 
/** An individual connection within the connection pool
 *
 * Defines connection counters, timestamps, and holds a pointer to the
 * connection handle itself.
 *
 * @see fr_pool_t
 */
struct fr_pool_connection_s {
        fr_pool_connection_t    *prev;                  //!< Previous connection in list.
        fr_pool_connection_t    *next;                  //!< Next connection in list.
        fr_heap_index_t heap_id;                        //!< For the next connection heap.
 
        fr_time_t       created;                //!< Time connection was created.
        fr_time_t       last_reserved;          //!< Last time the connection was reserved.
 
        fr_time_t       last_released;          //!< Time the connection was released.
 
        uint32_t        num_uses;               //!< Number of times the connection has been reserved.
        uint64_t        number;                 //!< Unique ID assigned when the connection is created,
                                                //!< these will monotonically increase over the
                                                //!< lifetime of the connection pool.
        void            *connection;            //!< Pointer to whatever the module uses for a connection
                                                //!< handle.
        bool            in_use;                 //!< Whether the connection is currently reserved.
 
        bool            needs_reconnecting;     //!< Reconnect this connection before use.
 
#ifdef PTHREAD_DEBUG
        pthread_t       pthread_id;             //!< When 'in_use == true'.
#endif
};
 
/** A connection pool
 *
 * Defines the configuration of the connection pool, all the counters and
 * timestamps related to the connection pool, the mutex that stops multiple
 * threads leaving the pool in an inconsistent state, and the callbacks
 * required to open, close and check the status of connections within the pool.
 *
 * @see connection
 */
struct fr_pool_s {
        int             ref;                    //!< Reference counter to prevent connection
                                                //!< pool being freed multiple times.
        uint32_t        start;                  //!< Number of initial connections.
        uint32_t        min;                    //!< Minimum number of concurrent connections to keep open.
        uint32_t        max;                    //!< Maximum number of concurrent connections to allow.
        uint32_t        max_pending;            //!< Max number of pending connections to allow.
        uint32_t        spare;                  //!< Number of spare connections to try.
        uint64_t        max_uses;               //!< Maximum number of times a connection can be used
                                                //!< before being closed.
        uint32_t        pending_window;         //!< Sliding window of pending connections.
 
        fr_time_delta_t retry_delay;            //!< seconds to delay re-open after a failed open.
        fr_time_delta_t cleanup_interval;       //!< Initial timer for how often we sweep the pool
                                                //!< for free connections. (0 is infinite).
        fr_time_delta_t delay_interval;         //!< When we next do a cleanup.  Initialized to
                                                //!< cleanup_interval, and increase from there based
                                                //!< on the delay.
        fr_time_delta_t lifetime;               //!< How long a connection can be open before being
                                                //!< closed (irrespective of whether it's idle or not).
        fr_time_delta_t idle_timeout;           //!< How long a connection can be idle before
                                                //!< being closed.
        fr_time_delta_t connect_timeout;        //!< New connection timeout, enforced by the create
                                                //!< callback.
 
        bool            spread;                 //!< If true we spread requests over the connections,
                                                //!< using the connection released longest ago, first.
 
        fr_heap_t       *heap;                  //!< For the next connection heap
 
        fr_pool_connection_t    *head;          //!< Start of the connection list.
        fr_pool_connection_t    *tail;          //!< End of the connection list.
 
        pthread_mutex_t mutex;                  //!< Mutex used to keep consistent state when making
                                                //!< modifications in threaded mode.
        pthread_cond_t  done_spawn;             //!< Threads that need to ensure no spawning is in progress,
                                                //!< should block on this condition if pending != 0.
        pthread_cond_t  done_reconnecting;      //!< Before calling the create callback, threads should
                                                //!< block on this condition if reconnecting == true.
 
        CONF_SECTION const *cs;                 //!< Configuration section holding the section of parsed
                                                //!< config file that relates to this pool.
        void            *opaque;                //!< Pointer to context data that will be passed to callbacks.
 
        char const      *log_prefix;            //!< Log prefix to prepend to all log messages created
                                                //!< by the connection pool code.
 
        bool            triggers_enabled;       //!< Whether we call the trigger functions.
 
        char const      *trigger_prefix;        //!< Prefix to prepend to names of all triggers
                                                //!< fired by the connection pool code.
        fr_pair_list_t  trigger_args;           //!< Arguments to make available in connection pool triggers.
 
        fr_time_delta_t held_trigger_min;       //!< If a connection is held for less than the specified
                                                //!< period, fire a trigger.
        fr_time_delta_t held_trigger_max;       //!< If a connection is held for longer than the specified
                                                //!< period, fire a trigger.
 
        fr_pool_connection_create_t     create; //!< Function used to create new connections.
        fr_pool_connection_alive_t      alive;  //!< Function used to check status of connections.
 
        fr_pool_reconnect_t     reconnect;      //!< Called during connection pool reconnect.
 
        fr_pool_state_t state;                  //!< Stats and state of the connection pool.
};
 
static const conf_parser_t pool_config[] = {
        { FR_CONF_OFFSET("start", fr_pool_t, start), .dflt = "0" },
        { FR_CONF_OFFSET("min", fr_pool_t, min), .dflt = "0" },
        { FR_CONF_OFFSET("max", fr_pool_t, max), .dflt_func = max_dflt },
        { FR_CONF_OFFSET("max_pending", fr_pool_t, max_pending), .dflt = "0" },
        { FR_CONF_OFFSET("spare", fr_pool_t, spare), .dflt = "3" },
        { FR_CONF_OFFSET("uses", fr_pool_t, max_uses), .dflt = "0" },
        { FR_CONF_OFFSET("lifetime", fr_pool_t, lifetime), .dflt = "0" },
        { FR_CONF_OFFSET("cleanup_interval", fr_pool_t, cleanup_interval), .dflt = "30" },
        { FR_CONF_OFFSET("idle_timeout", fr_pool_t, idle_timeout), .dflt = "60" },
        { FR_CONF_OFFSET("connect_timeout", fr_pool_t, connect_timeout), .dflt = "3.0" },
        { FR_CONF_OFFSET("held_trigger_min", fr_pool_t, held_trigger_min), .dflt = "0.0" },
        { FR_CONF_OFFSET("held_trigger_max", fr_pool_t, held_trigger_max), .dflt = "0.5" },
        { FR_CONF_OFFSET("retry_delay", fr_pool_t, retry_delay), .dflt = "1" },
        { FR_CONF_OFFSET("spread", fr_pool_t, spread), .dflt = "no" },
        CONF_PARSER_TERMINATOR
};
 
static int max_dflt(CONF_PAIR **out, UNUSED void *parent, CONF_SECTION *cs, fr_token_t quote, conf_parser_t const *rule)
{
        char            *strvalue;
 
        strvalue = talloc_asprintf(NULL, "%u", main_config->max_workers);
        *out = cf_pair_alloc(cs, rule->name1, strvalue, T_OP_EQ, T_BARE_WORD, quote);
        talloc_free(strvalue);
 
        return 0;
}
 
/** Order connections by reserved most recently
 */
static fr_cmp_ret_t last_reserved_cmp(void const *one, void const *two)
{
        fr_pool_connection_t const *a = one, *b = two;
 
        return fr_time_cmp(a->last_reserved, b->last_reserved);
}
 
/** Order connections by released longest ago
 */
static fr_cmp_ret_t last_released_cmp(void const *one, void const *two)
{
        fr_pool_connection_t const *a = one, *b = two;
 
        return fr_time_cmp(a->last_released, b->last_released);
}
 
/** Removes a connection from the connection list
 *
 * @note Must be called with the mutex held.
 *
 * @param[in] pool      to modify.
 * @param[in] this      Connection to delete.
 */
static void connection_unlink(fr_pool_t *pool, fr_pool_connection_t *this)
{
        if (this->prev) {
                fr_assert(pool->head != this);
                this->prev->next = this->next;
        } else {
                fr_assert(pool->head == this);
                pool->head = this->next;
        }
        if (this->next) {
                fr_assert(pool->tail != this);
                this->next->prev = this->prev;
        } else {
                fr_assert(pool->tail == this);
                pool->tail = this->prev;
        }
 
        this->prev = this->next = NULL;
}
 
/** Adds a connection to the head of the connection list
 *
 * @note Must be called with the mutex held.
 *
 * @param[in] pool      to modify.
 * @param[in] this      Connection to add.
 */
static void connection_link_head(fr_pool_t *pool, fr_pool_connection_t *this)
{
        fr_assert(pool != NULL);
        fr_assert(this != NULL);
        fr_assert(pool->head != this);
        fr_assert(pool->tail != this);
 
        if (pool->head) {
                pool->head->prev = this;
        }
 
        this->next = pool->head;
        this->prev = NULL;
        pool->head = this;
        if (!pool->tail) {
                fr_assert(this->next == NULL);
                pool->tail = this;
        } else {
                fr_assert(this->next != NULL);
        }
}
 
/** Send a connection pool trigger.
 *
 * @param[in] pool      to send trigger for.
 * @param[in] event     trigger name suffix.
 */
static inline void fr_pool_trigger(fr_pool_t *pool, char const *event)
{
        char    name[128];
 
        fr_assert(pool != NULL);
        fr_assert(event != NULL);
 
        if (!pool->triggers_enabled) return;
 
        snprintf(name, sizeof(name), "%s.%s", pool->trigger_prefix, event);
        trigger(unlang_interpret_get_thread_default(), pool->cs, NULL, name, true, &pool->trigger_args, pool);
}
 
/** Find a connection handle in the connection list
 *
 * Walks over the list of connections searching for a specified connection
 * handle and returns the first connection that contains that pointer.
 *
 * @note Must be called with the mutex held.
 *
 * @param[in] pool      to search in.
 * @param[in] conn      handle to search for.
 * @return
 *      - Connection containing the specified handle.
 *      - NULL if no such connection was found.
 */
static fr_pool_connection_t *connection_find(fr_pool_t *pool, void *conn)
{
        fr_pool_connection_t *this;
 
        /*
         *      FIXME: This loop could be avoided if we passed a 'void
         *      **connection' instead.  We could use "offsetof" in
         *      order to find top of the parent structure.
         */
        for (this = pool->head; this != NULL; this = this->next) {
                if (this->connection == conn) {
#ifdef PTHREAD_DEBUG
                        pthread_t pthread_id;
 
                        pthread_id = pthread_self();
                        fr_assert(pthread_equal(this->pthread_id, pthread_id) != 0);
#endif
 
                        fr_assert(this->in_use == true);
                        return this;
                }
        }
 
        return NULL;
}
 
/** Spawns a new connection
 *
 * Spawns a new connection using the create callback, and returns it for
 * adding to the connection list.
 *
 * @note Will call the 'open' trigger.
 * @note Must be called with the mutex held.  Returns with the mutex held.
 *       The mutex is released while the create callback runs.
 *
 * @param[in] pool      to modify.
 * @param[in] request   The current request.
 * @param[in] now       Current time.
 * @param[in] in_use    whether the new connection should be "in_use" or not
 * @return
 *      - New connection struct.
 *      - NULL on error.
 */
static fr_pool_connection_t *connection_spawn(fr_pool_t *pool, request_t *request, fr_time_t now, bool in_use)
{
        uint64_t                number;
        uint32_t                pending_window;
        TALLOC_CTX              *ctx;
 
        fr_pool_connection_t    *this;
        void                    *conn;
 
        fr_assert(pool != NULL);
        fr_assert(pool->state.num <= pool->max);
 
        /*
         *      If we have NO connections, and we've previously failed
         *      opening connections, don't open multiple connections until
         *      we successfully open at least one.
         */
        if ((pool->state.num == 0) &&
            pool->state.pending &&
            fr_time_gt(pool->state.last_failed, fr_time_wrap(0))) {
                return NULL;
        }
 
        /*
         *      Don't spawn too many connections at the same time.
         */
        if ((pool->state.num + pool->state.pending) >= pool->max) {
                ERROR("Cannot open new connection, already at max");
                return NULL;
        }
 
        /*
         *      If the last attempt failed, wait a bit before
         *      retrying.
         */
        if (fr_time_gt(pool->state.last_failed, fr_time_wrap(0)) &&
            fr_time_gt(fr_time_add(pool->state.last_failed, pool->retry_delay), now)) {
                bool complain = false;
                fr_time_delta_t retry_in = fr_time_sub(fr_time_add(pool->state.last_failed, pool->retry_delay), now);
 
                if (fr_time_delta_gteq(fr_time_sub(now, pool->state.last_throttled), fr_time_delta_from_sec(1))) {
                        complain = true;
 
                        pool->state.last_throttled = now;
                }
 
                if (!fr_rate_limit_enabled() || complain) {
                        ERROR("Last connection attempt failed, waiting %pV seconds before retrying",
                              fr_box_time_delta(retry_in));
                }
 
                return NULL;
        }
 
        /*
         *      We limit the rate of new connections after a failed attempt.
         */
        if (pool->state.pending > pool->pending_window) {
                RATE_LIMIT_GLOBAL_ROPTIONAL(RWARN, WARN,
                                            "Cannot open a new connection due to rate limit after failure");
 
                return NULL;
        }
 
        pool->state.pending++;
        number = pool->state.count++;
 
        /*
         *      Don't starve out the thread trying to reconnect
         *      the pool, by continuously opening new connections.
         */
        while (pool->state.reconnecting) pthread_cond_wait(&pool->done_reconnecting, &pool->mutex);
 
        /*
         *      The true value for pending_window is the smaller of
         *      free connection slots, or pool->pending_window.
         */
        pending_window = (pool->max - pool->state.num);
        if (pool->pending_window < pending_window) pending_window = pool->pending_window;
        ROPTIONAL(RDEBUG2, DEBUG2, "Opening additional connection (%" PRIu64 "), %u of %u pending slots used",
                  number, pool->state.pending, pending_window);
 
        /*
         *      Unlock the mutex while we try to open a new
         *      connection.  If there are issues with the back-end,
         *      opening a new connection may take a LONG time.  In
         *      that case, we want the other connections to continue
         *      to be used.
         */
        pthread_mutex_unlock(&pool->mutex);
 
        /*
         *      Allocate a new top level ctx for the create callback
         *      to hang its memory off of.
         */
        MEM(ctx = talloc_init("connection_ctx"));
 
        /*
         *      This may take a long time, which prevents other
         *      threads from releasing connections.  We don't care
         *      about other threads opening new connections, as we
         *      already have no free connections.
         */
        conn = pool->create(ctx, pool->opaque, pool->connect_timeout);
        if (!conn) {
                ERROR("Opening connection failed (%" PRIu64 ")", number);
 
                pthread_mutex_lock(&pool->mutex);
                pool->state.last_failed = now;
                pool->pending_window = 1;
                pool->state.pending--;
 
                /*
                 *      Must be done inside the mutex, reconnect callback
                 *      may modify args.
                 */
                fr_pool_trigger(pool, "fail");
                pthread_cond_broadcast(&pool->done_spawn);
 
                talloc_free(ctx);
 
                return NULL;
        }
 
        /*
         *      And lock the mutex again while we link the new
         *      connection back into the pool.
         */
        pthread_mutex_lock(&pool->mutex);
 
        this = talloc_zero(pool, fr_pool_connection_t);
        if (!this) {
                pool->state.last_failed = now;
                pool->state.pending--;
 
                pthread_cond_broadcast(&pool->done_spawn);
 
                ERROR("Memory allocation failed for new connection (%" PRIu64 ")", number);
 
                talloc_free(ctx);
 
                return NULL;
        }
        MEM(talloc_link_ctx(this, ctx) >= 0);
 
        this->created = now;
        this->connection = conn;
        this->in_use = in_use;
 
        this->number = number;
        this->last_reserved = fr_time();
        this->last_released = this->last_reserved;
 
        /*
         *      The connection pool is starting up.  Insert the
         *      connection into the heap.
         */
        if (!in_use) fr_heap_insert(&pool->heap, this);
 
        connection_link_head(pool, this);
 
        /*
         *      Do NOT insert the connection into the heap.  That's
         *      done when the connection is released.
         */
 
        pool->state.num++;
 
        fr_assert(pool->state.pending > 0);
        pool->state.pending--;
 
        /*
         *      We've successfully opened one more connection.  Allow
         *      more connections to open in parallel.
         */
        if ((pool->pending_window < pool->max) &&
            ((pool->max_pending == 0) || (pool->pending_window < pool->max_pending))) {
                pool->pending_window++;
        }
 
        pool->state.last_spawned = fr_time();
        pool->delay_interval = pool->cleanup_interval;
        pool->state.next_delay = pool->cleanup_interval;
        pool->state.last_failed = fr_time_wrap(0);
 
        /*
         *      Must be done inside the mutex, reconnect callback
         *      may modify args.
         */
        fr_pool_trigger(pool, "open");
 
        pthread_cond_broadcast(&pool->done_spawn);
 
        return this;
}
 
/** Close an existing connection.
 *
 * Removes the connection from the list, calls the delete callback to close
 * the connection, then frees memory allocated to the connection.
 *
 * @note Will call the 'close' trigger.
 * @note Must be called with the mutex held.
 *
 * @param[in] pool      to modify.
 * @param[in] this      Connection to delete.
 */
static void connection_close_internal(fr_pool_t *pool, fr_pool_connection_t *this)
{
        /*
         *      If it's in use, release it.
         */
        if (this->in_use) {
#ifdef PTHREAD_DEBUG
                pthread_t pthread_id = pthread_self();
                fr_assert(pthread_equal(this->pthread_id, pthread_id) != 0);
#endif
 
                this->in_use = false;
 
                fr_assert(pool->state.active != 0);
                pool->state.active--;
 
        } else {
                /*
                 *      Connection isn't used, remove it from the heap.
                 */
                fr_heap_extract(&pool->heap, this);
        }
 
        fr_pool_trigger(pool, "close");
 
        connection_unlink(pool, this);
 
        fr_assert(pool->state.num > 0);
        pool->state.num--;
        talloc_free(this);
}
 
/** Check whether a connection needs to be removed from the pool
 *
 * Will verify that the connection is within idle_timeout, max_uses, and
 * lifetime values. If it is not, the connection will be closed.
 *
 * @note Will only close connections not in use.
 * @note Must be called with the mutex held.
 *
 * @param[in] pool      to modify.
 * @param[in] request   The current request.
 * @param[in] this      Connection to manage.
 * @param[in] now       Current time.
 * @return
 *      - 0 if connection was closed.
 *      - 1 if connection handle was left open.
 */
static int connection_manage(fr_pool_t *pool, request_t *request, fr_pool_connection_t *this, fr_time_t now)
{
        fr_assert(pool != NULL);
        fr_assert(this != NULL);
 
        /*
         *      Don't terminated in-use connections
         */
        if (this->in_use) return 1;
 
        if (this->needs_reconnecting) {
                ROPTIONAL(RDEBUG2, DEBUG2, "Closing expired connection (%" PRIu64 "): Needs reconnecting",
                          this->number);
        do_delete:
                if (pool->state.num <= pool->min) {
                        ROPTIONAL(RDEBUG2, DEBUG2, "You probably need to lower \"min\"");
                }
                connection_close_internal(pool, this);
                return 0;
        }
 
        if ((pool->max_uses > 0) &&
            (this->num_uses >= pool->max_uses)) {
                ROPTIONAL(RDEBUG2, DEBUG2, "Closing expired connection (%" PRIu64 "): Hit max_uses limit",
                          this->number);
                goto do_delete;
        }
 
        if (fr_time_delta_ispos(pool->lifetime) &&
            (fr_time_lt(fr_time_add(this->created, pool->lifetime), now))) {
                ROPTIONAL(RDEBUG2, DEBUG2, "Closing expired connection (%" PRIu64 "): Hit lifetime limit",
                          this->number);
                goto do_delete;
        }
 
        if (fr_time_delta_ispos(pool->idle_timeout) &&
            (fr_time_lt(fr_time_add(this->last_released, pool->idle_timeout), now))) {
                ROPTIONAL(RINFO, INFO, "Closing connection (%" PRIu64 "): Hit idle_timeout, was idle for %pVs",
                          this->number, fr_box_time_delta(fr_time_sub(now, this->last_released)));
                goto do_delete;
        }
 
        return 1;
}
 
 
/** Check whether any connections need to be removed from the pool
 *
 * Maintains the number of connections in the pool as per the configuration
 * parameters for the connection pool.
 *
 * @note Will only run checks the first time it's called in a given second,
 * to throttle connection spawning/closing.
 * @note Will only close connections not in use.
 * @note Must be called with the mutex held.  Returns with the mutex held.
 *
 * @param[in] pool      to manage.
 * @param[in] request   The current request.
 * @return 1
 */
static int connection_check(fr_pool_t *pool, request_t *request)
{
        uint32_t                num, spare;
        fr_time_t               now = fr_time();
        fr_pool_connection_t    *this, *next;
 
        if (fr_time_delta_lt(fr_time_sub(now, pool->state.last_checked), fr_time_delta_from_sec(1))) return 1;
 
        /*
         *      Get "real" number of connections, and count pending
         *      connections as spare.
         */
        num = pool->state.num + pool->state.pending;
        spare = pool->state.pending + (pool->state.num - pool->state.active);
 
        /*
         *      The other end can close connections.  If so, we'll
         *      have fewer than "min".  When that happens, open more
         *      connections to enforce "min".
         *
         *      The code for spawning connections enforces that
         *      num + pending <= max.
         */
        if (num < pool->min) {
                ROPTIONAL(RINFO, INFO, "Need %" PRIu32 " more connections to reach min connections (%" PRIu32 ")", pool->min - num, pool->min);
                goto add_connection;
        }
 
        /*
         *      On the odd chance that we've opened too many
         *      connections, take care of that.
         */
        if (num > pool->max) {
                /*
                 *      Pending connections don't get closed as "spare".
                 */
                if (pool->state.pending > 0) goto manage_connections;
 
                /*
                 *      Otherwise close one of the connections to
                 *      bring us down to "max".
                 */
                goto close_connection;
        }
 
        /*
         *      Now that we've enforced min/max connections, try to
         *      keep the "spare" connections at the correct number.
         */
 
        /*
         *      Nothing to do?  Go check all of the connections for
         *      timeouts, etc.
         */
        if (spare == pool->spare) goto manage_connections;
 
        /*
         *      Too many spare connections, delete some.
         */
        if (spare > pool->spare) {
                fr_pool_connection_t *found;
 
                /*
                 *      Pending connections don't get closed as "spare".
                 */
                if (pool->state.pending > 0) goto manage_connections;
 
                /*
                 *      Don't close too many connections, even they
                 *      are spare.
                 */
                if (num <= pool->min) goto manage_connections;
 
                /*
                 *      Too many spares, go close one.
                 */
 
        close_connection:
                /*
                 *      Don't close connections too often, in order to
                 *      prevent flapping.
                 */
                if (fr_time_lt(now, fr_time_add(pool->state.last_spawned, pool->delay_interval))) goto manage_connections;
 
                /*
                 *      Find a connection to close.
                 */
                found = NULL;
                for (this = pool->tail; this != NULL; this = this->prev) {
                        if (this->in_use) continue;
 
                        if (!found || (fr_time_lt(this->last_reserved, found->last_reserved))) found = this;
                }
 
                if (!fr_cond_assert(found)) goto done;
 
                ROPTIONAL(RDEBUG2, DEBUG2, "Closing connection (%" PRIu64 ") as we have too many unused connections",
                          found->number);
                connection_close_internal(pool, found);
 
                /*
                 *      Decrease the delay for the next time we clean
                 *      up.
                 */
                pool->state.next_delay = fr_time_delta_wrap(fr_time_delta_unwrap(pool->state.next_delay) >> 1);
                if (!fr_time_delta_ispos(pool->state.next_delay)) pool->state.next_delay = fr_time_delta_wrap(1);
                pool->delay_interval = fr_time_delta_add(pool->delay_interval, pool->state.next_delay);
 
                goto manage_connections;
        }
 
        /*
         *      Too few connections, open some more.
         */
        if (spare < pool->spare) {
                /*
                 *      Don't open too many pending connections.
                 */
                if (pool->state.pending >= pool->pending_window) goto manage_connections;
 
                /*
                 *      Don't open too many connections, even if we
                 *      need more spares.
                 */
                if (num >= pool->max) goto manage_connections;
 
                /*
                 *      Too few spares, go add one.
                 */
                ROPTIONAL(RINFO, INFO, "Need %" PRIu32 " more connections to reach %" PRIu32 " spares", pool->spare - spare, pool->spare);
 
        add_connection:
                /*
                 *      Only try to open spares if we're not already attempting to open
                 *      a connection. Avoids spurious log messages.
                 */
                (void) connection_spawn(pool, request, now, false);
                goto manage_connections;
        }
 
        /*
         *      Pass over all of the connections in the pool, limiting
         *      lifetime, idle time, max requests, etc.
         */
manage_connections:
        for (this = pool->head; this != NULL; this = next) {
                next = this->next;
                connection_manage(pool, request, this, now);
        }
 
        pool->state.last_checked = now;
 
done:
        return 1;
}
 
/** Get a connection from the connection pool
 *
 * @note Must be called with the mutex free.
 *
 * @param[in] pool      to reserve the connection from.
 * @param[in] request   The current request.
 * @param[in] spawn     whether to spawn a new connection
 * @return
 *      - A pointer to the connection handle.
 *      - NULL on error.
 */
static void *connection_get_internal(fr_pool_t *pool, request_t *request, bool spawn)
{
        fr_time_t now;
        fr_pool_connection_t *this;
 
        if (!pool) return NULL;
 
        pthread_mutex_lock(&pool->mutex);
 
        now = fr_time();
 
        /*
         *      Grab the link with the lowest latency, and check it
         *      for limits.  If "connection manage" says the link is
         *      no longer usable, go grab another one.
         */
        do {
                this = fr_heap_peek(pool->heap);
                if (!this) break;
        } while (!connection_manage(pool, request, this, now));
 
        /*
         *      We have a working connection.  Extract it from the
         *      heap and use it.
         */
        if (this) {
                fr_heap_extract(&pool->heap, this);
                goto do_return;
        }
 
        if (pool->state.num == pool->max) {
                bool complain = false;
 
                /*
                 *      Rate-limit complaints.
                 */
                if (fr_time_delta_gt(fr_time_sub(now, pool->state.last_at_max), fr_time_delta_from_sec(1))) {
                        complain = true;
                        pool->state.last_at_max = now;
                }
 
                if (!fr_rate_limit_enabled() || complain) {
                        ERROR("No connections available and at max connection limit");
 
                        /*
                         *      Must be done inside the mutex, reconnect callback
                         *      may modify args.
                         */
                        fr_pool_trigger(pool, "none");
                }
                pthread_mutex_unlock(&pool->mutex);
 
                return NULL;
        }
 
        if (!spawn) {
                pthread_mutex_unlock(&pool->mutex);
                return NULL;
        }
 
        ROPTIONAL(RDEBUG2, DEBUG2, "%" PRIu32 " of %u connections in use.  You may need to increase \"spare\"",
               pool->state.active, pool->state.num);
 
        this = connection_spawn(pool, request, now, true);
        if (!this) {
                pthread_mutex_unlock(&pool->mutex);
                return NULL;
        }
 
do_return:
        pool->state.active++;
        this->num_uses++;
        this->last_reserved = fr_time();
        this->in_use = true;
 
#ifdef PTHREAD_DEBUG
        this->pthread_id = pthread_self();
#endif
        pthread_mutex_unlock(&pool->mutex);
 
        ROPTIONAL(RDEBUG2, DEBUG2, "Reserved connection (%" PRIu64 ")", this->number);
 
        return this->connection;
}
 
/** Enable triggers for a connection pool
 *
 * @param[in] pool              to enable triggers for.
 * @param[in] trigger_prefix    prefix to prepend to all trigger names.  Usually a path
 *                              to the module's trigger configuration e.g.
 *                              @verbatim modules.<name>.pool @endverbatim
 *                              @verbatim <trigger name> @endverbatim is appended to form
 *                              the complete path.
 * @param[in] trigger_args      to make available in any triggers executed by the connection pool.
 *                              These will usually be fr_pair_t (s) describing the host
 *                              associated with the pool.
 *                              Trigger args will be copied, input trigger_args should be freed
 *                              if necessary.
 */
void fr_pool_enable_triggers(fr_pool_t *pool, char const *trigger_prefix, fr_pair_list_t *trigger_args)
{
        pool->triggers_enabled = true;
 
        talloc_const_free(pool->trigger_prefix);
        MEM(pool->trigger_prefix = trigger_prefix ? talloc_strdup(pool, trigger_prefix) : "");
 
        fr_pair_list_free(&pool->trigger_args);
 
        if (!trigger_args) return;
 
        MEM(fr_pair_list_copy(pool, &pool->trigger_args, trigger_args) >= 0);
}
 
/** Create a new connection pool
 *
 * Allocates structures used by the connection pool, initialises the various
 * configuration options and counters, and sets the callback functions.
 *
 * Will also spawn the number of connections specified by the 'start' configuration
 * option.
 *
 * @note Will call the 'start' trigger.
 *
 * @param[in] ctx               Context to link pool's destruction to.
 * @param[in] cs                pool section.
 * @param[in] opaque data       pointer to pass to callbacks.
 * @param[in] c                 Callback to create new connections.
 * @param[in] a                 Callback to check the status of connections.
 * @param[in] log_prefix        prefix to prepend to all log messages.
 * @return
 *      - New connection pool.
 *      - NULL on error.
 */
fr_pool_t *fr_pool_init(TALLOC_CTX *ctx,
                        CONF_SECTION const *cs,
                        void *opaque,
                        fr_pool_connection_create_t c, fr_pool_connection_alive_t a,
                        char const *log_prefix)
{
        fr_pool_t               *pool = NULL;
 
        if (!cs || !opaque || !c) return NULL;
 
        /*
         *      Pool is allocated in the NULL context as
         *      threads are likely to allocate memory
         *      beneath the pool.
         */
        MEM(pool = talloc_zero(NULL, fr_pool_t));
        fr_pair_list_init(&pool->trigger_args);
 
        /*
         *      Ensure the pool is freed at the same time
         *      as its parent.
         */
        if (ctx && (talloc_link_ctx(ctx, pool) < 0)) {
                PERROR("%s: Failed linking pool ctx", __FUNCTION__);
                talloc_free(pool);
 
                return NULL;
        }
 
        pool->cs = cs;
        pool->opaque = opaque;
        pool->create = c;
        pool->alive = a;
 
        pool->head = pool->tail = NULL;
 
        /*
         *      We keep a heap of connections, sorted by the last time
         *      we STARTED using them.  Newly opened connections
         *      aren't in the heap.  They're only inserted in the list
         *      once they're released.
         *
         *      We do "most recently started" instead of "most
         *      recently used", because MRU is done as most recently
         *      *released*.  We want to order connections by
         *      responsiveness, and MRU prioritizes high latency
         *      connections.
         *
         *      We want most recently *started*, which gives
         *      preference to low latency links, and pushes high
         *      latency links down in the priority heap.
         *
         *      https://code.facebook.com/posts/1499322996995183/solving-the-mystery-of-link-imbalance-a-metastable-failure-state-at-scale/
         */
        if (!pool->spread) {
                pool->heap = fr_heap_talloc_alloc(pool, last_reserved_cmp, fr_pool_connection_t, heap_id, 0);
        /*
         *      For some types of connections we need to used a different
         *      algorithm, because load balancing benefits are secondary
         *      to maintaining a cache of open connections.
         *
         *      With libcurl's multihandle, connections can only be reused
         *      if all handles that make up the multihandle are done processing
         *      their requests.
         *
         *      We can't tell when that's happened using libcurl, and even
         *      if we could, blocking until all servers had responded
         *      would have huge cost.
         *
         *      The solution is to order the heap so that the connection that
         *      was released longest ago is at the top.
         *
         *      That way we maximise time between connection use.
         */
        } else {
                pool->heap = fr_heap_talloc_alloc(pool, last_released_cmp, fr_pool_connection_t, heap_id, 0);
        }
        if (!pool->heap) {
                ERROR("%s: Failed creating connection heap", __FUNCTION__);
        error:
                fr_pool_free(pool);
                return NULL;
        }
 
        pool->log_prefix = log_prefix ? talloc_strdup(pool, log_prefix) : "core";
        pthread_mutex_init(&pool->mutex, NULL);
        pthread_cond_init(&pool->done_spawn, NULL);
        pthread_cond_init(&pool->done_reconnecting, NULL);
 
        DEBUG2("Initialising connection pool");
 
        if (cf_section_rules_push(UNCONST(CONF_SECTION *, cs), pool_config) < 0) goto error;
        if (cf_section_parse(pool, pool, UNCONST(CONF_SECTION *, cs)) < 0) {
                PERROR("Configuration parsing failed");
                goto error;
        }
 
        /*
         *      Some simple limits
         */
        if (pool->max == 0) {
                cf_log_err(cs, "Cannot set 'max' to zero");
                goto error;
        }
 
        pool->pending_window = (pool->max_pending > 0) ? pool->max_pending : pool->max;
 
        if (pool->min > pool->max) {
                cf_log_err(cs, "Cannot set 'min' to more than 'max'");
                goto error;
        }
 
        FR_INTEGER_BOUND_CHECK("max", pool->max, <=, 1024);
        FR_INTEGER_BOUND_CHECK("start", pool->start, <=, pool->max);
        FR_INTEGER_BOUND_CHECK("spare", pool->spare, <=, (pool->max - pool->min));
 
        if (fr_time_delta_ispos(pool->lifetime)) {
                FR_TIME_DELTA_COND_CHECK("idle_timeout", pool->idle_timeout,
                                         fr_time_delta_lteq(pool->idle_timeout, pool->lifetime), fr_time_delta_wrap(0));
        }
 
        if (fr_time_delta_ispos(pool->idle_timeout)) {
                FR_TIME_DELTA_BOUND_CHECK("cleanup_interval", pool->cleanup_interval, <=, pool->idle_timeout);
        }
 
        /*
         *      Some libraries treat 0.0 as infinite timeout, others treat it
         *      as instantaneous timeout.  Solve the inconsistency by making
         *      the smallest allowable timeout 100ms.
         */
        FR_TIME_DELTA_BOUND_CHECK("connect_timeout", pool->connect_timeout, >=, fr_time_delta_from_msec(100));
 
        /*
         *      Don't open any connections.  Instead, force the limits
         *      to only 1 connection.
         *
         */
        if (check_config) pool->start = pool->min = pool->max = 1;
 
        return pool;
}
 
int fr_pool_start(fr_pool_t *pool)
{
        uint32_t                i;
        fr_pool_connection_t    *this;
 
        /*
         *      Don't spawn any connections
         */
        if (check_config) return 0;
 
        /*
         *      Create all of the connections, unless the admin says
         *      not to.
         */
        pthread_mutex_lock(&pool->mutex);
        for (i = 0; i < pool->start; i++) {
                /*
                 *      Call time() once for each spawn attempt as there
                 *      could be a significant delay.
                 */
                this = connection_spawn(pool, NULL, fr_time(), false);
                if (!this) {
                        pthread_mutex_unlock(&pool->mutex);
                        ERROR("Failed spawning initial connections");
                        return -1;
                }
        }
        pthread_mutex_unlock(&pool->mutex);
 
        fr_pool_trigger(pool, "start");
 
        return 0;
}
 
/** Allocate a new pool using an existing one as a template
 *
 * @param[in] ctx       to allocate new pool in.
 * @param[in] pool      to copy.
 * @param[in] opaque    data to pass to connection function.
 * @return
 *      - New connection pool.
 *      - NULL on error.
 */
fr_pool_t *fr_pool_copy(TALLOC_CTX *ctx, fr_pool_t *pool, void *opaque)
{
        fr_pool_t *copy;
 
        copy = fr_pool_init(ctx, pool->cs, opaque, pool->create, pool->alive, pool->log_prefix);
        if (!copy) return NULL;
 
        if (pool->trigger_prefix) fr_pool_enable_triggers(copy, pool->trigger_prefix, &pool->trigger_args);
 
        return copy;
}
 
/** Get the number of connections currently in the pool
 *
 * @param[in] pool to count connections for.
 * @return the number of connections in the pool
 */
fr_pool_state_t const *fr_pool_state(fr_pool_t *pool)
{
        return &pool->state;
}
 
/** Connection pool get timeout
 *
 * @param[in] pool to get connection timeout for.
 * @return the connection timeout configured for the pool.
 */
fr_time_delta_t fr_pool_timeout(fr_pool_t *pool)
{
        return pool->connect_timeout;
}
 
/** Connection pool get start
 *
 * @param[in] pool to get connection start for.
 * @return the connection start value configured for the pool.
 */
int fr_pool_start_num(fr_pool_t *pool)
{
        return pool->start;
}
 
/** Return the opaque data associated with a connection pool
 *
 * @param pool to return data for.
 * @return opaque data associated with pool.
 */
void const *fr_pool_opaque(fr_pool_t *pool)
{
        return pool->opaque;
}
 
/** Increment pool reference by one.
 *
 * @param[in] pool to increment reference counter for.
 */
void fr_pool_ref(fr_pool_t *pool)
{
        pool->ref++;
}
 
/** Set a reconnection callback for the connection pool
 *
 * This can be called at any time during the pool's lifecycle.
 *
 * @param[in] pool to set reconnect callback for.
 * @param reconnect callback to call when reconnecting pool's connections.
 */
void fr_pool_reconnect_func(fr_pool_t *pool, fr_pool_reconnect_t reconnect)
{
        pool->reconnect = reconnect;
}
 
/** Mark connections for reconnection, and spawn at least 'start' connections
 *
 * @note This call may block whilst waiting for pending connection attempts to complete.
 *
 * This intended to be called on a connection pool that's in use, to have it reflect
 * a configuration change, or because the administrator knows that all connections
 * in the pool are inviable and need to be reconnected.
 *
 * @param[in] pool      to reconnect.
 * @param[in] request   The current request.
 * @return
 *      -  0 On success.
 *      - -1 If we couldn't create start connections, this may be ignored
 *           depending on the context in which this function is being called.
 */
int fr_pool_reconnect(fr_pool_t *pool, request_t *request)
{
        uint32_t                i;
        fr_pool_connection_t    *this;
        fr_time_t               now;
 
        pthread_mutex_lock(&pool->mutex);
 
        /*
         *      Pause new spawn attempts (we release the mutex
         *      during our cond wait).
         */
        pool->state.reconnecting = true;
 
        /*
         *      When the loop exits, we'll hold the lock for the pool,
         *      and we're guaranteed the connection create callback
         *      will not be using the opaque data.
         */
        while (pool->state.pending) pthread_cond_wait(&pool->done_spawn, &pool->mutex);
 
        /*
         *      We want to ensure at least 'start' connections
         *      have been reconnected. We can't call reconnect
         *      because, we might get the same connection each
         *      time we reserve one, so we close 'start'
         *      connections, and then attempt to spawn them again.
         */
        for (i = 0; i < pool->start; i++) {
                this = fr_heap_peek(pool->heap);
                if (!this) break;       /* There wasn't 'start' connections available */
 
                connection_close_internal(pool, this);
        }
 
        /*
         *      Mark all remaining connections in the pool as
         *      requiring reconnection.
         */
        for (this = pool->head; this; this = this->next) this->needs_reconnecting = true;
 
        /*
         *      Call the reconnect callback (if one's set)
         *      This may modify the opaque data associated
         *      with the pool.
         */
        if (pool->reconnect) pool->reconnect(pool, pool->opaque);
 
        /*
         *      Must be done inside the mutex, reconnect callback
         *      may modify args.
         */
        fr_pool_trigger(pool, "reconnect");
 
        /*
         *      Allow new spawn attempts, and wakeup any threads
         *      waiting to spawn new connections.
         */
        pool->state.reconnecting = false;
        pthread_cond_broadcast(&pool->done_reconnecting);
 
        now = fr_time();
 
        /*
         *      Now attempt to spawn 'start' connections.
         */
        for (i = 0; i < pool->start; i++) {
                this = connection_spawn(pool, request, now, false);
                if (!this) {
                        pthread_mutex_unlock(&pool->mutex);
                        return -1;
                }
        }
        pthread_mutex_unlock(&pool->mutex);
 
        return 0;
}
 
/** Delete a connection pool
 *
 * Closes, unlinks and frees all connections in the connection pool, then frees
 * all memory used by the connection pool.
 *
 * @note Will call the 'stop' trigger.
 * @note Must be called with the mutex free.
 *
 * @param[in,out] pool to delete.
 */
void fr_pool_free(fr_pool_t *pool)
{
        fr_pool_connection_t *this;
 
        if (!pool) return;
 
        /*
         *      More modules hold a reference to this pool, don't free
         *      it yet.
         */
        if (pool->ref > 0) {
                pool->ref--;
                return;
        }
 
        DEBUG2("Removing connection pool");
 
        pthread_mutex_lock(&pool->mutex);
 
        /*
         *      Don't loop over the list.  Just keep removing the head
         *      until they're all gone.
         */
        while ((this = pool->head) != NULL) {
                INFO("Closing connection (%" PRIu64 ")", this->number);
 
                connection_close_internal(pool, this);
        }
 
        fr_pool_trigger(pool, "stop");
 
        fr_assert(pool->head == NULL);
        fr_assert(pool->tail == NULL);
        fr_assert(pool->state.num == 0);
 
        pthread_mutex_unlock(&pool->mutex);
        pthread_mutex_destroy(&pool->mutex);
        pthread_cond_destroy(&pool->done_spawn);
        pthread_cond_destroy(&pool->done_reconnecting);
 
        talloc_free(pool);
}
 
/** Reserve a connection in the connection pool
 *
 * Will attempt to find an unused connection in the connection pool, if one is
 * found, will mark it as in use, and increment the number of active connections
 * and return the connection handle.
 *
 * If no free connections are found will attempt to spawn a new one, conditional
 * on a connection spawning not already being in progress, and not being at the
 * 'max' connection limit.
 *
 * @note fr_pool_connection_release must be called once the caller has finished
 * using the connection.
 *
 * @see fr_pool_connection_release
 * @param[in] pool      to reserve the connection from.
 * @param[in] request   The current request.
 * @return
 *      - A pointer to the connection handle.
 *      - NULL on error.
 */
void *fr_pool_connection_get(fr_pool_t *pool, request_t *request)
{
        return connection_get_internal(pool, request, true);
}
 
/** Release a connection
 *
 * Will mark a connection as unused and decrement the number of active
 * connections.
 *
 * @see fr_pool_connection_get
 * @param[in] pool      to release the connection in.
 * @param[in] request   The current request.
 * @param[in] conn      to release.
 */
void fr_pool_connection_release(fr_pool_t *pool, request_t *request, void *conn)
{
        fr_pool_connection_t    *this;
        fr_time_delta_t         held;
        bool                    trigger_min = false, trigger_max = false;
 
        if (!conn) return;
 
        pthread_mutex_lock(&pool->mutex);
        this = connection_find(pool, conn);
        if (!this) {
                pthread_mutex_unlock(&pool->mutex);
                return;
        }
 
        this->in_use = false;
 
        /*
         *      Record when the connection was last released
         */
        this->last_released = fr_time();
        pool->state.last_released = this->last_released;
 
        /*
         *      This is done inside the mutex to ensure
         *      updates are atomic.
         */
        held = fr_time_sub(this->last_released, this->last_reserved);
 
        /*
         *      Check we've not exceeded out trigger limits
         *
         *      These should only fire once per second.
         */
        if (fr_time_delta_ispos(pool->held_trigger_min) &&
            (fr_time_delta_lt(held, pool->held_trigger_min)) &&
            (fr_time_delta_gteq(fr_time_sub(this->last_released, pool->state.last_held_min), fr_time_delta_from_sec(1)))) {
                trigger_min = true;
                pool->state.last_held_min = this->last_released;
        }
 
        if (fr_time_delta_ispos(pool->held_trigger_max) &&
            (fr_time_delta_gt(held, pool->held_trigger_max)) &&
            (fr_time_delta_gteq(fr_time_sub(this->last_released, pool->state.last_held_max), fr_time_delta_from_sec(1)))) {
                trigger_max = true;
                pool->state.last_held_max = this->last_released;
        }
 
        /*
         *      Insert the connection in the heap.
         *
         *      This will either be based on when we *started* using it
         *      (allowing fast links to be re-used, and slow links to be
         *      gradually expired), or when we released it (allowing
         *      the maximum amount of time between connection use).
         */
        fr_heap_insert(&pool->heap, this);
 
        fr_assert(pool->state.active != 0);
        pool->state.active--;
 
        ROPTIONAL(RDEBUG2, DEBUG2, "Released connection (%" PRIu64 ")", this->number);
 
        /*
         *      We mirror the "spawn on get" functionality by having
         *      "delete on release".  If there are too many spare
         *      connections, go manage the pool && clean some up.
         */
        connection_check(pool, request);
        pthread_mutex_unlock(&pool->mutex);
 
        if (trigger_min) fr_pool_trigger(pool, "min");
        if (trigger_max) fr_pool_trigger(pool, "max");
}
 
/** Reconnect a suspected inviable connection
 *
 * This should be called by the module if it suspects that a connection is
 * not viable (e.g. the server has closed it).
 *
 * When implementing a module that uses the connection pool API, it is advisable
 * to pass a pointer to the pointer to the handle (void **conn)
 * to all functions which may call reconnect. This is so that if a new handle
 * is created and returned, the handle pointer can be updated up the callstack,
 * and a function higher up the stack doesn't attempt to use a now invalid
 * connection handle.
 *
 * @note Will free any talloced memory hung off the context of the connection,
 *      being reconnected.
 *
 * @warning After calling reconnect the caller *MUST NOT* attempt to use
 *      the old handle in any other operations, as its memory will have been
 *      freed.
 *
 * @see fr_pool_connection_get
 * @param[in] pool      to reconnect the connection in.
 * @param[in] request   The current request.
 * @param[in] conn      to reconnect.
 * @return new connection handle if successful else NULL.
 */
void *fr_pool_connection_reconnect(fr_pool_t *pool, request_t *request, void *conn)
{
        fr_pool_connection_t    *this;
 
        if (!pool || !conn) return NULL;
 
        pthread_mutex_lock(&pool->mutex);
        this = connection_find(pool, conn);
        if (!this) {
                pthread_mutex_unlock(&pool->mutex);
                return NULL;
        }
 
        ROPTIONAL(RINFO, INFO, "Deleting inviable connection (%" PRIu64 ")", this->number);
 
        connection_close_internal(pool, this);
        connection_check(pool, request);
        pthread_mutex_unlock(&pool->mutex);
 
        /*
         *      Return an existing connection or spawn a new one.
         */
        return connection_get_internal(pool, request, true);
}
 
/** Delete a connection from the connection pool.
 *
 * Resolves the connection handle to a connection, then (if found)
 * closes, unlinks and frees that connection.
 *
 * @note Must be called with the mutex free.
 *
 * @param[in] pool      Connection pool to modify.
 * @param[in] request   The current request.
 * @param[in] conn      to delete.
 * @return
 *      - 0 If the connection could not be found.
 *      - 1 if the connection was deleted.
 */
int fr_pool_connection_close(fr_pool_t *pool, request_t *request, void *conn)
{
        fr_pool_connection_t *this;
 
        if (!conn) return 0;
 
        pthread_mutex_lock(&pool->mutex);
        this = connection_find(pool, conn);
        if (!this) {
                pthread_mutex_unlock(&pool->mutex);
                return 0;
        }
 
        /*
         *      Record the last time a connection was closed
         */
        pool->state.last_closed = fr_time();
 
        ROPTIONAL(RINFO, INFO, "Deleting connection (%" PRIu64 ")", this->number);
 
        connection_close_internal(pool, this);
        connection_check(pool, request);
        pthread_mutex_unlock(&pool->mutex);
 
        return 1;
}