channel.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: 693d7f9ff179420537b600091c13173c70ef1eda $
 *
 * @brief Two-way thread-safe channels.
 * @file io/channel.c
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSID("$Id: 693d7f9ff179420537b600091c13173c70ef1eda $")
 
#include <freeradius-devel/io/channel.h>
#include <freeradius-devel/util/debug.h>
 
#ifdef HAVE_STDATOMIC_H
#  include <stdatomic.h>
#else
#  include <freeradius-devel/util/stdatomic.h>
#endif
 
/*
 *      Debugging, mainly for channel_test
 */
#ifdef DEBUG_CHANNEL
#define MPRINT(...) fprintf(stdout, __VA_ARGS__)
#else
#define MPRINT(...)
#endif
 
typedef enum : uint8_t {
        TO_RESPONDER = 0,
        TO_REQUESTOR = 1
} fr_channel_direction_t;
 
#ifdef DEBUG_CHANNEL
static fr_table_num_sorted_t const channel_direction[] = {
        { L("to responder"),    TO_RESPONDER },
        { L("to requestor"),    TO_REQUESTOR },
};
size_t channel_direction_len = NUM_ELEMENTS(channel_direction);
#endif
 
#if 0
#define SIGNAL_INTERVAL (1000000)       //!< The minimum interval between responder signals.
#endif
 
/** Size of the atomic queues
 *
 * The queue reader MUST service the queue occasionally,
 * otherwise the writer will not be able to write.  If it's too
 * low, the writer will fail.  If it's too high, it will
 * unnecessarily use memory.  So we're better off putting it on
 * the high side.
 *
 * The reader SHOULD service the queues at inter-packet latency.
 * i.e. at 1M pps, the queue will get serviced every microsecond.
 */
#define ATOMIC_QUEUE_SIZE (1024)
 
/** The payload of every channel control message
 *
 *  Which event it is comes from the control message, not from the payload.
 */
typedef struct {
        uint64_t                ack;            //!< the sender's ack sequence, for data done.
        fr_channel_t            *ch;            //!< the channel
        fr_message_set_t        *reply_ms;      //!< the reply message set, for open only.
} fr_channel_control_t;
 
/** The callbacks one control plane's owner registered for channel events
 */
typedef struct {
        fr_control_t            *control;       //!< the control plane these were registered on
        fr_channel_callbacks_t  cb;             //!< what the owner does with each event
        void                    *uctx;          //!< passed to each callback
} fr_channel_hooks_t;
 
/** One end of a channel
 *
 * Consists of a kqueue descriptor, and an atomic queue.
 * The atomic queue is there to get bulk data through, because it's more efficient
 * than pushing 1M+ events per second through a kqueue.
 */
typedef struct {
        fr_channel_direction_t  direction;      //!< Use for debug messages.
 
        fr_control_t            *control;       //!< The control plane we signal this end through.
 
        void                    *uctx;          //!< Worker context.
 
        fr_channel_recv_callback_t recv;        //!< callback for receiving messages
        void                    *recv_uctx;     //!< context for receiving messages
 
        uint64_t                sequence;       //!< Sequence number for this channel.
        uint64_t                ack;            //!< Sequence number of the other end.
        uint64_t                their_view_of_my_sequence;      //!< Should be clear.
 
        uint64_t                sequence_at_last_signal;        //!< When we last signaled.
 
        fr_atomic_queue_t       *aq;            //!< The queue of messages - visible only to this channel.
 
        atomic_bool             active;         //!< Whether the channel is active.
 
        fr_channel_stats_t      stats;          //!< channel statistics
} fr_channel_end_t;
 
typedef struct fr_channel_s fr_channel_t;
 
/** A full channel, which consists of two ends
 *
 * A channel consists of an I/O identifier that can be placed in kequeue
 * and an atomic queue in each direction to allow for bidirectional communication.
 */
struct fr_channel_s {
        fr_time_delta_t         cpu_time;       //!< Total time used by the responder for this channel.
        fr_time_delta_t         processing_time; //!< Time spent by the responder processing requests.
 
        bool                    same_thread;    //!< are both ends in the same thread?
 
        fr_channel_end_t        end[2];         //!< Two ends of the channel.
};
 
fr_table_num_sorted_t const channel_signals[] = {
        { L("close"),                   FR_CHANNEL_CONTROL_CLOSE                },
        { L("close-ack"),               FR_CHANNEL_CONTROL_CLOSE_ACK            },
        { L("data-done-responder"),     FR_CHANNEL_CONTROL_DATA_DONE_RESPONDER  },
        { L("data-to-requestor"),       FR_CHANNEL_CONTROL_DATA_TO_REQUESTOR    },
        { L("data-to-responder"),       FR_CHANNEL_CONTROL_DATA_TO_RESPONDER    },
        { L("open"),                    FR_CHANNEL_CONTROL_OPEN                 },
};
size_t channel_signals_len = NUM_ELEMENTS(channel_signals);
 
fr_table_num_sorted_t const channel_packet_priority[] = {
        { L("high"),    PRIORITY_HIGH           },
        { L("low"),     PRIORITY_LOW            },
        { L("normal"),  PRIORITY_NORMAL         },
        { L("now"),     PRIORITY_NOW            }
};
size_t channel_packet_priority_len = NUM_ELEMENTS(channel_packet_priority);
 
 
/** Create a new channel
 *
 * @param[in] ctx       The talloc_ctx to allocate channel data in.
 * @param[in] requestor control plane.
 * @param[in] responder control plane.
 * @param[in] same      whether or not the channel is for the same thread
 * @return
 *      - NULL on error
 *      - channel on success
 */
fr_channel_t *fr_channel_create(TALLOC_CTX *ctx, fr_control_t *requestor, fr_control_t *responder, bool same)
{
        fr_time_t now;
        fr_channel_t *ch;
 
        ch = talloc_zero(ctx, fr_channel_t);
        if (!ch) {
        nomem:
                fr_strerror_const("Failed allocating memory");
                return NULL;
        }
 
        ch->same_thread = same;
 
        ch->end[TO_RESPONDER].direction = TO_RESPONDER;
        ch->end[TO_REQUESTOR].direction = TO_REQUESTOR;
 
        ch->end[TO_RESPONDER].aq = fr_atomic_queue_talloc(ch, ATOMIC_QUEUE_SIZE, sizeof(uintptr_t));
        if (!ch->end[TO_RESPONDER].aq) {
                talloc_free(ch);
                goto nomem;
        }
 
        ch->end[TO_REQUESTOR].aq = fr_atomic_queue_talloc(ch, ATOMIC_QUEUE_SIZE, sizeof(uintptr_t));
        if (!ch->end[TO_REQUESTOR].aq) {
                talloc_free(ch);
                goto nomem;
        }
 
        ch->end[TO_RESPONDER].control = responder;
        ch->end[TO_REQUESTOR].control = requestor;
 
        /*
         *      Initialize all of the timers to now.
         */
        now = fr_time();
 
        ch->end[TO_RESPONDER].stats.last_write = now;
        ch->end[TO_RESPONDER].stats.last_read_other = now;
        ch->end[TO_RESPONDER].stats.last_sent_signal = now;
        atomic_store(&ch->end[TO_RESPONDER].active, true);
 
        ch->end[TO_REQUESTOR].stats.last_write = now;
        ch->end[TO_REQUESTOR].stats.last_read_other = now;
        ch->end[TO_REQUESTOR].stats.last_sent_signal = now;
        atomic_store(&ch->end[TO_REQUESTOR].active, true);
 
        return ch;
}
 
 
/** Send a message via a kq user signal
 *
 * Note that the caller doesn't care about data in the event, that is
 * sent via the atomic queue.  The kevent code takes care of
 * delivering the signal once, even if it's sent by multiple requestor
 * threads.
 *
 * The thread watching the KQ knows which end it is.  So when it gets
 * the signal (and the channel pointer) it knows to look at end[0] or
 * end[1].  We also send which end in 'which' (0, 1) to further help
 * the recipient.
 *
 * @param[in] ch        the channel.
 * @param[in] when      the data was ready.  Typically taken from the message.
 * @param[in] end       of the channel that the message was written to.
 * @param[in] ev        the event to send.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
static int fr_channel_data_ready(fr_channel_t *ch, fr_time_t when, fr_channel_end_t *end, fr_channel_control_event_t ev)
{
        fr_channel_control_t cc;
 
        end->stats.last_sent_signal = when;
        end->stats.signals++;
 
        cc.ack = end->ack;
        cc.ch = ch;
        cc.reply_ms = NULL;
 
        MPRINT("Signalling %s, with %s\n",
               fr_table_str_by_value(channel_direction, end->direction, "<INVALID>"),
               fr_table_str_by_value(channel_signals, ev, "<INVALID>"));
 
        return fr_control_message_send(end->control, FR_CONTROL_EVENTSET_CHANNEL, ev, &cc, sizeof(cc));
}
 
#define IALPHA (8)
#define RTT(_old, _new) fr_time_delta_wrap((fr_time_delta_unwrap(_new) + (fr_time_delta_unwrap(_old) * (IALPHA - 1))) / IALPHA)
 
/** Send a request message into the channel
 *
 * The message should be initialized, other than "sequence" and "ack".
 *
 * This function automatically calls the recv_reply callback if there is a reply.
 *
 * @param[in] ch        the channel to send the request on.
 * @param[in] cd        the message to send.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_send_request(fr_channel_t *ch, fr_channel_data_t *cd)
{
        uint64_t sequence;
        fr_time_t when;
        fr_time_delta_t message_interval;
        fr_channel_end_t *requestor;
 
        if (!fr_cond_assert_msg(atomic_load(&ch->end[TO_RESPONDER].active), "Channel not active")) return -1;
 
        /*
         *      Same thread?  Just call the "recv" function directly.
         */
        if (ch->same_thread) {
                ch->end[TO_REQUESTOR].recv(ch, cd, ch->end[TO_REQUESTOR].recv_uctx);
                return 0;
        }
 
        requestor = &(ch->end[TO_RESPONDER]);
        when = cd->m.when;
 
        sequence = requestor->sequence + 1;
        cd->live.sequence = sequence;
        cd->live.ack = requestor->ack;
 
        /*
         *      Push the message onto the queue for the other end.  If
         *      the push fails, the caller should try another queue.
         */
        if (!fr_atomic_queue_spsc_push(requestor->aq, &cd)) {
                fr_strerror_printf("Failed pushing to atomic queue - full.  Queue contains %zu items",
                                   fr_atomic_queue_size(requestor->aq));
                while (fr_channel_recv_reply(ch));
                return -1;
        }
 
        requestor->sequence = sequence;
        message_interval = fr_time_sub(when, requestor->stats.last_write);
 
        if (!fr_time_delta_ispos(requestor->stats.message_interval)) {
                requestor->stats.message_interval = message_interval;
        } else {
                requestor->stats.message_interval = RTT(requestor->stats.message_interval, message_interval);
        }
 
        fr_assert_msg(fr_time_lteq(requestor->stats.last_write, when),
                      "Channel data timestamp (%" PRId64") older than last channel data sent (%" PRId64 ")",
                      fr_time_unwrap(when), fr_time_unwrap(requestor->stats.last_write));
        requestor->stats.last_write = when;
 
        requestor->stats.outstanding++;
        requestor->stats.packets++;
 
        MPRINT("REQUESTOR requests %"PRIu64", num_outstanding %"PRIu64"\n", requestor->stats.packets, requestor->stats.outstanding);
 
        /*
         *      The message is in the queue.  If the responder is
         *      awake it will find it before it next bsleeps, so
         *      the signal is only needed if it is already asleep.
         */
        if (!fr_control_is_sleeping(requestor->control)) {
                MPRINT("REQUESTOR SKIPS signal\n");
                return 0;
        }
 
        /*
         *      Tell the other end that there is new data ready.
         *
         *      Ignore errors on signalling.  The responder already has
         *      the packet in its inbound queue, so at some point, it
         *      will pick up the message.
         */
        MPRINT("REQUESTOR SIGNALS\n");
        (void) fr_channel_data_ready(ch, when, requestor, FR_CHANNEL_CONTROL_DATA_TO_RESPONDER);
        return 0;
}
 
/** Receive a reply message from the channel
 *
 * @param[in] ch        the channel to read data from.
 * @return
 *      - true if there was a message received
 *      - false if there are no more messages
 */
bool fr_channel_recv_reply(fr_channel_t *ch)
{
        fr_channel_data_t *cd;
        fr_channel_end_t *requestor;
        fr_atomic_queue_t *aq;
 
        fr_assert(ch->end[TO_RESPONDER].recv != NULL);
 
        aq = ch->end[TO_REQUESTOR].aq;
        requestor = &(ch->end[TO_RESPONDER]);
 
        /*
         *      It's OK for the queue to be empty.
         */
        if (!fr_atomic_queue_spsc_pop(aq, &cd)) return false;
 
        /*
         *      We want an exponential moving average for round trip
         *      time, where "alpha" is a number between [0,1)
         *
         *      RTT_new = alpha * RTT_old + (1 - alpha) * RTT_sample
         *
         *      BUT we use fixed-point arithmetic, so we need to use inverse alpha,
         *      which works out to the following equation:
         *
         *      RTT_new = (RTT_sample + (ialpha - 1) * RTT_old) / ialpha
         *
         *      NAKs have zero processing time, so we ignore them for
         *      the purpose of RTT.
         */
        if (fr_time_delta_ispos(cd->reply.processing_time)) {
                ch->processing_time = RTT(ch->processing_time, cd->reply.processing_time);
        }
        ch->cpu_time = cd->reply.cpu_time;
 
        /*
         *      Update the outbound channel with the knowledge that
         *      we've received one more reply, and with the responders
         *      ACK.
         */
        fr_assert(requestor->stats.outstanding > 0);
        fr_assert(cd->live.sequence > requestor->ack);
        fr_assert(cd->live.sequence <= requestor->sequence); /* must have fewer replies than requests */
 
        requestor->stats.outstanding--;
        requestor->ack = cd->live.sequence;
        requestor->their_view_of_my_sequence = cd->live.ack;
 
        fr_assert(fr_time_lteq(requestor->stats.last_read_other, cd->m.when));
        requestor->stats.last_read_other = cd->m.when;
 
        ch->end[TO_RESPONDER].recv(ch, cd, ch->end[TO_RESPONDER].recv_uctx);
 
        return true;
}
 
 
/** Receive a request message from the channel
 *
 * @param[in] ch the channel
 * @return
 *      - true if there was a message received
 *      - false if there are no more messages
 */
bool fr_channel_recv_request(fr_channel_t *ch)
{
        fr_channel_data_t *cd;
        fr_channel_end_t *responder;
        fr_atomic_queue_t *aq;
 
        aq = ch->end[TO_RESPONDER].aq;
        responder = &(ch->end[TO_REQUESTOR]);
 
        /*
         *      It's OK for the queue to be empty.
         */
        if (!fr_atomic_queue_spsc_pop(aq, &cd)) return false;
 
        fr_assert(cd->live.sequence > responder->ack);
        fr_assert(cd->live.sequence >= responder->sequence); /* must have more requests than replies */
 
        responder->stats.outstanding++;
        responder->ack = cd->live.sequence;
        responder->their_view_of_my_sequence = cd->live.ack;
 
        fr_assert(fr_time_lteq(responder->stats.last_read_other, cd->m.when));
        responder->stats.last_read_other = cd->m.when;
 
        ch->end[TO_REQUESTOR].recv(ch, cd, ch->end[TO_REQUESTOR].recv_uctx);
 
        return true;
}
 
/** Send a reply message into the channel
 *
 * The message should be initialized, other than "sequence" and "ack".
 *
 * @param[in] ch                the channel to send the reply on.
 * @param[in] cd                the message to send
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_send_reply(fr_channel_t *ch, fr_channel_data_t *cd)
{
        uint64_t                sequence;
        fr_time_t               when;
        fr_time_delta_t         message_interval;
        fr_channel_end_t        *responder;
 
        if (!fr_cond_assert_msg(atomic_load(&ch->end[TO_REQUESTOR].active), "Channel not active")) return -1;
 
        /*
         *      Same thread?  Just call the "recv" function directly.
         */
        if (ch->same_thread) {
                ch->end[TO_RESPONDER].recv(ch, cd, ch->end[TO_RESPONDER].recv_uctx);
                return 0;
        }
 
        responder = &(ch->end[TO_REQUESTOR]);
 
        when = cd->m.when;
 
        sequence = responder->sequence + 1;
        cd->live.sequence = sequence;
        cd->live.ack = responder->ack;
 
        if (!fr_atomic_queue_spsc_push(responder->aq, &cd)) {
                fr_strerror_printf("Failed pushing to atomic queue - full.  Queue contains %zu items",
                                   fr_atomic_queue_size(responder->aq));
                while (fr_channel_recv_request(ch));
                return -1;
        }
 
        fr_assert(responder->stats.outstanding > 0);
        responder->stats.outstanding--;
        responder->stats.packets++;
 
        MPRINT("\tRESPONDER replies %"PRIu64", num_outstanding %"PRIu64"\n", responder->stats.packets, responder->stats.outstanding);
 
        responder->sequence = sequence;
        message_interval = fr_time_sub(when, responder->stats.last_write);
        if (!fr_time_delta_ispos(responder->stats.message_interval)) {
                responder->stats.message_interval = message_interval;
        } else {
                responder->stats.message_interval = RTT(responder->stats.message_interval, message_interval);
        }
 
        fr_assert_msg(fr_time_lteq(responder->stats.last_write, when),
                      "Channel data timestamp (%" PRId64") older than last channel data sent (%" PRId64 ")",
                      fr_time_unwrap(when), fr_time_unwrap(responder->stats.last_write));
        responder->stats.last_write = when;
 
        /*
         *      Even if we think we have no more packets to process,
         *      the caller may have sent us one.  Go check the input
         *      channel.
         */
        while (fr_channel_recv_request(ch));
 
        /*
         *      The reply is in the queue.  If the requestor is awake
         *      it will find it before it next blocks, so the signal is
         *      only needed if it is already asleep.  The mirror of the
         *      check in fr_channel_send_request().
         */
        if (!fr_control_is_sleeping(responder->control)) {
                MPRINT("\tRESPONDER SKIPS signal\n");
                return 0;
        }
 
        /*
         *      No packets outstanding.  Tell the requestor how far we
         *      got, so it can re-signal us if it has sent more.
         */
        if (responder->stats.outstanding == 0) {
                (void) fr_channel_data_ready(ch, when, responder, FR_CHANNEL_CONTROL_DATA_DONE_RESPONDER);
                return 0;
        }
 
        MPRINT("\twhen - last_read_other = %"PRIu64" - %"PRIu64" = %"PRIu64"\n", when, responder->stats.last_read_other, when - responder->stats.last_read_other);
        MPRINT("\twhen - last signal = %"PRIu64" - %"PRIu64" = %"PRIu64"\n", when, responder->stats.last_sent_signal, when - responder->stats.last_sent_signal);
        MPRINT("\tsequence - ack = %"PRIu64" - %"PRIu64" = %"PRIu64"\n", responder->sequence, responder->their_view_of_my_sequence, responder->sequence - responder->their_view_of_my_sequence);
 
#ifdef __APPLE__
        /*
         *      If we've sent them a signal since the last ACK, they
         *      will receive it, and process the packets.  So we don't
         *      need to signal them again.
         *
         *      But... this doesn't appear to work on the Linux
         *      libkqueue implementation.
         */
        if (responder->sequence_at_last_signal > responder->their_view_of_my_sequence) return 0;
#endif
 
        /*
         *      If we've received a new packet in the last while, OR
         *      we've sent a signal in the last while, then we don't
         *      need to send a new signal.  But we DO send a signal if
         *      we haven't seen an ACK for a few packets.
         *
         *      FIXME: make these limits configurable, or include
         *      predictions about packet processing time?
         */
        fr_assert(responder->their_view_of_my_sequence <= responder->sequence);
#if 0
        if (((responder->sequence - their_view_of_my_sequence) <= 1000) &&
            ((when - responder->stats.last_read_other < SIGNAL_INTERVAL) ||
             ((when - responder->stats.last_sent_signal) < SIGNAL_INTERVAL))) {
                MPRINT("\tRESPONDER SKIPS signal\n");
                return 0;
        }
#endif
 
        MPRINT("\tRESPONDER SIGNALS num_outstanding %"PRIu64"\n", responder->stats.outstanding);
        (void) fr_channel_data_ready(ch, when, responder, FR_CHANNEL_CONTROL_DATA_TO_REQUESTOR);
        return 0;
}
 
 
/** Don't send a reply message into the channel
 *
 * The message should be the one we received from the network.
 *
 * @param[in] ch                the channel on which we're dropping a packet
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_null_reply(fr_channel_t *ch)
{
        fr_channel_end_t *responder;
 
        responder = &(ch->end[TO_REQUESTOR]);
 
        responder->sequence++;
        return 0;
}
 
 
 
/** Unpack the payload of a channel control message
 */
static inline void channel_message_unpack(fr_channel_control_t *cc, void const *data, NDEBUG_UNUSED size_t data_size)
{
        fr_assert(data_size == sizeof(*cc));
        memcpy(cc, data, sizeof(*cc));
}
 
/** Drain the requests queued for the responder
 *
 * Runs on the responder's thread, so a signal which finds nothing is
 * counted in the responder's own stats.  Each end's stats are only
 * ever written by that end's thread.
 */
static inline void channel_drain_requests(fr_channel_t *ch)
{
        if (!fr_channel_recv_request(ch)) {
                ch->end[TO_REQUESTOR].stats.spurious++;
                return;
        }
        while (fr_channel_recv_request(ch));
}
 
/** Drain the replies queued for the requestor
 *
 * Runs on the requestor's thread, see channel_drain_requests().
 */
static inline void channel_drain_replies(fr_channel_t *ch)
{
        if (!fr_channel_recv_reply(ch)) {
                ch->end[TO_RESPONDER].stats.spurious++;
                return;
        }
        while (fr_channel_recv_reply(ch));
}
 
/** The requestor has queued requests for the responder
 *
 * Must arrive on the responder's control plane.
 */
static void channel_data_to_responder(void const *data, size_t data_size, UNUSED fr_time_t now, void *uctx)
{
        fr_channel_hooks_t      *hooks = talloc_get_type_abort(uctx, fr_channel_hooks_t);
        fr_channel_control_t    cc;
 
        channel_message_unpack(&cc, data, data_size);
        MPRINT("channel got data_to_responder\n");
 
        if (!fr_cond_assert(hooks->control == cc.ch->end[TO_RESPONDER].control)) return;
        channel_drain_requests(cc.ch);
}
 
/** The responder has queued replies for the requestor
 *
 * Must arrive on the requestor's control plane.
 */
static void channel_data_to_requestor(void const *data, size_t data_size, UNUSED fr_time_t now, void *uctx)
{
        fr_channel_hooks_t      *hooks = talloc_get_type_abort(uctx, fr_channel_hooks_t);
        fr_channel_control_t    cc;
 
        channel_message_unpack(&cc, data, data_size);
        MPRINT("channel got data_to_requestor\n");
 
        if (!fr_cond_assert(hooks->control == cc.ch->end[TO_REQUESTOR].control)) return;
        channel_drain_replies(cc.ch);
}
 
/** The responder has serviced its requests, and says how many it has seen
 *
 * Only sent by the responder.  Replies may be waiting, so drain them.
 * First compare their ACK to the last sequence we sent, and if they
 * have not seen our latest, signal the responder to wake up.
 */
static void channel_data_done_responder(void const *data, size_t data_size, fr_time_t now, void *uctx)
{
        fr_channel_hooks_t      *hooks = talloc_get_type_abort(uctx, fr_channel_hooks_t);
        fr_channel_control_t    cc;
        fr_channel_t            *ch;
        fr_channel_end_t        *requestor;
 
        channel_message_unpack(&cc, data, data_size);
        ch = cc.ch;
        MPRINT("channel got data_done_responder\n");
 
        if (!fr_cond_assert(hooks->control == ch->end[TO_REQUESTOR].control)) return;
 
        requestor = &ch->end[TO_RESPONDER];
 
        /*
         *      We have told this end to close, so there is nothing left to
         *      wake it for: fr_channel_send_request() refuses on an inactive
         *      end, so no more data can be queued.  Signalling it anyway
         *      writes into the control plane which the responder may have
         *      already freed (this has been observed and caused a crash-on-exit).
         *
         *      Servicing the message is still correct: it was in flight
         *      before the close, and we must keep servicing to receive the
         *      ack at all.  The load is relaxed because only this thread
         *      clears "active" for this end.
         */
        if (unlikely(!atomic_load_explicit(&requestor->active, memory_order_relaxed))) goto drain;
 
        /*
         *      Their ack covers everything we have sent.  Anything
         *      sent later signals for itself.
         */
        fr_assert(cc.ack <= requestor->sequence);
        if (cc.ack == requestor->sequence) {
                MPRINT("REQUESTOR SKIPS signal AFTER data_done_responder num_outstanding %"PRIu64"\n",
                       requestor->stats.outstanding);
                goto drain;
        }
 
        /*
         *      They have not seen our latest.  Each of those sends
         *      already signalled, or was pushed while the responder was
         *      awake and so will be found by its final drain.  Signal
         *      again only if it has since blocked: cheap insurance
         *      against a dropped wakeup, free when it is running.
         */
        if (!fr_control_is_sleeping(requestor->control)) goto drain;
 
        requestor->stats.resignals++;
 
        MPRINT("REQUESTOR SIGNALS AFTER data_done_responder\n");
        if (fr_channel_data_ready(ch, now, requestor, FR_CHANNEL_CONTROL_DATA_TO_RESPONDER) < 0) return;
 
drain:
        channel_drain_replies(ch);
}
 
/** A requestor has opened a channel to us
 */
static void channel_open(void const *data, size_t data_size, fr_time_t now, void *uctx)
{
        fr_channel_hooks_t      *hooks = talloc_get_type_abort(uctx, fr_channel_hooks_t);
        fr_channel_control_t    cc;
 
        channel_message_unpack(&cc, data, data_size);
        MPRINT("channel got open\n");
 
        if (!fr_cond_assert(hooks->cb.open)) return;
        hooks->cb.open(cc.ch, cc.reply_ms, now, hooks->uctx);
}
 
/** The requestor is closing a channel to us
 */
static void channel_close(void const *data, size_t data_size, fr_time_t now, void *uctx)
{
        fr_channel_hooks_t      *hooks = talloc_get_type_abort(uctx, fr_channel_hooks_t);
        fr_channel_control_t    cc;
 
        channel_message_unpack(&cc, data, data_size);
        MPRINT("channel got close\n");
 
        if (!fr_cond_assert(hooks->cb.close)) return;
        hooks->cb.close(cc.ch, now, hooks->uctx);
}
 
/** The responder has finished closing a channel we asked it to close
 */
static void channel_close_ack(void const *data, size_t data_size, fr_time_t now, void *uctx)
{
        fr_channel_hooks_t      *hooks = talloc_get_type_abort(uctx, fr_channel_hooks_t);
        fr_channel_control_t    cc;
 
        channel_message_unpack(&cc, data, data_size);
        MPRINT("channel got close_ack\n");
 
        if (!fr_cond_assert(hooks->cb.closed)) return;
        hooks->cb.closed(cc.ch, now, hooks->uctx);
}
 
/** Register the owner's channel callbacks on its control plane
 *
 * Call before fr_control_open().  Every channel event gets its
 * own entry in the FR_CONTROL_EVENTSET_CHANNEL event set, so the control
 * plane dispatches straight to the handler for that event.
 *
 * @param[in] c         the control plane channels will signal.
 * @param[in] callbacks what the owner does with each event.  Copied.
 * @param[in] uctx      passed to each callback.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_callbacks_add(fr_control_t *c, fr_channel_callbacks_t const *callbacks, void *uctx)
{
        fr_channel_hooks_t *hooks;
 
        MEM(hooks = talloc_zero(c, fr_channel_hooks_t));
        hooks->control = c;
        hooks->cb = *callbacks;
        hooks->uctx = uctx;
 
        if ((fr_control_callback_add(c, FR_CONTROL_EVENTSET_CHANNEL, FR_CHANNEL_CONTROL_DATA_TO_RESPONDER,
                                     channel_data_to_responder, hooks) < 0) ||
            (fr_control_callback_add(c, FR_CONTROL_EVENTSET_CHANNEL, FR_CHANNEL_CONTROL_DATA_TO_REQUESTOR,
                                     channel_data_to_requestor, hooks) < 0) ||
            (fr_control_callback_add(c, FR_CONTROL_EVENTSET_CHANNEL, FR_CHANNEL_CONTROL_DATA_DONE_RESPONDER,
                                     channel_data_done_responder, hooks) < 0) ||
            (fr_control_callback_add(c, FR_CONTROL_EVENTSET_CHANNEL, FR_CHANNEL_CONTROL_OPEN,
                                     channel_open, hooks) < 0) ||
            (fr_control_callback_add(c, FR_CONTROL_EVENTSET_CHANNEL, FR_CHANNEL_CONTROL_CLOSE,
                                     channel_close, hooks) < 0) ||
            (fr_control_callback_add(c, FR_CONTROL_EVENTSET_CHANNEL, FR_CHANNEL_CONTROL_CLOSE_ACK,
                                     channel_close_ack, hooks) < 0)) {
                talloc_free(hooks);
                return -1;
        }
 
        return 0;
}
 
 
/** Service a control-plane event.
 *
 * The channels use control planes for internal signaling.  Note that
 * the caller does NOT pass the channel into this function.  Instead,
 * the channel is taken from the kevent.
 *
 * @param[in] ch        The channel to service.
 * @param[in] c         The control plane on which we received the kev.
 * @param[in] kev       The kevent data, should get passed to the control plane.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_service_kevent(fr_channel_t *ch, fr_control_t *c, UNUSED struct kevent const *kev)
{
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        if (c == ch->end[TO_RESPONDER].control) {
                ch->end[TO_RESPONDER].stats.kevents++;
        } else {
                ch->end[TO_REQUESTOR].stats.kevents++;
        }
 
        return 0;
}
 
 
/** Check if a channel is active.
 *
 * A channel may be closed by either end.  If so, it stays alive (but
 * inactive) until both ends acknowledge the close.
 *
 * @param[in] ch the channel
 * @return
 *      - false the channel is closing.
 *      - true the channel is active
 */
bool fr_channel_active(fr_channel_t *ch)
{
        return atomic_load(&ch->end[TO_REQUESTOR].active) && atomic_load(&ch->end[TO_RESPONDER].active);
}
 
/** Signal a responder that the channel is closing
 *
 * @param[in] ch        The channel.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_signal_responder_close(fr_channel_t *ch)
{
        int ret;
        bool active;
        fr_channel_control_t cc;
 
        active = atomic_load(&ch->end[TO_RESPONDER].active);
        if (!active) return 0;                                  /* Already signalled to close */
 
        atomic_store(&ch->end[TO_RESPONDER].active, false);     /* Prevent further requests */
 
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        cc.ack = 0;
        cc.ch = ch;
        cc.reply_ms = NULL;
 
        ret = fr_control_message_send(ch->end[TO_RESPONDER].control, FR_CONTROL_EVENTSET_CHANNEL,
                                      FR_CHANNEL_CONTROL_CLOSE, &cc, sizeof(cc));
 
        return ret;
}
 
/** Discard any requests the requestor queued but we never received
 *
 * The messages belong to the requestor's message set, and it cannot reclaim them
 * until they are marked done.  A responder that closes with requests still in
 * the queue would strand them there, and with them the memory they were
 * allocated from.
 *
 * @param[in] ch        to discard the queued requests of.
 * @return the number of requests discarded.
 */
unsigned int fr_channel_responder_discard(fr_channel_t *ch)
{
        fr_channel_data_t       *cd;
        fr_atomic_queue_t       *aq = ch->end[TO_RESPONDER].aq;
        unsigned int            num = 0;
 
        while (fr_atomic_queue_spsc_pop(aq, &cd)) {
                fr_message_done(&cd->m);
                num++;
        }
 
        return num;
}
 
/** Acknowledge that the channel is closing
 *
 * @param[in] ch        The channel.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_responder_ack_close(fr_channel_t *ch)
{
        int ret;
        bool active;
 
        fr_channel_control_t cc;
 
        active = atomic_load(&ch->end[TO_REQUESTOR].active);
        if (!active) return 0;                                  /* Already signalled to close */
 
        atomic_store(&ch->end[TO_REQUESTOR].active, false);     /* Prevent further responses */
 
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        cc.ack = 0;
        cc.ch = ch;
        cc.reply_ms = NULL;
 
        ret = fr_control_message_send(ch->end[TO_REQUESTOR].control, FR_CONTROL_EVENTSET_CHANNEL,
                                      FR_CHANNEL_CONTROL_CLOSE_ACK, &cc, sizeof(cc));
 
        return ret;
}
 
/** Add responder-specific data to a channel
 *
 * @param[in] ch        The channel.
 * @param[in] uctx      The context to add.
 */
void fr_channel_responder_uctx_add(fr_channel_t *ch, void *uctx)
{
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        ch->end[TO_REQUESTOR].uctx = uctx;
}
 
 
/** Get responder-specific data from a channel
 *
 * @param[in] ch        The channel.
 */
void *fr_channel_responder_uctx_get(fr_channel_t *ch)
{
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        return ch->end[TO_REQUESTOR].uctx;
}
 
 
/** Add network-specific data to a channel
 *
 * @param[in] ch        The channel.
 * @param[in] uctx      The context to add.
 */
void fr_channel_requestor_uctx_add(fr_channel_t *ch, void *uctx)
{
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        ch->end[TO_RESPONDER].uctx = uctx;
}
 
 
/** Get network-specific data from a channel
 *
 * @param[in] ch        The channel.
 */
void *fr_channel_requestor_uctx_get(fr_channel_t *ch)
{
        (void) talloc_get_type_abort(ch, fr_channel_t);
 
        return ch->end[TO_RESPONDER].uctx;
}
 
 
int fr_channel_set_recv_reply(fr_channel_t *ch, fr_channel_recv_callback_t recv_reply, void *uctx)
{
        ch->end[TO_RESPONDER].recv = recv_reply;
        ch->end[TO_RESPONDER].recv_uctx = uctx;
 
        return 0;
}
 
int fr_channel_set_recv_request(fr_channel_t *ch, fr_channel_recv_callback_t recv_request, void *uctx)
{
        ch->end[TO_REQUESTOR].recv = recv_request;
        ch->end[TO_REQUESTOR].recv_uctx = uctx;
        return 0;
}
 
/** Send a channel to a responder
 *
 * @param[in] ch        The channel.
 * @param[in] reply_ms  The message set the responder allocates replies from.
 * @return
 *      - <0 on error
 *      - 0 on success
 */
int fr_channel_signal_open(fr_channel_t *ch, fr_message_set_t *reply_ms)
{
        fr_channel_control_t cc;
 
        cc.ack = 0;
        cc.ch = ch;
        cc.reply_ms = reply_ms;
 
        return fr_control_message_send(ch->end[TO_RESPONDER].control, FR_CONTROL_EVENTSET_CHANNEL,
                                       FR_CHANNEL_CONTROL_OPEN, &cc, sizeof(cc));
}
 
void fr_channel_stats_log(fr_channel_t const *ch, fr_log_t const *log, char const *file, int line)
{
        fr_log(log, L_INFO, file, line, "requestor\n");
        fr_log(log, L_INFO, file, line, "\tsignals sent = %" PRIu64 "\n", ch->end[TO_RESPONDER].stats.signals);
        fr_log(log, L_INFO, file, line, "\tsignals re-sent = %" PRIu64 "\n", ch->end[TO_RESPONDER].stats.resignals);
        fr_log(log, L_INFO, file, line, "\tsignals received empty = %" PRIu64 "\n",
               ch->end[TO_RESPONDER].stats.spurious);
        fr_log(log, L_INFO, file, line, "\tkevents checked = %" PRIu64 "\n", ch->end[TO_RESPONDER].stats.kevents);
        fr_log(log, L_INFO, file, line, "\toutstanding = %" PRIu64 "\n", ch->end[TO_RESPONDER].stats.outstanding);
        fr_log(log, L_INFO, file, line, "\tpackets processed = %" PRIu64 "\n", ch->end[TO_RESPONDER].stats.packets);
        fr_log(log, L_INFO, file, line, "\tmessage interval (RTT) = %" PRId64 "\n", fr_time_delta_unwrap(ch->end[TO_RESPONDER].stats.message_interval));
        fr_log(log, L_INFO, file, line, "\tlast write = %" PRId64 "\n", fr_time_unwrap(ch->end[TO_RESPONDER].stats.last_write));
        fr_log(log, L_INFO, file, line, "\tlast read other end = %" PRId64 "\n", fr_time_unwrap(ch->end[TO_RESPONDER].stats.last_read_other));
        fr_log(log, L_INFO, file, line, "\tlast signal other = %" PRId64 "\n", fr_time_unwrap(ch->end[TO_RESPONDER].stats.last_sent_signal));
 
        fr_log(log, L_INFO, file, line, "responder\n");
        fr_log(log, L_INFO, file, line, "\tsignals sent = %" PRIu64"\n", ch->end[TO_REQUESTOR].stats.signals);
        fr_log(log, L_INFO, file, line, "\tsignals received empty = %" PRIu64 "\n",
               ch->end[TO_REQUESTOR].stats.spurious);
        fr_log(log, L_INFO, file, line, "\tkevents checked = %" PRIu64 "\n", ch->end[TO_REQUESTOR].stats.kevents);
        fr_log(log, L_INFO, file, line, "\tpackets processed = %" PRIu64 "\n", ch->end[TO_REQUESTOR].stats.packets);
        fr_log(log, L_INFO, file, line, "\tmessage interval (RTT) = %" PRId64 "\n", fr_time_delta_unwrap(ch->end[TO_REQUESTOR].stats.message_interval));
        fr_log(log, L_INFO, file, line, "\tlast write = %" PRId64 "\n", fr_time_unwrap(ch->end[TO_REQUESTOR].stats.last_write));
        fr_log(log, L_INFO, file, line, "\tlast read other end = %" PRId64 "\n", fr_time_unwrap(ch->end[TO_REQUESTOR].stats.last_read_other));
        fr_log(log, L_INFO, file, line, "\tlast signal other = %" PRId64 "\n", fr_time_unwrap(ch->end[TO_REQUESTOR].stats.last_sent_signal));
}
 
/** Statistics for the requestor's end of a channel
 *
 * Signals the requestor sent, and signals it received which found nothing.
 */
fr_channel_stats_t const *fr_channel_requestor_stats(fr_channel_t const *ch)
{
        return &ch->end[TO_RESPONDER].stats;
}
 
/** Statistics for the responder's end of a channel
 */
fr_channel_stats_t const *fr_channel_responder_stats(fr_channel_t const *ch)
{
        return &ch->end[TO_REQUESTOR].stats;
}