/*

 *   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;