udp_queue.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: 2c31f0cda126ffa7b0a19de62eb0b712c6f6743f $
 *
 * @file src/lib/server/udp_queue.c
 * @brief Handle queues of outgoing UDP packets
 *
 * @author Alan DeKok (aland@freeradius.org)
 *
 * @copyright 2021 Network RADIUS SAS (legal@networkradius.com)
 */
RCSID("$Id: 2c31f0cda126ffa7b0a19de62eb0b712c6f6743f $")
 
#include <freeradius-devel/util/debug.h>
#include <freeradius-devel/util/syserror.h>
 
#include <freeradius-devel/util/socket.h>
#include <freeradius-devel/util/udp_queue.h>
 
struct fr_udp_queue_s {
        fr_udp_queue_config_t const *config;            //!< configuration
        fr_dlist_head_t         queue;                  //!< list of queued packets to write, ordered by time
 
        fr_event_list_t         *el;
        int                     fd;
        int                     port;
        bool                    blocked;                //!< are we blocked?
 
        fr_udp_queue_resume_t   resume;
};
 
typedef struct {
        struct sockaddr_storage sockaddr;
        socklen_t       socklen;
 
        fr_udp_queue_t  *uq;
        fr_dlist_t      dlist;
 
        void            *rctx;
 
        fr_time_t       expires;
 
        size_t          packet_len;
        uint8_t         packet[];
} fr_udp_queue_entry_t;
 
static int _udp_queue_free(fr_udp_queue_t *uq)
{
        fr_dlist_foreach_safe(&uq->queue, fr_udp_queue_entry_t, entry) {
                talloc_free(entry);
        }}
 
        close(uq->fd);
 
        return 0;
}
 
static int _udp_queue_entry_free(fr_udp_queue_entry_t *entry)
{
        fr_udp_queue_t *uq = entry->uq;
        void *rctx = entry->rctx;
 
        fr_dlist_remove(&uq->queue, entry);
 
        if (uq->resume) uq->resume(false, rctx);
 
        return 0;
}
 
/** Allocate an outbound UDP queue.
 *
 * @param ctx   where the structure will be allocated.
 * @param config containing the IPs, ports, etc
 * @param el    the event list for adding events to see if the socket is writable
 * @param resume the function to call after a delayed packet has been written
 * @return
 *      - NULL on error
 *      - !NULL on success
 */
fr_udp_queue_t *fr_udp_queue_alloc(TALLOC_CTX *ctx, fr_udp_queue_config_t const *config, fr_event_list_t *el,
                                   fr_udp_queue_resume_t resume)
{
        fr_udp_queue_t  *uq;
        int             fd;
        fr_ipaddr_t     ipaddr = config->ipaddr;
        uint16_t        port = config->port;
 
        /*
         *      Open the socket.
         */
        fd = fr_socket_server_udp(&config->ipaddr, &port, NULL, false);
        if (fd < 0) return NULL;
 
        /*
         *      Set SO_REUSEPORT if we're binding to a specific port
         *      (e.g. DHCP), so that multiple threads can use the same
         *      port.
         */
        if (config->port != 0) {
                int on = 1;
 
                if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &on, sizeof(on)) < 0) {
                        fr_strerror_printf("SO_REUSEPORT said %s", fr_syserror(errno));
                        goto error;
                }
        }
 
        /*
         *      Bind to the given interface.
         */
        if (config->interface &&
            (fr_socket_bind(fd, config->interface, &ipaddr, &port) < 0)) goto error;
 
#ifdef SO_SNDBUF
        /*
         *      Set SO_SNDBUF size, if configured to do so.
         */
        if (config->send_buff_is_set) {
                int opt;
 
                opt = config->send_buff;
 
                if (opt < 65536) opt = 65536;
                if (opt > (1 << 30)) opt = 1<<30;
 
                (void) setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(int));
        }
#endif
 
#ifdef SO_RCVBUF
        /*
         *      Set SO_RECVBUFF to 4K, so that the kernel will quickly
         *      drop incoming packets.  We don't expect replies, and
         *      we never check the socket for readability, so this is
         *      fine.
         */
        {
                int opt = 4096;
 
                (void) setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(int));
        }
#endif
 
        uq = talloc_zero(ctx, fr_udp_queue_t);
        if (!uq) {
        error:
                close(fd);
                return NULL;
        }
 
        *uq = (fr_udp_queue_t) {
                .config = config,
                .el = el,
                .fd = fd,
                .port = port,
                .resume = resume,
        };
 
        fr_dlist_init(&uq->queue, fr_udp_queue_entry_t, dlist);
 
        talloc_set_destructor(uq, _udp_queue_free);
 
        return uq;
}
 
/** If the socket is writable, then flush packets until either it
 * returns EWOULDBLOCK, or there are no more packets to write.
 *
 */
static void udp_queue_writable(UNUSED fr_event_list_t *el, UNUSED int fd,
                               UNUSED int flags, void *uctx)
{
        fr_udp_queue_t  *uq = talloc_get_type_abort(uctx, fr_udp_queue_t);
        fr_time_t       now = fr_time();
 
        fr_dlist_foreach_safe(&uq->queue, fr_udp_queue_entry_t, entry) {
                ssize_t rcode;
                int retries = 0;
 
                /*
                 *      If the entry is expired, tell the caller that
                 *      it wasn't written to the socket.
                 */
                if (fr_time_gteq(now, entry->expires)) {
                        void *rctx = entry->rctx;
 
                        talloc_free(entry);
                        if (uq->resume) uq->resume(false, rctx);
                        continue;
                }
 
        retry:
                rcode = sendto(uq->fd, entry->packet, entry->packet_len, 0, (struct sockaddr *) &entry->sockaddr, entry->socklen);
                if (rcode >= 0) {
                        void *rctx = entry->rctx;
 
                        talloc_free(entry);
                        if (uq->resume) uq->resume(true, rctx);
                        continue;
                }
 
                if (rcode < 0) {
                        if (errno == EINTR) {
                                if (retries++ < 3) goto retry;
                                return;
                        }
 
#if EWOULDBLOCK != EAGAIN
                        if (!((errno == EWOULDBLOCK) || (errno == EAGAIN))) return;
#else
                        if (errno != EWOULDBLOCK) return;
#endif
                }
        }}
 
        /*
         *      Nothing more to write, delete the IO handler so that we don't get extraneous signals.
         */
        if (fr_dlist_num_elements(&uq->queue) == 0) {
                fr_event_fd_delete(uq->el, uq->fd, FR_EVENT_FILTER_IO);
                uq->blocked = false;
        }
}
 
/** Write packet to socket, OR enqueue it if we get EAGAIN
 *
 *  In most cases, the packet will get written to the socket immediately.
 *
 *  However, if the socket is blocked, then the packet is added to an
 *  outbound queue.  When the socket becomes unblocked, the packets
 *  will be sent.
 *
 * @param ctx   the talloc context for this packet to be saved in, usually request_t
 * @param uq    the local queue to write it to
 * @param packet the packet to write
 * @param packet_len how long the packet is
 * @param ipaddr the IP address we're sending the packet to
 * @param port   the port we're sending the packet to
 * @param rctx   for resumption, usually request_t, or a structure which holds a request_t
 * @return
 *      - <0 for error
 *      - 0 for "didn't write it to socket, but added it to the queue, and the caller should yield"
 *      - 1 for "wrote it to the socket, you're good to go".
 */
int fr_udp_queue_write(TALLOC_CTX *ctx, fr_udp_queue_t *uq, uint8_t const *packet, size_t packet_len,
                       fr_ipaddr_t const *ipaddr, int port, void *rctx)
{
        struct sockaddr_storage sockaddr;
        socklen_t               socklen;
        fr_udp_queue_entry_t    *entry;
 
        fr_ipaddr_to_sockaddr(&sockaddr, &socklen, ipaddr, port);
 
        if (!packet_len || !port) return 1;
 
        if (!uq->blocked) {
                int retries = 0;
                ssize_t rcode;
 
retry:
                rcode = sendto(uq->fd, packet, packet_len, 0, (struct sockaddr *) &sockaddr, socklen);
                if (rcode >= 0) return 1;
 
                if (rcode < 0) {
                        if (errno == EINTR) {
                                if (retries++ < 3) goto retry;
                                return -1;
                        }
 
#if EWOULDBLOCK != EAGAIN
                        if (!((errno == EWOULDBLOCK) || (errno == EAGAIN))) return -1;
#else
                        if (errno != EWOULDBLOCK) return -1;
#endif
                }
 
                /*
                 */
                if (fr_event_fd_insert(uq, NULL, uq->el, uq->fd, NULL,
                                       udp_queue_writable, NULL, uq) < 0) {
                        return -1;
                }
 
                uq->blocked = true;
        }
 
        /*
         *      Limit the number of packets in the queue.
         */
        if (uq->config->max_queued_packets &&
            (fr_dlist_num_elements(&uq->queue) >= uq->config->max_queued_packets)) {
                return -1;
        }
 
        entry = (fr_udp_queue_entry_t *) talloc_zero_array(ctx, uint8_t, sizeof(fr_udp_queue_entry_t) + packet_len);
        if (!entry) return -1;
 
        talloc_set_type(entry, fr_udp_queue_entry_t);
        talloc_set_destructor(entry, _udp_queue_entry_free);
 
        *entry = (fr_udp_queue_entry_t) {
                .sockaddr = sockaddr,
                .socklen = socklen,
                .uq = uq,
                .expires = fr_time_add(fr_time(), uq->config->max_queued_time),
                .rctx = rctx,
                .packet_len = packet_len,
        };
 
        memcpy(entry->packet, packet, packet_len);
        fr_dlist_insert_tail(&uq->queue, entry);
 
        /*
         *      Didn't do anything, say so.
         */
 
        return 0;
}