/*

 *   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: c133ceaca5bce535f3de3800c578b6d6b51bba5f $

 *

 * @brief Control-plane signaling

 * @file io/control.c

 *

 * @copyright 2016 Alan DeKok (aland@freeradius.org)

 */

RCSID("$Id: c133ceaca5bce535f3de3800c578b6d6b51bba5f $")

#include <freeradius-devel/io/control.h>

#include <freeradius-devel/util/strerror.h>

#include <freeradius-devel/util/syserror.h>

#include <freeradius-devel/util/misc.h>

#include <freeradius-devel/util/rand.h>

#include <fcntl.h>

#include <sys/event.h>

#include <poll.h>

/*

 *  Debugging, mainly for channel_test

 */

#if 0

#define MPRINT(...) fprintf(stderr, __VA_ARGS__)

#else

#define MPRINT(...)

#endif

/**

 *  Status of control messages

 */

typedef enum fr_control_message_status_t {

  FR_CONTROL_MESSAGE_FREE = 0,      //!< the message is free

  FR_CONTROL_MESSAGE_USED,      //!< the message is used (set only by originator)

  FR_CONTROL_MESSAGE_DONE       //!< the message is done (set only by receiver)

} fr_control_message_status_t;

/**

 *  The header for the control message

 */

typedef struct {

  fr_control_message_status_t status;   //!< status of this message

  uint32_t      id;   //!< ID of this message

  size_t        data_size;      //!< size of the data we're sending

} fr_control_message_t;

typedef struct {

  uint32_t      id;   //!< id of this callback

  void        *ctx;   //!< context for the callback

  fr_control_callback_t   callback; //!< the function to call

} fr_control_ctx_t;

/**

 *  The control structure.

 */

struct fr_control_s {

  fr_event_list_t   *el;      //!< our event list

  fr_atomic_queue_t *aq;      //!< destination AQ

  int     pipe[2];          //!< our pipes

  bool      same_thread;    //!< are the two ends in the same thread

  bool      opened;     //!< has the control path been opened.

  uint32_t    num_callbacks;    //!< the size of the callback array

  fr_control_ctx_t  type[];     //!< callbacks.  Must be at the end of the structure as these

              ///< are created by allocating a larger memory size.

};

static void pipe_read(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)

{

  fr_control_t *c = talloc_get_type_abort(uctx, fr_control_t);

  fr_time_t now;

  char read_buffer[256];

  uint8_t data[256];

  size_t message_size;

  uint32_t id = 0;

  /*

   *  The presence of data on the pipe fd is just a trigger to pop all

   *  available messages from the atomic queue, so the number of bytes

   *  read is not important.

   */

  /* coverity[check_return] */

  if (read(fd, read_buffer, sizeof(read_buffer)) <= 0) return;

  now = fr_time();

  while((message_size = fr_control_message_pop(c->aq, &id, data, sizeof(data)))) {

    if (id >= c->num_callbacks) continue;

    if (!c->type[id].callback) continue;

    c->type[id].callback(c->type[id].ctx, data, message_size, now);

  }

}

/** Free a control structure

 *

 *  This function really only calls the underlying "garbage collect".

 *

 * @param[in] c the control structure

 */

static int _control_free(fr_control_t *c)

{

  (void) talloc_get_type_abort(c, fr_control_t);

#ifndef NDEBUG

  (void) fr_event_fd_unarmour(c->el, c->pipe[0], FR_EVENT_FILTER_IO, (uintptr_t)c);

#endif

  (void) fr_event_fd_delete(c->el, c->pipe[0], FR_EVENT_FILTER_IO);

  close(c->pipe[0]);

  close(c->pipe[1]);

  return 0;

}

/** Create a control-plane signaling path.

 *

 * @param[in] ctx the talloc context

 * @param[in] el the event list for the control socket

 * @param[in] aq the atomic queue where we will be pushing message data

 * @param[in] num_callbacks the initial number of callback entries to allocate.

 * @return

 *  - NULL on error

 *  - fr_control_t on success

 */

fr_control_t *fr_control_create(TALLOC_CTX *ctx, fr_event_list_t *el, fr_atomic_queue_t *aq, size_t num_callbacks)

{

  fr_control_t *c;

  c = talloc_zero_size(ctx, sizeof(fr_control_t) + sizeof(fr_control_ctx_t) * num_callbacks);

  talloc_set_type(c, fr_control_t);

  if (!c) {

    fr_strerror_const("Failed allocating memory");

    return NULL;

  }

  c->el = el;

  c->aq = aq;

  c->num_callbacks = num_callbacks;

  return c;

}

/** Open the control-plane signalling path

 *

 * @param[in] c the control-plane to open

 * @return

 *  - 0 on success

 *  - -1 on failure

 */

int fr_control_open(fr_control_t *c)

{

  if (pipe(c->pipe) < 0) {

    fr_strerror_printf("Failed opening pipe for control socket: %s", fr_syserror(errno));

    return -1;

  }

  talloc_set_destructor(c, _control_free);

  /*

   *  We don't want reads from the pipe to be blocking.

   */

  (void) fr_nonblock(c->pipe[0]);

  (void) fr_nonblock(c->pipe[1]);

  (void) fr_cloexec(c->pipe[0]);

  (void) fr_cloexec(c->pipe[1]);

  if (fr_event_fd_insert(c, NULL, c->el, c->pipe[0], pipe_read, NULL, NULL, c) < 0) {

    fr_strerror_const_push("Failed adding FD to event list control socket");

    return -1;

  }

#ifndef NDEBUG

  (void) fr_event_fd_armour(c->el, c->pipe[0], FR_EVENT_FILTER_IO, (uintptr_t)c);

#endif

  c->opened = true;

  return 0;

}

/** Clean up messages in a control-plane buffer

 *

 *  Find the oldest messages which are marked FR_CONTROL_MESSAGE_DONE,

 *  and mark them FR_CONTROL_MESSAGE_FREE.

 *

 * @param[in] c the fr_control_t

 * @param[in] rb the callers ring buffer for message allocation.

 * @return

 *  - <0 there are still messages used

 *  - 0 the control list is empty.

 */

int fr_control_gc(UNUSED fr_control_t *c, fr_ring_buffer_t *rb)

{

  while (true) {

    size_t room, message_size;

    fr_control_message_t *m;

    (void) fr_ring_buffer_start(rb, (uint8_t **) &m, &room);

    if (room == 0) break;

    fr_assert(m != NULL);

    fr_assert(room >= sizeof(*m));

    fr_assert(m->status != FR_CONTROL_MESSAGE_FREE);

    if (m->status != FR_CONTROL_MESSAGE_DONE) break;

    m->status = FR_CONTROL_MESSAGE_FREE;

    /*

     *  Each message is aligned to a 64-byte boundary,

     *  for cache contention issues.

     */

    message_size = sizeof(*m);

    message_size += m->data_size;

    message_size += 63;

    message_size &= ~(size_t) 63;

    fr_ring_buffer_free(rb, message_size);

  }

  /*

   *  Maybe we failed to garbage collect everything?

   */

  if (fr_ring_buffer_used(rb) > 0) {

    fr_strerror_const("Data still in control buffers");

    return -1;

  }

  return 0;

}

/** Allocate a control message

 *

 * @param[in] c the control structure

 * @param[in] rb the callers ring buffer for message allocation.

 * @param[in] id the ident of this message.

 * @param[in] data the data to write to the control plane

 * @param[in] data_size the size of the data to write to the control plane.

 * @return

 *  - NULL on error

 *  - fr_message_t on success

 */

static fr_control_message_t *fr_control_message_alloc(fr_control_t *c, fr_ring_buffer_t *rb, uint32_t id, void *data, size_t data_size)

{

  size_t message_size;

  fr_control_message_t *m;

  uint8_t *p;

  message_size = sizeof(*m);

  message_size += data_size;

  message_size += 63;

  message_size &= ~(size_t) 63;

  m = (fr_control_message_t *) fr_ring_buffer_alloc(rb, message_size);

  if (!m) {

    (void) fr_control_gc(c, rb);

    m = (fr_control_message_t *) fr_ring_buffer_alloc(rb, message_size);

    if (!m) {

      fr_strerror_const_push("Failed allocating from ring buffer");

      return NULL;

    }

  }

  m->status = FR_CONTROL_MESSAGE_USED;

  m->id = id;

  m->data_size = data_size;

  p = (uint8_t *) m;

  memcpy(p + sizeof(*m), data, data_size);

  return m;

}

/** Push a control-plane message

 *

 *  This function is called ONLY from the originating threads.

 *

 * @param[in] c the control structure

 * @param[in] rb the callers ring buffer for message allocation.

 * @param[in] id the ident of this message.

 * @param[in] data the data to write to the control plane

 * @param[in] data_size the size of the data to write to the control plane.

 * @return

 *  - -2 on ring buffer full

 *  - <0 on error

 *  - 0 on success

 */

int fr_control_message_push(fr_control_t *c, fr_ring_buffer_t *rb, uint32_t id, void *data, size_t data_size)

{

  fr_control_message_t *m;

  (void) talloc_get_type_abort(c, fr_control_t);

  MPRINT("CONTROL push aq %p\n", c->aq);

  /*

   *  Get a message.  The alloc call attempts garbage collection

   *  if there is not enough free space.

   */

  m = fr_control_message_alloc(c, rb, id, data, data_size);

  if (!m) {

    fr_strerror_const_push("Failed allocating after GC");

    return -2;

  }

  if (!fr_atomic_queue_push(c->aq, m)) {

    m->status = FR_CONTROL_MESSAGE_DONE;

    fr_strerror_const("Failed pushing message to atomic queue.");

    return -1;

  }

  return 0;

}

/** Send a control-plane message

 *

 *  This function is called ONLY from the originating threads.

 *

 * @param[in] c the control structure

 * @param[in] rb the callers ring buffer for message allocation.

 * @param[in] id the ident of this message.

 * @param[in] data the data to write to the control plane

 * @param[in] data_size the size of the data to write to the control plane.

 * @return

 *  - <0 on error

 *  - 0 on success

 */

int fr_control_message_send(fr_control_t *c, fr_ring_buffer_t *rb, uint32_t id, void *data, size_t data_size)

{

  (void) talloc_get_type_abort(c, fr_control_t);

  if (c->same_thread) {

    if (!c->type[id].callback) return -1;

    c->type[id].callback(c->type[id].ctx, data, data_size, fr_time());

    return 0;

  }

  if (fr_control_message_push(c, rb, id, data, data_size) < 0) return -1;

redo:

  if (write(c->pipe[1], ".", 1) >= 0) return 0;

  if (errno == EINTR) goto redo;

  /*

   *  EAGAIN means that the pipe is full, which means that the other end will eventually

   *  read from it.

   */

  if (errno == EAGAIN) return 0;

#if defined(EWOULDBLOCK) && (EWOULDBLOCK != EAGAIN)

  if (errno == EWOULDBLOCK) return 0;

#endif

  /*

   *  Other error, that's an issue.

   */

  return -1;

}

/** Pop control-plane message

 *

 *  This function is called ONLY from the receiving thread.

 *

 * @param[in] aq the recipients atomic queue for control-plane messages

 * @param[out] p_id the ident of this message.

 * @param[in,out] data where the data is stored

 * @param[in] data_size the size of the buffer where we store the data.

 * @return

 *  - <0 the size of the data we need to read the next message

 *  - 0 this kevent is not for us.

 *  - >0 the amount of data we've read

 */

ssize_t fr_control_message_pop(fr_atomic_queue_t *aq, uint32_t *p_id, void *data, size_t data_size)

{

  uint8_t *p;

  fr_control_message_t *m;

  MPRINT("CONTROL pop aq %p\n", aq);

  if (!fr_atomic_queue_pop(aq, (void **) &m)) return 0;

  fr_assert_msg(m->status == FR_CONTROL_MESSAGE_USED, "Bad control message state, expected %u got %u",

          FR_CONTROL_MESSAGE_USED, m->status);

  /*

   *  There isn't enough room to store the data, die.

   */

  if (data_size < m->data_size) {

    fr_strerror_printf("Allocation size should be at least %zd", m->data_size);

    return -(m->data_size);

  }

  p = (uint8_t *) m;

  data_size = m->data_size;

  memcpy(data, p + sizeof(*m), data_size);

  m->status = FR_CONTROL_MESSAGE_DONE;

  *p_id = m->id;

  return data_size;

}

/** Register a callback for an ID

 *

 * @param[in] c the control structure

 * @param[in] id the ident of this message.

 * @param[in] ctx the context for the callback

 * @param[in] callback the callback function

 * @return

 *  - <0 on error

 *  - 0 on success

 */

int fr_control_callback_add(fr_control_t **c, uint32_t id, void *ctx, fr_control_callback_t callback)

{

  fr_control_t  *ctrl = talloc_get_type_abort(*c, fr_control_t);

  if (ctrl->opened) {

    fr_strerror_printf("Cannot add callbacks after the control is opened");

    return -1;

  }

  /*

   *  If there is not enough space in the array of callbacks

   *  re-allocate the fr_control_t with a larger array.

   */

  if (id >= ctrl->num_callbacks) {

    ctrl = talloc_realloc_size(talloc_parent(*c), *c, sizeof(fr_control_t) + sizeof(fr_control_ctx_t) * (id + 1));

    if (!ctrl) {

      fr_strerror_printf("Failed re-allocating control when registering callback ID %d", id);

      return -1;

    }

    talloc_set_type(ctrl, fr_control_t);

    /*

     *  Zero the additional callback entries.

     */

    memset((uint8_t *)ctrl + sizeof(fr_control_t) + sizeof(fr_control_ctx_t) * ctrl->num_callbacks, 0,

           (id + 1 - ctrl->num_callbacks) * sizeof(fr_control_ctx_t));

    ctrl->num_callbacks = id + 1;

    *c = ctrl;

  }

  /*

   *  Re-registering the same thing is OK.

   */

  if ((ctrl->type[id].ctx == ctx) &&

      (ctrl->type[id].callback == callback)) {

    return 0;

  }

  if (ctrl->type[id].callback != NULL) {

    fr_strerror_const("Callback is already set");

    return -1;

  }

  ctrl->type[id].id = id;

  ctrl->type[id].ctx = ctx;

  ctrl->type[id].callback = callback;

  return 0;

}

/** Delete a callback for an ID

 *

 * @param[in] c the control structure

 * @param[in] id the ident of this message.

 * @return

 *  - <0 on error

 *  - 0 on success

 */

int fr_control_callback_delete(fr_control_t *c, uint32_t id)

{

  (void) talloc_get_type_abort(c, fr_control_t);

  if (id >= c->num_callbacks) {

    fr_strerror_printf("Failed deleting unknown ID %d", id);

    return -1;

  }

  if (c->type[id].callback == NULL) return 0;

  c->type[id].id = 0;

  c->type[id].ctx = NULL;

  c->type[id].callback = NULL;

  return 0;

}

int fr_control_same_thread(fr_control_t *c)

{

  c->same_thread = true;

  (void) fr_event_fd_delete(c->el, c->pipe[0], FR_EVENT_FILTER_IO);

  close(c->pipe[0]);

  close(c->pipe[1]);

  /*

   *  Nothing more to do now that everything is gone.

   */

  talloc_set_destructor(c, NULL);

  return 0;

}

/** Wait for a plane control to become readable

 *

 * This is a blocking function so only to be used in rare cases such as waiting

 * for another thread to complete a task before proceeding.

 *

 * @param[in] c the control structure.

 */

void fr_control_wait(fr_control_t *c)

{

  struct pollfd fd;

  fd.fd = c->pipe[0];

  fd.events = POLLIN;

  (void) poll(&fd, 1, -1);

}