network.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: 980bbf74e52f0f1fc0d2c96c2d9fdeaa38feb37c $
 *
 * @brief Receiver of socket data, which sends messages to the workers.
 * @file io/network.c
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSID("$Id: 980bbf74e52f0f1fc0d2c96c2d9fdeaa38feb37c $")
 
#define LOG_PREFIX nr->name
 
#define LOG_DST nr->log
 
#include <freeradius-devel/util/event.h>
#include <freeradius-devel/util/rand.h>
#include <freeradius-devel/util/rb.h>
#include <freeradius-devel/util/syserror.h>
#include <freeradius-devel/util/atexit.h>
 
#include <freeradius-devel/io/channel.h>
#include <freeradius-devel/io/listen.h>
#include <freeradius-devel/io/network.h>
#include <freeradius-devel/io/queue.h>
 
#define MAX_WORKERS 64
 
static _Thread_local fr_ring_buffer_t *fr_network_rb;
 
typedef struct {
        fr_listen_t             *listen;
        uint8_t                 *packet;
        size_t                  packet_len;
        fr_time_t               recv_time;
} fr_network_inject_t;
 
/** Associate a worker thread with a network thread
 *
 */
typedef struct {
        fr_heap_index_t         heap_id;                //!< workers are in a heap
        fr_time_delta_t         cpu_time;               //!< how much CPU time this worker has spent
        fr_time_delta_t         predicted;              //!< predicted processing time for one packet
 
        bool                    blocked;                //!< is this worker blocked?
 
        fr_channel_t            *channel;               //!< channel to the worker
        fr_worker_t             *worker;                //!< worker pointer
        fr_io_stats_t           stats;
} fr_network_worker_t;
 
typedef struct {
        fr_rb_node_t            listen_node;            //!< rbtree node for looking up by listener.
        fr_rb_node_t            num_node;               //!< rbtree node for looking up by number.
 
        fr_network_t            *nr;                    //!< O(N) issues in talloc
        int                     number;                 //!< unique ID
        fr_heap_index_t         heap_id;                //!< for the sockets_by_num heap
 
        fr_event_filter_t       filter;                 //!< what type of filter it is
        fr_event_fd_t           *ef;                    //!< the I/O event, if it was inserted.  Parented off
                                                        ///< this socket, so it cannot outlive it.
 
        bool                    dead;                   //!< is it dead?
        bool                    blocked;                //!< is it blocked?
        bool                    closed;                 //!< the descriptor has been closed.
 
        unsigned int            outstanding;            //!< number of outstanding packets sent to the worker
        fr_listen_t             *listen;                //!< I/O ctx and functions.
 
        fr_message_set_t        *ms;                    //!< message buffers for this socket.
        fr_channel_data_t       *cd;                    //!< cached in case of allocation & read error
        size_t                  leftover;               //!< leftover data from a previous read
        size_t                  written;                //!< however much we did in a partial write
 
        fr_channel_data_t       *pending;               //!< the currently pending partial packet
        fr_heap_t               *waiting;               //!< packets waiting to be written
        fr_io_stats_t           stats;
} fr_network_socket_t;
 
/*
 *      We have an array of workers, so we can index the workers in
 *      O(1) time.  remove the heap of "workers ordered by CPU time"
 *      when we send a packet to a worker, just update the predicted
 *      CPU time in place.  when we receive a reply from a worker,
 *      just update the predicted CPU time in place.
 *
 *      when we need to choose a worker, pick 2 at random, and then
 *      choose the one with the lowe cpu time.  For background, see
 *      "Power of Two-Choices" and
 *      https://www.eecs.harvard.edu/~michaelm/postscripts/mythesis.pdf
 *      https://www.eecs.harvard.edu/~michaelm/postscripts/tpds2001.pdf
 */
struct fr_network_s {
        char const              *name;                  //!< Network ID for logging.
 
        pthread_t               thread_id;              //!< for self
 
        bool                    suspended;              //!< whether or not we're suspended.
 
        fr_log_t const          *log;                   //!< log destination
        fr_log_lvl_t            lvl;                    //!< debug log level
 
        fr_atomic_queue_t       *aq_control;            //!< atomic queue for control messages sent to me
 
        fr_control_t            *control;               //!< the control plane
 
        fr_ring_buffer_t        *rb;                    //!< ring buffer for my control-plane messages
 
        fr_event_list_t         *el;                    //!< our event list
 
        fr_heap_t               *replies;               //!< replies from the worker, ordered by priority / origin time
 
        fr_io_stats_t           stats;
 
        fr_rb_tree_t            *sockets;               //!< list of sockets we're managing, ordered by the listener
        fr_rb_tree_t            *sockets_by_num;        //!< ordered by number;
 
        int                     num_workers;            //!< number of active workers
        int                     num_blocked;            //!< number of blocked workers
        int                     num_pending_workers;    //!< number of workers we're waiting to start.
        int                     max_workers;            //!< maximum number of allowed workers
        int                     num_sockets;            //!< actually a counter...
 
        int                     signal_pipe[2];         //!< Pipe for signalling the worker in an orderly way.
                                                        ///< This is more deterministic than using async signals.
 
        bool                    exiting;                //!< are we exiting?
 
        fr_network_config_t     config;                 //!< configuration
        fr_network_worker_t     *workers[MAX_WORKERS];  //!< each worker
};
 
static void fr_network_post_event(fr_event_list_t *el, fr_time_t now, void *uctx);
static int fr_network_pre_event(fr_time_t now, fr_time_delta_t wake, void *uctx);
static void fr_network_socket_dead(fr_network_t *nr, fr_network_socket_t *s);
static int network_socket_close(fr_network_socket_t *s);
static void fr_network_read(UNUSED fr_event_list_t *el, int sockfd, UNUSED int flags, void *ctx);
 
static int8_t reply_cmp(void const *one, void const *two)
{
        fr_channel_data_t const *a = one, *b = two;
        int ret;
 
        ret = CMP(a->priority, b->priority);
        if (ret != 0) return ret;
 
        return fr_time_cmp(a->m.when, b->m.when);
}
 
static int8_t waiting_cmp(void const *one, void const *two)
{
        fr_channel_data_t const *a = one, *b = two;
        int ret;
 
        ret = CMP(a->priority, b->priority);
        if (ret != 0) return ret;
 
        return fr_time_cmp(a->reply.request_time, b->reply.request_time);
}
 
static int8_t socket_listen_cmp(void const *one, void const *two)
{
        fr_network_socket_t const *a = one, *b = two;
 
        return CMP(a->listen, b->listen);
}
 
static int8_t socket_num_cmp(void const *one, void const *two)
{
        fr_network_socket_t const *a = one, *b = two;
 
        return CMP(a->number, b->number);
}
 
/*
 *      Explicitly cleanup the memory allocated to the ring buffer,
 *      just in case valgrind complains about it.
 */
static int _fr_network_rb_free(void *arg)
{
        return talloc_free(arg);
}
 
/** Initialise thread local storage
 *
 * @return fr_ring_buffer_t for messages
 */
static inline fr_ring_buffer_t *fr_network_rb_init(void)
{
        fr_ring_buffer_t *rb;
 
        rb = fr_network_rb;
        if (rb) return rb;
 
        rb = fr_ring_buffer_create(NULL, FR_CONTROL_MAX_MESSAGES * FR_CONTROL_MAX_SIZE);
        if (!rb) {
                fr_perror("Failed allocating memory for network ring buffer");
                return NULL;
        }
 
        fr_atexit_thread_local(fr_network_rb, _fr_network_rb_free, rb);
 
        return rb;
}
 
static inline bool is_network_thread(fr_network_t const *nr)
{
        return (pthread_equal(pthread_self(), nr->thread_id) != 0);
}
 
static int fr_network_listen_add_self(fr_network_t *nr, fr_listen_t *listen);
 
/** Add a fr_listen_t to a network
 *
 * @param nr            the network
 * @param li            the listener
 */
int fr_network_listen_add(fr_network_t *nr, fr_listen_t *li)
{
        fr_ring_buffer_t *rb;
 
        /*
         *      Associate the protocol dictionary with the listener, so that the decode functions can check /
         *      use it.
         *
         *      A virtual server may start off with a "dictionary" block, and therefore define a local
         *      dictionary.  So the "root" dictionary of a virtual server may not be a protocol dict.
         */
        fr_assert(li->server_cs != NULL);
        li->dict = fr_dict_proto_dict(virtual_server_dict_by_cs(li->server_cs));
 
        fr_assert(li->dict != NULL);
 
        /*
         *      Skip a bunch of work if we're already in the network thread.
         */
        if (is_network_thread(nr) && !li->needs_full_setup) {
                return fr_network_listen_add_self(nr, li);
        }
 
        rb = fr_network_rb_init();
        if (!rb) return -1;
 
        return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_LISTEN, &li, sizeof(li));
}
 
 
/** Delete a socket from a network.  MUST be called only by the listener itself!.
 *
 * @param nr            the network
 * @param li            the listener
 */
int fr_network_listen_delete(fr_network_t *nr, fr_listen_t *li)
{
        fr_network_socket_t *s;
 
        fr_assert(is_network_thread(nr));
 
        s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
        if (!s) return -1;
 
        fr_network_socket_dead(nr, s);
 
        return 0;
}
 
/** Add a "watch directory" call to a network
 *
 * @param nr            the network
 * @param li            the listener
 */
int fr_network_directory_add(fr_network_t *nr, fr_listen_t *li)
{
        fr_ring_buffer_t *rb;
 
        rb = fr_network_rb_init();
        if (!rb) return -1;
 
        return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_DIRECTORY, &li, sizeof(li));
}
 
/** Add a worker to a network in a different thread
 *
 * @param nr the network
 * @param worker the worker
 */
int fr_network_worker_add(fr_network_t *nr, fr_worker_t *worker)
{
        fr_ring_buffer_t *rb;
 
        rb = fr_network_rb_init();
        if (!rb) return -1;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
        (void) talloc_get_type_abort(worker, fr_worker_t);
 
        return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_WORKER, &worker, sizeof(worker));
}
 
static void fr_network_worker_started_callback(void *ctx, void const *data, size_t data_size, fr_time_t now);
 
/** Add a worker to a network in the same thread
 *
 * @param nr the network
 * @param worker the worker
 */
void fr_network_worker_add_self(fr_network_t *nr, fr_worker_t *worker)
{
        fr_network_worker_started_callback(nr, &worker, sizeof(worker), fr_time_wrap(0));
}
 
 
/** Signal the network to read from a listener
 *
 * @param nr the network
 * @param li the listener to read from
 */
void fr_network_listen_read(fr_network_t *nr, fr_listen_t *li)
{
        fr_network_socket_t *s;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
        (void) talloc_get_type_abort_const(li, fr_listen_t);
 
        s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
        if (!s) return;
 
        /*
         *      Go read the socket.
         */
        fr_network_read(nr->el, s->listen->fd, 0, s);
}
 
 
/** Inject a packet for a listener to write
 *
 * @param nr            the network
 * @param li            the listener where the packet is being injected
 * @param packet        the packet to be written
 * @param packet_len    the length of the packet
 * @param packet_ctx    The packet context to write
 * @param request_time  when the packet was received.
 */
void fr_network_listen_write(fr_network_t *nr, fr_listen_t *li, uint8_t const *packet, size_t packet_len,
                             void *packet_ctx, fr_time_t request_time)
{
        fr_message_t *lm;
        fr_channel_data_t cd;
 
        cd = (fr_channel_data_t) {
                .m = (fr_message_t) {
                        .status = FR_MESSAGE_USED,
                        .data_size = packet_len,
                        .when = request_time,
                },
 
                .channel = {
                        .heap_id = FR_HEAP_INDEX_INVALID,
                },
 
                .listen = li,
                .priority = PRIORITY_NOW,
                .reply.request_time = request_time,
        };
 
        memcpy(&cd.m.data, &packet, sizeof(packet)); /* const issues */
        memcpy(&cd.packet_ctx, &packet_ctx, sizeof(packet_ctx)); /* const issues */
 
        /*
         *      Localize the message and insert it into the heap of pending messages.
         */
        lm = fr_message_localize(nr, &cd.m, sizeof(cd));
        if (!lm) return;
 
        if (fr_heap_insert(&nr->replies, lm) < 0) {
                fr_message_done(lm);
        }
}
 
 
/** Inject a packet for a listener to read
 *
 * @param nr            the network
 * @param li            the listener where the packet is being injected
 * @param packet        the packet to be injected
 * @param packet_len    the length of the packet
 * @param recv_time     when the packet was received.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_network_listen_inject(fr_network_t *nr, fr_listen_t *li, uint8_t const *packet, size_t packet_len, fr_time_t recv_time)
{
        int rcode;
        fr_ring_buffer_t *rb;
        fr_network_inject_t my_inject;
 
        /*
         *      Can't inject to injection-less destinations.
         */
        if (!li->app_io->inject) {
                fr_strerror_const("Listener cannot accept injected packet");
                return -1;
        }
 
        /*
         *      Avoid a bounce through the event loop if we're being called from the network thread.
         */
        if (is_network_thread(nr)) {
                fr_network_socket_t *s;
 
                s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
                if (!s) {
                        fr_strerror_const("Listener was not found for injected packet");
                        return -1;
                }
 
                /*
                 *      Inject the packet.  The master.c mod_read() routine will then take care of avoiding
                 *      IO, and instead return the packet to the network side.
                 */
                if (li->app_io->inject(li, packet, packet_len, recv_time) == 0) {
                        (void) fr_network_read(nr->el, li->fd, 0, s);
                }
 
                return 0;
        }
 
        rb = fr_network_rb_init();
        if (!rb) return -1;
 
        my_inject.listen = li;
        MEM(my_inject.packet = talloc_memdup(NULL, packet, packet_len));
        my_inject.packet_len = packet_len;
        my_inject.recv_time = recv_time;
 
        rcode = fr_control_message_send(nr->control, rb, FR_CONTROL_ID_INJECT, &my_inject, sizeof(my_inject));
        if (rcode < 0) talloc_free(my_inject.packet);
 
        return rcode;
}
 
static fr_event_update_t const pause_read[] = {
        FR_EVENT_SUSPEND(fr_event_io_func_t, read),
        { 0 }
};
 
static fr_event_update_t const resume_read[] = {
        FR_EVENT_RESUME(fr_event_io_func_t, read),
        { 0 }
};
 
static fr_event_update_t const pause_write[] = {
        FR_EVENT_SUSPEND(fr_event_io_func_t, write),
        { 0 }
};
 
static fr_event_update_t const resume_write[] = {
        FR_EVENT_RESUME(fr_event_io_func_t, write),
        { 0 }
};
 
static void fr_network_suspend(fr_network_t *nr)
{
        fr_rb_iter_inorder_t    iter;
        fr_network_socket_t     *s;
 
        if (nr->suspended) return;
 
        for (s = fr_rb_iter_init_inorder(nr->sockets, &iter);
             s != NULL;
             s = fr_rb_iter_next_inorder(nr->sockets, &iter)) {
                fr_event_filter_update(s->nr->el, s->listen->fd, FR_EVENT_FILTER_IO, pause_read);
        }
        nr->suspended = true;
}
 
static void fr_network_unsuspend(fr_network_t *nr)
{
        fr_rb_iter_inorder_t    iter;
        fr_network_socket_t     *s;
 
        if (!nr->suspended) return;
 
        for (s = fr_rb_iter_init_inorder(nr->sockets, &iter);
             s != NULL;
             s = fr_rb_iter_next_inorder(nr->sockets, &iter)) {
                fr_event_filter_update(s->nr->el, s->listen->fd, FR_EVENT_FILTER_IO, resume_read);
        }
        nr->suspended = false;
}
 
#define IALPHA (8)
#define RTT(_old, _new) fr_time_delta_wrap((fr_time_delta_unwrap(_new) + (fr_time_delta_unwrap(_old) * (IALPHA - 1))) / IALPHA)
 
/** Callback which handles a message being received on the network side.
 *
 * @param[in] ctx the network
 * @param[in] ch the channel that the message is on.
 * @param[in] cd the message (if any) to start with
 */
static void fr_network_recv_reply(void *ctx, fr_channel_t *ch, fr_channel_data_t *cd)
{
        fr_network_t *nr = ctx;
        fr_network_worker_t *worker;
 
        cd->channel.ch = ch;
 
        /*
         *      Update stats for the worker.
         */
        worker = fr_channel_requestor_uctx_get(ch);
        worker->stats.out++;
        worker->cpu_time = cd->reply.cpu_time;
        if (!fr_time_delta_ispos(worker->predicted)) {
                worker->predicted = cd->reply.processing_time;
        } else {
                worker->predicted = RTT(worker->predicted, cd->reply.processing_time);
        }
 
        /*
         *      Unblock the worker.
         */
        if (worker->blocked) {
                worker->blocked = false;
                nr->num_blocked--;
                fr_network_unsuspend(nr);
        }
 
        /*
         *      Ensure that heap insert works.
         */
        cd->channel.heap_id = FR_HEAP_INDEX_INVALID;
        if (fr_heap_insert(&nr->replies, cd) < 0) {
                fr_message_done(&cd->m);
                fr_assert(0 == 1);
        }
}
 
/** Handle a network control message callback for a channel
 *
 * This is called from the event loop when we get a notification
 * from the event signalling pipe.
 *
 * @param[in] ctx       the network
 * @param[in] data      the message
 * @param[in] data_size size of the data
 * @param[in] now       the current time
 */
static void fr_network_channel_callback(void *ctx, void const *data, size_t data_size, fr_time_t now)
{
        fr_channel_event_t      ce;
        fr_channel_t            *ch;
        fr_network_t            *nr = ctx;
 
        ce = fr_channel_service_message(now, &ch, data, data_size);
        DEBUG3("Channel %s",
               fr_table_str_by_value(channel_signals, ce, "<INVALID>"));
        switch (ce) {
        case FR_CHANNEL_ERROR:
                return;
 
        case FR_CHANNEL_EMPTY:
                return;
 
        case FR_CHANNEL_NOOP:
                break;
 
        case FR_CHANNEL_DATA_READY_REQUESTOR:
                fr_assert(ch != NULL);
                while (fr_channel_recv_reply(ch));
                break;
 
        case FR_CHANNEL_DATA_READY_RESPONDER:
                fr_assert(0 == 1);
                break;
 
        case FR_CHANNEL_OPEN:
                fr_assert(0 == 1);
                break;
 
        case FR_CHANNEL_CLOSE:
        {
                fr_network_worker_t     *w = talloc_get_type_abort(fr_channel_requestor_uctx_get(ch),
                                                                   fr_network_worker_t);
                int                     i;
 
                /*
                 *      Remove this worker from the array
                 */
                DEBUG3("Worker acked our close request");
                for (i = 0; i < nr->num_workers; i++) {
                        if (nr->workers[i] == w) {
                                if (i == (nr->num_workers - 1)) break;
 
                                /*
                                 *      Close the hole...
                                 */
                                memmove(&nr->workers[i], &nr->workers[i + 1],
                                        (uint8_t *) &nr->workers[nr->num_workers] - (uint8_t *) &nr->workers[i + 1]);
                                break;
                        }
                }
                nr->num_workers--;
                nr->workers[nr->num_workers] = NULL; /* over-write now unused pointer */
        }
                break;
        }
}
 
#define OUTSTANDING(_x) ((_x)->stats.in - (_x)->stats.out)
 
/** Send a message on the "best" channel.
 *
 * @param nr the network
 * @param cd the message we've received
 */
static int fr_network_send_request(fr_network_t *nr, fr_channel_data_t *cd)
{
        fr_network_worker_t *worker;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
 
        /*
         *      The workers have been signalled to close and are tearing their
         *      channels down.  Anything queued now is stranded: the worker
         *      discards it without a reply, so our outstanding count for the
         *      socket never comes back down and the socket can never be freed.
         *
         *      The listeners are closed before the close is signalled, so a
         *      socket read cannot get here.  Assert, so that whatever did shows
         *      itself with a backtrace, and drop the packet in release builds.
         */
        if (!fr_cond_assert_msg(!nr->exiting,
                                "Sending packet to worker after signalling the channel close")) return -1;
 
        if (!nr->num_workers) {
                RATE_LIMIT_GLOBAL(ERROR, "Failed sending packet to worker - "
                                  "No workers are available");
                return -1;
        }
 
retry:
        if (nr->num_workers == 1) {
                worker = nr->workers[0];
                if (worker->blocked) {
                        RATE_LIMIT_GLOBAL(ERROR, "Failed sending packet to worker - "
                                          "In single-threaded mode and worker is blocked");
        drop:
                        worker->stats.dropped++;
                        return -1;
                }
 
        } else if (nr->num_blocked == 0) {
                int64_t cmp;
                uint32_t one, two;
 
                one = fr_rand() % nr->num_workers;
                do {
                        two = fr_rand() % nr->num_workers;
                } while (two == one);
 
                /*
                 *      Choose a worker based on minimizing the amount
                 *      of future work it's being asked to do.
                 *
                 *      If both workers have the same number of
                 *      outstanding requests, then choose the worker
                 *      which has used the least total CPU time.
                 */
                cmp = (OUTSTANDING(nr->workers[one]) - OUTSTANDING(nr->workers[two]));
                if (cmp < 0) {
                        worker = nr->workers[one];
 
                } else if (cmp > 0) {
                        worker = nr->workers[two];
 
                } else if (fr_time_delta_lt(nr->workers[one]->cpu_time, nr->workers[two]->cpu_time)) {
                        worker = nr->workers[one];
 
                } else {
                        worker = nr->workers[two];
                }
        } else {
                int i;
                uint64_t min_outstanding = UINT64_MAX;
                fr_network_worker_t *found = NULL;
 
                /*
                 *      Some workers are blocked.  Pick the worker
                 *      with the least amount of future work to do.
                 */
                for (i = 0; i < nr->num_workers; i++) {
                        uint64_t outstanding;
 
                        worker = nr->workers[i];
                        if (worker->blocked) continue;
 
                        outstanding = OUTSTANDING(worker);
                        if ((outstanding < min_outstanding) || !found) {
                                found = worker;
                                min_outstanding = outstanding;
 
                        } else if (outstanding == min_outstanding) {
                                /*
                                 *      Queue lengths are the same.
                                 *      Choose this worker if it's
                                 *      less busy than the previous one we found.
                                 */
                                if (fr_time_delta_lt(worker->cpu_time, found->cpu_time)) {
                                        found = worker;
                                }
                        }
                }
 
                if (!found) {
                         RATE_LIMIT_GLOBAL(PERROR, "Failed sending packet to worker - Couldn't find active worker, "
                                           "%u/%u workers are blocked", nr->num_blocked, nr->num_workers);
                         return -1;
                }
 
                worker = found;
        }
 
        (void) talloc_get_type_abort(worker, fr_network_worker_t);
 
        /*
         *      Too many outstanding packets for this worker.  Drop
         *      the request.
         *
         *      If the worker we've picked has too many outstanding
         *      packets, then we have either only one worker, in which
         *      cae we should drop the packet.  Or, we were unable to
         *      find a worker with smaller than max_outstanding
         *      packets.  In which case all of the workers are likely
         *      at max_outstanding.
         *
         *      In both cases, we should just drop the new packet.
         */
        fr_assert(worker->stats.in >= worker->stats.out);
        if (nr->config.max_outstanding &&
            (OUTSTANDING(worker) >= nr->config.max_outstanding)) {
                RATE_LIMIT_GLOBAL(PERROR, "max_outstanding reached - dropping packet");
                goto drop;
        }
 
        /*
         *      Send the message to the channel.  If we fail, drop the
         *      packet.  The only reason for failure is that the
         *      worker isn't servicing it's input queue.  When that
         *      happens, we have no idea what to do, and the whole
         *      thing falls over.
         */
        if (fr_channel_send_request(worker->channel, cd) < 0) {
                worker->stats.dropped++;
                worker->blocked = true;
                nr->num_blocked++;
 
                RATE_LIMIT_GLOBAL(PERROR, "Failed sending packet to worker - %u/%u workers are blocked",
                                  nr->num_blocked, nr->num_workers);
 
                if (nr->num_blocked == nr->num_workers) {
                        fr_network_suspend(nr);
                        return -1;
                }
                goto retry;
        }
 
        worker->stats.in++;
 
        /*
         *      We're projecting that the worker will use more CPU
         *      time to process this request.  The CPU time will be
         *      updated with a more accurate number when we receive a
         *      reply from this channel.
         */
        worker->cpu_time = fr_time_delta_add(worker->cpu_time, worker->predicted);
 
        return 0;
}
 
 
/** Send a packet to the worker.
 *
 *  MUST only be called from the network thread.
 *
 * @param nr the network
 * @param parent the parent listener
 * @param li the listener that the packet was "read" from.  Can be "parent"
 * @param buffer the packet to send
 * @param buflen size of the packet to send
 * @param recv_time of the packet
 * @param packet_ctx for the packet
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_network_listen_send_packet(fr_network_t *nr, fr_listen_t *parent, fr_listen_t *li,
                                  const uint8_t *buffer, size_t buflen, fr_time_t recv_time, void *packet_ctx)
{
        fr_channel_data_t *cd;
        fr_network_socket_t *s;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
        (void) talloc_get_type_abort_const(li, fr_listen_t);
 
        s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
        if (!s) return -1;
 
        cd = (fr_channel_data_t *) fr_message_and_data_alloc(s->ms, buflen);
        if (!cd) return -1;
 
        cd->listen = parent;
        cd->priority = PRIORITY_NORMAL;
        cd->packet_ctx = packet_ctx;
        cd->request.recv_time = recv_time;
        memcpy(cd->m.data, buffer, buflen);
        cd->m.when = fr_time();
 
        if (fr_network_send_request(nr, cd) < 0) {
                talloc_free(cd->packet_ctx);
                fr_message_done(&cd->m);
                nr->stats.dropped++;
                s->stats.dropped++;
                return -1;
        }
 
        s->outstanding++;
        return 0;
}
 
/** Get the number of outstanding packets
 *
 * @param nr the network
 * @param li the listener that the packet was "read" from
 * @return
 *      - <0 on error
 *      - the number of outstanding packets
*/
ssize_t fr_network_listen_outstanding(fr_network_t *nr, fr_listen_t *li) {
        fr_network_socket_t *s;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
        (void) talloc_get_type_abort_const(li, fr_listen_t);
 
        s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
        if (!s) return -1;
 
        return s->outstanding;
}
 
/*
 *      Mark it as dead, but DON'T free it until all of the replies
 *      have come in.
 */
static void fr_network_socket_dead(fr_network_t *nr, fr_network_socket_t *s)
{
        int i;
 
        if (s->dead) return;
 
        s->dead = true;
 
        /*
         *      Nothing writes to a dead socket - fr_network_post_event()
         *      discards its replies - so the descriptor goes now.  The memory
         *      has to wait for the workers to finish with its messages.
         */
        if (network_socket_close(s) < 0) PWARN("Failed removing event for socket %d", s->number);
 
        for (i = 0; i < nr->max_workers; i++) {
                if (!nr->workers[i]) continue;
 
                (void) fr_worker_listen_cancel(nr->workers[i]->worker, s->listen);
        }
 
        /*
         *      If there are no outstanding packets, then we can free
         *      it now.
         */
        if (!s->outstanding) {
                talloc_free(s);
                return;
        }
 
        /*
         *      There are still outstanding packets.  Leave it in the
         *      socket tree, so that replies from the worker can find
         *      it.  When we've received all of the replies, then
         *      fr_network_post_event() will clean up this socket.
         */
}
 
/** Read a packet from the network.
 *
 * @param[in] el        the event list.
 * @param[in] sockfd    the socket which is ready to read.
 * @param[in] flags     from kevent.
 * @param[in] ctx       the network socket context.
 */
static void fr_network_read(UNUSED fr_event_list_t *el, int sockfd, UNUSED int flags, void *ctx)
{
        int                     num_messages = 0;
        fr_network_socket_t     *s = ctx;
        fr_network_t            *nr = s->nr;
        ssize_t                 data_size;
        fr_channel_data_t       *cd, *next;
 
        if (!fr_cond_assert_msg(s->listen->fd == sockfd, "Expected listen->fd (%u) to be equal event fd (%u)",
                                s->listen->fd, sockfd)) return;
 
        DEBUG3("Reading data from FD %u", sockfd);
 
        if (!s->cd) {
                cd = (fr_channel_data_t *) fr_message_and_data_reserve(s->ms, s->listen->default_message_size);
                if (!cd) {
                        ERROR("Failed allocating message size %zd! - Closing socket",
                              s->listen->default_message_size);
                        fr_network_socket_dead(nr, s);
                        return;
                }
        } else {
                cd = s->cd;
        }
 
        fr_assert(cd->m.data != NULL);
 
next_message:
        /*
         *      Poll this socket, but not too often.  We have to go
         *      service other sockets, too.
         */
        if (num_messages > 16) {
                s->cd = cd;
                return;
        }
 
        cd->priority = PRIORITY_NORMAL;
 
        /*
         *      Read data from the network.
         *
         *      Return of 0 means "no data", which is fine for UDP.
         *      For TCP, if an underlying read() on the TCP socket
         *      returns 0, (which signals that the FD is no longer
         *      usable) this function should return -1, so that the
         *      network side knows that it needs to close the
         *      connection.
         */
        data_size = s->listen->app_io->read(s->listen, &cd->packet_ctx, &cd->request.recv_time,
                                            cd->m.data, cd->m.rb_size, &s->leftover);
        if (data_size == 0) {
                /*
                 *      Cache the message only when there are leftover
                 *      bytes from a partial stream read.  The buffer
                 *      must be handed back on the next call so the
                 *      stream can append to it.
                 *
                 *      When app_io->read() calls fr_network_listen_send_packet()
                 *      internally, that function calls fr_message_and_data_alloc()
                 *      on the same message set.  Because the message ring uses
                 *      fr_ring_buffer_reserve() (which does not advance write_offset),
                 *      the new alloc returns the same ring slot as our reservation.
                 *      The memset inside message_reserve zeroes our struct, then the
                 *      new message commits into that slot.  We can detect this
                 *      because cd->m.data_size is non-zero after the alloc commits.
                 *
                 *      In the aliased case, cd now refers to the already-committed
                 *      message that was dispatched to a worker.  We must not reset
                 *      or modify it; just clear s->cd so the next call reserves fresh.
                 *
                 *      When neither leftover nor aliasing applies, the reservation
                 *      is clean and unused; reset it explicitly so its ring-buffer
                 *      slot is immediately reusable.
                 */
                if (s->leftover) {
                        s->cd = cd;
                } else if (cd->m.data_size != 0) {
                        s->cd = NULL;
                } else {
                        fr_message_and_data_reset(s->ms, &cd->m);
                        s->cd = NULL;
                }
                return;
        }
 
        /*
         *      Error: close the connection, and remove the fr_listen_t
         */
        if (data_size < 0) {
//              fr_log(nr->log, L_DBG_ERR, "error from transport read on socket %d", sockfd);
                fr_network_socket_dead(nr, s);
                return;
        }
        s->cd = NULL;
 
        DEBUG3("Read %zd byte(s) from FD %u", data_size, sockfd);
        if (s->listen->read_hexdump) HEXDUMP2(cd->m.data, data_size, "%s read ", s->listen->name);
        nr->stats.in++;
        s->stats.in++;
 
        /*
         *      Initialize the rest of the fields of the channel data.
         *
         *      We always use "now" as the time of the message, as the
         *      packet MAY be a duplicate packet magically resurrected
         *      from the past.  i.e. If the read routines are doing
         *      dedup, then they notice that the packet is a
         *      duplicate.  In that case, they send over a copy of the
         *      packet, BUT with the original timestamp.  This
         *      information tells the worker that the packet is a
         *      duplicate.
         */
        cd->m.when = fr_time();
        cd->listen = s->listen;
 
        /*
         *      Nothing in the buffer yet.  Allocate room for one
         *      packet.
         */
        if ((cd->m.data_size == 0) && (!s->leftover)) {
                (void) fr_message_and_data_commit(s->ms, &cd->m, data_size);
                next = NULL;
        } else {
                /*
                 *      There are leftover bytes in the buffer, feed
                 *      them to the next round of reading.
                 */
                if (s->leftover) {
                        next = (fr_channel_data_t *) fr_message_and_data_commit_with_leftover(s->ms, &cd->m, data_size, s->leftover,
                                                                              s->listen->default_message_size);
                        if (!next) {
                                PERROR("Failed reserving partial packet.");
                                // @todo - probably close the socket...
                                fr_assert(0 == 1);
                        }
                } else {
                        (void) fr_message_and_data_commit(s->ms, &cd->m, data_size);
                        next = NULL;
                }
        }
 
        /*
         *      Set the priority.  Which incidentally also checks if
         *      we're allowed to read this particular kind of packet.
         *
         *      That check is because the app_io handlers just read
         *      packets, and don't really have access to the parent
         *      "list of allowed packet types".  So we have to do the
         *      work here in a callback.
         *
         *      That should probably be fixed...
         */
        if (s->listen->app->priority) {
                int priority;
 
                priority = s->listen->app->priority(s->listen->app_instance, cd->m.data, data_size);
                if (priority <= 0) goto discard;
 
                cd->priority = priority;
        }
 
        if (fr_network_send_request(nr, cd) < 0) {
        discard:
                talloc_free(cd->packet_ctx); /* not sure what else to do here */
                fr_message_done(&cd->m);
                nr->stats.dropped++;
                s->stats.dropped++;
 
        } else {
                /*
                 *      One more packet sent to a worker.
                 */
                s->outstanding++;
        }
 
        /*
         *      If there is a next message, go read it from the buffer.
         *
         *      @todo - note that this calls read(), even if the
         *      app_io has paused the reader.  We likely want to be
         *      able to check that, too.  We might just remove this
         *      "goto"...
         */
        if (next) {
                cd = next;
                num_messages++;
                goto next_message;
        }
}
 
int fr_network_sendto_worker(fr_network_t *nr, fr_listen_t *li, void *packet_ctx, uint8_t const *data, size_t data_len, fr_time_t recv_time)
{
        fr_channel_data_t *cd;
        fr_network_socket_t *s;
 
        s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
        if (!s) return -1;
 
        cd = (fr_channel_data_t *) fr_message_and_data_alloc(s->ms, data_len);
        if (!cd) return -1;
 
        s->stats.in++;
 
        cd->priority = PRIORITY_NORMAL;
 
        cd->m.when = recv_time;
        cd->listen = li;
        cd->packet_ctx = packet_ctx;
 
        memcpy(cd->m.data, data, data_len);
 
        if (fr_network_send_request(nr, cd) < 0) {
                talloc_free(packet_ctx);
                fr_message_done(&cd->m);
                nr->stats.dropped++;
                s->stats.dropped++;
                return -1;
        }
 
        /*
         *      One more packet sent to a worker.
         */
        s->outstanding++;
        return 0;
}
 
 
/** Get a notification that a vnode changed
 *
 * @param[in] el        the event list.
 * @param[in] sockfd    the socket which is ready to read.
 * @param[in] fflags    from kevent.
 * @param[in] ctx       the network socket context.
 */
static void fr_network_vnode_extend(UNUSED fr_event_list_t *el, int sockfd, int fflags, void *ctx)
{
        fr_network_socket_t *s = ctx;
        fr_network_t *nr = s->nr;
 
        if (!fr_cond_assert(s->listen->fd == sockfd)) return;
 
        DEBUG3("network vnode");
 
        /*
         *      Tell the IO handler that something has happened to the
         *      file.
         */
        s->listen->app_io->vnode(s->listen, fflags);
}
 
 
/** Handle errors for a socket.
 *
 * @param[in] el                the event list
 * @param[in] sockfd            the socket which has a fatal error.
 * @param[in] flags             returned by kevent.
 * @param[in] fd_errno          returned by kevent.
 * @param[in] ctx               the network socket context.
 */
static void fr_network_error(UNUSED fr_event_list_t *el, UNUSED int sockfd, int flags,
                             int fd_errno, void *ctx)
{
        fr_network_socket_t *s = ctx;
        fr_network_t *nr = s->nr;
 
        if (s->listen->app_io->error) {
                s->listen->app_io->error(s->listen);
 
        } else if (flags & EV_EOF) {
                DEBUG2("Socket %s closed by peer", s->listen->name);
 
        } else {
                ERROR("Socket %s errored - %s", s->listen->name, fr_syserror(fd_errno));
        }
 
        fr_network_socket_dead(s->nr, s);
}
 
 
/** Write packets to the network.
 *
 * @param el the event list
 * @param sockfd the socket which is ready to write
 * @param flags returned by kevent.
 * @param ctx the network socket context.
 */
static void fr_network_write(UNUSED fr_event_list_t *el, UNUSED int sockfd, UNUSED int flags, void *ctx)
{
        fr_network_socket_t *s = ctx;
        fr_listen_t *li = s->listen;
        fr_network_t *nr = s->nr;
        fr_channel_data_t *cd;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
 
        /*
         *      Start with the currently pending message, and then
         *      work through the priority heap.
         */
        if (s->pending) {
                cd = s->pending;
                s->pending = NULL;
 
        } else {
                cd = fr_heap_pop(&s->waiting);
        }
 
        while (cd != NULL) {
                int rcode;
 
                fr_assert(li == cd->listen);
                if (li->write_hexdump) HEXDUMP2(cd->m.data, cd->m.data_size, "%s writing ", li->name);
                rcode = li->app_io->write(li, cd->packet_ctx,
                                          cd->reply.request_time,
                                          cd->m.data, cd->m.data_size, s->written);
 
                /*
                 *      Write of 0 bytes means an OS bug, and we just discard this packet.
                 */
                if (rcode == 0) {
                        RATE_LIMIT_GLOBAL(ERROR, "Discarding packet due to write returning zero for socket %s",
                                          s->listen->name);
                        goto discard;
                }
 
                /*
                 *      Or we have a write error.
                 */
                if (rcode < 0) {
                        /*
                         *      Stop processing the heap, and set the
                         *      pending message to the current one.
                         */
                        if (errno == EWOULDBLOCK) {
                        save_pending:
                                fr_assert(!s->pending);
 
                                if (cd->m.status != FR_MESSAGE_LOCALIZED) {
                                        fr_message_t *lm;
 
                                        lm = fr_message_localize(s, &cd->m, sizeof(*cd));
                                        if (!lm) {
                                                ERROR("Failed saving pending packet");
                                                goto dead;
                                        }
 
                                        cd = (fr_channel_data_t *) lm;
                                }
 
                                if (!s->blocked) {
                                        if (fr_event_filter_update(nr->el, s->listen->fd, FR_EVENT_FILTER_IO, resume_write) < 0) {
                                                PERROR("Failed adding write callback to event loop");
                                                goto dead;
                                        }
 
                                        s->blocked = true;
                                }
 
                                s->pending = cd;
                                return;
                        }
 
                        PERROR("Failed writing to socket %s", s->listen->name);
 
                        switch (errno) {
                        /*
                         *      As a special hack, check for something
                         *      that will never be returned from a
                         *      real write() routine.  Which then
                         *      signals to us that we have to close
                         *      the socket, but NOT complain about it.
                         */
                        case ECONNREFUSED:
                        case ECONNRESET:
                                break;
 
                                /*
                                 *      These are temporary errors.  We just discard the data.
                                 */
                        case ENETDOWN:
                        case ENETUNREACH:
                                if (li->app_io->error) li->app_io->error(li);
 
                                fr_message_done(&cd->m);
                                return;
 
                        default:
                                if (li->app_io->error) li->app_io->error(li);
                                break;
                        }
 
                dead:
                        fr_message_done(&cd->m);
                        fr_network_socket_dead(nr, s);
                        return;
                }
 
                /*
                 *      If we've done a partial write, localize the message and continue.
                 */
                if ((size_t) rcode < cd->m.data_size) {
                        s->written = rcode;
                        goto save_pending;
                }
 
        discard:
                s->written = 0;
 
                /*
                 *      Reset for the next message.
                 */
                fr_message_done(&cd->m);
                nr->stats.out++;
                s->stats.out++;
 
                /*
                 *      Grab the net entry.
                 */
                cd = fr_heap_pop(&s->waiting);
        }
 
        /*
         *      We've successfully written all of the packets.  Remove
         *      the write callback.
         */
        if (fr_event_filter_update(nr->el, s->listen->fd, FR_EVENT_FILTER_IO, pause_write) < 0) {
                PERROR("Failed removing write callback from event loop");
                fr_network_socket_dead(nr, s);
        }
 
        s->blocked = false;
}
 
/** Stop a socket's I/O events and close its descriptor
 *
 * Separate from freeing the socket because freeing it also frees s->ms, and
 * the workers hold messages allocated from that until they ack the channel
 * close.  The descriptor can go as soon as we are done with it; the memory
 * cannot.
 *
 * Idempotent, so it is safe to call on a socket that is already dead.
 *
 * @param[in] s to close.
 * @return
 *      - 0 on success.
 *      - -1 if the I/O event could not be removed.  The descriptor is closed
 *        either way, so this is worth reporting but not worth stopping for.
 */
static int network_socket_close(fr_network_socket_t *s)
{
        int ret = 0;
 
        if (s->closed) return 0;
        s->closed = true;
 
        /*
         *      NULL if the socket never made it into the event loop, which the
         *      setup error paths rely on.  This is the only place the event is
         *      removed, so the handle cannot have gone stale under us.
         */
        if (s->ef) {
                ret = fr_event_fd_delete_handle(s->ef);
                s->ef = NULL;
        }
 
        if (s->listen->app_io->close) {
                s->listen->app_io->close(s->listen);
        } else {
                close(s->listen->fd);
        }
 
        return ret;
}
 
static int _network_socket_free(fr_network_socket_t *s)
{
        fr_network_t *nr = s->nr;
        fr_channel_data_t *cd;
 
        /*
         *      Closing is a separate step, so that the descriptor can be
         *      released while the message set stays alive for the workers.
         */
        fr_assert_msg(s->closed, "socket %d freed without being closed", s->number);
        fr_assert(s->outstanding == 0);
 
        fr_rb_delete(nr->sockets, s);
        fr_rb_delete(nr->sockets_by_num, s);
 
        if (s->pending) {
                fr_message_done(&s->pending->m);
                s->pending = NULL;
        }
 
        /*
         *      Clean up any queued entries.
         */
        while ((cd = fr_heap_pop(&s->waiting)) != NULL) {
                fr_message_done(&cd->m);
        }
 
        /* s->waiting is already talloc parented from s */
        talloc_free(s->listen);
 
        return 0;
}
 
 
/** Handle a network control message callback for a new listener
 *
 * @param[in] ctx the network
 * @param[in] data the message
 * @param[in] data_size size of the data
 * @param[in] now the current time
 */
static void fr_network_listen_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
{
        fr_network_t            *nr = talloc_get_type_abort(ctx, fr_network_t);
        fr_listen_t             *li;
 
        fr_assert(data_size == sizeof(li));
 
        if (data_size != sizeof(li)) return;
 
        li = talloc_get_type_abort(*((void * const *)data), fr_listen_t);
 
        (void) fr_network_listen_add_self(nr, li);
}
 
static void fr_network_limit_ringbuffer(fr_network_socket_t *s, int *num_messages_p, size_t *size_p)
{
        int num_messages;
        size_t size;
 
        num_messages = s->listen->num_messages;
        if (num_messages < 8) num_messages = 8;
        if (num_messages > (1 << 20)) num_messages = (1 << 20);
 
        size = s->listen->default_message_size * num_messages;
        if (size < (1 << 17)) size = (1 << 17);
        if (size > (100 * 1024 * 1024)) size = (100 * 1024 * 1024);
 
        *num_messages_p = num_messages;
        *size_p = size;
}
 
static int fr_network_listen_add_self(fr_network_t *nr, fr_listen_t *li)
{
        fr_network_socket_t     *s;
        fr_app_io_t const       *app_io;
        size_t                  size;
        int                     num_messages;
 
        fr_assert(li->app_io != NULL);
 
        /*
         *      Non-socket listeners just get told about the event
         *      list, and nothing else.
         */
        if (li->non_socket_listener) {
                fr_assert(li->app_io->event_list_set != NULL);
                fr_assert(!li->app_io->read);
                fr_assert(!li->app_io->write);
 
                li->app_io->event_list_set(li, nr->el, nr);
 
                /*
                 *      We use fr_log() here to avoid the "Network - " prefix.
                 */
                fr_log(nr->log, L_DBG, __FILE__, __LINE__, "Listener %s bound to virtual server %s",
                       li->name, cf_section_name2(li->server_cs));
 
                return 0;
        }
 
        s = talloc_zero(nr, fr_network_socket_t);
        fr_assert(s != NULL);
        talloc_steal(s, li);
 
        s->nr = nr;
        s->listen = li;
        s->number = nr->num_sockets++;
 
        MEM(s->waiting = fr_heap_alloc(s, waiting_cmp, fr_channel_data_t, channel.heap_id, 0));
 
        talloc_set_destructor(s, _network_socket_free);
 
        /*
         *      Put reasonable limits on the ring buffer size.  Then
         *      round it up to the nearest power of 2, which is
         *      required by the ring buffer code.
         */
        fr_network_limit_ringbuffer(s, &num_messages, &size);
 
        /*
         *      Allocate the ring buffer for messages and packets.
         */
        s->ms = fr_message_set_create(s, num_messages,
                                      sizeof(fr_channel_data_t),
                                      size, false);
        if (!s->ms) {
                PERROR("Failed creating message buffers for network IO");
                if (network_socket_close(s) < 0) PWARN("Failed removing event for socket %d", s->number);
                talloc_free(s);
                return -1;
        }
 
        app_io = s->listen->app_io;
        s->filter = FR_EVENT_FILTER_IO;
 
        if (fr_event_fd_insert(s, &s->ef, nr->el, s->listen->fd,
                               fr_network_read,
                               s->listen->no_write_callback ? NULL : fr_network_write,
                               fr_network_error,
                               s) < 0) {
                PERROR("Failed adding new socket to network event loop");
                if (network_socket_close(s) < 0) PWARN("Failed removing event for socket %d", s->number);
                talloc_free(s);
                return -1;
        }
 
        /*
         *      Start of with write updates being paused.  We don't
         *      care about being able to write if there's nothing to
         *      write.
         */
        (void) fr_event_filter_update(nr->el, s->listen->fd, FR_EVENT_FILTER_IO, pause_write);
 
        /*
         *      Add the listener before calling the app_io, so that
         *      the app_io can find the listener which we're adding
         *      here.
         */
        (void) fr_rb_insert(nr->sockets, s);
        (void) fr_rb_insert(nr->sockets_by_num, s);
 
        if (app_io->event_list_set) app_io->event_list_set(s->listen, nr->el, nr);
 
        /*
         *      We use fr_log() here to avoid the "Network - " prefix.
         */
        fr_log(nr->log, L_DBG, __FILE__, __LINE__, "Listening on %s bound to virtual server %s",
              s->listen->name, cf_section_name2(s->listen->server_cs));
 
        DEBUG3("Using new socket %s with FD %d", s->listen->name, s->listen->fd);
 
        return 0;
}
 
/** Handle a network control message callback for a new "watch directory"
 *
 * @param[in] ctx the network
 * @param[in] data the message
 * @param[in] data_size size of the data
 * @param[in] now the current time
 */
static void fr_network_directory_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
{
        int                     num_messages;
        size_t                  size;
        fr_network_t            *nr = talloc_get_type_abort(ctx, fr_network_t);
        fr_listen_t             *li = talloc_get_type_abort(*((void * const *)data), fr_listen_t);
        fr_network_socket_t     *s;
        fr_app_io_t const       *app_io;
        fr_event_vnode_func_t   funcs = { .extend = fr_network_vnode_extend };
 
        if (!fr_cond_assert(data_size == sizeof(li))) return;
 
        memcpy(&li, data, sizeof(li));
 
        s = talloc_zero(nr, fr_network_socket_t);
        fr_assert(s != NULL);
        talloc_steal(s, li);
 
        s->nr = nr;
        s->listen = li;
        s->number = nr->num_sockets++;
 
        MEM(s->waiting = fr_heap_alloc(s, waiting_cmp, fr_channel_data_t, channel.heap_id, 0));
 
        talloc_set_destructor(s, _network_socket_free);
 
        /*
         *      Allocate the ring buffer for messages and packets.
         */
        fr_network_limit_ringbuffer(s, &num_messages, &size);
 
        s->ms = fr_message_set_create(s, num_messages,
                                      sizeof(fr_channel_data_t),
                                      size, false);
        if (!s->ms) {
                PERROR("Failed creating message buffers for directory IO");
                if (network_socket_close(s) < 0) PWARN("Failed removing event for socket %d", s->number);
                talloc_free(s);
                return;
        }
 
        app_io = s->listen->app_io;
 
        if (app_io->event_list_set) app_io->event_list_set(s->listen, nr->el, nr);
 
        s->filter = FR_EVENT_FILTER_VNODE;
 
        if (fr_event_filter_insert(s, &s->ef, nr->el, s->listen->fd, s->filter,
                                   &funcs,
                                   app_io->error ? fr_network_error : NULL,
                                   s) < 0) {
                PERROR("Failed adding directory monitor event loop");
                if (network_socket_close(s) < 0) PWARN("Failed removing event for socket %d", s->number);
                talloc_free(s);
                return;
        }
 
        (void) fr_rb_insert(nr->sockets, s);
        (void) fr_rb_insert(nr->sockets_by_num, s);
 
        DEBUG3("Using new socket with FD %d", s->listen->fd);
}
 
/** Handle a network control message callback for a new worker
 *
 * @param[in] ctx the network
 * @param[in] data the message
 * @param[in] data_size size of the data
 * @param[in] now the current time
 */
static void fr_network_worker_started_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
{
        int i;
        fr_network_t *nr = ctx;
        fr_worker_t *worker;
        fr_network_worker_t *w;
 
        fr_assert(data_size == sizeof(worker));
 
        if (nr->num_workers >= nr->max_workers) {
                ERROR("Too many workers");
                return;
        }
 
        memcpy(&worker, data, data_size);
        (void) talloc_get_type_abort(worker, fr_worker_t);
 
        MEM(w = talloc_zero(nr, fr_network_worker_t));
 
        w->worker = worker;
        w->channel = fr_worker_channel_create(worker, w, nr->control);
        w->predicted = fr_time_delta_from_msec(10);
        fr_fatal_assert_msg(w->channel, "Failed creating new channel");
 
        fr_channel_requestor_uctx_add(w->channel, w);
        fr_channel_set_recv_reply(w->channel, nr, fr_network_recv_reply);
 
        /*
         *      Insert the worker into the array of workers.
         */
        for (i = 0; i < nr->max_workers; i++) {
                if (nr->workers[i]) continue;
 
                nr->workers[i] = w;
                nr->num_workers++;
                return;
        }
}
 
/** Handle a network control message callback for a packet sent to a socket
 *
 * @param[in] ctx the network
 * @param[in] data the message
 * @param[in] data_size size of the data
 * @param[in] now the current time
 */
static void fr_network_inject_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
{
        fr_network_t *nr = ctx;
        fr_network_inject_t my_inject;
        fr_network_socket_t *s;
 
        fr_assert(data_size == sizeof(my_inject));
 
        memcpy(&my_inject, data, data_size);
        s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = my_inject.listen });
        if (!s) {
                talloc_free(my_inject.packet); /* MUST be it's own TALLOC_CTX */
                return;
        }
 
        /*
         *      Inject the packet, and then read it back from the
         *      network.
         */
        if (s->listen->app_io->inject(s->listen, my_inject.packet, my_inject.packet_len, my_inject.recv_time) == 0) {
                fr_network_read(nr->el, s->listen->fd, 0, s);
        }
 
        talloc_free(my_inject.packet);
}
 
/** Run the event loop 'pre' callback
 *
 *  This function MUST DO NO WORK.  All it does is check if there's
 *  work, and tell the event code to return to the main loop if
 *  there's work to do.
 *
 * @param[in] now       the current time.
 * @param[in] wake      the time when the event loop will wake up.
 * @param[in] uctx      the network
 */
static int fr_network_pre_event(UNUSED fr_time_t now, UNUSED fr_time_delta_t wake, void *uctx)
{
        fr_network_t *nr = talloc_get_type_abort(uctx, fr_network_t);
 
        if (fr_heap_num_elements(nr->replies) > 0) return 1;
 
        return 0;
}
 
/** Handle replies after all FD and timer events have been serviced
 *
 * @param el    the event loop
 * @param now   the current time (mostly)
 * @param uctx  the fr_network_t
 */
static void fr_network_post_event(UNUSED fr_event_list_t *el, UNUSED fr_time_t now, void *uctx)
{
        fr_channel_data_t *cd;
        fr_network_t *nr = talloc_get_type_abort(uctx, fr_network_t);
 
        /*
         *      Pull the replies off of our global heap, and try to
         *      push them to the individual sockets.
         */
        while ((cd = fr_heap_pop(&nr->replies)) != NULL) {
                fr_listen_t *li;
                fr_network_socket_t *s;
 
                li = cd->listen;
 
                /*
                 *      @todo - cache this somewhere so we don't need
                 *      to do an rbtree lookup for every packet.
                 */
                s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
 
                /*
                 *      This shouldn't happen, but be safe...
                 */
                if (!s) {
                        fr_message_done(&cd->m);
                        continue;
                }
 
                if (cd->m.status != FR_MESSAGE_LOCALIZED) {
                        fr_assert(s->outstanding > 0);
                        s->outstanding--;
                }
 
                /*
                 *      Just mark the message done, and skip it.
                 */
                if (s->dead) {
                        fr_message_done(&cd->m);
 
                        /*
                         *      No more packets, it's safe to delete
                         *      the socket.
                         */
                        if (!s->outstanding) talloc_free(s);
 
                        continue;
                }
 
                /*
                 *      No data to write to the socket, so we skip the message.
                 */
                if (!cd->m.data_size) {
                        fr_message_done(&cd->m);
                        continue;
                }
 
                (void) fr_heap_insert(&s->waiting, cd);
 
                /*
                 *      No pending message, write it.  If there is a pending write, the message will be left
                 *      in the waiting queue.
                 */
                if (!s->pending) {
                        fr_assert(!s->blocked);
                        fr_network_write(nr->el, s->listen->fd, 0, s);
                }
        }
}
 
/** Stop a network thread in an orderly way
 *
 * @param[in] nr the network to stop
 */
int fr_network_destroy(fr_network_t *nr)
{
        fr_channel_data_t       *cd;
        int                     ret = 0;
 
        (void) talloc_get_type_abort(nr, fr_network_t);
 
        /*
         *      Close the network sockets, but leave them allocated.
         *
         *      Freeing a socket frees its message set, and the workers hold
         *      messages allocated from it until they ack the channel close
         *      signalled below.  fr_network() frees them after its loop, which
         *      already runs until every worker has acked.
         */
        {
                fr_network_socket_t     **sockets;
                size_t                  len;
                size_t                  i;
 
                if (fr_rb_flatten_inorder(nr, (void ***)&sockets, nr->sockets) < 0) return -1;
                len = talloc_array_length(sockets);
 
                for (i = 0; i < len; i++) {
                        if (network_socket_close(sockets[i]) < 0) {
                                PWARN("Failed removing event for socket %d", sockets[i]->number);
                                ret = -1;
                        }
                }
 
                talloc_free(sockets);
        }
 
 
        /*
         *      Clean up all outstanding replies.
         *
         *      We can't do this after signalling the
         *      workers to close, because they free
         *      their message sets, and we end up
         *      getting random use-after-free errors
         *      as there's a race between the network
         *      popping replies, and the workers
         *      freeing their message sets.
         *
         *      This isn't perfect, and we might still
         *      lose some replies, but it's good enough
         *      for now.
         *
         *      @todo - call transport "done" for the reply, so that
         *      it knows the replies are done, too.
         */
        while ((cd = fr_heap_pop(&nr->replies)) != NULL) {
                fr_message_done(&cd->m);
        }
 
        /*
         *      Nothing may be queued from here on: a worker that has been
         *      signalled discards whatever arrives late, without a reply.
         */
        nr->exiting = true;
 
        /*
         *      Signal the workers that we're closing
         *
         *      nr->num_workers is decremented every
         *      time a worker closes a socket.
         *
         *      When nr->num_workers == 0, the event
         *      loop (fr_network()) will exit.
         */
        {
                int i;
 
                for (i = 0; i < nr->num_workers; i++) {
                        fr_network_worker_t *worker = nr->workers[i];
 
                        fr_channel_signal_responder_close(worker->channel);
                }
        }
 
        (void) fr_event_pre_delete(nr->el, fr_network_pre_event, nr);
        (void) fr_event_post_delete(nr->el, fr_network_post_event, nr);
        fr_event_fd_delete(nr->el, nr->signal_pipe[0], FR_EVENT_FILTER_IO);
 
        return ret;
}
 
/** Read handler for signal pipe
 *
 */
static void _signal_pipe_read(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
{
        fr_network_t    *nr = talloc_get_type_abort(uctx, fr_network_t);
        uint8_t         buff;
 
        if (read(fd, &buff, sizeof(buff)) < 0) {
                ERROR("Failed reading signal - %s", fr_syserror(errno));
                return;
        }
 
        fr_assert(buff == 1);
 
        /*
         *      fr_network_stop() will signal the workers
         *      to exit (by closing their channels).
         *
         *      When we get the ack, we decrement our
         *      nr->num_workers counter.
         *
         *      When the counter reaches 0, the event loop
         *      exits.
         */
        DEBUG2("Signalled to exit");
 
        if (unlikely(fr_network_destroy(nr) < 0)) {
                PERROR("Failed stopping network");
        }
}
 
/** The main network worker function.
 *
 * @param[in] nr the network data structure to run.
 */
void fr_network(fr_network_t *nr)
{
        /*
         *      Run until we're told to exit AND the number of
         *      workers has dropped to zero.
         *
         *      This is important as if we exit too early we
         *      free the channels out from underneath the
         *      workers and they read uninitialised memory.
         *
         *      Whenever a worker ACKs our close notification
         *      nr->num_workers is decremented, so when
         *      nr->num_workers == 0, all workers have ACKd
         *      our close and are no longer using the channel.
         */
        while (likely(!(nr->exiting && (nr->num_workers == 0)))) {
                bool wait_for_event;
                int num_events;
 
                /*
                 *      There are runnable requests.  We still service
                 *      the event loop, but we don't wait for events.
                 */
                wait_for_event = (fr_heap_num_elements(nr->replies) == 0);
 
                /*
                 *      Check the event list.  If there's an error
                 *      (e.g. exit), we stop looping and clean up.
                 */
                DEBUG4("Gathering events - %s", wait_for_event ? "will wait" : "Will not wait");
                num_events = fr_event_corral(nr->el, fr_time(), wait_for_event);
                DEBUG4("%u event(s) pending%s",
                       num_events == -1 ? 0 : num_events, num_events == -1 ? " - event loop exiting" : "");
                if (num_events < 0) break;
 
                /*
                 *      Service outstanding events.
                 */
                if (num_events > 0) {
                        DEBUG4("Servicing event(s)");
                        fr_event_service(nr->el);
                }
        }
 
        /*
         *      Free the sockets.  The loop above only exits once every worker
         *      has acked, so the messages allocated from the sockets' message
         *      sets have all been returned, and fr_network_destroy() left the
         *      sockets closed but allocated for exactly this point.  It cannot do
         *      this itself, as it returns long before the first ack arrives.  The
         *      error exit above lands here too, where no ack is ever coming.
         */
        {
                fr_network_socket_t     **sockets;
                size_t                  len;
                size_t                  i;
 
                if (fr_rb_flatten_inorder(nr, (void ***)&sockets, nr->sockets) < 0) {
                        PWARN("Failed enumerating sockets, leaving them to be freed with the network");
                        return;
                }
                len = talloc_array_length(sockets);
 
                for (i = 0; i < len; i++) talloc_free(sockets[i]);
 
                talloc_free(sockets);
        }
}
 
/** Signal a network thread to exit
 *
 * @note Request to exit will be processed asynchronously.
 *
 * @param[in] nr the network data structure to manage
 * @return
 *      - 0 on success.
 *      - -1 on failure.
 */
int fr_network_exit(fr_network_t *nr)
{
        if (write(nr->signal_pipe[1], &(uint8_t){ 0x01 }, 1) < 0) {
                fr_strerror_printf("Failed signalling network thread to exit - %s", fr_syserror(errno));
                return -1;
        }
 
        return 0;
}
 
/** Free any resources associated with a network thread
 *
 */
static int _fr_network_free(fr_network_t *nr)
{
        if (nr->signal_pipe[0] >= 0) close(nr->signal_pipe[0]);
        if (nr->signal_pipe[1] >= 0) close(nr->signal_pipe[1]);
 
        return 0;
}
 
/** Create a network
 *
 * @param[in] ctx       The talloc ctx
 * @param[in] el        The event list
 * @param[in] name      Networker identifier.
 * @param[in] logger    The destination for all logging messages
 * @param[in] lvl       Log level
 * @param[in] config    configuration structure.
 * @return
 *      - NULL on error
 *      - fr_network_t on success
 */
fr_network_t *fr_network_create(TALLOC_CTX *ctx, fr_event_list_t *el, char const *name,
                                fr_log_t const *logger, fr_log_lvl_t lvl,
                                fr_network_config_t const *config)
{
        fr_network_t *nr;
 
        nr = talloc_zero(ctx, fr_network_t);
        if (!nr) {
                fr_strerror_const("Failed allocating memory");
                return NULL;
        }
        talloc_set_destructor(nr, _fr_network_free);
 
        nr->name = talloc_strdup(nr, name);
 
        nr->thread_id = pthread_self();
        nr->el = el;
        nr->log = logger;
        nr->lvl = lvl;
 
        nr->max_workers = MAX_WORKERS;
        nr->num_workers = 0;
        nr->signal_pipe[0] = -1;
        nr->signal_pipe[1] = -1;
        if (config) nr->config = *config;
 
        nr->aq_control = fr_atomic_queue_talloc(nr, 1024);
        if (!nr->aq_control) {
                talloc_free(nr);
                return NULL;
        }
 
        nr->control = fr_control_create(nr, el, nr->aq_control, 6);
        if (!nr->control) {
                fr_strerror_const_push("Failed creating control queue");
        fail:
                talloc_free(nr);
                return NULL;
        }
 
        /*
         *      @todo - rely on thread-local variables.  And then the
         *      various users of this can check if (rb == nr->rb), and
         *      if so, skip the whole control plane / kevent /
         *      whatever roundabout thing.
         */
        nr->rb = fr_ring_buffer_create(nr, FR_CONTROL_MAX_MESSAGES * FR_CONTROL_MAX_SIZE);
        if (!nr->rb) {
                fr_strerror_const_push("Failed creating ring buffer");
        fail2:
                talloc_free(nr->control);
                goto fail;
        }
 
        if (fr_control_callback_add(&nr->control, FR_CONTROL_ID_CHANNEL, nr, fr_network_channel_callback) < 0) {
                fr_strerror_const_push("Failed adding channel callback");
                goto fail2;
        }
 
        if (fr_control_callback_add(&nr->control, FR_CONTROL_ID_LISTEN, nr, fr_network_listen_callback) < 0) {
                fr_strerror_const_push("Failed adding socket callback");
                goto fail2;
        }
 
        if (fr_control_callback_add(&nr->control, FR_CONTROL_ID_DIRECTORY, nr, fr_network_directory_callback) < 0) {
                fr_strerror_const_push("Failed adding socket callback");
                goto fail2;
        }
 
        if (fr_control_callback_add(&nr->control, FR_CONTROL_ID_WORKER, nr, fr_network_worker_started_callback) < 0) {
                fr_strerror_const_push("Failed adding worker callback");
                goto fail2;
        }
 
        if (fr_control_callback_add(&nr->control, FR_CONTROL_ID_INJECT, nr, fr_network_inject_callback) < 0) {
                fr_strerror_const_push("Failed adding packet injection callback");
                goto fail2;
        }
 
        if (fr_control_open(nr->control) < 0) {
                fr_strerror_const_push("Failed opening control queue");
                goto fail2;
        }
 
        /*
         *      Create the various heaps.
         */
        nr->sockets = fr_rb_inline_talloc_alloc(nr, fr_network_socket_t, listen_node, socket_listen_cmp, NULL);
        if (!nr->sockets) {
                fr_strerror_const_push("Failed creating listen tree for sockets");
                goto fail2;
        }
 
        nr->sockets_by_num = fr_rb_inline_talloc_alloc(nr, fr_network_socket_t, num_node, socket_num_cmp, NULL);
        if (!nr->sockets_by_num) {
                fr_strerror_const_push("Failed creating number tree for sockets");
                goto fail2;
        }
 
        nr->replies = fr_heap_alloc(nr, reply_cmp, fr_channel_data_t, channel.heap_id, 0);
        if (!nr->replies) {
                fr_strerror_const_push("Failed creating heap for replies");
                goto fail2;
        }
 
        if (fr_event_pre_insert(nr->el, fr_network_pre_event, nr) < 0) {
                fr_strerror_const("Failed adding pre-check to event list");
                goto fail2;
        }
 
        if (fr_event_post_insert(nr->el, fr_network_post_event, nr) < 0) {
                fr_strerror_const("Failed inserting post-processing event");
                goto fail2;
        }
 
        if (pipe(nr->signal_pipe) < 0) {
                fr_strerror_printf("Failed initialising signal pipe - %s", fr_syserror(errno));
                goto fail2;
        }
        if (fr_nonblock(nr->signal_pipe[0]) < 0) goto fail2;
        if (fr_nonblock(nr->signal_pipe[1]) < 0) goto fail2;
 
        if (fr_event_fd_insert(nr, NULL, nr->el, nr->signal_pipe[0], _signal_pipe_read, NULL, NULL, nr) < 0) {
                fr_strerror_const("Failed inserting event for signal pipe");
                goto fail2;
        }
 
        return nr;
}
 
int fr_network_stats(fr_network_t const *nr, int num, uint64_t *stats)
{
        if (num < 0) return -1;
        if (num == 0) return 0;
 
        stats[0] = nr->stats.in;
        if (num >= 2) stats[1] = nr->stats.out;
        if (num >= 3) stats[2] = nr->stats.dup;
        if (num >= 4) stats[3] = nr->stats.dropped;
        if (num >= 5) stats[4] = nr->num_workers;
 
        if (num <= 5) return num;
 
        return 5;
}
 
void fr_network_stats_log(fr_network_t const *nr, fr_log_t const *log)
{
        int i;
 
        /*
         *      Dump all of the channel statistics.
         */
        for (i = 0; i < nr->max_workers; i++) {
                if (!nr->workers[i]) continue;
 
                fr_channel_stats_log(nr->workers[i]->channel, log, __FILE__, __LINE__);
        }
}
 
static int cmd_stats_self(FILE *fp, UNUSED FILE *fp_err, void *ctx, UNUSED fr_cmd_info_t const *info)
{
        fr_network_t const *nr = ctx;
 
        fprintf(fp, "count.in\t%" PRIu64 "\n", nr->stats.in);
        fprintf(fp, "count.out\t%" PRIu64 "\n", nr->stats.out);
        fprintf(fp, "count.dup\t%" PRIu64 "\n", nr->stats.dup);
        fprintf(fp, "count.dropped\t%" PRIu64 "\n", nr->stats.dropped);
        fprintf(fp, "count.sockets\t%u\n", fr_rb_num_elements(nr->sockets));
 
        return 0;
}
 
static int cmd_socket_list(FILE *fp, UNUSED FILE *fp_err, void *ctx, UNUSED fr_cmd_info_t const *info)
{
        fr_network_t const              *nr = ctx;
        fr_rb_iter_inorder_t    iter;
        fr_network_socket_t             *s;
 
        // @todo - note that this isn't thread-safe!
 
        for (s = fr_rb_iter_init_inorder(nr->sockets, &iter);
             s != NULL;
             s = fr_rb_iter_next_inorder(nr->sockets, &iter)) {
                if (!s->listen->app_io->get_name) {
                        fprintf(fp, "%s\n", s->listen->app_io->common.name);
                } else {
                        fprintf(fp, "%d\t%s\n", s->number, s->listen->app_io->get_name(s->listen));
                }
        }
        return 0;
}
 
static int cmd_stats_socket(FILE *fp, FILE *fp_err, void *ctx, fr_cmd_info_t const *info)
{
        fr_network_t const *nr = ctx;
        fr_network_socket_t *s;
 
        s = fr_rb_find(nr->sockets_by_num, &(fr_network_socket_t){ .number = info->box[0]->vb_uint32 });
        if (!s) {
                fprintf(fp_err, "No such socket number '%s'.\n", info->argv[0]);
                return -1;
        }
 
        fprintf(fp, "count.in\t%" PRIu64 "\n", s->stats.in);
        fprintf(fp, "count.out\t%" PRIu64 "\n", s->stats.out);
        fprintf(fp, "count.dup\t%" PRIu64 "\n", s->stats.dup);
        fprintf(fp, "count.dropped\t%" PRIu64 "\n", s->stats.dropped);
 
        return 0;
}
 
 
fr_cmd_table_t cmd_network_table[] = {
        {
                .parent = "stats",
                .name = "network",
                .help = "Statistics for network threads.",
                .read_only = true
        },
 
        {
                .parent = "stats network",
                .add_name = true,
                .name = "self",
                .func = cmd_stats_self,
                .help = "Show statistics for a specific network thread.",
                .read_only = true
        },
 
        {
                .parent = "stats network",
                .add_name = true,
                .name = "socket",
                .syntax = "INTEGER",
                .func = cmd_stats_socket,
                .help = "Show statistics for a specific socket",
                .read_only = true
        },
 
        {
                .parent = "show",
                .name = "network",
                .help = "Show information about network threads.",
                .read_only = true
        },
 
        {
                .parent = "show network",
                .add_name = true,
                .name = "socket",
                .syntax = "list",
                .func = cmd_socket_list,
                .help = "List the sockets associated with this network thread.",
                .read_only = true
        },
 
        CMD_TABLE_END
};