io.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
 */
 
/**
 * $Id: 49f07b0c05c976ccc22ce1a5bbb740ae2a22dddd $
 * @file lib/redis/io.c
 * @brief Common functions for interacting with Redis via hiredis
 *
 * @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/redis/io.h>
#include <freeradius-devel/util/debug.h>
 
 
/** Called by hiredis to indicate the connection is dead
 *
 */
static void _redis_disconnected(redisAsyncContext const *ac, UNUSED int status)
{
        connection_t            *conn = talloc_get_type_abort(ac->data, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        /*
         *      redisAsyncFree was called with a live
         *      connection, but inside the talloc
         *      destructor of the fr_redis_handle_t.
         *
         *      Don't signal the connection state
         *      machine that it needs reconnecting,
         *      the connection is being destroyed.
         */
        if (h->ignore_disconnect_cb) return;
 
        DEBUG4("Signalled by hiredis, connection disconnected");
 
        connection_signal_reconnect(conn, CONNECTION_FAILED);
}
 
/** Called by hiredis to indicate the connection is live
 *
 */
static void _redis_connected(redisAsyncContext const *ac, UNUSED int status)
{
        connection_t            *conn = talloc_get_type_abort(ac->data, connection_t);
 
        DEBUG4("Signalled by hiredis, connection is open");
 
        connection_signal_connected(conn);
}
 
/** Redis FD became readable
 *
 */
static void _redis_io_service_readable(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
{
        connection_t const      *conn = talloc_get_type_abort_const(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - FD %i now readable", h, fd);
 
        redisAsyncHandleRead(h->ac);
}
 
/** Redis FD became writable
 *
 */
static void _redis_io_service_writable(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
{
        connection_t const      *conn = talloc_get_type_abort_const(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - FD %i now writable", h, fd);
 
        redisAsyncHandleWrite(h->ac);
}
 
/** Redis FD errored - Automatically removes registered events
 *
 */
static void _redis_io_service_errored(UNUSED fr_event_list_t *el, int fd, UNUSED int flags,
                                      int fd_errno, void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - FD %i errored: %s", h, fd, fr_syserror(fd_errno));
 
        /*
         *      Connection state machine will handle reconnecting
         */
        connection_signal_reconnect(conn, CONNECTION_FAILED);
}
 
/** Deal with the method hiredis uses to register/unregister interest in a file descriptor
 *
 */
static void _redis_io_common(connection_t *conn, fr_redis_handle_t *h, bool read, bool write)
{
        redisContext            *c = &(h->ac->c);
        fr_event_list_t         *el = conn->el;
 
        /*
         *      hiredis doesn't even attempt to dedup registration
         *      requests *sigh*...
         */
        if ((h->read_set == read) && (h->write_set == write)) return;
 
        if (!read && !write) {
                DEBUG4("redis handle %p - De-registering FD %i", h, c->fd);
 
                if (fr_event_fd_delete(el, c->fd, FR_EVENT_FILTER_IO) < 0) {
                        PERROR("redis handle %p - De-registration failed for FD %i", h, c->fd);
                }
                return;
        }
 
        DEBUG4("redis handle %p - Registered for %s%serror events on FD %i",
               h, read ? "read+" : "", write ? "write+" : "", c->fd);
 
        if (fr_event_fd_insert(h, NULL, el, c->fd,
                               read ? _redis_io_service_readable : NULL,
                               write ? _redis_io_service_writable : NULL,
                               _redis_io_service_errored,
                               conn) < 0) {
                PERROR("redis handle %p - Registration failed for %s%serror events on FD %i",
                       h, read ? "read+" : "", write ? "write+" : "", c->fd);
                return;
        }
 
        h->read_set = read;
        h->write_set = write;
}
 
/** Register FD for reads
 *
 */
static void _redis_io_add_read(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, true, h->write_set);
}
 
/** De-register FD for reads
 *
 */
static void _redis_io_del_read(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, false, h->write_set);
}
 
/** Register FD for writes
 *
 */
static void _redis_io_add_write(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, h->read_set, true);
}
 
/** De-register FD for writes
 *
 */
static void _redis_io_del_write(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        _redis_io_common(conn, h, h->read_set, false);
}
 
#ifdef HAVE_REDIS_TIMEOUT
/** Connection timer expired
 *
 */
static void _redis_io_service_timer_expired(UNUSED fr_event_list_t *el, UNUSED fr_time_t now, void *uctx)
{
        connection_t const      *conn = talloc_get_type_abort_const(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - Timeout", h);
 
        redisAsyncHandleTimeout(h->ac);
}
 
/** Modify the connection I/O timer
 *
 */
static void _redis_io_timer_modify(void *uctx, struct timeval tv)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
        fr_time_delta_t         timeout;
 
        timeout = fr_time_delta_from_timeval(&tv);
 
        DEBUG4("redis handle %p - Timeout in %pV seconds", h, fr_box_time_delta(timeout));
 
        if (fr_timer_in(h, conn->el, &h->timer,
                              timeout, _redis_io_service_timer_expired, conn) < 0) {
                PERROR("redis timeout %p - Failed adding timeout", h);
        }
}
#endif
 
/** Handle freeing the redisAsyncContext
 *
 * delRead and delWrite don't seem to be called when the redisAsyncContext is freed
 *
 * As the IO events must be removed from the event loop *before* the FD is closed
 * and as the IO events will only be automatically de-registered when when the
 * fr_redis_handle_t is freed.
 *
 * Unfortunately the destructor for the fr_redis_handle_t will be run before
 * the IO events are de-registered, which'll free the redisAsycCtx, which'll close
 * the FD.
 *
 * This means there'd be a brief period of time between the FD is closed, and
 * it being removed from the event loop.
 *
 * We use the cleanup callback (which is called before the FD is closed) to remove
 * the events now, and ensure there's no chance of issues.
 */
static void _redis_io_free(void *uctx)
{
        connection_t            *conn = talloc_get_type_abort(uctx, connection_t);
        fr_redis_handle_t       *h = conn->h;
 
        DEBUG4("redis handle %p - Freed", h);
 
        _redis_io_common(conn, h, false, false);
}
 
/** Configures async I/O callbacks for an existing redisAsyncContext
 *
 */
static int fr_redis_io_setup(redisAsyncContext *ac, connection_t const *conn)
{
        if (ac->ev.data != NULL) return REDIS_ERR;
 
        ac->ev.addRead = _redis_io_add_read;
        ac->ev.delRead = _redis_io_del_read;
        ac->ev.addWrite = _redis_io_add_write;
        ac->ev.delWrite = _redis_io_del_write;
#ifdef HAVE_REDIS_TIMEOUT
        ac->ev.scheduleTimer = _redis_io_timer_modify;
#endif
        ac->ev.cleanup = _redis_io_free;
        memcpy(&ac->ev.data, &conn, sizeof(ac->ev.data));
 
        return REDIS_OK;
}
 
/** Free the redis async context when the handle is freed
 *
 */
static int _redis_handle_free(fr_redis_handle_t *h)
{
        /*
         *      Don't fire the reconnect callback if we're
         *      freeing the handle.
         */
        h->ignore_disconnect_cb = true;
        if (h->ac) redisAsyncFree(h->ac);
 
        return 0;
}
 
/** Callback for the initialise state
 *
 * Should attempt to open a non-blocking connection and return it in h_out.
 *
 * @param[out] h_out    Where to write the new handle
 * @param[in] conn      This connection.  Opaque, should only be used for
 *                      signalling the connection state machine.
 * @param[in] uctx      User context.
 * @return
 *      - #CONNECTION_STATE_CONNECTING  if a file descriptor was successfully created.
 *      - #CONNECTION_STATE_FAILED              if we could not open a valid handle.
 */
CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
static connection_state_t _redis_io_connection_init(void **h_out, connection_t *conn, void *uctx)
{
        fr_redis_io_conf_t      *conf = uctx;
        char const              *host = conf->hostname;
        uint16_t                port = conf->port;
        fr_redis_handle_t       *h;
        int                     ret;
 
        /*
         *      Allocate a structure to wrap the
         *      redis async context.
         */
        MEM(h = talloc_zero(conn, fr_redis_handle_t));
        talloc_set_destructor(h, _redis_handle_free);
 
        h->ac = redisAsyncConnect(host, port);
        if (!h->ac) {
                ERROR("Failed allocating handle for %s:%u", host, port);
                return CONNECTION_STATE_FAILED;
        }
 
        if (h->ac->err) {
                ERROR("Failed allocating handle for %s:%u: %s", host, port, h->ac->errstr);
        error:
                redisAsyncFree(h->ac);
                return CONNECTION_STATE_FAILED;
        }
 
        /*
         *      Store the connection in private data,
         *      so we can use it for signalling.
         *
         *      Comments in the redis src indicate
         *      it doesn't mess with this.
         */
        memcpy(&h->ac->data, &conn, sizeof(h->ac->data));
 
        /*
         *      Handle has to be associated with the
         *      conn in case I/O handlers want to get
         *      at it.
         */
        *h_out = h;
 
        /*
         *      Install the I/O service functions
         *
         *      Event library must be set first
         *      before calling redisAsyncSetConnectCallback
         *      as just setting the callback triggers
         *      I/O events to be registered.
         */
        fr_redis_io_setup(h->ac, conn);
 
        /*
         *      Setup callbacks so we're notified
         *      when the connection state changes.
         *
         *      We then signal the connection state
         *      machine, to let it handle
         *      reconnecting.
         */
        ret = redisAsyncSetConnectCallback(h->ac, _redis_connected);
        if (ret != REDIS_OK) {
                ERROR("Failed setting connected callback: Error %i", ret);
                goto error;
        }
 
        ret = redisAsyncSetDisconnectCallback(h->ac, _redis_disconnected);
        if (ret != REDIS_OK) {
                ERROR("Failed setting disconnected callback: Error %i", ret);
                goto error;
        }
 
        fr_dlist_talloc_init(&h->ignore, fr_redis_sqn_ignore_t, entry);
 
        return CONNECTION_STATE_CONNECTING;
}
 
/** Gracefully signal that the connection should shutdown
 *
 */
static connection_state_t _redis_io_connection_shutdown(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
{
        fr_redis_handle_t       *our_h = talloc_get_type_abort(h, fr_redis_handle_t);
 
        redisAsyncDisconnect(our_h->ac);        /* Should not free the handle */
 
        return CONNECTION_STATE_SHUTDOWN;
}
 
/** Notification that the connection has errored and must be closed
 *
 * This should be used to close the file descriptor.  It is assumed
 * that the file descriptor is invalid after this callback has been executed.
 *
 * If this callback does not close the file descriptor, the server will leak
 * file descriptors.
 *
 * @param[in] el        to remove event handlers from.
 * @param[in] h         to close.
 * @param[in] uctx      User context.
 */
static void _redis_io_connection_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
{
        fr_redis_handle_t       *our_h = talloc_get_type_abort(h, fr_redis_handle_t);
 
        /*
         *      The destructor will free the redisAsyncContext
         *      which'll close the connection, after removing
         *      it from the event loop.
         */
        talloc_free(our_h);
}
 
/** Allocate an async redis I/O connection
 *
 */
connection_t *fr_redis_connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el,
                                           connection_conf_t const *conn_conf, fr_redis_io_conf_t const *io_conf,
                                           char const *log_prefix)
{
        connection_t *conn;
        /*
         *      We don't specify an open callback
         *      as hiredis handles switching over
         *      all the I/O handlers internally
         *      within hireds, and calls us when
         *      the connection is open.
         */
        conn = connection_alloc(ctx, el,
                                   &(connection_funcs_t){
                                        .init = _redis_io_connection_init,
                                        .close = _redis_io_connection_close,
                                        .shutdown = _redis_io_connection_shutdown
                                   },
                                   conn_conf,
                                   log_prefix,
                                   io_conf);
        if (!conn) return NULL;
 
        return conn;
}
 
/** Return the redisAsyncContext associated with the connection
 *
 * This is needed to issue commands to the redis server.
 *
 * @param[in] conn      To retrieve async ctx from.
 * @return The async ctx.
 */
redisAsyncContext *fr_redis_connection_get_async_ctx(connection_t *conn)
{
        fr_redis_handle_t       *h = conn->h;
        return h->ac;
}