pipeline.c

Go to the documentation of this file.

/*
 *   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: 2cf5735c8be7881c78af69da0b0ce96a85a1ef0d $
 * @file lib/redis/pipeline.c
 * @brief Functions for pipelining commands.
 *
 * @copyright 2019 The FreeRADIUS server project
 * @copyright 2019 Network RADIUS SAS (legal@networkradius.com)
 *
 * @author Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 */
 
#include <freeradius-devel/server/connection.h>
#include <freeradius-devel/server/trunk.h>
 
#include "pipeline.h"
#include "io.h"
 
 
/** Thread local state for a cluster
 *
 * MOVE ME TO NEW ASYNC CLUSTER CODE
 */
struct fr_redis_cluster_thread_s {
        fr_event_list_t                 *el;
        trunk_conf_t    const           *tconf;         //!< Configuration for all trunks in the cluster.
        char                            *log_prefix;    //!< Common log prefix to use for all cluster related
                                                        ///< messages.
        bool                            delay_start;    //!< Prevent connections from spawning immediately.
};
 
/** The thread local free list
 *
 * Any entries remaining in the list will be freed when the thread is joined
 */
static _Thread_local fr_dlist_head_t *command_set_free_list;
 
typedef enum {
        FR_REDIS_COMMAND_NORMAL = 0,                    //!< A normal, non-transactional command.
        FR_REDIS_COMMAND_TRANSACTION_START,             //!< Start of a transaction block. Either WATCH or MULTI.
                                                        ///< if a transaction is started with WATCH, then multi
                                                        ///< is not marked up as a transaction start.
        FR_REDIS_COMMAND_TRANSACTION_END                //!< End of a transaction block. Either EXEC or DISCARD.
                                                        ///< If this command fails with
                                                        ///< MOVED or ASK, all commands back to the previous
                                                        ///< MULTI command must be requeued.
} fr_redis_command_type_t;
 
/** Represents a single command
 *
 */
struct fr_redis_command_s {
        fr_redis_command_set_t          *cmds;          //!< Command set this entry belongs to.
        fr_dlist_t                      entry;          //!< Entry in the command buffer.
 
        fr_redis_command_type_t         type;           //!< Redis command type.
 
        char const                      *str;           //!< The command string.
        size_t                          len;            //!< Length of the command string.
 
        uint64_t                        sqn;            //!< The sequence number of the command.  This is only
                                                        ///< valid for a specific handle, and is unique within
                                                        ///< the handle.
 
        redisReply                      *result;        //!< The result from the REDIS server.
};
 
/** Represents a collection of pipelined commands
 *
 * Commands MUST map to the same cluster node if using clustering.
 */
struct fr_redis_command_set_s {
        fr_dlist_t                      entry;
 
        /** @name Command state lists
         * @{
         */
        fr_dlist_head_t                 pending;        //!< Commands yet to be sent.
        fr_dlist_head_t                 sent;           //!< Commands sent.
        fr_dlist_head_t                 completed;      //!< Commands complete with replies.
        /** @} */
 
        uint8_t                         redirected;     //!< How many times this command set was redirected.
 
        /** @name Request state
         *
         * treq and request are duplicated here with the trunk code.
         * The reason for this, is because a fr_command_set_t, may need to be transferred
         * between trunks when redirects are being followed, and so we need this information
         * encapsulated within the command set, not just within the trunk.
         * @{
         */
        trunk_request_t         *treq;          //!< Trunk request this command set is associated with.
        request_t                       *request;       //!< Request this commands set is associated with (if any).
        void                            *rctx;          //!< Resume context to write results to.
        /** @} */
 
        /** @name Callback functions
         * @{
         */
        fr_redis_command_set_complete_t complete;       //!< Notify the creator of the command set
                                                        ///< that the command set has executed to
                                                        ///< to completion.  We have results for
                                                        ///< all commands.
 
        fr_redis_command_set_fail_t     fail;           //!< Notify the creator of the command set
                                                        ///< that the command set failed to execute
                                                        ///< to completion.  Partial results will
                                                        ///< be available.
        /** @} */
 
        /** @name Command set transaction stats
         *
         * We do these checks as REDIS commands from a great number of requests may pipeline
         * requests on the same connection and leaving a transaction open would be fairly
         * catastrophic, potentially causing errors across all future command sets set to
         * the connection.
         * @{
         */
        bool                            txn_watch;      //!< Transaction was started with a watch statement.
        uint16_t                        txn_start;      //!< Number of times a transaction block was started
                                                        ///< in this command set.
        uint16_t                        txn_end;        //!< The number of times a transaction block ended
                                                        ///< in this command set.
 
        /** @} */
};
 
struct fr_redis_trunk_s {
        fr_redis_io_conf_t const        *io_conf;       //!< Redis I/O configuration.  Specifies how to connect
                                                        ///< to the host this trunk is used to communicate with.
        trunk_t                 *trunk;         //!< Trunk containing all the connections to a specific
                                                        ///< host.
        fr_redis_cluster_thread_t       *cluster;       //!< Cluster this trunk belongs to.
};
 
/** Free any free requests when the thread is joined
 *
 */
static int _command_set_free_list_free_on_exit(void *arg)
{
        fr_dlist_head_t         *list = talloc_get_type_abort(arg, fr_dlist_head_t);
        fr_redis_command_set_t  *cmds;
 
        /*
         *      See the destructor for why this works
         */
        while ((cmds = fr_dlist_head(list))) if (talloc_free(cmds) < 0) return -1;
        return talloc_free(list);
}
 
/** Free a command set
 *
 */
static int _redis_command_set_free(fr_redis_command_set_t *cmds)
{
        if (fr_dlist_num_elements(command_set_free_list) >= 1024) return 0;     /* Keep a buffer of 1024 */
 
        /*
         *      Freed from the free list....
         */
        if (unlikely(fr_dlist_entry_in_list(&cmds->entry))) {
                fr_dlist_entry_unlink(&cmds->entry);    /* Don't trust the list head to be available */
                return 0;
        }
 
        talloc_free_children(cmds);
        memset(&cmds, 0, sizeof(cmds));
 
        fr_dlist_insert_head(command_set_free_list, cmds);
 
        return -1;      /* Prevent the free */
}
 
/** Allocate a new command set
 *
 * This is a set of commands that the calling module wants to execute
 * on the redis server in sequence.
 *
 * Control will be returned to the caller via the registered complete
 * and fail functions.
 *
 * @param[in] ctx       to bind the command set's lifetime to.
 * @param[in] request   to pass to places that need it.
 * @param[in] complete  Function to call when all commands have been processed.
 * @param[in] fail      Function to call if the command set was not executed
 *                      or was partially executed.
 * @param[in] rctx      Resume context to pass to complete and fail functions.
 * @return A new or refurbished command set.
 */
fr_redis_command_set_t *fr_redis_command_set_alloc(TALLOC_CTX *ctx,
                                                   request_t *request,
                                                   fr_redis_command_set_complete_t complete,
                                                   fr_redis_command_set_fail_t fail,
                                                   void *rctx)
 
{
        fr_redis_command_set_t  *cmds;
        fr_dlist_head_t         *free_list;
 
#define COMMAND_PRE_ALLOC_COUNT 8       //!< How much room we pre-allocate for commands.
#define COMMAND_PRE_ALLOC_LEN   64      //!< How much we allocate for each command string.
 
        /*
         *      Initialise the free list
         */
        if (unlikely(!command_set_free_list)) {
                MEM(free_list = talloc(NULL, fr_dlist_head_t));
                fr_dlist_init(free_list, fr_redis_command_set_t, entry);
                fr_atexit_thread_local(command_set_free_list, _command_set_free_list_free_on_exit, free_list);
        } else {
                free_list = command_set_free_list;
        }
 
        /*
         *      Pull an element out of the free list
         *      or allocate a new one.
         */
        cmds = fr_dlist_head(free_list);
        if (!cmds) {
                MEM(cmds = talloc_zero_pooled_object(NULL, fr_redis_command_set_t,
                                                     COMMAND_PRE_ALLOC_COUNT,
                                                     COMMAND_PRE_ALLOC_COUNT * (sizeof(fr_redis_command_t) +
                                                     COMMAND_PRE_ALLOC_LEN)));
                talloc_set_destructor(cmds, _redis_command_set_free);
                fr_dlist_entry_init(&cmds->entry);
        } else {
                fr_dlist_remove(free_list, cmds);
        }
 
        fr_dlist_talloc_init(&cmds->pending, fr_redis_command_t, entry);
        fr_dlist_talloc_init(&cmds->sent, fr_redis_command_t, entry);
        fr_dlist_talloc_init(&cmds->completed, fr_redis_command_t, entry);
        cmds->request = request;
        cmds->complete = complete;
        cmds->fail = fail;
        cmds->rctx = rctx;
 
        if (ctx) talloc_link_ctx(ctx, cmds);
 
        return cmds;
}
 
/** Free any result associated with the command
 *
 * @param[in] cmd to free.  Frees any redis results associated with the command.
 */
static int _redis_command_free(fr_redis_command_t *cmd)
{
        //if (cmd->result) fr_redis_reply_free(&cmd->result);
 
        return 0;
}
 
redisReply *fr_redis_command_get_result(fr_redis_command_t *cmd)
{
        return cmd->result;
}
 
/** Add a preformatted/expanded command to the command set
 *
 * The command must either be entirely static, or parented by the command set.
 *
 * @note Caller should disallow "SUBSCRIBE" et al, if they're not appropriate.
 *       As subscribing to a stream where we're not expecting it would break
 *       things, badly.
 *
 * @param[in] cmds      Command set to add command to.
 * @param[in] cmd_str   A fully expanded/formatted command to send to redis.
 *                      Must be static, or have the same lifetime as the
 *                      command set (allocated with the command set as the parent).
 * @param[in] cmd_len   Length of the command.
 * @return
 *      - FR_REDIS_PIPELINE_BAD_CMDS if a bad command sequence is enqueued.
 *      - FR_REDIS_PIPELINE_OK if command was enqueued successfully.
 */
fr_redis_pipeline_status_t fr_redis_command_preformatted_add(fr_redis_command_set_t *cmds,
                                                             char const *cmd_str, size_t cmd_len)
{
        request_t                       *request = cmds->request;
        fr_redis_command_t      *cmd;
        fr_redis_command_type_t type = FR_REDIS_COMMAND_NORMAL;
 
        /*
         *      Transaction sanity checks.
         *
         *      Because commands from many different requests share the same connection
         *      we need to ensure that transaction blocks aren't left dangling and
         *      that the commands are all in the right order.
         *
         *      We try very hard to do this without incurring a performance penalty
         *      for non-transactional commands.
         */
        switch (tolower(cmd_str[0])) {
        case 'm':
                if (tolower(cmd_str[1] != 'u')) break;
                if (strncasecmp(cmd_str, "multi", sizeof("multi") - 1) != 0) break;
                /*
                 *      There should only ever be a difference of
                 *      1 between txn starts and txn ends.
                 */
                if ((cmds->txn_end < cmds->txn_start) && ((cmds->txn_start - cmds->txn_end) > 1)) {
                        ROPTIONAL(ERROR, REDEBUG, "Too many consecutive \"MULTI\" commands");
                        return FR_REDIS_PIPELINE_BAD_CMDS;
                }
                /*
                 *      If we have a watch before the MULTI,
                 *      that's marked as the start of the transaction
                 *      block.
                 */
                type = cmds->txn_watch ? FR_REDIS_COMMAND_TRANSACTION_START : FR_REDIS_COMMAND_NORMAL;
                cmds->txn_start++;      /* Yes MULTI increments start, not WATCH */
                break;
 
        case 'e':
                if (tolower(cmd_str[1] != 'e')) break;
                if (strncasecmp(cmd_str, "exec", sizeof("exec") - 1) != 0) break;
                goto txn_end;
 
        /*
         *      It's useful to allow discard as it allows command syntax checks
         *      to be performed against the REDIS server without actually
         *      executing the commands.
         */
        case 'd':
                if (tolower(cmd_str[1] != 'i')) break;
                if (strncasecmp(cmd_str, "discard", sizeof("discard") - 1) != 0) break;
        txn_end:
                if (cmds->txn_start <= cmds->txn_end) {
                        ROPTIONAL(ERROR, REDEBUG, "Transaction not started, missing \"MULTI\" command");
                        return FR_REDIS_PIPELINE_BAD_CMDS;
                }
                type = FR_REDIS_COMMAND_TRANSACTION_END;
                cmds->txn_end++;
                break;
 
        case 'w':
                if (tolower(cmd_str[1] != 'a')) break;
                if (strncasecmp(cmd_str, "watch", sizeof("watch") - 1) != 0) break;
                if (cmds->txn_watch) {
                        ROPTIONAL(ERROR, REDEBUG, "Too many consecutive \"WATCH\" commands");
                        return FR_REDIS_PIPELINE_BAD_CMDS;
                }
                if (cmds->txn_start > cmds->txn_end) {
                        ROPTIONAL(ERROR, REDEBUG, "\"WATCH\" can only be used before \"MULTI\"");
                        return FR_REDIS_PIPELINE_BAD_CMDS;
                }
                FALL_THROUGH;
 
        default:
                break;
        }
 
        MEM(cmd = talloc_zero(cmds, fr_redis_command_t));
        talloc_set_destructor(cmd, _redis_command_free);
        cmd->cmds = cmds;
        cmd->type = type;
        cmd->str = cmd_str;
        cmd->len = cmd_len;
        fr_dlist_insert_tail(&cmds->pending, cmd);
 
        return FR_REDIS_PIPELINE_OK;
}
 
/** Enqueue a command set on a specific trunk
 *
 * The command set may be passed around several trunks before it is complete.
 * This is to allow it to follow MOVED and ASK responses.
 *
 * @param[in] rtrunk    to enqueue command set on.
 * @param[in] cmds      Command set to enqueue.
 * @return
 *      - FR_REDIS_PIPELINE_OK if commands were immediately enqueued or placed in the backlog.
 *      - FR_REDIS_PIPELINE_DST_UNAVAILABLE if the REDIS host is unreachable.
 *      - FR_REDIS_PIPELINE_FAIL any other general error.
 */
fr_redis_pipeline_status_t redis_command_set_enqueue(fr_redis_trunk_t *rtrunk, fr_redis_command_set_t *cmds)
{
        if (cmds->txn_start != cmds->txn_end) {
                ERROR("Refusing to enqueue - Unbalanced transaction start/stop commands");
                return FR_REDIS_PIPELINE_BAD_CMDS;
        }
 
        switch (trunk_request_enqueue(&cmds->treq, rtrunk->trunk, cmds->request, cmds, cmds->rctx)) {
        case TRUNK_ENQUEUE_OK:
        case TRUNK_ENQUEUE_IN_BACKLOG:
                return FR_REDIS_PIPELINE_OK;
 
        case TRUNK_ENQUEUE_DST_UNAVAILABLE:
                return FR_REDIS_PIPELINE_DST_UNAVAILABLE;
 
        default:
                return FR_REDIS_PIPELINE_FAIL;
        }
}
 
/** Callback for for receiving Redis replies
 *
 * This is called by hiredis for each response is receives.  privData is set to the
 * fr_command_set
 *
 * @note Called only from hiredis, not the trunk itself.
 *
 * @param[in] ac                The async context the command was enqueued on.
 * @param[in] vreply            redisReply containing the result of the command.
 * @param[in] privdata          fr_redis_command_t that was sent to the Redis server.
 *                              The fr_redis_command_t contains a pointer to the
 *                              fr_redis_command_set_t which holds the treq which
 *                              we use to signal that we have responses for all
 *                              commands.
 */
static void _redis_pipeline_demux(struct redisAsyncContext *ac, void *vreply, void *privdata)
{
        fr_redis_command_t      *cmd;
        fr_redis_command_set_t  *cmds;
        connection_t            *conn = talloc_get_type_abort(ac->ev.data, connection_t);
        fr_redis_handle_t       *h = talloc_get_type_abort(conn->h, fr_redis_handle_t);
        redisReply              *reply = vreply;
        /*
         *      First check if we should ignore the response
         */
        if (!fr_redis_connection_process_response(h)) {
                DEBUG4("Ignoring response with SQN %"PRIu64, (h->rsp_sqn - 1)); /* Already incremented */
                fr_redis_reply_free((redisReply **)&reply);
                return;
        }
 
        /*
         *      FIXME - Need to check TRYAGAIN, MOVED etc...
         *      I guess we might want to wait for the end of
         *      the command set to do that.
         */
        cmd = talloc_get_type_abort(privdata, fr_redis_command_t);
        cmds = cmd->cmds;
        cmd->result = reply;
 
        fr_dlist_remove(&cmds->sent, cmd);
        fr_dlist_insert_tail(&cmds->completed, cmd);
 
        /*
         *      Check is the command set is complete,
         *      and if it is, tell the trunk the treq
         *      is complete.
         */
        if ((fr_dlist_num_elements(&cmds->pending) == 0) &&
            (fr_dlist_num_elements(&cmds->sent) == 0)) trunk_request_signal_complete(cmds->treq);
}
 
static connection_t *_redis_pipeline_connection_alloc(trunk_connection_t *tconn, fr_event_list_t *el,
                                                         connection_conf_t const *conf,
                                                         char const *log_prefix, void *uctx)
{
        fr_redis_trunk_t *rtrunk = talloc_get_type_abort(uctx, fr_redis_trunk_t);
 
        return fr_redis_connection_alloc(tconn, el, conf, rtrunk->io_conf, log_prefix);
}
 
/** Enqueue one or more command sets onto a redis handle
 *
 * Because the trunk is in always writable mode, _redis_pipeline_mux
 * will be called any time trunk_request_enqueue is called, so there'll only
 * ever be one command to dequeue.
 *
 * @param[in] tconn             Trunk connection holding the commands to enqueue.
 * @param[in] conn              Connection handle containing the fr_redis_handle_t.
 * @param[in] uctx              fr_redis_cluster_t.  Unused.
 */
static void _redis_pipeline_mux(trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
{
        trunk_request_t *treq;
        fr_redis_command_set_t  *cmds;
        fr_redis_command_t      *cmd;
        fr_redis_handle_t       *h = talloc_get_type_abort(conn->h, fr_redis_handle_t);
        request_t                       *request;
 
        treq = trunk_connection_pop_request(&request, (void *)&cmds, NULL, tconn);
        while ((cmd = fr_dlist_head(&cmds->pending))) {
                /*
                 *      If this fails it probably means the connection
                 *      is disconnecting, but if that's happening then
                 *      we shouldn't be enqueueing new requests?
                 */
                if (unlikely(redisAsyncCommand(h->ac, _redis_pipeline_demux, cmd, "%s", cmd->str) != REDIS_OK)) {
                        ROPTIONAL(ERROR, REDEBUG, "Unexpected error queueing REDIS command");
 
                        while ((cmd = fr_dlist_head(&cmds->sent))) {
                                fr_redis_connection_ignore_response(h, cmd->sqn);
                                fr_dlist_remove(&cmds->sent, cmd);
                                fr_dlist_insert_tail(&cmds->pending, cmd);
                        }
                        trunk_request_signal_fail(treq);
                        return;
                }
                cmd->sqn = fr_redis_connection_sent_request(h);
                fr_dlist_remove(&cmds->pending, cmd);
                fr_dlist_insert_tail(&cmds->sent, cmd);
        }
        trunk_request_signal_sent(treq);
}
 
/** Deal with cancellation of sent requests
 *
 * We can't actually signal redis to not process the request, so depending
 * on why the commands were cancelled, we either tell the handle to ignore
 * them, or move them back into the pending list.
 */
static void _redis_pipeline_command_set_cancel(connection_t *conn, UNUSED trunk_request_t *treq, void *preq,
                                               trunk_cancel_reason_t reason, UNUSED void *uctx)
{
        fr_redis_command_set_t  *cmds = talloc_get_type_abort(preq, fr_redis_command_set_t);
        fr_redis_handle_t       *h = conn->h;
 
        /*
         *      How we cancel is very different depending
         *      on _WHY_ we're cancelling.
         */
        switch (reason) {
        /*
         *      Cancel is only called for requests that
         *      have been sent, and only when the connection
         *      is about to be closed for some reason.
         *
         *      We don't need to tell the handle to ignore
         *      the responses, we just need to get the
         *      command set back into the correct state for
         *      execution by another handle.
         */
        case TRUNK_CANCEL_REASON_MOVE:
                fr_dlist_move(&cmds->pending, &cmds->sent);
                return;
 
        /*
         *      If the request was cancelled due to a signal
         *      we'll have a response coming back for a
         *      request, pctx and rctx that no longer exist.
         *      Tell the handle to signal that the response
         *      should be ignored when it's received.
         *
         *      Free will take care of cleaning up the
         *      pending commands.
         */
        case TRUNK_CANCEL_REASON_SIGNAL:
        {
                fr_redis_command_t      *cmd;
 
                for (cmd = fr_dlist_head(&cmds->sent);
                     cmd;
                     cmd = fr_dlist_next(&cmds->sent, cmd)) {
                        fr_redis_connection_ignore_response(h, cmd->sqn);
                }
        }
 
        case TRUNK_CANCEL_REASON_NONE:
                fr_assert(0);
                return;
        }
}
 
/** Signal the API client that we got a complete set of responses to a command set
 *
 */
static void _redis_pipeline_command_set_complete(UNUSED request_t *request, void *preq,
                                                 UNUSED void *rctx, UNUSED void *uctx)
{
        fr_redis_command_set_t  *cmds = talloc_get_type_abort(preq, fr_redis_command_set_t);
 
        if (cmds->complete) cmds->complete(cmds->request, &cmds->completed, cmds->rctx);
}
 
/** Signal the API client that we failed enqueuing the commands
 *
 */
static void _redis_pipeline_command_set_fail(UNUSED request_t *request, void *preq,
                                             UNUSED void *rctx, UNUSED void *uctx)
{
        fr_redis_command_set_t  *cmds = talloc_get_type_abort(preq, fr_redis_command_set_t);
 
        if (cmds->fail) cmds->fail(cmds->request, &cmds->completed, cmds->rctx);
}
 
/** Free the command set
 *
 */
static void _redis_pipeline_command_set_free(UNUSED request_t *request, void *preq,
                                             UNUSED void *uctx)
{
        fr_redis_command_set_t  *cmds = talloc_get_type_abort(preq, fr_redis_command_set_t);
 
        talloc_free(cmds);
}
 
/** Allocate a new trunk
 *
 * @param[in] cluster_thread    to allocate the trunk for.
 * @param[in] io_conf           Describing the connection to a single REDIS host.
 * @return
 *      - On success, a new fr_redis_trunk_t which can be used for pipelining commands.
 *      - NULL on failure.
 */
fr_redis_trunk_t *fr_redis_trunk_alloc(fr_redis_cluster_thread_t *cluster_thread, fr_redis_io_conf_t const *io_conf)
{
        fr_redis_trunk_t        *rtrunk;
        trunk_io_funcs_t        io_funcs = {
                                        .connection_alloc       = _redis_pipeline_connection_alloc,
                                        .request_mux            = _redis_pipeline_mux,
                                        /* demux called directly by hiredis */
                                        .request_cancel         = _redis_pipeline_command_set_cancel,
                                        .request_complete       = _redis_pipeline_command_set_complete,
                                        .request_fail           = _redis_pipeline_command_set_fail,
                                        .request_free           = _redis_pipeline_command_set_free
                                };
 
        MEM(rtrunk = talloc_zero(cluster_thread, fr_redis_trunk_t));
        rtrunk->io_conf = io_conf;
        rtrunk->trunk = trunk_alloc(rtrunk, cluster_thread->el,
                                       &io_funcs, cluster_thread->tconf, cluster_thread->log_prefix, rtrunk,
                                       cluster_thread->delay_start);
        if (!rtrunk->trunk) {
                talloc_free(rtrunk);
                return NULL;
        }
 
        return rtrunk;
}
 
/** Allocate per-thread, per-cluster instance
 *
 * This structure represents all the connections for a given thread for a given cluster.
 * The structures holds the trunk connections to talk to each cluster member.
 *
 */
fr_redis_cluster_thread_t *fr_redis_cluster_thread_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, trunk_conf_t const *tconf)
{
        fr_redis_cluster_thread_t *cluster_thread;
        trunk_conf_t *our_tconf;
 
        MEM(cluster_thread = talloc_zero(ctx, fr_redis_cluster_thread_t));
        MEM(our_tconf = talloc_memdup(cluster_thread, tconf, sizeof(*tconf)));
        our_tconf->always_writable = true;
 
        cluster_thread->el = el;
        cluster_thread->tconf = our_tconf;
 
        return cluster_thread;
}